add config covert uniq to pk

JNSimba · Dec 5, 2024 · bb66a3b · bb66a3b
1 parent 908bddb
commit bb66a3b
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 5 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java
@@ -463,6 +463,11 @@ public void setTableSchemaBuckets(
     private void tryCreateTableIfAbsent(
             DorisSystem dorisSystem, String targetDb, String dorisTable, SourceSchema schema) {
         if (!dorisSystem.tableExists(targetDb, dorisTable)) {
+            if (dorisTableConfig.isConvertUniqToPk()
+                    && CollectionUtil.isNullOrEmpty(schema.primaryKeys)
+                    && !CollectionUtil.isNullOrEmpty(schema.uniqueIndexs)) {
+                schema.primaryKeys = new ArrayList<>(schema.uniqueIndexs);
+            }
             TableSchema dorisSchema =
                     DorisSchemaFactory.createTableSchema(
                             database,

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DorisTableConfig.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DorisTableConfig.java
@@ -32,13 +32,16 @@ public class DorisTableConfig implements Serializable {
     public static final String REPLICATION_NUM = "replication_num";
     public static final String TABLE_BUCKETS = "table-buckets";
     public static final String TABLE_PARTITIONS = "table-partitions";
+    public static final String CONVERT_UNIQ_TO_PK = "convert-uniq-to-pk";
 
     private final Map<String, String> tableProperties;
     // The specific parameters extracted from --table-conf need to be parsed and integrated into the
     // doris table creation statement. such as: table-buckets="tbl1:10,tbl2:20,a.*:30,b.*:40,.*:50".
     private Map<String, Integer> tableBuckets;
     // table:partitionColumn:interval
     private Map<String, Tuple2<String, String>> tablePartitions;
+    // uniq index to primary key
+    private boolean convertUniqToPk = false;
 
     // Only for testing
     @VisibleForTesting
@@ -64,6 +67,11 @@ public DorisTableConfig(Map<String, String> tableConfig) {
             tableConfig.remove(TABLE_PARTITIONS);
         }
 
+        if (tableConfig.containsKey(CONVERT_UNIQ_TO_PK)) {
+            this.convertUniqToPk = Boolean.parseBoolean(tableConfig.get(CONVERT_UNIQ_TO_PK));
+            tableConfig.remove(CONVERT_UNIQ_TO_PK);
+        }
+
         tableProperties = tableConfig;
     }
 
@@ -79,6 +87,10 @@ public Map<String, Tuple2<String, String>> getTablePartitions() {
         return tablePartitions;
     }
 
+    public boolean isConvertUniqToPk() {
+        return convertUniqToPk;
+    }
+
     /**
      * Build table bucket Map.
      *

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/JdbcSourceSchema.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/JdbcSourceSchema.java
@@ -49,10 +49,7 @@ public JdbcSourceSchema(
         super(databaseName, schemaName, tableName, tableComment);
         fields = getColumnInfo(metaData, databaseName, schemaName, tableName);
         primaryKeys = getPrimaryKeys(metaData, databaseName, schemaName, tableName);
-        if (primaryKeys.isEmpty()) {
-            List<String> uniqIndex = getUniqIndex(metaData, databaseName, schemaName, tableName);
-            primaryKeys.addAll(uniqIndex);
-        }
+        uniqueIndexs = getUniqIndex(metaData, databaseName, schemaName, tableName);
     }
 
     public LinkedHashMap<String, FieldSchema> getColumnInfo(

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/SourceSchema.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/SourceSchema.java
@@ -37,6 +37,7 @@ public abstract class SourceSchema {
     protected final String tableComment;
     protected LinkedHashMap<String, FieldSchema> fields;
     public List<String> primaryKeys;
+    public List<String> uniqueIndexs;
     public DataModel model = DataModel.UNIQUE;
 
     public SourceSchema(
@@ -64,7 +65,6 @@ public static String getString(String databaseName, String schemaName, String ta
         if (!StringUtils.isNullOrWhitespaceOnly(schemaName)) {
             identifier.add(schemaName);
         }
-
         if (!StringUtils.isNullOrWhitespaceOnly(tableName)) {
             identifier.add(tableName);
         }
@@ -115,6 +115,10 @@ public List<String> getPrimaryKeys() {
         return primaryKeys;
     }
 
+    public List<String> getUniqueIndexs() {
+        return uniqueIndexs;
+    }
+
     public String getTableComment() {
         return tableComment;
     }