Repository: apache/iceberg (language: python)

Issue:
Flink: Pre-create fieldGetters to avoid constructing them for each row
I found that RowDataUtil.clone consumes a significant amount of CPU time by analyzing the flame graph. It constructs a FieldGetter object for each field of every row being processed. This commit optimizes this by constructing all FieldGetter objects upfront, avoiding repeated object creation.


Discussion / hints:
In my case, the average task execution time decreased from 344 seconds to 328 seconds.

![reader](https://github.com/apache/iceberg/assets/16379560/2521b016-3965-4818-b10a-89aadbe4775d)

Hi @nastra , the failing test cases seem unrelated to this change. Could you please re-trigger the tests? Thanks!
@fengjiajie Rerunning the failed tests 👍 
Thanks for the detailed explanation and the PR @fengjiajie!

Left a few small comments.

As a general rule we create a PR for a single version. Usually the latest. So if there are any review comments, they need to be fixed only once.
Once the first PR is merged, we create a backport PR for the other supported versions.
> Thanks for the detailed explanation and the PR @fengjiajie!
> 
> Left a few small comments.
> 
> As a general rule we create a PR for a single version. Usually the latest. So if there are any review comments, they need to be fixed only once. Once the first PR is merged, we create a backport PR for the other supported versions.

@pvary  Thanks for taking the time to review this! Please let me know if you have any suggestions.
> +1 from me. Let's wait a bit, to see if someone else is interested in reviewing. If there're no more comments, I will merge next Tuesday. (Ping me if I forgot 😄)

Thanks
Thanks @fengjiajie for the improvement.
Could you please create a backport PR to the other Flink versions?

Fix patch:
diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/RowDataUtil.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/RowDataUtil.java
index c5cb51b7eae4..3a8f5ccc6c03 100644
--- a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/RowDataUtil.java
+++ b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/RowDataUtil.java
@@ -79,22 +79,45 @@ public static Object convertConstant(Type type, Object value) {
    * the arity check.
    */
   public static RowData clone(
-      RowData from, RowData reuse, RowType rowType, TypeSerializer[] fieldSerializers) {
+      RowData from,
+      RowData reuse,
+      RowType rowType,
+      TypeSerializer[] fieldSerializers,
+      RowData.FieldGetter[] fieldGetters) {
     GenericRowData ret;
     if (reuse instanceof GenericRowData) {
       ret = (GenericRowData) reuse;
     } else {
       ret = new GenericRowData(from.getArity());
     }
+
     ret.setRowKind(from.getRowKind());
     for (int i = 0; i < rowType.getFieldCount(); i++) {
       if (!from.isNullAt(i)) {
-        RowData.FieldGetter getter = RowData.createFieldGetter(rowType.getTypeAt(i), i);
-        ret.setField(i, fieldSerializers[i].copy(getter.getFieldOrNull(from)));
+        ret.setField(i, fieldSerializers[i].copy(fieldGetters[i].getFieldOrNull(from)));
       } else {
         ret.setField(i, null);
       }
     }
+
     return ret;
   }
+
+  /**
+   * @deprecated will be removed in 1.7.0; Not reusing FieldGetter in this method could lead to
+   *     performance degradation, use {@link #clone(RowData, RowData, RowType, TypeSerializer[],
+   *     RowData.FieldGetter[])} instead.
+   */
+  @Deprecated
+  public static RowData clone(
+      RowData from, RowData reuse, RowType rowType, TypeSerializer[] fieldSerializers) {
+    RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[rowType.getFieldCount()];
+    for (int i = 0; i < rowType.getFieldCount(); ++i) {
+      if (!from.isNullAt(i)) {
+        fieldGetters[i] = RowData.createFieldGetter(rowType.getTypeAt(i), i);
+      }
+    }
+
+    return clone(from, reuse, rowType, fieldSerializers, fieldGetters);
+  }
 }
diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataRecordFactory.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataRecordFactory.java
index 1e265b2663ce..40d5c28d7bc7 100644
--- a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataRecordFactory.java
+++ b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataRecordFactory.java
@@ -28,10 +28,12 @@
 class RowDataRecordFactory implements RecordFactory<RowData> {
   private final RowType rowType;
   private final TypeSerializer[] fieldSerializers;
+  private final RowData.FieldGetter[] fieldGetters;
 
   RowDataRecordFactory(RowType rowType) {
     this.rowType = rowType;
     this.fieldSerializers = createFieldSerializers(rowType);
+    this.fieldGetters = createFieldGetters(rowType);
   }
 
   static TypeSerializer[] createFieldSerializers(RowType rowType) {
@@ -40,6 +42,15 @@ static TypeSerializer[] createFieldSerializers(RowType rowType) {
         .toArray(TypeSerializer[]::new);
   }
 
+  static RowData.FieldGetter[] createFieldGetters(RowType rowType) {
+    RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[rowType.getFieldCount()];
+    for (int i = 0; i < rowType.getFieldCount(); ++i) {
+      fieldGetters[i] = RowData.createFieldGetter(rowType.getTypeAt(i), i);
+    }
+
+    return fieldGetters;
+  }
+
   @Override
   public RowData[] createBatch(int batchSize) {
     RowData[] arr = new RowData[batchSize];
@@ -56,6 +67,7 @@ public void clone(RowData from, RowData[] batch, int position) {
     // Clone method will allocate a new GenericRowData object
     // if the target object is NOT a GenericRowData.
     // So we should always set the clone return value back to the array.
-    batch[position] = RowDataUtil.clone(from, batch[position], rowType, fieldSerializers);
+    batch[position] =
+        RowDataUtil.clone(from, batch[position], rowType, fieldSerializers, fieldGetters);
   }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/RowDataUtil.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/RowDataUtil.java
index c5cb51b7eae4..3a8f5ccc6c03 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/RowDataUtil.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/RowDataUtil.java
@@ -79,22 +79,45 @@ public static Object convertConstant(Type type, Object value) {
    * the arity check.
    */
   public static RowData clone(
-      RowData from, RowData reuse, RowType rowType, TypeSerializer[] fieldSerializers) {
+      RowData from,
+      RowData reuse,
+      RowType rowType,
+      TypeSerializer[] fieldSerializers,
+      RowData.FieldGetter[] fieldGetters) {
     GenericRowData ret;
     if (reuse instanceof GenericRowData) {
       ret = (GenericRowData) reuse;
     } else {
       ret = new GenericRowData(from.getArity());
     }
+
     ret.setRowKind(from.getRowKind());
     for (int i = 0; i < rowType.getFieldCount(); i++) {
       if (!from.isNullAt(i)) {
-        RowData.FieldGetter getter = RowData.createFieldGetter(rowType.getTypeAt(i), i);
-        ret.setField(i, fieldSerializers[i].copy(getter.getFieldOrNull(from)));
+        ret.setField(i, fieldSerializers[i].copy(fieldGetters[i].getFieldOrNull(from)));
       } else {
         ret.setField(i, null);
       }
     }
+
     return ret;
   }
+
+  /**
+   * @deprecated will be removed in 1.7.0; Not reusing FieldGetter in this method could lead to
+   *     performance degradation, use {@link #clone(RowData, RowData, RowType, TypeSerializer[],
+   *     RowData.FieldGetter[])} instead.
+   */
+  @Deprecated
+  public static RowData clone(
+      RowData from, RowData reuse, RowType rowType, TypeSerializer[] fieldSerializers) {
+    RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[rowType.getFieldCount()];
+    for (int i = 0; i < rowType.getFieldCount(); ++i) {
+      if (!from.isNullAt(i)) {
+        fieldGetters[i] = RowData.createFieldGetter(rowType.getTypeAt(i), i);
+      }
+    }
+
+    return clone(from, reuse, rowType, fieldSerializers, fieldGetters);
+  }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataRecordFactory.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataRecordFactory.java
index 1e265b2663ce..40d5c28d7bc7 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataRecordFactory.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataRecordFactory.java
@@ -28,10 +28,12 @@
 class RowDataRecordFactory implements RecordFactory<RowData> {
   private final RowType rowType;
   private final TypeSerializer[] fieldSerializers;
+  private final RowData.FieldGetter[] fieldGetters;
 
   RowDataRecordFactory(RowType rowType) {
     this.rowType = rowType;
     this.fieldSerializers = createFieldSerializers(rowType);
+    this.fieldGetters = createFieldGetters(rowType);
   }
 
   static TypeSerializer[] createFieldSerializers(RowType rowType) {
@@ -40,6 +42,15 @@ static TypeSerializer[] createFieldSerializers(RowType rowType) {
         .toArray(TypeSerializer[]::new);
   }
 
+  static RowData.FieldGetter[] createFieldGetters(RowType rowType) {
+    RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[rowType.getFieldCount()];
+    for (int i = 0; i < rowType.getFieldCount(); ++i) {
+      fieldGetters[i] = RowData.createFieldGetter(rowType.getTypeAt(i), i);
+    }
+
+    return fieldGetters;
+  }
+
   @Override
   public RowData[] createBatch(int batchSize) {
     RowData[] arr = new RowData[batchSize];
@@ -56,6 +67,7 @@ public void clone(RowData from, RowData[] batch, int position) {
     // Clone method will allocate a new GenericRowData object
     // if the target object is NOT a GenericRowData.
     // So we should always set the clone return value back to the array.
-    batch[position] = RowDataUtil.clone(from, batch[position], rowType, fieldSerializers);
+    batch[position] =
+        RowDataUtil.clone(from, batch[position], rowType, fieldSerializers, fieldGetters);
   }
 }


Test patch:
diff --git a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
index a243be5834a8..8cebf950c5f0 100644
--- a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
+++ b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
@@ -87,7 +87,12 @@ public static RowData copyRowData(RowData from, RowType rowType) {
         rowType.getChildren().stream()
             .map((LogicalType type) -> InternalSerializers.create(type))
             .toArray(TypeSerializer[]::new);
-    return RowDataUtil.clone(from, null, rowType, fieldSerializers);
+    RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[rowType.getFieldCount()];
+    for (int i = 0; i < rowType.getFieldCount(); ++i) {
+      fieldGetters[i] = RowData.createFieldGetter(rowType.getTypeAt(i), i);
+    }
+
+    return RowDataUtil.clone(from, null, rowType, fieldSerializers, fieldGetters);
   }
 
   public static void readRowData(FlinkInputFormat input, Consumer<RowData> visitor)
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
index a243be5834a8..8cebf950c5f0 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java
@@ -87,7 +87,12 @@ public static RowData copyRowData(RowData from, RowType rowType) {
         rowType.getChildren().stream()
             .map((LogicalType type) -> InternalSerializers.create(type))
             .toArray(TypeSerializer[]::new);
-    return RowDataUtil.clone(from, null, rowType, fieldSerializers);
+    RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[rowType.getFieldCount()];
+    for (int i = 0; i < rowType.getFieldCount(); ++i) {
+      fieldGetters[i] = RowData.createFieldGetter(rowType.getTypeAt(i), i);
+    }
+
+    return RowDataUtil.clone(from, null, rowType, fieldSerializers, fieldGetters);
   }
 
   public static void readRowData(FlinkInputFormat input, Consumer<RowData> visitor)