modified based on CR

Author: loserwang1024

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.DataTypeEqualsWithFieldId;
 import org.apache.fluss.types.DataTypes;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -430,6 +431,11 @@ void testAlterTableColumn() throws Exception {
                         .build();
         SchemaInfo schemaInfo = admin.getTableSchema(tablePath).get();
         assertThat(schemaInfo).isEqualTo(new SchemaInfo(expectedSchema, 2));
+        // test field_id of rowType
+        assertThat(
+                        DataTypeEqualsWithFieldId.equals(
+                                schemaInfo.getSchema().getRowType(), expectedSchema.getRowType()))
+                .isTrue();
     }
 
     @Test

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

@@ -815,9 +815,10 @@ private static void checkFieldIds(List<Column> columns, int highestFieldId) {
         Integer maximumFieldId = allFieldIds.stream().max(Integer::compareTo).orElse(-1);
         checkState(
                 columns.isEmpty() || highestFieldId >= maximumFieldId,
-                "Highest field ID (%s) must be greater than or equal to the maximum field ID (%s) including nested fields.",
+                "Highest field ID (%s) must be greater than or equal to the maximum field ID (%s) including nested fields. Current columns is %s",
                 highestFieldId,
-                maximumFieldId);
+                maximumFieldId,
+                columns);
     }
 
     /**

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

@@ -25,7 +25,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
@@ -51,7 +50,9 @@ public RowType(boolean isNullable, List<DataField> fields) {
         super(isNullable, DataTypeRoot.ROW);
         this.fields =
                 Collections.unmodifiableList(
-                        validateFields(checkNotNull(fields, "Fields must not be null.")));
+                        new ArrayList<>(checkNotNull(fields, "Fields must not be null.")));
+
+        validateFields(fields);
     }
 
     public RowType(List<DataField> fields) {
@@ -163,106 +164,41 @@ public boolean equals(Object o) {
             return false;
         }
         RowType rowType = (RowType) o;
-        return fields.equals(rowType.fields);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(super.hashCode(), fields);
-    }
-
-    /**
-     * Compares this RowType with another RowType, ignoring field IDs.
-     *
-     * @param other the other RowType to compare with
-     * @return true if the RowTypes are equal ignoring field IDs, false otherwise
-     */
-    public boolean equalsIgnoreFieldId(RowType other) {
-        if (this == other) {
-            return true;
-        }
-        if (other == null) {
-            return false;
-        }
-        if (this.isNullable() != other.isNullable()) {
-            return false;
-        }
-        if (this.fields.size() != other.fields.size()) {
-            return false;
-        }
+        //  Compares this RowType with another RowType, ignoring field IDs.
         for (int i = 0; i < fields.size(); i++) {
             DataField thisField = fields.get(i);
-            DataField otherField = other.fields.get(i);
-            if (!thisField.getName().equals(otherField.getName())) {
-                return false;
-            }
-            if (!thisField.getType().equals(otherField.getType())) {
-                return false;
-            }
-            if (!Objects.equals(thisField.getDescription(), otherField.getDescription())) {
+            DataField otherField = rowType.fields.get(i);
+            if (!thisField.getName().equals(otherField.getName())
+                    || !thisField.getType().equals(otherField.getType())
+                    || !Objects.equals(thisField.getDescription(), otherField.getDescription())) {
                 return false;
             }
         }
         return true;
     }
 
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), fields);
+    }
+
     // --------------------------------------------------------------------------------------------
 
-    private static List<DataField> validateFields(List<DataField> fields) {
-        // Validate field names
+    private static void validateFields(List<DataField> fields) {
         final List<String> fieldNames =
                 fields.stream().map(DataField::getName).collect(Collectors.toList());
         if (fieldNames.stream().anyMatch(StringUtils::isNullOrWhitespaceOnly)) {
             throw new IllegalArgumentException(
                     "Field names must contain at least one non-whitespace character.");
         }
-        final Set<String> duplicateNames =
+        final Set<String> duplicates =
                 fieldNames.stream()
                         .filter(n -> Collections.frequency(fieldNames, n) > 1)
                         .collect(Collectors.toSet());
-        if (!duplicateNames.isEmpty()) {
+        if (!duplicates.isEmpty()) {
             throw new IllegalArgumentException(
-                    String.format(
-                            "Field names must be unique. Found duplicates: %s", duplicateNames));
+                    String.format("Field names must be unique. Found duplicates: %s", duplicates));
         }
-
-        // Validate and process field IDs
-        final List<Integer> fieldIds =
-                fields.stream().map(DataField::getFieldId).collect(Collectors.toList());
-        long negativeIdCount = fieldIds.stream().filter(id -> id == -1).count();
-
-        List<DataField> processedFields;
-        if (negativeIdCount == fields.size()) {
-            // All fields have ID -1, assign IDs in order
-            processedFields = new ArrayList<>();
-            for (int i = 0; i < fields.size(); i++) {
-                DataField field = fields.get(i);
-                processedFields.add(
-                        new DataField(
-                                field.getName(),
-                                field.getType(),
-                                field.getDescription().orElse(null),
-                                i));
-            }
-        } else if (negativeIdCount > 0) {
-            // Some fields have ID -1 and some don't, throw error
-            throw new IllegalArgumentException(
-                    "Field IDs must be either all -1 or all non-negative. "
-                            + "Mixed field IDs are not allowed.");
-        } else {
-            final Set<Integer> duplicateIds =
-                    fieldIds.stream()
-                            .filter(id -> Collections.frequency(fieldIds, id) > 1)
-                            .collect(Collectors.toSet());
-            if (!duplicateIds.isEmpty()) {
-                throw new IllegalArgumentException(
-                        String.format(
-                                "Field IDs must be unique. Found duplicates: %s", duplicateIds));
-            }
-            processedFields = fields;
-        }
-
-        return processedFields;
     }
 
     public static RowType of(DataType... types) {
@@ -294,11 +230,7 @@ public static Builder builder() {
     }
 
     public static Builder builder(boolean isNullable) {
-        return new Builder(isNullable, new AtomicInteger(-1));
-    }
-
-    public static Builder builder(boolean isNullable, AtomicInteger fieldId) {
-        return new Builder(isNullable, fieldId);
+        return new Builder(isNullable);
     }
 
     /** Builder of {@link RowType}. */
@@ -307,20 +239,18 @@ public static class Builder {
         private final List<DataField> fields = new ArrayList<>();
 
         private final boolean isNullable;
-        private final AtomicInteger fieldId;
 
-        private Builder(boolean isNullable, AtomicInteger fieldId) {
+        private Builder(boolean isNullable) {
             this.isNullable = isNullable;
-            this.fieldId = fieldId;
         }
 
         public Builder field(String name, DataType type) {
-            fields.add(new DataField(name, type, fieldId.incrementAndGet()));
+            fields.add(new DataField(name, type));
             return this;
         }
 
         public Builder field(String name, DataType type, String description) {
-            fields.add(new DataField(name, type, description, fieldId.incrementAndGet()));
+            fields.add(new DataField(name, type, description));
             return this;
         }
 

fluss-common/src/main/java/org/apache/fluss/utils/json/DataTypeJsonSerde.java

@@ -295,10 +295,6 @@ private static DataType deserializeRow(JsonNode dataTypeNode) {
         final ArrayNode fieldNodes = (ArrayNode) dataTypeNode.get(FIELD_NAME_FIELDS);
         final List<DataField> fields = new ArrayList<>();
 
-        boolean hasFieldWithId = false;
-        boolean hasFieldWithoutId = false;
-        int autoFieldId = 0;
-
         for (JsonNode fieldNode : fieldNodes) {
             final String fieldName = fieldNode.get(FIELD_NAME_FIELD_NAME).asText();
             final DataType fieldType =
@@ -308,24 +304,13 @@ private static DataType deserializeRow(JsonNode dataTypeNode) {
                             ? fieldNode.get(FIELD_NAME_FIELD_DESCRIPTION).asText()
                             : null;
 
-            final int fieldId;
-            if (fieldNode.has(FIELD_NAME_FIELD_ID)) {
-                hasFieldWithId = true;
-                fieldId = fieldNode.get(FIELD_NAME_FIELD_ID).asInt();
-            } else {
-                hasFieldWithoutId = true;
-                fieldId = autoFieldId++;
-            }
-
+            final int fieldId =
+                    fieldNode.has(FIELD_NAME_FIELD_ID)
+                            ? fieldNode.get(FIELD_NAME_FIELD_ID).asInt()
+                            : -1;
             fields.add(new DataField(fieldName, fieldType, fieldDescription, fieldId));
         }
 
-        if (hasFieldWithId && hasFieldWithoutId) {
-            throw new TableException(
-                    "Field ID inconsistency detected in row type: "
-                            + "all fields must either have field IDs or none should have field IDs.");
-        }
-
         return new RowType(fields);
     }
 }