feat(variant): support nested-ROW variant shredding and align with Java

- Discover, infer and shred VARIANT columns nested inside ROW types on
  the write path and reassemble them on the read path (struct-only
  recursion by field-index path, as in Java)
- Share one variant.shredding.maxSchemaWidth budget across all variant
  columns of a schema (the Java MaxFields semantics)
- Parse and validate the variant.* options in CoreOptions::FromMap and
  create VariantShreddingWritePlanFactory from the validated options
  instead of deferring configuration errors to write time
- Write files unshredded when the configured shredding schema matches no
  variant column, as in Java
- Skip rows whose enclosing struct is null when sampling and shredding
  nested variants; child slot contents under a null ancestor are
  unspecified in Arrow and must not be decoded
- Accept microsecond TIMESTAMP typed_value columns when building variant
  schemas and reject other precisions
- Render DOUBLE/FLOAT to STRING casts with Java Double/Float.toString
  semantics and emit at least two significant digits in scientific form
  (Double.MIN_VALUE is 4.9E-324)
- Reassemble within the caller-provided memory pools, include an error
  message in MalformedVariant, and move output parameters to the end of
  parameter lists
- Use explicit types in PAIMON_ASSIGN_OR_RAISE, checked_cast for known
  array types, a pure-virtual CreateMetadataFinalizer, and document the
  ShreddingFileReader position in the reader chains
- Extend tests: CoreOptions variant options, shared-budget, nested and
  null-ancestor shredding, Java Double/Float.toString special values,
  variant-null cast semantics, deep-nested JSON and IO-exception
  integration cases

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: SteNicholas

docs/source/user_guide/data_types.rst

@@ -241,14 +241,17 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
        parquet columns per the parquet-format
        `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 variant column names to their shredding types.
-       Alternatively, setting ``variant.inferShreddingSchema`` to ``true``
-       infers a shredding schema per file from the first written rows
-       (tuned by ``variant.shredding.maxSchemaWidth``,
+       whose fields map top-level variant column names to their shredding
+       types. 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,
        ``variant.shredding.maxSchemaDepth``,
        ``variant.shredding.minFieldCardinalityRatio`` and
-       ``variant.shredding.maxInferBufferRow``). Readers reassemble shredded
-       columns transparently.
+       ``variant.shredding.maxInferBufferRow``). Inference also covers variant
+       columns nested inside ROW columns (variants inside arrays or maps stay
+       unshredded, as in Java Paimon). Readers reassemble shredded columns
+       transparently.
 
        When reading, instead of the full variant, specific paths can be
        extracted by replacing the variant column in the read schema with a

src/paimon/CMakeLists.txt

@@ -466,6 +466,7 @@ if(PAIMON_BUILD_TESTS)
                     common/data/variant/infer_variant_shredding_schema_test.cpp
                     common/data/variant/variant_shredding_write_plan_factory_test.cpp
                     common/data/variant/variant_get_test.cpp
+                    common/data/variant/variant_json_utils_test.cpp
                     common/data/variant/variant_shredding_test.cpp
                     common/data/variant/variant_test.cpp
                     common/data/variant/variant_type_utils_test.cpp

src/paimon/common/data/shredding/shredding_write_plan_factories.cpp

@@ -35,8 +35,7 @@ std::shared_ptr<ShreddingWritePlanFactory> ShreddingWritePlanFactories::SelectAc
             return map_factory;
         }
     }
-    auto variant_factory =
-        std::make_shared<VariantShreddingWritePlanFactory>(options, write_schema, pool);
+    auto variant_factory = VariantShreddingWritePlanFactory::Create(options, write_schema, pool);
     if (variant_factory->ShouldCreateWritePlan()) {
         return variant_factory;
     }

src/paimon/common/data/shredding/shredding_write_plan_factory.h

@@ -61,9 +61,7 @@ class ShreddingWritePlanFactory {
     /// The per-file metadata finalizer persisted into the file footer, or nullptr when the
     /// physical schema is self-describing (as it is for VARIANT shredding).
     virtual MetadataFinalizer CreateMetadataFinalizer(
-        const std::shared_ptr<ShreddingBatchConverter>& converter) const {
-        return nullptr;
-    }
+        const std::shared_ptr<ShreddingBatchConverter>& converter) const = 0;
 };
 
 }  // namespace paimon

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

@@ -25,6 +25,7 @@
 #include <utility>
 
 #include "arrow/util/base64.h"
+#include "fmt/format.h"
 #include "paimon/common/data/variant/variant_builder.h"
 #include "paimon/common/data/variant/variant_defs.h"
 #include "paimon/common/data/variant/variant_json_utils.h"
@@ -170,7 +171,7 @@ Status ToJsonImpl(std::string_view value, std::string_view metadata, int32_t pos
             return Status::OK();
         }
     }
-    return VariantBinaryUtil::MalformedVariant();
+    return VariantBinaryUtil::MalformedVariant("unknown variant value type in JSON rendering");
 }
 
 }  // namespace
@@ -187,7 +188,7 @@ Result<std::shared_ptr<GenericVariant>> GenericVariant::Create(std::shared_ptr<B
     // There is currently only one allowed version.
     if (metadata->size() < 1 || (static_cast<uint8_t>((*metadata)[0]) &
                                  VariantDefs::kVersionMask) != VariantDefs::kVersion) {
-        return VariantBinaryUtil::MalformedVariant();
+        return VariantBinaryUtil::MalformedVariant("unsupported variant metadata version");
     }
     // Don't attempt to use a Variant larger than 128 MiB. We'll never produce one, and it risks
     // memory instability.
@@ -389,7 +390,8 @@ Result<int32_t> GenericVariant::GetDictionaryIdAtIndex(int32_t index) const {
     PAIMON_ASSIGN_OR_RAISE(VariantBinaryUtil::ObjectInfo info,
                            VariantBinaryUtil::GetObjectInfo(raw, pos_));
     if (index < 0 || index >= info.num_elements) {
-        return VariantBinaryUtil::MalformedVariant();
+        return VariantBinaryUtil::MalformedVariant(fmt::format(
+            "object field index {} is out of bounds for {} fields", index, info.num_elements));
     }
     return VariantBinaryUtil::ReadUnsigned(raw, info.id_start + info.id_size * index, info.id_size);
 }

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

@@ -59,11 +59,6 @@ struct SimpleSchema {
     }
 };
 
-// Tracks the total number of shredded fields remaining, shared across the whole schema.
-struct MaxFields {
-    int32_t remaining;
-};
-
 std::shared_ptr<SimpleSchema> MergeSchema(const std::shared_ptr<SimpleSchema>& s1,
                                           const std::shared_ptr<SimpleSchema>& s2);
 
@@ -180,9 +175,10 @@ Result<std::shared_ptr<SimpleSchema>> SchemaOf(const GenericVariant& variant, in
             result->is_object = true;
             result->fields.reserve(size);
             for (int32_t i = 0; i < size; ++i) {
-                PAIMON_ASSIGN_OR_RAISE(auto field, variant.GetFieldAtIndex(i));
+                PAIMON_ASSIGN_OR_RAISE(std::optional<GenericVariant::ObjectField> field,
+                                       variant.GetFieldAtIndex(i));
                 if (!field.has_value()) {
-                    return VariantBinaryUtil::MalformedVariant();
+                    return VariantBinaryUtil::MalformedVariant("an object field is missing");
                 }
                 PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<SimpleSchema> field_schema,
                                        SchemaOf(*field->value, max_depth - 1));
@@ -275,9 +271,9 @@ Result<std::shared_ptr<SimpleSchema>> SchemaOf(const GenericVariant& variant, in
 
 // Finalizes the inferred schema: 1) widen integer types to int64, 2) replace empty objects with
 // VARIANT, 3) limit the total number of shredded fields in the schema.
-std::shared_ptr<arrow::DataType> FinalizeSimpleSchema(const std::shared_ptr<SimpleSchema>& schema,
-                                                      int64_t min_cardinality,
-                                                      MaxFields* max_fields) {
+std::shared_ptr<arrow::DataType> FinalizeSimpleSchema(
+    const std::shared_ptr<SimpleSchema>& schema, int64_t min_cardinality,
+    InferVariantShreddingSchema::MaxFields* max_fields) {
     // Every field uses a value column.
     --max_fields->remaining;
     if (max_fields->remaining <= 0) {
@@ -332,7 +328,7 @@ std::shared_ptr<arrow::DataType> FinalizeSimpleSchema(const std::shared_ptr<Simp
 }  // namespace
 
 Result<std::shared_ptr<arrow::DataType>> InferVariantShreddingSchema::InferColumnShreddingType(
-    const std::vector<std::shared_ptr<GenericVariant>>& samples) const {
+    const std::vector<std::shared_ptr<GenericVariant>>& samples, MaxFields* max_fields) const {
     int64_t num_non_null_values = 0;
     std::shared_ptr<SimpleSchema> simple_schema;
     for (const auto& sample : samples) {
@@ -348,9 +344,8 @@ Result<std::shared_ptr<arrow::DataType>> InferVariantShreddingSchema::InferColum
     // the rows.
     auto min_cardinality = static_cast<int64_t>(
         std::ceil(static_cast<double>(num_non_null_values) * min_field_cardinality_ratio_));
-    MaxFields max_fields{max_schema_width_};
     std::shared_ptr<arrow::DataType> finalized =
-        FinalizeSimpleSchema(simple_schema, min_cardinality, &max_fields);
+        FinalizeSimpleSchema(simple_schema, min_cardinality, max_fields);
     if (finalized->id() == arrow::Type::NA) {
         // The whole column stays unshredded.
         return std::shared_ptr<arrow::DataType>(nullptr);

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

@@ -34,18 +34,30 @@ namespace paimon {
 /// fields is limited.
 class InferVariantShreddingSchema {
  public:
+    /// The mutable budget of shredded fields remaining. One instance is shared across all
+    /// variant columns of a schema so that the total inferred width stays within
+    /// `variant.shredding.maxSchemaWidth` (mirroring the Java `MaxFields`).
+    struct MaxFields {
+        int32_t remaining;
+    };
+
     InferVariantShreddingSchema(int32_t max_schema_width, int32_t max_schema_depth,
                                 double min_field_cardinality_ratio)
         : max_schema_width_(max_schema_width),
           max_schema_depth_(max_schema_depth),
           min_field_cardinality_ratio_(min_field_cardinality_ratio) {}
 
+    /// Creates the shared shredded-field budget for one schema inference.
+    MaxFields CreateMaxFieldsBudget() const {
+        return MaxFields{max_schema_width_};
+    }
+
     /// Infers the shredding type of one variant column from its sampled non-null values, e.g.
     /// `struct{a: int64, b: string}`. `arrow::null()` leaves denote untyped variant sub-values.
-    /// Returns nullptr when no useful shredding schema was found (the column should stay
-    /// unshredded).
+    /// `max_fields` is the budget shared across all columns of the schema. Returns nullptr when
+    /// no useful shredding schema was found (the column should stay unshredded).
     Result<std::shared_ptr<arrow::DataType>> InferColumnShreddingType(
-        const std::vector<std::shared_ptr<GenericVariant>>& samples) const;
+        const std::vector<std::shared_ptr<GenericVariant>>& samples, MaxFields* max_fields) const;
 
  private:
     int32_t max_schema_width_;

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

@@ -30,6 +30,14 @@ namespace paimon::test {
 
 class InferVariantShreddingSchemaTest : public ::testing::Test {
  public:
+    // Infers one column with a fresh shared-width budget.
+    static Result<std::shared_ptr<arrow::DataType>> InferColumn(
+        const InferVariantShreddingSchema& infer,
+        const std::vector<std::shared_ptr<GenericVariant>>& samples) {
+        InferVariantShreddingSchema::MaxFields max_fields = infer.CreateMaxFieldsBudget();
+        return infer.InferColumnShreddingType(samples, &max_fields);
+    }
+
     std::vector<std::shared_ptr<GenericVariant>> Samples(const std::vector<const char*>& jsons) {
         std::vector<std::shared_ptr<GenericVariant>> samples;
         for (const char* json : jsons) {
@@ -57,8 +65,7 @@ TEST_F(InferVariantShreddingSchemaTest, InferObjectSchema) {
         nullptr,
         R"({"age": 120000000000, "city": "Shanghai"})",
     });
-    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
-                         infer_.InferColumnShreddingType(samples));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred, InferColumn(infer_, samples));
     ASSERT_NE(inferred, nullptr);
     // Integers widen to int64, strings stay, arrays of small ints infer as list<int64>.
     auto expected =
@@ -73,8 +80,7 @@ TEST_F(InferVariantShreddingSchemaTest, MixedTypesFallToVariant) {
         R"({"x": 1, "y": 1.5e0})",
         R"({"x": "string now", "y": 2.5e0})",
     });
-    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
-                         infer_.InferColumnShreddingType(samples));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred, InferColumn(infer_, samples));
     ASSERT_NE(inferred, nullptr);
     // x saw both int and string: untyped variant leaf; y stays double (exponent notation
     // parses as double, plain decimals parse as DECIMAL).
@@ -91,8 +97,7 @@ TEST_F(InferVariantShreddingSchemaTest, RareFieldsDropped) {
     }
     jsons.push_back(R"({"common": 2, "rare": true})");
     auto samples = Samples(jsons);
-    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
-                         infer_.InferColumnShreddingType(samples));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred, InferColumn(infer_, samples));
     ASSERT_NE(inferred, nullptr);
     // "rare" appears in 1/20 rows (< 0.1 ratio): dropped from the typed schema.
     auto expected = arrow::struct_({arrow::field("common", arrow::int64())});
@@ -105,8 +110,7 @@ TEST_F(InferVariantShreddingSchemaTest, DecimalMerging) {
         "{\"d\": 1.5}",
         "{\"d\": 42}",
     });
-    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
-                         infer_.InferColumnShreddingType(samples));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred, InferColumn(infer_, samples));
     ASSERT_NE(inferred, nullptr);
     // Decimals merge to a widened decimal (scale 2, enough integer digits), capped at 18 digits.
     auto expected = arrow::struct_({arrow::field("d", arrow::decimal128(18, 2))});
@@ -116,20 +120,20 @@ TEST_F(InferVariantShreddingSchemaTest, DecimalMerging) {
 TEST_F(InferVariantShreddingSchemaTest, NoUsefulSchema) {
     auto scalar_samples = Samples({"1", "2"});
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> scalar_inferred,
-                         infer_.InferColumnShreddingType(scalar_samples));
+                         InferColumn(infer_, scalar_samples));
     ASSERT_NE(scalar_inferred, nullptr);
     ASSERT_TRUE(scalar_inferred->Equals(*arrow::int64())) << scalar_inferred->ToString();
 
     // Conflicting top-level types stay unshredded.
     auto mixed_samples = Samples({"1", "\"a string\""});
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> mixed_inferred,
-                         infer_.InferColumnShreddingType(mixed_samples));
+                         InferColumn(infer_, mixed_samples));
     ASSERT_EQ(mixed_inferred, nullptr);
 
     // All-null columns stay unshredded.
     auto null_samples = Samples({nullptr, nullptr});
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> null_inferred,
-                         infer_.InferColumnShreddingType(null_samples));
+                         InferColumn(infer_, null_samples));
     ASSERT_EQ(null_inferred, nullptr);
 }
 
@@ -138,19 +142,33 @@ TEST_F(InferVariantShreddingSchemaTest, MaxSchemaWidthLimit) {
                                              /*min_field_cardinality_ratio=*/0.1};
     auto samples = Samples({R"({"a": 1, "b": 2, "c": 3, "d": 4, "e": 5})"});
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
-                         narrow_infer.InferColumnShreddingType(samples));
-    if (inferred != nullptr) {
-        ASSERT_EQ(inferred->id(), arrow::Type::STRUCT);
-        ASSERT_LT(inferred->num_fields(), 5);
-    }
+                         InferColumn(narrow_infer, samples));
+    ASSERT_NE(inferred, nullptr);
+    // Budget of 3: the root object costs 1, "a" costs 2 (value + typed_value); the remaining
+    // fields exceed the budget and are dropped from the typed schema.
+    auto expected = arrow::struct_({arrow::field("a", arrow::int64())});
+    ASSERT_TRUE(inferred->Equals(*expected)) << inferred->ToString();
+
+    // One budget serves all variant columns of a schema: after the first column consumes it, a
+    // second column cannot shred anymore.
+    InferVariantShreddingSchema::MaxFields max_fields = narrow_infer.CreateMaxFieldsBudget();
+    auto first_samples = Samples({R"({"a": 1, "b": 2})"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> first,
+                         narrow_infer.InferColumnShreddingType(first_samples, &max_fields));
+    ASSERT_NE(first, nullptr);
+    ASSERT_TRUE(first->Equals(*expected)) << first->ToString();
+    auto second_samples = Samples({R"({"c": 1})"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> second,
+                         narrow_infer.InferColumnShreddingType(second_samples, &max_fields));
+    ASSERT_EQ(second, nullptr);
 }
 
 TEST_F(InferVariantShreddingSchemaTest, MaxSchemaDepthLimit) {
     InferVariantShreddingSchema shallow_infer{/*max_schema_width=*/300, /*max_schema_depth=*/1,
                                               /*min_field_cardinality_ratio=*/0.1};
     auto samples = Samples({R"({"outer": {"inner": 1}})"});
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
-                         shallow_infer.InferColumnShreddingType(samples));
+                         InferColumn(shallow_infer, samples));
     ASSERT_NE(inferred, nullptr);
     // Depth 1: the nested object stays an untyped variant leaf.
     auto expected = arrow::struct_({arrow::field("outer", arrow::null())});
@@ -162,14 +180,13 @@ TEST_F(InferVariantShreddingSchemaTest, TrailingZeroDecimalNormalized) {
     // scale or reassembling the shredded file would be rejected. After normalization the value
     // is an integral decimal, which finalization widens to int64.
     auto samples = Samples({"100.00"});
-    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
-                         infer_.InferColumnShreddingType(samples));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred, InferColumn(infer_, samples));
     ASSERT_NE(inferred, nullptr);
     ASSERT_TRUE(inferred->Equals(*arrow::int64())) << inferred->ToString();
 
     auto mixed = Samples({"100.00", "1.5"});
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> mixed_inferred,
-                         infer_.InferColumnShreddingType(mixed));
+                         InferColumn(infer_, mixed));
     ASSERT_NE(mixed_inferred, nullptr);
     ASSERT_TRUE(mixed_inferred->Equals(*arrow::decimal128(18, 1))) << mixed_inferred->ToString();
 }
@@ -181,14 +198,14 @@ TEST_F(InferVariantShreddingSchemaTest, TemporalValuesStayUnshredded) {
     ASSERT_OK(date_builder.AppendDate(19000));
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<GenericVariant> date_variant, date_builder.Build(pool_));
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> date_inferred,
-                         infer_.InferColumnShreddingType({date_variant}));
+                         InferColumn(infer_, {date_variant}));
     ASSERT_EQ(date_inferred, nullptr);
 
     VariantBuilder ts_builder(/*allow_duplicate_keys=*/false);
     ASSERT_OK(ts_builder.AppendTimestamp(1700000000000000));
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<GenericVariant> ts_variant, ts_builder.Build(pool_));
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> ts_inferred,
-                         infer_.InferColumnShreddingType({ts_variant}));
+                         InferColumn(infer_, {ts_variant}));
     ASSERT_EQ(ts_inferred, nullptr);
 }
 

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

@@ -45,14 +45,14 @@ class Variant::Impl {
         return pool_;
     }
 
-    arrow::MemoryPool* GetArrowMemoryPool() const {
-        return arrow_pool_.get();
+    const std::shared_ptr<arrow::MemoryPool>& GetArrowMemoryPool() const {
+        return arrow_pool_;
     }
 
  private:
     std::shared_ptr<GenericVariant> variant_;
     std::shared_ptr<MemoryPool> pool_;
-    std::unique_ptr<arrow::MemoryPool> arrow_pool_;
+    std::shared_ptr<arrow::MemoryPool> arrow_pool_;
 };
 
 Variant::Variant(std::unique_ptr<Impl>&& impl) : impl_(std::move(impl)) {}
@@ -148,7 +148,7 @@ Status VariantAccessBuilder::AddField(struct ArrowSchema* target_type, const std
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Field> target,
                                       arrow::ImportField(target_type));
     // Validate the path eagerly so mistakes fail at build time, not at read time.
-    PAIMON_RETURN_NOT_OK(VariantPathSegment::Parse(path).status());
+    PAIMON_RETURN_NOT_OK(VariantPathSegment::Parse(path));
     // Keep the target field's own metadata (e.g. the variant extension marker of a
     // `Variant::ArrowField` target, which drives the deep re-encode cast) and add the access
     // description to it.