[spark] Make RowIdIndexFieldsExtractor Serializable

Author: JingsongLi

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java

@@ -45,6 +45,8 @@
 /** Default global index builder. */
 public class DefaultGlobalIndexBuilder implements Serializable {
 
+    private static final long serialVersionUID = 1L;
+
     private final FileStoreTable table;
     private final BinaryRow partition;
     private final RowType readType;

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/RowIdIndexFieldsExtractor.java

@@ -22,37 +22,59 @@
 import org.apache.paimon.codegen.Projection;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.InternalRow.FieldGetter;
 import org.apache.paimon.table.SpecialFields;
 import org.apache.paimon.types.RowType;
 
 import javax.annotation.Nullable;
 
+import java.io.Serializable;
 import java.util.List;
 
 /** The extractor to get partition, index field and row id from records. */
-public class RowIdIndexFieldsExtractor {
+public class RowIdIndexFieldsExtractor implements Serializable {
+
+    private static final long serialVersionUID = 1L;
 
-    private final Projection partitionProjection;
-    private final InternalRow.FieldGetter indexFieldGetter;
     private final int rowIdPos;
+    private final RowType readType;
+    private final List<String> partitionKeys;
+    private final String indexField;
+
+    private transient Projection lazyPartitionProjection;
+    private transient FieldGetter lazyIndexFieldGetter;
 
     public RowIdIndexFieldsExtractor(
             RowType readType, List<String> partitionKeys, String indexField) {
-        this.partitionProjection = CodeGenUtils.newProjection(readType, partitionKeys);
-        int indexFieldPos = readType.getFieldIndex(indexField);
-        this.indexFieldGetter =
-                InternalRow.createFieldGetter(readType.getTypeAt(indexFieldPos), indexFieldPos);
+        this.readType = readType;
+        this.partitionKeys = partitionKeys;
+        this.indexField = indexField;
         this.rowIdPos = readType.getFieldIndex(SpecialFields.ROW_ID.name());
     }
 
+    private Projection partitionProjection() {
+        if (lazyPartitionProjection == null) {
+            lazyPartitionProjection = CodeGenUtils.newProjection(readType, partitionKeys);
+        }
+        return lazyPartitionProjection;
+    }
+
+    private FieldGetter indexFieldGetter() {
+        if (lazyIndexFieldGetter == null) {
+            int indexFieldPos = readType.getFieldIndex(indexField);
+            lazyIndexFieldGetter =
+                    InternalRow.createFieldGetter(readType.getTypeAt(indexFieldPos), indexFieldPos);
+        }
+        return lazyIndexFieldGetter;
+    }
+
     public BinaryRow extractPartition(InternalRow record) {
-        // projection will reuse returning record, copy is necessary
-        return partitionProjection.apply(record).copy();
+        return partitionProjection().apply(record).copy();
     }
 
     @Nullable
     public Object extractIndexField(InternalRow record) {
-        return indexFieldGetter.getFieldOrNull(record);
+        return indexFieldGetter().getFieldOrNull(record);
     }
 
     public Long extractRowId(InternalRow record) {

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/btree/BTreeGlobalIndexBuilder.java

@@ -54,6 +54,8 @@
  */
 public class BTreeGlobalIndexBuilder implements Serializable {
 
+    private static final long serialVersionUID = 1L;
+
     private static final double FLOATING = 1.2;
 
     private final FileStoreTable table;