fix(variant): write parquet field ids in shredded physical schema

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

@@ -31,12 +31,22 @@
 #include "paimon/common/data/variant/variant_schema.h"
 #include "paimon/common/data/variant/variant_shredding_utils.h"
 #include "paimon/common/data/variant/variant_shredding_writer.h"
+#include "paimon/common/types/data_field.h"
 #include "paimon/common/utils/checked_cast.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
 
+namespace {
+
+int32_t ShreddedFieldId(const std::shared_ptr<arrow::Field>& field) {
+    Result<DataField> data_field = DataField::ConvertArrowFieldToDataField(field);
+    return data_field.ok() ? data_field.value().Id() : -1;
+}
+
+}  // namespace
+
 class VariantShreddingTest : public ::testing::Test {
  public:
     // Shreds the given JSON documents (nullptr = null variant) with the given shredding type,
@@ -217,6 +227,102 @@ TEST_F(VariantShreddingTest, ShreddingSchemaShape) {
         VariantShreddingUtils::VariantShreddingSchema(arrow::map(arrow::utf8(), arrow::int32())));
 }
 
+TEST_F(VariantShreddingTest, ShreddingSchemaFieldIds) {
+    auto shredding_type = arrow::struct_(
+        {arrow::field("a", arrow::int32()), arrow::field("b", arrow::list(arrow::utf8())),
+         arrow::field("c", arrow::struct_({arrow::field("d", arrow::null())}))});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> physical,
+                         VariantShreddingUtils::VariantShreddingSchema(shredding_type));
+    ASSERT_EQ(physical->num_fields(), 3);
+    ASSERT_EQ(ShreddedFieldId(physical->field(0)), 0);
+    ASSERT_EQ(ShreddedFieldId(physical->field(1)), 1);
+    ASSERT_EQ(ShreddedFieldId(physical->field(2)), 2);
+
+    const std::shared_ptr<arrow::DataType>& object = physical->field(2)->type();
+    ASSERT_EQ(object->num_fields(), 3);
+    ASSERT_EQ(ShreddedFieldId(object->field(0)), 0);
+    ASSERT_EQ(ShreddedFieldId(object->field(1)), 1);
+    ASSERT_EQ(ShreddedFieldId(object->field(2)), 2);
+
+    const std::shared_ptr<arrow::DataType>& scalar = object->field(0)->type();
+    ASSERT_EQ(scalar->num_fields(), 2);
+    ASSERT_EQ(ShreddedFieldId(scalar->field(0)), 0);
+    ASSERT_EQ(ShreddedFieldId(scalar->field(1)), 1);
+
+    const std::shared_ptr<arrow::DataType>& array = object->field(1)->type();
+    ASSERT_EQ(array->num_fields(), 2);
+    ASSERT_EQ(ShreddedFieldId(array->field(0)), 0);
+    ASSERT_EQ(ShreddedFieldId(array->field(1)), 1);
+    ASSERT_EQ(array->field(1)->type()->id(), arrow::Type::LIST);
+    const std::shared_ptr<arrow::Field>& element = array->field(1)->type()->field(0);
+    ASSERT_EQ(ShreddedFieldId(element), 536871936);
+    ASSERT_EQ(element->type()->num_fields(), 2);
+    ASSERT_EQ(ShreddedFieldId(element->type()->field(0)), 0);
+    ASSERT_EQ(ShreddedFieldId(element->type()->field(1)), 1);
+
+    const std::shared_ptr<arrow::DataType>& nested = object->field(2)->type();
+    ASSERT_EQ(nested->num_fields(), 2);
+    ASSERT_EQ(ShreddedFieldId(nested->field(0)), 0);
+    ASSERT_EQ(ShreddedFieldId(nested->field(1)), 1);
+    const std::shared_ptr<arrow::DataType>& nested_object = nested->field(1)->type();
+    ASSERT_EQ(nested_object->num_fields(), 1);
+    ASSERT_EQ(ShreddedFieldId(nested_object->field(0)), 0);
+    const std::shared_ptr<arrow::DataType>& untyped = nested_object->field(0)->type();
+    ASSERT_EQ(untyped->num_fields(), 1);
+    ASSERT_EQ(ShreddedFieldId(untyped->field(0)), 0);
+
+    auto configured = arrow::struct_(
+        {arrow::field("a", arrow::int32(), true,
+                      arrow::KeyValueMetadata::Make({DataField::FIELD_ID}, {"7"})),
+         arrow::field("b", arrow::utf8(), true,
+                      arrow::KeyValueMetadata::Make({DataField::DESCRIPTION}, {"description"}))});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> configured_physical,
+                         VariantShreddingUtils::VariantShreddingSchema(configured));
+    ASSERT_EQ(configured_physical->num_fields(), 3);
+    ASSERT_EQ(configured_physical->field(2)->type()->num_fields(), 2);
+    ASSERT_EQ(ShreddedFieldId(configured_physical->field(2)->type()->field(0)), 7);
+    ASSERT_EQ(ShreddedFieldId(configured_physical->field(2)->type()->field(1)), 1);
+}
+
+TEST_F(VariantShreddingTest, ShreddingSchemaRejectsInvalidFieldIds) {
+    for (const std::string field_id : {"invalid", "", "2147483648"}) {
+        SCOPED_TRACE(field_id);
+        auto shredding_type = arrow::struct_(
+            {arrow::field("a", arrow::int32(), true,
+                          arrow::KeyValueMetadata::Make({DataField::FIELD_ID}, {field_id}))});
+        ASSERT_NOK_WITH_MSG(VariantShreddingUtils::VariantShreddingSchema(shredding_type),
+                            "cannot cast field id");
+    }
+}
+
+TEST_F(VariantShreddingTest, ShreddingArraySchemaFieldIds) {
+    for (const auto& leaf_type : {arrow::utf8(), arrow::null()}) {
+        SCOPED_TRACE(leaf_type->ToString());
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<arrow::DataType> physical,
+            VariantShreddingUtils::VariantShreddingSchema(arrow::list(arrow::list(leaf_type))));
+        ASSERT_EQ(physical->num_fields(), 3);
+        ASSERT_EQ(ShreddedFieldId(physical->field(0)), 0);
+        ASSERT_EQ(ShreddedFieldId(physical->field(1)), 1);
+        ASSERT_EQ(ShreddedFieldId(physical->field(2)), 2);
+        ASSERT_EQ(physical->field(2)->type()->id(), arrow::Type::LIST);
+        const auto& outer_element = physical->field(2)->type()->field(0);
+        ASSERT_EQ(ShreddedFieldId(outer_element), 536872960);
+        ASSERT_EQ(outer_element->type()->num_fields(), 2);
+        ASSERT_EQ(ShreddedFieldId(outer_element->type()->field(0)), 0);
+        ASSERT_EQ(ShreddedFieldId(outer_element->type()->field(1)), 1);
+        const auto& inner_array = outer_element->type()->field(1)->type();
+        ASSERT_EQ(inner_array->id(), arrow::Type::LIST);
+        const auto& inner_element = inner_array->field(0);
+        ASSERT_EQ(ShreddedFieldId(inner_element), 536871936);
+        ASSERT_EQ(inner_element->type()->num_fields(), leaf_type->id() == arrow::Type::NA ? 1 : 2);
+        ASSERT_EQ(ShreddedFieldId(inner_element->type()->field(0)), 0);
+        if (leaf_type->id() != arrow::Type::NA) {
+            ASSERT_EQ(ShreddedFieldId(inner_element->type()->field(1)), 1);
+        }
+    }
+}
+
 TEST_F(VariantShreddingTest, ShredObject) {
     // Mirrors the Java GenericVariantTest#testShredding scenarios.
     auto variant_json = R"({"a": 1, "b": "hello"})";

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

@@ -31,6 +31,8 @@
 #include "fmt/format.h"
 #include "paimon/common/data/variant/variant_defs.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
 #include "paimon/common/utils/checked_cast.h"
 
 namespace paimon {
@@ -42,14 +44,34 @@ Status InvalidVariantShreddingSchema(const std::shared_ptr<arrow::DataType>& typ
         fmt::format("Invalid variant shredding schema: {}", type ? type->ToString() : "null"));
 }
 
+std::shared_ptr<arrow::KeyValueMetadata> FieldIdMetadata(int32_t field_id) {
+    return arrow::KeyValueMetadata::Make({DataField::FIELD_ID}, {std::to_string(field_id)});
+}
+
+// Preserve configured IDs; inferred object fields use their positions, as in Java RowType.
+Result<int32_t> GetObjectFieldId(const std::shared_ptr<arrow::Field>& field, int32_t index) {
+    if (!field->metadata() || !field->metadata()->Contains(DataField::FIELD_ID)) {
+        return index;
+    }
+    PAIMON_ASSIGN_OR_RAISE(DataField data_field, DataField::ConvertArrowFieldToDataField(field));
+    return data_field.Id();
+}
+
 // Mirrors the Java `PaimonShreddingUtils.variantShreddingSchema(dataType, isTopLevel,
 // isObjectField)`.
 Result<std::shared_ptr<arrow::DataType>> VariantShreddingSchemaImpl(
     const std::shared_ptr<arrow::DataType>& data_type, bool is_top_level, bool is_object_field) {
     arrow::FieldVector fields;
+    // Java's Parquet reader requires field IDs; RowType.builder() numbers each row from zero.
+    int32_t next_field_id = 0;
+    auto shredded_field = [&next_field_id](const std::string& name,
+                                           const std::shared_ptr<arrow::DataType>& type,
+                                           bool nullable) {
+        return arrow::field(name, type, nullable, FieldIdMetadata(next_field_id++));
+    };
     if (is_top_level) {
-        fields.push_back(arrow::field(VariantDefs::kMetadataFieldName, arrow::binary(),
-                                      /*nullable=*/false));
+        fields.push_back(shredded_field(VariantDefs::kMetadataFieldName, arrow::binary(),
+                                        /*nullable=*/false));
     }
     switch (data_type->id()) {
         case arrow::Type::LIST: {
@@ -58,10 +80,15 @@ Result<std::shared_ptr<arrow::DataType>> VariantShreddingSchemaImpl(
                                    VariantShreddingSchemaImpl(list_type->value_type(),
                                                               /*is_top_level=*/false,
                                                               /*is_object_field=*/false));
-            fields.push_back(
-                arrow::field(VariantDefs::kValueFieldName, arrow::binary(), /*nullable=*/true));
-            fields.push_back(arrow::field(VariantDefs::kTypedValueFieldName,
-                                          arrow::list(element_type), /*nullable=*/true));
+            fields.push_back(shredded_field(VariantDefs::kValueFieldName, arrow::binary(),
+                                            /*nullable=*/true));
+            // Java resets collection depth at each ROW; every shredded array element is a ROW.
+            const int32_t element_id =
+                SpecialFields::GetArrayElementFieldId(next_field_id, /*depth=*/1);
+            std::shared_ptr<arrow::Field> element_field =
+                arrow::list(element_type)->field(0)->WithMetadata(FieldIdMetadata(element_id));
+            fields.push_back(shredded_field(VariantDefs::kTypedValueFieldName,
+                                            arrow::list(element_field), /*nullable=*/true));
             break;
         }
         case arrow::Type::STRUCT: {
@@ -70,27 +97,30 @@ Result<std::shared_ptr<arrow::DataType>> VariantShreddingSchemaImpl(
             // "value" and "typed_value" to null.
             const auto& struct_type = checked_pointer_cast<arrow::StructType>(data_type);
             arrow::FieldVector shredded_fields;
-            for (const auto& field : struct_type->fields()) {
+            for (int32_t index = 0; index < struct_type->num_fields(); ++index) {
+                const std::shared_ptr<arrow::Field>& field = struct_type->field(index);
+                PAIMON_ASSIGN_OR_RAISE(int32_t field_id, GetObjectFieldId(field, index));
                 PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::DataType> field_type,
                                        VariantShreddingSchemaImpl(field->type(),
                                                                   /*is_top_level=*/false,
                                                                   /*is_object_field=*/true));
-                shredded_fields.push_back(
-                    arrow::field(field->name(), field_type, /*nullable=*/false));
+                shredded_fields.push_back(arrow::field(field->name(), field_type,
+                                                       /*nullable=*/false,
+                                                       FieldIdMetadata(field_id)));
             }
-            fields.push_back(
-                arrow::field(VariantDefs::kValueFieldName, arrow::binary(), /*nullable=*/true));
-            fields.push_back(arrow::field(VariantDefs::kTypedValueFieldName,
-                                          arrow::struct_(shredded_fields), /*nullable=*/true));
+            fields.push_back(shredded_field(VariantDefs::kValueFieldName, arrow::binary(),
+                                            /*nullable=*/true));
+            fields.push_back(shredded_field(VariantDefs::kTypedValueFieldName,
+                                            arrow::struct_(shredded_fields), /*nullable=*/true));
             break;
         }
         case arrow::Type::NA: {
             // `arrow::null()` denotes an untyped VARIANT leaf in shredding types. It doesn't
             // need a typed column. If there is no typed column, value is required for array
             // elements or top-level fields, but optional for objects (where a null represents a
             // missing field).
-            fields.push_back(arrow::field(VariantDefs::kValueFieldName, arrow::binary(),
-                                          /*nullable=*/is_object_field));
+            fields.push_back(shredded_field(VariantDefs::kValueFieldName, arrow::binary(),
+                                            /*nullable=*/is_object_field));
             break;
         }
         case arrow::Type::STRING:
@@ -103,10 +133,10 @@ Result<std::shared_ptr<arrow::DataType>> VariantShreddingSchemaImpl(
         case arrow::Type::INT64:
         case arrow::Type::FLOAT:
         case arrow::Type::DOUBLE: {
-            fields.push_back(
-                arrow::field(VariantDefs::kValueFieldName, arrow::binary(), /*nullable=*/true));
-            fields.push_back(arrow::field(VariantDefs::kTypedValueFieldName, data_type,
-                                          /*nullable=*/true));
+            fields.push_back(shredded_field(VariantDefs::kValueFieldName, arrow::binary(),
+                                            /*nullable=*/true));
+            fields.push_back(shredded_field(VariantDefs::kTypedValueFieldName, data_type,
+                                            /*nullable=*/true));
             break;
         }
         default:

src/paimon/common/table/special_fields.h

@@ -74,6 +74,17 @@ struct SpecialFields {
         return data_field;
     }
 
+    /// Field ID range and stride used by Java SpecialFields for structured types.
+    static constexpr int32_t STRUCTURED_TYPE_FIELD_ID_BASE =
+        std::numeric_limits<int32_t>::max() / 4;
+    static constexpr int32_t STRUCTURED_TYPE_FIELD_DEPTH_LIMIT = 1 << 10;
+
+    /// Match Java's Parquet array-element IDs. Depth starts at 1 and resets at each ROW.
+    static constexpr int32_t GetArrayElementFieldId(int32_t array_field_id, int32_t depth) {
+        return STRUCTURED_TYPE_FIELD_ID_BASE + array_field_id * STRUCTURED_TYPE_FIELD_DEPTH_LIMIT +
+               depth;
+    }
+
     static bool IsSystemField(const std::string& field_name) {
         if (StringUtils::StartsWith(field_name, KEY_FIELD_PREFIX)) {
             return true;

src/paimon/common/table/special_fields_test.cpp

@@ -66,6 +66,13 @@ TEST(SpecialFieldsTest, TestKeyValueSpecialFieldCount) {
     ASSERT_EQ(SpecialFields::KEY_VALUE_SPECIAL_FIELD_COUNT, 2);
 }
 
+TEST(SpecialFieldsTest, TestGetArrayElementFieldId) {
+    ASSERT_EQ(SpecialFields::STRUCTURED_TYPE_FIELD_ID_BASE, 536870911);
+    ASSERT_EQ(SpecialFields::STRUCTURED_TYPE_FIELD_DEPTH_LIMIT, 1024);
+    ASSERT_EQ(SpecialFields::GetArrayElementFieldId(1, 1), 536871936);
+    ASSERT_EQ(SpecialFields::GetArrayElementFieldId(2, 2), 536872961);
+}
+
 TEST(SpecialFieldsTest, TestIsSystemField) {
     ASSERT_TRUE(SpecialFields::IsSystemField("_SEQUENCE_NUMBER"));
     ASSERT_TRUE(SpecialFields::IsSystemField("_VALUE_KIND"));