[cdc] Kafka database sync new table shouldn't require primary keys (a…

…pache#4202)
xiangyuf · Sep 18, 2024 · 102e5a9 · 102e5a9
1 parent cc573ea
commit 102e5a9
Show file tree

Hide file tree

Showing 6 changed files with 28 additions and 3 deletions.
diff --git a/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java b/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
@@ -148,6 +148,7 @@ protected EventParser.Factory<RichCdcMultiplexRecord> buildEventParserFactory()
                         allowUpperCase,
                         partitionKeys,
                         primaryKeys,
+                        requirePrimaryKeys(),
                         partitionKeyMultiple,
                         metadataConverters);
         Pattern includingPattern = Pattern.compile(includingTables);
@@ -170,6 +171,8 @@ protected EventParser.Factory<RichCdcMultiplexRecord> buildEventParserFactory()
                         createdTables);
     }
 
+    protected abstract boolean requirePrimaryKeys();
+
     @Override
     protected void buildSink(
             DataStream<RichCdcMultiplexRecord> input,

diff --git a/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java b/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
@@ -40,4 +40,9 @@ public KafkaSyncDatabaseAction(
     protected CdcTimestampExtractor createCdcTimestampExtractor() {
         return new MessageQueueCdcTimestampExtractor();
     }
+
+    @Override
+    protected boolean requirePrimaryKeys() {
+        return false;
+    }
 }
diff --git a/...c/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java b/...c/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
@@ -73,4 +73,9 @@ protected MongoDBSource<CdcSourceRecord> buildSource() {
                         includingTables,
                         Collections.emptyList()));
     }
+
+    @Override
+    protected boolean requirePrimaryKeys() {
+        return true;
+    }
 }
diff --git a/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java b/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -259,4 +259,9 @@ public List<Identifier> monitoredTables() {
     public List<Identifier> excludedTables() {
         return excludedTables;
     }
+
+    @Override
+    protected boolean requirePrimaryKeys() {
+        return true;
+    }
 }
diff --git a/...cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java b/...cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
@@ -40,4 +40,9 @@ public PulsarSyncDatabaseAction(
     protected CdcTimestampExtractor createCdcTimestampExtractor() {
         return new MessageQueueCdcTimestampExtractor();
     }
+
+    @Override
+    protected boolean requirePrimaryKeys() {
+        return false;
+    }
 }
diff --git a/...aimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java b/...aimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
@@ -24,7 +24,6 @@
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -38,21 +37,24 @@ public class NewTableSchemaBuilder implements Serializable {
     private final boolean caseSensitive;
     private final List<String> partitionKeys;
     private final List<String> primaryKeys;
+    private final boolean requirePrimaryKeys;
     private final CdcMetadataConverter[] metadataConverters;
-    protected Map<String, List<String>> partitionKeyMultiple = new HashMap<>();
+    private final Map<String, List<String>> partitionKeyMultiple;
 
     public NewTableSchemaBuilder(
             Map<String, String> tableConfig,
             boolean caseSensitive,
             List<String> partitionKeys,
             List<String> primaryKeys,
+            boolean requirePrimaryKeys,
             Map<String, List<String>> partitionKeyMultiple,
             CdcMetadataConverter[] metadataConverters) {
         this.tableConfig = tableConfig;
         this.caseSensitive = caseSensitive;
         this.metadataConverters = metadataConverters;
         this.partitionKeys = partitionKeys;
         this.primaryKeys = primaryKeys;
+        this.requirePrimaryKeys = requirePrimaryKeys;
         this.partitionKeyMultiple = partitionKeyMultiple;
     }
 
@@ -84,6 +86,6 @@ public Optional<Schema> build(RichCdcMultiplexRecord record) {
                         metadataConverters,
                         caseSensitive,
                         false,
-                        true));
+                        requirePrimaryKeys));
     }
 }