[cdc] Correct equalsIgnoreFieldId in UpdatedDataFieldsProcessFunction…

…Base
JingsongLi · Dec 18, 2024 · f3813f0 · f3813f0
1 parent c93040e
commit f3813f0
Show file tree

Hide file tree

Showing 6 changed files with 25 additions and 19 deletions.
diff --git a/paimon-common/src/main/java/org/apache/paimon/types/ArrayType.java b/paimon-common/src/main/java/org/apache/paimon/types/ArrayType.java
@@ -101,7 +101,7 @@ public boolean equals(Object o) {
     }
 
     @Override
-    public boolean equalsIgnoreFieldId(Object o) {
+    public boolean equalsIgnoreFieldId(DataType o) {
         if (this == o) {
             return true;
         }

diff --git a/paimon-common/src/main/java/org/apache/paimon/types/DataType.java b/paimon-common/src/main/java/org/apache/paimon/types/DataType.java
@@ -23,8 +23,6 @@
 
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonGenerator;
 
-import javax.annotation.Nonnull;
-
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.Arrays;
@@ -124,15 +122,6 @@ public final DataType copy() {
         return copy(isNullable);
     }
 
-    /**
-     * Compare two data types without nullable.
-     *
-     * @param o the target data type
-     */
-    public boolean equalsIgnoreNullable(@Nonnull DataType o) {
-        return Objects.equals(this.copy(true), o.copy(true));
-    }
-
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -145,7 +134,21 @@ public boolean equals(Object o) {
         return isNullable == that.isNullable && typeRoot == that.typeRoot;
     }
 
-    public boolean equalsIgnoreFieldId(Object o) {
+    /**
+     * Compare two data types without nullable.
+     *
+     * @param o the target data type
+     */
+    public boolean equalsIgnoreNullable(DataType o) {
+        return Objects.equals(this.copy(true), o.copy(true));
+    }
+
+    /**
+     * Compare two data types without field id.
+     *
+     * @param o the target data type
+     */
+    public boolean equalsIgnoreFieldId(DataType o) {
         return equals(o);
     }
 

diff --git a/paimon-common/src/main/java/org/apache/paimon/types/MapType.java b/paimon-common/src/main/java/org/apache/paimon/types/MapType.java
@@ -110,7 +110,7 @@ public boolean equals(Object o) {
     }
 
     @Override
-    public boolean equalsIgnoreFieldId(Object o) {
+    public boolean equalsIgnoreFieldId(DataType o) {
         if (this == o) {
             return true;
         }

diff --git a/paimon-common/src/main/java/org/apache/paimon/types/RowType.java b/paimon-common/src/main/java/org/apache/paimon/types/RowType.java
@@ -213,7 +213,8 @@ public boolean equals(Object o) {
         return fields.equals(rowType.fields);
     }
 
-    public boolean equalsIgnoreFieldId(Object o) {
+    @Override
+    public boolean equalsIgnoreFieldId(DataType o) {
         if (this == o) {
             return true;
         }

diff --git a/...-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunction.java b/...-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunction.java
@@ -77,7 +77,7 @@ public void processElement(
                 extractSchemaChanges(schemaManager, actualUpdatedDataFields)) {
             applySchemaChange(schemaManager, schemaChange, identifier);
         }
-        /**
+        /*
          * Here, actualUpdatedDataFields cannot be used to update latestFields because there is a
          * non-SchemaChange.AddColumn scenario. Otherwise, the previously existing fields cannot be
          * modified again.

diff --git a/.../src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java b/.../src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
@@ -219,9 +219,11 @@ protected List<SchemaChange> extractSchemaChanges(
             String newFieldName = StringUtils.toLowerCaseIfNeed(newField.name(), caseSensitive);
             if (oldFields.containsKey(newFieldName)) {
                 DataField oldField = oldFields.get(newFieldName);
-                // we compare by ignoring nullable, because partition keys and primary keys might be
-                // nullable in source database, but they can't be null in Paimon
-                if (oldField.type().equalsIgnoreNullable(newField.type())) {
+                // 1. we compare by ignoring nullable, because partition keys and primary keys might
+                // be nullable in source database, but they can't be null in Paimon
+                // 2. we compare by ignoring field id, the field ID is newly created and may be
+                // different, we should ignore it
+                if (oldField.type().copy(true).equalsIgnoreFieldId(newField.type().copy(true))) {
                     // update column comment
                     if (newField.description() != null
                             && !newField.description().equals(oldField.description())) {