address comments

Author: polyzos

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java

@@ -163,7 +163,7 @@ public ScanRecords poll(Duration timeout) {
                 }
             } while (System.nanoTime() - startNanos < timeoutNanos);
 
-            return ScanRecords.empty();
+            return ScanRecords.EMPTY;
         } finally {
             release();
             scannerMetricGroup.recordPollEnd(System.currentTimeMillis());

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java

@@ -37,9 +37,7 @@
  */
 @PublicEvolving
 public class ScanRecords implements Iterable<ScanRecord> {
-    public static final ScanRecords empty() {
-        return new ScanRecords(Collections.emptyMap());
-    }
+    public static final ScanRecords EMPTY = new ScanRecords(Collections.emptyMap());
 
     private final Map<TableBucket, List<ScanRecord>> records;
 

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

@@ -41,7 +41,6 @@ public AppendWriter createWriter() {
 
     @Override
     public <T> TypedAppendWriter<T> createTypedWriter(Class<T> pojoClass) {
-        AppendWriterImpl delegate = new AppendWriterImpl(tablePath, tableInfo, writerClient);
-        return new TypedAppendWriterImpl<>(delegate, pojoClass, tableInfo);
+        return new TypedAppendWriterImpl<>(createWriter(), pojoClass, tableInfo);
     }
 }

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

@@ -98,8 +98,6 @@ public UpsertWriter createWriter() {
 
     @Override
     public <T> TypedUpsertWriter<T> createTypedWriter(Class<T> pojoClass) {
-        UpsertWriterImpl delegate =
-                new UpsertWriterImpl(tablePath, tableInfo, targetColumns, writerClient);
-        return new TypedUpsertWriterImpl<>(delegate, pojoClass, tableInfo, targetColumns);
+        return new TypedUpsertWriterImpl<>(createWriter(), pojoClass, tableInfo, targetColumns);
     }
 }

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

@@ -28,7 +28,7 @@
  * @since 0.1
  */
 @PublicEvolving
-public interface TableWriter extends AutoCloseable {
+public interface TableWriter {
 
     /**
      * Flush data written that have not yet been sent to the server, forcing the client to send the
@@ -38,9 +38,4 @@ public interface TableWriter extends AutoCloseable {
      * results in an error.
      */
     void flush();
-
-    @Override
-    default void close() throws Exception {
-        // by default do nothing
-    }
 }

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

@@ -30,14 +30,12 @@
  */
 class TypedAppendWriterImpl<T> implements TypedAppendWriter<T> {
 
-    private final AppendWriterImpl delegate;
-    private final Class<T> pojoClass;
+    private final AppendWriter delegate;
     private final RowType tableSchema;
     private final PojoToRowConverter<T> pojoToRowConverter;
 
-    TypedAppendWriterImpl(AppendWriterImpl delegate, Class<T> pojoClass, TableInfo tableInfo) {
+    TypedAppendWriterImpl(AppendWriter delegate, Class<T> pojoClass, TableInfo tableInfo) {
         this.delegate = delegate;
-        this.pojoClass = pojoClass;
         this.tableSchema = tableInfo.getRowType();
         this.pojoToRowConverter = PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
     }
@@ -47,11 +45,6 @@ public void flush() {
         delegate.flush();
     }
 
-    @Override
-    public void close() throws Exception {
-        delegate.close();
-    }
-
     @Override
     public CompletableFuture<AppendResult> append(T record) {
         if (record instanceof InternalRow) {

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

@@ -33,22 +33,10 @@
  */
 class TypedUpsertWriterImpl<T> implements TypedUpsertWriter<T> {
 
-    @Override
-    public void flush() {
-        delegate.flush();
-    }
-
-    private final UpsertWriterImpl delegate;
-
-    @Override
-    public void close() throws Exception {
-        delegate.close();
-    }
-
-    private final Class<T> pojoClass;
+    private final UpsertWriter delegate;
     private final TableInfo tableInfo;
     private final RowType tableSchema;
-    private final int[] targetColumns; // may be null
+    @Nullable private final int[] targetColumns;
 
     private final RowType pkProjection;
     @Nullable private final RowType targetProjection;
@@ -58,12 +46,8 @@ public void close() throws Exception {
     @Nullable private final PojoToRowConverter<T> targetConverter;
 
     TypedUpsertWriterImpl(
-            UpsertWriterImpl delegate,
-            Class<T> pojoClass,
-            TableInfo tableInfo,
-            int[] targetColumns) {
+            UpsertWriter delegate, Class<T> pojoClass, TableInfo tableInfo, int[] targetColumns) {
         this.delegate = delegate;
-        this.pojoClass = pojoClass;
         this.tableInfo = tableInfo;
         this.tableSchema = tableInfo.getRowType();
         this.targetColumns = targetColumns;
@@ -82,6 +66,11 @@ public void close() throws Exception {
                         : PojoToRowConverter.of(pojoClass, tableSchema, targetProjection);
     }
 
+    @Override
+    public void flush() {
+        delegate.flush();
+    }
+
     @Override
     public CompletableFuture<UpsertResult> upsert(T record) {
         if (record instanceof InternalRow) {
@@ -124,6 +113,9 @@ private InternalRow convertPojo(T pojo, boolean forDelete) {
             // set PK fields, others null
             for (String pk : tableInfo.getPhysicalPrimaryKeys()) {
                 int projIndex = projection.getFieldIndex(pk);
+
+                // TODO: this can be optimized by pre-computing
+                // the index mapping in the constructor?
                 int fullIndex = tableSchema.getFieldIndex(pk);
                 full.setField(fullIndex, projected.getField(projIndex));
             }