[cdc] Optimize Duplicate Field Detection in MySQL CDC Schema Building. (

apache#2078)
wg1026688210 · Sep 28, 2023 · ae1dde1 · ae1dde1
1 parent 10540c7
commit ae1dde1
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 13 deletions.
diff --git a/...imon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java b/...imon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
@@ -140,8 +140,9 @@ public static String columnCaseConvertAndDuplicateCheck(
         if (caseSensitive) {
             return column;
         }
-        checkArgument(existedFields.add(column.toLowerCase()), columnDuplicateErrMsg.apply(column));
-        return column.toLowerCase();
+        String columnLowerCase = column.toLowerCase();
+        checkArgument(existedFields.add(columnLowerCase), columnDuplicateErrMsg.apply(column));
+        return columnLowerCase;
     }
 
     public static Schema buildPaimonSchema(

diff --git a/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTableSchemaBuilder.java b/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTableSchemaBuilder.java
@@ -27,15 +27,17 @@
 import io.debezium.relational.Table;
 import io.debezium.relational.history.TableChanges;
 
-import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
 
+import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.columnCaseConvertAndDuplicateCheck;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.columnDuplicateErrMsg;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.listCaseConvert;
 import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TO_NULLABLE;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /** Schema builder for MySQL cdc. */
 public class MySqlTableSchemaBuilder implements NewTableSchemaBuilder<TableChanges.TableChange> {
@@ -57,8 +59,10 @@ public Optional<Schema> build(TableChanges.TableChange tableChange) {
         String tableName = tableChange.getId().toString();
         List<Column> columns = table.columns();
 
+        Set<String> existedFields = new HashSet<>();
+        Function<String, String> columnDuplicateErrMsg = columnDuplicateErrMsg(tableName);
+
         Schema.Builder builder = Schema.newBuilder();
-        Map<String, Integer> duplicateFields = new HashMap<>();
 
         // column
         for (Column column : columns) {
@@ -71,16 +75,12 @@ public Optional<Schema> build(TableChanges.TableChange tableChange) {
 
             dataType = dataType.copy(typeMapping.containsMode(TO_NULLABLE) || column.isOptional());
 
-            String columnName = column.name();
-            if (!caseSensitive) {
-                checkArgument(
-                        !duplicateFields.containsKey(columnName.toLowerCase()),
-                        columnDuplicateErrMsg(tableName).apply(columnName));
-                columnName = columnName.toLowerCase();
-            }
+            String columnName =
+                    columnCaseConvertAndDuplicateCheck(
+                            column.name(), existedFields, caseSensitive, columnDuplicateErrMsg);
+
             // TODO : add table comment and column comment when we upgrade flink cdc to 2.4
             builder.column(columnName, dataType, null);
-            duplicateFields.put(columnName, 1);
         }
 
         // primaryKey