[cdc] Fix database sync performance issue of schema evolution (#5382)

Author: yuzelin

paimon-common/src/main/java/org/apache/paimon/types/FieldIdentifier.java

@@ -22,9 +22,9 @@
 
 /** Used to indicate the uniqueness of a field. */
 public class FieldIdentifier {
-    private String name;
-    private DataType type;
-    private String description;
+    private final String name;
+    private final DataType type;
+    private final String description;
 
     public FieldIdentifier(DataField dataField) {
         this.name = dataField.name();

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/MultiTableUpdatedDataFieldsProcessFunction.java

@@ -25,6 +25,8 @@
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.FieldIdentifier;
 
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.streaming.api.functions.ProcessFunction;
@@ -33,8 +35,11 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 
 /**
  * A {@link ProcessFunction} to handle schema changes. New schema is represented by a {@link
@@ -51,6 +56,8 @@ public class MultiTableUpdatedDataFieldsProcessFunction
 
     private final Map<Identifier, SchemaManager> schemaManagers = new HashMap<>();
 
+    private final Map<Identifier, Set<FieldIdentifier>> latestFieldsMap = new HashMap<>();
+
     public MultiTableUpdatedDataFieldsProcessFunction(
             CatalogLoader catalogLoader, TypeMapping typeMapping) {
         super(catalogLoader, typeMapping);
@@ -73,14 +80,34 @@ public void processElement(
                             }
                             return new SchemaManager(table.fileIO(), table.location());
                         });
-
         if (Objects.isNull(schemaManager)) {
             LOG.error("Failed to get schema manager for table " + tableId);
-        } else {
-            for (SchemaChange schemaChange :
-                    extractSchemaChanges(schemaManager, updatedSchema.f1)) {
-                applySchemaChange(schemaManager, schemaChange, tableId);
-            }
+            return;
+        }
+
+        Set<FieldIdentifier> latestFields =
+                latestFieldsMap.computeIfAbsent(tableId, id -> new HashSet<>());
+        List<DataField> actualUpdatedDataFields =
+                actualUpdatedDataFields(updatedSchema.f1.fields(), latestFields);
+
+        if (actualUpdatedDataFields.isEmpty() && updatedSchema.f1.comment() == null) {
+            return;
+        }
+
+        CdcSchema actualUpdatedSchema =
+                new CdcSchema(
+                        actualUpdatedDataFields,
+                        updatedSchema.f1.primaryKeys(),
+                        updatedSchema.f1.comment());
+
+        for (SchemaChange schemaChange : extractSchemaChanges(schemaManager, actualUpdatedSchema)) {
+            applySchemaChange(schemaManager, schemaChange, tableId);
         }
+        /*
+         * 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.
+         */
+        latestFieldsMap.put(tableId, updateLatestFields(schemaManager));
     }
 }

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunction.java

@@ -25,17 +25,13 @@
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.FieldIdentifier;
-import org.apache.paimon.types.RowType;
 
-import org.apache.commons.collections.CollectionUtils;
 import org.apache.flink.streaming.api.functions.ProcessFunction;
 import org.apache.flink.util.Collector;
 
 import java.util.HashSet;
 import java.util.List;
-import java.util.Objects;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
  * A {@link ProcessFunction} to handle schema changes. New schema is represented by a {@link
@@ -68,12 +64,8 @@ public UpdatedDataFieldsProcessFunction(
     public void processElement(CdcSchema updatedSchema, Context context, Collector<Void> collector)
             throws Exception {
         List<DataField> actualUpdatedDataFields =
-                updatedSchema.fields().stream()
-                        .filter(
-                                dataField ->
-                                        !latestDataFieldContain(new FieldIdentifier(dataField)))
-                        .collect(Collectors.toList());
-        if (CollectionUtils.isEmpty(actualUpdatedDataFields) && updatedSchema.comment() == null) {
+                actualUpdatedDataFields(updatedSchema.fields(), latestFields);
+        if (actualUpdatedDataFields.isEmpty() && updatedSchema.comment() == null) {
             return;
         }
         CdcSchema actualUpdatedSchema =
@@ -89,19 +81,6 @@ public void processElement(CdcSchema updatedSchema, Context context, Collector<V
          * non-SchemaChange.AddColumn scenario. Otherwise, the previously existing fields cannot be
          * modified again.
          */
-        updateLatestFields();
-    }
-
-    private boolean latestDataFieldContain(FieldIdentifier dataField) {
-        return latestFields.stream().anyMatch(previous -> Objects.equals(previous, dataField));
-    }
-
-    private void updateLatestFields() {
-        RowType oldRowType = schemaManager.latest().get().logicalRowType();
-        Set<FieldIdentifier> fieldIdentifiers =
-                oldRowType.getFields().stream()
-                        .map(item -> new FieldIdentifier(item))
-                        .collect(Collectors.toSet());
-        latestFields = fieldIdentifiers;
+        latestFields = updateLatestFields(schemaManager);
     }
 }

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java

@@ -29,6 +29,7 @@
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypeChecks;
 import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.FieldIdentifier;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Preconditions;
 import org.apache.paimon.utils.StringUtils;
@@ -44,6 +45,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 /** Base class for update data fields process function. */
 public abstract class UpdatedDataFieldsProcessFunctionBase<I, O> extends ProcessFunction<I, O> {
@@ -280,6 +283,20 @@ protected List<SchemaChange> extractSchemaChanges(
         return result;
     }
 
+    protected List<DataField> actualUpdatedDataFields(
+            List<DataField> newFields, Set<FieldIdentifier> latestFields) {
+        return newFields.stream()
+                .filter(dataField -> !latestFields.contains(new FieldIdentifier(dataField)))
+                .collect(Collectors.toList());
+    }
+
+    protected Set<FieldIdentifier> updateLatestFields(SchemaManager schemaManager) {
+        RowType oldRowType = schemaManager.latest().get().logicalRowType();
+        return oldRowType.getFields().stream()
+                .map(FieldIdentifier::new)
+                .collect(Collectors.toSet());
+    }
+
     @Override
     public void close() throws Exception {
         if (catalog != null) {