[flink] Optimize the getLakeTableFactory in LakeTableFactory (#1816)

Author: luoyuxia

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

@@ -23,7 +23,7 @@
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.InvalidTableException;
-import org.apache.fluss.flink.lake.LakeCatalog;
+import org.apache.fluss.flink.lake.LakeFlinkCatalog;
 import org.apache.fluss.flink.procedure.ProcedureManager;
 import org.apache.fluss.flink.utils.CatalogExceptionUtils;
 import org.apache.fluss.flink.utils.FlinkConversions;
@@ -114,7 +114,7 @@ public class FlinkCatalog extends AbstractCatalog {
     protected final String defaultDatabase;
     protected final String bootstrapServers;
     protected final Map<String, String> securityConfigs;
-    private final LakeCatalog lakeCatalog;
+    private final LakeFlinkCatalog lakeFlinkCatalog;
     protected Connection connection;
     protected Admin admin;
 
@@ -130,12 +130,12 @@ public FlinkCatalog(
         this.bootstrapServers = bootstrapServers;
         this.classLoader = classLoader;
         this.securityConfigs = securityConfigs;
-        this.lakeCatalog = new LakeCatalog(catalogName, classLoader);
+        this.lakeFlinkCatalog = new LakeFlinkCatalog(catalogName, classLoader);
     }
 
     @Override
     public Optional<Factory> getFactory() {
-        return Optional.of(new FlinkTableFactory(lakeCatalog));
+        return Optional.of(new FlinkTableFactory(lakeFlinkCatalog));
     }
 
     @Override
@@ -340,7 +340,7 @@ protected CatalogBaseTable getLakeTable(
             // Need to reconstruct: table_name + $snapshots
             tableName = String.join("", tableComponents);
         }
-        return lakeCatalog
+        return lakeFlinkCatalog
                 .getLakeCatalog(properties)
                 .getTable(new ObjectPath(databaseName, tableName));
     }

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

@@ -20,7 +20,7 @@
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.flink.FlinkConnectorOptions;
-import org.apache.fluss.flink.lake.LakeCatalog;
+import org.apache.fluss.flink.lake.LakeFlinkCatalog;
 import org.apache.fluss.flink.lake.LakeTableFactory;
 import org.apache.fluss.flink.sink.FlinkTableSink;
 import org.apache.fluss.flink.source.FlinkTableSource;
@@ -69,11 +69,11 @@
 /** Factory to create table source and table sink for Fluss. */
 public class FlinkTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
 
-    private final LakeCatalog lakeCatalog;
+    private final LakeFlinkCatalog lakeFlinkCatalog;
     private volatile LakeTableFactory lakeTableFactory;
 
-    public FlinkTableFactory(LakeCatalog lakeCatalog) {
-        this.lakeCatalog = lakeCatalog;
+    public FlinkTableFactory(LakeFlinkCatalog lakeFlinkCatalog) {
+        this.lakeFlinkCatalog = lakeFlinkCatalog;
     }
 
     @Override
@@ -257,7 +257,7 @@ private LakeTableFactory mayInitLakeTableFactory() {
         if (lakeTableFactory == null) {
             synchronized (this) {
                 if (lakeTableFactory == null) {
-                    lakeTableFactory = new LakeTableFactory(lakeCatalog);
+                    lakeTableFactory = new LakeTableFactory(lakeFlinkCatalog);
                 }
             }
         }

…apache/fluss/flink/lake/LakeCatalog.java …e/fluss/flink/lake/LakeFlinkCatalog.javafluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeCatalog.java renamed to fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeFlinkCatalog.java fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeCatalog.java renamed to fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeFlinkCatalog.java

@@ -21,7 +21,6 @@
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.flink.utils.DataLakeUtils;
 import org.apache.fluss.metadata.DataLakeFormat;
-import org.apache.fluss.utils.MapUtils;
 
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
@@ -34,59 +33,64 @@
 
 import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
 import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
-/** A lake catalog to delegate the operations on lake table. */
-public class LakeCatalog {
-    private static final Map<DataLakeFormat, Catalog> LAKE_CATALOG_CACHE =
-            MapUtils.newConcurrentHashMap();
+/** A lake flink catalog to delegate the operations on lake table. */
+public class LakeFlinkCatalog {
 
     private final String catalogName;
     private final ClassLoader classLoader;
 
-    public LakeCatalog(String catalogName, ClassLoader classLoader) {
+    private volatile Catalog catalog;
+    private volatile DataLakeFormat lakeFormat;
+
+    public LakeFlinkCatalog(String catalogName, ClassLoader classLoader) {
         this.catalogName = catalogName;
         this.classLoader = classLoader;
     }
 
     public Catalog getLakeCatalog(Configuration tableOptions) {
-        DataLakeFormat lakeFormat = tableOptions.get(ConfigOptions.TABLE_DATALAKE_FORMAT);
-        if (lakeFormat == null) {
-            throw new IllegalArgumentException(
-                    "DataLake format is not specified in table options. "
-                            + "Please ensure '"
-                            + ConfigOptions.TABLE_DATALAKE_FORMAT.key()
-                            + "' is set.");
-        }
-        return LAKE_CATALOG_CACHE.computeIfAbsent(
-                lakeFormat,
-                (dataLakeFormat) -> {
-                    if (dataLakeFormat == PAIMON) {
-                        return PaimonCatalogFactory.create(catalogName, tableOptions, classLoader);
-                    } else if (dataLakeFormat == ICEBERG) {
-                        return IcebergCatalogFactory.create(catalogName, tableOptions);
+        // TODO: Currently, a Fluss cluster only supports a single DataLake storage.
+        // However, in the
+        //  future, it may support multiple DataLakes. The following code assumes
+        // that a single
+        //  lakeCatalog is shared across multiple tables, which will no longer be
+        // valid in such
+        //  cases and should be updated accordingly.
+        if (catalog == null) {
+            synchronized (this) {
+                if (catalog == null) {
+                    DataLakeFormat lakeFormat =
+                            tableOptions.get(ConfigOptions.TABLE_DATALAKE_FORMAT);
+                    if (lakeFormat == null) {
+                        throw new IllegalArgumentException(
+                                "DataLake format is not specified in table options. "
+                                        + "Please ensure '"
+                                        + ConfigOptions.TABLE_DATALAKE_FORMAT.key()
+                                        + "' is set.");
+                    }
+                    if (lakeFormat == PAIMON) {
+                        catalog =
+                                PaimonCatalogFactory.create(catalogName, tableOptions, classLoader);
+                        this.lakeFormat = PAIMON;
+                    } else if (lakeFormat == ICEBERG) {
+                        catalog = IcebergCatalogFactory.create(catalogName, tableOptions);
+                        this.lakeFormat = ICEBERG;
                     } else {
                         throw new UnsupportedOperationException(
-                                "Unsupported datalake format: " + dataLakeFormat);
+                                "Unsupported data lake format: " + lakeFormat);
                     }
-                });
+                }
+            }
+        }
+        return catalog;
     }
 
-    public Catalog getLakeCatalog(Configuration tableOptions, DataLakeFormat lakeFormat) {
-        if (lakeFormat == null) {
-            throw new IllegalArgumentException("DataLake format cannot be null");
-        }
-        return LAKE_CATALOG_CACHE.computeIfAbsent(
+    public DataLakeFormat getLakeFormat() {
+        checkNotNull(
                 lakeFormat,
-                (dataLakeFormat) -> {
-                    if (dataLakeFormat == PAIMON) {
-                        return PaimonCatalogFactory.create(catalogName, tableOptions, classLoader);
-                    } else if (dataLakeFormat == ICEBERG) {
-                        return IcebergCatalogFactory.create(catalogName, tableOptions);
-                    } else {
-                        throw new UnsupportedOperationException(
-                                "Unsupported datalake format: " + dataLakeFormat);
-                    }
-                });
+                "DataLake format is null, must call getLakeCatalog first to initialize lake format.");
+        return lakeFormat;
     }
 
     /**

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

@@ -17,20 +17,20 @@
 
 package org.apache.fluss.flink.lake;
 
+import org.apache.fluss.config.Configuration;
+
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.factories.DynamicTableFactory;
 import org.apache.flink.table.factories.DynamicTableSourceFactory;
 import org.apache.flink.table.factories.FactoryUtil;
 
-import java.util.Map;
-
 /** A factory to create {@link DynamicTableSource} for lake table. */
 public class LakeTableFactory {
-    private final LakeCatalog lakeCatalog;
+    private final LakeFlinkCatalog lakeFlinkCatalog;
 
-    public LakeTableFactory(LakeCatalog lakeCatalog) {
-        this.lakeCatalog = lakeCatalog;
+    public LakeTableFactory(LakeFlinkCatalog lakeFlinkCatalog) {
+        this.lakeFlinkCatalog = lakeFlinkCatalog;
     }
 
     public DynamicTableSource createDynamicTableSource(
@@ -42,21 +42,6 @@ public DynamicTableSource createDynamicTableSource(
                         originIdentifier.getDatabaseName(),
                         tableName);
 
-        // Determine the lake format from the table options
-        Map<String, String> tableOptions = context.getCatalogTable().getOptions();
-
-        // If not present, fallback to 'fluss.table.datalake.format' (set by Fluss)
-        String connector = tableOptions.get("connector");
-        if (connector == null) {
-            connector = tableOptions.get("fluss.table.datalake.format");
-        }
-
-        if (connector == null) {
-            // For Paimon system tables (like table_name$options), the table options are empty
-            // Default to Paimon for backward compatibility
-            connector = "paimon";
-        }
-
         // For Iceberg and Paimon, pass the table name as-is to their factory.
         // Metadata tables will be handled internally by their respective factories.
         DynamicTableFactory.Context newContext =
@@ -69,44 +54,41 @@ public DynamicTableSource createDynamicTableSource(
                         context.isTemporary());
 
         // Get the appropriate factory based on connector type
-        DynamicTableSourceFactory factory = getLakeTableFactory(connector, tableOptions);
+        DynamicTableSourceFactory factory = getLakeTableFactory();
         return factory.createDynamicTableSource(newContext);
     }
 
-    private DynamicTableSourceFactory getLakeTableFactory(
-            String connector, Map<String, String> tableOptions) {
-        if ("paimon".equalsIgnoreCase(connector)) {
-            return getPaimonFactory();
-        } else if ("iceberg".equalsIgnoreCase(connector)) {
-            return getIcebergFactory(tableOptions);
-        } else {
-            throw new UnsupportedOperationException(
-                    "Unsupported lake connector: "
-                            + connector
-                            + ". Only 'paimon' and 'iceberg' are supported.");
+    private DynamicTableSourceFactory getLakeTableFactory() {
+        switch (lakeFlinkCatalog.getLakeFormat()) {
+            case PAIMON:
+                return getPaimonFactory();
+            case ICEBERG:
+                return getIcebergFactory();
+            default:
+                throw new UnsupportedOperationException(
+                        "Unsupported lake connector: "
+                                + lakeFlinkCatalog.getLakeFormat()
+                                + ". Only 'paimon' and 'iceberg' are supported.");
         }
     }
 
     private DynamicTableSourceFactory getPaimonFactory() {
         return new org.apache.paimon.flink.FlinkTableFactory();
     }
 
-    private DynamicTableSourceFactory getIcebergFactory(Map<String, String> tableOptions) {
+    private DynamicTableSourceFactory getIcebergFactory() {
         try {
-            // Get the Iceberg FlinkCatalog instance from LakeCatalog
-            org.apache.fluss.config.Configuration flussConfig =
-                    org.apache.fluss.config.Configuration.fromMap(tableOptions);
-
             // Get catalog with explicit ICEBERG format
             org.apache.flink.table.catalog.Catalog catalog =
-                    lakeCatalog.getLakeCatalog(
-                            flussConfig, org.apache.fluss.metadata.DataLakeFormat.ICEBERG);
+                    lakeFlinkCatalog.getLakeCatalog(
+                            // we can pass empty configuration to get catalog
+                            // since the catalog should already be initialized
+                            new Configuration());
 
             // Create FlinkDynamicTableFactory with the catalog
             Class<?> icebergFactoryClass =
                     Class.forName("org.apache.iceberg.flink.FlinkDynamicTableFactory");
             Class<?> flinkCatalogClass = Class.forName("org.apache.iceberg.flink.FlinkCatalog");
-
             return (DynamicTableSourceFactory)
                     icebergFactoryClass
                             .getDeclaredConstructor(flinkCatalogClass)

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadPrimaryKeyTableITCase.java

@@ -278,7 +278,7 @@ void testUnionReadInStreamMode(Boolean isPartitioned) throws Exception {
 
     @ParameterizedTest
     @ValueSource(booleans = {false, true})
-    void testReadIcebergLakeTableAndSystemTable(boolean isPartitioned) throws Exception {
+    void testReadIcebergLakeTable(boolean isPartitioned) throws Exception {
         // first of all, start tiering
         JobClient jobClient = buildTieringJob(execEnv);
 
@@ -308,13 +308,6 @@ void testReadIcebergLakeTableAndSystemTable(boolean isPartitioned) throws Except
         int expectedUserRowCount = isPartitioned ? 2 * waitUntilPartitions(t1).size() : 2;
         assertThat(icebergRows).hasSize(expectedUserRowCount);
 
-        // verify rows have expected number of columns
-        int userColumnCount = lakeTableResult.getResolvedSchema().getColumnCount();
-        Row firstRow = icebergRows.get(0);
-        assertThat(firstRow.getArity())
-                .as("Iceberg row should have at least user columns")
-                .isGreaterThanOrEqualTo(userColumnCount);
-
         // Test 2: Read Iceberg system table (snapshots) using $lake$snapshots suffix
         TableResult snapshotsResult =
                 batchTEnv.executeSql(String.format("select * from %s$lake$snapshots", tableName));