[hotfix] Flink catalog should pass to lake properties to lake table (apache#2205)

Author: luoyuxia

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

@@ -335,6 +335,14 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
             Map<String, String> newOptions = new HashMap<>(catalogBaseTable.getOptions());
             newOptions.put(BOOTSTRAP_SERVERS.key(), bootstrapServers);
             newOptions.putAll(securityConfigs);
+            // add lake properties
+            if (tableInfo.getTableConfig().isDataLakeEnabled()) {
+                for (Map.Entry<String, String> lakePropertyEntry :
+                        getLakeCatalogProperties().entrySet()) {
+                    String key = "table.datalake." + lakePropertyEntry.getKey();
+                    newOptions.put(key, lakePropertyEntry.getValue());
+                }
+            }
             if (CatalogBaseTable.TableKind.TABLE == catalogBaseTable.getTableKind()) {
                 return ((CatalogTable) catalogBaseTable).copy(newOptions);
             } else if (CatalogBaseTable.TableKind.MATERIALIZED_TABLE

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

@@ -35,6 +35,7 @@
 import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.api.config.TableConfigOptions;
 import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.catalog.ObjectPath;
 import org.apache.flink.table.catalog.exceptions.CatalogException;
@@ -819,7 +820,7 @@ void testCreateCatalogWithUnexistedDatabase() {
     }
 
     @Test
-    void testCreateCatalogWithLakeProperties() {
+    void testCreateCatalogWithLakeProperties() throws Exception {
         Map<String, String> properties = new HashMap<>();
         properties.put("paimon.jdbc.password", "pass");
         tEnv.executeSql(
@@ -830,6 +831,18 @@ void testCreateCatalogWithLakeProperties() {
                 (FlinkCatalog) tEnv.getCatalog("test_catalog_with_lake_properties").get();
 
         assertOptionsEqual(catalog.getLakeCatalogProperties(), properties);
+
+        String ddl =
+                "create table test_get_lake_table ("
+                        + "a string, "
+                        + "b int) "
+                        + "with ('bucket.num' = '5', 'table.datalake.enabled' = 'true')";
+        tEnv.executeSql(ddl);
+
+        CatalogBaseTable catalogTable =
+                catalog.getTable(new ObjectPath(DEFAULT_DB, "test_get_lake_table"));
+        assertThat(catalogTable.getOptions())
+                .containsEntry("table.datalake.paimon.jdbc.password", "pass");
     }
 
     /**