rebase and resolve conflicts

Author: wuchong

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

@@ -210,15 +210,24 @@ public static TableDescriptor toFlussTable(ResolvedCatalogBaseTable<?> catalogBa
                         .collect(Collectors.toList()));
 
         // convert some flink options to fluss table configs.
-        Map<String, String> properties = convertFlinkOptionsToFlussTableProperties(flinkTableConf);
+        Map<String, String> storageProperties =
+                convertFlinkOptionsToFlussTableProperties(flinkTableConf);
 
-        if (properties.containsKey(TABLE_AUTO_INCREMENT_FIELDS.key())) {
+        if (storageProperties.containsKey(TABLE_AUTO_INCREMENT_FIELDS.key())) {
             for (String autoIncrementColumn :
-                    properties.get(TABLE_AUTO_INCREMENT_FIELDS.key()).split(",")) {
+                    storageProperties.get(TABLE_AUTO_INCREMENT_FIELDS.key()).split(",")) {
                 schemBuilder.enableAutoIncrement(autoIncrementColumn);
             }
         }
 
+        // serialize computed column and watermark spec to custom properties
+        Map<String, String> customProperties =
+                extractCustomProperties(flinkTableConf, storageProperties);
+        CatalogPropertiesUtils.serializeComputedColumns(
+                customProperties, resolvedSchema.getColumns());
+        CatalogPropertiesUtils.serializeWatermarkSpecs(
+                customProperties, catalogBaseTable.getResolvedSchema().getWatermarkSpecs());
+
         Schema schema = schemBuilder.build();
 
         resolvedSchema.getColumns().stream()
@@ -236,11 +245,6 @@ public static TableDescriptor toFlussTable(ResolvedCatalogBaseTable<?> catalogBa
                         ? ((ResolvedCatalogTable) catalogBaseTable).getPartitionKeys()
                         : ((ResolvedCatalogMaterializedTable) catalogBaseTable).getPartitionKeys();
 
-        Map<String, String> properties = flinkTableConf.toMap();
-        CatalogPropertiesUtils.serializeComputedColumns(properties, resolvedSchema.getColumns());
-        CatalogPropertiesUtils.serializeWatermarkSpecs(
-                properties, catalogBaseTable.getResolvedSchema().getWatermarkSpecs());
-
         // Set materialized table flags to fluss table custom properties
         if (CatalogTableAdapter.isMaterializedTable(tableKind)) {
             CatalogMaterializedTable.RefreshMode refreshMode =
@@ -254,11 +258,6 @@ public static TableDescriptor toFlussTable(ResolvedCatalogBaseTable<?> catalogBa
         }
 
         String comment = catalogBaseTable.getComment();
-        // convert some flink options to fluss table configs.
-        Map<String, String> flussTableProperties =
-                convertFlinkOptionsToFlussTableProperties(flinkTableConf);
-        Map<String, String> customProperties =
-                extractCustomProperties(properties, flussTableProperties);
 
         // then set distributed by information
         List<String> bucketKey;
@@ -287,7 +286,7 @@ public static TableDescriptor toFlussTable(ResolvedCatalogBaseTable<?> catalogBa
                 .partitionedBy(partitionKeys)
                 .distributedBy(bucketNum, bucketKey)
                 .comment(comment)
-                .properties(flussTableProperties)
+                .properties(storageProperties)
                 .customProperties(customProperties)
                 .build();
     }
@@ -650,8 +649,8 @@ private static CatalogMaterializedTable toFlinkMaterializedTable(
     }
 
     private static Map<String, String> extractCustomProperties(
-            Map<String, String> allProperties, Map<String, String> flussTableProperties) {
-        Map<String, String> customProperties = new HashMap<>(allProperties);
+            Configuration allProperties, Map<String, String> flussTableProperties) {
+        Map<String, String> customProperties = new HashMap<>(allProperties.toMap());
         customProperties.keySet().removeAll(flussTableProperties.keySet());
         return customProperties;
     }