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 2558209c55 [cdc] Pair canal UPDATE rows by position instead of by
content (#10370)
2558209c55 is described below
commit 2558209c5525c17d585ec2faee9201a572c4e758
Author: jackylee <[email protected]>
AuthorDate: Mon Oct 5 20:45:16 2026 +0800
[cdc] Pair canal UPDATE rows by position instead of by content (#10370)
---
.../action/cdc/format/canal/CanalRecordParser.java | 18 ++---
.../cdc/format/canal/CanalRecordParserTest.java | 84 ++++++++++++++++++++++
2 files changed, 90 insertions(+), 12 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
index 21d5935aac..034e6734dd 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
@@ -46,8 +46,6 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
import static org.apache.paimon.utils.JsonSerdeUtil.getNodeAs;
import static org.apache.paimon.utils.JsonSerdeUtil.isNull;
@@ -101,15 +99,17 @@ public class CanalRecordParser extends
AbstractJsonRecordParser {
String type = getAndCheck(FIELD_TYPE).asText();
- for (JsonNode data : arrayData) {
+ for (int i = 0; i < arrayData.size(); i++) {
+ JsonNode data = arrayData.get(i);
switch (type) {
case OP_UPDATE:
ArrayNode oldArrayData = getNodeAs(root, FIELD_OLD,
ArrayNode.class);
checkNotNull(oldArrayData, FIELD_OLD, FIELD_TYPE, type);
- Map<JsonNode, JsonNode> matchedOldRecords =
- matchOldRecords(arrayData, oldArrayData);
- JsonNode old = matchedOldRecords.get(data);
+ // Pair the i-th new row with the i-th old row by
position. The previous
+ // content-keyed lookup threw IllegalStateException on a
batch that held two
+ // equal row images (natural for a table without a primary
key).
+ JsonNode old = oldArrayData.get(i);
processRecord(mergeOldRecord(data, old), RowKind.DELETE,
records);
processRecord(data, RowKind.INSERT, records);
break;
@@ -190,12 +190,6 @@ public class CanalRecordParser extends
AbstractJsonRecordParser {
return "canal-json";
}
- private Map<JsonNode, JsonNode> matchOldRecords(ArrayNode newData,
ArrayNode oldData) {
- return IntStream.range(0, newData.size())
- .boxed()
- .collect(Collectors.toMap(newData::get, oldData::get));
- }
-
private String transformValue(@Nullable String oldValue, String shortType,
String mySqlType) {
if (oldValue == null) {
return null;
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParserTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParserTest.java
new file mode 100644
index 0000000000..3447ba240d
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParserTest.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.format.canal;
+
+import org.apache.paimon.flink.action.cdc.CdcSourceRecord;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+import org.apache.paimon.types.RowKind;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link CanalRecordParser}. */
+public class CanalRecordParserTest {
+
+ /**
+ * A Canal {@code UPDATE} event may carry two equal row images in one
batch (natural for a table
+ * without a primary key, or a bulk update that sets several rows to the
same values). The old
+ * positional pairing was built with {@code Collectors.toMap} keyed by the
row node, so equal
+ * images collided and the parser threw {@code IllegalStateException:
Duplicate key}. Pairing by
+ * position must instead emit a DELETE + INSERT for every row.
+ */
+ @Test
+ public void testUpdateWithDuplicateRowImagesPairsByPosition() throws
Exception {
+ String json =
+ "{"
+ + "\"database\":\"test_db\",\"table\":\"test_table\","
+ +
"\"pkNames\":[\"id\"],\"isDdl\":false,\"type\":\"UPDATE\","
+ +
"\"mysqlType\":{\"id\":\"int\",\"name\":\"varchar(20)\"},"
+ +
"\"data\":[{\"id\":\"1\",\"name\":\"x\"},{\"id\":\"1\",\"name\":\"x\"}],"
+ + "\"old\":[{\"name\":\"a\"},{\"name\":\"b\"}]"
+ + "}";
+
+ CanalRecordParser parser =
+ new CanalRecordParser(TypeMapping.defaultMapping(),
Collections.emptyList());
+ JsonNode rootNode = new ObjectMapper().readValue(json, JsonNode.class);
+ CdcSourceRecord cdcRecord = new CdcSourceRecord(rootNode);
+ parser.buildSchema(cdcRecord);
+
+ List<RichCdcMultiplexRecord> records = parser.extractRecords();
+
+ // Two updated rows, each emits a DELETE (old image) + an INSERT (new
image).
+ assertThat(records).hasSize(4);
+ assertThat(
+ records.stream()
+ .filter(
+ r ->
+
r.toRichCdcRecord().toCdcRecord().kind()
+ == RowKind.DELETE)
+ .count())
+ .isEqualTo(2);
+ assertThat(
+ records.stream()
+ .filter(
+ r ->
+
r.toRichCdcRecord().toCdcRecord().kind()
+ == RowKind.INSERT)
+ .count())
+ .isEqualTo(2);
+ }
+}
