(paimon) branch master updated: [flink-cdc] Fix partition key conflicts and support append-only parti… (#6706)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi 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 47d87c1d85 [flink-cdc] Fix partition key conflicts and support 
append-only parti… (#6706)
47d87c1d85 is described below

commit 47d87c1d85d6823234422e9fe0433a205551a5c8
Author: prashant8530 <prashant8...@gmail.com>
AuthorDate: Sun May 24 08:16:49 2026 +0530

    [flink-cdc] Fix partition key conflicts and support append-only parti… 
(#6706)
---
 .../flink/pipeline/cdc/schema/PaimonMetadataApplier.java      | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/PaimonMetadataApplier.java
 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/PaimonMetadataApplier.java
index 4798d9a6e7..9fcf197a85 100644
--- 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/PaimonMetadataApplier.java
+++ 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/PaimonMetadataApplier.java
@@ -189,15 +189,18 @@ public class PaimonMetadataApplier implements 
MetadataApplier {
                                                             .getLogicalType()),
                                             column.getComment()));
             List<String> partitionKeys = new ArrayList<>();
-            List<String> primaryKeys = schema.primaryKeys();
+            List<String> primaryKeys = new ArrayList<>(schema.primaryKeys());
             if (partitionMaps.containsKey(event.tableId())) {
                 partitionKeys.addAll(partitionMaps.get(event.tableId()));
             } else if (schema.partitionKeys() != null && 
!schema.partitionKeys().isEmpty()) {
                 partitionKeys.addAll(schema.partitionKeys());
             }
-            for (String partitionColumn : partitionKeys) {
-                if (!primaryKeys.contains(partitionColumn)) {
-                    primaryKeys.add(partitionColumn);
+            // Only add partition keys to primary keys if primary keys exist
+            if (!primaryKeys.isEmpty()) {
+                for (String partitionColumn : partitionKeys) {
+                    if (!primaryKeys.contains(partitionColumn)) {
+                        primaryKeys.add(partitionColumn);
+                    }
                 }
             }
             builder.primaryKey(primaryKeys)