[flink] Improved Handling of Custom Properties During Fluss Table Creation

[flink] Improved Handling of Custom Properties During Fluss Table Creation

Author: rionmonster

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

@@ -45,14 +45,7 @@
 import org.apache.flink.types.RowKind;
 
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 import java.util.stream.Collectors;
 
 import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;
@@ -202,16 +195,18 @@ public static TableDescriptor toFlussTable(ResolvedCatalogTable catalogTable) {
                                     "Metadata column " + col + " is not supported.");
                         });
 
-        Map<String, String> customProperties = flinkTableConf.toMap();
-        CatalogPropertiesUtils.serializeComputedColumns(
-                customProperties, resolvedSchema.getColumns());
+        Map<String, String> properties = flinkTableConf.toMap();
+        CatalogPropertiesUtils.serializeComputedColumns(properties, resolvedSchema.getColumns());
         CatalogPropertiesUtils.serializeWatermarkSpecs(
-                customProperties, catalogTable.getResolvedSchema().getWatermarkSpecs());
+                properties, catalogTable.getResolvedSchema().getWatermarkSpecs());
 
         String comment = catalogTable.getComment();
 
         // convert some flink options to fluss table configs.
-        Map<String, String> properties = convertFlinkOptionsToFlussTableProperties(flinkTableConf);
+        Map<String, String> flussTableProperties =
+                convertFlinkOptionsToFlussTableProperties(flinkTableConf);
+        Map<String, String> customProperties =
+                extractCustomProperties(properties, flussTableProperties);
 
         // then set distributed by information
         List<String> bucketKey;
@@ -240,7 +235,7 @@ public static TableDescriptor toFlussTable(ResolvedCatalogTable catalogTable) {
                 .partitionedBy(catalogTable.getPartitionKeys())
                 .distributedBy(bucketNum, bucketKey)
                 .comment(comment)
-                .properties(properties)
+                .properties(flussTableProperties)
                 .customProperties(customProperties)
                 .build();
     }
@@ -352,4 +347,11 @@ private static void convertFlussTablePropertiesToFlinkOptions(
             }
         }
     }
+
+    private static Map<String, String> extractCustomProperties(
+            Map<String, String> allProperties, Map<String, String> flussTableProperties) {
+        Map<String, String> customProperties = new HashMap<>(allProperties);
+        customProperties.keySet().removeAll(flussTableProperties.keySet());
+        return customProperties;
+    }
 }

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java

@@ -254,6 +254,43 @@ void testTableConversionWithOptions() {
                 .containsEntry(ConfigOptions.CLIENT_WRITER_BATCH_TIMEOUT.key(), "32s");
     }
 
+    @Test
+    void testTableConversionForCustomProperties() {
+        Map<String, String> options = new HashMap<>();
+        // forward table option & enum type
+        options.put(ConfigOptions.TABLE_LOG_FORMAT.key(), "indexed");
+        // forward client memory option
+        options.put(ConfigOptions.CLIENT_WRITER_BUFFER_MEMORY_SIZE.key(), "64mb");
+        // forward client duration option
+        options.put(ConfigOptions.CLIENT_WRITER_BATCH_TIMEOUT.key(), "32s");
+
+        ResolvedSchema schema =
+                new ResolvedSchema(
+                        Collections.singletonList(
+                                Column.physical(
+                                        "order_id",
+                                        org.apache.flink.table.api.DataTypes.STRING().notNull())),
+                        Collections.emptyList(),
+                        null);
+        CatalogTable flinkTable =
+                CatalogTable.of(
+                        Schema.newBuilder().fromResolvedSchema(schema).build(),
+                        "test comment",
+                        Collections.emptyList(),
+                        options);
+
+        TableDescriptor flussTable =
+                FlinkConversions.toFlussTable(new ResolvedCatalogTable(flinkTable, schema));
+
+        assertThat(flussTable.getProperties())
+                .containsEntry(ConfigOptions.TABLE_LOG_FORMAT.key(), "indexed");
+
+        HashMap<String, String> customProperties = new HashMap<>();
+        customProperties.put(ConfigOptions.CLIENT_WRITER_BUFFER_MEMORY_SIZE.key(), "64mb");
+        customProperties.put(ConfigOptions.CLIENT_WRITER_BATCH_TIMEOUT.key(), "32s");
+        assertThat(flussTable.getCustomProperties()).containsExactlyEntriesOf(customProperties);
+    }
+
     @Test
     void testOptionConversions() {
         ConfigOption<?> flinkOption = FlinkConversions.toFlinkOption(ConfigOptions.TABLE_KV_FORMAT);