Use flat List for projection column lookup instead of HashMap (should be faster)

Author: yuxiqian

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformProcessor.java

@@ -31,7 +31,6 @@
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -57,7 +56,7 @@ public class PostTransformProcessor {
     private @Nullable TransformFilter transformFilter;
     private String timezone;
     private Map<String, ProjectionColumnProcessor> projectionColumnProcessorMap;
-    private Map<String, ProjectionColumn> cachedProjectionColumns;
+    private List<ProjectionColumn> cachedProjectionColumns;
 
     public PostTransformProcessor(
             TableInfo tableInfo,
@@ -117,9 +116,11 @@ public Schema processSchemaChangeEvent(Schema schema) {
 
     public BinaryRecordData processData(BinaryRecordData payload, long epochTime) {
         List<Object> valueList = new ArrayList<>();
-        for (Column column : tableInfo.getSchema().getColumns()) {
-            ProjectionColumn projectionColumn = cachedProjectionColumns.get(column.getName());
-            if (cachedProjectionColumns.containsKey(column.getName())) {
+        List<Column> columns = tableInfo.getSchema().getColumns();
+
+        for (int i = 0; i < columns.size(); i++) {
+            if (cachedProjectionColumns.get(i) != null) {
+                ProjectionColumn projectionColumn = cachedProjectionColumns.get(i);
                 projectionColumnProcessorMap.putIfAbsent(
                         projectionColumn.getColumnName(),
                         ProjectionColumnProcessor.of(tableInfo, projectionColumn, timezone));
@@ -130,6 +131,7 @@ public BinaryRecordData processData(BinaryRecordData payload, long epochTime) {
                                 projectionColumnProcessor.evaluate(payload, epochTime),
                                 projectionColumn.getDataType()));
             } else {
+                Column column = columns.get(i);
                 valueList.add(
                         getValueFromBinaryRecordData(
                                 column.getName(),
@@ -139,6 +141,7 @@ public BinaryRecordData processData(BinaryRecordData payload, long epochTime) {
                                 tableInfo.getOriginalFieldGetters()));
             }
         }
+
         return tableInfo.getRecordDataGenerator().generate(valueList.toArray(new Object[0]));
     }
 
@@ -157,23 +160,25 @@ private Object getValueFromBinaryRecordData(
         return null;
     }
 
-    private Map<String, ProjectionColumn> cacheProjectionColumnMap(
+    private List<ProjectionColumn> cacheProjectionColumnMap(
             TableInfo tableInfo, TransformProjection transformProjection) {
-        Map<String, ProjectionColumn> cachedMap = new HashMap<>();
+        List<ProjectionColumn> cachedProjectionColumns = new ArrayList<>();
         if (!hasTableInfo()) {
-            return cachedMap;
+            return cachedProjectionColumns;
         }
 
         for (Column column : tableInfo.getSchema().getColumns()) {
+            ProjectionColumn matchedProjectionColumn = null;
             for (ProjectionColumn projectionColumn : transformProjection.getProjectionColumns()) {
                 if (column.getName().equals(projectionColumn.getColumnName())
                         && projectionColumn.isValidTransformedProjectionColumn()) {
-                    cachedMap.put(column.getName(), projectionColumn);
+                    matchedProjectionColumn = projectionColumn;
                     break;
                 }
             }
+            cachedProjectionColumns.add(matchedProjectionColumn);
         }
 
-        return cachedMap;
+        return cachedProjectionColumns;
     }
 }

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformProcessor.java

@@ -28,9 +28,7 @@
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 
 /**
  * The processor of pre-transform projection in {@link PreTransformOperator}.
@@ -47,7 +45,7 @@ public class PreTransformProcessor {
     private TableChangeInfo tableChangeInfo;
     private TransformProjection transformProjection;
     private @Nullable TransformFilter transformFilter;
-    private Set<String> cachedProjectionColumns;
+    private List<Boolean> cachedProjectionColumnsState;
 
     public PreTransformProcessor(
             TableChangeInfo tableChangeInfo,
@@ -56,7 +54,7 @@ public PreTransformProcessor(
         this.tableChangeInfo = tableChangeInfo;
         this.transformProjection = transformProjection;
         this.transformFilter = transformFilter;
-        this.cachedProjectionColumns =
+        this.cachedProjectionColumnsState =
                 cacheIsProjectionColumnMap(tableChangeInfo, transformProjection);
     }
 
@@ -82,19 +80,21 @@ public CreateTableEvent preTransformCreateTableEvent(CreateTableEvent createTabl
 
     public BinaryRecordData processFillDataField(BinaryRecordData data) {
         List<Object> valueList = new ArrayList<>();
-        for (Column column : tableChangeInfo.getTransformedSchema().getColumns()) {
-            if (cachedProjectionColumns.contains(column.getName())) {
+        List<Column> columns = tableChangeInfo.getTransformedSchema().getColumns();
+
+        for (int i = 0; i < columns.size(); i++) {
+            if (cachedProjectionColumnsState.get(i)) {
                 valueList.add(null);
-                break;
             } else {
                 valueList.add(
                         getValueFromBinaryRecordData(
-                                column.getName(),
+                                columns.get(i).getName(),
                                 data,
                                 tableChangeInfo.getOriginalSchema().getColumns(),
                                 tableChangeInfo.getFieldGetters()));
             }
         }
+
         return tableChangeInfo.getRecordDataGenerator().generate(valueList.toArray(new Object[0]));
     }
 
@@ -112,21 +112,23 @@ private Object getValueFromBinaryRecordData(
         return null;
     }
 
-    private Set<String> cacheIsProjectionColumnMap(
+    private List<Boolean> cacheIsProjectionColumnMap(
             TableChangeInfo tableChangeInfo, TransformProjection transformProjection) {
-        Set<String> cachedMap = new HashSet<>();
+        List<Boolean> cachedMap = new ArrayList<>();
         if (!hasTableChangeInfo()) {
             return cachedMap;
         }
 
         for (Column column : tableChangeInfo.getTransformedSchema().getColumns()) {
+            boolean isProjectionColumn = false;
             for (ProjectionColumn projectionColumn : transformProjection.getProjectionColumns()) {
                 if (column.getName().equals(projectionColumn.getColumnName())
                         && projectionColumn.isValidTransformedProjectionColumn()) {
-                    cachedMap.add(column.getName());
+                    isProjectionColumn = true;
                     break;
                 }
             }
+            cachedMap.add(isProjectionColumn);
         }
 
         return cachedMap;