From a2d114c5b2e9cf110418b720831bb83cba2c4910 Mon Sep 17 00:00:00 2001
From: lxy264173 <lxy264173@alibaba-inc.com>
Date: Wed, 4 Nov 2026 17:12:31 +0800
Subject: [PATCH 1/3] test: expand coverage for indexes, variant shredding, and
 core I/O

---
 src/paimon/CMakeLists.txt                     |   4 +
 .../data/variant/generic_variant_test.cpp     |  39 +++++
 .../infer_variant_shredding_schema_test.cpp   | 101 +++++++++++
 .../common/data/variant/variant_get_test.cpp  |  12 ++
 ...riant_shredding_read_plan_factory_test.cpp |  66 +++++++
 .../data/variant/variant_shredding_test.cpp   |  39 +++++
 .../apply_bitmap_index_batch_reader_test.cpp  |  68 ++++++++
 .../bitmap_global_index_format_test.cpp       |  70 ++++++++
 .../bitmap/bitmap_global_index_test.cpp       |  68 ++++++++
 .../btree_global_index_integration_test.cpp   |  84 +++++++++
 .../range_bitmap_global_index_test.cpp        |   3 +
 .../sorted_file_global_index_reader_test.cpp  | 139 +++++++++++++++
 .../sorted_file_meta_selector_test.cpp        |  23 ++-
 .../common/utils/arrow/arrow_utils_test.cpp   |  49 ++++++
 src/paimon/core/io/data_file_meta_test.cpp    |  69 ++++++++
 .../io/infer_shredding_file_writer_test.cpp   |  37 +++-
 .../io/row_to_arrow_array_converter_test.cpp  | 101 +++++++++++
 ...ey_value_data_file_writer_factory_test.cpp | 151 ++++++++++++++++
 .../aggregate/field_aggregate_utils_test.cpp  | 131 ++++++++++++++
 .../field_nested_update_agg_test.cpp          |  54 ++++++
 .../partial_update_merge_function_test.cpp    |  48 +++++
 .../data_evolution_split_read_test.cpp        | 165 ++++++++++++++++++
 src/paimon/core/snapshot_file_scan_test.cpp   | 140 ++++++++++++++-
 .../format/mosaic/mosaic_file_format_test.cpp |  21 ++-
 .../orc/orc_input_output_stream_test.cpp      |  11 ++
 test/inte/variant_table_inte_test.cpp         |  34 ++++
 26 files changed, 1714 insertions(+), 13 deletions(-)
 create mode 100644 src/paimon/common/global_index/sorted_file_global_index_reader_test.cpp
 create mode 100644 src/paimon/core/io/row_to_arrow_array_converter_test.cpp
 create mode 100644 src/paimon/core/io/shredding_key_value_data_file_writer_factory_test.cpp
 create mode 100644 src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils_test.cpp

diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index 4c5c91bb8..004621a9f 100644
--- a/src/paimon/CMakeLists.txt
+++ b/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,6 +866,7 @@ 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
@@ -872,6 +874,7 @@ if(PAIMON_BUILD_TESTS)
                     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
diff --git a/src/paimon/common/data/variant/generic_variant_test.cpp b/src/paimon/common/data/variant/generic_variant_test.cpp
index bc46a5730..72c5f8357 100644
--- a/src/paimon/common/data/variant/generic_variant_test.cpp
+++ b/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.
diff --git a/src/paimon/common/data/variant/infer_variant_shredding_schema_test.cpp b/src/paimon/common/data/variant/infer_variant_shredding_schema_test.cpp
index 13e77dbd0..f286efe35 100644
--- a/src/paimon/common/data/variant/infer_variant_shredding_schema_test.cpp
+++ b/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
diff --git a/src/paimon/common/data/variant/variant_get_test.cpp b/src/paimon/common/data/variant/variant_get_test.cpp
index 159a573dc..7d34d68ea 100644
--- a/src/paimon/common/data/variant/variant_get_test.cpp
+++ b/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\":"
diff --git a/src/paimon/common/data/variant/variant_shredding_read_plan_factory_test.cpp b/src/paimon/common/data/variant/variant_shredding_read_plan_factory_test.cpp
index c515fd713..79129aaf3 100644
--- a/src/paimon/common/data/variant/variant_shredding_read_plan_factory_test.cpp
+++ b/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
diff --git a/src/paimon/common/data/variant/variant_shredding_test.cpp b/src/paimon/common/data/variant/variant_shredding_test.cpp
index c0b54eb07..5b8014723 100644
--- a/src/paimon/common/data/variant/variant_shredding_test.cpp
+++ b/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
diff --git a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
index 85f257cae..bc8680efb 100644
--- a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
+++ b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
@@ -18,8 +18,11 @@
 
 #include "paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h"
 
+#include <cstdint>
 #include <cstdlib>
+#include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "arrow/api.h"
@@ -48,6 +51,31 @@ class FileSystem;
 }  // namespace paimon
 
 namespace paimon::test {
+namespace {
+
+class OffsetFileBatchReader : public MockFileBatchReader {
+ public:
+    OffsetFileBatchReader(const std::shared_ptr<arrow::Array>& data, uint64_t offset)
+        : MockFileBatchReader(data, data->type(), /*batch_size=*/2), offset_(offset) {
+        EnableRandomizeBatchSize(false);
+    }
+
+    Result<uint64_t> GetPreviousBatchFileRowId(uint64_t batch_row_id) const override {
+        PAIMON_ASSIGN_OR_RAISE(uint64_t row_id,
+                               MockFileBatchReader::GetPreviousBatchFileRowId(batch_row_id));
+        return offset_ + row_id;
+    }
+
+    Result<uint64_t> GetNumberOfRows() const override {
+        return offset_ + static_cast<uint64_t>(data_->length());
+    }
+
+ private:
+    uint64_t offset_;
+};
+
+}  // namespace
+
 class ApplyBitmapIndexBatchReaderTest : public ::testing::Test,
                                         public ::testing::WithParamInterface<bool> {
  public:
@@ -208,6 +236,46 @@ TEST(ApplyBitmapIndexBatchReaderWarmupTest, WarmupForwardsToInnerReader) {
     ASSERT_EQ(1, inner_reader->GetWarmupCount());
 }
 
+TEST(ApplyBitmapIndexBatchReaderMetadataTest, ForwardsFileMetadata) {
+    auto type = arrow::struct_({arrow::field("f1", arrow::int32())});
+    auto data =
+        arrow::ipc::internal::json::ArrayFromJSON(type, "[[10], [11], [12], [13]]").ValueOrDie();
+    ApplyBitmapIndexBatchReader reader(
+        std::make_unique<MockFileBatchReader>(data, type, /*batch_size=*/2),
+        RoaringBitmap32::From({1, 3}));
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ArrowSchema> c_schema, reader.GetFileSchema());
+    auto schema_result = arrow::ImportSchema(c_schema.get());
+    ASSERT_TRUE(schema_result.ok()) << schema_result.status().ToString();
+    auto schema = schema_result.ValueOrDie();
+    ASSERT_TRUE(schema->Equals(*arrow::schema(type->fields())));
+    ASSERT_OK_AND_ASSIGN(uint64_t row_count, reader.GetNumberOfRows());
+    ASSERT_EQ(row_count, 4);
+    ASSERT_FALSE(reader.SupportPreciseBitmapSelection());
+}
+
+TEST(ApplyBitmapIndexBatchReaderBoundaryTest, FiltersAtBitmapRowIdLimit) {
+    auto type = arrow::struct_({arrow::field("f1", arrow::int32())});
+    auto data =
+        arrow::ipc::internal::json::ArrayFromJSON(type, "[[10], [11], [12], [13]]").ValueOrDie();
+    const uint64_t max_row_id = RoaringBitmap32::MAX_VALUE;
+    for (uint64_t offset : {max_row_id - 1, max_row_id + 1}) {
+        SCOPED_TRACE(offset);
+        auto reader = std::make_unique<ApplyBitmapIndexBatchReader>(
+            std::make_unique<OffsetFileBatchReader>(data, offset),
+            RoaringBitmap32::From({RoaringBitmap32::MAX_VALUE - 1, RoaringBitmap32::MAX_VALUE}));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+                             ReadResultCollector::CollectResult(std::move(reader)));
+        if (offset > max_row_id) {
+            ASSERT_FALSE(result);
+        } else {
+            ASSERT_TRUE(result);
+            ASSERT_TRUE(result->Equals(arrow::ChunkedArray(data->Slice(0, 2))))
+                << result->ToString();
+        }
+    }
+}
+
 INSTANTIATE_TEST_SUITE_P(EnablePrefetch, ApplyBitmapIndexBatchReaderTest,
                          ::testing::Values(false, true));
 
diff --git a/src/paimon/common/global_index/bitmap/bitmap_global_index_format_test.cpp b/src/paimon/common/global_index/bitmap/bitmap_global_index_format_test.cpp
index 2ad25cf05..79c9a48b3 100644
--- a/src/paimon/common/global_index/bitmap/bitmap_global_index_format_test.cpp
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_format_test.cpp
@@ -19,9 +19,11 @@
 
 #include "paimon/common/global_index/bitmap/bitmap_global_index_format.h"
 
+#include <cstdint>
 #include <cstring>
 #include <memory>
 #include <string>
+#include <vector>
 
 #include "gtest/gtest.h"
 #include "paimon/common/io/byte_array_output_stream.h"
@@ -64,6 +66,17 @@ std::shared_ptr<ByteArrayOutputStream> CreateOutput(const std::shared_ptr<Memory
 
 }  // namespace
 
+TEST(BitmapGlobalIndexFormatTest, CompareNullSerializedKeys) {
+    auto pool = GetDefaultPool();
+    BitmapGlobalIndexFormat::SerializedKey null_key(nullptr);
+    BitmapGlobalIndexFormat::SerializedKey another_null_key(nullptr);
+    auto empty_key = SerializedString("", pool.get());
+
+    ASSERT_EQ(null_key.CompareTo(another_null_key), 0);
+    ASSERT_LT(null_key.CompareTo(empty_key), 0);
+    ASSERT_GT(empty_key.CompareTo(null_key), 0);
+}
+
 TEST(BitmapGlobalIndexFormatTest, RoundTripMultipleDictionaryBlocks) {
     std::shared_ptr<MemoryPool> pool = GetDefaultPool();
     std::shared_ptr<ByteArrayOutputStream> output = CreateOutput(pool);
@@ -140,4 +153,61 @@ TEST(BitmapGlobalIndexFormatTest, RoundTripEmptyNullBitmap) {
     ASSERT_EQ(RoaringBitmap64::From({0}), non_null_rows);
 }
 
+TEST(BitmapGlobalIndexFormatTest, RoundTripWithCompressionAndUncompressedFallback) {
+    auto pool = GetDefaultPool();
+    for (BlockCompressionType compression_type :
+         {BlockCompressionType::ZSTD, BlockCompressionType::LZ4}) {
+        SCOPED_TRACE(static_cast<int32_t>(compression_type));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlockCompressionFactory> compression_factory,
+                             BlockCompressionFactory::Create(compression_type));
+        for (int32_t key_length : {1, 1024}) {
+            SCOPED_TRACE(key_length);
+            // Tiny blocks do not benefit from compression; repeated long keys do.
+            std::vector<std::string> keys = {std::string(key_length, 'a'),
+                                             std::string(key_length, 'b')};
+            auto output = CreateOutput(pool);
+            ASSERT_OK_AND_ASSIGN(
+                std::unique_ptr<BitmapGlobalIndexFormat::StreamingWriter> writer,
+                BitmapGlobalIndexFormat::StreamingWriter::Create(
+                    output, /*dictionary_block_size=*/1, compression_factory, pool));
+            for (int32_t i = 0; i < 2; ++i) {
+                ASSERT_OK(writer->Write(SerializedString(keys[i], pool.get()),
+                                        RoaringBitmap64::From({i, i + 2})));
+            }
+            ASSERT_OK(
+                writer->Finish(RoaringBitmap64::From({4}), RoaringBitmap64::From({0, 1, 2, 3})));
+            ASSERT_OK_AND_ASSIGN(std::shared_ptr<Bytes> bytes, output->Finish(pool.get()));
+
+            BytesSeekableReader reader(bytes, pool);
+            ASSERT_OK_AND_ASSIGN(BitmapGlobalIndexFormat::Footer footer,
+                                 BitmapGlobalIndexFormat::ReadFooter(bytes->size(), &reader));
+            ASSERT_OK_AND_ASSIGN(
+                std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta> block_metas,
+                BitmapGlobalIndexFormat::ReadIndexBlock(footer.IndexBlock(), &reader, pool.get()));
+            ASSERT_EQ(block_metas.size(), 2);
+
+            for (int32_t i = 0; i < 2; ++i) {
+                ASSERT_OK_AND_ASSIGN(BitmapGlobalIndexFormat::DictionaryBlock block,
+                                     BitmapGlobalIndexFormat::ReadDictionaryBlock(
+                                         block_metas[i], &reader, pool.get()));
+                ASSERT_EQ(block.Entries().size(), 1);
+                const auto& entry = block.Entries()[0];
+                ASSERT_EQ(keys[i], std::string(entry.Key().GetBytes()->data(),
+                                               entry.Key().GetBytes()->size()));
+                ASSERT_OK_AND_ASSIGN(RoaringBitmap64 bitmap, BitmapGlobalIndexFormat::ReadBitmap(
+                                                                 entry.BitmapBlock(), &reader));
+                ASSERT_EQ(bitmap, RoaringBitmap64::From({i, i + 2}));
+            }
+
+            ASSERT_OK_AND_ASSIGN(RoaringBitmap64 null_rows, BitmapGlobalIndexFormat::ReadBitmap(
+                                                                footer.NullRowsBlock(), &reader));
+            ASSERT_EQ(null_rows, RoaringBitmap64::From({4}));
+            ASSERT_OK_AND_ASSIGN(
+                RoaringBitmap64 non_null_rows,
+                BitmapGlobalIndexFormat::ReadBitmap(footer.NonNullRowsBlock(), &reader));
+            ASSERT_EQ(non_null_rows, RoaringBitmap64::From({0, 1, 2, 3}));
+        }
+    }
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp b/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
index 22b259d70..972fb00c9 100644
--- a/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
@@ -112,6 +112,8 @@ TEST_F(BitmapGlobalIndexTest, WriteAndReadStringPredicates) {
 
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapIndexReader> reader,
                          CreateReader(field->type(), metas[0]));
+    ASSERT_FALSE(reader->IsThreadSafe());
+    ASSERT_EQ(reader->GetIndexType(), "bitmap");
     CheckResult(reader->VisitIsNull(), {2, 6});
     CheckResult(reader->VisitIsNotNull(), {0, 1, 3, 4, 5});
     CheckResult(reader->VisitEqual(StringLiteral("apple")), {0, 1});
@@ -152,6 +154,31 @@ TEST_F(BitmapGlobalIndexTest, WriteAndReadStringPredicates) {
     CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
                     std::dynamic_pointer_cast<LeafPredicate>(like), reader),
                 {3});
+
+    CheckResult(reader->VisitEqual(StringLiteral("aardvark")), {});
+    CheckResult(reader->VisitStartsWith(StringLiteral("")), {0, 1, 3, 4, 5});
+    CheckResult(reader->VisitEndsWith(StringLiteral("")), {0, 1, 3, 4, 5});
+    CheckResult(reader->VisitContains(StringLiteral("")), {0, 1, 3, 4, 5});
+
+    // NULL operands do not match rows; IN ignores NULL while NOT IN with NULL matches nothing.
+    Literal null_literal(FieldType::STRING);
+    CheckResult(reader->VisitEqual(null_literal), {});
+    CheckResult(reader->VisitNotEqual(null_literal), {});
+    CheckResult(reader->VisitLessThan(null_literal), {});
+    CheckResult(reader->VisitLessOrEqual(null_literal), {});
+    CheckResult(reader->VisitGreaterThan(null_literal), {});
+    CheckResult(reader->VisitGreaterOrEqual(null_literal), {});
+    CheckResult(reader->VisitStartsWith(null_literal), {});
+    CheckResult(reader->VisitEndsWith(null_literal), {});
+    CheckResult(reader->VisitContains(null_literal), {});
+    CheckResult(reader->VisitLike(null_literal), {});
+    CheckResult(reader->VisitIn({null_literal, StringLiteral("apple"), null_literal}), {0, 1});
+    CheckResult(reader->VisitNotIn({StringLiteral("apple"), null_literal}), {});
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Bytes> empty_read, reader->Read(metas[0].file_size, 0));
+    ASSERT_TRUE(empty_read);
+    ASSERT_EQ(empty_read->size(), 0);
+
     ASSERT_NOK_WITH_MSG(reader->VisitVectorSearch(nullptr),
                         "Vector search is not supported in BitmapIndexReader");
     ASSERT_NOK_WITH_MSG(reader->VisitFullTextSearch(nullptr),
@@ -177,8 +204,49 @@ TEST_F(BitmapGlobalIndexTest, WriteAndReadLogicalIntegerOrder) {
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapIndexReader> reader,
                          CreateReader(field->type(), metas[0]));
     CheckResult(reader->VisitEqual(Literal(static_cast<int32_t>(256))), {4});
+    CheckResult(reader->VisitEqual(Literal(static_cast<int32_t>(-258))), {});
     CheckResult(reader->VisitLessThan(Literal(static_cast<int32_t>(1))), {0, 1, 2});
     CheckResult(reader->VisitGreaterOrEqual(Literal(static_cast<int32_t>(0))), {2, 3, 4, 5});
+    // LIKE does not match non-string dictionary values.
+    CheckResult(reader->VisitLike(StringLiteral("%")), {});
+}
+
+TEST_F(BitmapGlobalIndexTest, StringPrefixBoundaries) {
+    auto field = arrow::field("f0", arrow::utf8());
+    // Exercise the same queries with entries spread across blocks and grouped in one block.
+    for (int32_t dictionary_block_size : {1, 4096}) {
+        SCOPED_TRACE(dictionary_block_size);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapGlobalIndexWriter> writer,
+                             CreateWriter(field, dictionary_block_size));
+        std::vector<std::string> values = {"a",        "ba",          "bb", "c",
+                                           "c\xFF",    "c\xFFtail",   "d",  "\xFF\xFE",
+                                           "\xFF\xFF", "\xFF\xFFtail"};
+        arrow::StringBuilder value_builder;
+        for (const auto& value : values) {
+            ASSERT_TRUE(value_builder.Append(value).ok());
+        }
+        ASSERT_TRUE(value_builder.AppendNull().ok());
+        std::shared_ptr<arrow::Array> value_array;
+        ASSERT_TRUE(value_builder.Finish(&value_array).ok());
+        auto array = arrow::StructArray::Make({value_array}, {field}).ValueOrDie();
+        ArrowArray c_array;
+        ASSERT_TRUE(arrow::ExportArray(*array, &c_array).ok());
+        std::vector<int64_t> row_ids(array->length());
+        std::iota(row_ids.begin(), row_ids.end(), 0);
+        ASSERT_OK(writer->AddBatch(&c_array, std::move(row_ids)));
+        ASSERT_OK_AND_ASSIGN(std::vector<GlobalIndexIOMeta> metas, writer->Finish());
+        ASSERT_EQ(metas.size(), 1);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapIndexReader> reader,
+                             CreateReader(field->type(), metas[0]));
+
+        CheckResult(reader->VisitStartsWith(StringLiteral("b")), {1, 2});
+        CheckResult(reader->VisitStartsWith(StringLiteral("0")), {});
+        CheckResult(reader->VisitStartsWith(StringLiteral("bbbb")), {});
+        // A trailing 0xFF carries into the preceding byte; an all-0xFF prefix has no upper bound.
+        CheckResult(reader->VisitStartsWith(StringLiteral("c\xFF")), {4, 5});
+        CheckResult(reader->VisitStartsWith(StringLiteral("\xFF\xFF")), {8, 9});
+        CheckResult(reader->VisitStartsWith(StringLiteral("\xFF\xFFz")), {});
+    }
 }
 
 TEST_F(BitmapGlobalIndexTest, WriteAndReadTimestamp) {
diff --git a/src/paimon/common/global_index/btree/btree_global_index_integration_test.cpp b/src/paimon/common/global_index/btree/btree_global_index_integration_test.cpp
index 20db545ba..ce2afefc7 100644
--- a/src/paimon/common/global_index/btree/btree_global_index_integration_test.cpp
+++ b/src/paimon/common/global_index/btree/btree_global_index_integration_test.cpp
@@ -24,6 +24,7 @@
 #include "gtest/gtest.h"
 #include "paimon/common/compression/block_compression_factory.h"
 #include "paimon/common/factories/io_hook.h"
+#include "paimon/common/global_index/btree/btree_global_index_reader.h"
 #include "paimon/common/global_index/btree/btree_global_index_writer.h"
 #include "paimon/common/global_index/btree/btree_global_indexer.h"
 #include "paimon/common/global_index/btree/lazy_filtered_btree_reader.h"
@@ -346,6 +347,33 @@ TEST_P(BTreeGlobalIndexIntegrationTest, WriteAndReadIntData) {
         ASSERT_OK_AND_ASSIGN(auto result, filtered_reader->VisitIsNull());
         CheckResult(result, {2, 5});
     }
+
+    // Call the file reader directly, since the outer reader handles these predicates itself.
+    {
+        auto lazy_reader = std::dynamic_pointer_cast<LazyFilteredBTreeReader>(filtered_reader);
+        ASSERT_TRUE(lazy_reader);
+        ASSERT_OK_AND_ASSIGN(auto file_index_reader, lazy_reader->GetOrCreateReader(metas[0]));
+        auto btree_reader = std::dynamic_pointer_cast<BTreeGlobalIndexReader>(file_index_reader);
+        ASSERT_TRUE(btree_reader);
+
+        ASSERT_OK_AND_ASSIGN(auto result, btree_reader->VisitIsNotNull());
+        CheckResult(result, {1, 3, 4, 6, 7, 8, 10});
+
+        ASSERT_OK_AND_ASSIGN(result, btree_reader->VisitNotEqual(Literal(1)));
+        CheckResult(result, {3, 4, 6, 7, 8, 10});
+        ASSERT_OK_AND_ASSIGN(result, btree_reader->VisitNotEqual(Literal(99)));
+        CheckResult(result, {1, 3, 4, 6, 7, 8, 10});
+
+        ASSERT_OK_AND_ASSIGN(result,
+                             btree_reader->VisitNotIn({Literal(1), Literal(5), Literal(99)}));
+        CheckResult(result, {3, 4, 6, 7});
+        ASSERT_OK_AND_ASSIGN(result, btree_reader->VisitNotIn({}));
+        CheckResult(result, {1, 3, 4, 6, 7, 8, 10});
+
+        // A non-string prefix falls back to all non-null rows within the row-id filter.
+        ASSERT_OK_AND_ASSIGN(result, btree_reader->VisitStartsWith(Literal(1)));
+        CheckResult(result, {1, 3, 4, 6, 7, 8, 10});
+    }
 }
 
 TEST_P(BTreeGlobalIndexIntegrationTest, WriteAndReadStringData) {
@@ -529,6 +557,62 @@ TEST_P(BTreeGlobalIndexIntegrationTest, WriteAndReadStringData) {
     }
 }
 
+TEST_P(BTreeGlobalIndexIntegrationTest, StringPrefixByteOverflow) {
+    auto file_writer = std::make_shared<FakeGlobalIndexFileWriter>(fs_, base_path_);
+    auto field = arrow::field("str_field", arrow::utf8());
+    auto c_schema = CreateArrowSchema(field);
+    std::map<std::string, std::string> options = {{BtreeDefs::kBtreeIndexBlockSize, "128"},
+                                                  {BtreeDefs::kBtreeIndexCompression, GetParam()}};
+    ASSERT_OK_AND_ASSIGN(auto indexer, BTreeGlobalIndexer::Create(options));
+    ASSERT_OK_AND_ASSIGN(auto writer,
+                         indexer->CreateWriter("str_field", c_schema.get(), file_writer, pool_));
+
+    // Build byte strings directly so that 0xFF and embedded zero bytes are preserved.
+    std::vector<std::string> values = {"a\xFE",    "a\xFF",    "a\xFFz",   std::string("b\0", 2),
+                                       "\xFF\xFE", "\xFF\xFF", "\xFF\xFFz"};
+    arrow::StringBuilder value_builder;
+    for (const auto& value : values) {
+        ASSERT_TRUE(value_builder.Append(value).ok());
+    }
+    ASSERT_TRUE(value_builder.AppendNull().ok());
+    std::shared_ptr<arrow::Array> value_array;
+    ASSERT_TRUE(value_builder.Finish(&value_array).ok());
+    auto array = arrow::StructArray::Make({value_array}, {field}).ValueOrDie();
+
+    ArrowArray c_array;
+    ASSERT_TRUE(arrow::ExportArray(*array, &c_array).ok());
+    std::vector<int64_t> row_ids(array->length());
+    std::iota(row_ids.begin(), row_ids.end(), 0);
+    ASSERT_OK(writer->AddBatch(&c_array, std::move(row_ids)));
+    ASSERT_OK_AND_ASSIGN(auto metas, writer->Finish());
+    ASSERT_EQ(metas.size(), 1);
+
+    auto file_reader = std::make_shared<FakeGlobalIndexFileReader>(fs_, base_path_);
+    c_schema = CreateArrowSchema(field);
+    ASSERT_OK_AND_ASSIGN(
+        auto reader, indexer->CreateReader(c_schema.get(), file_reader, metas, TotalRowCount(metas),
+                                           /*row_ranges=*/std::nullopt, pool_));
+    auto lazy_reader = std::dynamic_pointer_cast<LazyFilteredBTreeReader>(reader);
+    ASSERT_TRUE(lazy_reader);
+    ASSERT_OK_AND_ASSIGN(auto file_index_reader, lazy_reader->GetOrCreateReader(metas[0]));
+    auto btree_reader = std::dynamic_pointer_cast<BTreeGlobalIndexReader>(file_index_reader);
+    ASSERT_TRUE(btree_reader);
+
+    // Carry into the preceding byte; the exclusive upper bound is "b\0".
+    Literal carry_prefix(FieldType::STRING, values[1].data(), values[1].size());
+    ASSERT_OK_AND_ASSIGN(auto result, btree_reader->VisitStartsWith(carry_prefix));
+    CheckResult(result, {1, 2});
+
+    // All bytes overflow, so max_key_ must be included in the result.
+    Literal overflow_prefix(FieldType::STRING, values[5].data(), values[5].size());
+    ASSERT_OK_AND_ASSIGN(result, btree_reader->VisitStartsWith(overflow_prefix));
+    CheckResult(result, {5, 6});
+
+    // Suffix queries return all non-null candidates for subsequent filtering.
+    ASSERT_OK_AND_ASSIGN(result, btree_reader->VisitEndsWith(Literal(FieldType::STRING, "z", 1)));
+    CheckResult(result, {0, 1, 2, 3, 4, 5, 6});
+}
+
 TEST_P(BTreeGlobalIndexIntegrationTest, WriteEmptyStringKeyMetadata) {
     auto file_writer = std::make_shared<FakeGlobalIndexFileWriter>(fs_, base_path_);
     auto field = arrow::field("str_field", arrow::utf8());
diff --git a/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp b/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
index bbba051a1..d5287d3de 100644
--- a/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
+++ b/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
@@ -146,6 +146,9 @@ TEST_F(RangeBitmapGlobalIndexTest, TestIntType) {
     ASSERT_OK_AND_ASSIGN(auto meta, WriteGlobalIndex(test_root, type, array, Range(0, 4)));
 
     auto reader = CreateGlobalIndexReader(test_root, type, meta);
+    ASSERT_TRUE(reader);
+    ASSERT_FALSE(reader->IsThreadSafe());
+    ASSERT_EQ(reader->GetIndexType(), "range-bitmap");
 
     Literal lit_0(static_cast<int32_t>(0));
     Literal lit_1(static_cast<int32_t>(1));
diff --git a/src/paimon/common/global_index/sorted_file_global_index_reader_test.cpp b/src/paimon/common/global_index/sorted_file_global_index_reader_test.cpp
new file mode 100644
index 000000000..2029f0ab3
--- /dev/null
+++ b/src/paimon/common/global_index/sorted_file_global_index_reader_test.cpp
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+#include "paimon/common/global_index/sorted_file_global_index_reader.h"
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/global_index/key_serializer.h"
+#include "paimon/common/global_index/sorted_index_file_meta.h"
+#include "paimon/testing/mock/mock_global_index_reader.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+
+class TestSortedFileGlobalIndexReader : public SortedFileGlobalIndexReader {
+ public:
+    TestSortedFileGlobalIndexReader(std::unique_ptr<SortedFileMetaSelector> selector,
+                                    int64_t fallback_scan_max_size,
+                                    const std::shared_ptr<GlobalIndexReader>& reader)
+        : SortedFileGlobalIndexReader(std::move(selector), fallback_scan_max_size,
+                                      /*total_row_count=*/3, /*executor=*/nullptr),
+          reader_(reader) {}
+
+    Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitVectorSearch(
+        const std::shared_ptr<VectorSearch>& search) override {
+        return reader_->VisitVectorSearch(search);
+    }
+
+    Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitFullTextSearch(
+        const std::shared_ptr<FullTextSearch>& search) override {
+        return reader_->VisitFullTextSearch(search);
+    }
+
+    std::string GetIndexType() const override {
+        return reader_->GetIndexType();
+    }
+
+ protected:
+    Result<std::shared_ptr<GlobalIndexReader>> OpenReader(const GlobalIndexIOMeta&) override {
+        return reader_;
+    }
+
+ private:
+    std::shared_ptr<GlobalIndexReader> reader_;
+};
+
+}  // namespace
+
+class SortedFileGlobalIndexReaderTest : public ::testing::Test {
+ public:
+    void SetUp() override {
+        auto pool = GetDefaultPool();
+        ASSERT_OK_AND_ASSIGN(serializer_, KeySerializer::Create(arrow::utf8(), pool));
+        ASSERT_OK_AND_ASSIGN(auto first,
+                             serializer_->Serialize(Literal(FieldType::STRING, "a", 1)));
+        ASSERT_OK_AND_ASSIGN(auto last, serializer_->Serialize(Literal(FieldType::STRING, "z", 1)));
+        SortedIndexFileMeta meta(first, last, /*has_nulls=*/true);
+        files_.emplace_back("index", /*file_size=*/1, /*row_count=*/3, meta.Serialize(pool.get()));
+        backend_ = std::make_shared<MockGlobalIndexReader>();
+        backend_->SetReturnNullptr();
+    }
+
+    std::unique_ptr<TestSortedFileGlobalIndexReader> MakeReader(int64_t fallback_scan_max_size) {
+        EXPECT_OK_AND_ASSIGN(auto selector, SortedFileMetaSelector::Create(files_, serializer_));
+        return std::make_unique<TestSortedFileGlobalIndexReader>(std::move(selector),
+                                                                 fallback_scan_max_size, backend_);
+    }
+
+ protected:
+    std::shared_ptr<KeySerializer> serializer_;
+    std::vector<GlobalIndexIOMeta> files_;
+    std::shared_ptr<MockGlobalIndexReader> backend_;
+};
+
+TEST_F(SortedFileGlobalIndexReaderTest, UnsupportedResultsRemainUnavailableForComplements) {
+    auto reader = MakeReader(/*fallback_scan_max_size=*/100);
+    ASSERT_OK_AND_ASSIGN(auto not_null, reader->VisitIsNotNull());
+    ASSERT_EQ(not_null, nullptr);
+    const Literal value(FieldType::STRING, "m", 1);
+    ASSERT_OK_AND_ASSIGN(auto not_equal, reader->VisitNotEqual(value));
+    ASSERT_EQ(not_equal, nullptr);
+    ASSERT_OK_AND_ASSIGN(auto not_in, reader->VisitNotIn({value}));
+    ASSERT_EQ(not_in, nullptr);
+}
+
+TEST_F(SortedFileGlobalIndexReaderTest, NullLiteralsDoNotProduceIndexResults) {
+    auto reader = MakeReader(/*fallback_scan_max_size=*/100);
+    const Literal null_literal(FieldType::STRING);
+    for (auto visit :
+         {&SortedFileGlobalIndexReader::VisitLessThan,
+          &SortedFileGlobalIndexReader::VisitLessOrEqual,
+          &SortedFileGlobalIndexReader::VisitGreaterThan,
+          &SortedFileGlobalIndexReader::VisitGreaterOrEqual,
+          &SortedFileGlobalIndexReader::VisitStartsWith,
+          &SortedFileGlobalIndexReader::VisitEndsWith, &SortedFileGlobalIndexReader::VisitContains,
+          &SortedFileGlobalIndexReader::VisitLike}) {
+        ASSERT_OK_AND_ASSIGN(auto result, (reader.get()->*visit)(null_literal));
+        ASSERT_EQ(result, nullptr);
+    }
+}
+
+TEST_F(SortedFileGlobalIndexReaderTest, DisabledFallbackLeavesScanPredicatesUnavailable) {
+    auto reader = MakeReader(/*fallback_scan_max_size=*/0);
+    const Literal value(FieldType::STRING, "a_b", 3);
+    for (auto visit :
+         {&SortedFileGlobalIndexReader::VisitLessThan,
+          &SortedFileGlobalIndexReader::VisitLessOrEqual,
+          &SortedFileGlobalIndexReader::VisitGreaterThan,
+          &SortedFileGlobalIndexReader::VisitGreaterOrEqual,
+          &SortedFileGlobalIndexReader::VisitEndsWith, &SortedFileGlobalIndexReader::VisitContains,
+          &SortedFileGlobalIndexReader::VisitLike}) {
+        ASSERT_OK_AND_ASSIGN(auto result, (reader.get()->*visit)(value));
+        ASSERT_EQ(result, nullptr);
+    }
+}
+
+}  // namespace paimon::test
diff --git a/src/paimon/common/global_index/sorted_file_meta_selector_test.cpp b/src/paimon/common/global_index/sorted_file_meta_selector_test.cpp
index 0d8cc8c56..ddd74f300 100644
--- a/src/paimon/common/global_index/sorted_file_meta_selector_test.cpp
+++ b/src/paimon/common/global_index/sorted_file_meta_selector_test.cpp
@@ -225,6 +225,21 @@ TEST_F(SortedFileMetaSelectorTest, TestVisitNotIn) {
     CheckResult(result, {"file1", "file2", "file3", "file4", "file5"});
 }
 
+TEST_F(SortedFileMetaSelectorTest, NullLiteralsSelectNoFiles) {
+    ASSERT_OK_AND_ASSIGN(auto selector, SortedFileMetaSelector::Create(files_, key_serializer_));
+    const Literal null_literal(FieldType::INT);
+    for (auto visit :
+         {&SortedFileMetaSelector::VisitEqual, &SortedFileMetaSelector::VisitNotEqual,
+          &SortedFileMetaSelector::VisitLessThan, &SortedFileMetaSelector::VisitLessOrEqual,
+          &SortedFileMetaSelector::VisitGreaterThan,
+          &SortedFileMetaSelector::VisitGreaterOrEqual}) {
+        ASSERT_OK_AND_ASSIGN(auto result, (selector.get()->*visit)(null_literal));
+        ASSERT_TRUE(result.empty());
+    }
+    ASSERT_OK_AND_ASSIGN(auto result, selector->VisitNotIn({Literal(1), null_literal}));
+    ASSERT_TRUE(result.empty());
+}
+
 TEST_F(SortedFileMetaSelectorTest, TestVisitStartsWith) {
     std::shared_ptr<MemoryPool> pool = GetDefaultPool();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<KeySerializer> key_serializer,
@@ -241,12 +256,15 @@ TEST_F(SortedFileMetaSelectorTest, TestVisitStartsWith) {
                                                              /*has_nulls=*/false);
     auto third_meta = std::make_shared<SortedIndexFileMeta>(serialize("dog"), serialize("zoo"),
                                                             /*has_nulls=*/false);
+    auto high_byte_meta = std::make_shared<SortedIndexFileMeta>(
+        serialize(std::string("\xff", 1)), serialize(std::string("\xff\xff", 2)), false);
     auto only_nulls_meta =
         std::make_shared<SortedIndexFileMeta>(nullptr, nullptr, /*has_nulls=*/true);
     std::vector<GlobalIndexIOMeta> files = {
         GlobalIndexIOMeta("first", 1, /*row_count=*/1, first_meta->Serialize(pool.get())),
         GlobalIndexIOMeta("second", 1, /*row_count=*/1, second_meta->Serialize(pool.get())),
         GlobalIndexIOMeta("third", 1, /*row_count=*/1, third_meta->Serialize(pool.get())),
+        GlobalIndexIOMeta("high_byte", 1, /*row_count=*/1, high_byte_meta->Serialize(pool.get())),
         GlobalIndexIOMeta("only_nulls", 1, /*row_count=*/1, only_nulls_meta->Serialize(pool.get())),
     };
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<SortedFileMetaSelector> selector,
@@ -260,7 +278,10 @@ TEST_F(SortedFileMetaSelectorTest, TestVisitStartsWith) {
     ASSERT_TRUE(result.empty());
 
     ASSERT_OK_AND_ASSIGN(result, selector->VisitStartsWith(Literal(FieldType::STRING, "", 0)));
-    CheckResult(result, {"first", "second", "third"});
+    CheckResult(result, {"first", "second", "third", "high_byte"});
+
+    ASSERT_OK_AND_ASSIGN(result, selector->VisitStartsWith(Literal(FieldType::STRING, "\xff", 1)));
+    CheckResult(result, {"high_byte"});
 
     ASSERT_OK_AND_ASSIGN(result, selector->VisitStartsWith(Literal(FieldType::STRING)));
     ASSERT_TRUE(result.empty());
diff --git a/src/paimon/common/utils/arrow/arrow_utils_test.cpp b/src/paimon/common/utils/arrow/arrow_utils_test.cpp
index 6625e15e5..6bbd0f625 100644
--- a/src/paimon/common/utils/arrow/arrow_utils_test.cpp
+++ b/src/paimon/common/utils/arrow/arrow_utils_test.cpp
@@ -29,6 +29,7 @@
 #include "arrow/c/bridge.h"
 #include "arrow/c/helpers.h"
 #include "arrow/ipc/api.h"
+#include "arrow/util/byte_size.h"
 #include "gtest/gtest.h"
 #include "paimon/common/types/data_field.h"
 #include "paimon/common/utils/checked_cast.h"
@@ -250,6 +251,16 @@ TEST(ArrowUtilsTest, TestCheckNullableMatchParentNullMasksChildren) {
     ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, batch));
 }
 
+TEST(ArrowUtilsTest, TestCheckNullableMatchAllNullParentMasksNestedStruct) {
+    auto leaf = arrow::field("leaf", arrow::int32(), /*nullable=*/false);
+    auto inner = arrow::field("inner", arrow::struct_({leaf}), /*nullable=*/false);
+    auto parent = arrow::field("parent", arrow::struct_({inner}), /*nullable=*/true);
+    auto null_parents = arrow::MakeArrayOfNull(parent->type(), 3).ValueOrDie();
+    auto batch = arrow::StructArray::Make({null_parents}, {parent}).ValueOrDie();
+    ASSERT_TRUE(batch->ValidateFull().ok());
+    ASSERT_OK(ArrowUtils::CheckNullabilityMatch(arrow::schema({parent}), batch));
+}
+
 TEST(ArrowUtilsTest, TestCheckNullableMatchRejectsVisibleChildNulls) {
     auto child = arrow::field("child", arrow::int32(), /*nullable=*/false);
     auto struct_field = arrow::field("parent", arrow::struct_({child}), /*nullable=*/true);
@@ -995,6 +1006,26 @@ TEST(ArrowUtilsTest, TestNormalizeRecordBatchOffsetsFallsBackForDictionary) {
     ASSERT_FALSE(IsViewInto(*normalized->column_data(0)->buffers[1], *array->data()->buffers[1]));
 }
 
+TEST(ArrowUtilsTest, TestNormalizeZeroWidthBinarySlice) {
+    auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow::fixed_size_binary(0),
+                                                           R"(["", null, "", ""])")
+                     .ValueOrDie();
+    ASSERT_TRUE(array->ValidateFull().ok());
+    CheckNormalizedSlice(array, /*offset=*/1, /*length=*/2);
+}
+
+TEST(ArrowUtilsTest, TestDictionaryArrayMemoryUsage) {
+    auto indices =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[0, null, 1]").ValueOrDie();
+    auto dictionary =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::utf8(), R"(["alpha", "beta"])")
+            .ValueOrDie();
+    auto array = arrow::DictionaryArray::FromArrays(
+                     arrow::dictionary(arrow::int32(), arrow::utf8()), indices, dictionary)
+                     .ValueOrDie();
+    ASSERT_EQ(ArrowUtils::GetArrayMemoryUsage(array->data()), arrow::util::TotalBufferSize(*array));
+}
+
 TEST(ArrowUtilsTest, TestEqualsIgnoreNullable) {
     {
         // test simple
@@ -1136,6 +1167,11 @@ TEST(ArrowUtilsTest, TestResolveDictionaryStructTypeFromLayout) {
     auto logical_type =
         arrow::struct_({arrow::field("s", arrow::utf8()), arrow::field("i", arrow::int32())});
 
+    {
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> resolved,
+                             resolve(ints, arrow::int32()));
+        ASSERT_TRUE(resolved->Equals(*arrow::int32()));
+    }
     {
         // No dictionary: the very same type instance comes back.
         auto batch = arrow::StructArray::Make({strings, ints}, std::vector<std::string>{"s", "i"})
@@ -1252,6 +1288,19 @@ TEST(ArrowUtilsTest, TestFlattenUnresolvableDictionaries) {
                 .ValueOrDie())
             .ValueOrDie();
 
+    {
+        // A column outside the requested logical schema retains its original encoding.
+        auto batch = arrow::StructArray::Make({layout_unrecoverable, ints},
+                                              std::vector<std::string>{"unprojected", "i"})
+                         .ValueOrDie();
+        auto projected_type = arrow::struct_({arrow::field("i", arrow::int32())});
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::StructArray> flattened,
+                             ArrowUtils::FlattenUnresolvableDictionaries(
+                                 batch, projected_type, pool,
+                                 /*preserve_layout_recoverable_dictionaries=*/true));
+        ASSERT_TRUE(flattened->Equals(*batch));
+        ASSERT_EQ(flattened->field(0)->type_id(), arrow::Type::DICTIONARY);
+    }
     {
         // Only the column that would not survive the export is decoded; the one that would keeps
         // its encoding, which is what makes this selective rather than an all-or-nothing flatten.
diff --git a/src/paimon/core/io/data_file_meta_test.cpp b/src/paimon/core/io/data_file_meta_test.cpp
index b35631e42..551bc3ffe 100644
--- a/src/paimon/core/io/data_file_meta_test.cpp
+++ b/src/paimon/core/io/data_file_meta_test.cpp
@@ -18,9 +18,11 @@
 
 #include "paimon/core/io/data_file_meta.h"
 
+#include <functional>
 #include <limits>
 
 #include "gtest/gtest.h"
+#include "paimon/memory/bytes.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/binary_row_generator.h"
 #include "paimon/testing/utils/testharness.h"
@@ -61,6 +63,73 @@ TEST(DataFileMetaTest, TestCopyWithoutStats) {
     ASSERT_TRUE(upgraded->value_stats_cols->empty());
 }
 
+TEST(DataFileMetaTest, TestCopyWithColumnMaxSequenceNumbers) {
+    auto pool = GetDefaultPool();
+    auto key_stats = BinaryRowGenerator::GenerateStats({1}, {5}, {0}, pool.get());
+    auto value_stats = BinaryRowGenerator::GenerateStats({1, std::string("a")},
+                                                         {5, std::string("z")}, {0, 1}, pool.get());
+    DataFileMeta file_meta(
+        "data-0.orc", /*file_size=*/645, /*row_count=*/5,
+        BinaryRowGenerator::GenerateRow({1}, pool.get()),
+        BinaryRowGenerator::GenerateRow({5}, pool.get()), key_stats, value_stats,
+        /*min_sequence_number=*/100, /*max_sequence_number=*/104, /*schema_id=*/2, /*level=*/1,
+        /*extra_files=*/{"data-0.orc.index"}, Timestamp(1737111915429ll, 0),
+        /*delete_row_count=*/2, std::make_shared<Bytes>("index", pool.get()), FileSource::Append(),
+        /*value_stats_cols=*/std::vector<std::string>{"f0", "f1"},
+        /*external_path=*/"external/data-0.orc", /*first_row_id=*/10,
+        /*write_cols=*/std::vector<std::string>{"f0", "f1"},
+        /*column_max_sequence_numbers=*/std::vector<int64_t>{103, 104});
+    const DataFileMeta original = file_meta;
+    const std::vector<std::optional<std::vector<int64_t>>> replacements = {
+        std::vector<int64_t>{104, 104}, std::nullopt};
+    for (const auto& sequences : replacements) {
+        auto copied = file_meta.CopyWithColumnMaxSequenceNumbers(sequences);
+        ASSERT_TRUE(copied);
+        DataFileMeta expected = original;
+        expected.column_max_sequence_numbers = sequences;
+        ASSERT_EQ(*copied, expected);
+        ASSERT_EQ(file_meta, original);
+    }
+}
+
+TEST(DataFileMetaTest, TestComparisonOfOptionalMetadata) {
+    auto pool = GetDefaultPool();
+    ASSERT_OK_AND_ASSIGN(auto file_meta,
+                         DataFileMeta::ForAppend("data-0.orc", 100, 2, SimpleStats::EmptyStats(), 0,
+                                                 1, 0, FileSource::Append(), std::nullopt,
+                                                 "external/data-0.orc", 10, std::nullopt));
+    file_meta->extra_files = {"changelog-0.orc", std::nullopt};
+    file_meta->embedded_index = std::make_shared<Bytes>("index", pool.get());
+    file_meta->ToString();
+    ASSERT_TRUE(file_meta->TEST_Equal(*file_meta));
+    DataFileMeta renamed = *file_meta;
+    renamed.file_name = "data-1.orc";
+    renamed.file_size = 200;
+    renamed.creation_time = Timestamp(1737111915429ll, 0);
+    renamed.extra_files = {"changelog-1.orc", std::nullopt};
+    renamed.external_path = "other/data-1.orc";
+    ASSERT_TRUE(file_meta->TEST_Equal(renamed));
+    ASSERT_TRUE(renamed.TEST_Equal(*file_meta));
+
+    const std::vector<std::pair<const char*, std::function<void(DataFileMeta&)>>> changes = {
+        {"extra file count", [](DataFileMeta& meta) { meta.extra_files.pop_back(); }},
+        {"missing extra file", [](DataFileMeta& meta) { meta.extra_files[0] = std::nullopt; }},
+        {"missing embedded index", [](DataFileMeta& meta) { meta.embedded_index.reset(); }},
+        {"different embedded index",
+         [&](DataFileMeta& meta) {
+             meta.embedded_index = std::make_shared<Bytes>("other", pool.get());
+         }},
+        {"missing external path", [](DataFileMeta& meta) { meta.external_path.reset(); }},
+    };
+    for (const auto& [name, change] : changes) {
+        SCOPED_TRACE(name);
+        DataFileMeta other = *file_meta;
+        change(other);
+        ASSERT_FALSE(file_meta->TEST_Equal(other));
+        ASSERT_FALSE(other.TEST_Equal(*file_meta));
+    }
+}
+
 TEST(DataFileMetaTest, TestAddRowCount) {
     DataFileMeta file_meta("data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0.orc", /*file_size=*/645,
                            /*row_count=*/5, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
diff --git a/src/paimon/core/io/infer_shredding_file_writer_test.cpp b/src/paimon/core/io/infer_shredding_file_writer_test.cpp
index ff1cc4053..2db82487e 100644
--- a/src/paimon/core/io/infer_shredding_file_writer_test.cpp
+++ b/src/paimon/core/io/infer_shredding_file_writer_test.cpp
@@ -47,7 +47,9 @@ class CollectingFileWriter : public SingleFileWriter<::ArrowArray*, std::shared_
         : SingleFileWriter("", std::function<Status(::ArrowArray*, ::ArrowArray*)>()),
           converter_(converter),
           logical_type_(logical_type),
-          sink_(sink) {}
+          sink_(sink) {
+        path_ = "collected.parquet";
+    }
 
     Status Write(::ArrowArray* record) override {
         record_count_ += record->length;
@@ -75,20 +77,24 @@ class CollectingFileWriter : public SingleFileWriter<::ArrowArray*, std::shared_
     }
 
     Result<bool> ReachTargetSize(bool suggested_check, int64_t target_size) override {
-        return false;
+        return reach_target_size;
     }
 
     Result<AbortExecutor> GetAbortExecutor() const override {
         return AbortExecutor(nullptr, "");
     }
 
-    void Abort() override {}
+    void Abort() override {
+        aborted = true;
+    }
 
     int64_t RecordCount() const override {
         return record_count_;
     }
 
     bool closed = false;
+    bool aborted = false;
+    bool reach_target_size = false;
 
  private:
     std::shared_ptr<ShreddingBatchConverter> converter_;
@@ -211,4 +217,29 @@ TEST_F(InferShreddingFileWriterTest, EmptyFileUsesUntypedVariantPlan) {
     ASSERT_TRUE(inner_->closed);
 }
 
+TEST_F(InferShreddingFileWriterTest, AbortWhileBufferingDoesNotCreateInnerWriter) {
+    auto writer = MakeWriter(/*buffer_rows=*/4);
+    ASSERT_OK(WriteBatch(writer.get(), {R"({"a":1})"}));
+    ASSERT_EQ(writer->GetPath(), "");
+    writer->Abort();
+    ASSERT_TRUE(captured_converters_.empty());
+    ASSERT_TRUE(sink_.empty());
+}
+
+TEST_F(InferShreddingFileWriterTest, ForwardsFinalizedWriterOperations) {
+    auto writer = MakeWriter(/*buffer_rows=*/1);
+    ASSERT_OK(WriteBatch(writer.get(), {R"({"a":1})"}));
+    ASSERT_NE(inner_, nullptr);
+    ASSERT_EQ(writer->GetPath(), "collected.parquet");
+    ASSERT_OK_AND_ASSIGN(bool below_target, writer->ReachTargetSize(true, 100));
+    ASSERT_FALSE(below_target);
+    inner_->reach_target_size = true;
+    ASSERT_OK_AND_ASSIGN(bool reached_target, writer->ReachTargetSize(true, 100));
+    ASSERT_TRUE(reached_target);
+
+    ASSERT_OK(writer->Close());
+    writer->Abort();
+    ASSERT_TRUE(inner_->aborted);
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/core/io/row_to_arrow_array_converter_test.cpp b/src/paimon/core/io/row_to_arrow_array_converter_test.cpp
new file mode 100644
index 000000000..066d78d14
--- /dev/null
+++ b/src/paimon/core/io/row_to_arrow_array_converter_test.cpp
@@ -0,0 +1,101 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/io/row_to_arrow_array_converter.h"
+
+#include <memory>
+#include <string>
+
+#include "arrow/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+
+// The concrete converters select view access, so expose the owning-value appender here.
+class RowToArrowArrayConverterTestAccess
+    : public RowToArrowArrayConverter<BinaryRow, BatchReader::ReadBatch> {
+ public:
+    using RowToArrowArrayConverter::AppendField;
+};
+
+}  // namespace
+
+TEST(RowToArrowArrayConverterTest, AppendStringWithoutView) {
+    auto pool = GetDefaultPool();
+    auto arrow_pool = GetArrowPool(pool);
+    arrow::StringBuilder builder(arrow_pool.get());
+    int32_t reserve_count = 0;
+    ASSERT_OK_AND_ASSIGN(auto append, RowToArrowArrayConverterTestAccess::AppendField(
+                                          /*use_view=*/false, &builder, &reserve_count));
+
+    const std::string embedded_null("a\0b", 3);
+    auto row =
+        BinaryRowGenerator::GenerateRow({"paimon", "", NullType(), embedded_null}, pool.get());
+    for (int32_t i = 0; i < row.GetFieldCount(); ++i) {
+        ASSERT_TRUE(append(row, i).ok());
+    }
+
+    std::shared_ptr<arrow::Array> result;
+    ASSERT_TRUE(builder.Finish(&result).ok());
+    auto strings = std::dynamic_pointer_cast<arrow::StringArray>(result);
+    ASSERT_TRUE(strings);
+    ASSERT_EQ(strings->length(), 4);
+    ASSERT_EQ(strings->null_count(), 1);
+    ASSERT_EQ(strings->GetString(0), "paimon");
+    ASSERT_TRUE(strings->IsValid(1));
+    ASSERT_EQ(strings->GetString(1), "");
+    ASSERT_TRUE(strings->IsNull(2));
+    ASSERT_EQ(strings->GetString(3), embedded_null);
+}
+
+TEST(RowToArrowArrayConverterTest, AppendBinaryWithoutView) {
+    auto pool = GetDefaultPool();
+    auto arrow_pool = GetArrowPool(pool);
+    arrow::BinaryBuilder builder(arrow_pool.get());
+    int32_t reserve_count = 0;
+    ASSERT_OK_AND_ASSIGN(auto append, RowToArrowArrayConverterTestAccess::AppendField(
+                                          /*use_view=*/false, &builder, &reserve_count));
+
+    const std::string binary_value("\0\xff\x01", 3);
+    auto row =
+        BinaryRowGenerator::GenerateRow({"paimon", "", NullType(), binary_value}, pool.get());
+    for (int32_t i = 0; i < row.GetFieldCount(); ++i) {
+        ASSERT_TRUE(append(row, i).ok());
+    }
+
+    std::shared_ptr<arrow::Array> result;
+    ASSERT_TRUE(builder.Finish(&result).ok());
+    auto binaries = std::dynamic_pointer_cast<arrow::BinaryArray>(result);
+    ASSERT_TRUE(binaries);
+    ASSERT_EQ(binaries->length(), 4);
+    ASSERT_EQ(binaries->null_count(), 1);
+    ASSERT_EQ(binaries->GetString(0), "paimon");
+    ASSERT_TRUE(binaries->IsValid(1));
+    ASSERT_EQ(binaries->GetString(1), "");
+    ASSERT_TRUE(binaries->IsNull(2));
+    ASSERT_EQ(binaries->GetString(3), binary_value);
+}
+
+}  // namespace paimon::test
diff --git a/src/paimon/core/io/shredding_key_value_data_file_writer_factory_test.cpp b/src/paimon/core/io/shredding_key_value_data_file_writer_factory_test.cpp
new file mode 100644
index 000000000..231bc9ad0
--- /dev/null
+++ b/src/paimon/core/io/shredding_key_value_data_file_writer_factory_test.cpp
@@ -0,0 +1,151 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/io/shredding_key_value_data_file_writer_factory.h"
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/json_simple.h"
+#include "gtest/gtest.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/defs.h"
+#include "paimon/format/file_format.h"
+#include "paimon/format/reader_builder.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/read_result_collector.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+
+// A valid plan may decide that a particular file does not benefit from conversion.
+class PlainFallbackWritePlanFactory : public ShreddingWritePlanFactory {
+ public:
+    explicit PlainFallbackWritePlanFactory(int32_t buffer_rows) : buffer_rows_(buffer_rows) {}
+
+    bool ShouldCreateWritePlan() const override {
+        return true;
+    }
+    bool ShouldInferWritePlan() const override {
+        return buffer_rows_ > 0;
+    }
+    int32_t InferBufferRowCount() const override {
+        return buffer_rows_;
+    }
+    Result<std::shared_ptr<ShreddingBatchConverter>> CreateConverter(
+        const std::string&, const std::vector<std::shared_ptr<arrow::Array>>&) override {
+        return std::shared_ptr<ShreddingBatchConverter>();
+    }
+    MetadataFinalizer CreateMetadataFinalizer(const std::shared_ptr<ShreddingBatchConverter>&,
+                                              const std::string&) const override {
+        return nullptr;
+    }
+    Status OnFileCompleted(const std::shared_ptr<ShreddingBatchConverter>&) override {
+        ++completed_files;
+        return Status::OK();
+    }
+
+    int32_t completed_files = 0;
+
+ private:
+    int32_t buffer_rows_;
+};
+
+}  // namespace
+
+TEST(ShreddingKeyValueDataFileWriterFactoryTest, PlainFallbackPreservesBatchesAndMetadata) {
+    using Writer = SingleFileWriter<KeyValueBatch, std::shared_ptr<DataFileMeta>>;
+    auto pool = GetDefaultPool();
+    auto schema = SpecialFields::CompleteSequenceAndValueKindField(
+        arrow::schema({arrow::field("k", arrow::int32()), arrow::field("v", arrow::utf8())}));
+    auto data = arrow::ipc::internal::json::ArrayFromJSON(
+                    arrow::struct_(schema->fields()),
+                    R"([[10, 0, 1, "a"], [11, 0, 2, null], [12, 0, 3, "c"]])")
+                    .ValueOrDie();
+    auto record_batch = arrow::RecordBatch::FromStructArray(data).ValueOrDie();
+    // Cover immediate planning, inference with subsequent writes, inference on the last batch,
+    // and inference at Close().
+    for (int32_t buffer_rows : {0, 2, 3, 10}) {
+        SCOPED_TRACE(buffer_rows);
+        auto dir = UniqueTestDirectory::Create();
+        ASSERT_TRUE(dir);
+        ASSERT_OK_AND_ASSIGN(CoreOptions options,
+                             CoreOptions::FromMap({{Options::FILE_FORMAT, "parquet"}}));
+        auto path_factory = std::make_shared<DataFilePathFactory>();
+        ASSERT_OK(path_factory->Init(dir->Str(), "parquet", "data-", nullptr));
+        auto plan_factory = std::make_shared<PlainFallbackWritePlanFactory>(buffer_rows);
+        ShreddingKeyValueDataFileWriterFactory factory(
+            options, /*schema_id=*/7, schema, /*level=*/0, FileSource::Append(),
+            /*primary_keys=*/{"k"}, path_factory, /*create_stats_extractor=*/true, plan_factory,
+            /*is_changelog=*/false, pool);
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<Writer> writer, factory.CreateWriter());
+        for (int32_t offset : {0, 2}) {
+            auto slice = record_batch->Slice(offset, 2);
+            KeyValueBatch batch;
+            batch.min_sequence_number = 10 + offset;
+            batch.max_sequence_number = 10 + offset + slice->num_rows() - 1;
+            batch.min_key = BinaryRowGenerator::GenerateRowPtr({offset + 1}, pool.get());
+            batch.max_key = BinaryRowGenerator::GenerateRowPtr(
+                {offset + static_cast<int32_t>(slice->num_rows())}, pool.get());
+            batch.batch = std::make_unique<ArrowArray>();
+            auto export_status = arrow::ExportRecordBatch(*slice, batch.batch.get());
+            ASSERT_TRUE(export_status.ok()) << export_status.ToString();
+            ASSERT_OK(writer->Write(std::move(batch)));
+        }
+        ASSERT_EQ(plan_factory->completed_files, 0);
+        ASSERT_OK(writer->Close());
+        ASSERT_EQ(plan_factory->completed_files, 1);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataFileMeta> meta, writer->GetResult());
+        ASSERT_EQ(meta->row_count, 3);
+        ASSERT_EQ(meta->schema_id, 7);
+        ASSERT_EQ(meta->min_sequence_number, 10);
+        ASSERT_EQ(meta->max_sequence_number, 12);
+        ASSERT_EQ(meta->min_key.GetInt(0), 1);
+        ASSERT_EQ(meta->max_key.GetInt(0), 3);
+
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReaderBuilder> reader_builder,
+                             options.GetFileFormat()->CreateReaderBuilder(/*batch_size=*/2));
+        reader_builder->WithMemoryPool(pool);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input,
+                             options.GetFileSystem()->Open(writer->GetPath()));
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileBatchReader> reader, reader_builder->Build(input));
+        ArrowSchema c_schema;
+        ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok());
+        ASSERT_OK(reader->SetReadSchema(&c_schema, /*predicate=*/nullptr,
+                                        /*selection_bitmap=*/std::nullopt));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+                             ReadResultCollector::CollectResult(std::move(reader)));
+        ASSERT_TRUE(result);
+        ASSERT_TRUE(result->Equals(arrow::ChunkedArray(data))) << result->ToString();
+    }
+}
+
+}  // namespace paimon::test
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils_test.cpp b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils_test.cpp
new file mode 100644
index 000000000..0b0364322
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils_test.cpp
@@ -0,0 +1,131 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h"
+
+#include <cstdint>
+#include <memory>
+#include <string_view>
+#include <unordered_set>
+#include <vector>
+
+#include "arrow/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/generic_array.h"
+#include "paimon/common/data/generic_map.h"
+#include "paimon/common/data/generic_row.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(FieldAggregateUtilsTest, OwnedBinaryReusesOwnedBuffer) {
+    auto pool = GetDefaultPool();
+    auto bytes = std::make_shared<Bytes>("owned-value", pool.get());
+    VariantType result = FieldAggregateUtils::OwnedBinary(VariantType(bytes), pool.get());
+    ASSERT_EQ(DataDefine::GetVariantValue<std::shared_ptr<Bytes>>(result), bytes);
+    ASSERT_EQ(DataDefine::GetStringView(result), "owned-value");
+}
+
+TEST(FieldAggregateUtilsTest, OwnedBinaryCopiesEmptyView) {
+    auto pool = GetDefaultPool();
+    VariantType result =
+        FieldAggregateUtils::OwnedBinary(VariantType(std::string_view()), pool.get());
+    const auto* bytes = DataDefine::GetVariantPtr<std::shared_ptr<Bytes>>(result);
+    ASSERT_TRUE(bytes);
+    ASSERT_TRUE(*bytes);
+    ASSERT_EQ((*bytes)->size(), 0);
+}
+
+TEST(FieldAggregateUtilsTest, PrimitiveExtractionEqualityAndHashing) {
+    struct TestCase {
+        std::shared_ptr<arrow::DataType> type;
+        VariantType value;
+        VariantType other;
+    };
+    std::vector<TestCase> cases = {{arrow::boolean(), true, false},
+                                   {arrow::int8(), char{7}, char{8}},
+                                   {arrow::int16(), int16_t{-1234}, int16_t{1234}},
+                                   {arrow::date32(), int32_t{20000}, int32_t{20001}},
+                                   {arrow::int64(), int64_t{1} << 40, -(int64_t{1} << 40)}};
+    for (const auto& test_case : cases) {
+        SCOPED_TRACE(test_case.type->ToString());
+        GenericRow row(2);
+        row.SetField(0, test_case.value);
+        row.SetField(1, test_case.other);
+        ASSERT_OK_AND_ASSIGN(VariantType value,
+                             FieldAggregateUtils::GetValue(row, 0, test_case.type));
+        ASSERT_OK_AND_ASSIGN(VariantType other,
+                             FieldAggregateUtils::GetValue(row, 1, test_case.type));
+        ASSERT_EQ(value, test_case.value);
+        ASSERT_EQ(other, test_case.other);
+
+        ASSERT_OK_AND_ASSIGN(bool equal,
+                             FieldAggregateUtils::Equals(value, test_case.value, test_case.type));
+        ASSERT_TRUE(equal);
+        ASSERT_OK_AND_ASSIGN(bool different,
+                             FieldAggregateUtils::Equals(value, other, test_case.type));
+        ASSERT_FALSE(different);
+
+        ASSERT_TRUE(FieldAggregateUtils::IsHashableType(test_case.type));
+        std::unordered_set<VariantType, detail::SemanticHash, detail::SemanticEqual> values(
+            0, detail::SemanticHash{test_case.type}, detail::SemanticEqual{test_case.type});
+        ASSERT_TRUE(values.insert(test_case.value).second);
+        ASSERT_FALSE(values.insert(value).second);
+        ASSERT_TRUE(values.insert(other).second);
+        ASSERT_EQ(values.size(), 2);
+    }
+}
+
+TEST(FieldAggregateUtilsTest, RowsWithDifferentFieldCountsAreNotEqual) {
+    std::shared_ptr<InternalRow> lhs = std::make_shared<GenericRow>(1);
+    std::shared_ptr<InternalRow> rhs = std::make_shared<GenericRow>(2);
+    auto type = arrow::struct_({arrow::field("f", arrow::int32())});
+    ASSERT_OK_AND_ASSIGN(bool equal,
+                         FieldAggregateUtils::Equals(VariantType(lhs), VariantType(rhs), type));
+    ASSERT_FALSE(equal);
+}
+
+TEST(FieldAggregateUtilsTest, ArraysWithDifferentLengthsAreNotEqual) {
+    std::shared_ptr<InternalArray> lhs =
+        std::make_shared<GenericArray>(std::vector<VariantType>{int32_t{1}});
+    std::shared_ptr<InternalArray> rhs =
+        std::make_shared<GenericArray>(std::vector<VariantType>{int32_t{1}, int32_t{2}});
+    ASSERT_OK_AND_ASSIGN(bool equal, FieldAggregateUtils::Equals(VariantType(lhs), VariantType(rhs),
+                                                                 arrow::list(arrow::int32())));
+    ASSERT_FALSE(equal);
+}
+
+TEST(FieldAggregateUtilsTest, MapsWithMultipleEntriesCompareByContent) {
+    auto make_map = []() -> std::shared_ptr<InternalMap> {
+        auto keys =
+            std::make_shared<GenericArray>(std::vector<VariantType>{int32_t{1}, int32_t{2}});
+        auto values =
+            std::make_shared<GenericArray>(std::vector<VariantType>{int32_t{10}, int32_t{20}});
+        return std::make_shared<GenericMap>(keys, values);
+    };
+    VariantType lhs(make_map());
+    VariantType rhs(make_map());
+    ASSERT_OK_AND_ASSIGN(bool equal, FieldAggregateUtils::Equals(
+                                         lhs, rhs, arrow::map(arrow::int32(), arrow::int32())));
+    ASSERT_TRUE(equal);
+}
+
+}  // namespace paimon::test
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_nested_update_agg_test.cpp b/src/paimon/core/mergetree/compact/aggregate/field_nested_update_agg_test.cpp
index 805850d46..d57732cdb 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_nested_update_agg_test.cpp
+++ b/src/paimon/core/mergetree/compact/aggregate/field_nested_update_agg_test.cpp
@@ -123,6 +123,60 @@ TEST(FieldNestedUpdateAggTest, AppendsNonNullRowsUpToLimit) {
     ASSERT_TRUE(FindRow(result, 2));
 }
 
+TEST(FieldNestedUpdateAggTest, NullFieldsLeaveAccumulatorUnchanged) {
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldNestedUpdateAgg> agg, MakeAgg({}));
+    VariantType null_field = VariantType(NullType());
+    VariantType accumulator = Rows({Row(int32_t{1}, 1, 10)});
+
+    ASSERT_OK_AND_ASSIGN(VariantType aggregated, agg->Agg(accumulator, null_field));
+    ASSERT_EQ(1, GetRows(aggregated)->Size());
+    ASSERT_TRUE(FindRow(aggregated, 1));
+    ASSERT_EQ(10, FindRow(aggregated, 1)->GetInt(2));
+
+    ASSERT_OK_AND_ASSIGN(VariantType retracted, agg->Retract(accumulator, null_field));
+    ASSERT_EQ(1, GetRows(retracted)->Size());
+    ASSERT_TRUE(FindRow(retracted, 1));
+    ASSERT_EQ(10, FindRow(retracted, 1)->GetInt(2));
+
+    ASSERT_OK_AND_ASSIGN(VariantType empty, agg->Retract(null_field, accumulator));
+    ASSERT_TRUE(DataDefine::IsVariantNull(empty));
+}
+
+TEST(FieldNestedUpdateAggTest, SkipsNullElementsWhenUpsertingByKey) {
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldNestedUpdateAgg> agg,
+                         MakeAgg({{"fields.f.nested-key", "id"}}));
+    VariantType accumulator = Rows({VariantType(NullType()), Row(int32_t{1}, 1, 10)});
+    VariantType input = Rows({VariantType(NullType()), Row(int32_t{1}, 2, 20),
+                              VariantType(NullType()), Row(int32_t{2}, 1, 30)});
+    ASSERT_OK_AND_ASSIGN(VariantType result, agg->Agg(accumulator, input));
+
+    ASSERT_EQ(2, GetRows(result)->Size());
+    ASSERT_TRUE(FindRow(result, 1));
+    ASSERT_EQ(20, FindRow(result, 1)->GetInt(2));
+    ASSERT_TRUE(FindRow(result, 2));
+    ASSERT_EQ(30, FindRow(result, 2)->GetInt(2));
+}
+
+TEST(FieldNestedUpdateAggTest, SkipsNullElementsWhenRetracting) {
+    for (bool keyed : {false, true}) {
+        SCOPED_TRACE(keyed);
+        std::map<std::string, std::string> options;
+        if (keyed) {
+            options.emplace("fields.f.nested-key", "id");
+        }
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldNestedUpdateAgg> agg, MakeAgg(options));
+        VariantType accumulator =
+            Rows({VariantType(NullType()), Row(int32_t{1}, 1, 10), Row(int32_t{2}, 1, 20)});
+        VariantType input = Rows({VariantType(NullType()), Row(int32_t{1}, 1, 10)});
+        ASSERT_OK_AND_ASSIGN(VariantType result, agg->Retract(accumulator, input));
+
+        ASSERT_EQ(1, GetRows(result)->Size());
+        ASSERT_FALSE(FindRow(result, 1));
+        ASSERT_TRUE(FindRow(result, 2));
+        ASSERT_EQ(20, FindRow(result, 2)->GetInt(2));
+    }
+}
+
 // count limit is measured against the raw element count, so null elements consume the limit even
 // though they are dropped from the result
 TEST(FieldNestedUpdateAggTest, CountLimitCountsNullElementsOfAccumulator) {
diff --git a/src/paimon/core/mergetree/compact/partial_update_merge_function_test.cpp b/src/paimon/core/mergetree/compact/partial_update_merge_function_test.cpp
index f937baf28..1f3000343 100644
--- a/src/paimon/core/mergetree/compact/partial_update_merge_function_test.cpp
+++ b/src/paimon/core/mergetree/compact/partial_update_merge_function_test.cpp
@@ -851,6 +851,54 @@ TEST_F(PartialUpdateMergeFunctionTest, TestDeleteReproduceCorrectSequenceNumber)
     ASSERT_EQ(result->sequence_number, 1);
 }
 
+TEST_F(PartialUpdateMergeFunctionTest, TestIgnoreDeleteInitializesFromFirstRetract) {
+    auto mfunc = CreateMergeFunction(/*value_arity=*/3, {{"ignore-delete", "true"}});
+    mfunc->Reset();
+    Add(mfunc, RowKind::Delete(), {1, 10, 20});
+    Add(mfunc, RowKind::Delete(), {1, 100, 200});
+    CheckResult(mfunc, {1, 10, 20});
+
+    Add(mfunc, RowKind::UpdateAfter(), {1, NullType(), 30});
+    CheckResult(mfunc, {1, 10, 30});
+    ASSERT_OK_AND_ASSIGN(auto result, mfunc->GetResult());
+    ASSERT_TRUE(result.has_value());
+    ASSERT_EQ(result->value_kind, RowKind::Insert());
+}
+
+TEST_F(PartialUpdateMergeFunctionTest, TestRemoveRecordOnDeleteStartsWithUpdateBefore) {
+    auto mfunc = CreateMergeFunction(/*value_arity=*/3,
+                                     {{"partial-update.remove-record-on-delete", "true"}});
+    mfunc->Reset();
+    Add(mfunc, RowKind::UpdateBefore(), {1, 10, 20});
+    Add(mfunc, RowKind::UpdateBefore(), {1, 100, 200});
+    CheckResult(mfunc, {1, 10, 20});
+
+    Add(mfunc, RowKind::UpdateAfter(), {1, NullType(), 30});
+    CheckResult(mfunc, {1, 10, 30});
+    ASSERT_OK_AND_ASSIGN(auto result, mfunc->GetResult());
+    ASSERT_TRUE(result.has_value());
+    ASSERT_EQ(result->value_kind, RowKind::Insert());
+    ASSERT_EQ(result->sequence_number, 2);
+}
+
+TEST_F(PartialUpdateMergeFunctionTest, TestSequenceGroupStartsWithRetract) {
+    auto mfunc = CreateMergeFunction(/*value_arity=*/4, {{"fields.f2.sequence-group", "f1"}});
+    mfunc->Reset();
+    Add(mfunc, RowKind::Delete(), {1, 10, 5, 30});
+    CheckResult(mfunc, {1, NullType(), 5, 30});
+    ASSERT_OK_AND_ASSIGN(auto deleted, mfunc->GetResult());
+    ASSERT_TRUE(deleted.has_value());
+    ASSERT_EQ(deleted->value_kind, RowKind::Delete());
+
+    mfunc->Reset();
+    Add(mfunc, RowKind::UpdateBefore(), {1, 10, 5, 30});
+    Add(mfunc, RowKind::UpdateAfter(), {1, 20, 6, NullType()});
+    CheckResult(mfunc, {1, 20, 6, 30});
+    ASSERT_OK_AND_ASSIGN(auto updated, mfunc->GetResult());
+    ASSERT_TRUE(updated.has_value());
+    ASSERT_EQ(updated->value_kind, RowKind::Insert());
+}
+
 TEST_F(PartialUpdateMergeFunctionTest, TestInitRowWithNullableFieldOnDelete) {
     std::map<std::string, std::string> options_map = {
         {"partial-update.remove-record-on-delete", "true"}};
diff --git a/src/paimon/core/operation/data_evolution_split_read_test.cpp b/src/paimon/core/operation/data_evolution_split_read_test.cpp
index a68dfdf8b..3bd6beb0b 100644
--- a/src/paimon/core/operation/data_evolution_split_read_test.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read_test.cpp
@@ -40,6 +40,7 @@
 #include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h"
 #include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
 #include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
+#include "paimon/core/global_index/indexed_split_impl.h"
 #include "paimon/core/io/data_file_meta.h"
 #include "paimon/core/manifest/file_source.h"
 #include "paimon/core/operation/internal_read_context.h"
@@ -438,6 +439,51 @@ TEST_F(DataEvolutionSplitReadTest, TestBlobFallbackValidatesRangesBeforeOpeningF
                         "should not have overlapping row id ranges");
 }
 
+TEST_F(DataEvolutionSplitReadTest, TestEmptyBlobReaders) {
+    auto dir = UniqueTestDirectory::Create("local");
+    auto schema =
+        DataField::ConvertDataFieldsToArrowSchema({DataField(0, arrow::field("id", arrow::int32())),
+                                                   DataField(1, BlobUtils::ToArrowField("b0"))});
+    const std::map<std::string, std::string> options = {{Options::FILE_FORMAT, "parquet"},
+                                                        {Options::ROW_TRACKING_ENABLED, "true"},
+                                                        {Options::DATA_EVOLUTION_ENABLED, "true"}};
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                         TableSchema::Create(0, schema, {}, {}, options));
+    ReadContextBuilder builder(dir->Str());
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context, builder.Finish());
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<InternalReadContext> context,
+                         InternalReadContext::Create(read_context, table_schema, options));
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<FileStorePathFactory> path_factory,
+        FileStorePathFactory::Create(dir->Str(), schema, {}, "default", "parquet", "data-",
+                                     /*legacy_partition_name_enabled=*/true, {}, std::nullopt,
+                                     /*index_file_in_data_file_dir=*/false, GetDefaultPool()));
+    DataEvolutionSplitRead split_read(path_factory, context, GetDefaultPool(),
+                                      /*executor=*/nullptr);
+    auto blob = CreateBlobFile("payload", 100, 10, 1, {{"b0"}});
+    auto blob_schema = arrow::schema({schema->field(1)});
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<BatchReader> fallback_reader,
+        split_read.CreateBlobFallbackReader(
+            BinaryRow::EmptyRow(), {blob}, Range(100, 109), blob_schema,
+            /*row_ranges=*/std::vector<Range>{}, /*data_file_path_factory=*/nullptr,
+            /*group_dv_factory=*/nullptr, /*group_dv=*/std::nullopt));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+                         ReadResultCollector::CollectResult(std::move(fallback_reader)));
+    ASSERT_FALSE(result);
+
+    // Dedicated payload files do not contain blob-view references for the pre-read pass.
+    ASSERT_OK_AND_ASSIGN(std::string bucket_path,
+                         path_factory->BucketPath(BinaryRow::EmptyRow(), 0));
+    DataSplitImpl::Builder split_builder(BinaryRow::EmptyRow(), 0, bucket_path, {blob});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataSplitImpl> split,
+                         split_builder.RawConvertible(false).Build());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> view_reader,
+                         split_read.CreateBlobViewReader(split, {"b0"}, std::nullopt));
+    ASSERT_OK_AND_ASSIGN(result, ReadResultCollector::CollectResult(std::move(view_reader)));
+    ASSERT_FALSE(result);
+}
+
 TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap) {
     auto blob_entry =
         CreateBlobFile("blob1", /*first_row_id=*/0, /*row_count=*/10,
@@ -861,6 +907,30 @@ TEST_F(DataEvolutionSplitReadTest, TestPlanIndependentlyRolledVectorReadRanges)
     ASSERT_EQ(vector1->row_count, 4);
 }
 
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesWithBlobAndNewerUpdate) {
+    auto main = CreateNormalFile("main.parquet", 100, 10, 0);
+    auto older = CreateDataFileMeta("older.vector.vortex", 100, 5, 1);
+    auto newer = CreateDataFileMeta("newer.vector.vortex", 100, 5, 2);
+    auto blob = CreateBlobFile("payload", 100, 10, 3, {{"payload"}});
+    ASSERT_OK_AND_ASSIGN(std::vector<std::vector<std::shared_ptr<DataFileMeta>>> groups,
+                         DataEvolutionSplitRead::MergeRangesAndSort({blob, older, main, newer}));
+    ASSERT_EQ(groups.size(), 1);
+    ASSERT_OK_AND_ASSIGN(
+        std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+        DataEvolutionSplitRead::PlanVectorReadRanges(groups[0], {0, 1, 2}, EmbeddingFieldIds));
+    ASSERT_EQ(ranges.size(), 2);
+    ASSERT_EQ(ranges[0].range, Range(100, 104));
+    ASSERT_EQ(ranges[0].files, (std::vector<std::shared_ptr<DataFileMeta>>{newer, main, blob}));
+    ASSERT_EQ(ranges[1].range, Range(105, 109));
+    ASSERT_EQ(ranges[1].files, (std::vector<std::shared_ptr<DataFileMeta>>{main, blob}));
+}
+
+TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesWithNoFiles) {
+    ASSERT_OK_AND_ASSIGN(std::vector<DataEvolutionSplitRead::VectorReadRange> ranges,
+                         DataEvolutionSplitRead::PlanVectorReadRanges({}, {1}, EmbeddingFieldIds));
+    ASSERT_TRUE(ranges.empty());
+}
+
 TEST_F(DataEvolutionSplitReadTest, TestPlanSequentialVectorRead) {
     auto main = CreateNormalFile("main.parquet", 100, 10, 0);
     auto vector0 = CreateDataFileMeta("vector0.vector.vortex", 100, 3, 1);
@@ -1058,6 +1128,70 @@ TEST_F(DataEvolutionSplitReadTest, TestSequentialVectorReadReusesMainReader) {
         ASSERT_TRUE(result->Equals(arrow::ChunkedArray(projected))) << result->ToString();
         ASSERT_EQ(counting_fs->OpenCount(), 1);
     }
+
+    // A selection confined to the first VECTOR file must not include rows from the second.
+    {
+        ReadContextBuilder builder(dir->Str());
+        builder.WithFileSystem(fs).EnablePrefetch(false).SetReadFieldNames(
+            {"id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"});
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context, builder.Finish());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<InternalReadContext> context,
+                             InternalReadContext::Create(read_context, table_schema, options));
+        DataEvolutionSplitRead split_read(path_factory, context, GetDefaultPool(), executor);
+        auto selected_split =
+            std::make_shared<IndexedSplitImpl>(split, std::vector<Range>{Range(0, 1)});
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader,
+                             split_read.CreateReader(selected_split));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+                             ReadResultCollector::CollectResult(std::move(reader)));
+        ASSERT_TRUE(result);
+        ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected->Slice(0, 2))))
+            << result->ToString();
+    }
+
+    // Older full-schema files may omit write_cols, while VECTOR metadata includes system fields.
+    {
+        SchemaManager schema_manager(fs, dir->Str());
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> stored_schema,
+                             schema_manager.CreateTable(schema, {}, {}, options));
+        ASSERT_EQ(stored_schema->Id(), 0);
+        auto evolved_fields = schema->fields();
+        evolved_fields.push_back(DataField(2, arrow::field("added", arrow::utf8())).ArrowField());
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> evolved_schema,
+            TableSchema::Create(1, arrow::schema(evolved_fields), {}, {}, options));
+        auto legacy_data = arrow::ipc::internal::json::ArrayFromJSON(
+                               arrow::struct_(schema->fields()),
+                               "[[0, null], [1, null], [2, null], [3, null], [4, null], [5, null]]")
+                               .ValueOrDie();
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataFileMeta> legacy_main,
+                             write_file("legacy.parquet", 0, legacy_data, 3));
+        legacy_main->write_cols = std::nullopt;
+        DataSplitImpl::Builder legacy_split_builder(BinaryRow::EmptyRow(), 0, bucket_path,
+                                                    {vector1, legacy_main, vector0});
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataSplitImpl> legacy_split,
+                             legacy_split_builder.RawConvertible(false).Build());
+        ReadContextBuilder builder(dir->Str());
+        builder.WithFileSystem(fs).EnablePrefetch(false).SetReadFieldNames(
+            {"id", "embedding", "_ROW_ID", "_SEQUENCE_NUMBER"});
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context, builder.Finish());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<InternalReadContext> context,
+                             InternalReadContext::Create(read_context, evolved_schema, options));
+        DataEvolutionSplitRead split_read(path_factory, context, GetDefaultPool(), executor);
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader,
+                             split_read.CreateReader(legacy_split));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+                             ReadResultCollector::CollectResult(std::move(reader)));
+        ASSERT_TRUE(result);
+        ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected))) << result->ToString();
+
+        vector0->write_cols->push_back(SpecialFields::RowId().Name());
+        vector0->write_cols->push_back(SpecialFields::SequenceNumber().Name());
+        ASSERT_OK_AND_ASSIGN(reader, split_read.CreateReader(legacy_split));
+        ASSERT_OK_AND_ASSIGN(result, ReadResultCollector::CollectResult(std::move(reader)));
+        ASSERT_TRUE(result);
+        ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected))) << result->ToString();
+    }
 }
 
 TEST_F(DataEvolutionSplitReadTest, TestPlanVectorReadRangesPreservesOldPrefixAndSuffix) {
@@ -1353,6 +1487,33 @@ TEST_F(DataEvolutionSplitReadTest, TestFileIndexSelectionOnlyPushesDownNormalGro
         ASSERT_TRUE(selection.has_value());
         ASSERT_TRUE(selection->IsEmpty());
     }
+
+    // The per-file reader also applies indexes when no group selection was prepared.
+    auto check_reader = [&](const std::shared_ptr<Predicate>& predicate,
+                            const std::optional<std::vector<Range>>& row_ranges,
+                            const std::string& expected_json) {
+        ASSERT_OK_AND_ASSIGN(
+            std::unique_ptr<FileBatchReader> reader,
+            split_read.ApplyIndexAndDvReaderIfNeeded(
+                std::make_unique<MockFileBatchReader>(data, data->type(), 2), normal, index_schema,
+                index_schema, predicate, /*dv_factory=*/nullptr, row_ranges,
+                /*data_file_path_factory=*/nullptr, /*file_index_options=*/{},
+                /*index_scores=*/nullptr));
+        if (expected_json.empty()) {
+            ASSERT_FALSE(reader);
+            return;
+        }
+        ASSERT_TRUE(reader);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+                             ReadResultCollector::CollectResult(std::move(reader)));
+        auto expected =
+            arrow::ipc::internal::json::ArrayFromJSON(data->type(), expected_json).ValueOrDie();
+        ASSERT_TRUE(result);
+        ASSERT_TRUE(result->Equals(arrow::ChunkedArray(expected))) << result->ToString();
+    };
+    check_reader(match, std::nullopt, "[[1]]");
+    check_reader(match, std::vector<Range>{Range(102, 103)}, "");
+    check_reader(miss, std::nullopt, "");
 }
 
 namespace {
@@ -1615,6 +1776,10 @@ TEST_F(DataEvolutionSplitReadTest, TestExcludeDeletedRowIds) {
     // anchor range starts at 100, deleted anchor positions {3, 4, 7} are row ids {103, 104, 107}
     DataEvolutionSplitRead::GroupDeletionVector group_dv{MakeBitmapDv({3, 4, 7}), Range(100, 199)};
 
+    ASSERT_OK_AND_ASSIGN(std::vector<Range> empty,
+                         DataEvolutionSplitRead::ExcludeDeletedRowIds({}, group_dv));
+    ASSERT_TRUE(empty.empty());
+
     ASSERT_OK_AND_ASSIGN(std::vector<Range> remaining,
                          DataEvolutionSplitRead::ExcludeDeletedRowIds({Range(102, 108)}, group_dv));
     std::vector<Range> expected = {Range(102, 102), Range(105, 106), Range(108, 108)};
diff --git a/src/paimon/core/snapshot_file_scan_test.cpp b/src/paimon/core/snapshot_file_scan_test.cpp
index 36e61cb4f..3f8bba11b 100644
--- a/src/paimon/core/snapshot_file_scan_test.cpp
+++ b/src/paimon/core/snapshot_file_scan_test.cpp
@@ -32,16 +32,25 @@
 #include <vector>
 
 #include "gtest/gtest.h"
+#include "paimon/commit_context.h"
 #include "paimon/common/factories/io_hook.h"
 #include "paimon/common/utils/path_util.h"
 #include "paimon/common/utils/scope_guard.h"
 #include "paimon/common/utils/string_utils.h"
+#include "paimon/core/operation/commit/realtime_commit_properties.h"
+#include "paimon/core/snapshot.h"
+#include "paimon/core/table/sink/commit_message_impl.h"
+#include "paimon/core/utils/snapshot_manager.h"
 #include "paimon/defs.h"
+#include "paimon/file_store_commit.h"
+#include "paimon/file_store_write.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/predicate/predicate_builder.h"
+#include "paimon/realtime/realtime_context.h"
 #include "paimon/scan_context.h"
 #include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
+#include "paimon/write_context.h"
 
 namespace paimon::test {
 namespace {
@@ -49,8 +58,9 @@ namespace {
 Result<std::set<std::string>> ListFiles(const std::string& table_path,
                                         const std::optional<int64_t>& snapshot_id = std::nullopt,
                                         const std::shared_ptr<ScanFilter>& scan_filter = nullptr,
-                                        const std::string& branch = "main") {
-    return SnapshotFileScan::ListFiles(table_path, branch, snapshot_id, scan_filter, /*options=*/{},
+                                        const std::string& branch = "main",
+                                        const std::map<std::string, std::string>& options = {}) {
+    return SnapshotFileScan::ListFiles(table_path, branch, snapshot_id, scan_filter, options,
                                        /*file_system=*/nullptr, /*executor=*/nullptr,
                                        /*memory_pool=*/nullptr);
 }
@@ -184,6 +194,15 @@ TEST(SnapshotFileScanTest, TestSingleBucketManifestAvoidsExtraIO) {
                          ListFiles(table_path, std::nullopt, CreateFilter({}, 0)));
     ASSERT_EQ(bucket_files, all_files);
     ASSERT_EQ(io_hook->IOCount(), full_read_io_count);
+
+    SnapshotManager snapshot_manager(dir->GetFileSystem(), table_path);
+    ASSERT_OK_AND_ASSIGN(Snapshot snapshot, snapshot_manager.LoadSnapshot(1));
+    ASSERT_OK_AND_ASSIGN(std::set<std::string> unmatched_bucket_files,
+                         ListFiles(table_path, std::nullopt, CreateFilter({}, 1)));
+    ASSERT_EQ(ExpectedFiles(table_path, {"schema/schema-0", "snapshot/snapshot-1",
+                                         "manifest/" + snapshot.BaseManifestList(),
+                                         "manifest/" + snapshot.DeltaManifestList()}),
+              unmatched_bucket_files);
 }
 
 TEST(SnapshotFileScanTest, TestExplicitSnapshot) {
@@ -329,6 +348,20 @@ TEST(SnapshotFileScanTest, TestDeletionVectorIndexBucketFilter) {
                                    "manifest/manifest-list-540cf68e-698e-4b66-af30-eb558d8db43d-7",
                                    "schema/schema-0", "snapshot/snapshot-4"}),
         bucket_one_files);
+
+    ASSERT_OK_AND_ASSIGN(
+        std::set<std::string> matching_partition_files,
+        ListFiles(table_path, /*snapshot_id=*/4, CreateFilter({{{"f1", "10"}}}, 0)));
+    ASSERT_EQ(matching_partition_files, bucket_zero_files);
+    ASSERT_OK_AND_ASSIGN(
+        std::set<std::string> unmatched_partition_files,
+        ListFiles(table_path, /*snapshot_id=*/4, CreateFilter({{{"f1", "999"}}}, 0)));
+    ASSERT_EQ(unmatched_partition_files.count(
+                  table_path + "/index/index-86356766-3238-46e6-990b-656cd7409eaa-0"),
+              0);
+    ASSERT_EQ(unmatched_partition_files.count(
+                  table_path + "/manifest/index-manifest-59bcb792-830f-4b57-b838-cfcc50d29266-0"),
+              1);
 }
 
 TEST(SnapshotFileScanTest, TestGlobalIndex) {
@@ -355,7 +388,8 @@ TEST(SnapshotFileScanTest, TestBranch) {
                              "append_table_with_append_pt_branch";
 
     ASSERT_OK_AND_ASSIGN(std::set<std::string> files,
-                         ListFiles(table_path, std::nullopt, /*scan_filter=*/nullptr, "test"));
+                         ListFiles(table_path, std::nullopt, /*scan_filter=*/nullptr, "test",
+                                   {{Options::BRANCH, "main"}}));
     ASSERT_EQ(
         ExpectedFiles(table_path,
                       {"branch/branch-test/schema/schema-0", "branch/branch-test/schema/schema-1",
@@ -369,6 +403,106 @@ TEST(SnapshotFileScanTest, TestBranch) {
         files);
 }
 
+TEST(SnapshotFileScanTest, TestTableWithoutSnapshot) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto schema = arrow::schema({arrow::field("value", arrow::utf8())});
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<TestHelper> helper,
+        TestHelper::Create(dir->Str(), schema, {}, {}, {{Options::FILE_FORMAT, "orc"}}, false));
+    ASSERT_OK_AND_ASSIGN(std::set<std::string> files, ListFiles(dir->Str() + "/foo.db/bar"));
+    ASSERT_TRUE(files.empty());
+}
+
+TEST(SnapshotFileScanTest, TestCommittedChangelogFiles) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto schema =
+        arrow::schema({arrow::field("id", arrow::int64()), arrow::field("value", arrow::utf8())});
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TestHelper> helper,
+                         TestHelper::Create(dir->Str(), schema, {}, {"id"},
+                                            {{Options::FILE_FORMAT, "orc"},
+                                             {Options::BUCKET, "1"},
+                                             {Options::CHANGELOG_PRODUCER, "input"}},
+                                            /*is_streaming_mode=*/true));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
+                         TestHelper::MakeRecordBatch(arrow::struct_(schema->fields()),
+                                                     R"([[1,"a"],[2,"b"]])", {}, 0, {}));
+    ASSERT_OK_AND_ASSIGN(auto messages, helper->WriteAndCommit(std::move(batch), 0, std::nullopt));
+    const std::string table_path = dir->Str() + "/foo.db/bar";
+    SnapshotManager snapshot_manager(dir->GetFileSystem(), table_path);
+    ASSERT_OK_AND_ASSIGN(Snapshot snapshot, snapshot_manager.LoadSnapshot(1));
+    ASSERT_TRUE(snapshot.ChangelogManifestList().has_value());
+    ASSERT_OK_AND_ASSIGN(std::set<std::string> files, ListFiles(table_path));
+    ASSERT_EQ(files.count(table_path + "/manifest/" + snapshot.ChangelogManifestList().value()), 1);
+    ASSERT_EQ(messages.size(), 1);
+    auto message = std::dynamic_pointer_cast<CommitMessageImpl>(messages[0]);
+    ASSERT_TRUE(message);
+    const auto& changelog_files = message->GetNewFilesIncrement().ChangelogFiles();
+    ASSERT_FALSE(changelog_files.empty());
+    for (const auto& file : changelog_files) {
+        ASSERT_EQ(files.count(table_path + "/bucket-0/" + file->file_name), 1);
+    }
+    std::vector<BasicFileStatus> manifests;
+    ASSERT_OK(dir->GetFileSystem()->ListDir(table_path + "/manifest", &manifests));
+    for (const auto& manifest : manifests) {
+        ASSERT_EQ(files.count(manifest.GetPath()), 1);
+    }
+}
+
+TEST(SnapshotFileScanTest, TestCommittedRealtimeOffsetFiles) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto schema =
+        arrow::schema({arrow::field("id", arrow::int64()), arrow::field("value", arrow::utf8())});
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<TestHelper> helper,
+        TestHelper::Create(
+            dir->Str(), schema, {}, {},
+            {{Options::FILE_FORMAT, "orc"}, {Options::BUCKET, "1"}, {Options::BUCKET_KEY, "id"}},
+            /*is_streaming_mode=*/false));
+    helper.reset();
+    const std::string table_path = dir->Str() + "/foo.db/bar";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
+                         RealtimeContext::Create());
+    WriteContextBuilder write_builder(table_path, "test");
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context,
+                         write_builder.AddOption(Options::REALTIME_ENABLED, "true")
+                             .WithStreamingMode(true)
+                             .WithRealtimeContext(realtime_context)
+                             .Finish());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> writer,
+                         FileStoreWrite::Create(std::move(write_context)));
+    auto input_type = arrow::struct_({arrow::field("_REALTIME_OFFSET", arrow::int64(), false),
+                                      schema->field(0), schema->field(1)});
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<RecordBatch> batch,
+        TestHelper::MakeRecordBatch(input_type, R"([[0,1,"a"],[1,2,"b"]])", {}, 0, {}));
+    ASSERT_OK(writer->Write(std::move(batch)));
+    ASSERT_OK_AND_ASSIGN(auto progress, writer->PrepareCommitWithProgress(/*commit_identifier=*/0));
+    ASSERT_FALSE(progress.empty());
+    CommitContextBuilder commit_builder(table_path, "test");
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> commit_context,
+                         commit_builder.AddOption(Options::REALTIME_ENABLED, "true").Finish());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> commit,
+                         FileStoreCommit::Create(std::move(commit_context)));
+    ASSERT_OK_AND_ASSIGN(int64_t snapshot_id,
+                         commit->CommitWithProgress(progress, 0, std::nullopt));
+    ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot_id));
+    ASSERT_OK(writer->Close());
+
+    SnapshotManager snapshot_manager(dir->GetFileSystem(), table_path);
+    ASSERT_OK_AND_ASSIGN(Snapshot snapshot, snapshot_manager.LoadSnapshot(snapshot_id));
+    auto offsets_path = RealtimeCommitProperties::GetOffsetsPath(snapshot);
+    ASSERT_TRUE(offsets_path.has_value());
+    ASSERT_OK_AND_ASSIGN(std::set<std::string> files, ListFiles(table_path));
+    ASSERT_EQ(files.count(offsets_path.value()), 1);
+    for (const auto& path : files) {
+        ASSERT_OK_AND_ASSIGN(bool exists, dir->GetFileSystem()->Exists(path));
+        ASSERT_TRUE(exists) << path;
+    }
+}
+
 TEST(SnapshotFileScanTest, TestInvalidArguments) {
     ASSERT_NOK_WITH_MSG(ListFiles(""), "table path is empty");
     ASSERT_NOK_WITH_MSG(ListFiles("unused", /*snapshot_id=*/0),
diff --git a/src/paimon/format/mosaic/mosaic_file_format_test.cpp b/src/paimon/format/mosaic/mosaic_file_format_test.cpp
index 3646b06ff..ed94feda0 100644
--- a/src/paimon/format/mosaic/mosaic_file_format_test.cpp
+++ b/src/paimon/format/mosaic/mosaic_file_format_test.cpp
@@ -319,7 +319,7 @@ TEST_F(MosaicFileFormatTest, ExtractStatistics) {
     std::map<std::string, std::string> options = {
         {"file.format", "mosaic"},
         {Options::FILE_BLOCK_SIZE, "1 B"},
-        {MOSAIC_STATS_COLUMNS, " id, name, ts, amount, all_null "},
+        {MOSAIC_STATS_COLUMNS, " id, name, ts, amount, all_null, flag, f, d "},
     };
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> configured_format,
                          FileFormatFactory::Get("mosaic", options));
@@ -331,18 +331,24 @@ TEST_F(MosaicFileFormatTest, ExtractStatistics) {
         arrow::field("amount", arrow::decimal128(10, 2)),
         arrow::field("untracked", arrow::int64()),
         arrow::field("all_null", arrow::int32()),
+        arrow::field("flag", arrow::boolean()),
+        arrow::field("f", arrow::float32()),
+        arrow::field("d", arrow::float64()),
     };
     std::shared_ptr<arrow::Schema> schema = arrow::schema(fields);
     std::shared_ptr<arrow::Array> data =
         arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
-          [3,"three","1970-01-01 00:00:00.000000003","3.30",30,null],
-          [1,"one","1970-01-01 00:00:00.000000001","1.10",10,null],
-          [null,null,null,null,null,null],
-          [5,"five","1970-01-01 00:00:00.000000005","5.50",50,null],
-          [2,"","1970-01-01 00:00:00.000000002","2.20",20,null]
+          [3,"three","1970-01-01 00:00:00.000000003","3.30",30,null,true,1.0,1.0],
+          [1,"one","1970-01-01 00:00:00.000000001","1.10",10,null,true,3.0,3.0],
+          [null,null,null,null,null,null,false,-2.0,-2.0],
+          [5,"five","1970-01-01 00:00:00.000000005","5.50",50,null,false,null,null],
+          [2,"","1970-01-01 00:00:00.000000002","2.20",20,null,null,10.0,10.0]
         ])")
             .ValueOrDie();
     ASSERT_OK(WriteFile(path, schema, data, /*batch_size=*/2, configured_format));
+    ASSERT_OK_AND_ASSIGN(FooterLayout footer_layout, ReadFooterLayout(path));
+    ASSERT_EQ(footer_layout.second, 3);
+    AssertReadWithBatchSizes(path, schema, data, {2, 5});
 
     ::ArrowSchema ffi_schema = {};
     ASSERT_TRUE(arrow::ExportSchema(*schema, &ffi_schema).ok());
@@ -357,6 +363,9 @@ TEST_F(MosaicFileFormatTest, ExtractStatistics) {
         "min 1.10, max 5.50, null count 1",
         "min null, max null, null count null",
         "min null, max null, null count 5",
+        "min false, max true, null count 1",
+        "min -2, max 10, null count 1",
+        "min -2, max 10, null count 1",
     };
     ASSERT_EQ(result.first.size(), expected_stats.size());
     for (size_t i = 0; i < expected_stats.size(); ++i) {
diff --git a/src/paimon/format/orc/orc_input_output_stream_test.cpp b/src/paimon/format/orc/orc_input_output_stream_test.cpp
index fcbb13049..4c0f7bf86 100644
--- a/src/paimon/format/orc/orc_input_output_stream_test.cpp
+++ b/src/paimon/format/orc/orc_input_output_stream_test.cpp
@@ -18,6 +18,7 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <future>
 #include <list>
 #include <memory>
 #include <string>
@@ -73,6 +74,16 @@ TEST(OrcInputOutputStreamTest, TestInOutStream) {
     char ret[10];
     in_stream->read(ret, data.length(), /*offset=*/0);
     ASSERT_EQ(data, std::string(ret, data.length()));
+
+    std::string full_data(data.size(), '\0');
+    std::string partial_data(3, '\0');
+    auto full_read = in_stream->readAsync(full_data.data(), full_data.size(), /*offset=*/0);
+    auto partial_read =
+        in_stream->readAsync(partial_data.data(), partial_data.size(), /*offset=*/1);
+    full_read.get();
+    partial_read.get();
+    ASSERT_EQ(full_data, data);
+    ASSERT_EQ(partial_data, data.substr(1, partial_data.size()));
 }
 
 TEST(OrcInputOutputStreamTest, TestSimple) {
diff --git a/test/inte/variant_table_inte_test.cpp b/test/inte/variant_table_inte_test.cpp
index 14f05899e..b45f8f94f 100644
--- a/test/inte/variant_table_inte_test.cpp
+++ b/test/inte/variant_table_inte_test.cpp
@@ -618,6 +618,40 @@ TEST_F(VariantTableInteTest, TestPrimaryKeyTable) {
     ReadAndCheck(helper.get(), splits, {0, 1, 2}, {"{\"a\": 1}", R"({"b": "updated"})", "[42]"});
 }
 
+TEST_F(VariantTableInteTest, TestPrimaryKeyInferenceWithBitmapIndex) {
+    std::map<std::string, std::string> options = {
+        {Options::FILE_FORMAT, "parquet"},
+        {Options::BUCKET, "1"},
+        {Options::WRITE_ONLY, "true"},
+        {Options::VARIANT_INFER_SHREDDING_SCHEMA, "true"},
+        {Options::VARIANT_SHREDDING_MAX_INFER_BUFFER_ROW, "2"},
+        {"file-index.bitmap.columns", "id"},
+        {Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1MB"}};
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TestHelper> helper,
+                         TestHelper::Create(test_dir_, schema_, /*partition_keys=*/{},
+                                            /*primary_keys=*/{"id"}, options,
+                                            /*is_streaming_mode=*/true));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> initial,
+                         MakeBatch(BuildArray({R"({"a": 1})", R"({"a": 2})", nullptr})));
+    ASSERT_OK(helper->WriteAndCommit(std::move(initial), /*commit_identifier=*/0,
+                                     /*expected_commit_messages=*/std::nullopt));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> update,
+                         MakeBatch(BuildArray({R"({"a": 20})"}, /*id_offset=*/1)));
+    ASSERT_OK(helper->WriteAndCommit(std::move(update), /*commit_identifier=*/1,
+                                     /*expected_commit_messages=*/std::nullopt));
+    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits,
+                         helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
+    ASSERT_FALSE(splits.empty());
+    for (const auto& split : splits) {
+        auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+        ASSERT_TRUE(data_split);
+        for (const auto& file : data_split->DataFiles()) {
+            ASSERT_TRUE(file->embedded_index);
+        }
+    }
+    ReadAndCheck(helper.get(), splits, {0, 1, 2}, {R"({"a": 1})", R"({"a": 20})", nullptr});
+}
+
 TEST_F(VariantTableInteTest, TestVariantAccessRead) {
     std::map<std::string, std::string> options = {
         {Options::FILE_FORMAT, "parquet"},

From 225fc5913fb831ba5ae4ea024294f69d352ec2b5 Mon Sep 17 00:00:00 2001
From: lxy264173 <lxy264173@alibaba-inc.com>
Date: Wed, 4 Nov 2026 18:35:18 +0800
Subject: [PATCH 2/3] fix ci

---
 .../file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
index bc8680efb..85e3d6002 100644
--- a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
+++ b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
@@ -56,7 +56,7 @@ namespace {
 class OffsetFileBatchReader : public MockFileBatchReader {
  public:
     OffsetFileBatchReader(const std::shared_ptr<arrow::Array>& data, uint64_t offset)
-        : MockFileBatchReader(data, data->type(), /*batch_size=*/2), offset_(offset) {
+        : MockFileBatchReader(data, data->type(), /*read_batch_size=*/2), offset_(offset) {
         EnableRandomizeBatchSize(false);
     }
 

From ecbfb79647894888e46d7836c5aab4a7688539e3 Mon Sep 17 00:00:00 2001
From: lxy264173 <lxy264173@alibaba-inc.com>
Date: Wed, 4 Nov 2026 20:11:30 +0800
Subject: [PATCH 3/3] fix ci

---
 src/paimon/core/io/row_to_arrow_array_converter_test.cpp | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/paimon/core/io/row_to_arrow_array_converter_test.cpp b/src/paimon/core/io/row_to_arrow_array_converter_test.cpp
index 066d78d14..ca6a246a8 100644
--- a/src/paimon/core/io/row_to_arrow_array_converter_test.cpp
+++ b/src/paimon/core/io/row_to_arrow_array_converter_test.cpp
@@ -51,8 +51,8 @@ TEST(RowToArrowArrayConverterTest, AppendStringWithoutView) {
                                           /*use_view=*/false, &builder, &reserve_count));
 
     const std::string embedded_null("a\0b", 3);
-    auto row =
-        BinaryRowGenerator::GenerateRow({"paimon", "", NullType(), embedded_null}, pool.get());
+    auto row = BinaryRowGenerator::GenerateRow(
+        {std::string("paimon"), std::string(), NullType(), embedded_null}, pool.get());
     for (int32_t i = 0; i < row.GetFieldCount(); ++i) {
         ASSERT_TRUE(append(row, i).ok());
     }
@@ -79,8 +79,8 @@ TEST(RowToArrowArrayConverterTest, AppendBinaryWithoutView) {
                                           /*use_view=*/false, &builder, &reserve_count));
 
     const std::string binary_value("\0\xff\x01", 3);
-    auto row =
-        BinaryRowGenerator::GenerateRow({"paimon", "", NullType(), binary_value}, pool.get());
+    auto row = BinaryRowGenerator::GenerateRow(
+        {std::string("paimon"), std::string(), NullType(), binary_value}, pool.get());
     for (int32_t i = 0; i < row.GetFieldCount(); ++i) {
         ASSERT_TRUE(append(row, i).ok());
     }