[lake/iceberg] Support custom Iceberg lake table path

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<String> 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<Duration> TABLE_DATALAKE_FRESHNESS =
             key("table.datalake.freshness")

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<String> 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();
     }
 }

fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/TestingLakeCatalogContext.java

@@ -18,27 +18,43 @@
 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}. */
 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;

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)))

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<String, String> 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

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<String, String> 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);

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

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<TableChange> 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<String> 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.
+     *
+     * <p>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,

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeTieringFactory.java

@@ -58,7 +58,7 @@ public LakeWriter<IcebergWriteResult> 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<IcebergWriteResult> getWriteResultSerializer()
     @Override
     public LakeCommitter<IcebergWriteResult, IcebergCommittable> createLakeCommitter(
             CommitterInitContext committerInitContext) throws IOException {
-        return new IcebergLakeCommitter(icebergCatalogProvider, committerInitContext.tablePath());
+        return new IcebergLakeCommitter(
+                icebergCatalogProvider, committerInitContext.tableInfo().getLakeTablePath());
     }
 
     @Override

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