POC: Support Add Column.

Author: loserwang1024

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

@@ -21,6 +21,7 @@
 import org.apache.fluss.client.admin.Admin;
 import org.apache.fluss.client.table.Table;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.SchemaInfo;
 import org.apache.fluss.metadata.TablePath;
 
 import javax.annotation.concurrent.ThreadSafe;
@@ -56,6 +57,16 @@ public interface Connection extends AutoCloseable {
     /** Retrieve a new Table client to operate data in table. */
     Table getTable(TablePath tablePath);
 
+    /**
+     * Retrieve a new Table client to operate data in table with specific schema. When performing
+     * read, write and loop operations, this schema will be used.
+     *
+     * @param tablePath the path of the table to operate on
+     * @param schemaInfo the schema information to be used for the table operations
+     * @return a new Table client instance for the specified table and schema
+     */
+    Table getTable(TablePath tablePath, SchemaInfo schemaInfo);
+
     /** Close the connection and release all resources. */
     @Override
     void close() throws Exception;

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

@@ -33,6 +33,7 @@
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.fs.FileSystem;
+import org.apache.fluss.metadata.SchemaInfo;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.metrics.registry.MetricRegistry;
@@ -104,6 +105,14 @@ public Table getTable(TablePath tablePath) {
         return new FlussTable(this, tablePath, tableInfo);
     }
 
+    @Override
+    public Table getTable(TablePath tablePath, SchemaInfo schemaInfo) {
+        metadataUpdater.updateTableOrPartitionMetadata(tablePath, null);
+        TableInfo tableInfo = metadataUpdater.getTableInfoOrElseThrow(tablePath);
+        tableInfo = TableInfo.of(tableInfo, schemaInfo);
+        return new FlussTable(this, tablePath, tableInfo);
+    }
+
     public RpcClient getRpcClient() {
         return rpcClient;
     }

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

@@ -90,6 +90,7 @@
 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.makeCreatePartitionRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeDropPartitionRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeListOffsetsRequest;
@@ -249,12 +250,17 @@ 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()

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

@@ -22,13 +22,14 @@
 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.decode.RowDecoder;
+import org.apache.fluss.row.PruneRow;
 import org.apache.fluss.row.encode.KeyEncoder;
 import org.apache.fluss.row.encode.ValueDecoder;
-import org.apache.fluss.types.DataType;
 import org.apache.fluss.types.RowType;
 
 import javax.annotation.Nullable;
@@ -68,8 +69,11 @@ class PrefixKeyLookuper implements Lookuper {
     /** 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) {
@@ -91,10 +95,8 @@ public PrefixKeyLookuper(
                         ? new PartitionGetter(lookupRowType, tableInfo.getPartitionKeys())
                         : null;
         this.kvValueDecoder =
-                new ValueDecoder(
-                        RowDecoder.create(
-                                tableInfo.getTableConfig().getKvFormat(),
-                                tableInfo.getRowType().getChildren().toArray(new DataType[0])));
+                new ValueDecoder(schemaGetter, tableInfo.getTableConfig().getKvFormat());
+        this.schemaGetter = schemaGetter;
     }
 
     private void validatePrefixLookup(TableInfo tableInfo, List<String> lookupColumns) {
@@ -174,7 +176,17 @@ public CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {
                                 if (valueBytes == null) {
                                     continue;
                                 }
-                                rowList.add(kvValueDecoder.decodeValue(valueBytes).row);
+                                ValueDecoder.Value value = kvValueDecoder.decodeValue(valueBytes);
+                                InternalRow row;
+                                if (value.schemaId == tableInfo.getSchemaId()) {
+                                    row = value.row;
+                                } else {
+                                    Schema schema = schemaGetter.getSchema(value.schemaId);
+                                    row =
+                                            PruneRow.from(schema, tableInfo.getSchema())
+                                                    .replaceRow(value.row);
+                                }
+                                rowList.add(row);
                             }
                             return new LookupResult(rowList);
                         });

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

@@ -22,13 +22,14 @@
 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.decode.RowDecoder;
+import org.apache.fluss.row.PruneRow;
 import org.apache.fluss.row.encode.KeyEncoder;
 import org.apache.fluss.row.encode.ValueDecoder;
-import org.apache.fluss.types.DataType;
 import org.apache.fluss.types.RowType;
 
 import javax.annotation.Nullable;
@@ -38,6 +39,7 @@
 
 import static org.apache.fluss.client.utils.ClientUtils.getPartitionId;
 import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
 /** An implementation of {@link Lookuper} that lookups by primary key. */
 class PrimaryKeyLookuper implements Lookuper {
@@ -65,8 +67,13 @@ class PrimaryKeyLookuper implements Lookuper {
     /** Decode the lookup bytes to result row. */
     private final ValueDecoder kvValueDecoder;
 
+    private final SchemaGetter schemaGetter;
+
     public PrimaryKeyLookuper(
-            TableInfo tableInfo, MetadataUpdater metadataUpdater, LookupClient lookupClient) {
+            TableInfo tableInfo,
+            SchemaGetter schemaGetter,
+            MetadataUpdater metadataUpdater,
+            LookupClient lookupClient) {
         checkArgument(
                 tableInfo.hasPrimaryKey(),
                 "Log table %s doesn't support lookup",
@@ -94,10 +101,8 @@ public PrimaryKeyLookuper(
                         ? new PartitionGetter(lookupRowType, tableInfo.getPartitionKeys())
                         : null;
         this.kvValueDecoder =
-                new ValueDecoder(
-                        RowDecoder.create(
-                                tableInfo.getTableConfig().getKvFormat(),
-                                tableInfo.getRowType().getChildren().toArray(new DataType[0])));
+                new ValueDecoder(schemaGetter, tableInfo.getTableConfig().getKvFormat());
+        this.schemaGetter = schemaGetter;
     }
 
     @Override
@@ -129,10 +134,20 @@ public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {
                 .lookup(tableBucket, pkBytes)
                 .thenApply(
                         valueBytes -> {
-                            InternalRow row =
-                                    valueBytes == null
-                                            ? null
-                                            : kvValueDecoder.decodeValue(valueBytes).row;
+                            InternalRow row = null;
+                            if (valueBytes != null) {
+                                ValueDecoder.Value value = kvValueDecoder.decodeValue(valueBytes);
+                                if (value.schemaId == tableInfo.getSchemaId()) {
+                                    row = value.row;
+                                } else {
+                                    Schema schema = schemaGetter.getSchema(value.schemaId);
+                                    checkNotNull(schema, "schema is null");
+                                    row =
+                                            PruneRow.from(schema, tableInfo.getSchema())
+                                                    .replaceRow(value.row);
+                                }
+                            }
+
                             return new LookupResult(row);
                         });
     }

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

@@ -18,6 +18,7 @@
 package org.apache.fluss.client.lookup;
 
 import org.apache.fluss.client.metadata.MetadataUpdater;
+import org.apache.fluss.metadata.SchemaGetter;
 import org.apache.fluss.metadata.TableInfo;
 
 import javax.annotation.Nullable;
@@ -28,39 +29,46 @@
 public class TableLookup implements Lookup {
 
     private final TableInfo tableInfo;
+    private final SchemaGetter schemaGetter;
     private final MetadataUpdater metadataUpdater;
     private final LookupClient lookupClient;
 
     @Nullable private final List<String> lookupColumnNames;
 
     public TableLookup(
-            TableInfo tableInfo, MetadataUpdater metadataUpdater, LookupClient lookupClient) {
-        this(tableInfo, metadataUpdater, lookupClient, null);
+            TableInfo tableInfo,
+            SchemaGetter schemaGetter,
+            MetadataUpdater metadataUpdater,
+            LookupClient lookupClient) {
+        this(tableInfo, schemaGetter, metadataUpdater, lookupClient, null);
     }
 
     private TableLookup(
             TableInfo tableInfo,
+            SchemaGetter schemaGetter,
             MetadataUpdater metadataUpdater,
             LookupClient lookupClient,
             @Nullable List<String> lookupColumnNames) {
         this.tableInfo = tableInfo;
+        this.schemaGetter = schemaGetter;
         this.metadataUpdater = metadataUpdater;
         this.lookupClient = lookupClient;
         this.lookupColumnNames = lookupColumnNames;
     }
 
     @Override
     public Lookup lookupBy(List<String> lookupColumnNames) {
-        return new TableLookup(tableInfo, metadataUpdater, lookupClient, lookupColumnNames);
+        return new TableLookup(
+                tableInfo, schemaGetter, metadataUpdater, lookupClient, lookupColumnNames);
     }
 
     @Override
     public Lookuper createLookuper() {
         if (lookupColumnNames == null) {
-            return new PrimaryKeyLookuper(tableInfo, metadataUpdater, lookupClient);
+            return new PrimaryKeyLookuper(tableInfo, schemaGetter, metadataUpdater, lookupClient);
         } else {
             return new PrefixKeyLookuper(
-                    tableInfo, metadataUpdater, lookupClient, lookupColumnNames);
+                    tableInfo, schemaGetter, metadataUpdater, lookupClient, lookupColumnNames);
         }
     }
 }

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

@@ -0,0 +1,87 @@
+/*
+ * 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.metadata;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.SchemaGetter;
+import org.apache.fluss.metadata.SchemaInfo;
+import org.apache.fluss.metadata.TablePath;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/** 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;
+
+    public ClientSchemaGetter(TablePath tablePath, SchemaInfo latestSchemaInfo, Admin admin) {
+        this.tablePath = tablePath;
+        this.latestSchemaInfo = latestSchemaInfo;
+        this.admin = admin;
+        this.schemasById = new HashMap<>();
+        schemasById.put(latestSchemaInfo.getSchemaId(), latestSchemaInfo.getSchema());
+    }
+
+    @Override
+    public Schema getSchema(int schemaId) {
+        return schemasById.computeIfAbsent(
+                schemaId,
+                (id) -> {
+                    try {
+                        // todo: 测试这一步总是会一定概率卡住(有时候卡几分钟恢复），这个可以排查一下。
+                        SchemaInfo schemaInfo =
+                                admin.getTableSchema(tablePath, schemaId).get(5, TimeUnit.MINUTES);
+                        if (id > latestSchemaInfo.getSchemaId()) {
+                            latestSchemaInfo = schemaInfo;
+                        }
+                        return schemaInfo.getSchema();
+
+                    } catch (Exception e) {
+                        LOG.warn("Failed to get schema for table: " + tablePath);
+                        throw new RuntimeException(e);
+                    }
+                });
+    }
+
+    @Override
+    public SchemaInfo getLatestSchemaInfo() {
+        return latestSchemaInfo;
+    }
+
+    @Override
+    public void release() {
+        try {
+            admin.close();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}