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 78a49c3ac3..e9e5b102d3 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 @@ -30,9 +30,11 @@ import org.apache.fluss.metadata.TablePath; import org.apache.fluss.utils.IOUtils; +import org.apache.iceberg.PartitionField; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.RowLevelOperationMode; import org.apache.iceberg.Schema; +import org.apache.iceberg.SortField; import org.apache.iceberg.SortOrder; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; @@ -112,15 +114,30 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co PartitionSpec partitionSpec = createPartitionSpec(tableDescriptor, icebergSchema, isPkTable); SortOrder sortOrder = createSortOrder(icebergSchema); - tableBuilder.withProperties(buildTableProperties(tableDescriptor, isPkTable)); + Map expectedProperties = buildTableProperties(tableDescriptor, isPkTable); + tableBuilder.withProperties(expectedProperties); tableBuilder.withPartitionSpec(partitionSpec); tableBuilder.withSortOrder(sortOrder); try { - createTable(tablePath, tableBuilder); + createTable( + tablePath, + tableBuilder, + icebergSchema, + partitionSpec, + sortOrder, + expectedProperties, + context.isCreatingFlussTable()); } catch (NoSuchNamespaceException e) { createDatabase(tablePath.getDatabaseName()); try { - createTable(tablePath, tableBuilder); + createTable( + tablePath, + tableBuilder, + icebergSchema, + partitionSpec, + sortOrder, + expectedProperties, + context.isCreatingFlussTable()); } catch (NoSuchNamespaceException t) { // shouldn't happen in normal cases throw new RuntimeException( @@ -293,14 +310,189 @@ private TableIdentifier toIcebergTableIdentifier(TablePath tablePath) { return TableIdentifier.of(tablePath.getDatabaseName(), tablePath.getTableName()); } - private void createTable(TablePath tablePath, Catalog.TableBuilder tableBuilder) { + private void createTable( + TablePath tablePath, + Catalog.TableBuilder tableBuilder, + Schema newIcebergSchema, + PartitionSpec expectedSpec, + SortOrder expectedSortOrder, + Map expectedProperties, + boolean isCreatingFlussTable) { try { tableBuilder.create(); } catch (AlreadyExistsException e) { - throw new TableAlreadyExistException("Table " + tablePath + " already exists."); + // Table already exists in Iceberg. Check schema compatibility for idempotency, + TableIdentifier icebergId = toIcebergTableIdentifier(tablePath); + Table existingTable = icebergCatalog.loadTable(icebergId); + Schema existingSchema = existingTable.schema(); + if (!isIcebergSchemaCompatibleWithSchema(existingSchema, newIcebergSchema)) { + throw new TableAlreadyExistException( + String.format( + "The table %s already exists in Iceberg catalog, but the table schema is not compatible. " + + "Existing schema: %s, new schema: %s. " + + "Please first drop the table in Iceberg catalog or use a new table name.", + tablePath, existingSchema, newIcebergSchema), + e); + } + + if (!isIcebergPartitionSpecCompatible( + existingTable.spec(), existingSchema, expectedSpec, newIcebergSchema)) { + throw new TableAlreadyExistException( + String.format( + "The table %s already exists in Iceberg catalog, but the partition spec is not compatible. " + + "Existing spec: %s, new spec: %s. " + + "Please first drop the table in Iceberg catalog or use a new table name.", + tablePath, existingTable.spec(), expectedSpec), + e); + } + + if (!isIcebergSortOrderCompatible( + existingTable.sortOrder(), + existingSchema, + expectedSortOrder, + newIcebergSchema)) { + throw new TableAlreadyExistException( + String.format( + "The table %s already exists in Iceberg catalog, but the sort order is not compatible. " + + "Existing sort order: %s, new sort order: %s. " + + "Please first drop the table in Iceberg catalog, or pre-create the " + + "Iceberg table without a custom sort order, or with ASC(%s) set explicitly, " + + "or use a new table name.", + tablePath, + existingTable.sortOrder(), + expectedSortOrder, + OFFSET_COLUMN_NAME), + e); + } + + if (!isIcebergPropertiesCompatible(existingTable.properties(), expectedProperties)) { + throw new TableAlreadyExistException( + String.format( + "The table %s already exists in Iceberg catalog, but the table properties are not compatible. " + + "Existing properties: %s, new properties: %s. " + + "Please first drop the table in Iceberg catalog or use a new table name.", + tablePath, existingTable.properties(), expectedProperties), + e); + } + + // if creating a new fluss table, we should ensure the lake table is empty + if (isCreatingFlussTable && existingTable.currentSnapshot() != null) { + throw new TableAlreadyExistException( + String.format( + "The table %s already exists in Iceberg catalog, and the table is not empty. " + + "Please first drop the table in Iceberg catalog or use a new table name.", + tablePath), + e); + } } } + @VisibleForTesting + boolean isIcebergSchemaCompatibleWithSchema(Schema existingSchema, Schema newSchema) { + Schema normalizedExisting = TypeUtil.assignIncreasingFreshIds(existingSchema); + Schema normalizedNew = TypeUtil.assignIncreasingFreshIds(newSchema); + + List existingFields = normalizedExisting.columns(); + List newFields = normalizedNew.columns(); + + if (existingFields.size() != newFields.size()) { + return false; + } + + for (int i = 0; i < existingFields.size(); i++) { + Types.NestedField existing = existingFields.get(i); + Types.NestedField expected = newFields.get(i); + if (!existing.name().equals(expected.name()) + || !existing.type().equals(expected.type()) + || existing.isOptional() != expected.isOptional()) { + return false; + } + } + + return true; + } + + /** Checks whether the existing Iceberg partition spec is compatible with the expected one. */ + @VisibleForTesting + boolean isIcebergPartitionSpecCompatible( + PartitionSpec existingSpec, + Schema existingSchema, + PartitionSpec expectedSpec, + Schema expectedSchema) { + if (existingSpec.fields().size() != expectedSpec.fields().size()) { + return false; + } + + for (int i = 0; i < existingSpec.fields().size(); i++) { + PartitionField existing = existingSpec.fields().get(i); + PartitionField expected = expectedSpec.fields().get(i); + + String existingSource = existingSchema.findColumnName(existing.sourceId()); + String expectedSource = expectedSchema.findColumnName(expected.sourceId()); + if (existingSource == null || !existingSource.equals(expectedSource)) { + return false; + } + + if (!existing.name().equals(expected.name())) { + return false; + } + + if (!existing.transform().equals(expected.transform())) { + return false; + } + } + return true; + } + + @VisibleForTesting + boolean isIcebergSortOrderCompatible( + SortOrder existingOrder, + Schema existingSchema, + SortOrder expectedOrder, + Schema expectedSchema) { + if (existingOrder.fields().size() != expectedOrder.fields().size()) { + return false; + } + + for (int i = 0; i < existingOrder.fields().size(); i++) { + SortField existing = existingOrder.fields().get(i); + SortField expected = expectedOrder.fields().get(i); + + String existingSource = existingSchema.findColumnName(existing.sourceId()); + String expectedSource = expectedSchema.findColumnName(expected.sourceId()); + if (existingSource == null || !existingSource.equals(expectedSource)) { + return false; + } + + if (!existing.transform().equals(expected.transform())) { + return false; + } + + if (existing.direction() != expected.direction()) { + return false; + } + + if (existing.nullOrder() != expected.nullOrder()) { + return false; + } + } + + return true; + } + + @VisibleForTesting + boolean isIcebergPropertiesCompatible( + Map existingProperties, Map expectedProperties) { + for (Map.Entry entry : expectedProperties.entrySet()) { + String actual = existingProperties.get(entry.getKey()); + if (actual == null || !actual.equals(entry.getValue())) { + return false; + } + } + + return true; + } + public Schema convertToIcebergSchema(TableDescriptor tableDescriptor, boolean isPkTable) { List fields = new ArrayList<>(); int fieldId = 0; 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 5bae68d3c2..8c75bdbe41 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 @@ -21,6 +21,7 @@ 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; import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext; import org.apache.fluss.metadata.Schema; @@ -30,11 +31,17 @@ import org.apache.fluss.server.coordinator.SchemaUpdate; import org.apache.fluss.types.DataTypes; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.PartitionField; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.RowLevelOperationMode; import org.apache.iceberg.SortDirection; import org.apache.iceberg.SortField; +import org.apache.iceberg.SortOrder; import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.types.Types; @@ -51,6 +58,7 @@ import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.stream.Collectors; @@ -1131,6 +1139,469 @@ void testAlterTableComplexTypeMismatch() { .hasMessageContaining("Iceberg schema is not compatible with Fluss schema"); } + /** + * If the Iceberg table already exists with a compatible schema, re-creating the same Fluss + * table should succeed idempotently as long as the caller does not claim it is the first + * creation of the Fluss table. + */ + @Test + void testCreateTableIdempotentWhenCompatibleLakeTableExists() { + String database = "reuse_existing_db"; + String tableName = "reuse_existing_table"; + TablePath tablePath = TablePath.of(database, tableName); + + TableDescriptor td = getTableDescriptor(FLUSS_SCHEMA); + + // First creation should succeed + flussIcebergCatalog.createTable(tablePath, td, new TestingLakeCatalogContext()); + + // Second creation with the same schema and isCreatingFlussTable=false should succeed + // idempotently. + flussIcebergCatalog.createTable(tablePath, td, new TestingLakeCatalogContext()); + + Table table = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + assertThat(table).isNotNull(); + } + + /** + * When creating a fresh Fluss table, if a compatible but empty Iceberg table already exists, + * the creation should still succeed (allowing a Fluss table to bind to a pre-created empty lake + * table). + */ + @Test + void testCreateFlussTableWithExistingEmptyLakeTable() { + String database = "empty_lake_db"; + String tableName = "empty_lake_table"; + TablePath tablePath = TablePath.of(database, tableName); + + TableDescriptor td = getTableDescriptor(FLUSS_SCHEMA); + + // First creation: prepares the Iceberg table (still empty, no snapshot) + flussIcebergCatalog.createTable(tablePath, td, new TestingLakeCatalogContext()); + Table before = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + assertThat(before.currentSnapshot()).isNull(); + + // A brand-new Fluss table binding to an existing empty compatible Iceberg table + // should be allowed. + TestingLakeCatalogContext firstFlussCreationCtx = + new TestingLakeCatalogContext() { + @Override + public boolean isCreatingFlussTable() { + return true; + } + }; + + flussIcebergCatalog.createTable(tablePath, td, firstFlussCreationCtx); + } + + /** + * When creating a fresh Fluss table, if a compatible but non-empty Iceberg table already + * exists, the creation should fail with a clear "not empty" message. + */ + @Test + void testCreateFlussTableFailsWhenExistingLakeTableIsNonEmpty() { + String database = "non_empty_lake_db"; + String tableName = "non_empty_lake_table"; + TablePath tablePath = TablePath.of(database, tableName); + + TableDescriptor td = getTableDescriptor(FLUSS_SCHEMA); + + // First: create the Iceberg table. + flussIcebergCatalog.createTable(tablePath, td, new TestingLakeCatalogContext()); + + // Then: append a dummy data file so that the table has a snapshot. + Table icebergTable = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + appendDummyDataFile(icebergTable); + Table afterAppend = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + assertThat(afterAppend.currentSnapshot()).isNotNull(); + + // Now creating a brand-new Fluss table on top of a non-empty Iceberg table must fail. + TestingLakeCatalogContext firstFlussCreationCtx = + new TestingLakeCatalogContext() { + @Override + public boolean isCreatingFlussTable() { + return true; + } + }; + + assertThatThrownBy( + () -> flussIcebergCatalog.createTable(tablePath, td, firstFlussCreationCtx)) + .isInstanceOf(TableAlreadyExistException.class) + .hasMessageContaining("already exists in Iceberg catalog") + .hasMessageContaining("not empty"); + } + + /** + * If the existing Iceberg table has an incompatible schema, the creation should fail regardless + * of whether it is a first Fluss creation or a replay. + */ + @Test + void testCreateTableFailsWithIncompatibleLakeSchema() { + String database = "incompat_lake_db"; + String tableName = "incompat_lake_table"; + TablePath tablePath = TablePath.of(database, tableName); + + // First: create the Iceberg table with the base FLUSS_SCHEMA. + TableDescriptor tdOrig = getTableDescriptor(FLUSS_SCHEMA); + flussIcebergCatalog.createTable(tablePath, tdOrig, new TestingLakeCatalogContext()); + + // Then: attempt to create a Fluss table pointing at the same lake table, but with a + // different schema (an extra column). Should be rejected as incompatible. + Schema mismatchedSchema = + Schema.newBuilder() + .column("id", DataTypes.BIGINT()) + .column("name", DataTypes.STRING()) + .column("amount", DataTypes.INT()) + .column("address", DataTypes.STRING()) + .column("extra", DataTypes.STRING()) + .build(); + TableDescriptor tdMismatch = getTableDescriptor(mismatchedSchema); + + assertThatThrownBy( + () -> + flussIcebergCatalog.createTable( + tablePath, tdMismatch, new TestingLakeCatalogContext())) + .isInstanceOf(TableAlreadyExistException.class) + .hasMessageContaining("already exists in Iceberg catalog") + .hasMessageContaining("not compatible"); + } + + /** Direct unit tests for {@link IcebergLakeCatalog#isIcebergSchemaCompatibleWithSchema}. */ + @Test + void testIsIcebergSchemaCompatibleWithSchema() { + // baseline + org.apache.iceberg.Schema base = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "name", Types.StringType.get())); + + // identical schema -> compatible + org.apache.iceberg.Schema sameStruct = + new org.apache.iceberg.Schema( + Types.NestedField.required(11, "id", Types.LongType.get()), + Types.NestedField.optional(22, "name", Types.StringType.get())); + assertThat(flussIcebergCatalog.isIcebergSchemaCompatibleWithSchema(base, sameStruct)) + .isTrue(); + + // extra column -> incompatible + org.apache.iceberg.Schema extra = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "name", Types.StringType.get()), + Types.NestedField.optional(3, "extra", Types.StringType.get())); + assertThat(flussIcebergCatalog.isIcebergSchemaCompatibleWithSchema(base, extra)).isFalse(); + + // different name -> incompatible + org.apache.iceberg.Schema renamed = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "full_name", Types.StringType.get())); + assertThat(flussIcebergCatalog.isIcebergSchemaCompatibleWithSchema(base, renamed)) + .isFalse(); + + // different type -> incompatible + org.apache.iceberg.Schema differentType = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "name", Types.IntegerType.get())); + assertThat(flussIcebergCatalog.isIcebergSchemaCompatibleWithSchema(base, differentType)) + .isFalse(); + + // different nullability -> incompatible + org.apache.iceberg.Schema differentNullability = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get())); + assertThat( + flussIcebergCatalog.isIcebergSchemaCompatibleWithSchema( + base, differentNullability)) + .isFalse(); + } + + /** + * Simulates the real-world case where a user pre-creates the Iceberg table with a different + * bucket count (partition transform {@code bucket[N]}) than what Fluss expects, then attempts + * to bind a Fluss table on top. The mismatch on the partition transform must be detected and + * rejected. + */ + @Test + void testCreateTableFailsWithIncompatiblePartitionBucketCount() { + String database = "spec_bucket_db"; + String tableName = "spec_bucket_table"; + TablePath tablePath = TablePath.of(database, tableName); + + // Pre-create the Iceberg table directly with bucket count = 2. + Schema flussSchema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .primaryKey("id") + .build(); + TableDescriptor td2Buckets = + TableDescriptor.builder().schema(flussSchema).distributedBy(2, "id").build(); + flussIcebergCatalog.createTable(tablePath, td2Buckets, new TestingLakeCatalogContext()); + + // Now Fluss tries to (re-)create the same table with bucket count = 4. + TableDescriptor td4Buckets = + TableDescriptor.builder().schema(flussSchema).distributedBy(4, "id").build(); + + assertThatThrownBy( + () -> + flussIcebergCatalog.createTable( + tablePath, td4Buckets, new TestingLakeCatalogContext())) + .isInstanceOf(TableAlreadyExistException.class) + .hasMessageContaining("already exists in Iceberg catalog") + .hasMessageContaining("partition spec is not compatible"); + } + + /** + * When a log table has partition keys, mismatched partition columns should be rejected as + * partition spec incompatible. + */ + @Test + void testCreateTableFailsWithIncompatiblePartitionKeys() { + String database = "spec_partkeys_db"; + String tableName = "spec_partkeys_table"; + TablePath tablePath = TablePath.of(database, tableName); + + Schema flussSchema = + Schema.newBuilder() + .column("id", DataTypes.BIGINT()) + .column("name", DataTypes.STRING()) + .column("region", DataTypes.STRING()) + .column("dt", DataTypes.STRING()) + .build(); + + // First: partitioned by "region" + TableDescriptor tdRegion = + TableDescriptor.builder() + .schema(flussSchema) + .distributedBy(3) + .partitionedBy("region") + .build(); + flussIcebergCatalog.createTable(tablePath, tdRegion, new TestingLakeCatalogContext()); + + // Second: partitioned by "dt" -> mismatched + TableDescriptor tdDt = + TableDescriptor.builder() + .schema(flussSchema) + .distributedBy(3) + .partitionedBy("dt") + .build(); + assertThatThrownBy( + () -> + flussIcebergCatalog.createTable( + tablePath, tdDt, new TestingLakeCatalogContext())) + .isInstanceOf(TableAlreadyExistException.class) + .hasMessageContaining("partition spec is not compatible"); + } + + /** + * If someone alters the sort order of the existing Iceberg table (e.g. removes the fluss sort + * order), Fluss should detect the mismatch when re-binding. + */ + @Test + void testCreateTableFailsWithIncompatibleSortOrder() { + String database = "sortorder_db"; + String tableName = "sortorder_table"; + TablePath tablePath = TablePath.of(database, tableName); + + TableDescriptor td = getTableDescriptor(FLUSS_SCHEMA); + flussIcebergCatalog.createTable(tablePath, td, new TestingLakeCatalogContext()); + + // Manually replace the sort order of the existing Iceberg table. + Table iceberg = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + iceberg.replaceSortOrder().asc("id").commit(); + + assertThatThrownBy( + () -> + flussIcebergCatalog.createTable( + tablePath, td, new TestingLakeCatalogContext())) + .isInstanceOf(TableAlreadyExistException.class) + .hasMessageContaining("sort order is not compatible") + .hasMessageContaining("ASC(__offset)") + .hasMessageContaining("pre-create the Iceberg table without a custom sort order"); + } + + /** + * If someone changes a fluss-managed property on the existing Iceberg table (e.g. write mode), + * the mismatch should be surfaced. + */ + @Test + void testCreateTableFailsWithIncompatibleProperties() { + String database = "props_db"; + String tableName = "props_table"; + TablePath tablePath = TablePath.of(database, tableName); + + // pk table -> we set MERGE_MODE / UPDATE_MODE / DELETE_MODE to merge-on-read. + Schema pkSchema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .primaryKey("id") + .build(); + TableDescriptor td = + TableDescriptor.builder().schema(pkSchema).distributedBy(4, "id").build(); + flussIcebergCatalog.createTable(tablePath, td, new TestingLakeCatalogContext()); + + // Manually change a required property to an incompatible value. + Table iceberg = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, tableName)); + iceberg.updateProperties() + .set(TableProperties.MERGE_MODE, RowLevelOperationMode.COPY_ON_WRITE.modeName()) + .commit(); + + assertThatThrownBy( + () -> + flussIcebergCatalog.createTable( + tablePath, td, new TestingLakeCatalogContext())) + .isInstanceOf(TableAlreadyExistException.class) + .hasMessageContaining("table properties are not compatible"); + } + + /** White-box tests for {@link IcebergLakeCatalog#isIcebergPartitionSpecCompatible}. */ + @Test + void testIsIcebergPartitionSpecCompatible() { + // baseline schema with an "id" column (required). + org.apache.iceberg.Schema schemaA = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "name", Types.StringType.get())); + org.apache.iceberg.Schema schemaB = + new org.apache.iceberg.Schema( + Types.NestedField.required(11, "id", Types.LongType.get()), + Types.NestedField.optional(22, "name", Types.StringType.get())); + + PartitionSpec bucket4A = PartitionSpec.builderFor(schemaA).bucket("id", 4).build(); + PartitionSpec bucket4B = PartitionSpec.builderFor(schemaB).bucket("id", 4).build(); + PartitionSpec bucket2A = PartitionSpec.builderFor(schemaA).bucket("id", 2).build(); + PartitionSpec identityNameA = PartitionSpec.builderFor(schemaA).identity("name").build(); + + // Same transform + same source column, different field ids -> compatible. + assertThat( + flussIcebergCatalog.isIcebergPartitionSpecCompatible( + bucket4A, schemaA, bucket4B, schemaB)) + .isTrue(); + + // Different bucket count on same column -> incompatible. + assertThat( + flussIcebergCatalog.isIcebergPartitionSpecCompatible( + bucket4A, schemaA, bucket2A, schemaA)) + .isFalse(); + + // Different source column -> incompatible. + assertThat( + flussIcebergCatalog.isIcebergPartitionSpecCompatible( + bucket4A, schemaA, identityNameA, schemaA)) + .isFalse(); + + // Different partition field count -> incompatible. + PartitionSpec twoFields = + PartitionSpec.builderFor(schemaA).identity("name").bucket("id", 4).build(); + assertThat( + flussIcebergCatalog.isIcebergPartitionSpecCompatible( + bucket4A, schemaA, twoFields, schemaA)) + .isFalse(); + } + + /** White-box tests for {@link IcebergLakeCatalog#isIcebergSortOrderCompatible}. */ + @Test + void testIsIcebergSortOrderCompatible() { + org.apache.iceberg.Schema schemaA = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "name", Types.StringType.get())); + org.apache.iceberg.Schema schemaB = + new org.apache.iceberg.Schema( + Types.NestedField.required(11, "id", Types.LongType.get()), + Types.NestedField.optional(22, "name", Types.StringType.get())); + + SortOrder ascIdA = SortOrder.builderFor(schemaA).asc("id").build(); + SortOrder ascIdB = SortOrder.builderFor(schemaB).asc("id").build(); + SortOrder descIdA = SortOrder.builderFor(schemaA).desc("id").build(); + SortOrder ascNameA = SortOrder.builderFor(schemaA).asc("name").build(); + + // Same source + direction, different ids -> compatible. + assertThat( + flussIcebergCatalog.isIcebergSortOrderCompatible( + ascIdA, schemaA, ascIdB, schemaB)) + .isTrue(); + + // Different direction -> incompatible. + assertThat( + flussIcebergCatalog.isIcebergSortOrderCompatible( + ascIdA, schemaA, descIdA, schemaA)) + .isFalse(); + + // Different source column -> incompatible. + assertThat( + flussIcebergCatalog.isIcebergSortOrderCompatible( + ascIdA, schemaA, ascNameA, schemaA)) + .isFalse(); + + // Different sort field count -> incompatible. + SortOrder twoFields = SortOrder.builderFor(schemaA).asc("id").asc("name").build(); + assertThat( + flussIcebergCatalog.isIcebergSortOrderCompatible( + ascIdA, schemaA, twoFields, schemaA)) + .isFalse(); + } + + /** White-box tests for {@link IcebergLakeCatalog#isIcebergPropertiesCompatible}. */ + @Test + void testIsIcebergPropertiesCompatible() { + Map expected = new java.util.HashMap<>(); + expected.put("k1", "v1"); + expected.put("k2", "v2"); + + // Existing has all expected keys with same values -> compatible. + Map existing = new java.util.HashMap<>(expected); + assertThat(flussIcebergCatalog.isIcebergPropertiesCompatible(existing, expected)).isTrue(); + + // Existing also has extra keys -> still compatible (extras are ignored). + existing = new java.util.HashMap<>(expected); + existing.put("engine.internal", "whatever"); + assertThat(flussIcebergCatalog.isIcebergPropertiesCompatible(existing, expected)).isTrue(); + + // Existing missing an expected key -> incompatible. + existing = new java.util.HashMap<>(); + existing.put("k1", "v1"); + assertThat(flussIcebergCatalog.isIcebergPropertiesCompatible(existing, expected)).isFalse(); + + // Existing has key but different value -> incompatible. + existing = new java.util.HashMap<>(expected); + existing.put("k2", "different"); + assertThat(flussIcebergCatalog.isIcebergPropertiesCompatible(existing, expected)).isFalse(); + } + + private void appendDummyDataFile(Table icebergTable) { + DataFile dataFile = + DataFiles.builder(PartitionSpec.unpartitioned()) + .withPath("/tmp/fluss-fake-file.parquet") + .withFileSizeInBytes(1024) + .withRecordCount(1L) + .withFormat(FileFormat.PARQUET) + .build(); + icebergTable.newAppend().appendFile(dataFile).commit(); + } + private void createLogTable(String database, String tableName) { TableDescriptor td = getTableDescriptor(FLUSS_SCHEMA); TablePath tablePath = TablePath.of(database, tableName);