[lake/paimon] Support custom Paimon lake table path

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1646,6 +1646,20 @@ public class ConfigOptions {
                                     + "The `table.datalake.format` can be pre-defined before enabling `table.datalake.enabled`. This allows the data lake feature to be dynamically enabled on the table without requiring table recreation. "
                                     + "If `table.datalake.format` is not explicitly set during table creation, the table will default to the format specified by the `datalake.format` configuration in the Fluss cluster.");
 
+    public static final ConfigOption<String> TABLE_DATALAKE_DATABASE_NAME =
+            key("table.datalake.database-name")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The database name of the datalake table. If not set, it will be the same as the Fluss database.");
+
+    public static final ConfigOption<String> TABLE_DATALAKE_TABLE_NAME =
+            key("table.datalake.table-name")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The table name of the datalake table. If not set, it will be the same as the Fluss table.");
+
     public static final ConfigOption<Duration> TABLE_DATALAKE_FRESHNESS =
             key("table.datalake.freshness")
                     .durationType()

fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java

@@ -47,6 +47,8 @@ public class FlussConfigUtils {
         ALTERABLE_TABLE_OPTIONS =
                 Arrays.asList(
                         ConfigOptions.TABLE_DATALAKE_ENABLED.key(),
+                        ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key(),
+                        ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key(),
                         ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(),
                         ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(),
                         ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS.key(),

fluss-common/src/main/java/org/apache/fluss/metadata/LakeTableUtil.java

@@ -0,0 +1,101 @@
+/*
+ * 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.metadata;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+
+import java.util.Map;
+import java.util.Optional;
+
+/** Utility methods for resolving external lake table metadata. */
+@Internal
+public final class LakeTableUtil {
+
+    private LakeTableUtil() {}
+
+    /** Returns the table path used to access the external datalake table. */
+    public static TablePath getLakeTablePath(
+            TablePath flussTablePath, Map<String, String> tableProperties) {
+        return getLakeTablePath(flussTablePath, Configuration.fromMap(tableProperties));
+    }
+
+    /** Returns the table path used to access the external datalake table. */
+    public static TablePath getLakeTablePath(TablePath flussTablePath, Configuration tableConf) {
+        String lakeDatabaseName =
+                getDataLakeDatabaseName(tableConf).orElse(flussTablePath.getDatabaseName());
+        String lakeTableName =
+                getDataLakeTableName(tableConf).orElse(flussTablePath.getTableName());
+        return TablePath.of(lakeDatabaseName, lakeTableName);
+    }
+
+    /** Returns whether the table has explicit custom datalake path options. */
+    public static boolean hasCustomLakePath(Map<String, String> tableProperties) {
+        return hasCustomLakePath(Configuration.fromMap(tableProperties));
+    }
+
+    /** Returns whether the table has explicit custom datalake path options. */
+    public static boolean hasCustomLakePath(Configuration tableConf) {
+        return getDataLakeDatabaseName(tableConf).isPresent()
+                || getDataLakeTableName(tableConf).isPresent();
+    }
+
+    /** Returns the lake table name with the metadata table suffix from the requested table name. */
+    public static String getLakeTableName(String lakeTableName, String requestedTableName) {
+        if (lakeTableName == null) {
+            return requestedTableName;
+        }
+
+        int metadataTableIndex = requestedTableName.indexOf('$');
+        if (metadataTableIndex < 0) {
+            return lakeTableName;
+        }
+
+        String metadataTableSuffix = requestedTableName.substring(metadataTableIndex);
+        if (lakeTableName.endsWith(metadataTableSuffix)) {
+            return lakeTableName;
+        }
+        return lakeTableName + metadataTableSuffix;
+    }
+
+    /**
+     * Returns the lake table name with the metadata table suffix from a table name containing the
+     * given lake table splitter.
+     */
+    public static String getLakeTableName(
+            String lakeTableName, String requestedTableName, String lakeTableSplitter) {
+        int splitterIndex = requestedTableName.indexOf(lakeTableSplitter);
+        if (splitterIndex < 0) {
+            return requestedTableName;
+        }
+
+        String requestedLakeTableName =
+                requestedTableName.substring(0, splitterIndex)
+                        + requestedTableName.substring(splitterIndex + lakeTableSplitter.length());
+        return getLakeTableName(lakeTableName, requestedLakeTableName);
+    }
+
+    private static Optional<String> getDataLakeDatabaseName(Configuration tableConf) {
+        return tableConf.getOptional(ConfigOptions.TABLE_DATALAKE_DATABASE_NAME);
+    }
+
+    private static Optional<String> getDataLakeTableName(Configuration tableConf) {
+        return tableConf.getOptional(ConfigOptions.TABLE_DATALAKE_TABLE_NAME);
+    }
+}

fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java

@@ -112,6 +112,22 @@ public TablePath getTablePath() {
         return tablePath;
     }
 
+    /**
+     * Returns the table path used to access the external datalake table.
+     *
+     * <p>The returned path is derived from the Fluss table path and table-level datalake path
+     * options. If no custom datalake path option is configured, this returns the same path as
+     * {@link #getTablePath()}.
+     */
+    public TablePath getLakeTablePath() {
+        return LakeTableUtil.getLakeTablePath(tablePath, properties);
+    }
+
+    /** Returns whether the table has explicit custom datalake path options. */
+    public boolean hasCustomLakePath() {
+        return LakeTableUtil.hasCustomLakePath(properties);
+    }
+
     /**
      * Returns the unique identifier for the table within the cluster.
      *

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

@@ -34,6 +34,7 @@
 import org.apache.fluss.flink.utils.FlinkConversions;
 import org.apache.fluss.metadata.DatabaseChange;
 import org.apache.fluss.metadata.DatabaseDescriptor;
+import org.apache.fluss.metadata.LakeTableUtil;
 import org.apache.fluss.metadata.PartitionInfo;
 import org.apache.fluss.metadata.PartitionSpec;
 import org.apache.fluss.metadata.TableChange;
@@ -404,9 +405,10 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
                                             objectPath.getDatabaseName(),
                                             tableName.split("\\" + LAKE_TABLE_SPLITTER)[0])));
                 }
+                TablePath lakeTablePath = tableInfo.getLakeTablePath();
 
                 return getLakeTable(
-                        objectPath.getDatabaseName(),
+                        lakeTablePath,
                         tableName,
                         tableInfo.getProperties(),
                         getLakeCatalogProperties());
@@ -455,23 +457,19 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
     }
 
     protected CatalogBaseTable getLakeTable(
-            String databaseName,
-            String tableName,
+            TablePath lakeTablePath,
+            String requestedTableName,
             Configuration properties,
             Map<String, String> lakeCatalogProperties)
             throws TableNotExistException, CatalogException {
-        String[] tableComponents = tableName.split("\\" + LAKE_TABLE_SPLITTER);
-        if (tableComponents.length == 1) {
-            // should be pattern like table_name$lake
-            tableName = tableComponents[0];
-        } else {
-            // pattern is table_name$lake$snapshots
-            // Need to reconstruct: table_name + $snapshots
-            tableName = String.join("", tableComponents);
-        }
-        return lakeFlinkCatalog
-                .getLakeCatalog(properties, lakeCatalogProperties)
-                .getTable(new ObjectPath(databaseName, tableName));
+        String lakeObjectName =
+                LakeTableUtil.getLakeTableName(
+                        lakeTablePath.getTableName(), requestedTableName, LAKE_TABLE_SPLITTER);
+        CatalogBaseTable lakeTable =
+                lakeFlinkCatalog
+                        .getLakeCatalog(properties, lakeCatalogProperties)
+                        .getTable(new ObjectPath(lakeTablePath.getDatabaseName(), lakeObjectName));
+        return lakeTable;
     }
 
     @Override

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeTableFactory.java

@@ -17,7 +17,9 @@
 
 package org.apache.fluss.flink.lake;
 
+import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.LakeTableUtil;
 
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.connector.source.DynamicTableSource;
@@ -26,23 +28,30 @@
 import org.apache.flink.table.factories.FactoryUtil;
 
 import java.util.Collections;
+import java.util.Map;
 
 /** A factory to create {@link DynamicTableSource} for lake table. */
 public class LakeTableFactory {
+
+    private static final String FLUSS_CONF_PREFIX = "fluss.";
+    private static final String FLUSS_TABLE_DATALAKE_DATABASE_NAME =
+            FLUSS_CONF_PREFIX + ConfigOptions.TABLE_DATALAKE_DATABASE_NAME.key();
+    private static final String FLUSS_TABLE_DATALAKE_TABLE_NAME =
+            FLUSS_CONF_PREFIX + ConfigOptions.TABLE_DATALAKE_TABLE_NAME.key();
+
     private final LakeFlinkCatalog lakeFlinkCatalog;
 
     public LakeTableFactory(LakeFlinkCatalog lakeFlinkCatalog) {
         this.lakeFlinkCatalog = lakeFlinkCatalog;
     }
 
     public DynamicTableSource createDynamicTableSource(
-            DynamicTableFactory.Context context, String tableName) {
-        ObjectIdentifier originIdentifier = context.getObjectIdentifier();
+            DynamicTableFactory.Context context, String requestedTableName) {
         ObjectIdentifier lakeIdentifier =
-                ObjectIdentifier.of(
-                        originIdentifier.getCatalogName(),
-                        originIdentifier.getDatabaseName(),
-                        tableName);
+                toLakeIdentifier(
+                        context.getObjectIdentifier(),
+                        context.getCatalogTable().getOptions(),
+                        requestedTableName);
 
         // For Iceberg and Paimon, pass the table name as-is to their factory.
         // Metadata tables will be handled internally by their respective factories.
@@ -60,6 +69,22 @@ public DynamicTableSource createDynamicTableSource(
         return factory.createDynamicTableSource(newContext);
     }
 
+    static ObjectIdentifier toLakeIdentifier(
+            ObjectIdentifier originIdentifier,
+            Map<String, String> lakeTableOptions,
+            String requestedTableName) {
+        String lakeDatabaseName =
+                lakeTableOptions.getOrDefault(
+                        FLUSS_TABLE_DATALAKE_DATABASE_NAME, originIdentifier.getDatabaseName());
+        String lakeObjectName =
+                LakeTableUtil.getLakeTableName(
+                        lakeTableOptions.getOrDefault(
+                                FLUSS_TABLE_DATALAKE_TABLE_NAME, requestedTableName),
+                        requestedTableName);
+        return ObjectIdentifier.of(
+                originIdentifier.getCatalogName(), lakeDatabaseName, lakeObjectName);
+    }
+
     private DynamicTableSourceFactory getLakeTableFactory() {
         switch (lakeFlinkCatalog.getLakeFormat()) {
             case PAIMON:

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/LakeSourceUtils.java

@@ -24,6 +24,7 @@
 import org.apache.fluss.lake.lakestorage.LakeStoragePluginSetUp;
 import org.apache.fluss.lake.source.LakeSource;
 import org.apache.fluss.lake.source.LakeSplit;
+import org.apache.fluss.metadata.LakeTableUtil;
 import org.apache.fluss.metadata.TablePath;
 
 import org.slf4j.Logger;
@@ -67,8 +68,9 @@ public static LakeSource<LakeSplit> createLakeSource(
                             dataLake, dataLake.toLowerCase()));
         }
         LakeStorage lakeStorage = checkNotNull(lakeStoragePlugin).createLakeStorage(lakeConfig);
+        TablePath lakeTablePath = LakeTableUtil.getLakeTablePath(tablePath, properties);
         try {
-            return (LakeSource<LakeSplit>) lakeStorage.createLakeSource(tablePath);
+            return (LakeSource<LakeSplit>) lakeStorage.createLakeSource(lakeTablePath);
         } catch (UnsupportedOperationException e) {
             throw new UnsupportedOperationException(
                     String.format(

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java

@@ -25,6 +25,7 @@
 import org.apache.fluss.flink.adapter.ResolvedCatalogMaterializedTableAdapter;
 import org.apache.fluss.flink.lake.LakeFlinkCatalog;
 import org.apache.fluss.flink.utils.FlinkConversionsTest;
+import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.server.testutils.FlussClusterExtension;
 import org.apache.fluss.utils.ExceptionUtils;
 
@@ -74,8 +75,10 @@
 import java.util.Optional;
 
 import static org.apache.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS;
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_DATABASE_NAME;
 import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
 import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_TABLE_NAME;
 import static org.apache.fluss.flink.FlinkConnectorOptions.BUCKET_KEY;
 import static org.apache.fluss.flink.FlinkConnectorOptions.BUCKET_NUMBER;
 import static org.apache.fluss.flink.FlinkConnectorOptions.SCAN_STARTUP_MODE;
@@ -344,6 +347,43 @@ void testCreateAlreadyExistsLakeTable() throws Exception {
         catalog.createTable(lakeTablePath, table, false);
     }
 
+    @Test
+    void testGetLakeTableWithCustomLakePath() throws Exception {
+        String flussTableName = "fluss_table";
+        TablePath lakeTablePath = TablePath.of("custom_lake_db", "custom_lake_table");
+
+        Map<String, String> options = new HashMap<>();
+        options.put(TABLE_DATALAKE_ENABLED.key(), "true");
+        options.put(TABLE_DATALAKE_FORMAT.key(), PAIMON.name());
+        options.put(TABLE_DATALAKE_DATABASE_NAME.key(), lakeTablePath.getDatabaseName());
+        options.put(TABLE_DATALAKE_TABLE_NAME.key(), lakeTablePath.getTableName());
+
+        ObjectPath flussTablePath = new ObjectPath(DEFAULT_DB, flussTableName);
+        catalog.createTable(flussTablePath, newCatalogTable(options), false);
+
+        CatalogTable lakeTable = newCatalogTable(Collections.emptyMap());
+        mockLakeCatalog.catalog.createDatabase(
+                lakeTablePath.getDatabaseName(),
+                new CatalogDatabaseImpl(Collections.emptyMap(), null),
+                true);
+        mockLakeCatalog.registerLakeTable(
+                new ObjectPath(lakeTablePath.getDatabaseName(), lakeTablePath.getTableName()),
+                lakeTable);
+        CatalogBaseTable gottenLakeTable =
+                catalog.getTable(new ObjectPath(DEFAULT_DB, flussTableName + "$lake"));
+        assertThat(gottenLakeTable).isEqualTo(lakeTable);
+
+        CatalogTable snapshotsTable = newCatalogTable(Collections.emptyMap());
+        mockLakeCatalog.registerLakeTable(
+                new ObjectPath(
+                        lakeTablePath.getDatabaseName(),
+                        lakeTablePath.getTableName() + "$snapshots"),
+                snapshotsTable);
+        CatalogBaseTable gottenSnapshotsTable =
+                catalog.getTable(new ObjectPath(DEFAULT_DB, flussTableName + "$lake$snapshots"));
+        assertThat(gottenSnapshotsTable).isEqualTo(snapshotsTable);
+    }
+
     @Test
     void testCreateTableWithBucket() throws Exception {
         // for pk table;

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java

@@ -64,6 +64,7 @@ public class PaimonLakeCommitter implements LakeCommitter<PaimonWriteResult, Pai
     private final Catalog paimonCatalog;
     private final FileStoreTable fileStoreTable;
     private final TablePath tablePath;
+    private final TablePath lakeTablePath;
     private final long tableId;
     private final Configuration flussClientConfig;
     private TableCommitImpl tableCommit;
@@ -75,11 +76,12 @@ public PaimonLakeCommitter(
             throws IOException {
         this.paimonCatalog = paimonCatalogProvider.get();
         this.tablePath = committerInitContext.tablePath();
+        this.lakeTablePath = committerInitContext.tableInfo().getLakeTablePath();
         this.tableId = committerInitContext.tableInfo().getTableId();
         this.flussClientConfig = committerInitContext.flussClientConfig();
         this.fileStoreTable =
                 getTable(
-                        committerInitContext.tablePath(),
+                        lakeTablePath,
                         committerInitContext
                                         .tableInfo()
                                         .getTableConfig()
@@ -161,7 +163,7 @@ public LakeCommitResult commit(
     @Nullable
     private TieringStats computeTableStats() {
         Identifier identifier =
-                new Identifier(tablePath.getDatabaseName(), tablePath.getTableName());
+                new Identifier(lakeTablePath.getDatabaseName(), lakeTablePath.getTableName());
         try {
             Optional<TableSnapshot> snapshot = paimonCatalog.loadSnapshot(identifier);
             if (!snapshot.isPresent()) {