[server][client] Improve the design and performance of ADD COLUMN operation

1. Make lookup requests async.
2. Introduce ServerProjectionCache to share ProjectionInfo in TabletServer level.
3. Revert FlinkAsFlussRow changes, and introduce PaddingRow for the padding null columns for schema changes.
4. Make PutKv zero-copy to persist kv records.
5. Rename SchemaMetadataManager to ServerSchemaCache and use TableId instead of TablePath to track schemas which is safer.
6. Revert source related changes to still include the projectedFieldIndexes parameter.

Author: wuchong

fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java

@@ -88,9 +88,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
 
-import static org.apache.fluss.client.utils.ClientRpcMessageUtils.addPbAlterSchemas;
+import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeAlterTableRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeCreatePartitionRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeDropPartitionRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeListOffsetsRequest;
@@ -249,23 +248,8 @@ public CompletableFuture<Void> createTable(
     public CompletableFuture<Void> alterTable(
             TablePath tablePath, List<TableChange> tableChanges, boolean ignoreIfNotExists) {
         tablePath.validate();
-        AlterTableRequest request = new AlterTableRequest();
-        List<PbAlterConfig> pbFlussTableChanges =
-                tableChanges.stream()
-                        .filter(tableChange -> !(tableChange instanceof TableChange.SchemaChange))
-                        .map(ClientRpcMessageUtils::toPbAlterConfigs)
-                        .collect(Collectors.toList());
-
-        List<TableChange> schemaChanges =
-                tableChanges.stream()
-                        .filter(tableChange -> tableChange instanceof TableChange.SchemaChange)
-                        .collect(Collectors.toList());
-        addPbAlterSchemas(request, schemaChanges);
-        request.addAllConfigChanges(pbFlussTableChanges)
-                .setIgnoreIfNotExists(ignoreIfNotExists)
-                .setTablePath()
-                .setDatabaseName(tablePath.getDatabaseName())
-                .setTableName(tablePath.getTableName());
+        AlterTableRequest request =
+                makeAlterTableRequest(tablePath, tableChanges, ignoreIfNotExists);
         return gateway.alterTable(request).thenApply(r -> null);
     }
 

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

@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.lookup;
+
+import org.apache.fluss.client.metadata.MetadataUpdater;
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.SchemaGetter;
+import org.apache.fluss.metadata.SchemaInfo;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.decode.FixedSchemaDecoder;
+import org.apache.fluss.utils.CopyOnWriteMap;
+import org.apache.fluss.utils.concurrent.FutureUtils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/** Abstract lookuper implementation for common methods. */
+abstract class AbstractLookuper implements Lookuper {
+    protected final TableInfo tableInfo;
+
+    protected final MetadataUpdater metadataUpdater;
+
+    protected final LookupClient lookupClient;
+
+    protected final short targetSchemaId;
+
+    private final SchemaGetter schemaGetter;
+
+    /**
+     * Cache for row decoders for different schema ids. Use CopyOnWriteMap for fast access, as it is
+     * not frequently updated.
+     */
+    private final CopyOnWriteMap<Short, FixedSchemaDecoder> decoders;
+
+    AbstractLookuper(
+            TableInfo tableInfo,
+            MetadataUpdater metadataUpdater,
+            LookupClient lookupClient,
+            SchemaGetter schemaGetter) {
+        this.tableInfo = tableInfo;
+        this.metadataUpdater = metadataUpdater;
+        this.lookupClient = lookupClient;
+        this.targetSchemaId = (short) tableInfo.getSchemaId();
+        this.schemaGetter = schemaGetter;
+        this.decoders = new CopyOnWriteMap<>();
+        // initialize the decoder for the same schema
+        this.decoders.put(
+                targetSchemaId,
+                new FixedSchemaDecoder(
+                        tableInfo.getTableConfig().getKvFormat(), tableInfo.getSchema()));
+    }
+
+    protected void handleLookupResponse(
+            List<byte[]> result, CompletableFuture<LookupResult> lookupFuture) {
+        List<MemorySegment> valueList = new ArrayList<>(result.size());
+        Set<Short> schemaIdsToRequest = new HashSet<>();
+        boolean allTargetSchema = true;
+        for (byte[] valueBytes : result) {
+            if (valueBytes == null) {
+                continue;
+            }
+            MemorySegment memorySegment = MemorySegment.wrap(valueBytes);
+            short schemaId = memorySegment.getShort(0);
+            if (targetSchemaId != schemaId) {
+                allTargetSchema = false;
+                if (!decoders.containsKey(schemaId)) {
+                    schemaIdsToRequest.add(schemaId);
+                }
+            }
+            valueList.add(memorySegment);
+        }
+        // all schema ids are the target schema id, fast path
+        if (allTargetSchema) {
+            lookupFuture.complete(processAllTargetSchemaRows(valueList));
+            return;
+        }
+
+        // all schema ids and decoders are cached in local
+        if (schemaIdsToRequest.isEmpty()) {
+            lookupFuture.complete(processSchemaMismatchedRows(valueList));
+            return;
+        }
+
+        // need to fetch schema infos for schema ids
+        List<CompletableFuture<SchemaInfo>> schemaFutures =
+                new ArrayList<>(schemaIdsToRequest.size());
+        for (short schemaId : schemaIdsToRequest) {
+            CompletableFuture<SchemaInfo> schemaFuture = schemaGetter.getSchemaInfoAsync(schemaId);
+            schemaFutures.add(schemaFuture);
+        }
+        FutureUtils.ConjunctFuture<Collection<SchemaInfo>> allFutures =
+                FutureUtils.combineAll(schemaFutures);
+        // normal async path
+        allFutures.whenComplete(
+                (schemas, error) -> {
+                    if (error != null) {
+                        lookupFuture.completeExceptionally(
+                                new RuntimeException(
+                                        "Failed to get schema infos for lookup", error));
+                    } else {
+                        LookupResult lookupResult = processSchemaRequestedRows(schemas, valueList);
+                        lookupFuture.complete(lookupResult);
+                    }
+                });
+    }
+
+    protected LookupResult processAllTargetSchemaRows(List<MemorySegment> valueList) {
+        FixedSchemaDecoder decoder = decoders.get(targetSchemaId);
+        List<InternalRow> rowList = new ArrayList<>(valueList.size());
+        for (MemorySegment value : valueList) {
+            rowList.add(decoder.decode(value));
+        }
+        return new LookupResult(rowList);
+    }
+
+    protected LookupResult processSchemaMismatchedRows(List<MemorySegment> valueList) {
+        List<InternalRow> rowList = new ArrayList<>(valueList.size());
+        for (MemorySegment value : valueList) {
+            short schemaId = value.getShort(0);
+            FixedSchemaDecoder decoder = decoders.get(schemaId);
+            checkArgument(decoder != null, "Decoder for schema id %s not found", schemaId);
+            InternalRow row = decoder.decode(value);
+            rowList.add(row);
+        }
+        return new LookupResult(rowList);
+    }
+
+    protected LookupResult processSchemaRequestedRows(
+            Collection<SchemaInfo> schemaInfos, List<MemorySegment> valueList) {
+        // build a map from schema id to the target schema decoder
+        Map<Short, Schema> schemaMap = new HashMap<>();
+        for (SchemaInfo schemaInfo : schemaInfos) {
+            schemaMap.put((short) schemaInfo.getSchemaId(), schemaInfo.getSchema());
+        }
+
+        // process the value list to convert to target schema
+        List<InternalRow> rowList = new ArrayList<>(valueList.size());
+        for (MemorySegment value : valueList) {
+            short schemaId = value.getShort(0);
+            FixedSchemaDecoder decoder =
+                    decoders.computeIfAbsent(
+                            schemaId,
+                            (id) -> {
+                                Schema sourceSchema = schemaMap.get(id);
+                                return new FixedSchemaDecoder(
+                                        tableInfo.getTableConfig().getKvFormat(),
+                                        sourceSchema,
+                                        tableInfo.getSchema());
+                            });
+            InternalRow row = decoder.decode(value);
+            rowList.add(row);
+        }
+        return new LookupResult(rowList);
+    }
+}

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

@@ -22,14 +22,11 @@
 import org.apache.fluss.client.table.getter.PartitionGetter;
 import org.apache.fluss.exception.PartitionNotExistException;
 import org.apache.fluss.metadata.DataLakeFormat;
-import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.SchemaGetter;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.row.InternalRow;
-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;
 
 import javax.annotation.Nullable;
@@ -49,13 +46,7 @@
  * of the primary key.
  */
 @NotThreadSafe
-class PrefixKeyLookuper implements Lookuper {
-
-    private final TableInfo tableInfo;
-
-    private final MetadataUpdater metadataUpdater;
-
-    private final LookupClient lookupClient;
+class PrefixKeyLookuper extends AbstractLookuper {
 
     /** Extract bucket key from prefix lookup key row. */
     private final KeyEncoder bucketKeyEncoder;
@@ -68,24 +59,16 @@ class PrefixKeyLookuper implements Lookuper {
      */
     private @Nullable final PartitionGetter partitionGetter;
 
-    /** Decode the lookup bytes to result row. */
-    private final ValueDecoder kvValueDecoder;
-
-    private final SchemaGetter schemaGetter;
-
     public PrefixKeyLookuper(
             TableInfo tableInfo,
             SchemaGetter schemaGetter,
             MetadataUpdater metadataUpdater,
             LookupClient lookupClient,
             List<String> lookupColumnNames) {
+        super(tableInfo, metadataUpdater, lookupClient, schemaGetter);
         // sanity check
         validatePrefixLookup(tableInfo, lookupColumnNames);
-        // initialization
-        this.tableInfo = tableInfo;
         this.numBuckets = tableInfo.getNumBuckets();
-        this.metadataUpdater = metadataUpdater;
-        this.lookupClient = lookupClient;
         // the row type of the input lookup row
         RowType lookupRowType = tableInfo.getRowType().project(lookupColumnNames);
         DataLakeFormat lakeFormat = tableInfo.getTableConfig().getDataLakeFormat().orElse(null);
@@ -96,9 +79,6 @@ public PrefixKeyLookuper(
                 tableInfo.isPartitioned()
                         ? new PartitionGetter(lookupRowType, tableInfo.getPartitionKeys())
                         : null;
-        this.kvValueDecoder =
-                new ValueDecoder(schemaGetter, tableInfo.getTableConfig().getKvFormat());
-        this.schemaGetter = schemaGetter;
     }
 
     private void validatePrefixLookup(TableInfo tableInfo, List<String> lookupColumns) {
@@ -168,36 +148,22 @@ public CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {
             }
         }
 
+        CompletableFuture<LookupResult> lookupFuture = new CompletableFuture<>();
         TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), partitionId, bucketId);
-        CompletableFuture<LookupResult> future = new CompletableFuture<>();
-
-        CompletableFuture.runAsync(
-                () -> {
-                    try {
-                        List<byte[]> result =
-                                lookupClient.prefixLookup(tableBucket, bucketKeyBytes).get();
-                        List<InternalRow> rowList = new ArrayList<>(result.size());
-                        for (byte[] valueBytes : result) {
-                            if (valueBytes == null) {
-                                continue;
-                            }
-                            ValueDecoder.Value value = kvValueDecoder.decodeValue(valueBytes);
-                            InternalRow row;
-                            if (value.schemaId == tableInfo.getSchemaId()) {
-                                row = value.row;
+        lookupClient
+                .prefixLookup(tableBucket, bucketKeyBytes)
+                .whenComplete(
+                        (result, error) -> {
+                            if (error != null) {
+                                lookupFuture.completeExceptionally(
+                                        new RuntimeException(
+                                                "Failed to perform prefix lookup for table: "
+                                                        + tableInfo.getTablePath(),
+                                                error));
                             } else {
-                                Schema schema = schemaGetter.getSchema(value.schemaId);
-                                row =
-                                        ProjectedRow.from(schema, tableInfo.getSchema())
-                                                .replaceRow(value.row);
+                                handleLookupResponse(result, lookupFuture);
                             }
-                            rowList.add(row);
-                        }
-                        future.complete(new LookupResult(rowList));
-                    } catch (Exception e) {
-                        future.complete(new LookupResult(Collections.emptyList()));
-                    }
-                });
-        return future;
+                        });
+        return lookupFuture;
     }
 }

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

@@ -23,7 +23,8 @@
 import java.util.List;
 
 /**
- * Used to describe the operation to prefix lookup by {@link PrefixLookuper} to a primary key table.
+ * Used to describe the operation to prefix lookup by {@link PrefixKeyLookuper} to a primary key
+ * table.
  *
  * @since 0.6
  */
@@ -37,7 +38,7 @@ public class PrefixLookup {
      * <p>For partitioned table, the lookupColumnNames exclude partition fields should be a prefix
      * of primary key exclude partition fields.
      *
-     * <p>See {@link PrefixLookuper#prefixLookup(InternalRow)} for more details.
+     * <p>See {@link PrefixKeyLookuper#lookup(InternalRow)} for more details.
      */
     private final List<String> lookupColumnNames;