[FLINK-38359][pipeline-connector][paimon] Avoid exception when writing update event to Append only table. (apache#4120)

Author: lvyanquan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonRecordEventSerializer.java

@@ -80,10 +80,12 @@ public PaimonEvent serialize(Event event) {
             return new PaimonEvent(tableId, null, true);
         } else if (event instanceof DataChangeEvent) {
             DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
+            TableSchemaInfo schemaInfo = schemaMaps.get(dataChangeEvent.tableId());
             List<GenericRow> genericRows =
                     PaimonWriterHelper.convertEventToFullGenericRows(
                             dataChangeEvent,
-                            schemaMaps.get(dataChangeEvent.tableId()).getFieldGetters());
+                            schemaInfo.getFieldGetters(),
+                            schemaInfo.hasPrimaryKey());
             return new PaimonEvent(tableId, genericRows, false, bucket);
         } else {
             throw new IllegalArgumentException(

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelper.java

@@ -222,7 +222,9 @@ public static GenericRow convertEventToGenericRow(
 
     /** create full {@link GenericRow}s from a {@link DataChangeEvent} for {@link PaimonWriter}. */
     public static List<GenericRow> convertEventToFullGenericRows(
-            DataChangeEvent dataChangeEvent, List<RecordData.FieldGetter> fieldGetters) {
+            DataChangeEvent dataChangeEvent,
+            List<RecordData.FieldGetter> fieldGetters,
+            boolean hasPrimaryKey) {
         List<GenericRow> fullGenericRows = new ArrayList<>();
         switch (dataChangeEvent.op()) {
             case INSERT:
@@ -235,19 +237,25 @@ public static List<GenericRow> convertEventToFullGenericRows(
             case UPDATE:
             case REPLACE:
                 {
-                    fullGenericRows.add(
-                            convertRecordDataToGenericRow(
-                                    dataChangeEvent.before(), fieldGetters, RowKind.UPDATE_BEFORE));
+                    if (hasPrimaryKey) {
+                        fullGenericRows.add(
+                                convertRecordDataToGenericRow(
+                                        dataChangeEvent.before(),
+                                        fieldGetters,
+                                        RowKind.UPDATE_BEFORE));
+                    }
                     fullGenericRows.add(
                             convertRecordDataToGenericRow(
                                     dataChangeEvent.after(), fieldGetters, RowKind.UPDATE_AFTER));
                     break;
                 }
             case DELETE:
                 {
-                    fullGenericRows.add(
-                            convertRecordDataToGenericRow(
-                                    dataChangeEvent.before(), fieldGetters, RowKind.DELETE));
+                    if (hasPrimaryKey) {
+                        fullGenericRows.add(
+                                convertRecordDataToGenericRow(
+                                        dataChangeEvent.before(), fieldGetters, RowKind.DELETE));
+                    }
                     break;
                 }
             default:

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/TableSchemaInfo.java

@@ -30,9 +30,12 @@ public class TableSchemaInfo {
 
     private final List<RecordData.FieldGetter> fieldGetters;
 
+    private final boolean hasPrimaryKey;
+
     public TableSchemaInfo(Schema schema, ZoneId zoneId) {
         this.schema = schema;
         this.fieldGetters = PaimonWriterHelper.createFieldGetters(schema, zoneId);
+        this.hasPrimaryKey = !schema.primaryKeys().isEmpty();
     }
 
     public Schema getSchema() {
@@ -42,4 +45,8 @@ public Schema getSchema() {
     public List<RecordData.FieldGetter> getFieldGetters() {
         return fieldGetters;
     }
+
+    public boolean hasPrimaryKey() {
+        return hasPrimaryKey;
+    }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java

@@ -356,6 +356,33 @@ public void testSinkWithDataChangeForAppendOnlyTable(String metastore, boolean e
                         Row.ofKind(RowKind.INSERT, "1", "1"),
                         Row.ofKind(RowKind.INSERT, "2", "2"),
                         Row.ofKind(RowKind.INSERT, "3", "3"));
+
+        // Delete
+        writeAndCommit(
+                writer,
+                committer,
+                generateDelete(
+                        table1, Arrays.asList(Tuple2.of(STRING(), "3"), Tuple2.of(STRING(), "3"))));
+        Assertions.assertThat(fetchResults(table1))
+                .containsExactlyInAnyOrder(
+                        Row.ofKind(RowKind.INSERT, "1", "1"),
+                        Row.ofKind(RowKind.INSERT, "2", "2"),
+                        Row.ofKind(RowKind.INSERT, "3", "3"));
+
+        // Update
+        writeAndCommit(
+                writer,
+                committer,
+                generateUpdate(
+                        table1,
+                        Arrays.asList(Tuple2.of(STRING(), "3"), Tuple2.of(STRING(), "3")),
+                        Arrays.asList(Tuple2.of(STRING(), "3"), Tuple2.of(STRING(), "x"))));
+        Assertions.assertThat(fetchResults(table1))
+                .containsExactlyInAnyOrder(
+                        Row.ofKind(RowKind.INSERT, "1", "1"),
+                        Row.ofKind(RowKind.INSERT, "2", "2"),
+                        Row.ofKind(RowKind.INSERT, "3", "3"),
+                        Row.ofKind(RowKind.INSERT, "3", "x"));
     }
 
     @ParameterizedTest