[server] Get table.datalake.format from dynamic configs when get table info.

Author: loserwang1024

fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java

@@ -562,10 +562,14 @@ public TableInfo getTable(TablePath tablePath) throws TableNotExistException {
         }
         TableRegistration tableReg = optionalTable.get();
         SchemaInfo schemaInfo = getLatestSchema(tablePath);
-        Map<String, String> tableLakeOptions =
-                lakeCatalogDynamicLoader.getLakeCatalogContainer().getDefaultTableLakeOptions();
-        removeSensitiveTableOptions(tableLakeOptions);
-        return tableReg.toTableInfo(tablePath, schemaInfo, tableLakeOptions);
+        LakeCatalogDynamicLoader.LakeCatalogContainer lakeCatalogContainer =
+                lakeCatalogDynamicLoader.getLakeCatalogContainer();
+        removeSensitiveTableOptions(lakeCatalogContainer.getDefaultTableLakeOptions());
+        return tableReg.toTableInfo(
+                tablePath,
+                schemaInfo,
+                lakeCatalogContainer.getDefaultTableLakeOptions(),
+                lakeCatalogContainer.getDataLakeFormat());
     }
 
     public Map<TablePath, TableInfo> getTables(Collection<TablePath> tablePaths)
@@ -587,15 +591,16 @@ public Map<TablePath, TableInfo> getTables(Collection<TablePath> tablePaths)
                 }
                 TableRegistration tableReg = tablePath2TableRegistrations.get(tablePath);
                 SchemaInfo schemaInfo = tablePath2SchemaInfos.get(tablePath);
+                LakeCatalogDynamicLoader.LakeCatalogContainer lakeCatalogContainer =
+                        lakeCatalogDynamicLoader.getLakeCatalogContainer();
 
                 result.put(
                         tablePath,
                         tableReg.toTableInfo(
                                 tablePath,
                                 schemaInfo,
-                                lakeCatalogDynamicLoader
-                                        .getLakeCatalogContainer()
-                                        .getDefaultTableLakeOptions()));
+                                lakeCatalogContainer.getDefaultTableLakeOptions(),
+                                lakeCatalogContainer.getDataLakeFormat()));
             }
         } catch (Exception e) {
             throw new FlussRuntimeException(

fluss-server/src/main/java/org/apache/fluss/server/zk/data/TableRegistration.java

@@ -20,6 +20,7 @@
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.SchemaInfo;
 import org.apache.fluss.metadata.TableDescriptor;
@@ -87,18 +88,27 @@ public TableConfig getTableConfig() {
     }
 
     public TableInfo toTableInfo(TablePath tablePath, SchemaInfo schemaInfo) {
-        return toTableInfo(tablePath, schemaInfo, null);
+        return toTableInfo(tablePath, schemaInfo, null, null);
     }
 
     public TableInfo toTableInfo(
             TablePath tablePath,
             SchemaInfo schemaInfo,
-            @Nullable Map<String, String> defaultTableLakeOptions) {
+            @Nullable Map<String, String> defaultTableLakeOptions,
+            @Nullable DataLakeFormat lakeFormat) {
         Configuration properties = Configuration.fromMap(this.properties);
-        if (defaultTableLakeOptions != null) {
+        if (defaultTableLakeOptions != null || lakeFormat != null) {
             if (properties.get(ConfigOptions.TABLE_DATALAKE_ENABLED)) {
-                // only make the lake options visible when the datalake is enabled on the table
-                defaultTableLakeOptions.forEach(properties::setString);
+                if (defaultTableLakeOptions != null) {
+                    // only make the lake options visible when the datalake is enabled on the table
+                    defaultTableLakeOptions.forEach(properties::setString);
+                }
+                if (lakeFormat != null
+                        && !properties
+                                .getOptional(ConfigOptions.TABLE_DATALAKE_FORMAT)
+                                .isPresent()) {
+                    properties.set(ConfigOptions.TABLE_DATALAKE_FORMAT, lakeFormat);
+                }
             }
         }
         return new TableInfo(