[common] Add field_id for Nested Row.

Author: loserwang1024

fluss-common/src/main/java/org/apache/fluss/types/ReassignFieldId.java

@@ -21,7 +21,7 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
-/** Reassign field id by given field id. */
+/** Visitor that recursively reassigns field IDs in nested data types using a provided counter. */
 public class ReassignFieldId extends DataTypeDefaultVisitor<DataType> {
 
     private final AtomicInteger highestFieldId;

fluss-common/src/main/java/org/apache/fluss/types/RowType.java

@@ -250,10 +250,6 @@ private static List<DataField> validateFields(List<DataField> fields) {
                     "Field IDs must be either all -1 or all non-negative. "
                             + "Mixed field IDs are not allowed.");
         } else {
-            // All fields have non-negative IDs
-            if (fieldIds.stream().anyMatch(id -> id < 0)) {
-                throw new IllegalArgumentException("Field ID must not be negative.");
-            }
             final Set<Integer> duplicateIds =
                     fieldIds.stream()
                             .filter(id -> Collections.frequency(fieldIds, id) > 1)

fluss-common/src/test/java/org/apache/fluss/utils/json/DataTypeJsonSerdeTest.java

@@ -143,8 +143,8 @@ protected String[] expectedJsons() {
             "{\"type\":\"TIMESTAMP_WITH_LOCAL_TIME_ZONE\",\"nullable\":false,\"precision\":3}",
             "{\"type\":\"ARRAY\",\"element_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
             "{\"type\":\"ARRAY\",\"nullable\":false,\"element_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
-            "{\"type\":\"MAP\",\"key_type\":{\"type\":\"BIGINT\"},\"nullable\":false},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
-            "{\"type\":\"MAP\",\"nullable\":false,\"key_type\":{\"type\":\"BIGINT\"},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
+            "{\"type\":\"MAP\",\"key_type\":{\"type\":\"BIGINT\",\"nullable\":false},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
+            "{\"type\":\"MAP\",\"nullable\":false,\"key_type\":{\"type\":\"BIGINT\",\"nullable\":false},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
             "{\"type\":\"ROW\",\"fields\":[{\"name\":\"f0\",\"field_type\":{\"type\":\"BIGINT\"},\"field_id\":0},{\"name\":\"f1\",\"field_type\":{\"type\":\"INTEGER\",\"nullable\":false},\"field_id\":1},{\"name\":\"f2\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":2}]}",
             "{\"type\":\"ROW\",\"nullable\":false,\"fields\":[{\"name\":\"f0\",\"field_type\":{\"type\":\"BIGINT\"},\"field_id\":0},{\"name\":\"f1\",\"field_type\":{\"type\":\"INTEGER\",\"nullable\":false},\"field_id\":1},{\"name\":\"f2\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":2}]}"
         };
@@ -200,8 +200,8 @@ void testCompatibilityFromJsonLackOfFieldId() {
                     "{\"type\":\"TIMESTAMP_WITH_LOCAL_TIME_ZONE\",\"nullable\":false,\"precision\":3}",
                     "{\"type\":\"ARRAY\",\"element_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
                     "{\"type\":\"ARRAY\",\"nullable\":false,\"element_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
-                    "{\"type\":\"MAP\",\"key_type\":{\"type\":\"BIGINT\"},\"nullable\":false},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
-                    "{\"type\":\"MAP\",\"nullable\":false,\"key_type\":{\"type\":\"BIGINT\"},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
+                    "{\"type\":\"MAP\",\"key_type\":{\"type\":\"BIGINT\",\"nullable\":false},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
+                    "{\"type\":\"MAP\",\"nullable\":false,\"key_type\":{\"type\":\"BIGINT\",\"nullable\":false},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
                     "{\"type\":\"ROW\",\"fields\":[{\"name\":\"f0\",\"field_type\":{\"type\":\"BIGINT\"}},{\"name\":\"f1\",\"field_type\":{\"type\":\"INTEGER\",\"nullable\":false}},{\"name\":\"f2\",\"field_type\":{\"type\":\"STRING\"}}]}",
                     "{\"type\":\"ROW\",\"nullable\":false,\"fields\":[{\"name\":\"f0\",\"field_type\":{\"type\":\"BIGINT\"}},{\"name\":\"f1\",\"field_type\":{\"type\":\"INTEGER\",\"nullable\":false}},{\"name\":\"f2\",\"field_type\":{\"type\":\"STRING\"}}]}",
                 };

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkComplexTypeITCase.java

@@ -625,8 +625,8 @@ void testProjectionAndAddColumnInLogTable() throws Exception {
 
         // Currently, flink not supported push down nested row projection because
         // FlinkTableSource.supportsNestedProjection returns false.
-        // Todo: supported nested row projection down in https://github.com/apache/fluss/issues/2311
-        // later.
+        // Todo: support nested row projection pushdown in
+        // https://github.com/apache/fluss/issues/2311 later.
         String s = tEnv.explainSql("select id, simple_row.a, nested_row.y.z from row_log_test");
         assertThat(s)
                 .contains(