This is an automated email from the ASF dual-hosted git repository. lzljs3620320 pushed a commit to branch release-0.8 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 5802088285f12f70941602f846a8b83e5523b327 Author: chenxi0599 <[email protected]> AuthorDate: Fri May 10 13:56:38 2024 +0800 [cdc] Fixed when the order of the same field differs, it is considered a schema change. (#3314) --- .../apache/paimon/flink/sink/cdc/RichEventParser.java | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/RichEventParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/RichEventParser.java index 756875781..10dbdcc8d 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/RichEventParser.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/RichEventParser.java @@ -45,7 +45,10 @@ public class RichEventParser implements EventParser<RichCdcRecord> { .forEach( dataField -> { DataField previous = previousDataFields.get(dataField.name()); - if (!Objects.equals(previous, dataField)) { + // When the order of the same field is different, its ID may also be + // different, + // so the comparison should not include the ID. + if (!dataFieldEqualsIgnoreId(previous, dataField)) { previousDataFields.put(dataField.name(), dataField); change.add(dataField); } @@ -53,6 +56,18 @@ public class RichEventParser implements EventParser<RichCdcRecord> { return change; } + private boolean dataFieldEqualsIgnoreId(DataField dataField1, DataField dataField2) { + if (dataField1 == dataField2) { + return true; + } else if (dataField1 != null && dataField2 != null) { + return Objects.equals(dataField1.name(), dataField2.name()) + && Objects.equals(dataField1.type(), dataField2.type()) + && Objects.equals(dataField1.description(), dataField2.description()); + } else { + return false; + } + } + @Override public List<CdcRecord> parseRecords() { if (record.hasPayload()) {
