This is an automated email from the ASF dual-hosted git repository.
polyzos pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 5d1e18f1f [lake/iceberg] Support ADD COLUMN schema evolution for
Iceberg lake catalog (#2593)
5d1e18f1f is described below
commit 5d1e18f1f28f40d0456e12641b6789529de359ed
Author: Anton Borisov <[email protected]>
AuthorDate: Mon May 25 12:09:50 2026 +0100
[lake/iceberg] Support ADD COLUMN schema evolution for Iceberg lake catalog
(#2593)
* Iceberg Schema Evolution ADD COLUMN
* improve tests by adding rows after alter
---------
Co-authored-by: ipolyzos <[email protected]>
---
.../fluss/lake/iceberg/IcebergLakeCatalog.java | 167 ++++++-
.../fluss/lake/iceberg/IcebergLakeCatalogTest.java | 523 +++++++++++++++++++++
.../tiering/IcebergSchemaEvolutionITCase.java | 422 +++++++++++++++++
3 files changed, 1092 insertions(+), 20 deletions(-)
diff --git
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java
index 9bd9d54e6..78a49c3ac 100644
---
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java
+++
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java
@@ -19,6 +19,7 @@ package org.apache.fluss.lake.iceberg;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.TableAlreadyExistException;
import org.apache.fluss.exception.TableNotExistException;
@@ -36,6 +37,7 @@ import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
@@ -44,8 +46,11 @@ import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -133,35 +138,157 @@ public class IcebergLakeCatalog implements LakeCatalog {
throws TableNotExistException {
try {
Table table =
icebergCatalog.loadTable(toIcebergTableIdentifier(tablePath));
- UpdateProperties updateProperties = table.updateProperties();
- for (TableChange tableChange : tableChanges) {
- if (tableChange instanceof TableChange.SetOption) {
- TableChange.SetOption option = (TableChange.SetOption)
tableChange;
- checkArgument(
- !RESERVED_PROPERTIES.contains(option.getKey()),
- "Cannot set table property '%s'",
- option.getKey());
- updateProperties.set(
- convertFlussPropertyKeyToIceberg(option.getKey()),
option.getValue());
- } else if (tableChange instanceof TableChange.ResetOption) {
- TableChange.ResetOption option = (TableChange.ResetOption)
tableChange;
- checkArgument(
- !RESERVED_PROPERTIES.contains(option.getKey()),
- "Cannot reset table property '%s'",
- option.getKey());
-
updateProperties.remove(convertFlussPropertyKeyToIceberg(option.getKey()));
+
+ List<TableChange> schemaChanges = new ArrayList<>();
+ List<TableChange> propertyChanges = new ArrayList<>();
+ for (TableChange change : tableChanges) {
+ if (change instanceof TableChange.SchemaChange) {
+ schemaChanges.add(change);
} else {
- throw new UnsupportedOperationException(
- "Unsupported table change: " +
tableChange.getClass());
+ propertyChanges.add(change);
}
}
- updateProperties.commit();
+ if (!schemaChanges.isEmpty()) {
+ applySchemaChanges(table, schemaChanges, context);
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ applyPropertyChanges(table, propertyChanges);
+ }
} catch (NoSuchTableException e) {
throw new TableNotExistException("Table " + tablePath + " does not
exist.");
}
}
+ private void applyPropertyChanges(Table table, List<TableChange>
propertyChanges) {
+ UpdateProperties updateProperties = table.updateProperties();
+ for (TableChange tableChange : propertyChanges) {
+ if (tableChange instanceof TableChange.SetOption) {
+ TableChange.SetOption option = (TableChange.SetOption)
tableChange;
+ checkArgument(
+ !RESERVED_PROPERTIES.contains(option.getKey()),
+ "Cannot set table property '%s'",
+ option.getKey());
+ updateProperties.set(
+ convertFlussPropertyKeyToIceberg(option.getKey()),
option.getValue());
+ } else if (tableChange instanceof TableChange.ResetOption) {
+ TableChange.ResetOption option = (TableChange.ResetOption)
tableChange;
+ checkArgument(
+ !RESERVED_PROPERTIES.contains(option.getKey()),
+ "Cannot reset table property '%s'",
+ option.getKey());
+
updateProperties.remove(convertFlussPropertyKeyToIceberg(option.getKey()));
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported table change: " + tableChange.getClass());
+ }
+ }
+ updateProperties.commit();
+ }
+
+ private void applySchemaChanges(Table table, List<TableChange>
schemaChanges, Context context) {
+ Schema currentIcebergSchema = table.schema();
+
+ // Check schema compatibility to handle crash recovery idempotency.
+ boolean skipAddColumns;
+ if (isIcebergSchemaCompatible(currentIcebergSchema,
context.getCurrentTable())) {
+ // Iceberg schema matches current Fluss schema, apply all changes.
+ skipAddColumns = false;
+ } else if (isIcebergSchemaCompatible(currentIcebergSchema,
context.getExpectedTable())) {
+ // Iceberg schema already matches expected (post-alter) schema,
+ // skip AddColumn changes since they were already applied.
+ skipAddColumns = true;
+ } else {
+ throw new InvalidAlterTableException(
+ String.format(
+ "Iceberg schema is not compatible with Fluss
schema: "
+ + "Iceberg schema: %s, Fluss schema: %s. "
+ + "therefore you need to add the diff
columns all at once, "
+ + "rather than applying other table
changes: %s.",
+ currentIcebergSchema,
+ context.getCurrentTable().getSchema(),
+ schemaChanges));
+ }
+
+ UpdateSchema updateSchema = table.updateSchema();
+ String firstSystemColumnName =
SYSTEM_COLUMNS.keySet().iterator().next();
+ boolean hasChanges = false;
+
+ for (TableChange tableChange : schemaChanges) {
+ if (tableChange instanceof TableChange.AddColumn) {
+ if (skipAddColumns) {
+ continue;
+ }
+ TableChange.AddColumn addColumn = (TableChange.AddColumn)
tableChange;
+
+ if (!(addColumn.getPosition() instanceof TableChange.Last)) {
+ throw new UnsupportedOperationException(
+ "Only support to add column at last for iceberg
table.");
+ }
+
+ org.apache.fluss.types.DataType flussDataType =
addColumn.getDataType();
+ if (!flussDataType.isNullable()) {
+ throw new UnsupportedOperationException(
+ "Only support to add nullable column for iceberg
table.");
+ }
+
+ Type icebergType = flussDataType.accept(new
FlussDataTypeToIcebergDataType());
+ updateSchema.addColumn(addColumn.getName(), icebergType,
addColumn.getComment());
+ updateSchema.moveBefore(addColumn.getName(),
firstSystemColumnName);
+ hasChanges = true;
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported table change: " + tableChange.getClass());
+ }
+ }
+
+ if (hasChanges) {
+ updateSchema.commit();
+ }
+ }
+
+ /**
+ * Checks whether the current Iceberg schema is compatible with the given
Fluss table
+ * descriptor. Compatibility means the user columns and system columns
match in name, type, and
+ * nullability (ignoring Iceberg-assigned field IDs).
+ *
+ * <p>Iceberg reassigns field IDs during table creation, so the IDs in the
stored schema differ
+ * from those we generate via {@link #convertToIcebergSchema}. We
normalize both schemas to the
+ * same fresh ID space using {@link TypeUtil#assignIncreasingFreshIds}
before comparing, so that
+ * {@link Type#equals} works correctly for all types including complex
ones (Map, List, Struct).
+ */
+ @VisibleForTesting
+ boolean isIcebergSchemaCompatible(
+ Schema icebergSchema, @Nullable TableDescriptor
flussTableDescriptor) {
+ if (flussTableDescriptor == null) {
+ return false;
+ }
+ // isPkTable=false: identifier fields don't affect the comparison
+ Schema expectedSchema = convertToIcebergSchema(flussTableDescriptor,
false);
+
+ Schema normalizedIceberg =
TypeUtil.assignIncreasingFreshIds(icebergSchema);
+ Schema normalizedExpected =
TypeUtil.assignIncreasingFreshIds(expectedSchema);
+
+ List<Types.NestedField> currentFields = normalizedIceberg.columns();
+ List<Types.NestedField> expectedFields = normalizedExpected.columns();
+
+ if (currentFields.size() != expectedFields.size()) {
+ return false;
+ }
+
+ for (int i = 0; i < currentFields.size(); i++) {
+ Types.NestedField current = currentFields.get(i);
+ Types.NestedField expected = expectedFields.get(i);
+ if (!current.name().equals(expected.name())
+ || !current.type().equals(expected.type())
+ || current.isOptional() != expected.isOptional()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
private TableIdentifier toIcebergTableIdentifier(TablePath tablePath) {
return TableIdentifier.of(tablePath.getDatabaseName(),
tablePath.getTableName());
}
diff --git
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java
index 0625f1506..5bae68d3c 100644
---
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java
+++
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java
@@ -19,6 +19,7 @@ package org.apache.fluss.lake.iceberg;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.TableNotExistException;
import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext;
@@ -26,6 +27,7 @@ import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.coordinator.SchemaUpdate;
import org.apache.fluss.types.DataTypes;
import org.apache.iceberg.PartitionField;
@@ -50,6 +52,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.stream.Collectors;
import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
@@ -60,6 +63,14 @@ import static
org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
/** Unit test for {@link IcebergLakeCatalog}. */
class IcebergLakeCatalogTest {
+ private static final Schema FLUSS_SCHEMA =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("name", DataTypes.STRING())
+ .column("amount", DataTypes.INT())
+ .column("address", DataTypes.STRING())
+ .build();
+
@TempDir private File tempWarehouseDir;
private IcebergLakeCatalog flussIcebergCatalog;
@@ -624,4 +635,516 @@ class IcebergLakeCatalogTest {
.hasMessage("Cannot reset table property '%s'", property);
}
}
+
+ @Test
+ void testAlterTableAddColumnLastNullable() {
+ String database = "test_alter_add_col_db";
+ String tableName = "test_alter_add_col_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_col",
+ DataTypes.INT(),
+ "new_col comment",
+ TableChange.ColumnPosition.last()));
+
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+
+ Table table =
+ flussIcebergCatalog
+ .getIcebergCatalog()
+ .loadTable(TableIdentifier.of(database, tableName));
+ List<String> fieldNames =
+ table.schema().columns().stream()
+ .map(Types.NestedField::name)
+ .collect(Collectors.toList());
+ assertThat(fieldNames)
+ .containsExactly(
+ "id",
+ "name",
+ "amount",
+ "address",
+ "new_col",
+ BUCKET_COLUMN_NAME,
+ OFFSET_COLUMN_NAME,
+ TIMESTAMP_COLUMN_NAME);
+
+ // Verify the new column's type and nullability
+ Types.NestedField newCol = table.schema().findField("new_col");
+ assertThat(newCol.type()).isEqualTo(Types.IntegerType.get());
+ assertThat(newCol.isOptional()).isTrue();
+ assertThat(newCol.doc()).isEqualTo("new_col comment");
+ }
+
+ @Test
+ void testAlterTableAddColumnNotLast() {
+ String database = "test_alter_add_col_not_last_db";
+ String tableName = "test_alter_add_col_not_last_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_col",
+ DataTypes.INT(),
+ null,
+ TableChange.ColumnPosition.first()));
+
+ // Use a simple context since SchemaUpdate rejects non-LAST position
+ TableDescriptor td = getTableDescriptor(FLUSS_SCHEMA);
+ assertThatThrownBy(
+ () ->
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, new
TestingLakeCatalogContext(td)))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Only support to add column at last for iceberg
table.");
+ }
+
+ @Test
+ void testAlterTableAddColumnNotNullable() {
+ String database = "test_alter_add_col_not_nullable_db";
+ String tableName = "test_alter_add_col_not_nullable_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_col",
+ DataTypes.INT().copy(false),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ // Use a simple context since SchemaUpdate rejects non-nullable columns
+ TableDescriptor td = getTableDescriptor(FLUSS_SCHEMA);
+ assertThatThrownBy(
+ () ->
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, new
TestingLakeCatalogContext(td)))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Only support to add nullable column for iceberg
table.");
+ }
+
+ @Test
+ void testAlterTableAddExistingColumns() {
+ String database = "test_alter_add_existing_col_db";
+ String tableName = "test_alter_add_existing_col_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ // Simulate crash recovery: add columns to iceberg first, then retry
+ List<TableChange> changes =
+ Arrays.asList(
+ TableChange.addColumn(
+ "new_column",
+ DataTypes.INT(),
+ null,
+ TableChange.ColumnPosition.last()),
+ TableChange.addColumn(
+ "new_column2",
+ DataTypes.STRING(),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ // First call succeeds - adds columns to iceberg
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+
+ // Second call with same changes should succeed (idempotent via schema
compatibility)
+ // because iceberg schema now matches the expected schema
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+ }
+
+ @Test
+ void testAlterTableAddColumnWhenIcebergSchemaNotMatch() {
+ String database = "test_alter_add_col_schema_mismatch_db";
+ String tableName = "test_alter_add_col_schema_mismatch_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_col",
+ DataTypes.INT(),
+ "new_col comment",
+ TableChange.ColumnPosition.last()));
+
+ // Test column number mismatch: pass a wider Fluss schema that doesn't
match iceberg
+ Schema widerFlussSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("name", DataTypes.STRING())
+ .column("amount", DataTypes.INT())
+ .column("address", DataTypes.STRING())
+ .column("phone", DataTypes.INT())
+ .build();
+ assertThatThrownBy(
+ () ->
+ flussIcebergCatalog.alterTable(
+ tablePath,
+ changes,
+
getLakeCatalogContext(widerFlussSchema, changes)))
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining("Iceberg schema is not compatible with
Fluss schema");
+
+ // Test column order mismatch
+ Schema disorderFlussSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("amount", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("address", DataTypes.STRING())
+ .build();
+ assertThatThrownBy(
+ () ->
+ flussIcebergCatalog.alterTable(
+ tablePath,
+ changes,
+
getLakeCatalogContext(disorderFlussSchema, changes)))
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining("Iceberg schema is not compatible with
Fluss schema");
+ }
+
+ @Test
+ void testAlterTableAddColumnWithComplexTypeTable() {
+ String database = "test_alter_add_col_complex_db";
+ String tableName = "test_alter_add_col_complex_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+
+ // Create table with complex types (Map, Array)
+ Schema complexSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("tags", DataTypes.ARRAY(DataTypes.STRING()))
+ .column("metadata", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT()))
+ .build();
+
+ TableDescriptor td = getTableDescriptor(complexSchema);
+ flussIcebergCatalog.createTable(tablePath, td, new
TestingLakeCatalogContext());
+
+ // ADD COLUMN should succeed despite Iceberg reassigning field IDs for
complex types
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_col",
+ DataTypes.STRING(),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(complexSchema,
changes));
+
+ Table table =
+ flussIcebergCatalog
+ .getIcebergCatalog()
+ .loadTable(TableIdentifier.of(database, tableName));
+ List<String> fieldNames =
+ table.schema().columns().stream()
+ .map(Types.NestedField::name)
+ .collect(Collectors.toList());
+ assertThat(fieldNames)
+ .containsExactly(
+ "id",
+ "tags",
+ "metadata",
+ "new_col",
+ BUCKET_COLUMN_NAME,
+ OFFSET_COLUMN_NAME,
+ TIMESTAMP_COLUMN_NAME);
+
+ // Verify the new column
+ Types.NestedField newCol = table.schema().findField("new_col");
+ assertThat(newCol.type()).isEqualTo(Types.StringType.get());
+ assertThat(newCol.isOptional()).isTrue();
+ }
+
+ @Test
+ void testAlterTableAddColumnIdempotencyWithComplexTypes() {
+ String database = "test_alter_idempotent_complex_db";
+ String tableName = "test_alter_idempotent_complex_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+
+ Schema complexSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("data", DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING()))
+ .build();
+
+ TableDescriptor td = getTableDescriptor(complexSchema);
+ flussIcebergCatalog.createTable(tablePath, td, new
TestingLakeCatalogContext());
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_col",
+ DataTypes.INT(),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ // First call succeeds
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(complexSchema,
changes));
+
+ // Second call (crash recovery) should succeed via idempotency check
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(complexSchema,
changes));
+ }
+
+ @Test
+ void testAlterTableAddArrayColumn() {
+ String database = "test_alter_add_array_db";
+ String tableName = "test_alter_add_array_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_arr",
+ DataTypes.ARRAY(DataTypes.STRING()),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+
+ Table table =
+ flussIcebergCatalog
+ .getIcebergCatalog()
+ .loadTable(TableIdentifier.of(database, tableName));
+ Types.NestedField field = table.schema().findField("new_arr");
+ assertThat(field).isNotNull();
+ assertThat(field.isOptional()).isTrue();
+ assertThat(field.type().isListType()).isTrue();
+ Types.ListType listType = field.type().asListType();
+ assertThat(listType.elementType()).isEqualTo(Types.StringType.get());
+ assertThat(listType.elementId()).isNotEqualTo(field.fieldId());
+ assertThat(table.schema().columns())
+ .extracting(Types.NestedField::name)
+ .containsSubsequence("new_arr", BUCKET_COLUMN_NAME);
+ }
+
+ @Test
+ void testAlterTableAddMapColumn() {
+ String database = "test_alter_add_map_db";
+ String tableName = "test_alter_add_map_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_map",
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT()),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+
+ Table table =
+ flussIcebergCatalog
+ .getIcebergCatalog()
+ .loadTable(TableIdentifier.of(database, tableName));
+ Types.NestedField field = table.schema().findField("new_map");
+ assertThat(field.type().isMapType()).isTrue();
+ Types.MapType mapType = field.type().asMapType();
+ assertThat(mapType.keyType()).isEqualTo(Types.StringType.get());
+ assertThat(mapType.valueType()).isEqualTo(Types.IntegerType.get());
+ assertThat(mapType.keyId()).isNotEqualTo(mapType.valueId());
+ assertThat(mapType.keyId()).isNotEqualTo(field.fieldId());
+ assertThat(mapType.valueId()).isNotEqualTo(field.fieldId());
+ }
+
+ @Test
+ void testAlterTableAddRowColumn() {
+ String database = "test_alter_add_row_db";
+ String tableName = "test_alter_add_row_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_row",
+ DataTypes.ROW(
+ DataTypes.FIELD("a", DataTypes.INT()),
+ DataTypes.FIELD("b",
DataTypes.STRING())),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+
+ Table table =
+ flussIcebergCatalog
+ .getIcebergCatalog()
+ .loadTable(TableIdentifier.of(database, tableName));
+ Types.NestedField field = table.schema().findField("new_row");
+ assertThat(field.type().isStructType()).isTrue();
+ Types.StructType struct = field.type().asStructType();
+
assertThat(struct.fields()).extracting(Types.NestedField::name).containsExactly("a",
"b");
+
assertThat(struct.field("a").type()).isEqualTo(Types.IntegerType.get());
+ assertThat(struct.field("b").type()).isEqualTo(Types.StringType.get());
+
assertThat(struct.field("a").fieldId()).isNotEqualTo(struct.field("b").fieldId());
+ assertThat(struct.field("a").fieldId()).isNotEqualTo(field.fieldId());
+ }
+
+ @Test
+ void testAlterTableAddNestedComplexColumn() {
+ String database = "test_alter_add_nested_db";
+ String tableName = "test_alter_add_nested_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_nested",
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD("x",
DataTypes.INT()),
+ DataTypes.FIELD("y",
DataTypes.STRING()))),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+
+ Table table =
+ flussIcebergCatalog
+ .getIcebergCatalog()
+ .loadTable(TableIdentifier.of(database, tableName));
+ Types.NestedField field = table.schema().findField("new_nested");
+ Types.ListType listType = field.type().asListType();
+ Types.StructType struct = listType.elementType().asStructType();
+
assertThat(struct.fields()).extracting(Types.NestedField::name).containsExactly("x",
"y");
+
+ Set<Integer> ids = new HashSet<>();
+ ids.add(field.fieldId());
+ ids.add(listType.elementId());
+ ids.add(struct.field("x").fieldId());
+ ids.add(struct.field("y").fieldId());
+ assertThat(ids).hasSize(4);
+ }
+
+ @Test
+ void testAlterTableAddDecimalColumnPreservesPrecisionScale() {
+ String database = "test_alter_add_decimal_db";
+ String tableName = "test_alter_add_decimal_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_dec",
+ DataTypes.DECIMAL(20, 5),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+
+ Table table =
+ flussIcebergCatalog
+ .getIcebergCatalog()
+ .loadTable(TableIdentifier.of(database, tableName));
+ assertThat(table.schema().findField("new_dec").type())
+ .isEqualTo(Types.DecimalType.of(20, 5));
+ }
+
+ @Test
+ void testAlterTableAddTimestampVariants() {
+ String database = "test_alter_add_ts_db";
+ String tableName = "test_alter_add_ts_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+ createLogTable(database, tableName);
+
+ List<TableChange> changes =
+ Arrays.asList(
+ TableChange.addColumn(
+ "new_ts",
+ DataTypes.TIMESTAMP(6),
+ null,
+ TableChange.ColumnPosition.last()),
+ TableChange.addColumn(
+ "new_ts_ltz",
+ DataTypes.TIMESTAMP_LTZ(6),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ flussIcebergCatalog.alterTable(
+ tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA,
changes));
+
+ Table table =
+ flussIcebergCatalog
+ .getIcebergCatalog()
+ .loadTable(TableIdentifier.of(database, tableName));
+ assertThat(table.schema().findField("new_ts").type())
+ .isEqualTo(Types.TimestampType.withoutZone());
+ assertThat(table.schema().findField("new_ts_ltz").type())
+ .isEqualTo(Types.TimestampType.withZone());
+ }
+
+ @Test
+ void testAlterTableComplexTypeMismatch() {
+ String database = "test_alter_complex_mismatch_db";
+ String tableName = "test_alter_complex_mismatch_table";
+ TablePath tablePath = TablePath.of(database, tableName);
+
+ Schema actualSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("tags", DataTypes.ARRAY(DataTypes.STRING()))
+ .build();
+ flussIcebergCatalog.createTable(
+ tablePath, getTableDescriptor(actualSchema), new
TestingLakeCatalogContext());
+
+ Schema mismatched =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("tags", DataTypes.ARRAY(DataTypes.INT()))
+ .build();
+ List<TableChange> changes =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_col",
+ DataTypes.STRING(),
+ null,
+ TableChange.ColumnPosition.last()));
+
+ assertThatThrownBy(
+ () ->
+ flussIcebergCatalog.alterTable(
+ tablePath,
+ changes,
+ getLakeCatalogContext(mismatched,
changes)))
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining("Iceberg schema is not compatible with
Fluss schema");
+ }
+
+ private void createLogTable(String database, String tableName) {
+ TableDescriptor td = getTableDescriptor(FLUSS_SCHEMA);
+ TablePath tablePath = TablePath.of(database, tableName);
+ flussIcebergCatalog.createTable(tablePath, td, new
TestingLakeCatalogContext());
+ }
+
+ private TestingLakeCatalogContext getLakeCatalogContext(
+ Schema schema, List<TableChange> schemaChanges) {
+ Schema expectedSchema = SchemaUpdate.applySchemaChanges(schema,
schemaChanges);
+ return new TestingLakeCatalogContext(
+ getTableDescriptor(schema),
getTableDescriptor(expectedSchema));
+ }
+
+ private TableDescriptor getTableDescriptor(Schema schema) {
+ return
TableDescriptor.builder().schema(schema).distributedBy(3).build();
+ }
}
diff --git
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergSchemaEvolutionITCase.java
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergSchemaEvolutionITCase.java
new file mode 100644
index 000000000..d4177a305
--- /dev/null
+++
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergSchemaEvolutionITCase.java
@@ -0,0 +1,422 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.iceberg.tiering;
+
+import org.apache.fluss.lake.iceberg.testutils.FlinkIcebergTieringTestBase;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableChange;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.DataTypes;
+
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg;
+import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** ITCase for schema evolution (ADD COLUMN) with Iceberg tiering. */
+class IcebergSchemaEvolutionITCase extends FlinkIcebergTieringTestBase {
+
+ private static final String DEFAULT_DB = "fluss";
+
+ private static StreamExecutionEnvironment execEnv;
+
+ private static final Schema INITIAL_LOG_SCHEMA =
+ Schema.newBuilder()
+ .column("f_int", DataTypes.INT())
+ .column("f_str", DataTypes.STRING())
+ .build();
+
+ private static final Schema INITIAL_PK_SCHEMA =
+ Schema.newBuilder()
+ .column("f_int", DataTypes.INT())
+ .column("f_str", DataTypes.STRING())
+ .primaryKey("f_int")
+ .build();
+
+ private static final Schema COMPLEX_LOG_SCHEMA =
+ Schema.newBuilder()
+ .column("f_int", DataTypes.INT())
+ .column("f_str", DataTypes.STRING())
+ .column("f_tags", DataTypes.ARRAY(DataTypes.STRING()))
+ .column("f_meta", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT()))
+ .build();
+
+ @BeforeAll
+ protected static void beforeAll() {
+ FlinkIcebergTieringTestBase.beforeAll();
+ execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ execEnv.setParallelism(2);
+ execEnv.enableCheckpointing(1000);
+ }
+
+ @Test
+ void testSchemaEvolutionLogTable() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "schemaEvoLogTable");
+ long tableId = createLogTable(tablePath, 1, false, INITIAL_LOG_SCHEMA);
+ TableBucket bucket = new TableBucket(tableId, 0);
+
+ // Write initial rows and start tiering
+ List<InternalRow> initialRows = Arrays.asList(row(1, "v1"), row(2,
"v2"), row(3, "v3"));
+ writeRows(tablePath, initialRows, true);
+
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ // Wait until initial data is tiered
+ assertReplicaStatus(bucket, 3);
+ checkDataInIcebergAppendOnlyTable(tablePath, initialRows, 0);
+
+ // Verify initial Iceberg schema has no extra columns
+ org.apache.iceberg.Table icebergTable =
icebergCatalog.loadTable(toIceberg(tablePath));
+ assertThat(icebergTable.schema().findField("f_new")).isNull();
+
+ // ALTER TABLE ADD COLUMN via Admin API
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "f_new",
+ DataTypes.STRING(),
+ null,
+
TableChange.ColumnPosition.last())),
+ false)
+ .get();
+
+ // Verify the Iceberg schema now has the new column before system
columns
+ icebergTable.refresh();
+ Types.NestedField newField =
icebergTable.schema().findField("f_new");
+ assertThat(newField).isNotNull();
+ assertThat(newField.type()).isEqualTo(Types.StringType.get());
+ assertThat(newField.isOptional()).isTrue();
+
+ // Verify column ordering: new column should be before __bucket
+ List<String> fieldNames =
+ icebergTable.schema().columns().stream()
+ .map(Types.NestedField::name)
+ .collect(Collectors.toList());
+ assertThat(fieldNames.indexOf("f_new"))
+ .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME));
+
+ // Post-ALTER rows: new rows carry f_new values; old rows must
surface NULL
+ // via Iceberg field-ID schema evolution.
+ List<InternalRow> postAlterRows =
+ Arrays.asList(row(4, "v4", "newA"), row(5, "v5", "newB"));
+ writeRows(tablePath, postAlterRows, true);
+ assertReplicaStatus(bucket, 5);
+
+ icebergTable.refresh();
+ List<Record> records = getIcebergRecords(tablePath);
+ assertThat(records).hasSize(5);
+ Map<Integer, Object> fNewByFInt = new HashMap<>();
+ for (Record record : records) {
+ fNewByFInt.put((Integer) record.getField("f_int"),
record.getField("f_new"));
+ }
+ assertThat(fNewByFInt.get(1)).isNull();
+ assertThat(fNewByFInt.get(2)).isNull();
+ assertThat(fNewByFInt.get(3)).isNull();
+ assertThat(fNewByFInt.get(4)).isEqualTo("newA");
+ assertThat(fNewByFInt.get(5)).isEqualTo("newB");
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ @Test
+ void testSchemaEvolutionPkTable() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "schemaEvoPkTable");
+ long tableId = createPkTable(tablePath, 1, false, INITIAL_PK_SCHEMA);
+ TableBucket bucket = new TableBucket(tableId, 0);
+
+ // Write initial rows and trigger snapshot
+ List<InternalRow> initialRows = Arrays.asList(row(1, "v1"), row(2,
"v2"), row(3, "v3"));
+ writeRows(tablePath, initialRows, false);
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tablePath);
+
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ // Wait until initial data is tiered
+ assertReplicaStatus(bucket, 3);
+
+ // Verify initial data in Iceberg
+ List<Record> records = getIcebergRecords(tablePath);
+ assertThat(records).hasSize(3);
+
+ // ALTER TABLE ADD COLUMN
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "f_new",
+ DataTypes.INT(),
+ "new column",
+
TableChange.ColumnPosition.last())),
+ false)
+ .get();
+
+ // Verify the Iceberg schema now has the new column
+ org.apache.iceberg.Table icebergTable =
icebergCatalog.loadTable(toIceberg(tablePath));
+ Types.NestedField newField =
icebergTable.schema().findField("f_new");
+ assertThat(newField).isNotNull();
+ assertThat(newField.type()).isEqualTo(Types.IntegerType.get());
+ assertThat(newField.isOptional()).isTrue();
+ assertThat(newField.doc()).isEqualTo("new column");
+
+ // Verify column ordering
+ List<String> fieldNames =
+ icebergTable.schema().columns().stream()
+ .map(Types.NestedField::name)
+ .collect(Collectors.toList());
+ assertThat(fieldNames.indexOf("f_new"))
+ .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME));
+
+ // Post-ALTER upserts + snapshot trigger; verify all rows tier.
+ List<InternalRow> postAlterRows = Arrays.asList(row(4, "v4", 100),
row(5, "v5", 200));
+ writeRows(tablePath, postAlterRows, false);
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tablePath);
+ assertReplicaStatus(bucket, 5);
+
+ icebergTable.refresh();
+ List<Record> postAlterRecords = getIcebergRecords(tablePath);
+ assertThat(postAlterRecords).hasSize(5);
+ Map<Integer, Object> fNewByFInt = new HashMap<>();
+ for (Record record : postAlterRecords) {
+ fNewByFInt.put((Integer) record.getField("f_int"),
record.getField("f_new"));
+ }
+ assertThat(fNewByFInt.get(1)).isNull();
+ assertThat(fNewByFInt.get(2)).isNull();
+ assertThat(fNewByFInt.get(3)).isNull();
+ assertThat(fNewByFInt.get(4)).isEqualTo(100);
+ assertThat(fNewByFInt.get(5)).isEqualTo(200);
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ @Test
+ void testSchemaEvolutionLogTableWithComplexTypes() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB,
"schemaEvoComplexLogTable");
+ long tableId = createLogTable(tablePath, 1, false, COMPLEX_LOG_SCHEMA);
+ TableBucket bucket = new TableBucket(tableId, 0);
+
+ // Write initial rows (nulls for complex type columns)
+ List<InternalRow> initialRows =
+ Arrays.asList(
+ row(1, "v1", null, null),
+ row(2, "v2", null, null),
+ row(3, "v3", null, null));
+ writeRows(tablePath, initialRows, true);
+
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ // Wait until initial data is tiered
+ assertReplicaStatus(bucket, 3);
+
+ // Verify initial Iceberg schema has complex type columns
+ org.apache.iceberg.Table icebergTable =
icebergCatalog.loadTable(toIceberg(tablePath));
+
assertThat(icebergTable.schema().findField("f_tags").type().isListType()).isTrue();
+
assertThat(icebergTable.schema().findField("f_meta").type().isMapType()).isTrue();
+
+ // ALTER TABLE ADD COLUMN — this exercises the compatibility check
+ // with complex types whose field IDs were reassigned by Iceberg
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "f_new",
+ DataTypes.STRING(),
+ null,
+
TableChange.ColumnPosition.last())),
+ false)
+ .get();
+
+ // Verify the Iceberg schema now has the new column
+ icebergTable.refresh();
+ Types.NestedField newField =
icebergTable.schema().findField("f_new");
+ assertThat(newField).isNotNull();
+ assertThat(newField.type()).isEqualTo(Types.StringType.get());
+ assertThat(newField.isOptional()).isTrue();
+
+ // Verify column ordering: new column before system columns
+ List<String> fieldNames =
+ icebergTable.schema().columns().stream()
+ .map(Types.NestedField::name)
+ .collect(Collectors.toList());
+ assertThat(fieldNames.indexOf("f_new"))
+ .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME));
+
+
assertThat(icebergTable.schema().findField("f_tags").type().isListType()).isTrue();
+
assertThat(icebergTable.schema().findField("f_meta").type().isMapType()).isTrue();
+
+ // Post-ALTER rows alongside pre-existing complex columns.
+ List<InternalRow> postAlterRows =
+ Arrays.asList(
+ row(4, "v4", null, null, "newA"), row(5, "v5",
null, null, "newB"));
+ writeRows(tablePath, postAlterRows, true);
+ assertReplicaStatus(bucket, 5);
+
+ icebergTable.refresh();
+ List<Record> records = getIcebergRecords(tablePath);
+ assertThat(records).hasSize(5);
+ Map<Integer, Object> fNewByFInt = new HashMap<>();
+ for (Record record : records) {
+ fNewByFInt.put((Integer) record.getField("f_int"),
record.getField("f_new"));
+ }
+ assertThat(fNewByFInt.get(1)).isNull();
+ assertThat(fNewByFInt.get(2)).isNull();
+ assertThat(fNewByFInt.get(3)).isNull();
+ assertThat(fNewByFInt.get(4)).isEqualTo("newA");
+ assertThat(fNewByFInt.get(5)).isEqualTo("newB");
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ @Test
+ void testSchemaEvolutionLogTableAddComplexColumn() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB,
"schemaEvoLogAddComplex");
+ long tableId = createLogTable(tablePath, 1, false, INITIAL_LOG_SCHEMA);
+ TableBucket bucket = new TableBucket(tableId, 0);
+ List<InternalRow> initialRows = Arrays.asList(row(1, "v1"), row(2,
"v2"), row(3, "v3"));
+ writeRows(tablePath, initialRows, true);
+
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ assertReplicaStatus(bucket, 3);
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "f_tags",
+
DataTypes.ARRAY(DataTypes.STRING()),
+ null,
+
TableChange.ColumnPosition.last())),
+ false)
+ .get();
+
+ org.apache.iceberg.Table icebergTable =
icebergCatalog.loadTable(toIceberg(tablePath));
+ Types.NestedField added =
icebergTable.schema().findField("f_tags");
+ assertThat(added).isNotNull();
+ assertThat(added.type().isListType()).isTrue();
+
assertThat(added.type().asListType().elementType()).isEqualTo(Types.StringType.get());
+ List<String> fieldNames =
+ icebergTable.schema().columns().stream()
+ .map(Types.NestedField::name)
+ .collect(Collectors.toList());
+ assertThat(fieldNames.indexOf("f_tags"))
+ .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME));
+
+ // Post-ALTER rows for a complex new column. Writing null
exercises the
+ // data-plane path through the new ARRAY field ID without forcing
the test
+ // to build an InternalArray literal.
+ List<InternalRow> postAlterRows = Arrays.asList(row(4, "v4",
null), row(5, "v5", null));
+ writeRows(tablePath, postAlterRows, true);
+ assertReplicaStatus(bucket, 5);
+
+ icebergTable.refresh();
+ List<Record> records = getIcebergRecords(tablePath);
+ assertThat(records).hasSize(5);
+ for (Record record : records) {
+ assertThat(record.getField("f_tags")).isNull();
+ }
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ @Test
+ void testSchemaEvolutionPkTableWithComplexPreExisting() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "schemaEvoPkComplex");
+ Schema pkComplexSchema =
+ Schema.newBuilder()
+ .column("f_int", DataTypes.INT())
+ .column("f_str", DataTypes.STRING())
+ .column("f_tags", DataTypes.ARRAY(DataTypes.STRING()))
+ .primaryKey("f_int")
+ .build();
+ long tableId = createPkTable(tablePath, 1, false, pkComplexSchema);
+ TableBucket bucket = new TableBucket(tableId, 0);
+
+ List<InternalRow> initialRows =
+ Arrays.asList(row(1, "v1", null), row(2, "v2", null), row(3,
"v3", null));
+ writeRows(tablePath, initialRows, false);
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tablePath);
+
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ assertReplicaStatus(bucket, 3);
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "f_new",
+ DataTypes.INT(),
+ null,
+
TableChange.ColumnPosition.last())),
+ false)
+ .get();
+
+ org.apache.iceberg.Table icebergTable =
icebergCatalog.loadTable(toIceberg(tablePath));
+ assertThat(icebergTable.schema().findField("f_new").type())
+ .isEqualTo(Types.IntegerType.get());
+
assertThat(icebergTable.schema().findField("f_tags").type().isListType()).isTrue();
+ List<String> fieldNames =
+ icebergTable.schema().columns().stream()
+ .map(Types.NestedField::name)
+ .collect(Collectors.toList());
+ assertThat(fieldNames.indexOf("f_new"))
+ .isLessThan(fieldNames.indexOf(BUCKET_COLUMN_NAME));
+
+ // Post-ALTER upserts with f_new values + snapshot trigger.
+ List<InternalRow> postAlterRows =
+ Arrays.asList(row(4, "v4", null, 100), row(5, "v5", null,
200));
+ writeRows(tablePath, postAlterRows, false);
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tablePath);
+ assertReplicaStatus(bucket, 5);
+
+ icebergTable.refresh();
+ List<Record> records = getIcebergRecords(tablePath);
+ assertThat(records).hasSize(5);
+ Map<Integer, Object> fNewByFInt = new HashMap<>();
+ for (Record record : records) {
+ fNewByFInt.put((Integer) record.getField("f_int"),
record.getField("f_new"));
+ }
+ assertThat(fNewByFInt.get(1)).isNull();
+ assertThat(fNewByFInt.get(2)).isNull();
+ assertThat(fNewByFInt.get(3)).isNull();
+ assertThat(fNewByFInt.get(4)).isEqualTo(100);
+ assertThat(fNewByFInt.get(5)).isEqualTo(200);
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+}
