[api] Require field ids for schema fields, auto-assign them for parameter lists

paimon-api/src/main/java/org/apache/paimon/types/DataTypeJsonParser.java

@@ -42,13 +42,19 @@ public static DataField parseDataField(JsonNode json) {
         return parseDataField(json, null);
     }
 
-    private static DataField parseDataField(JsonNode json, AtomicInteger fieldId) {
+    /**
+     * Parses a field, drawing its id from {@code fieldId} when the json carries none. Callers that
+     * parse a sequence of fields pass one counter for the whole sequence so the ids stay distinct;
+     * pass {@code null} to require an explicit id.
+     */
+    public static DataField parseDataField(JsonNode json, AtomicInteger fieldId) {
         int id;
         JsonNode idNode = json.get("id");
         if (idNode != null) {
             checkState(fieldId == null || fieldId.get() == -1, "Partial field id is not allowed.");
             id = idNode.asInt();
         } else {
+            checkState(fieldId != null, "Field id is required but the field carries none.");
             id = fieldId.incrementAndGet();
         }
         String name = json.get("name").asText();

paimon-api/src/test/java/org/apache/paimon/types/DataTypeJsonParserTest.java

@@ -0,0 +1,93 @@
+/*
+ * 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.paimon.types;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link DataTypeJsonParser}. */
+class DataTypeJsonParserTest {
+
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
+    @Test
+    void parseDataFieldWithoutIdAndWithoutCounterIsRejected() {
+        ObjectNode json = MAPPER.createObjectNode();
+        json.put("name", "x");
+        json.put("type", "INT");
+
+        // a table schema must carry its field ids: they drive projection and schema evolution,
+        // so silently assigning one would be worse than refusing to parse
+        assertThatThrownBy(() -> DataTypeJsonParser.parseDataField(json))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("Field id is required");
+    }
+
+    @Test
+    void parseDataFieldDrawsIdsFromOneCounter() {
+        AtomicInteger fieldId = new AtomicInteger(-1);
+
+        assertThat(DataTypeJsonParser.parseDataField(fieldJson("a"), fieldId).id()).isZero();
+        assertThat(DataTypeJsonParser.parseDataField(fieldJson("b"), fieldId).id()).isEqualTo(1);
+        assertThat(DataTypeJsonParser.parseDataField(fieldJson("c"), fieldId).id()).isEqualTo(2);
+    }
+
+    @Test
+    void parseDataFieldKeepsExplicitId() {
+        ObjectNode json = MAPPER.createObjectNode();
+        json.put("id", 7);
+        json.put("name", "x");
+        json.put("type", "INT");
+
+        DataField field = DataTypeJsonParser.parseDataField(json);
+        assertThat(field.id()).isEqualTo(7);
+    }
+
+    @Test
+    void parseRowWithoutFieldIdsAutoAssignsSequentially() throws Exception {
+        JsonNode json =
+                MAPPER.readTree(
+                        "{\"type\":\"ROW\",\"fields\":[{\"name\":\"a\",\"type\":\"INT\"},"
+                                + "{\"name\":\"b\",\"type\":\"STRING\"}]}");
+
+        DataType type = DataTypeJsonParser.parseDataType(json);
+        assertThat(type)
+                .isEqualTo(
+                        new RowType(
+                                Arrays.asList(
+                                        new DataField(0, "a", new IntType()),
+                                        new DataField(1, "b", DataTypes.STRING()))));
+    }
+
+    private static ObjectNode fieldJson(String name) {
+        ObjectNode json = MAPPER.createObjectNode();
+        json.put("name", name);
+        json.put("type", "INT");
+        return json;
+    }
+}

paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java

@@ -34,6 +34,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -141,12 +142,26 @@ public static List<DataField> parseDataFieldArray(String data) {
         if (data != null) {
             JsonNode jsonArray = JsonSerdeUtil.fromJson(data, JsonNode.class);
             if (jsonArray.isArray()) {
+                // A counter only for a list that carries no ids at all, and one counter for the
+                // whole list so each field gets its own. Supplying it when some field already has
+                // an id would let the rest silently draw a colliding one, so in that case pass
+                // null and let the parser reject the list.
+                AtomicInteger fieldId = carriesAnyFieldId(jsonArray) ? null : new AtomicInteger(-1);
                 for (JsonNode objNode : jsonArray) {
-                    DataField dataField = DataTypeJsonParser.parseDataField(objNode);
+                    DataField dataField = DataTypeJsonParser.parseDataField(objNode, fieldId);
                     list.add(dataField);
                 }
             }
         }
         return list;
     }
+
+    private static boolean carriesAnyFieldId(JsonNode jsonArray) {
+        for (JsonNode objNode : jsonArray) {
+            if (objNode.get("id") != null) {
+                return true;
+            }
+        }
+        return false;
+    }
 }

paimon-common/src/test/java/org/apache/paimon/utils/ParameterUtilsTest.java

@@ -18,16 +18,79 @@
 
 package org.apache.paimon.utils;
 
+import org.apache.paimon.types.DataField;
+
 import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
+import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link ParameterUtils}. */
 class ParameterUtilsTest {
 
+    @Test
+    void testParseDataFieldArrayWithoutIds() {
+        // create_function passes a user-written parameter list, which may omit the ids; each
+        // field still has to get its own instead of every one landing on 0
+        List<DataField> fields =
+                ParameterUtils.parseDataFieldArray(
+                        "[{\"name\":\"a\",\"type\":\"INT\"},"
+                                + "{\"name\":\"b\",\"type\":\"STRING\"},"
+                                + "{\"name\":\"c\",\"type\":\"BIGINT\"}]");
+
+        assertThat(fields).extracting(DataField::id).containsExactly(0, 1, 2);
+        assertThat(fields).extracting(DataField::name).containsExactly("a", "b", "c");
+    }
+
+    @Test
+    void testParseDataFieldArrayKeepsExplicitIds() {
+        List<DataField> fields =
+                ParameterUtils.parseDataFieldArray(
+                        "[{\"id\":3,\"name\":\"a\",\"type\":\"INT\"},"
+                                + "{\"id\":9,\"name\":\"b\",\"type\":\"STRING\"}]");
+
+        assertThat(fields).extracting(DataField::id).containsExactly(3, 9);
+    }
+
+    @Test
+    void testParseDataFieldArrayRejectsPartialIds() {
+        // both orders must be rejected: supplying a counter to a list that already carries an id
+        // would let the id-less fields silently draw a colliding one
+        assertThatThrownBy(
+                        () ->
+                                ParameterUtils.parseDataFieldArray(
+                                        "[{\"name\":\"a\",\"type\":\"INT\"},"
+                                                + "{\"id\":7,\"name\":\"b\",\"type\":\"STRING\"}]"))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("Field id is required");
+
+        assertThatThrownBy(
+                        () ->
+                                ParameterUtils.parseDataFieldArray(
+                                        "[{\"id\":0,\"name\":\"a\",\"type\":\"INT\"},"
+                                                + "{\"name\":\"b\",\"type\":\"STRING\"}]"))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("Field id is required");
+    }
+
+    @Test
+    void testParseDataFieldArrayRejectsIdLessNestedField() {
+        // a nested row inside an explicitly numbered list would otherwise draw id 0 and collide
+        // with the first top-level field
+        assertThatThrownBy(
+                        () ->
+                                ParameterUtils.parseDataFieldArray(
+                                        "[{\"id\":0,\"name\":\"a\",\"type\":\"INT\"},"
+                                                + "{\"id\":1,\"name\":\"b\",\"type\":"
+                                                + "{\"type\":\"ROW\",\"fields\":"
+                                                + "[{\"name\":\"x\",\"type\":\"INT\"}]}}]"))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("Field id is required");
+    }
+
     @Test
     void testParseIntegerRanges() {
         assertThat(ParameterUtils.parseIntegerRanges("0-2, 4, 2, 6 - 7", 8))