address comments

Author: polyzos

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

@@ -34,6 +34,7 @@
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.OffsetDateTime;
+import java.util.List;
 
 /**
  * Converter for writer path: converts POJO instances to Fluss InternalRow according to a (possibly
@@ -44,12 +45,14 @@ public final class PojoToRowConverter<T> {
     private final PojoType<T> pojoType;
     private final RowType tableSchema;
     private final RowType projection;
+    private final List<String> projectionFieldNames;
     private final FieldToRow[] fieldConverters; // index corresponds to projection position
 
     private PojoToRowConverter(PojoType<T> pojoType, RowType tableSchema, RowType projection) {
         this.pojoType = pojoType;
         this.tableSchema = tableSchema;
         this.projection = projection;
+        this.projectionFieldNames = projection.getFieldNames();
         ConverterCommons.validatePojoMatchesTable(pojoType, tableSchema);
         ConverterCommons.validateProjectionSubset(projection, tableSchema);
         this.fieldConverters = createFieldConverters();
@@ -74,7 +77,7 @@ public GenericRow toRow(@Nullable T pojo) {
             } catch (Exception e) {
                 throw new IllegalStateException(
                         "Failed to access field '"
-                                + projection.getFieldNames().get(i)
+                                + projectionFieldNames.get(i)
                                 + "' from POJO "
                                 + pojoType.getPojoClass().getName(),
                         e);
@@ -87,7 +90,7 @@ public GenericRow toRow(@Nullable T pojo) {
     private FieldToRow[] createFieldConverters() {
         FieldToRow[] arr = new FieldToRow[projection.getFieldCount()];
         for (int i = 0; i < projection.getFieldCount(); i++) {
-            String fieldName = projection.getFieldNames().get(i);
+            String fieldName = projectionFieldNames.get(i);
             DataType fieldType = projection.getTypeAt(i);
             PojoType.Property prop = requireProperty(fieldName);
             ConverterCommons.validateCompatibility(fieldType, prop);

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

@@ -37,6 +37,7 @@
 import java.time.LocalTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.util.List;
 
 /**
  * Converter for scanner path: converts InternalRow (possibly projected) to POJO, leaving
@@ -47,12 +48,14 @@ public final class RowToPojoConverter<T> {
     private final PojoType<T> pojoType;
     private final RowType tableSchema;
     private final RowType projection;
+    private final List<String> projectionFieldNames;
     private final RowToField[] rowReaders;
 
     private RowToPojoConverter(PojoType<T> pojoType, RowType tableSchema, RowType projection) {
         this.pojoType = pojoType;
         this.tableSchema = tableSchema;
         this.projection = projection;
+        this.projectionFieldNames = projection.getFieldNames();
         ConverterCommons.validatePojoMatchesTable(pojoType, tableSchema);
         ConverterCommons.validateProjectionSubset(projection, tableSchema);
         this.rowReaders = createRowReaders();
@@ -73,7 +76,7 @@ public T fromRow(@Nullable InternalRow row) {
                 if (!row.isNullAt(i)) {
                     Object v = rowReaders[i].convert(row, i);
                     PojoType.Property prop =
-                            pojoType.getProperty(projection.getFieldNames().get(i));
+                            pojoType.getProperty(projectionFieldNames.get(i));
                     if (v != null) {
                         prop.write(pojo, v);
                     }
@@ -97,7 +100,7 @@ public T fromRow(@Nullable InternalRow row) {
     private RowToField[] createRowReaders() {
         RowToField[] arr = new RowToField[projection.getFieldCount()];
         for (int i = 0; i < projection.getFieldCount(); i++) {
-            String name = projection.getFieldNames().get(i);
+            String name = projectionFieldNames.get(i);
             DataType type = projection.getTypeAt(i);
             PojoType.Property prop = requireProperty(name);
             ConverterCommons.validateCompatibility(type, prop);
@@ -147,10 +150,14 @@ private static RowToField createRowReader(DataType fieldType, PojoType.Property
                 return RowToPojoConverter::convertDateValue;
             case TIME_WITHOUT_TIME_ZONE:
                 return RowToPojoConverter::convertTimeValue;
-            case TIMESTAMP_WITHOUT_TIME_ZONE:
-                return (row, pos) -> convertTimestampNtzValue(fieldType, row, pos);
-            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-                return (row, pos) -> convertTimestampLtzValue(fieldType, prop, row, pos);
+            case TIMESTAMP_WITHOUT_TIME_ZONE: {
+                final int precision = DataTypeChecks.getPrecision(fieldType);
+                return (row, pos) -> convertTimestampNtzValue(precision, row, pos);
+            }
+            case TIMESTAMP_WITH_LOCAL_TIME_ZONE: {
+                final int precision = DataTypeChecks.getPrecision(fieldType);
+                return (row, pos) -> convertTimestampLtzValue(precision, prop, row, pos);
+            }
             default:
                 throw new UnsupportedOperationException(
                         String.format(
@@ -224,8 +231,7 @@ private static LocalTime convertTimeValue(InternalRow row, int pos) {
     }
 
     /** Converts a TIMESTAMP_WITHOUT_TIME_ZONE value to a LocalDateTime honoring precision. */
-    private static Object convertTimestampNtzValue(DataType fieldType, InternalRow row, int pos) {
-        final int precision = DataTypeChecks.getPrecision(fieldType);
+    private static Object convertTimestampNtzValue(int precision, InternalRow row, int pos) {
         TimestampNtz t = row.getTimestampNtz(pos, precision);
         return t.toLocalDateTime();
     }
@@ -235,8 +241,7 @@ private static Object convertTimestampNtzValue(DataType fieldType, InternalRow r
      * depending on the target POJO property type.
      */
     private static Object convertTimestampLtzValue(
-            DataType fieldType, PojoType.Property prop, InternalRow row, int pos) {
-        final int precision = DataTypeChecks.getPrecision(fieldType);
+            int precision, PojoType.Property prop, InternalRow row, int pos) {
         TimestampLtz t = row.getTimestampLtz(pos, precision);
         if (prop.type == Instant.class) {
             return t.toInstant();