test: expand coverage for indexes, variant shredding, and core I/O

src/paimon/CMakeLists.txt

@@ -674,6 +674,7 @@ if(PAIMON_BUILD_TESTS)
                     common/global_index/btree/btree_global_index_integration_test.cpp
                     common/global_index/btree/btree_compatibility_test.cpp
                     common/global_index/sorted_file_meta_selector_test.cpp
+                    common/global_index/sorted_file_global_index_reader_test.cpp
                     common/global_index/btree/lazy_filtered_btree_reader_test.cpp
                     common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
                     common/global_index/wrap/file_index_reader_wrapper_test.cpp
@@ -865,13 +866,15 @@ if(PAIMON_BUILD_TESTS)
                     core/index/index_file_handler_test.cpp
                     core/io/compact_increment_test.cpp
                     core/io/infer_shredding_file_writer_test.cpp
+                    core/io/shredding_key_value_data_file_writer_factory_test.cpp
                     core/io/concat_key_value_record_reader_test.cpp
                     core/io/data_file_meta_serializer_test.cpp
                     core/io/data_file_path_factory_test.cpp
                     core/io/data_increment_test.cpp
                     core/io/field_mapping_reader_test.cpp
                     core/io/key_value_data_file_record_reader_test.cpp
                     core/io/key_value_projection_reader_test.cpp
+                    core/io/row_to_arrow_array_converter_test.cpp
                     core/io/key_value_in_memory_record_reader_test.cpp
                     core/io/merged_key_value_record_reader_test.cpp
                     core/io/complete_row_tracking_fields_reader_test.cpp
@@ -922,6 +925,7 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/lookup_file_test.cpp
                     core/mergetree/lookup_levels_test.cpp
                     core/mergetree/compact/aggregate/aggregate_merge_function_test.cpp
+                    core/mergetree/compact/aggregate/field_aggregate_utils_test.cpp
                     core/mergetree/compact/aggregate/field_aggregator_factory_test.cpp
                     core/mergetree/compact/aggregate/field_bool_agg_test.cpp
                     core/mergetree/compact/aggregate/field_collect_agg_test.cpp

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

@@ -217,6 +217,45 @@ TEST_F(GenericVariantTest, DuplicateKeys) {
     ASSERT_EQ(to_json, "{\"a\":2}");
 }
 
+TEST_F(GenericVariantTest, DuplicateKeysWithInterleavedFieldsKeepLastValues) {
+    ASSERT_OK_AND_ASSIGN(
+        auto variant,
+        VariantBuilder::ParseJson(
+            R"({"z":1,"a":{"old":true},"m":[1,2],"z":4,"a":{"new":[3,4]},"b":"keep"})",
+            /*allow_duplicate_keys=*/true, pool_));
+    ASSERT_OK_AND_ASSIGN(std::string json, variant->ToJson());
+    ASSERT_EQ(json, R"({"a":{"new":[3,4]},"b":"keep","m":[1,2],"z":4})");
+    ASSERT_OK_AND_ASSIGN(int32_t size, variant->ObjectSize());
+    ASSERT_EQ(size, 4);
+}
+
+TEST_F(GenericVariantTest, HighScaleDecimalFallsBackToDouble) {
+    ASSERT_OK_AND_ASSIGN(auto variant,
+                         GenericVariant::FromJson("0." + std::string(38, '0') + "1", pool_));
+    ASSERT_OK_AND_ASSIGN(VariantValueType type, variant->GetType());
+    ASSERT_EQ(type, VariantValueType::kDouble);
+    ASSERT_OK_AND_ASSIGN(double value, variant->GetDouble());
+    ASSERT_DOUBLE_EQ(value, 1e-39);
+}
+
+TEST_F(GenericVariantTest, LargeArrayElementsRoundTrip) {
+    for (int32_t size : {1 << 16, 1 << 24}) {
+        SCOPED_TRACE(size);
+        const std::string text(size, 'x');
+        VariantBuilder builder(/*allow_duplicate_keys=*/false);
+        ASSERT_OK(builder.AppendString(text));
+        ASSERT_OK(builder.FinishWritingArray(/*start=*/0, /*offsets=*/{0}));
+        ASSERT_OK_AND_ASSIGN(auto variant, builder.Build(pool_));
+        ASSERT_OK_AND_ASSIGN(int32_t count, variant->ArraySize());
+        ASSERT_EQ(count, 1);
+        ASSERT_OK_AND_ASSIGN(auto element, variant->GetElementAtIndex(0));
+        ASSERT_TRUE(element);
+        ASSERT_OK_AND_ASSIGN(std::string_view value, element->GetString());
+        ASSERT_EQ(value.size(), text.size());
+        ASSERT_TRUE(value == text);
+    }
+}
+
 TEST_F(GenericVariantTest, CorruptedLayoutRejected) {
     // Headers claiming a near-INT32_MAX element count must be rejected by the 64-bit layout
     // bound instead of overflowing the 32-bit offset arithmetic.

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

@@ -21,6 +21,7 @@
 
 #include <functional>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "arrow/api.h"
@@ -534,4 +535,104 @@ TEST_F(InferVariantShreddingSchemaTest, ObjectWithAllRareFieldsStaysUnshredded)
     ASSERT_EQ(inferred, nullptr);
 }
 
+TEST_F(InferVariantShreddingSchemaTest, AdaptiveScalingPreservesNestedFieldRatios) {
+    auto samples =
+        Samples({R"({"items":[{"a":1}],"missing":null})", R"({"items":[{"a":2}],"missing":null})",
+                 R"({"items":[{"a":3}],"missing":null})", R"({"items":[{"a":4}],"missing":null})"});
+    auto max_fields = infer_.CreateMaxFieldsBudget();
+    ASSERT_OK_AND_ASSIGN(
+        auto initial, infer_.InferInitialColumn(samples, /*effective_sample_size=*/2, &max_fields));
+    auto expected = arrow::struct_(
+        {arrow::field("items", arrow::list(arrow::struct_({arrow::field("a", arrow::int64())}))),
+         arrow::field("missing", arrow::null())});
+    ASSERT_TRUE(initial.selected_schema->Equals(*expected)) << initial.selected_schema->ToString();
+    ASSERT_EQ(initial.evidence.root_value_count, 2);
+
+    auto current =
+        Samples({R"({"items":[{"a":5}],"new":true})", R"({"items":[{"a":6}],"new":true})"});
+    max_fields = infer_.CreateMaxFieldsBudget();
+    ASSERT_OK_AND_ASSIGN(auto updated, infer_.InferAdaptiveColumn(
+                                           initial.evidence, initial.selected_schema, current,
+                                           /*effective_sample_size=*/2, /*admission_ratio=*/0.6,
+                                           /*retention_ratio=*/0.2, &max_fields));
+    // The old null field is retained at 50%; the new field is below its admission threshold.
+    ASSERT_TRUE(updated.selected_schema->Equals(*expected)) << updated.selected_schema->ToString();
+    ASSERT_EQ(updated.evidence.root_value_count, 2);
+}
+
+TEST_F(InferVariantShreddingSchemaTest, AdaptiveDecimalMergingAndTypeChanges) {
+    struct TestCase {
+        const char* previous;
+        const char* current;
+        std::shared_ptr<arrow::DataType> expected;
+    };
+    const std::vector<TestCase> cases = {
+        {"1.25", "2.125", arrow::decimal128(19, 3)},
+        {"1.25", "1000000000000000000", arrow::decimal128(21, 2)},
+        {R"("old")", "1.25", arrow::decimal128(18, 2)},
+        {R"("old")", "1234567890123456789.125", arrow::decimal128(38, 3)},
+    };
+    for (const auto& test_case : cases) {
+        SCOPED_TRACE(std::string(test_case.previous) + " -> " + test_case.current);
+        auto max_fields = infer_.CreateMaxFieldsBudget();
+        ASSERT_OK_AND_ASSIGN(auto initial,
+                             infer_.InferInitialColumn(Samples({test_case.previous}),
+                                                       /*effective_sample_size=*/10, &max_fields));
+        max_fields = infer_.CreateMaxFieldsBudget();
+        ASSERT_OK_AND_ASSIGN(auto updated,
+                             infer_.InferAdaptiveColumn(
+                                 initial.evidence, initial.selected_schema,
+                                 Samples({test_case.current}), /*effective_sample_size=*/10,
+                                 /*admission_ratio=*/0.1, /*retention_ratio=*/0.05, &max_fields));
+        ASSERT_TRUE(updated.selected_schema->Equals(*test_case.expected))
+            << updated.selected_schema->ToString();
+    }
+}
+
+TEST_F(InferVariantShreddingSchemaTest, AdaptiveArraySelectionSurvivesMixedHistoryAndEmptyBatch) {
+    auto max_fields = infer_.CreateMaxFieldsBudget();
+    ASSERT_OK_AND_ASSIGN(
+        auto previous,
+        infer_.InferInitialColumn(Samples({"1"}), /*effective_sample_size=*/10, &max_fields));
+    auto expected = arrow::list(arrow::int64());
+    const std::vector<std::vector<const char*>> batches = {{"[1]"}, {"[2]"}, {}};
+    for (const auto& batch : batches) {
+        max_fields = infer_.CreateMaxFieldsBudget();
+        ASSERT_OK_AND_ASSIGN(
+            auto updated,
+            infer_.InferAdaptiveColumn(previous.evidence, previous.selected_schema, Samples(batch),
+                                       /*effective_sample_size=*/10, /*admission_ratio=*/0.1,
+                                       /*retention_ratio=*/0.05, &max_fields));
+        ASSERT_TRUE(updated.selected_schema->Equals(*expected))
+            << updated.selected_schema->ToString();
+        previous = std::move(updated);
+    }
+}
+
+TEST_F(InferVariantShreddingSchemaTest, AdaptiveEmptyAndNullSamplesStayUnshredded) {
+    const std::vector<std::vector<const char*>> batches = {{}, {nullptr}, {"null", "null"}};
+    for (const auto& batch : batches) {
+        auto max_fields = infer_.CreateMaxFieldsBudget();
+        ASSERT_OK_AND_ASSIGN(auto result,
+                             infer_.InferAdaptiveColumn(
+                                 /*previous_evidence=*/{}, /*previous_selected=*/nullptr,
+                                 Samples(batch), /*effective_sample_size=*/1,
+                                 /*admission_ratio=*/0.1, /*retention_ratio=*/0.05, &max_fields));
+        ASSERT_TRUE(result.selected_schema->Equals(*arrow::null()));
+    }
+}
+
+TEST_F(InferVariantShreddingSchemaTest, AdaptiveWidthLimitLeavesColumnUnshredded) {
+    InferVariantShreddingSchema narrow_infer{empty_logical_schema_, pool_,
+                                             /*max_schema_width=*/1, /*max_schema_depth=*/50,
+                                             /*min_field_cardinality_ratio=*/0.1};
+    auto max_fields = narrow_infer.CreateMaxFieldsBudget();
+    ASSERT_OK_AND_ASSIGN(
+        auto result, narrow_infer.InferAdaptiveColumn(
+                         /*previous_evidence=*/{}, /*previous_selected=*/nullptr, Samples({"1"}),
+                         /*effective_sample_size=*/10,
+                         /*admission_ratio=*/0.1, /*retention_ratio=*/0.05, &max_fields));
+    ASSERT_TRUE(result.selected_schema->Equals(*arrow::null()));
+}
+
 }  // namespace paimon::test

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

@@ -155,6 +155,18 @@ TEST_F(VariantGetTest, ScalarTargets) {
     ASSERT_EQ(elem->GetValue<int64_t>(), 4);
 }
 
+TEST_F(VariantGetTest, DecimalWithTrailingZerosCastsToScalarTargets) {
+    ASSERT_OK_AND_ASSIGN(auto variant, GenericVariant::FromJson("100.00", pool_));
+    ASSERT_OK_AND_ASSIGN(auto literal,
+                         VariantGetExecutor::Get(variant, "$", arrow::int64(), cast_args_));
+    ASSERT_TRUE(literal.has_value());
+    ASSERT_EQ(literal->GetValue<int64_t>(), 100);
+    ASSERT_OK_AND_ASSIGN(auto array, VariantGetExecutor::GetAsArrow(
+                                         variant, "$", arrow::field("x", arrow::decimal128(10, 2)),
+                                         cast_args_, pool_, arrow_pool_));
+    ASSERT_EQ(static_cast<const arrow::Decimal128Array&>(*array).FormatValue(0), "100.00");
+}
+
 TEST_F(VariantGetTest, ContainerToJsonString) {
     ASSERT_EQ(GetString("$.object", arrow::utf8()),
               "{\"address\":{\"city\":\"Hangzhou\",\"street\":\"Main St\"},\"age\":2,\"name\":"

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

@@ -308,4 +308,70 @@ TEST_F(VariantShreddingReadPlanFactoryTest, ColumnAbsentInFileSkipped) {
     EXPECT_TRUE(plans.empty());
 }
 
+TEST_F(VariantShreddingReadPlanFactoryTest, AccessProjectionOnShreddedArray) {
+    std::shared_ptr<arrow::DataType> physical;
+    std::shared_ptr<arrow::StructArray> shredded;
+    MakeFullShredded(arrow::list(arrow::int64()), Variant(R"([10,"hi",null])"), &physical,
+                     &shredded);
+    ASSERT_FALSE(HasFatalFailure());
+
+    auto read_field =
+        arrow::field("v", arrow::struct_({AccessChild("first", arrow::int64(), "$[0]"),
+                                          AccessChild("second", arrow::utf8(), "$[1]"),
+                                          AccessChild("null_element", arrow::utf8(), "$[2]"),
+                                          AccessChild("out_of_bounds", arrow::int64(), "$[5]")}));
+    auto plan = CreatePlan(read_field, arrow::field("v", physical));
+    ASSERT_NE(plan, nullptr);
+    ASSERT_OK_AND_ASSIGN(auto assembled, plan->Assemble(shredded, arrow::default_memory_pool()));
+    auto row = checked_pointer_cast<arrow::StructArray>(assembled);
+    ASSERT_EQ(row->length(), 1);
+    ASSERT_EQ(static_cast<const arrow::Int64Array&>(*row->field(0)).Value(0), 10);
+    // The string does not fit the shredded int64 element type and is read from its binary value.
+    ASSERT_EQ(static_cast<const arrow::StringArray&>(*row->field(1)).GetString(0), "hi");
+    ASSERT_TRUE(row->field(2)->IsNull(0));
+    ASSERT_TRUE(row->field(3)->IsNull(0));
+}
+
+TEST_F(VariantShreddingReadPlanFactoryTest, AccessThroughScalarReturnsNull) {
+    auto unshredded = MakeUnshredded(Variant(R"({"a":1})"));
+    ASSERT_FALSE(HasFatalFailure());
+    auto read_field =
+        arrow::field("v", arrow::struct_({AccessChild("object_child", arrow::int64(), "$.a.x"),
+                                          AccessChild("array_child", arrow::int64(), "$.a[0]")}));
+    auto plan = CreatePlan(read_field, arrow::field("v", unshredded->type()));
+    ASSERT_NE(plan, nullptr);
+    ASSERT_OK_AND_ASSIGN(auto assembled, plan->Assemble(unshredded, arrow::default_memory_pool()));
+    auto row = checked_pointer_cast<arrow::StructArray>(assembled);
+    ASSERT_EQ(row->length(), 1);
+    ASSERT_TRUE(row->field(0)->IsNull(0));
+    ASSERT_TRUE(row->field(1)->IsNull(0));
+}
+
+TEST_F(VariantShreddingReadPlanFactoryTest, NestedSchemaEvolutionWithoutVariantPlan) {
+    auto read_field = arrow::field("s", arrow::struct_({VariantTypeUtils::ToArrowField("v")}));
+    const arrow::FieldVector file_fields = {
+        arrow::field("s", arrow::struct_({arrow::field("other", arrow::int32())})),
+        arrow::field("s", arrow::int32()),
+    };
+    for (const auto& file_field : file_fields) {
+        ASSERT_OK_AND_ASSIGN(auto plans,
+                             VariantShreddingReadPlanFactory::CreateReadPlans(
+                                 arrow::schema({read_field}), arrow::schema({file_field}), pool_));
+        ASSERT_TRUE(plans.empty());
+    }
+}
+
+TEST_F(VariantShreddingReadPlanFactoryTest, ReorderedStructFieldsInsideListSkipPlan) {
+    auto variant_field = VariantTypeUtils::ToArrowField("v");
+    auto other_field = arrow::field("x", arrow::int32());
+    auto read_field =
+        arrow::field("items", arrow::list(arrow::struct_({variant_field, other_field})));
+    auto file_field =
+        arrow::field("items", arrow::list(arrow::struct_({other_field, variant_field})));
+    ASSERT_OK_AND_ASSIGN(auto plans,
+                         VariantShreddingReadPlanFactory::CreateReadPlans(
+                             arrow::schema({read_field}), arrow::schema({file_field}), pool_));
+    ASSERT_TRUE(plans.empty());
+}
+
 }  // namespace paimon::test

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

@@ -18,6 +18,7 @@
  */
 
 #include <functional>
+#include <limits>
 #include <memory>
 #include <string>
 #include <vector>
@@ -429,6 +430,44 @@ TEST_F(VariantShreddingTest, ShredScalarsAndMismatches) {
                            "{\"arr\": [1, 2]}", R"({"arr": {"k": "v"}})"});
 }
 
+TEST_F(VariantShreddingTest, DecimalExactDownscaleUsesTypedValue) {
+    auto variant =
+        BuildVariant([](VariantBuilder& builder) { return builder.AppendDecimal({12300, 3}); });
+    auto shredded = ShredAndCheck(Physical(arrow::decimal128(9, 1)), {variant});
+    ASSERT_FALSE(HasFatalFailure());
+    const auto& typed = static_cast<const arrow::Decimal128Array&>(*shredded->field(2));
+    ASSERT_FALSE(typed.IsNull(0));
+    ASSERT_EQ(typed.FormatValue(0), "12.3");
+    ASSERT_TRUE(shredded->field(1)->IsNull(0));
+}
+
+TEST_F(VariantShreddingTest, DecimalToLongUsesTypedValueOrPreservesResidual) {
+    auto integral =
+        BuildVariant([](VariantBuilder& builder) { return builder.AppendDecimal({120, 1}); });
+    auto fractional =
+        BuildVariant([](VariantBuilder& builder) { return builder.AppendDecimal({125, 1}); });
+    auto wide = BuildVariant([](VariantBuilder& builder) {
+        return builder.AppendDecimal(
+            {static_cast<__int128_t>(std::numeric_limits<int64_t>::max()) + 1, 0});
+    });
+    auto shredded = ShredAndCheck(Physical(arrow::int64()), {integral, fractional, wide});
+    ASSERT_FALSE(HasFatalFailure());
+    const auto& typed = static_cast<const arrow::Int64Array&>(*shredded->field(2));
+    ASSERT_FALSE(typed.IsNull(0));
+    ASSERT_EQ(typed.Value(0), 12);
+    ASSERT_TRUE(typed.IsNull(1));
+    ASSERT_TRUE(typed.IsNull(2));
+    ASSERT_FALSE(shredded->field(1)->IsNull(1));
+    ASSERT_FALSE(shredded->field(1)->IsNull(2));
+}
+
+TEST_F(VariantShreddingTest, LongPreservedAsResidualForBooleanTarget) {
+    auto shredded = RoundTrip(arrow::boolean(), {"1"});
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(shredded->field(2)->IsNull(0));
+    ASSERT_FALSE(shredded->field(1)->IsNull(0));
+}
+
 TEST_F(VariantShreddingTest, ScalarShreddingIntegerWidths) {
     // int8/int16 targets: in-range longs shred into the narrow typed column, while out-of-range
     // values fall back to the residual value column. Decimal-integral inputs take the