1. reduce the re-calculate of file projection.

2. fix test

Author: loserwang1024

fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java

@@ -237,7 +237,6 @@ void testUpsertWithSmallBuffer() throws Exception {
         }
     }
 
-    // todo: 测试写入一个过于提前或者为0的schema后会如何
     @Test
     void testPutAndLookup() throws Exception {
         TablePath tablePath = TablePath.of("test_db_1", "test_put_and_lookup_table");

fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/DefaultCompletedFetchTest.java

@@ -266,7 +266,6 @@ private MemoryLogRecords genRecordsWithProjection(
         fileLogRecords.flush();
 
         FileLogProjection fileLogProjection = new FileLogProjection();
-        // todo: 调整测试
         fileLogProjection.setCurrentProjection(
                 DATA2_TABLE_ID,
                 testingSchemaGetter,

fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java

@@ -35,10 +35,10 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -56,11 +56,9 @@
 public final class Schema implements Serializable {
 
     private static final long serialVersionUID = 1L;
-
     private static final Schema EMPTY = Schema.newBuilder().build();
 
     private final List<Column> columns;
-    private final Map<Integer, Column> columnById;
     private final @Nullable PrimaryKey primaryKey;
     private final RowType rowType;
 
@@ -82,8 +80,6 @@ private Schema(List<Column> columns, @Nullable PrimaryKey primaryKey, int highes
                                                 new DataField(
                                                         column.getName(), column.getDataType()))
                                 .collect(Collectors.toList()));
-        columnById =
-                columns.stream().collect(Collectors.toMap(Column::getColumnId, column -> column));
         this.highestFieldId = highestFieldId;
     }
 
@@ -207,10 +203,11 @@ public static Schema.Builder newBuilder() {
     public static final class Builder {
         private final List<Column> columns;
         private @Nullable PrimaryKey primaryKey;
-        private @Nullable Integer highestFieldId;
+        private AtomicInteger highestFieldId;
 
         private Builder() {
             columns = new ArrayList<>();
+            highestFieldId = new AtomicInteger(-1);
         }
 
         /** Adopts all members from the given schema. */
@@ -219,11 +216,12 @@ public Builder fromSchema(Schema schema) {
             if (schema.primaryKey != null) {
                 primaryKeyNamed(schema.primaryKey.constraintName, schema.primaryKey.columnNames);
             }
+            this.highestFieldId = new AtomicInteger(schema.highestFieldId);
             return this;
         }
 
         public Builder highestFieldId(int highestFieldId) {
-            this.highestFieldId = highestFieldId;
+            this.highestFieldId = new AtomicInteger(highestFieldId);
             return this;
         }
 
@@ -287,8 +285,7 @@ public Builder fromColumns(List<Column> inputColumns) {
         public Builder column(String columnName, DataType dataType) {
             checkNotNull(columnName, "Column name must not be null.");
             checkNotNull(dataType, "Data type must not be null.");
-            short defaultColumnId = (short) columns.size();
-            columns.add(new Column(columnName, dataType, null, defaultColumnId));
+            columns.add(new Column(columnName, dataType, null, highestFieldId.incrementAndGet()));
             return this;
         }
 
@@ -362,18 +359,15 @@ public Builder primaryKeyNamed(String constraintName, List<String> columnNames)
         public Schema build() {
             Integer maximumColumnId =
                     columns.stream().map(Column::getColumnId).max(Integer::compareTo).orElse(0);
-            if (highestFieldId == null) {
-                highestFieldId = maximumColumnId;
-            } else {
-                checkState(
-                        highestFieldId >= maximumColumnId,
-                        "Highest field id must be greater than or equal to the maximum column id.");
-            }
+
+            checkState(
+                    columns.isEmpty() || highestFieldId.get() >= maximumColumnId,
+                    "Highest field id must be greater than or equal to the maximum column id.");
 
             checkState(
                     columns.stream().map(Column::getColumnId).distinct().count() == columns.size(),
                     "Column ids must be unique.");
-            return new Schema(columns, primaryKey, highestFieldId);
+            return new Schema(columns, primaryKey, highestFieldId.get());
         }
     }
 

fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java

@@ -103,7 +103,7 @@ public class FileLogProjection {
     private SchemaGetter schemaGetter;
     private long tableId;
     private ArrowCompressionInfo compressionInfo;
-    private int[] selectedFieldIds;
+    private int[] selectedFieldPositions;
 
     public FileLogProjection() {
         this.outputStream = new ByteArrayOutputStream();
@@ -118,11 +118,18 @@ public void setCurrentProjection(
             long tableId,
             SchemaGetter schemaGetter,
             ArrowCompressionInfo compressionInfo,
-            int[] selectedFields) {
+            int[] selectedFieldIds) {
         this.tableId = tableId;
         this.schemaGetter = schemaGetter;
         this.compressionInfo = compressionInfo;
-        this.selectedFieldIds = selectedFields;
+
+        // Currently, only add last column is supported.Thus selectedFieldPositions is always same
+        // from same selectedFieldIds.
+        // TODO: if support drop column or add column in middle, this selectedFieldPositions should
+        // be re-calculated for each schema.
+        this.selectedFieldPositions =
+                selectedFieldPositions(
+                        schemaGetter.getLatestSchemaInfo().getSchema(), selectedFieldIds);
     }
 
     /**
@@ -405,16 +412,14 @@ private void setCurrentSchema(short schemaId) {
         if (projectionsCache.containsKey(projectionKey)) {
             // the schema and projection should identical for the same table id.
             currentProjection = projectionsCache.get(projectionKey);
-            if (!Arrays.equals(currentProjection.selectedFieldIds, selectedFieldIds)
+            if (!Arrays.equals(currentProjection.selectedFieldPositions, selectedFieldPositions)
                     || !currentProjection.schema.equals(rowType)) {
                 throw new InvalidColumnProjectionException(
                         "The schema and projection should be identical for the same table id.");
             }
             return;
         }
 
-        int[] selectedFieldPositions = checkProjection(schema, selectedFieldIds);
-
         // initialize the projection util information
         Schema arrowSchema = ArrowUtils.toArrowSchema(rowType);
         BitSet selection = toBitSet(arrowSchema.getFields().size(), selectedFieldPositions);
@@ -454,12 +459,11 @@ private void setCurrentSchema(short schemaId) {
                         rowType,
                         metadataLength,
                         bodyCompression,
-                        selectedFieldPositions,
-                        selectedFieldIds);
+                        selectedFieldPositions);
         projectionsCache.put(projectionKey, currentProjection);
     }
 
-    int[] checkProjection(org.apache.fluss.metadata.Schema schema, int[] projectedFields) {
+    int[] selectedFieldPositions(org.apache.fluss.metadata.Schema schema, int[] projectedFields) {
         Map<Integer, Integer> columnIdPositions = new HashMap<>();
         List<Integer> columnIds = schema.getColumnIds();
         for (int i = 0; i < columnIds.size(); i++) {
@@ -521,7 +525,6 @@ static final class ProjectionInfo {
         final int arrowMetadataLength;
         final ArrowBodyCompression bodyCompression;
         final int[] selectedFieldPositions;
-        final int[] selectedFieldIds;
 
         private ProjectionInfo(
                 BitSet nodesProjection,
@@ -530,16 +533,14 @@ private ProjectionInfo(
                 RowType schema,
                 int arrowMetadataLength,
                 ArrowBodyCompression bodyCompression,
-                int[] selectedFieldPositions,
-                int[] selectedFieldIds) {
+                int[] selectedFieldPositions) {
             this.nodesProjection = nodesProjection;
             this.buffersProjection = buffersProjection;
             this.bufferCount = bufferCount;
             this.schema = schema;
             this.arrowMetadataLength = arrowMetadataLength;
             this.bodyCompression = bodyCompression;
             this.selectedFieldPositions = selectedFieldPositions;
-            this.selectedFieldIds = selectedFieldIds;
         }
     }
 

fluss-common/src/main/java/org/apache/fluss/record/KvRecordReadContext.java

@@ -31,8 +31,6 @@ public class KvRecordReadContext implements KvRecordBatch.ReadContext {
 
     private final KvFormat kvFormat;
     private final SchemaGetter schemaGetter;
-
-    // todo: Modify to use cache.
     private final Map<Integer, RowDecoder> rowDecoderCache;
 
     private KvRecordReadContext(KvFormat kvFormat, SchemaGetter schemaGetter) {
@@ -48,8 +46,6 @@ public static KvRecordReadContext createReadContext(
 
     @Override
     public RowDecoder getRowDecoder(int schemaId) {
-        // assume schema id are always not changed for now.
-        // todo: however, schema will change here.
         return rowDecoderCache.computeIfAbsent(
                 schemaId,
                 (id) -> {

fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java

@@ -28,8 +28,6 @@
 
 /** A default implementation of {@link ValueRecordBatch.ReadContext} . */
 public class ValueRecordReadContext implements ValueRecordBatch.ReadContext {
-
-    // todo: use cache later.
     private final Map<Integer, RowDecoder> rowDecoderCache;
     private final SchemaGetter schemaGetter;
     private final KvFormat kvFormat;

fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java

@@ -25,24 +25,23 @@
 import org.apache.fluss.row.decode.RowDecoder;
 import org.apache.fluss.types.DataType;
 
-import java.util.HashMap;
 import java.util.Map;
 
 import static org.apache.fluss.row.encode.ValueEncoder.SCHEMA_ID_LENGTH;
+import static org.apache.fluss.utils.MapUtils.newConcurrentHashMap;
 
 /**
  * A decoder to decode a schema id and {@link BinaryRow} from a byte array value which is encoded by
  * {@link ValueEncoder#encodeValue(short, BinaryRow)}.
  */
 public class ValueDecoder {
 
-    // todo: use cache.
     private final Map<Short, RowDecoder> rowDecoders;
     private final SchemaGetter schemaGetter;
     private final KvFormat kvFormat;
 
     public ValueDecoder(SchemaGetter schemaGetter, KvFormat kvFormat) {
-        this.rowDecoders = new HashMap<>();
+        this.rowDecoders = newConcurrentHashMap();
         this.schemaGetter = schemaGetter;
         this.kvFormat = kvFormat;
     }

fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/SchemaChangeEvent.java

@@ -20,6 +20,8 @@
 import org.apache.fluss.metadata.SchemaInfo;
 import org.apache.fluss.metadata.TablePath;
 
+import java.util.Objects;
+
 /** An event for schema change. */
 public class SchemaChangeEvent implements CoordinatorEvent {
     private final TablePath tablePath;
@@ -37,4 +39,24 @@ public TablePath getTablePath() {
     public SchemaInfo getSchemaInfo() {
         return schemaInfo;
     }
+
+    @Override
+    public String toString() {
+        return "SchemaChangeEvent{" + "tablePath=" + tablePath + ", schemaInfo=" + schemaInfo + '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof SchemaChangeEvent)) {
+            return false;
+        }
+        SchemaChangeEvent that = (SchemaChangeEvent) o;
+        return Objects.equals(tablePath, that.tablePath)
+                && Objects.equals(schemaInfo, that.schemaInfo);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tablePath, schemaInfo);
+    }
 }