instantiate converters once

Author: polyzos

fluss-client/src/main/java/org/apache/fluss/client/converter/PojoToRowConverter.java

@@ -188,8 +188,6 @@ private static FieldToRow createFieldConverter(PojoType.Property prop, DataType
         BigDecimal bd = (BigDecimal) v;
         BigDecimal scaled = bd.setScale(scale, RoundingMode.HALF_UP);
 
-        // Validate precision after scaling; precision is the number of digits in the unscaled
-        // value.
         if (scaled.precision() > precision) {
             throw new IllegalArgumentException(
                     String.format(

fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java

@@ -37,7 +37,6 @@ final class PojoType<T> {
     private final Class<T> pojoClass;
     private final Constructor<T> defaultConstructor;
     private final Map<String, Property> properties; // property name -> property
-    // Mapping of primitive types to their boxed counterparts to avoid long if-chains
     private static final Map<Class<?>, Class<?>> PRIMITIVE_TO_BOXED = createPrimitiveToBoxedMap();
 
     private PojoType(Class<T> pojoClass, Constructor<T> ctor, Map<String, Property> props) {

fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedAppendWriter.java

@@ -33,11 +33,13 @@ class TypedAppendWriter<T> implements AppendWriter<T> {
     private final AppendWriterImpl delegate;
     private final Class<T> pojoClass;
     private final RowType tableSchema;
+    private final PojoToRowConverter<T> pojoToRowConverter;
 
     TypedAppendWriter(AppendWriterImpl delegate, Class<T> pojoClass, TableInfo tableInfo) {
         this.delegate = delegate;
         this.pojoClass = pojoClass;
         this.tableSchema = tableInfo.getRowType();
+        this.pojoToRowConverter = PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
     }
 
     @Override
@@ -50,10 +52,7 @@ public CompletableFuture<AppendResult> append(T record) {
         if (record instanceof InternalRow) {
             return delegate.append((InternalRow) record);
         }
-        // TODO: initialize this on the constructor and reuse
-        PojoToRowConverter<T> converter =
-                PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
-        InternalRow row = converter.toRow(record);
+        InternalRow row = pojoToRowConverter.toRow(record);
         return delegate.append(row);
     }
 }

fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedUpsertWriter.java

@@ -23,6 +23,8 @@
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.types.RowType;
 
+import javax.annotation.Nullable;
+
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -42,6 +44,13 @@ public void flush() {
     private final RowType tableSchema;
     private final int[] targetColumns; // may be null
 
+    private final RowType pkProjection;
+    @Nullable private final RowType targetProjection;
+
+    private final PojoToRowConverter<T> pojoToRowConverter;
+    private final PojoToRowConverter<T> pkConverter;
+    @Nullable private final PojoToRowConverter<T> targetConverter;
+
     TypedUpsertWriter(
             UpsertWriterImpl delegate,
             Class<T> pojoClass,
@@ -52,6 +61,19 @@ public void flush() {
         this.tableInfo = tableInfo;
         this.tableSchema = tableInfo.getRowType();
         this.targetColumns = targetColumns;
+
+        // Precompute projections
+        this.pkProjection = this.tableSchema.project(tableInfo.getPhysicalPrimaryKeys());
+        this.targetProjection =
+                (targetColumns == null) ? null : this.tableSchema.project(targetColumns);
+
+        // Initialize reusable converters
+        this.pojoToRowConverter = PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
+        this.pkConverter = PojoToRowConverter.of(pojoClass, tableSchema, pkProjection);
+        this.targetConverter =
+                (targetProjection == null)
+                        ? null
+                        : PojoToRowConverter.of(pojoClass, tableSchema, targetProjection);
     }
 
     @Override
@@ -73,18 +95,19 @@ public CompletableFuture<DeleteResult> delete(T record) {
     }
 
     private InternalRow convertPojo(T pojo, boolean forDelete) {
-        RowType projection;
+        final RowType projection;
+        final PojoToRowConverter<T> converter;
         if (forDelete) {
-            // for delete we only need primary key columns
-            projection = tableSchema.project(tableInfo.getPhysicalPrimaryKeys());
-        } else if (targetColumns != null) {
-            projection = tableSchema.project(targetColumns);
+            projection = pkProjection;
+            converter = pkConverter;
+        } else if (targetProjection != null && targetConverter != null) {
+            projection = targetProjection;
+            converter = targetConverter;
         } else {
             projection = tableSchema;
+            converter = pojoToRowConverter;
         }
 
-        // TODO: initialize this on the constructor and reuse
-        PojoToRowConverter<T> converter = PojoToRowConverter.of(pojoClass, tableSchema, projection);
         GenericRow projected = converter.toRow(pojo);
         if (projection == tableSchema) {
             return projected;