[common] Add field_id for Nested Row.

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

@@ -530,7 +530,7 @@ Map<Integer, FetchLogRequest> prepareFetchLogRequests() {
                             assert projection != null;
                             reqForTable
                                     .setProjectionPushdownEnabled(true)
-                                    .setProjectedFields(projection.getProjectionIdInOrder());
+                                    .setProjectedFields(projection.getProjectionInOrder());
                         } else {
                             reqForTable.setProjectionPushdownEnabled(false);
                         }

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

@@ -123,7 +123,7 @@ private Projection sanityProjection(@Nullable int[] projectedFields, TableInfo t
                                     + tableRowType);
                 }
             }
-            return Projection.of(projectedFields, tableInfo.getSchema());
+            return Projection.of(projectedFields);
         } else {
             return null;
         }

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -71,6 +71,7 @@
 import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle;
 import org.apache.fluss.server.zk.ZooKeeperClient;
 import org.apache.fluss.server.zk.data.ServerTags;
+import org.apache.fluss.types.DataTypeChecks;
 import org.apache.fluss.types.DataTypes;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -394,7 +395,12 @@ void testAlterTableColumn() throws Exception {
 
         admin.alterTable(
                         tablePath,
-                        Collections.singletonList(
+                        Arrays.asList(
+                                TableChange.addColumn(
+                                        "nested_row",
+                                        DataTypes.ROW(DataTypes.STRING(), DataTypes.INT()),
+                                        "new nested column",
+                                        TableChange.ColumnPosition.last()),
                                 TableChange.addColumn(
                                         "c1",
                                         DataTypes.STRING(),
@@ -408,23 +414,28 @@ void testAlterTableColumn() throws Exception {
                         .primaryKey("id")
                         .fromColumns(
                                 Arrays.asList(
+                                        new Schema.Column("id", DataTypes.INT(), "person id", 0),
                                         new Schema.Column(
-                                                "id", DataTypes.INT(), "person id", (short) 0),
+                                                "name", DataTypes.STRING(), "person name", 1),
+                                        new Schema.Column("age", DataTypes.INT(), "person age", 2),
                                         new Schema.Column(
-                                                "name",
-                                                DataTypes.STRING(),
-                                                "person name",
-                                                (short) 1),
+                                                "nested_row",
+                                                DataTypes.ROW(
+                                                        DataTypes.FIELD(
+                                                                "f0", DataTypes.STRING(), 4),
+                                                        DataTypes.FIELD("f1", DataTypes.INT(), 5)),
+                                                "new nested column",
+                                                3),
                                         new Schema.Column(
-                                                "age", DataTypes.INT(), "person age", (short) 2),
-                                        new Schema.Column(
-                                                "c1",
-                                                DataTypes.STRING(),
-                                                "new column c1",
-                                                (short) 3)))
+                                                "c1", DataTypes.STRING(), "new column c1", 6)))
                         .build();
         SchemaInfo schemaInfo = admin.getTableSchema(tablePath).get();
         assertThat(schemaInfo).isEqualTo(new SchemaInfo(expectedSchema, 2));
+        // test field_id of rowType
+        assertThat(
+                        DataTypeChecks.equalsWithFieldId(
+                                schemaInfo.getSchema().getRowType(), expectedSchema.getRowType()))
+                .isTrue();
     }
 
     @Test

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

@@ -174,7 +174,7 @@ void testProjection(LogFormat logFormat, byte magic) throws Exception {
         long fetchOffset = 0L;
         int bucketId = 0; // records for 0-10.
         TableBucket tb = new TableBucket(DATA2_TABLE_ID, bucketId);
-        Projection projection = Projection.of(new int[] {0, 2}, schema);
+        Projection projection = Projection.of(new int[] {0, 2});
         MemoryLogRecords memoryLogRecords;
         if (logFormat == LogFormat.ARROW) {
             memoryLogRecords = genRecordsWithProjection(DATA2, projection, magic);
@@ -210,7 +210,7 @@ void testProjection(LogFormat logFormat, byte magic) throws Exception {
         // test projection reorder.
         defaultCompletedFetch =
                 makeCompletedFetch(
-                        tb, resultForBucket0, fetchOffset, Projection.of(new int[] {2, 0}, schema));
+                        tb, resultForBucket0, fetchOffset, Projection.of(new int[] {2, 0}));
         scanRecords = defaultCompletedFetch.fetchRecords(8);
         assertThat(scanRecords.size()).isEqualTo(8);
         for (int i = 0; i < scanRecords.size(); i++) {
@@ -427,7 +427,7 @@ private MemoryLogRecords genRecordsWithProjection(
                 DATA2_TABLE_ID,
                 testingSchemaGetter,
                 DEFAULT_COMPRESSION,
-                projection.getProjectionIdInOrder());
+                projection.getProjectionInOrder());
         ByteBuffer buffer =
                 toByteBuffer(
                         fileLogProjection

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

@@ -226,10 +226,7 @@ void testProjection(String format) throws Exception {
                 createFileLogRecords(tableBucket, DATA2_PHYSICAL_TABLE_PATH, DATA2, logFormat);
         RemoteCompletedFetch completedFetch =
                 makeCompletedFetch(
-                        tableBucket,
-                        fileLogRecords,
-                        fetchOffset,
-                        Projection.of(new int[] {0, 2}, schema));
+                        tableBucket, fileLogRecords, fetchOffset, Projection.of(new int[] {0, 2}));
 
         List<ScanRecord> scanRecords = completedFetch.fetchRecords(8);
         List<Object[]> expectedObjects =
@@ -255,10 +252,7 @@ void testProjection(String format) throws Exception {
 
         completedFetch =
                 makeCompletedFetch(
-                        tableBucket,
-                        fileLogRecords,
-                        fetchOffset,
-                        Projection.of(new int[] {2, 0}, schema));
+                        tableBucket, fileLogRecords, fetchOffset, Projection.of(new int[] {2, 0}));
         scanRecords = completedFetch.fetchRecords(8);
         assertThat(scanRecords.size()).isEqualTo(8);
         for (int i = 0; i < scanRecords.size(); i++) {