fix(types): assign omitted field IDs in type JSON

docs/source/user_guide/data_types.rst

@@ -271,7 +271,10 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
        `Variant Shredding <https://github.com/apache/parquet-format/blob/master/VariantShredding.md>`_
        specification by setting ``variant.shreddingSchema`` to a ROW type JSON
        whose fields map top-level variant column names to their shredding
-       types. Alternatively, setting ``variant.inferShreddingSchema`` to
+       types. Explicit field IDs are preserved. If every ROW field, including
+       nested fields, omits ``id``, IDs are assigned in preorder starting at 0,
+       as in Java Paimon. Partially specified IDs are rejected.
+       Alternatively, setting ``variant.inferShreddingSchema`` to
        ``true`` infers a shredding schema per file from the first written rows
        (tuned by ``variant.shredding.maxSchemaWidth``, which bounds the total
        number of shredded fields across all variant columns of the schema,

include/paimon/defs.h

@@ -511,7 +511,9 @@ struct PAIMON_EXPORT Options {
     static const char MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY[];
 
     /// "variant.shreddingSchema" - The Variant shredding schema for writing: a ROW type JSON
-    /// whose fields map variant column names to their shredding types. No default value.
+    /// whose fields map variant column names to their shredding types. All ROW fields, including
+    /// nested fields, must all specify 'id' or all omit it. Omitted IDs are assigned in preorder
+    /// starting at 0. No default value.
     static const char VARIANT_SHREDDING_SCHEMA[];
     /// "parquet.variant.shreddingSchema" - Fallback key of "variant.shreddingSchema".
     static const char PARQUET_VARIANT_SHREDDING_SCHEMA[];

src/paimon/common/data/variant/variant_shredding_write_plan.h

@@ -75,7 +75,8 @@ class VariantShreddingWritePlan {
 
     /// Creates a plan from the `variant.shreddingSchema` option value: a ROW type JSON whose
     /// fields map top-level variant column names to their shredding types (nested variant
-    /// columns cannot be configured, as in Java).
+    /// columns cannot be configured, as in Java). All ROW fields, including nested fields, must
+    /// all specify 'id' or all omit it. Omitted IDs are assigned in preorder starting at 0.
     static Result<std::shared_ptr<VariantShreddingWritePlan>> FromConfiguredSchema(
         const std::shared_ptr<arrow::Schema>& logical_schema,
         const std::string& configured_schema_json);

src/paimon/common/data/variant/variant_shredding_write_plan_factory_test.cpp

@@ -95,20 +95,63 @@ TEST_F(VariantShreddingWritePlanFactoryTest, ConfiguredSchema) {
             }
         } ]
     })";
-    ASSERT_OK_AND_ASSIGN(CoreOptions options,
-                         MakeOptions({{"variant.shreddingSchema", shredding_schema_json}}));
-    auto factory = VariantShreddingWritePlanFactory::Create(options, schema_, pool_);
-    ASSERT_TRUE(factory->ShouldCreateWritePlan());
-    ASSERT_FALSE(factory->ShouldInferWritePlan());
-    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ShreddingBatchConverter> converter,
-                         factory->CreateConverter("parquet", {}));
-    ASSERT_NE(converter, nullptr);
-    auto variant_field = converter->GetPhysicalSchema()->GetFieldByName("v");
-    ASSERT_NE(variant_field, nullptr);
-    const auto& physical_type = static_cast<const arrow::StructType&>(*variant_field->type());
-    ASSERT_NE(physical_type.GetFieldByName("typed_value"), nullptr);
-    // Variant shredding only supports the parquet format.
-    ASSERT_TRUE(factory->CreateConverter("orc", {}).status().IsNotImplemented());
+    const char* shredding_schema_json_without_ids = R"({
+        "type": "ROW",
+        "fields": [ {
+            "name": "v",
+            "type": {
+                "type": "ROW",
+                "fields": [
+                    {"name": "age", "type": "INT"},
+                    {"name": "city", "type": "STRING"}
+                ]
+            }
+        } ]
+    })";
+    std::shared_ptr<arrow::Schema> physical_schema;
+    for (const char* option_key : {"variant.shreddingSchema", "parquet.variant.shreddingSchema"}) {
+        SCOPED_TRACE(option_key);
+        for (const char* configured_schema :
+             {shredding_schema_json, shredding_schema_json_without_ids}) {
+            SCOPED_TRACE(configured_schema);
+            ASSERT_OK_AND_ASSIGN(CoreOptions options,
+                                 MakeOptions({{option_key, configured_schema}}));
+            auto factory = VariantShreddingWritePlanFactory::Create(options, schema_, pool_);
+            ASSERT_TRUE(factory->ShouldCreateWritePlan());
+            ASSERT_FALSE(factory->ShouldInferWritePlan());
+            ASSERT_OK_AND_ASSIGN(std::shared_ptr<ShreddingBatchConverter> converter,
+                                 factory->CreateConverter("parquet", {}));
+            ASSERT_NE(converter, nullptr);
+            auto variant_field = converter->GetPhysicalSchema()->GetFieldByName("v");
+            ASSERT_NE(variant_field, nullptr);
+            const auto& physical_type =
+                static_cast<const arrow::StructType&>(*variant_field->type());
+            ASSERT_NE(physical_type.GetFieldByName("typed_value"), nullptr);
+            if (physical_schema == nullptr) {
+                physical_schema = converter->GetPhysicalSchema();
+            } else {
+                ASSERT_TRUE(converter->GetPhysicalSchema()->Equals(*physical_schema,
+                                                                   /*check_metadata=*/true));
+            }
+            ASSERT_TRUE(factory->CreateConverter("orc", {}).status().IsNotImplemented());
+        }
+    }
+}
+
+TEST_F(VariantShreddingWritePlanFactoryTest, ConfiguredSchemaRejectsPartialFieldIds) {
+    for (const char* configured_schema : {
+             R"({"type":"ROW","fields":[{"id":0,"name":"v","type":{
+                    "type":"ROW","fields":[{"name":"age","type":"INT"}]}}]})",
+             R"({"type":"ROW","fields":[{"name":"v","type":{
+                    "type":"ROW","fields":[{"id":1,"name":"age","type":"INT"}]}}]})",
+         }) {
+        SCOPED_TRACE(configured_schema);
+        ASSERT_OK_AND_ASSIGN(CoreOptions options,
+                             MakeOptions({{"variant.shreddingSchema", configured_schema}}));
+        auto factory = VariantShreddingWritePlanFactory::Create(options, schema_, pool_);
+        ASSERT_NOK_WITH_MSG(factory->CreateConverter("parquet", {}),
+                            "Partial field id is not allowed.");
+    }
 }
 
 TEST_F(VariantShreddingWritePlanFactoryTest, InferredSchema) {

src/paimon/common/types/data_field.cpp

@@ -65,21 +65,13 @@ rapidjson::Value DataField::ToJson(rapidjson::Document::AllocatorType* allocator
 }
 
 void DataField::FromJson(const rapidjson::Value& obj) noexcept(false) {
-    id_ = RapidJsonUtil::DeserializeKeyValue<int32_t>(obj, "id");
-    auto name = RapidJsonUtil::DeserializeKeyValue<std::string>(obj, "name");
-    assert(obj.IsObject());
-    if (!obj.HasMember("type")) {
-        throw std::invalid_argument("key 'type' must exist");
-    }
-    auto field_result = DataTypeJsonParser::ParseType(name, obj["type"]);
+    // Serialized table schemas require explicit IDs for projection and schema evolution.
+    Result<DataField> field_result = DataTypeJsonParser::ParseDataField(obj);
     if (!field_result.ok()) {
-        throw std::invalid_argument(
-            fmt::format("parse data type failed, error msg: {}", field_result.status().ToString()));
+        throw std::invalid_argument(field_result.status().ToString());
     }
-    field_ = field_result.value();
+    *this = std::move(field_result).value();
     assert(field_);
-    description_ = RapidJsonUtil::DeserializeKeyValue<std::optional<std::string>>(
-        obj, "description", description_);
 }
 
 std::shared_ptr<arrow::Field> DataField::ConvertDataFieldToArrowField(const DataField& field) {

src/paimon/common/types/data_field_test.cpp

@@ -20,6 +20,7 @@
 #include "paimon/common/types/data_field.h"
 
 #include <stdexcept>
+#include <utility>
 
 #include "arrow/api.h"
 #include "gtest/gtest.h"
@@ -137,7 +138,8 @@ TEST_F(DataFieldTest, ConvertArrowFieldToDataField) {
 TEST_F(DataFieldTest, FromJson) {
     const char* json = R"({
     "id" : 0,
-    "name" : "f0",
+    "name" : "f0\u0000tail",
+    "description" : "d\u0000tail",
     "type" : {
       "type" : "ROW",
       "fields" : [ {
@@ -161,7 +163,8 @@ TEST_F(DataFieldTest, FromJson) {
     DataField field;
     field.FromJson(doc);
     EXPECT_EQ(field.Id(), 0);
-    EXPECT_EQ(field.Name(), "f0");
+    ASSERT_EQ(field.Name(), std::string("f0\0tail", 7));
+    ASSERT_EQ(field.Description(), std::string("d\0tail", 6));
     EXPECT_EQ(field.Type()->id(), arrow::Type::STRUCT);
 
     auto sub_fields = field.Type()->fields();
@@ -226,6 +229,30 @@ TEST_F(DataFieldTest, FromJsonFailed) {
 })";
         check_result(json_str, "parse data type failed, error msg: ");
     }
+    const std::vector<std::pair<const char*, const char*>> test_cases = {
+        {R"([{"id":0,"name":"a","type":"INT"}])", "data field must be an object"},
+        {R"({"id":"0","name":"a","type":"INT"})", "value of key 'id' must be int"},
+        {R"({"name":"a","type":"INT"})", "key 'id' must exist"},
+        {R"({"id":null,"name":"a","type":"INT"})", "key 'id' must exist"},
+        {R"({"id":0,"type":"INT"})", "key 'name' must exist"},
+        {R"({"id":0,"name":0,"type":"INT"})", "value of key 'name' must be string"},
+        {R"({"id":0,"name":"a"})", "key 'type' must exist"},
+        {R"({"id":0,"name":"a","type":"INT","description":0})",
+         "value of key 'description' must be string"},
+        {R"({"id":0,"name":"a","type":{"type":"ROW",
+               "fields":[{"name":"b","type":"INT"}]}})",
+         "key 'id' must exist"},
+        {R"({"id":0,"name":"a","type":{"type":"ARRAY","element":{"type":"ROW",
+               "fields":[{"name":"b","type":"INT"}]}}})",
+         "key 'id' must exist"},
+        {R"({"id":0,"name":"a","type":{"type":"MAP","key":"STRING",
+               "value":{"type":"ROW","fields":[{"name":"b","type":"INT"}]}}})",
+         "key 'id' must exist"},
+    };
+    for (const auto& [json, error_msg] : test_cases) {
+        SCOPED_TRACE(json);
+        ASSERT_NO_FATAL_FAILURE(check_result(json, error_msg));
+    }
 }
 
 TEST_F(DataFieldTest, ToJson) {