This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 102e5a916 [cdc] Kafka database sync new table shouldn't require
primary keys (#4202)
102e5a916 is described below
commit 102e5a9168633f68c05d2487a1f90d708279c76d
Author: yuzelin <[email protected]>
AuthorDate: Wed Sep 18 14:26:24 2024 +0800
[cdc] Kafka database sync new table shouldn't require primary keys (#4202)
---
.../apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java | 3 +++
.../paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java | 5 +++++
.../flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java | 5 +++++
.../paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java | 5 +++++
.../paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java | 5 +++++
.../org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java | 8 +++++---
6 files changed, 28 insertions(+), 3 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
index c513aff10..4ab56bdcf 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
@@ -148,6 +148,7 @@ public abstract class SyncDatabaseActionBase extends
SynchronizationActionBase {
allowUpperCase,
partitionKeys,
primaryKeys,
+ requirePrimaryKeys(),
partitionKeyMultiple,
metadataConverters);
Pattern includingPattern = Pattern.compile(includingTables);
@@ -170,6 +171,8 @@ public abstract class SyncDatabaseActionBase extends
SynchronizationActionBase {
createdTables);
}
+ protected abstract boolean requirePrimaryKeys();
+
@Override
protected void buildSink(
DataStream<RichCdcMultiplexRecord> input,
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
index cce4c4e68..089bd035c 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
@@ -40,4 +40,9 @@ public class KafkaSyncDatabaseAction extends
SyncDatabaseActionBase {
protected CdcTimestampExtractor createCdcTimestampExtractor() {
return new MessageQueueCdcTimestampExtractor();
}
+
+ @Override
+ protected boolean requirePrimaryKeys() {
+ return false;
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
index 589513960..9f3ed1085 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
@@ -73,4 +73,9 @@ public class MongoDBSyncDatabaseAction extends
SyncDatabaseActionBase {
includingTables,
Collections.emptyList()));
}
+
+ @Override
+ protected boolean requirePrimaryKeys() {
+ return true;
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
index 316168e1e..01be020f7 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -259,4 +259,9 @@ public class MySqlSyncDatabaseAction extends
SyncDatabaseActionBase {
public List<Identifier> excludedTables() {
return excludedTables;
}
+
+ @Override
+ protected boolean requirePrimaryKeys() {
+ return true;
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
index d097feec6..0b99040f1 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
@@ -40,4 +40,9 @@ public class PulsarSyncDatabaseAction extends
SyncDatabaseActionBase {
protected CdcTimestampExtractor createCdcTimestampExtractor() {
return new MessageQueueCdcTimestampExtractor();
}
+
+ @Override
+ protected boolean requirePrimaryKeys() {
+ return false;
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
index 8ac36d075..ce1bcc47d 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
@@ -24,7 +24,6 @@ import org.apache.paimon.schema.Schema;
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,14 +37,16 @@ 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;
@@ -53,6 +54,7 @@ public class NewTableSchemaBuilder implements Serializable {
this.metadataConverters = metadataConverters;
this.partitionKeys = partitionKeys;
this.primaryKeys = primaryKeys;
+ this.requirePrimaryKeys = requirePrimaryKeys;
this.partitionKeyMultiple = partitionKeyMultiple;
}
@@ -84,6 +86,6 @@ public class NewTableSchemaBuilder implements Serializable {
metadataConverters,
caseSensitive,
false,
- true));
+ requirePrimaryKeys));
}
}
