diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 9c93ada557a..b8344108e2d 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -1951,16 +1951,16 @@ public class ConfigOptions { .stringType() .noDefaultValue() .withDescription( - "Specifies the database name of the datalake table. This option is currently supported only for Paimon. " - + "If not set, the Fluss database name is used. The option may be configured before the Paimon table is created and cannot be changed after creation."); + "Specifies the database name of the datalake table. This option is currently supported only for Paimon and Iceberg. " + + "If not set, the Fluss database name is used. The option may be configured before the lake table is created and cannot be changed after creation."); public static final ConfigOption TABLE_DATALAKE_TABLE_NAME = key("table.datalake.table-name") .stringType() .noDefaultValue() .withDescription( - "Specifies the table name of the datalake table. This option is currently supported only for Paimon. " - + "If not set, the Fluss table name is used. The option may be configured before the Paimon table is created and cannot be changed after creation."); + "Specifies the table name of the datalake table. This option is currently supported only for Paimon and Iceberg. " + + "If not set, the Fluss table name is used. The option may be configured before the lake table is created and cannot be changed after creation."); public static final ConfigOption TABLE_DATALAKE_FRESHNESS = key("table.datalake.freshness") diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/LakeTableUtil.java b/fluss-common/src/main/java/org/apache/fluss/metadata/LakeTableUtil.java index c3283ee9af9..f4629a3f9f7 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metadata/LakeTableUtil.java +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/LakeTableUtil.java @@ -21,6 +21,8 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.ReadableConfig; +import java.util.Optional; + /** Utility methods for resolving external lake table metadata. */ @Internal public final class LakeTableUtil { @@ -41,17 +43,25 @@ public static TablePath resolveLakeTablePath( return TablePath.of(lakeDatabaseName, lakeTableName); } - /** Returns whether the table change affects the resolved lake table path. */ - public static boolean isLakeTablePathChange(TableChange tableChange) { + /** Returns the changed lake table path option key, if present. */ + public static Optional getLakeTablePathOptionKey(TableChange tableChange) { String optionKey; if (tableChange instanceof TableChange.SetOption) { optionKey = ((TableChange.SetOption) tableChange).getKey(); } else if (tableChange instanceof TableChange.ResetOption) { optionKey = ((TableChange.ResetOption) tableChange).getKey(); } else { - return false; + return Optional.empty(); + } + if (ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key().equals(optionKey) + || ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key().equals(optionKey)) { + return Optional.of(optionKey); } - return ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key().equals(optionKey) - || ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key().equals(optionKey); + return Optional.empty(); + } + + /** Returns whether the table change affects the resolved lake table path. */ + public static boolean isLakeTablePathChange(TableChange tableChange) { + return getLakeTablePathOptionKey(tableChange).isPresent(); } } diff --git a/fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/TestingLakeCatalogContext.java b/fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/TestingLakeCatalogContext.java index d3efd712ea6..bca7345dd91 100644 --- a/fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/TestingLakeCatalogContext.java +++ b/fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/TestingLakeCatalogContext.java @@ -18,6 +18,7 @@ package org.apache.fluss.lake.lakestorage; import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; import org.apache.fluss.security.acl.FlussPrincipal; /** A testing implementation of {@link LakeCatalog.Context}. */ @@ -25,20 +26,35 @@ public class TestingLakeCatalogContext implements LakeCatalog.Context { private final TableDescriptor currentTable; private final TableDescriptor expectedTable; + private final TablePath currentLakeTablePath; public TestingLakeCatalogContext(TableDescriptor tableDescriptor) { this(tableDescriptor, tableDescriptor); } public TestingLakeCatalogContext(TableDescriptor currentTable, TableDescriptor expectedTable) { + this(currentTable, expectedTable, null); + } + + private TestingLakeCatalogContext( + TableDescriptor currentTable, + TableDescriptor expectedTable, + TablePath currentLakeTablePath) { this.currentTable = currentTable; this.expectedTable = expectedTable; + this.currentLakeTablePath = currentLakeTablePath; } public TestingLakeCatalogContext() { this(null); } + /** Creates a testing context with the lake table path currently associated with the table. */ + public static TestingLakeCatalogContext withCurrentLakeTablePath( + TablePath currentLakeTablePath) { + return new TestingLakeCatalogContext(null, null, currentLakeTablePath); + } + @Override public boolean isCreatingFlussTable() { return false; @@ -54,6 +70,13 @@ public TableDescriptor getCurrentTable() { return currentTable; } + @Override + public TablePath getCurrentLakeTablePath() { + return currentLakeTablePath == null + ? LakeCatalog.Context.super.getCurrentLakeTablePath() + : currentLakeTablePath; + } + @Override public TableDescriptor getExpectedTable() { return expectedTable; diff --git a/fluss-common/src/test/java/org/apache/fluss/metadata/LakeTableUtilTest.java b/fluss-common/src/test/java/org/apache/fluss/metadata/LakeTableUtilTest.java index 896b46ff475..a718294183e 100644 --- a/fluss-common/src/test/java/org/apache/fluss/metadata/LakeTableUtilTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/metadata/LakeTableUtilTest.java @@ -53,6 +53,17 @@ void testIsLakeTablePathChange() { String databaseNameKey = ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key(); String tableNameKey = ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key(); + assertThat(LakeTableUtil.getLakeTablePathOptionKey(TableChange.set(databaseNameKey, "db"))) + .contains(databaseNameKey); + assertThat(LakeTableUtil.getLakeTablePathOptionKey(TableChange.reset(tableNameKey))) + .contains(tableNameKey); + assertThat( + LakeTableUtil.getLakeTablePathOptionKey( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true"))) + .isEmpty(); + assertThat(LakeTableUtil.getLakeTablePathOptionKey(TableChange.dropColumn("c1"))).isEmpty(); + assertThat(LakeTableUtil.isLakeTablePathChange(TableChange.set(databaseNameKey, "db"))) .isTrue(); assertThat(LakeTableUtil.isLakeTablePathChange(TableChange.reset(databaseNameKey))) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java index 25ceb4917ae..aeeab424435 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java @@ -139,6 +139,13 @@ public class FlinkCatalog extends AbstractCatalog { public static final String CHANGELOG_TABLE_SUFFIX = "$changelog"; public static final String BINLOG_TABLE_SUFFIX = "$binlog"; + private static final String FLUSS_CONF_PREFIX = "fluss."; + + static final String LAKE_TABLE_DATABASE_NAME_OPTION = + FLUSS_CONF_PREFIX + ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key(); + static final String LAKE_TABLE_NAME_OPTION = + FLUSS_CONF_PREFIX + ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key(); + protected final ClassLoader classLoader; protected final String catalogName; @@ -411,12 +418,11 @@ public CatalogBaseTable getTable(ObjectPath objectPath) TableInfo tableInfo; // table name contains $lake, means to read from datalake if (tableName.contains(LAKE_TABLE_SPLITTER)) { - tableInfo = - admin.getTableInfo( - TablePath.of( - objectPath.getDatabaseName(), - tableName.split("\\" + LAKE_TABLE_SPLITTER)[0])) - .get(); + TablePath flussTablePath = + TablePath.of( + objectPath.getDatabaseName(), + tableName.split("\\" + LAKE_TABLE_SPLITTER)[0]); + tableInfo = admin.getTableInfo(flussTablePath).get(); // we need to make sure the table enable datalake if (!tableInfo.getTableConfig().isDataLakeEnabled()) { throw new UnsupportedOperationException( @@ -430,11 +436,13 @@ public CatalogBaseTable getTable(ObjectPath objectPath) String lakeObjectName = resolveToLakeObjectName(lakeTablePath.getTableName(), tableName); - return getLakeTable( - lakeTablePath.getDatabaseName(), - lakeObjectName, - tableInfo.getProperties(), - getLakeCatalogProperties()); + CatalogBaseTable lakeTable = + getLakeTable( + lakeTablePath.getDatabaseName(), + lakeObjectName, + tableInfo.getProperties(), + getLakeCatalogProperties()); + return withResolvedLakeTablePath(lakeTable, flussTablePath, lakeTablePath); } else { tableInfo = admin.getTableInfo(tablePath).get(); } @@ -514,6 +522,20 @@ protected CatalogBaseTable getLakeTable( .getTable(new ObjectPath(lakeDatabaseName, lakeObjectName)); } + private static CatalogBaseTable withResolvedLakeTablePath( + CatalogBaseTable lakeTable, TablePath flussTablePath, TablePath lakeTablePath) { + if (!(lakeTable instanceof CatalogTable) || lakeTablePath.equals(flussTablePath)) { + return lakeTable; + } + + Map options = new HashMap<>(lakeTable.getOptions()); + // Some lake metadata tables do not retain the original Fluss options. Use the resolved lake + // path as the source of truth for factory identifier resolution. + options.put(LAKE_TABLE_DATABASE_NAME_OPTION, lakeTablePath.getDatabaseName()); + options.put(LAKE_TABLE_NAME_OPTION, lakeTablePath.getTableName()); + return ((CatalogTable) lakeTable).copy(options); + } + @Override public boolean tableExists(ObjectPath objectPath) throws CatalogException { // For virtual tables ($changelog, $binlog), check if the base physical table exists diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java index 29f87a45180..86eec87205c 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java @@ -64,6 +64,8 @@ import static org.apache.fluss.config.ConfigOptions.TABLE_DELETE_BEHAVIOR; import static org.apache.fluss.config.FlussConfigUtils.CLIENT_PREFIX; import static org.apache.fluss.config.FlussConfigUtils.TABLE_PREFIX; +import static org.apache.fluss.flink.catalog.FlinkCatalog.LAKE_TABLE_DATABASE_NAME_OPTION; +import static org.apache.fluss.flink.catalog.FlinkCatalog.LAKE_TABLE_NAME_OPTION; import static org.apache.fluss.flink.catalog.FlinkCatalog.LAKE_TABLE_SPLITTER; import static org.apache.fluss.flink.catalog.FlinkCatalog.resolveToLakeObjectName; import static org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils.getBucketKeyIndexes; @@ -74,16 +76,6 @@ /** Factory to create table source and table sink for Fluss. */ public class FlinkTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory { - // Custom lake paths are currently supported only for Paimon. The Paimon integration - // deliberately persists the original Fluss table options with this prefix, and lake identifier - // resolution relies on that persisted contract. Keep the prefix in sync with - // PaimonConversions.FLUSS_CONF_PREFIX. - private static final String PERSISTED_FLUSS_OPTION_PREFIX = "fluss."; - private static final String PERSISTED_FLUSS_TABLE_DATALAKE_DATABASE_NAME = - PERSISTED_FLUSS_OPTION_PREFIX + ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key(); - private static final String PERSISTED_FLUSS_TABLE_DATALAKE_TABLE_NAME = - PERSISTED_FLUSS_OPTION_PREFIX + ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key(); - protected final LakeFlinkCatalog lakeFlinkCatalog; private volatile LakeTableFactory lakeTableFactory; @@ -277,11 +269,10 @@ static ObjectIdentifier resolveToLakeIdentifier( ObjectIdentifier flinkIdentifier, Map lakeTableOptions) { String lakeDatabaseName = lakeTableOptions.getOrDefault( - PERSISTED_FLUSS_TABLE_DATALAKE_DATABASE_NAME, - flinkIdentifier.getDatabaseName()); + LAKE_TABLE_DATABASE_NAME_OPTION, flinkIdentifier.getDatabaseName()); String lakeObjectName = resolveToLakeObjectName( - lakeTableOptions.get(PERSISTED_FLUSS_TABLE_DATALAKE_TABLE_NAME), + lakeTableOptions.get(LAKE_TABLE_NAME_OPTION), flinkIdentifier.getObjectName()); return ObjectIdentifier.of( flinkIdentifier.getCatalogName(), lakeDatabaseName, lakeObjectName); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java index a5e348c7af7..09487390827 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java @@ -382,7 +382,12 @@ void testGetLakeTableWithCustomLakePath() throws Exception { lakeTable); CatalogBaseTable gottenLakeTable = catalog.getTable(new ObjectPath(DEFAULT_DB, flussTableName + "$lake")); - assertThat(gottenLakeTable).isEqualTo(lakeTable); + assertThat(gottenLakeTable.getOptions()) + .containsAllEntriesOf(lakeTable.getOptions()) + .containsEntry( + FlinkCatalog.LAKE_TABLE_DATABASE_NAME_OPTION, + lakeTablePath.getDatabaseName()) + .containsEntry(FlinkCatalog.LAKE_TABLE_NAME_OPTION, lakeTablePath.getTableName()); CatalogTable snapshotsTable = newCatalogTable(Collections.emptyMap()); mockLakeCatalog.registerLakeTable( @@ -392,7 +397,12 @@ void testGetLakeTableWithCustomLakePath() throws Exception { snapshotsTable); CatalogBaseTable gottenSnapshotsTable = catalog.getTable(new ObjectPath(DEFAULT_DB, flussTableName + "$lake$snapshots")); - assertThat(gottenSnapshotsTable).isEqualTo(snapshotsTable); + assertThat(gottenSnapshotsTable.getOptions()) + .containsAllEntriesOf(snapshotsTable.getOptions()) + .containsEntry( + FlinkCatalog.LAKE_TABLE_DATABASE_NAME_OPTION, + lakeTablePath.getDatabaseName()) + .containsEntry(FlinkCatalog.LAKE_TABLE_NAME_OPTION, lakeTablePath.getTableName()); } @Test 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 70886c9c839..ff118c3d1a7 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 @@ -27,6 +27,7 @@ import org.apache.fluss.lake.iceberg.utils.IcebergPartitionSpecUtils; import org.apache.fluss.lake.iceberg.utils.IcebergUtils; import org.apache.fluss.lake.lakestorage.LakeCatalog; +import org.apache.fluss.metadata.LakeTableUtil; import org.apache.fluss.metadata.TableChange; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; @@ -60,6 +61,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import static org.apache.fluss.lake.iceberg.IcebergSchemaUtils.LEGACY_SYSTEM_COLUMNS; @@ -100,6 +102,7 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co keys.size() <= 1, "Iceberg format supports at most one bucket key, but got: %s", keys); + validateLakeTablePath(tablePath, context); // convert Fluss table path to iceberg table boolean isPkTable = tableDescriptor.hasPrimaryKey(); @@ -164,6 +167,7 @@ public void alterTable(TablePath tablePath, List tableChanges, Cont if (change instanceof TableChange.SchemaChange) { schemaChanges.add(change); } else { + rejectLakeTablePathChange(change); propertyChanges.add(change); } } @@ -308,6 +312,45 @@ private TableIdentifier toIcebergTableIdentifier(TablePath tablePath) { return TableIdentifier.of(tablePath.getDatabaseName(), tablePath.getTableName()); } + private static void rejectLakeTablePathChange(TableChange tableChange) { + Optional optionKey = LakeTableUtil.getLakeTablePathOptionKey(tableChange); + if (!optionKey.isPresent()) { + return; + } + throw new InvalidAlterTableException( + String.format( + "Cannot alter lake table path option '%s' after the Iceberg table has been " + + "created.", + optionKey.get())); + } + + /** + * Prevents an existing Iceberg table from being rebound to a different physical path. + * + *

For a disabled Fluss table, the current path may only be a candidate resolved from its + * options. When the requested path differs, the current Iceberg table must exist before the + * mapping is treated as immutable. This still allows a custom path to be configured when lake + * storage is enabled for the first time. + */ + private void validateLakeTablePath(TablePath targetLakeTablePath, Context context) { + TablePath currentLakeTablePath = context.getCurrentLakeTablePath(); + if (context.isCreatingFlussTable() || currentLakeTablePath == null) { + return; + } + if (currentLakeTablePath.equals(targetLakeTablePath)) { + // Re-enable the existing mapping without querying Iceberg. + return; + } + if (icebergCatalog.tableExists(toIcebergTableIdentifier(currentLakeTablePath))) { + // An existing table at the current path proves that the mapping has already been used. + throw new InvalidAlterTableException( + String.format( + "The Iceberg table path can only be altered before the Iceberg table " + + "is created. Current path: %s, target path: %s.", + currentLakeTablePath, targetLakeTablePath)); + } + } + private void createTable( TablePath tablePath, Catalog.TableBuilder tableBuilder, diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeTieringFactory.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeTieringFactory.java index 367e66bd285..20313ce406d 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeTieringFactory.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeTieringFactory.java @@ -58,7 +58,7 @@ public LakeWriter createLakeWriter(WriterInitContext writerI public void validateTable(TableInfo tableInfo) throws IOException { Catalog icebergCatalog = icebergCatalogProvider.get(); try { - Table icebergTable = icebergCatalog.loadTable(toIceberg(tableInfo.getTablePath())); + Table icebergTable = icebergCatalog.loadTable(toIceberg(tableInfo.getLakeTablePath())); IcebergPartitionSpecValidator.validate(icebergTable, tableInfo); } finally { if (icebergCatalog instanceof AutoCloseable) { @@ -76,7 +76,8 @@ public SimpleVersionedSerializer getWriteResultSerializer() @Override public LakeCommitter createLakeCommitter( CommitterInitContext committerInitContext) throws IOException { - return new IcebergLakeCommitter(icebergCatalogProvider, committerInitContext.tablePath()); + return new IcebergLakeCommitter( + icebergCatalogProvider, committerInitContext.tableInfo().getLakeTablePath()); } @Override diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java index 646bdddc456..5ab0eedad10 100644 --- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java +++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeWriter.java @@ -69,7 +69,7 @@ public IcebergLakeWriter( IcebergCatalogProvider icebergCatalogProvider, WriterInitContext writerInitContext) throws IOException { this.icebergCatalog = icebergCatalogProvider.get(); - this.icebergTable = getTable(writerInitContext.tablePath()); + this.icebergTable = getTable(writerInitContext.tableInfo().getLakeTablePath()); IcebergPartitionSpecValidator.validate(icebergTable, writerInitContext.tableInfo()); // Create a record writer 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 868a6bfd47b..8f3acf42120 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 @@ -807,6 +807,77 @@ void testAlterTableAddExistingColumns() { tablePath, changes, getLakeCatalogContext(FLUSS_SCHEMA, changes)); } + @Test + void testAlterRejectsNameMappingOptionAfterLakeTableCreated() { + String database = "test_reject_name_mapping_db"; + String tableName = "test_reject_name_mapping_table"; + TablePath tablePath = TablePath.of(database, tableName); + createLogTable(database, tableName); + + assertThatThrownBy( + () -> + flussIcebergCatalog.alterTable( + tablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_DATABASE_NAME + .key(), + "another_db")), + new TestingLakeCatalogContext())) + .isInstanceOf(InvalidAlterTableException.class) + .hasMessageContaining(ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key()) + .hasMessageContaining("after the Iceberg table has been created"); + + assertThatThrownBy( + () -> + flussIcebergCatalog.alterTable( + tablePath, + Collections.singletonList( + TableChange.reset( + ConfigOptions.TABLE_DATALAKE_TABLE_NAME + .key())), + new TestingLakeCatalogContext())) + .isInstanceOf(InvalidAlterTableException.class) + .hasMessageContaining(ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key()); + } + + @Test + void testCreateTableRejectsRebindAfterLakeTableCreated() { + String database = "test_rebind_db"; + String currentTableName = "current_lake_table"; + createLogTable(database, currentTableName); + + TablePath targetPath = TablePath.of(database, "target_lake_table"); + assertThatThrownBy( + () -> + flussIcebergCatalog.createTable( + targetPath, + getTableDescriptor(FLUSS_SCHEMA), + TestingLakeCatalogContext.withCurrentLakeTablePath( + TablePath.of(database, currentTableName)))) + .isInstanceOf(InvalidAlterTableException.class) + .hasMessageContaining("can only be altered before the Iceberg table is created"); + } + + @Test + void testCreateTableAllowsCustomLakePathWhenLakeTableNotCreated() { + String database = "test_custom_path_db"; + TablePath targetPath = TablePath.of(database, "target_lake_table"); + // The current mapping still points at a path that has no physical Iceberg table yet, so + // configuring a custom path while enabling lake storage for the first time is allowed. + flussIcebergCatalog.createTable( + targetPath, + getTableDescriptor(FLUSS_SCHEMA), + TestingLakeCatalogContext.withCurrentLakeTablePath( + TablePath.of(database, "never_created_table"))); + + Table created = + flussIcebergCatalog + .getIcebergCatalog() + .loadTable(TableIdentifier.of(database, "target_lake_table")); + assertThat(created).isNotNull(); + } + @Test void testAlterTableAddColumnWhenIcebergSchemaNotMatch() { String database = "test_alter_add_col_schema_mismatch_db"; diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadPrimaryKeyTableITCase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadPrimaryKeyTableITCase.java index 6cc48ede0db..47ec198e735 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadPrimaryKeyTableITCase.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadPrimaryKeyTableITCase.java @@ -40,6 +40,8 @@ import org.apache.flink.types.RowKind; import org.apache.flink.util.CloseableIterator; import org.apache.flink.util.CollectionUtil; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -52,11 +54,15 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsExactOrder; import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertRowResultsIgnoreOrder; +import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsWithTimeout; +import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg; import static org.apache.fluss.testutils.DataTestUtils.row; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test case for union read primary key table. */ public class FlinkUnionReadPrimaryKeyTableITCase extends FlinkUnionReadTestBase { @@ -358,6 +364,72 @@ void testReadIcebergLakeTable(boolean isPartitioned) throws Exception { jobClient.cancel().get(); } + @Test + void testUnionReadWithCustomLakeTablePath() throws Exception { + String tableName = "pk_table_custom_lake_mapping"; + TablePath tablePath = TablePath.of(DEFAULT_DB, tableName); + TablePath lakeTablePath = TablePath.of("custom_db", "pk_table_custom_lake_target"); + + Schema schema = + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .primaryKey("a") + .build(); + TableDescriptor descriptor = + TableDescriptor.builder() + .schema(schema) + .distributedBy(DEFAULT_BUCKET_NUM, "a") + .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true") + .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500)) + .property( + ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key(), + lakeTablePath.getDatabaseName()) + .property( + ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key(), + lakeTablePath.getTableName()) + .build(); + long tableId = createTable(tablePath, descriptor); + TableBucket tableBucket = new TableBucket(tableId, 0); + + writeRows(tablePath, Arrays.asList(row(1, "v1"), row(2, "v2")), false); + + JobClient jobClient = buildTieringJob(execEnv); + // wait until the two rows are tiered to the mapped Iceberg table + assertReplicaStatus(tableBucket, 2); + + // the physical Iceberg table is created at the mapped path, not the Fluss table path + icebergCatalog.loadTable(toIceberg(lakeTablePath)); + assertThatThrownBy(() -> icebergCatalog.loadTable(toIceberg(tablePath))) + .isInstanceOf(NoSuchTableException.class); + + // union read merges the Iceberg snapshot with the un-tiered Fluss changelog + try (CloseableIterator unionRows = + streamTEnv.executeSql("select a, b from " + tableName).collect()) { + assertRowResultsIgnoreOrder( + unionRows, Arrays.asList(Row.of(1, "v1"), Row.of(2, "v2")), false); + jobClient.cancel().get(1, TimeUnit.MINUTES); + writeRows(tablePath, Collections.singletonList(row(3, "v3")), false); + assertRowResultsIgnoreOrder( + unionRows, Collections.singletonList(Row.of(3, "v3")), false); + + // read the Iceberg lake table directly via $lake + try (CloseableIterator lakeRows = + batchTEnv.executeSql("select a, b from " + tableName + "$lake").collect()) { + assertRowResultsIgnoreOrder( + lakeRows, Arrays.asList(Row.of(1, "v1"), Row.of(2, "v2")), true); + } + + // read the Iceberg metadata table via $lake$snapshots + try (CloseableIterator snapshotRows = + batchTEnv + .executeSql("select * from " + tableName + "$lake$snapshots") + .collect()) { + assertThat(collectRowsWithTimeout(snapshotRows, 1)).hasSize(1); + } + } + } + private void writeFullTypeRow(TablePath tablePath, String partition) throws Exception { GenericRow nestedRow = new GenericRow(2); nestedRow.setField(0, 3); diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java index c3ecc0af390..c746eb63517 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java @@ -47,6 +47,7 @@ import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.data.IcebergGenerics; import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.io.CloseableIterator; import org.apache.iceberg.types.Types; import org.junit.jupiter.api.BeforeEach; @@ -235,6 +236,60 @@ void testEmptyCommitCreatesSnapshot() throws Exception { } } + @Test + void testTieringWritesToMappedLakeTablePath() throws Exception { + TablePath flussTablePath = TablePath.of("iceberg", "fluss_logical_tiering_table"); + TablePath lakeTablePath = TablePath.of("iceberg_lake_db", "iceberg_physical_tiering_table"); + + // only the mapped physical Iceberg table exists + createTable(lakeTablePath, false, false); + + TableInfo tableInfo = createTableInfoWithLakePath(flussTablePath, lakeTablePath); + + // the validator must resolve and load the mapped physical table + icebergLakeTieringFactory.validateTable(tableInfo); + + Table lakeTable = icebergCatalog.loadTable(toIceberg(lakeTablePath)); + SimpleVersionedSerializer writeResultSerializer = + icebergLakeTieringFactory.getWriteResultSerializer(); + + List writeResults = new ArrayList<>(); + Map> recordsByBucket = new HashMap<>(); + for (int bucket = 0; bucket < BUCKET_NUM; bucket++) { + try (LakeWriter writer = + createLakeWriter(flussTablePath, bucket, null, null, tableInfo)) { + Tuple2, List> writeAndExpect = + genLogTableRecords(null, bucket, 10); + recordsByBucket.put(bucket, writeAndExpect.f1); + for (LogRecord record : writeAndExpect.f0) { + writer.write(record); + } + IcebergWriteResult result = writer.complete(); + byte[] serialized = writeResultSerializer.serialize(result); + writeResults.add( + writeResultSerializer.deserialize( + writeResultSerializer.getVersion(), serialized)); + } + } + + try (LakeCommitter lakeCommitter = + createLakeCommitter(flussTablePath, tableInfo)) { + IcebergCommittable committable = lakeCommitter.toCommittable(writeResults); + lakeCommitter.commit(committable, Collections.singletonMap("k1", "v1")); + } + + lakeTable.refresh(); + assertThat(lakeTable.currentSnapshot()).isNotNull(); + for (int bucket = 0; bucket < BUCKET_NUM; bucket++) { + CloseableIterator actualRecords = getIcebergRows(lakeTable, null, bucket); + verifyTableRecords(actualRecords, recordsByBucket.get(bucket), bucket, null); + } + + // the Fluss logical path must never be materialized as a physical Iceberg table + assertThatThrownBy(() -> icebergCatalog.loadTable(toIceberg(flussTablePath))) + .isInstanceOf(NoSuchTableException.class); + } + @Test void testRejectIncompatiblePartitionSpec() { TablePath tablePath = TablePath.of("iceberg", "test_incompatible_partition_spec"); @@ -287,6 +342,29 @@ private TableInfo createTableInfo( tablePath, 0, 1, descriptorBuilder.build(), DEFAULT_REMOTE_DATA_DIR, 1L, 1L); } + private TableInfo createTableInfoWithLakePath( + TablePath flussTablePath, TablePath lakeTablePath) { + Schema schema = + Schema.newBuilder() + .column("c1", DataTypes.INT()) + .column("c2", DataTypes.STRING()) + .column("c3", DataTypes.STRING()) + .build(); + TableDescriptor descriptor = + TableDescriptor.builder() + .schema(schema) + .distributedBy(BUCKET_NUM) + .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true) + .property( + ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key(), + lakeTablePath.getDatabaseName()) + .property( + ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key(), + lakeTablePath.getTableName()) + .build(); + return TableInfo.of(flussTablePath, 0, 1, descriptor, DEFAULT_REMOTE_DATA_DIR, 1L, 1L); + } + private LakeWriter createLakeWriter( TablePath tablePath, int bucket, diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java index de40fbf01a8..3354999106c 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java @@ -253,9 +253,9 @@ private static void checkCustomLakePathSupported( tableConf .getOptional(ConfigOptions.TABLE_DATALAKE_FORMAT) .orElse(clusterDataLakeFormat); - if (dataLakeFormat != DataLakeFormat.PAIMON) { + if (dataLakeFormat != DataLakeFormat.PAIMON && dataLakeFormat != DataLakeFormat.ICEBERG) { throw new InvalidConfigException( - "Custom lake table path is only supported for Paimon."); + "Custom lake table path is only supported for Paimon and Iceberg."); } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java index 76834d342ed..052f9658bc7 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java @@ -153,6 +153,19 @@ void testCreateTableWithKvTTLMillisOverflowFails() { .hasMessageContaining("exceeds"); } + @Test + void testCustomLakePathRejectedForUnsupportedFormat() { + assertThatThrownBy( + () -> + validate( + tableDescriptorWithLakeName( + ConfigOptions.TABLE_DATALAKE_DATABASE_NAME, + "lance_db"), + DataLakeFormat.LANCE)) + .isInstanceOf(InvalidConfigException.class) + .hasMessageContaining("only supported for Paimon and Iceberg"); + } + @Test void testKvFormatVersionStillRejectsValuesAboveVersionTwo() { TableDescriptor descriptor = @@ -265,16 +278,15 @@ void testCustomLakePathValidation() { assertThatCode(() -> validate(tableDescriptor, DataLakeFormat.PAIMON)) .doesNotThrowAnyException(); - // custom lake paths are not supported for Iceberg - assertThatThrownBy( + // custom lake paths are now supported for Iceberg + assertThatCode( () -> validate( tableDescriptorWithLakeName( ConfigOptions.TABLE_DATALAKE_TABLE_NAME, "lake_table"), DataLakeFormat.ICEBERG)) - .isInstanceOf(InvalidConfigException.class) - .hasMessageContaining("Custom lake table path is only supported for Paimon"); + .doesNotThrowAnyException(); } private static Stream supportedKvTTLTimeColumnTypes() {