Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -548,9 +548,11 @@ void add(DataFileMeta file) {
}
}
if (!files.isEmpty()) {
checkArgument(
file.schemaId() == files.get(0).schemaId(),
"All files in this bunch should have the same schema id.");
if (!isBlobFile(file.fileName())) {
checkArgument(
file.schemaId() == files.get(0).schemaId(),
"All files in this bunch should have the same schema id.");
}
checkArgument(
file.writeCols().equals(files.get(0).writeCols()),
"All files in this bunch should have the same write columns.");
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypeCasts;
import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.MapType;
import org.apache.paimon.types.ReassignFieldId;
import org.apache.paimon.types.RowType;
Expand Down Expand Up @@ -404,6 +405,7 @@ protected void updateLastColumn(
} else if (change instanceof RenameColumn) {
RenameColumn rename = (RenameColumn) change;
assertNotUpdatingPartitionKeys(oldTableSchema, rename.fieldNames(), "rename");
assertNotRenamingBlobColumn(newFields, rename.fieldNames());
new NestedColumnModifier(rename.fieldNames(), lazyIdentifier) {
@Override
protected void updateLastColumn(
Expand Down Expand Up @@ -908,6 +910,19 @@ private static void assertNotUpdatingPrimaryKeys(
}
}

private static void assertNotRenamingBlobColumn(List<DataField> fields, String[] fieldNames) {
if (fieldNames.length > 1) {
return;
}
String fieldName = fieldNames[0];
for (DataField field : fields) {
if (field.name().equals(fieldName) && field.type().is(DataTypeRoot.BLOB)) {
throw new UnsupportedOperationException(
String.format("Cannot rename BLOB column: [%s]", fieldName));
}
}
}

private abstract static class NestedColumnModifier {

private final String[] updateFieldNames;
Expand Down
80 changes: 80 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
Original file line number Diff line number Diff line change
Expand Up @@ -791,6 +791,86 @@ public void testJavaWriteCompressedTextAppendTable() throws Exception {
}
}

@Test
@EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
public void testBlobWriteAlterCompact() throws Exception {
Identifier identifier = identifier("blob_alter_compact_test");
catalog.dropTable(identifier, true);
Schema schema =
Schema.newBuilder()
.column("f0", DataTypes.INT())
.column("f1", DataTypes.STRING())
.column("f2", DataTypes.BLOB())
.option("target-file-size", "100 MB")
.option("row-tracking.enabled", "true")
.option("data-evolution.enabled", "true")
.option("compaction.min.file-num", "2")
.option("bucket", "-1")
.build();
catalog.createTable(identifier, schema, false);

byte[] blobBytes = new byte[1024];
new java.util.Random(42).nextBytes(blobBytes);

// Batch 1: write with schemaId=0
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
StreamTableWrite write =
table.newStreamWriteBuilder().withCommitUser(commitUser).newWrite();
StreamTableCommit commit = table.newCommit(commitUser);
for (int i = 0; i < 100; i++) {
write.write(
GenericRow.of(
1,
BinaryString.fromString("batch1"),
new org.apache.paimon.data.BlobData(blobBytes)));
}
commit.commit(0, write.prepareCommit(false, 0));

// ALTER TABLE SET -> schemaId becomes 1
catalog.alterTable(
identifier,
org.apache.paimon.schema.SchemaChange.setOption("snapshot.num-retained.min", "5"),
false);

// Batch 2: write with schemaId=1
table = (FileStoreTable) catalog.getTable(identifier);
write = table.newStreamWriteBuilder().withCommitUser(commitUser).newWrite();
commit = table.newCommit(commitUser);
for (int i = 0; i < 100; i++) {
write.write(
GenericRow.of(
2,
BinaryString.fromString("batch2"),
new org.apache.paimon.data.BlobData(blobBytes)));
}
commit.commit(1, write.prepareCommit(false, 1));
write.close();
commit.close();

// Compact
table = (FileStoreTable) catalog.getTable(identifier);
org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator coordinator =
new org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator(
table, false, false);
List<org.apache.paimon.append.dataevolution.DataEvolutionCompactTask> tasks =
coordinator.plan();
assertThat(tasks.size()).isGreaterThan(0);
List<org.apache.paimon.table.sink.CommitMessage> compactMessages = new ArrayList<>();
for (org.apache.paimon.append.dataevolution.DataEvolutionCompactTask task : tasks) {
compactMessages.add(task.doCompact(table, commitUser));
}
StreamTableCommit compactCommit = table.newCommit(commitUser);
compactCommit.commit(2, compactMessages);
compactCommit.close();

FileStoreTable readTable = (FileStoreTable) catalog.getTable(identifier);
List<Split> splits = new ArrayList<>(readTable.newSnapshotReader().read().dataSplits());
TableRead read = readTable.newRead();
RowType rowType = readTable.rowType();
List<String> res = getResult(read, splits, row -> internalRowToString(row, rowType));
assertThat(res).hasSize(200);
}

// Helper method from TableTestBase
protected Identifier identifier(String tableName) {
return new Identifier(database, tableName);
Expand Down
Loading
Loading