1. modified based on CR

2. flink write with smaller row

Author: loserwang1024

fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java

@@ -112,7 +112,7 @@ public Table getTable(TablePath tablePath, SchemaInfo schemaInfo) {
         metadataUpdater.updateTableOrPartitionMetadata(tablePath, null);
         Admin admin = getOrCreateAdmin();
         TableInfo tableInfo = admin.getTableInfo(tablePath).join();
-        return new FlussTable(this, tablePath, TableInfo.of(tableInfo, schemaInfo));
+        return new FlussTable(this, tablePath, tableInfo.withNewSchema(schemaInfo));
     }
 
     public MetadataUpdater getMetadataUpdater() {

fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixKeyLookuper.java

@@ -27,7 +27,7 @@
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.row.InternalRow;
-import org.apache.fluss.row.PruneRow;
+import org.apache.fluss.row.ProjectedRow;
 import org.apache.fluss.row.encode.KeyEncoder;
 import org.apache.fluss.row.encode.ValueDecoder;
 import org.apache.fluss.types.RowType;
@@ -185,7 +185,7 @@ public CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {
                                 } else {
                                     Schema schema = schemaGetter.getSchema(value.schemaId);
                                     row =
-                                            PruneRow.from(schema, tableInfo.getSchema())
+                                            ProjectedRow.from(schema, tableInfo.getSchema())
                                                     .replaceRow(value.row);
                                 }
                                 rowList.add(row);

fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java

@@ -27,7 +27,7 @@
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.row.InternalRow;
-import org.apache.fluss.row.PruneRow;
+import org.apache.fluss.row.ProjectedRow;
 import org.apache.fluss.row.encode.KeyEncoder;
 import org.apache.fluss.row.encode.ValueDecoder;
 import org.apache.fluss.types.RowType;
@@ -132,7 +132,7 @@ public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {
 
         int bucketId = bucketingFunction.bucketing(bkBytes, numBuckets);
         TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), partitionId, bucketId);
-        return lookupClient
+         lookupClient
                 .lookup(tableBucket, pkBytes)
                 .thenApply(
                         valueBytes -> {
@@ -145,7 +145,7 @@ public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {
                                     Schema schema = schemaGetter.getSchema(value.schemaId);
                                     checkNotNull(schema, "schema is null");
                                     row =
-                                            PruneRow.from(schema, tableInfo.getSchema())
+                                            ProjectedRow.from(schema, tableInfo.getSchema())
                                                     .replaceRow(value.row);
                                 }
                             }

fluss-client/src/main/java/org/apache/fluss/client/metadata/ClientSchemaGetter.java

@@ -27,17 +27,17 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.fluss.utils.MapUtils.newConcurrentHashMap;
+
 /** Schema getter for client. */
 @Internal
 public class ClientSchemaGetter implements SchemaGetter {
     private static final Logger LOG = LoggerFactory.getLogger(ClientSchemaGetter.class);
 
     private final TablePath tablePath;
-    // todo: 改为cache.
     private final Map<Integer, Schema> schemasById;
     private final Admin admin;
     private SchemaInfo latestSchemaInfo;
@@ -46,7 +46,7 @@ public ClientSchemaGetter(TablePath tablePath, SchemaInfo latestSchemaInfo, Admi
         this.tablePath = tablePath;
         this.latestSchemaInfo = latestSchemaInfo;
         this.admin = admin;
-        this.schemasById = new HashMap<>();
+        this.schemasById = newConcurrentHashMap();
         schemasById.put(latestSchemaInfo.getSchemaId(), latestSchemaInfo.getSchema());
     }
 
@@ -56,9 +56,8 @@ public Schema getSchema(int schemaId) {
                 schemaId,
                 (id) -> {
                     try {
-                        // todo: 测试这一步总是会一定概率卡住(有时候卡几分钟恢复），这个可以排查一下。
                         SchemaInfo schemaInfo =
-                                admin.getTableSchema(tablePath, schemaId).get(5, TimeUnit.MINUTES);
+                                admin.getTableSchema(tablePath, schemaId).get(1, TimeUnit.MINUTES);
                         if (id > latestSchemaInfo.getSchemaId()) {
                             latestSchemaInfo = schemaInfo;
                         }
@@ -77,11 +76,5 @@ public SchemaInfo getLatestSchemaInfo() {
     }
 
     @Override
-    public void release() {
-        try {
-            admin.close();
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
+    public void release() {}
 }

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java

@@ -121,6 +121,16 @@ public KvSnapshotBatchScanner(
         initReaderAsynchronously();
     }
 
+    /**
+     * Fetch data from snapshot.
+     *
+     * <p>If the snapshot file reader is not ready in given maximum block time, return an empty
+     * iterator. If the reader is ready, always return the reader if there remains any record in the
+     * reader, otherwise, return null.
+     *
+     * @param timeout The maximum time to block (must not be greater than {@link Long#MAX_VALUE}
+     *     milliseconds)
+     */
     @Nullable
     @Override
     public CloseableIterator<InternalRow> pollBatch(Duration timeout) throws IOException {

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java

@@ -34,7 +34,6 @@
 import org.apache.fluss.row.GenericRow;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.ProjectedRow;
-import org.apache.fluss.row.PruneRow;
 import org.apache.fluss.rpc.gateway.TabletServerGateway;
 import org.apache.fluss.rpc.messages.LimitScanRequest;
 import org.apache.fluss.rpc.messages.LimitScanResponse;
@@ -144,7 +143,7 @@ private List<InternalRow> parseLimitScanResponse(LimitScanResponse limitScanResp
                 InternalRow row = record.getRow();
                 if (targetSchemaId != record.schemaId()) {
                     row =
-                            PruneRow.from(
+                            ProjectedRow.from(
                                             schemaGetter.getSchema(record.schemaId()),
                                             schemaGetter.getSchema(targetSchemaId))
                                     .replaceRow(row);

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java

@@ -26,7 +26,6 @@
 import org.apache.fluss.rocksdb.RocksIteratorWrapper;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.ProjectedRow;
-import org.apache.fluss.row.PruneRow;
 import org.apache.fluss.row.encode.ValueDecoder;
 import org.apache.fluss.utils.CloseableIterator;
 import org.apache.fluss.utils.CloseableRegistry;
@@ -160,7 +159,7 @@ public InternalRow next() {
         InternalRow originRow = originValue.row;
         if (targetSchemaId != originValue.schemaId) {
             originRow =
-                    PruneRow.from(schemaGetter.getSchema(originValue.schemaId), targetSchema)
+                    ProjectedRow.from(schemaGetter.getSchema(originValue.schemaId), targetSchema)
                             .replaceRow(originRow);
         }
 

fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java

@@ -921,6 +921,8 @@ public void close() {
 
         writerBufferPool.close();
         arrowWriterPool.close();
+        // Release all the memory segments.
+        bufferAllocator.releaseBytes(bufferAllocator.getAllocatedMemory());
         bufferAllocator.close();
     }
 

fluss-client/src/main/java/org/apache/fluss/client/write/WriteRecord.java

@@ -206,10 +206,6 @@ public int getEstimatedSizeInBytes() {
         return estimatedSizeInBytes;
     }
 
-    public TableInfo getTableInfo() {
-        return tableInfo;
-    }
-
     public int getSchemaId() {
         return tableInfo.getSchemaId();
     }

fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java

@@ -120,6 +120,7 @@ public WriterClient(
                             conf.get(ConfigOptions.CLIENT_WRITER_DYNAMIC_CREATE_PARTITION_ENABLED),
                             this::maybeAbortBatches);
         } catch (Throwable t) {
+            LOG.error("Failed to construct writer.", t);
             close(Duration.ofMillis(0));
             throw new FlussRuntimeException(
                     String.format(