You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2025/09/17 03:16:08 UTC
(flink-cdc) branch master updated: [FLINK-38359][pipeline-connector][paimon] Avoid exception when writing update event to Append only table. (#4120)
This is an automated email from the ASF dual-hosted git repository.
kunni pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 32ec9f57e [FLINK-38359][pipeline-connector][paimon] Avoid exception when writing update event to Append only table. (#4120)
32ec9f57e is described below
commit 32ec9f57e6be7f8152d5e27cbf2423dc23055b6b
Author: Kunni <de...@gmail.com>
AuthorDate: Wed Sep 17 11:16:04 2025 +0800
[FLINK-38359][pipeline-connector][paimon] Avoid exception when writing update event to Append only table. (#4120)
---
.../sink/v2/PaimonRecordEventSerializer.java | 4 +++-
.../paimon/sink/v2/PaimonWriterHelper.java | 22 ++++++++++++------
.../connectors/paimon/sink/v2/TableSchemaInfo.java | 7 ++++++
.../paimon/sink/v2/PaimonSinkITCase.java | 27 ++++++++++++++++++++++
4 files changed, 52 insertions(+), 8 deletions(-)
diff --git a/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 b/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
index d0eceabcd..52c9018ff 100644
--- a/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
+++ b/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 class PaimonRecordEventSerializer implements PaimonRecordSerializer<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(
diff --git a/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 b/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
index 3f515c24a..02bea1b16 100644
--- a/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
+++ b/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 class PaimonWriterHelper {
/** 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,9 +237,13 @@ public class PaimonWriterHelper {
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));
@@ -245,9 +251,11 @@ public class PaimonWriterHelper {
}
case DELETE:
{
- fullGenericRows.add(
- convertRecordDataToGenericRow(
- dataChangeEvent.before(), fieldGetters, RowKind.DELETE));
+ if (hasPrimaryKey) {
+ fullGenericRows.add(
+ convertRecordDataToGenericRow(
+ dataChangeEvent.before(), fieldGetters, RowKind.DELETE));
+ }
break;
}
default:
diff --git a/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 b/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
index 26b4a1b12..ed46baa7a 100644
--- a/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
+++ b/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 class TableSchemaInfo {
public List<RecordData.FieldGetter> getFieldGetters() {
return fieldGetters;
}
+
+ public boolean hasPrimaryKey() {
+ return hasPrimaryKey;
+ }
}
diff --git a/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 b/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
index 2a0286592..aa447ae04 100644
--- a/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
+++ b/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 class PaimonSinkITCase {
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