diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index ca8f7635b..4eb88dca7 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -464,6 +464,8 @@ if(PAIMON_BUILD_TESTS)
                     common/data/blob_utils_test.cpp
                     common/data/variant/generic_variant_test.cpp
                     common/data/variant/infer_variant_shredding_schema_test.cpp
+                    common/data/variant/variant_access_utils_test.cpp
+                    common/data/variant/variant_shredding_read_plan_factory_test.cpp
                     common/data/variant/variant_shredding_write_plan_factory_test.cpp
                     common/data/variant/variant_get_test.cpp
                     common/data/variant/variant_json_utils_test.cpp
diff --git a/src/paimon/common/compression/block_compression_factory_test.cpp b/src/paimon/common/compression/block_compression_factory_test.cpp
index 225c5206a..f2da459fc 100644
--- a/src/paimon/common/compression/block_compression_factory_test.cpp
+++ b/src/paimon/common/compression/block_compression_factory_test.cpp
@@ -115,4 +115,35 @@ TEST_P(CompressionFactoryTest, TestCompressInsufficientOutputBuffer) {
 INSTANTIATE_TEST_SUITE_P(BlockCompressionTypeGroup, CompressionFactoryTest,
                          ::testing::Values(BlockCompressionType::LZ4, BlockCompressionType::ZSTD));
 
+TEST(CompressionFactoryCreateTest, TestCreateFromCompressOptions) {
+    ASSERT_OK_AND_ASSIGN(auto none_factory,
+                         BlockCompressionFactory::Create(CompressOptions{"none", 0}));
+    ASSERT_EQ(BlockCompressionType::NONE, none_factory->GetCompressionType());
+
+    // Codec name matching is case-insensitive.
+    ASSERT_OK_AND_ASSIGN(auto zstd_factory,
+                         BlockCompressionFactory::Create(CompressOptions{"ZSTD", 1}));
+    ASSERT_EQ(BlockCompressionType::ZSTD, zstd_factory->GetCompressionType());
+
+    ASSERT_OK_AND_ASSIGN(auto lz4_factory,
+                         BlockCompressionFactory::Create(CompressOptions{"Lz4", 0}));
+    ASSERT_EQ(BlockCompressionType::LZ4, lz4_factory->GetCompressionType());
+
+    // Unsupported codec name returns an Invalid status.
+    auto unsupported = BlockCompressionFactory::Create(CompressOptions{"lzo", 0});
+    ASSERT_NOK(unsupported);
+    ASSERT_TRUE(unsupported.status().IsInvalid());
+}
+
+TEST(CompressionFactoryCreateTest, TestCreateFromCompressionType) {
+    ASSERT_OK_AND_ASSIGN(auto none_factory,
+                         BlockCompressionFactory::Create(BlockCompressionType::NONE));
+    ASSERT_EQ(BlockCompressionType::NONE, none_factory->GetCompressionType());
+
+    // LZO is declared but not yet implemented, so it hits the default branch.
+    auto lzo = BlockCompressionFactory::Create(BlockCompressionType::LZO);
+    ASSERT_NOK(lzo);
+    ASSERT_TRUE(lzo.status().IsInvalid());
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/common/data/variant/generic_variant_test.cpp b/src/paimon/common/data/variant/generic_variant_test.cpp
index 84e542a34..82a8d811a 100644
--- a/src/paimon/common/data/variant/generic_variant_test.cpp
+++ b/src/paimon/common/data/variant/generic_variant_test.cpp
@@ -16,8 +16,10 @@
 
 #include "paimon/common/data/variant/generic_variant.h"
 
+#include <functional>
 #include <limits>
 #include <string>
+#include <vector>
 
 #include "gtest/gtest.h"
 #include "paimon/common/data/variant/variant_builder.h"
@@ -345,4 +347,75 @@ TEST_F(GenericVariantTest, NonFiniteDoubleToJson) {
     ASSERT_EQ(json, "\"Infinity\"");
 }
 
+TEST_F(GenericVariantTest, GetTypeInfoReturnsHeaderBits) {
+    // GetTypeInfo exposes the primitive header's type-info bits; 42 is encoded as an int1.
+    auto v = FromJson("42");
+    ASSERT_OK_AND_ASSIGN(int32_t type_info, v->GetTypeInfo());
+    EXPECT_EQ(type_info, VariantDefs::kInt1);
+}
+
+TEST_F(GenericVariantTest, TypedGettersRejectMismatchedTypes) {
+    // Each accessor validates the value header and fails when the stored type differs.
+    auto number = FromJson("42");    // primitive int1
+    auto text = FromJson("\"hi\"");  // short string
+    auto real = FromJson("1.5e0");   // double
+
+    // A primitive long is neither boolean/double/decimal/float/binary/string/uuid, nor a
+    // container.
+    ASSERT_NOK(number->GetBoolean());
+    ASSERT_NOK(number->GetDouble());
+    ASSERT_NOK(number->GetDecimal());
+    ASSERT_NOK(number->GetFloat());
+    ASSERT_NOK(number->GetBinary());
+    ASSERT_NOK(number->GetString());
+    ASSERT_NOK(number->GetUuid());
+    ASSERT_NOK(number->ObjectSize());
+    ASSERT_NOK(number->ArraySize());
+    // A short string is not a primitive, so long/decimal decoding rejects it early.
+    ASSERT_NOK(text->GetLong());
+    ASSERT_NOK(text->GetDecimal());
+    // A double is a primitive but not an integer-like type.
+    ASSERT_NOK(real->GetLong());
+}
+
+TEST_F(GenericVariantTest, ValueSizeCoversAllPrimitiveWidths) {
+    // Builds an array whose elements span the primitive width branches of `ValueSize` (int4,
+    // decimal8, binary, uuid). Copying elements into the array and re-reading each element's
+    // value both exercise `ValueSize`.
+    auto build = [this](const std::function<Status(VariantBuilder&)>& append) {
+        VariantBuilder builder(/*allow_duplicate_keys=*/false);
+        EXPECT_OK(append(builder));
+        auto result = builder.Build(pool_);
+        EXPECT_TRUE(result.ok()) << result.status().ToString();
+        return result.value();
+    };
+    std::shared_ptr<GenericVariant> int4 =
+        build([](VariantBuilder& b) { return b.AppendLong(100000); });  // needs 4 bytes
+    std::shared_ptr<GenericVariant> decimal8 =
+        build([](VariantBuilder& b) { return b.AppendDecimal(VariantDecimal{1234567890, 2}); });
+    std::shared_ptr<GenericVariant> binary =
+        build([](VariantBuilder& b) { return b.AppendBinary(std::string_view("abc", 3)); });
+    std::string uuid_bytes(16, '\x07');
+    std::shared_ptr<GenericVariant> uuid =
+        build([&](VariantBuilder& b) { return b.AppendUuid(uuid_bytes); });
+
+    VariantBuilder array_builder(/*allow_duplicate_keys=*/false);
+    int32_t start = array_builder.GetWritePos();
+    std::vector<int32_t> offsets;
+    for (const std::shared_ptr<GenericVariant>* element : {&int4, &decimal8, &binary, &uuid}) {
+        offsets.push_back(array_builder.GetWritePos() - start);
+        ASSERT_OK(array_builder.AppendVariant(**element));
+    }
+    ASSERT_OK(array_builder.FinishWritingArray(start, offsets));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GenericVariant> array_variant, array_builder.Build(pool_));
+
+    ASSERT_OK_AND_ASSIGN(int32_t size, array_variant->ArraySize());
+    ASSERT_EQ(size, 4);
+    for (int32_t i = 0; i < size; ++i) {
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<GenericVariant> element,
+                             array_variant->GetElementAtIndex(i));
+        ASSERT_OK(element->Value());
+    }
+}
+
 }  // namespace paimon::test
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 427cc543c..4f2dfe548 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
@@ -16,6 +16,7 @@
 
 #include "paimon/common/data/variant/infer_variant_shredding_schema.h"
 
+#include <functional>
 #include <string>
 #include <vector>
 
@@ -52,6 +53,18 @@ class InferVariantShreddingSchemaTest : public ::testing::Test {
         return samples;
     }
 
+    // Builds a single variant using the direct append API, which can encode types (float, binary,
+    // uuid, decimals with a specific scale) that JSON parsing never produces.
+    std::shared_ptr<GenericVariant> BuildVariant(
+        const std::function<Status(VariantBuilder&)>& append) {
+        VariantBuilder builder(/*allow_duplicate_keys=*/false);
+        Status st = append(builder);
+        EXPECT_TRUE(st.ok()) << st.ToString();
+        auto result = builder.Build(pool_);
+        EXPECT_TRUE(result.ok()) << result.status().ToString();
+        return result.value();
+    }
+
  protected:
     std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
     InferVariantShreddingSchema infer_{/*max_schema_width=*/300, /*max_schema_depth=*/50,
@@ -209,4 +222,149 @@ TEST_F(InferVariantShreddingSchemaTest, TemporalValuesStayUnshredded) {
     ASSERT_EQ(ts_inferred, nullptr);
 }
 
+TEST_F(InferVariantShreddingSchemaTest, MergeObjectsWithDisjointFields) {
+    // Objects whose keys interleave exercise both single-side merge branches (field only in the
+    // first object, and field only in the second).
+    auto samples = Samples({R"({"a": 1, "c": 3})", R"({"b": 2, "c": 4})"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred, InferColumn(infer_, samples));
+    ASSERT_NE(inferred, nullptr);
+    auto expected =
+        arrow::struct_({arrow::field("a", arrow::int64()), arrow::field("b", arrow::int64()),
+                        arrow::field("c", arrow::int64())});
+    ASSERT_TRUE(inferred->Equals(*expected)) << inferred->ToString();
+}
+
+TEST_F(InferVariantShreddingSchemaTest, VariantNullSamplesAndFields) {
+    // A top-level variant null (JSON `null`, not a missing sample) merges away, leaving the other
+    // sample's inferred type.
+    auto scalar_and_null = Samples({"1", "null"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> merged,
+                         InferColumn(infer_, scalar_and_null));
+    ASSERT_NE(merged, nullptr);
+    ASSERT_TRUE(merged->Equals(*arrow::int64())) << merged->ToString();
+
+    // An object field that is always variant-null becomes an untyped variant leaf.
+    auto object_with_null = Samples({R"({"a": null, "b": 1})"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
+                         InferColumn(infer_, object_with_null));
+    ASSERT_NE(inferred, nullptr);
+    auto expected =
+        arrow::struct_({arrow::field("a", arrow::null()), arrow::field("b", arrow::int64())});
+    ASSERT_TRUE(inferred->Equals(*expected)) << inferred->ToString();
+}
+
+TEST_F(InferVariantShreddingSchemaTest, ArraysMerge) {
+    // Two arrays merge element-wise into a single typed element schema.
+    auto samples = Samples({"[1, 2]", "[3, 4]"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred, InferColumn(infer_, samples));
+    ASSERT_NE(inferred, nullptr);
+    ASSERT_TRUE(inferred->Equals(*arrow::list(arrow::int64()))) << inferred->ToString();
+}
+
+TEST_F(InferVariantShreddingSchemaTest, ArrayBeyondDepthLimitStaysVariant) {
+    InferVariantShreddingSchema shallow_infer{/*max_schema_width=*/300, /*max_schema_depth=*/1,
+                                              /*min_field_cardinality_ratio=*/0.1};
+    auto samples = Samples({R"({"arr": [1, 2]})"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
+                         InferColumn(shallow_infer, samples));
+    ASSERT_NE(inferred, nullptr);
+    // Depth 1: the nested array is beyond the limit and stays an untyped variant leaf.
+    auto expected = arrow::struct_({arrow::field("arr", arrow::null())});
+    ASSERT_TRUE(inferred->Equals(*expected)) << inferred->ToString();
+}
+
+TEST_F(InferVariantShreddingSchemaTest, ScalarLeafTypes) {
+    // Float and binary leaves shred to their arrow types.
+    std::shared_ptr<GenericVariant> float_variant =
+        BuildVariant([](VariantBuilder& b) { return b.AppendFloat(1.5f); });
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> float_inferred,
+                         InferColumn(infer_, {float_variant}));
+    ASSERT_NE(float_inferred, nullptr);
+    ASSERT_TRUE(float_inferred->Equals(*arrow::float32())) << float_inferred->ToString();
+
+    std::shared_ptr<GenericVariant> binary_variant = BuildVariant(
+        [](VariantBuilder& b) { return b.AppendBinary(std::string_view("\x01\x02", 2)); });
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> binary_inferred,
+                         InferColumn(infer_, {binary_variant}));
+    ASSERT_NE(binary_inferred, nullptr);
+    ASSERT_TRUE(binary_inferred->Equals(*arrow::binary())) << binary_inferred->ToString();
+
+    // A UUID has no shredding type, so the column stays unshredded.
+    std::string uuid_bytes(16, '\0');
+    std::shared_ptr<GenericVariant> uuid_variant =
+        BuildVariant([&](VariantBuilder& b) { return b.AppendUuid(uuid_bytes); });
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> uuid_inferred,
+                         InferColumn(infer_, {uuid_variant}));
+    ASSERT_EQ(uuid_inferred, nullptr);
+}
+
+TEST_F(InferVariantShreddingSchemaTest, LargeIntegerAndDecimalMerging) {
+    // A 19-digit long exceeds decimal(18) precision, so it stays a genuine int64 leaf.
+    auto big_long = Samples({"1000000000000000000"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> long_inferred,
+                         InferColumn(infer_, big_long));
+    ASSERT_NE(long_inferred, nullptr);
+    ASSERT_TRUE(long_inferred->Equals(*arrow::int64())) << long_inferred->ToString();
+
+    // A long (int64) merged with a fractional decimal widens via MergeDecimalWithLong.
+    auto long_then_decimal = Samples({"1000000000000000000", "1.5"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> ld,
+                         InferColumn(infer_, long_then_decimal));
+    ASSERT_NE(ld, nullptr);
+    ASSERT_TRUE(ld->Equals(*arrow::decimal128(38, 1))) << ld->ToString();
+
+    // The reversed order (decimal first, then long) hits the mirrored branch.
+    auto decimal_then_long = Samples({"1.5", "1000000000000000000"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> dl,
+                         InferColumn(infer_, decimal_then_long));
+    ASSERT_NE(dl, nullptr);
+    ASSERT_TRUE(dl->Equals(*arrow::decimal128(38, 1))) << dl->ToString();
+}
+
+TEST_F(InferVariantShreddingSchemaTest, LongMergedWithIntegralDecimalStaysLong) {
+    // A scale-0 decimal that fits in 18 digits merges with a long back to int64.
+    std::shared_ptr<GenericVariant> integral_decimal =
+        BuildVariant([](VariantBuilder& b) { return b.AppendDecimal(VariantDecimal{123, 0}); });
+    std::shared_ptr<GenericVariant> big_long = Samples({"1000000000000000000"})[0];
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
+                         InferColumn(infer_, {integral_decimal, big_long}));
+    ASSERT_NE(inferred, nullptr);
+    ASSERT_TRUE(inferred->Equals(*arrow::int64())) << inferred->ToString();
+}
+
+TEST_F(InferVariantShreddingSchemaTest, SmallFractionalDecimalPrecisionAdjusted) {
+    // 0.0015 has more fractional digits than significant digits; precision widens up to the scale.
+    auto samples = Samples({"0.0015"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred, InferColumn(infer_, samples));
+    ASSERT_NE(inferred, nullptr);
+    ASSERT_TRUE(inferred->Equals(*arrow::decimal128(18, 4))) << inferred->ToString();
+}
+
+TEST_F(InferVariantShreddingSchemaTest, DecimalMergeOverflowFallsToVariant) {
+    // A 38-digit integral decimal merged with a high-scale decimal would need precision > 38,
+    // which decimal cannot represent, so the column stays unshredded.
+    __int128 wide_unscaled = 0;
+    for (int i = 0; i < 38; ++i) {
+        wide_unscaled = wide_unscaled * 10 + 1;  // 38 ones, no trailing zeros
+    }
+    std::shared_ptr<GenericVariant> wide_decimal = BuildVariant(
+        [&](VariantBuilder& b) { return b.AppendDecimal(VariantDecimal{wide_unscaled, 0}); });
+    std::shared_ptr<GenericVariant> high_scale_decimal =
+        BuildVariant([](VariantBuilder& b) { return b.AppendDecimal(VariantDecimal{15, 20}); });
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
+                         InferColumn(infer_, {wide_decimal, high_scale_decimal}));
+    ASSERT_EQ(inferred, nullptr);
+}
+
+TEST_F(InferVariantShreddingSchemaTest, ObjectWithAllRareFieldsStaysUnshredded) {
+    InferVariantShreddingSchema strict_infer{/*max_schema_width=*/300, /*max_schema_depth=*/50,
+                                             /*min_field_cardinality_ratio=*/0.6};
+    // Two objects with disjoint single-occurrence keys: with a 0.6 ratio every field is below the
+    // cardinality threshold, so the object contributes no typed field and the column is dropped.
+    auto samples = Samples({R"({"a": 1})", R"({"b": 2})"});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> inferred,
+                         InferColumn(strict_infer, samples));
+    ASSERT_EQ(inferred, nullptr);
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/common/data/variant/variant_access_utils_test.cpp b/src/paimon/common/data/variant/variant_access_utils_test.cpp
new file mode 100644
index 000000000..6dc1b3545
--- /dev/null
+++ b/src/paimon/common/data/variant/variant_access_utils_test.cpp
@@ -0,0 +1,161 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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/data/variant/variant_access_utils.h"
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/util/key_value_metadata.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/variant/variant_defs.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+namespace {
+
+// An Arrow field carrying a `paimon.description` metadata entry.
+std::shared_ptr<arrow::Field> DescribedField(const std::string& name,
+                                             const std::shared_ptr<arrow::DataType>& type,
+                                             const std::string& description) {
+    return arrow::field(name, type, /*nullable=*/true,
+                        arrow::key_value_metadata({DataField::DESCRIPTION}, {description}));
+}
+
+std::shared_ptr<arrow::Field> AccessProjection(
+    const std::vector<std::shared_ptr<arrow::Field>>& children) {
+    return arrow::field("v", arrow::struct_(children));
+}
+
+// A shredded file field: struct{[metadata], value, typed_value: struct{<typed_children>}}.
+std::shared_ptr<arrow::Field> ShreddedFile(
+    const std::vector<std::shared_ptr<arrow::Field>>& typed_children, bool with_metadata = true) {
+    arrow::FieldVector fields;
+    if (with_metadata) {
+        fields.push_back(
+            arrow::field(std::string(VariantDefs::kMetadataFieldName), arrow::binary()));
+    }
+    fields.push_back(arrow::field(std::string(VariantDefs::kValueFieldName), arrow::binary()));
+    fields.push_back(arrow::field(std::string(VariantDefs::kTypedValueFieldName),
+                                  arrow::struct_(typed_children)));
+    return arrow::field("v", arrow::struct_(fields));
+}
+
+std::vector<VariantAccessSpec> ObjectKeySpecs() {
+    auto proj = AccessProjection({DescribedField(
+        "a", arrow::int32(), VariantAccessUtils::BuildVariantMetadata("$.a", false, "UTC"))});
+    auto result = VariantAccessUtils::ParseAccessSpecs(proj);
+    EXPECT_TRUE(result.ok()) << result.status().ToString();
+    return result.value();
+}
+
+}  // namespace
+
+TEST(VariantAccessUtilsTest, ParseAccessSpecsRejectsNonProjection) {
+    // A plain struct without access descriptions is not a variant-access projection.
+    auto plain = arrow::field("v", arrow::struct_({arrow::field("x", arrow::int32())}));
+    ASSERT_FALSE(VariantAccessUtils::IsVariantAccessType(plain->type()));
+    ASSERT_NOK(VariantAccessUtils::ParseAccessSpecs(plain));
+}
+
+TEST(VariantAccessUtilsTest, ParseAccessSpecsRejectsMalformedDescription) {
+    const std::string key = VariantAccessUtils::kMetadataKey;
+    // A description with no delimiter splits into a single part.
+    auto no_delim = AccessProjection({DescribedField("a", arrow::utf8(), key + "$.a")});
+    ASSERT_TRUE(VariantAccessUtils::IsVariantAccessType(no_delim->type()));
+    ASSERT_NOK(VariantAccessUtils::ParseAccessSpecs(no_delim));
+    // A single delimiter splits into two parts (still not the three that a spec needs).
+    auto one_delim = AccessProjection({DescribedField("a", arrow::utf8(), key + "$.a;true")});
+    ASSERT_TRUE(VariantAccessUtils::IsVariantAccessType(one_delim->type()));
+    ASSERT_NOK(VariantAccessUtils::ParseAccessSpecs(one_delim));
+}
+
+TEST(VariantAccessUtilsTest, ParseAccessSpecsRoundTripsBuildMetadata) {
+    auto proj = AccessProjection({DescribedField(
+        "a", arrow::int32(), VariantAccessUtils::BuildVariantMetadata("$.a", true, "+08:00"))});
+    ASSERT_OK_AND_ASSIGN(std::vector<VariantAccessSpec> specs,
+                         VariantAccessUtils::ParseAccessSpecs(proj));
+    ASSERT_EQ(specs.size(), 1);
+    EXPECT_EQ(specs[0].path, "$.a");
+    EXPECT_TRUE(specs[0].cast_args.fail_on_error);
+    EXPECT_EQ(specs[0].cast_args.zone_id, "+08:00");
+}
+
+TEST(VariantAccessUtilsTest, ClipRejectsNonStructFileField) {
+    auto non_struct = arrow::field("v", arrow::int32());
+    ASSERT_NOK(VariantAccessUtils::ClipShreddedFileField(ObjectKeySpecs(), non_struct));
+}
+
+TEST(VariantAccessUtilsTest, ClipUnprunablePathsReturnFileUnchanged) {
+    auto file = ShreddedFile({arrow::field("a", arrow::int32())});
+
+    // A root path needs the whole variant, so the file field is returned unchanged.
+    auto root_proj = AccessProjection({DescribedField(
+        "r", arrow::utf8(), VariantAccessUtils::BuildVariantMetadata("$", false, "UTC"))});
+    ASSERT_OK_AND_ASSIGN(std::vector<VariantAccessSpec> root_specs,
+                         VariantAccessUtils::ParseAccessSpecs(root_proj));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Field> root_clipped,
+                         VariantAccessUtils::ClipShreddedFileField(root_specs, file));
+    EXPECT_EQ(root_clipped, file);
+
+    // An array-first path likewise cannot be pruned to a top-level key.
+    auto array_proj = AccessProjection({DescribedField(
+        "e", arrow::utf8(), VariantAccessUtils::BuildVariantMetadata("$[0]", false, "UTC"))});
+    ASSERT_OK_AND_ASSIGN(std::vector<VariantAccessSpec> array_specs,
+                         VariantAccessUtils::ParseAccessSpecs(array_proj));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Field> array_clipped,
+                         VariantAccessUtils::ClipShreddedFileField(array_specs, file));
+    EXPECT_EQ(array_clipped, file);
+}
+
+TEST(VariantAccessUtilsTest, ClipUnshreddedFileFieldReturnedUnchanged) {
+    // No typed_value column: there is nothing to prune.
+    auto unshredded = arrow::field(
+        "v",
+        arrow::struct_({arrow::field(std::string(VariantDefs::kMetadataFieldName), arrow::binary()),
+                        arrow::field(std::string(VariantDefs::kValueFieldName), arrow::binary())}));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Field> clipped,
+                         VariantAccessUtils::ClipShreddedFileField(ObjectKeySpecs(), unshredded));
+    EXPECT_EQ(clipped, unshredded);
+}
+
+TEST(VariantAccessUtilsTest, ClipMissingMetadataFails) {
+    auto file_no_metadata =
+        ShreddedFile({arrow::field("a", arrow::int32())}, /*with_metadata=*/false);
+    ASSERT_NOK(VariantAccessUtils::ClipShreddedFileField(ObjectKeySpecs(), file_no_metadata));
+}
+
+TEST(VariantAccessUtilsTest, ClipNarrowsToRequestedKeys) {
+    // "$.a" is requested; the shredded typed_value keeps only "a" and drops the unrelated "b".
+    auto file = ShreddedFile({arrow::field("a", arrow::int32()), arrow::field("b", arrow::utf8())});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Field> clipped,
+                         VariantAccessUtils::ClipShreddedFileField(ObjectKeySpecs(), file));
+    ASSERT_EQ(clipped->type()->id(), arrow::Type::STRUCT);
+    const auto& clipped_struct = static_cast<const arrow::StructType&>(*clipped->type());
+    // metadata is always kept; value is dropped because "a" is shredded; typed_value keeps "a".
+    ASSERT_NE(clipped_struct.GetFieldByName(VariantDefs::kMetadataFieldName), nullptr);
+    EXPECT_EQ(clipped_struct.GetFieldByName(VariantDefs::kValueFieldName), nullptr);
+    auto typed = clipped_struct.GetFieldByName(VariantDefs::kTypedValueFieldName);
+    ASSERT_NE(typed, nullptr);
+    ASSERT_EQ(typed->type()->num_fields(), 1);
+    EXPECT_EQ(typed->type()->field(0)->name(), "a");
+}
+
+}  // 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 c8981160b..a31fa7b1d 100644
--- a/src/paimon/common/data/variant/variant_get_test.cpp
+++ b/src/paimon/common/data/variant/variant_get_test.cpp
@@ -16,6 +16,7 @@
 
 #include "paimon/common/data/variant/variant_get.h"
 
+#include <functional>
 #include <string>
 
 #include "arrow/api.h"
@@ -76,6 +77,18 @@ class VariantGetTest : public ::testing::Test {
                                               arrow_pool_);
     }
 
+    // Builds a single-scalar variant using the direct append API, which can encode types that
+    // JSON parsing never produces (uuid/date/float/binary/timestamp).
+    std::shared_ptr<GenericVariant> BuildScalar(
+        const std::function<Status(VariantBuilder&)>& append) {
+        VariantBuilder builder(/*allow_duplicate_keys=*/false);
+        Status st = append(builder);
+        EXPECT_TRUE(st.ok()) << st.ToString();
+        auto result = builder.Build(pool_);
+        EXPECT_TRUE(result.ok()) << result.status().ToString();
+        return result.value();
+    }
+
  protected:
     std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
     std::shared_ptr<arrow::MemoryPool> arrow_pool_ = GetArrowPool(pool_);
@@ -332,4 +345,121 @@ TEST_F(VariantGetTest, NestedTargetNullSemantics) {
     ASSERT_TRUE(array->IsNull(0));
 }
 
+TEST_F(VariantGetTest, UuidSourceCastsToStringOnly) {
+    std::string uuid_bytes(16, '\0');
+    for (int i = 0; i < 16; ++i) {
+        uuid_bytes[i] = static_cast<char>(i);
+    }
+    std::shared_ptr<GenericVariant> variant =
+        BuildScalar([&](VariantBuilder& b) { return b.AppendUuid(uuid_bytes); });
+    // A UUID has no Paimon type, so it can only be rendered as its canonical string.
+    ASSERT_OK_AND_ASSIGN(std::optional<Literal> as_string,
+                         VariantGetExecutor::Get(variant, "$", arrow::utf8(), cast_args_));
+    ASSERT_TRUE(as_string.has_value());
+    ASSERT_EQ(as_string->GetValue<std::string>(), "00010203-0405-0607-0809-0a0b0c0d0e0f");
+    // Any non-string target is an invalid cast, which is SQL NULL when fail_on_error is false.
+    ASSERT_OK_AND_ASSIGN(std::optional<Literal> as_long,
+                         VariantGetExecutor::Get(variant, "$", arrow::int64(), cast_args_));
+    ASSERT_FALSE(as_long.has_value());
+    cast_args_.fail_on_error = true;
+    ASSERT_NOK(VariantGetExecutor::Get(variant, "$", arrow::int64(), cast_args_));
+}
+
+TEST_F(VariantGetTest, FloatingPointToStringMatchesJava) {
+    // Doubles and floats are stringified via the Java formatting, not the arrow cast.
+    ASSERT_EQ(GetString("$.double", arrow::utf8()), "1.0123456789012346");
+    std::shared_ptr<GenericVariant> f =
+        BuildScalar([](VariantBuilder& b) { return b.AppendFloat(1.5f); });
+    ASSERT_OK_AND_ASSIGN(std::optional<Literal> as_string,
+                         VariantGetExecutor::Get(f, "$", arrow::utf8(), cast_args_));
+    ASSERT_TRUE(as_string.has_value());
+    ASSERT_EQ(as_string->GetValue<std::string>(), "1.5");
+    ASSERT_OK_AND_ASSIGN(std::optional<Literal> as_float,
+                         VariantGetExecutor::Get(f, "$", arrow::float32(), cast_args_));
+    ASSERT_TRUE(as_float.has_value());
+    ASSERT_EQ(as_float->GetValue<float>(), 1.5f);
+}
+
+TEST_F(VariantGetTest, DateSource) {
+    std::shared_ptr<GenericVariant> d =
+        BuildScalar([](VariantBuilder& b) { return b.AppendDate(19000); });
+    ASSERT_OK_AND_ASSIGN(std::optional<Literal> as_date,
+                         VariantGetExecutor::Get(d, "$", arrow::date32(), cast_args_));
+    ASSERT_TRUE(as_date.has_value());
+    ASSERT_EQ(as_date->GetValue<int32_t>(), 19000);
+}
+
+TEST_F(VariantGetTest, MissingCastExecutorYieldsNull) {
+    // No binary->int64 cast executor exists, so the cast is treated as invalid (SQL NULL).
+    std::shared_ptr<GenericVariant> bin = BuildScalar(
+        [](VariantBuilder& b) { return b.AppendBinary(std::string_view("\x01\x02", 2)); });
+    ASSERT_OK_AND_ASSIGN(std::optional<Literal> as_long,
+                         VariantGetExecutor::Get(bin, "$", arrow::int64(), cast_args_));
+    ASSERT_FALSE(as_long.has_value());
+}
+
+TEST_F(VariantGetTest, ScalarArrowBuilders) {
+    auto build_array = [&](const std::shared_ptr<GenericVariant>& variant, const std::string& path,
+                           const std::shared_ptr<arrow::DataType>& type) {
+        auto result = VariantGetExecutor::GetAsArrow(variant, path, arrow::field("x", type),
+                                                     cast_args_, pool_, arrow_pool_);
+        EXPECT_TRUE(result.ok()) << result.status().ToString();
+        return result.value();
+    };
+
+    std::shared_ptr<arrow::Array> boolean = build_array(variant_, "$.boolean1", arrow::boolean());
+    ASSERT_TRUE(static_cast<const arrow::BooleanArray&>(*boolean).Value(0));
+
+    // INT8/INT16/INT32 each require a narrowing cast from the int64 source.
+    std::shared_ptr<arrow::Array> int8 = build_array(variant_, "$.object.age", arrow::int8());
+    ASSERT_EQ(static_cast<const arrow::Int8Array&>(*int8).Value(0), 2);
+    std::shared_ptr<arrow::Array> int16 = build_array(variant_, "$.object.age", arrow::int16());
+    ASSERT_EQ(static_cast<const arrow::Int16Array&>(*int16).Value(0), 2);
+    std::shared_ptr<arrow::Array> int32 = build_array(variant_, "$.object.age", arrow::int32());
+    ASSERT_EQ(static_cast<const arrow::Int32Array&>(*int32).Value(0), 2);
+
+    std::shared_ptr<arrow::Array> float64 = build_array(variant_, "$.double", arrow::float64());
+    ASSERT_DOUBLE_EQ(static_cast<const arrow::DoubleArray&>(*float64).Value(0), 1.0123456789012346);
+
+    std::shared_ptr<GenericVariant> f =
+        BuildScalar([](VariantBuilder& b) { return b.AppendFloat(1.5f); });
+    std::shared_ptr<arrow::Array> float32 = build_array(f, "$", arrow::float32());
+    ASSERT_EQ(static_cast<const arrow::FloatArray&>(*float32).Value(0), 1.5f);
+
+    std::shared_ptr<arrow::Array> decimal =
+        build_array(variant_, "$.decimal", arrow::decimal128(5, 2));
+    ASSERT_EQ(static_cast<const arrow::Decimal128Array&>(*decimal).FormatValue(0), "100.99");
+
+    std::shared_ptr<GenericVariant> bin = BuildScalar(
+        [](VariantBuilder& b) { return b.AppendBinary(std::string_view("\x01\x02\x03", 3)); });
+    std::shared_ptr<arrow::Array> binary = build_array(bin, "$", arrow::binary());
+    ASSERT_EQ(static_cast<const arrow::BinaryArray&>(*binary).GetView(0),
+              std::string_view("\x01\x02\x03", 3));
+
+    std::shared_ptr<GenericVariant> d =
+        BuildScalar([](VariantBuilder& b) { return b.AppendDate(19000); });
+    std::shared_ptr<arrow::Array> date = build_array(d, "$", arrow::date32());
+    ASSERT_EQ(static_cast<const arrow::Date32Array&>(*date).Value(0), 19000);
+
+    // Only the microsecond unit matches the variant's internal timestamp representation; other
+    // units require a timestamp->timestamp literal cast that is intentionally unsupported.
+    std::shared_ptr<GenericVariant> ts =
+        BuildScalar([](VariantBuilder& b) { return b.AppendTimestamp(1700000000123456); });
+    std::shared_ptr<arrow::Array> timestamp =
+        build_array(ts, "$", arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"));
+    ASSERT_EQ(static_cast<const arrow::TimestampArray&>(*timestamp).Value(0), 1700000000123456);
+}
+
+TEST_F(VariantGetTest, ListFromNonArrayIsNull) {
+    auto target = arrow::field("l", arrow::list(arrow::int64()));
+    // An object cannot cast to a list: SQL NULL when fail_on_error is false ...
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> array, GetAsArrow("$.object", target));
+    ASSERT_TRUE(array->IsNull(0));
+    // ... and an error when fail_on_error is true.
+    cast_args_.fail_on_error = true;
+    auto result = VariantGetExecutor::GetAsArrow(variant_, "$.object", target, cast_args_, pool_,
+                                                 arrow_pool_);
+    ASSERT_NOK(result);
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/common/data/variant/variant_json_utils_test.cpp b/src/paimon/common/data/variant/variant_json_utils_test.cpp
index 260203a90..58a9ae119 100644
--- a/src/paimon/common/data/variant/variant_json_utils_test.cpp
+++ b/src/paimon/common/data/variant/variant_json_utils_test.cpp
@@ -18,8 +18,10 @@
 
 #include <cmath>
 #include <limits>
+#include <string>
 
 #include "gtest/gtest.h"
+#include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
 
@@ -69,4 +71,48 @@ TEST(VariantJsonUtilsTest, JavaFloatToString) {
     EXPECT_EQ(VariantJsonUtils::JavaFloatToString(2.5F), "2.5");
 }
 
+TEST(VariantJsonUtilsTest, AppendEscapedJsonControlChars) {
+    // Control characters without a named escape are emitted as \uXXXX.
+    std::string out;
+    VariantJsonUtils::AppendEscapedJson(std::string_view("\x01\x1f", 2), &out);
+    EXPECT_EQ(out, "\"\\u0001\\u001f\"");
+}
+
+TEST(VariantJsonUtilsTest, DateToStringNegativeYear) {
+    // A date far before the epoch renders a negative (BCE) year with a leading '-'.
+    std::string result = VariantJsonUtils::DateToString(-800000);
+    ASSERT_FALSE(result.empty());
+    EXPECT_EQ(result.front(), '-');
+}
+
+TEST(VariantJsonUtilsTest, TimestampToStringEdgeCases) {
+    // Sub-second fraction trailing zeros are trimmed.
+    EXPECT_EQ(VariantJsonUtils::TimestampToString(500000, 0, /*with_offset=*/false),
+              "1970-01-01 00:00:00.5");
+    // Negative epoch micros floor to the previous day (FloorDiv keeps a non-negative remainder).
+    EXPECT_EQ(VariantJsonUtils::TimestampToString(-1, 0, /*with_offset=*/false),
+              "1969-12-31 23:59:59.999999");
+    // A positive zone offset is appended as +HH:MM.
+    EXPECT_EQ(VariantJsonUtils::TimestampToString(0, 8 * 3600, /*with_offset=*/true),
+              "1970-01-01 08:00:00+08:00");
+}
+
+TEST(VariantJsonUtilsTest, ZoneOffsetParsing) {
+    // `UTC`/`GMT`/`UT` prefixes are stripped before the fixed offset is parsed.
+    ASSERT_OK_AND_ASSIGN(int32_t utc_prefixed,
+                         VariantJsonUtils::GetZoneOffsetSeconds("UTC+08:00", 0));
+    EXPECT_EQ(utc_prefixed, 8 * 3600);
+    ASSERT_OK_AND_ASSIGN(int32_t ut_prefixed, VariantJsonUtils::GetZoneOffsetSeconds("UT+05", 0));
+    EXPECT_EQ(ut_prefixed, 5 * 3600);
+    // HHMMSS form and a negative offset.
+    ASSERT_OK_AND_ASSIGN(int32_t hms, VariantJsonUtils::GetZoneOffsetSeconds("+18:30:15", 0));
+    EXPECT_EQ(hms, 18 * 3600 + 30 * 60 + 15);
+    ASSERT_OK_AND_ASSIGN(int32_t neg, VariantJsonUtils::GetZoneOffsetSeconds("-06:30", 0));
+    EXPECT_EQ(neg, -(6 * 3600 + 30 * 60));
+    // Invalid forms are rejected: a non-digit char, a wrong digit count, and an out-of-range hour.
+    ASSERT_NOK(VariantJsonUtils::GetZoneOffsetSeconds("+9A", 0));
+    ASSERT_NOK(VariantJsonUtils::GetZoneOffsetSeconds("+123", 0));
+    ASSERT_NOK(VariantJsonUtils::GetZoneOffsetSeconds("+19:00", 0));
+}
+
 }  // namespace paimon::test
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
new file mode 100644
index 000000000..7ab700e3a
--- /dev/null
+++ b/src/paimon/common/data/variant/variant_shredding_read_plan_factory_test.cpp
@@ -0,0 +1,274 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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/data/variant/variant_shredding_read_plan_factory.h"
+
+#include <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/util/key_value_metadata.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/variant/generic_variant.h"
+#include "paimon/common/data/variant/variant_access_utils.h"
+#include "paimon/common/data/variant/variant_defs.h"
+#include "paimon/common/data/variant/variant_shredding_utils.h"
+#include "paimon/common/data/variant/variant_shredding_writer.h"
+#include "paimon/common/data/variant/variant_type_utils.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class VariantShreddingReadPlanFactoryTest : public ::testing::Test {
+ public:
+    std::shared_ptr<GenericVariant> Variant(const std::string& json) {
+        auto result = GenericVariant::FromJson(json, pool_);
+        EXPECT_TRUE(result.ok()) << result.status().ToString();
+        return result.value();
+    }
+
+    // The full shredded StructArray (struct{metadata, value, typed_value}) for one variant, using
+    // the production writer.
+    void MakeFullShredded(const std::shared_ptr<arrow::DataType>& logical,
+                          const std::shared_ptr<GenericVariant>& variant,
+                          std::shared_ptr<arrow::DataType>* physical_type,
+                          std::shared_ptr<arrow::StructArray>* array) {
+        ASSERT_OK_AND_ASSIGN(*physical_type,
+                             VariantShreddingUtils::VariantShreddingSchema(logical));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<VariantSchema> schema,
+                             VariantShreddingUtils::BuildVariantSchema(*physical_type));
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<VariantShreddedColumnWriter> writer,
+                             VariantShreddedColumnWriter::Create(schema, *physical_type,
+                                                                 arrow::default_memory_pool()));
+        ASSERT_OK(writer->Append(*variant));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> arr, writer->Finish());
+        *array = std::static_pointer_cast<arrow::StructArray>(arr);
+    }
+
+    // The unshredded physical StructArray (struct{value, metadata}) for one variant.
+    std::shared_ptr<arrow::StructArray> MakeUnshredded(const std::shared_ptr<GenericVariant>& v) {
+        auto value = v->Value();
+        EXPECT_TRUE(value.ok()) << value.status().ToString();
+        return MakeUnshreddedRaw(value.value(), v->Metadata(), /*metadata_null=*/false);
+    }
+
+    std::shared_ptr<arrow::StructArray> MakeUnshreddedRaw(std::string_view value,
+                                                          std::string_view metadata,
+                                                          bool metadata_null) {
+        arrow::BinaryBuilder value_builder;
+        EXPECT_TRUE(value_builder.Append(value).ok());
+        std::shared_ptr<arrow::Array> value_array;
+        EXPECT_TRUE(value_builder.Finish(&value_array).ok());
+        arrow::BinaryBuilder metadata_builder;
+        if (metadata_null) {
+            EXPECT_TRUE(metadata_builder.AppendNull().ok());
+        } else {
+            EXPECT_TRUE(metadata_builder.Append(metadata).ok());
+        }
+        std::shared_ptr<arrow::Array> metadata_array;
+        EXPECT_TRUE(metadata_builder.Finish(&metadata_array).ok());
+        auto made = arrow::StructArray::Make({value_array, metadata_array},
+                                             std::vector<std::string>{"value", "metadata"});
+        EXPECT_TRUE(made.ok()) << made.status().ToString();
+        return made.ValueOrDie();
+    }
+
+    // A variant-access child: an arrow field carrying a `__VARIANT_METADATA` description.
+    static std::shared_ptr<arrow::Field> AccessChild(const std::string& name,
+                                                     const std::shared_ptr<arrow::DataType>& type,
+                                                     const std::string& path) {
+        return arrow::field(name, type, /*nullable=*/true,
+                            arrow::key_value_metadata({DataField::DESCRIPTION},
+                                                      {VariantAccessUtils::BuildVariantMetadata(
+                                                          path, /*fail_on_error=*/false, "UTC")}));
+    }
+
+    std::shared_ptr<ShreddingColumnReadPlan> CreatePlan(
+        const std::shared_ptr<arrow::Field>& read_field,
+        const std::shared_ptr<arrow::Field>& file_field) {
+        auto plans = VariantShreddingReadPlanFactory::CreateReadPlans(
+            arrow::schema({read_field}), arrow::schema({file_field}), pool_);
+        EXPECT_TRUE(plans.ok()) << plans.status().ToString();
+        auto it = plans.value().find(read_field->name());
+        EXPECT_NE(it, plans.value().end());
+        return it->second;
+    }
+
+    static std::shared_ptr<arrow::Array> Int32Array(int32_t value) {
+        arrow::Int32Builder builder;
+        EXPECT_TRUE(builder.Append(value).ok());
+        std::shared_ptr<arrow::Array> array;
+        EXPECT_TRUE(builder.Finish(&array).ok());
+        return array;
+    }
+
+ protected:
+    std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
+};
+
+TEST_F(VariantShreddingReadPlanFactoryTest, FullVariantReadOfShreddedFile) {
+    std::shared_ptr<GenericVariant> variant = Variant(R"({"a": 5})");
+    std::shared_ptr<arrow::DataType> physical;
+    std::shared_ptr<arrow::StructArray> shredded;
+    MakeFullShredded(arrow::struct_({arrow::field("a", arrow::int64())}), variant, &physical,
+                     &shredded);
+    ASSERT_FALSE(HasFatalFailure());
+
+    auto read_field = VariantTypeUtils::ToArrowField("v");
+    auto file_field = arrow::field("v", physical);
+    std::shared_ptr<ShreddingColumnReadPlan> plan = CreatePlan(read_field, file_field);
+    ASSERT_NE(plan, nullptr);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> assembled,
+                         plan->Assemble(shredded, arrow::default_memory_pool()));
+    auto assembled_struct = std::static_pointer_cast<arrow::StructArray>(assembled);
+    auto value_column = std::static_pointer_cast<arrow::BinaryArray>(assembled_struct->field(0));
+    auto metadata_column = std::static_pointer_cast<arrow::BinaryArray>(assembled_struct->field(1));
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<GenericVariant> rebuilt,
+        GenericVariant::Create(value_column->GetView(0), metadata_column->GetView(0), pool_));
+    ASSERT_OK_AND_ASSIGN(std::string json, rebuilt->ToJson());
+    EXPECT_EQ(json, R"({"a":5})");
+
+    // Assembling a non-struct physical array is an error.
+    auto ints = Int32Array(1);
+    ASSERT_NOK(plan->Assemble(ints, arrow::default_memory_pool()));
+}
+
+TEST_F(VariantShreddingReadPlanFactoryTest, AccessProjectionOnUnshreddedFile) {
+    std::shared_ptr<GenericVariant> variant = Variant(R"({"a": 5, "b": [10, 20], "c": "hi"})");
+    std::shared_ptr<arrow::StructArray> unshredded = MakeUnshredded(variant);
+    ASSERT_FALSE(HasFatalFailure());
+
+    auto read_field =
+        arrow::field("v", arrow::struct_({AccessChild("a", arrow::int64(), "$.a"),
+                                          AccessChild("c", arrow::utf8(), "$.c"),
+                                          AccessChild("b0", arrow::int64(), "$.b[0]"),
+                                          AccessChild("missing", arrow::utf8(), "$.missing")}));
+    // Unshredded file column: struct{value, metadata}.
+    auto file_field = arrow::field("v", VariantTypeUtils::UnshreddedStructType());
+    std::shared_ptr<ShreddingColumnReadPlan> plan = CreatePlan(read_field, file_field);
+    ASSERT_NE(plan, nullptr);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> assembled,
+                         plan->Assemble(unshredded, arrow::default_memory_pool()));
+    auto row = std::static_pointer_cast<arrow::StructArray>(assembled);
+    ASSERT_EQ(row->length(), 1);
+    EXPECT_EQ(static_cast<const arrow::Int64Array&>(*row->field(0)).Value(0), 5);
+    EXPECT_EQ(static_cast<const arrow::StringArray&>(*row->field(1)).GetString(0), "hi");
+    EXPECT_EQ(static_cast<const arrow::Int64Array&>(*row->field(2)).Value(0), 10);
+    EXPECT_TRUE(row->field(3)->IsNull(0));
+
+    // A non-struct physical array is an error.
+    auto ints = Int32Array(1);
+    ASSERT_NOK(plan->Assemble(ints, arrow::default_memory_pool()));
+}
+
+TEST_F(VariantShreddingReadPlanFactoryTest, AccessProjectionRejectsNullMetadata) {
+    std::shared_ptr<GenericVariant> variant = Variant(R"({"a": 5})");
+    ASSERT_OK_AND_ASSIGN(std::string_view value, variant->Value());
+    std::shared_ptr<arrow::StructArray> bad =
+        MakeUnshreddedRaw(value, variant->Metadata(), /*metadata_null=*/true);
+    ASSERT_FALSE(HasFatalFailure());
+
+    auto read_field = arrow::field("v", arrow::struct_({AccessChild("a", arrow::int64(), "$.a")}));
+    auto file_field = arrow::field("v", VariantTypeUtils::UnshreddedStructType());
+    std::shared_ptr<ShreddingColumnReadPlan> plan = CreatePlan(read_field, file_field);
+    ASSERT_NE(plan, nullptr);
+    ASSERT_NOK(plan->Assemble(bad, arrow::default_memory_pool()));
+}
+
+TEST_F(VariantShreddingReadPlanFactoryTest, AccessProjectionOnShreddedFile) {
+    // Extracting shredded object keys reads the typed sub-columns directly (with pruning).
+    std::shared_ptr<GenericVariant> variant = Variant(R"({"a": 5, "b": "hi"})");
+    std::shared_ptr<arrow::DataType> physical;
+    std::shared_ptr<arrow::StructArray> shredded;
+    MakeFullShredded(
+        arrow::struct_({arrow::field("a", arrow::int64()), arrow::field("b", arrow::utf8())}),
+        variant, &physical, &shredded);
+    ASSERT_FALSE(HasFatalFailure());
+
+    auto read_field = arrow::field("v", arrow::struct_({AccessChild("a", arrow::int64(), "$.a"),
+                                                        AccessChild("b", arrow::utf8(), "$.b")}));
+    auto file_field = arrow::field("v", physical);
+    std::shared_ptr<ShreddingColumnReadPlan> plan = CreatePlan(read_field, file_field);
+    ASSERT_NE(plan, nullptr);
+
+    // With both keys shredded and requested, pruning keeps {metadata, typed_value} and drops the
+    // top-level `value` column; project the full array down to match the pushed-down physical.
+    auto made = arrow::StructArray::Make({shredded->field(0), shredded->field(2)},
+                                         {std::string(VariantDefs::kMetadataFieldName),
+                                          std::string(VariantDefs::kTypedValueFieldName)});
+    ASSERT_TRUE(made.ok()) << made.status().ToString();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> assembled,
+                         plan->Assemble(made.ValueOrDie(), arrow::default_memory_pool()));
+    auto row = std::static_pointer_cast<arrow::StructArray>(assembled);
+    EXPECT_EQ(static_cast<const arrow::Int64Array&>(*row->field(0)).Value(0), 5);
+    EXPECT_EQ(static_cast<const arrow::StringArray&>(*row->field(1)).GetString(0), "hi");
+}
+
+TEST_F(VariantShreddingReadPlanFactoryTest, NestedVariantColumnAndTypeMismatch) {
+    // A struct column holding a variant child: the plan rebuilds the struct around the reassembled
+    // variant.
+    std::shared_ptr<GenericVariant> variant = Variant(R"({"a": 5})");
+    std::shared_ptr<arrow::DataType> physical;
+    std::shared_ptr<arrow::StructArray> shredded;
+    MakeFullShredded(arrow::struct_({arrow::field("a", arrow::int64())}), variant, &physical,
+                     &shredded);
+    ASSERT_FALSE(HasFatalFailure());
+
+    auto read_field = arrow::field("s", arrow::struct_({arrow::field("x", arrow::int32()),
+                                                        VariantTypeUtils::ToArrowField("v")}));
+    auto file_field = arrow::field(
+        "s", arrow::struct_({arrow::field("x", arrow::int32()), arrow::field("v", physical)}));
+    // A sibling plain struct with no variant exercises `ContainsNestedVariant` returning false.
+    auto plain_field = arrow::field("plain", arrow::struct_({arrow::field("y", arrow::int32())}));
+    auto plans_result = VariantShreddingReadPlanFactory::CreateReadPlans(
+        arrow::schema({plain_field, read_field}), arrow::schema({plain_field, file_field}), pool_);
+    ASSERT_OK_AND_ASSIGN(auto plans, std::move(plans_result));
+    ASSERT_EQ(plans.count("plain"), 0);
+    ASSERT_EQ(plans.count("s"), 1);
+    std::shared_ptr<ShreddingColumnReadPlan> plan = plans["s"];
+
+    auto x_array = Int32Array(7);
+    auto made = arrow::StructArray::Make({x_array, shredded}, std::vector<std::string>{"x", "v"});
+    ASSERT_TRUE(made.ok()) << made.status().ToString();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> assembled,
+                         plan->Assemble(made.ValueOrDie(), arrow::default_memory_pool()));
+    auto row = std::static_pointer_cast<arrow::StructArray>(assembled);
+    EXPECT_EQ(static_cast<const arrow::Int32Array&>(*row->field(0)).Value(0), 7);
+    ASSERT_EQ(row->field(1)->type_id(), arrow::Type::STRUCT);
+
+    // Assembling a physical array whose type differs from the logical struct is an error.
+    auto ints = Int32Array(1);
+    ASSERT_NOK(plan->Assemble(ints, arrow::default_memory_pool()));
+}
+
+TEST_F(VariantShreddingReadPlanFactoryTest, ColumnAbsentInFileSkipped) {
+    // Schema evolution: the variant column is missing from the file, so no plan is produced.
+    auto read_field = VariantTypeUtils::ToArrowField("v");
+    auto other = arrow::field("other", arrow::int32());
+    ASSERT_OK_AND_ASSIGN(auto plans,
+                         VariantShreddingReadPlanFactory::CreateReadPlans(
+                             arrow::schema({read_field}), arrow::schema({other}), pool_));
+    EXPECT_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 e4edef951..d66b6af3f 100644
--- a/src/paimon/common/data/variant/variant_shredding_test.cpp
+++ b/src/paimon/common/data/variant/variant_shredding_test.cpp
@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+#include <functional>
 #include <memory>
 #include <string>
 #include <vector>
@@ -21,6 +22,7 @@
 #include "arrow/api.h"
 #include "gtest/gtest.h"
 #include "paimon/common/data/variant/generic_variant.h"
+#include "paimon/common/data/variant/variant_builder.h"
 #include "paimon/common/data/variant/variant_defs.h"
 #include "paimon/common/data/variant/variant_reassembler.h"
 #include "paimon/common/data/variant/variant_schema.h"
@@ -87,6 +89,78 @@ class VariantShreddingTest : public ::testing::Test {
     }
 
  protected:
+    // The physical shredded arrow type for a logical shredding type.
+    std::shared_ptr<arrow::DataType> Physical(const std::shared_ptr<arrow::DataType>& logical) {
+        EXPECT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> physical,
+                             VariantShreddingUtils::VariantShreddingSchema(logical));
+        return physical;
+    }
+
+    std::shared_ptr<GenericVariant> Json(const char* json) {
+        EXPECT_OK_AND_ASSIGN(std::shared_ptr<GenericVariant> variant,
+                             GenericVariant::FromJson(json, pool_));
+        return variant;
+    }
+
+    // Builds a single variant using the direct append API, which can encode types (float, binary,
+    // date, timestamp) that JSON parsing never produces.
+    std::shared_ptr<GenericVariant> BuildVariant(
+        const std::function<Status(VariantBuilder&)>& append) {
+        VariantBuilder builder(/*allow_duplicate_keys=*/false);
+        Status st = append(builder);
+        EXPECT_TRUE(st.ok()) << st.ToString();
+        EXPECT_OK_AND_ASSIGN(std::shared_ptr<GenericVariant> variant, builder.Build(pool_));
+        return variant;
+    }
+
+    // Shreds the given variants (nullptr = null row) against a physical shredded type, reassembles
+    // them, asserts each reassembled variant renders back to the same JSON, and returns the
+    // shredded array. Unlike `RoundTrip`, the physical type is provided directly so that typed
+    // columns unsupported by `VariantShreddingSchema` (date/timestamp) can be exercised.
+    std::shared_ptr<arrow::StructArray> ShredAndCheck(
+        const std::shared_ptr<arrow::DataType>& physical,
+        const std::vector<std::shared_ptr<GenericVariant>>& variants) {
+        EXPECT_OK_AND_ASSIGN(std::shared_ptr<VariantSchema> schema,
+                             VariantShreddingUtils::BuildVariantSchema(physical));
+        EXPECT_OK_AND_ASSIGN(
+            std::unique_ptr<VariantShreddedColumnWriter> writer,
+            VariantShreddedColumnWriter::Create(schema, physical, arrow::default_memory_pool()));
+        std::vector<std::string> expected_jsons;
+        for (const auto& variant : variants) {
+            if (variant == nullptr) {
+                EXPECT_OK(writer->AppendNull());
+                expected_jsons.emplace_back();
+                continue;
+            }
+            EXPECT_OK_AND_ASSIGN(std::string expected_json, variant->ToJson());
+            expected_jsons.push_back(std::move(expected_json));
+            EXPECT_OK(writer->Append(*variant));
+        }
+        EXPECT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> shredded_array, writer->Finish());
+        auto shredded = std::static_pointer_cast<arrow::StructArray>(shredded_array);
+
+        EXPECT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> assembled_array,
+                             VariantReassembler::AssembleVariantArray(
+                                 shredded, schema, pool_, arrow::default_memory_pool()));
+        auto assembled = std::static_pointer_cast<arrow::StructArray>(assembled_array);
+        auto value_column = std::static_pointer_cast<arrow::BinaryArray>(assembled->field(0));
+        auto metadata_column = std::static_pointer_cast<arrow::BinaryArray>(assembled->field(1));
+        for (size_t i = 0; i < variants.size(); ++i) {
+            SCOPED_TRACE("row " + std::to_string(i));
+            if (variants[i] == nullptr) {
+                EXPECT_TRUE(assembled->IsNull(i));
+                continue;
+            }
+            EXPECT_FALSE(assembled->IsNull(i));
+            EXPECT_OK_AND_ASSIGN(std::shared_ptr<GenericVariant> variant,
+                                 GenericVariant::Create(value_column->GetView(i),
+                                                        metadata_column->GetView(i), pool_));
+            EXPECT_OK_AND_ASSIGN(std::string actual_json, variant->ToJson());
+            EXPECT_EQ(actual_json, expected_jsons[i]);
+        }
+        return shredded;
+    }
+
     std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
 };
 
@@ -239,4 +313,155 @@ TEST_F(VariantShreddingTest, ShredScalarsAndMismatches) {
                            "{\"arr\": [1, 2]}", R"({"arr": {"k": "v"}})"});
 }
 
+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
+    // decimal->integer shredding path.
+    ShredAndCheck(Physical(arrow::struct_({arrow::field("x", arrow::int8())})),
+                  {Json(R"({"x": 5})"), Json(R"({"x": 200})"), Json(R"({"x": 5.0})")});
+    ShredAndCheck(Physical(arrow::struct_({arrow::field("x", arrow::int16())})),
+                  {Json(R"({"x": 5})"), Json(R"({"x": 40000})"), Json(R"({"x": 5.0})")});
+}
+
+TEST_F(VariantShreddingTest, ScalarShreddingFloatAndBinary) {
+    // Float and binary variants are not producible from JSON, so build them directly. They shred
+    // into their typed columns and round-trip back to the same value.
+    ShredAndCheck(Physical(arrow::float32()),
+                  {BuildVariant([](VariantBuilder& b) { return b.AppendFloat(1.5f); }), nullptr});
+    ShredAndCheck(Physical(arrow::binary()), {BuildVariant([](VariantBuilder& b) {
+                      return b.AppendBinary(std::string_view("\x01\x02\x03", 3));
+                  })});
+}
+
+TEST_F(VariantShreddingTest, ScalarShreddingDate) {
+    // date typed columns are produced by external engines; `VariantShreddingSchema` itself never
+    // emits them, so build the physical shredded type directly.
+    auto physical = arrow::struct_({arrow::field("metadata", arrow::binary(), false),
+                                    arrow::field("value", arrow::binary(), true),
+                                    arrow::field("typed_value", arrow::date32(), true)});
+    ShredAndCheck(physical,
+                  {BuildVariant([](VariantBuilder& b) { return b.AppendDate(19000); }), nullptr});
+}
+
+TEST_F(VariantShreddingTest, TimestampSchemaParsing) {
+    auto make_physical = [](const std::shared_ptr<arrow::DataType>& ts) {
+        return arrow::struct_({arrow::field("metadata", arrow::binary(), false),
+                               arrow::field("value", arrow::binary(), true),
+                               arrow::field("typed_value", ts, true)});
+    };
+    // A microsecond timestamp with a timezone parses as TIMESTAMP_LTZ; without, TIMESTAMP_NTZ.
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<VariantSchema> ltz,
+                         VariantShreddingUtils::BuildVariantSchema(
+                             make_physical(arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"))));
+    ASSERT_TRUE(ltz->scalar_schema.has_value());
+    ASSERT_EQ(ltz->scalar_schema->kind, VariantSchema::ScalarKind::kTimestampLtz);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<VariantSchema> ntz,
+                         VariantShreddingUtils::BuildVariantSchema(
+                             make_physical(arrow::timestamp(arrow::TimeUnit::MICRO))));
+    ASSERT_TRUE(ntz->scalar_schema.has_value());
+    ASSERT_EQ(ntz->scalar_schema->kind, VariantSchema::ScalarKind::kTimestampNtz);
+    // Non-microsecond timestamps cannot represent the variant's microsecond values, so they are
+    // rejected as an invalid shredding schema.
+    ASSERT_NOK(VariantShreddingUtils::BuildVariantSchema(
+        make_physical(arrow::timestamp(arrow::TimeUnit::MILLI, "UTC"))));
+}
+
+TEST_F(VariantShreddingTest, TimestampReassembly) {
+    // The writer never produces timestamp typed columns, but the reassembler must handle files
+    // written by engines that do. Build a shredded array with a populated timestamp typed_value
+    // and verify it reassembles into the same variant.
+    auto reassemble_one = [&](const std::shared_ptr<arrow::DataType>& ts_type,
+                              const std::shared_ptr<GenericVariant>& reference, int64_t micros) {
+        auto physical = arrow::struct_({arrow::field("metadata", arrow::binary(), false),
+                                        arrow::field("value", arrow::binary(), true),
+                                        arrow::field("typed_value", ts_type, true)});
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<VariantSchema> schema,
+                             VariantShreddingUtils::BuildVariantSchema(physical));
+        std::string metadata(reference->Metadata());
+        ASSERT_OK_AND_ASSIGN(std::string expected_json, reference->ToJson());
+
+        arrow::BinaryBuilder meta_builder;
+        ASSERT_TRUE(meta_builder.Append(metadata).ok());
+        std::shared_ptr<arrow::Array> meta_array;
+        ASSERT_TRUE(meta_builder.Finish(&meta_array).ok());
+        arrow::BinaryBuilder value_builder;
+        ASSERT_TRUE(value_builder.AppendNull().ok());
+        std::shared_ptr<arrow::Array> value_array;
+        ASSERT_TRUE(value_builder.Finish(&value_array).ok());
+        arrow::TimestampBuilder ts_builder(ts_type, arrow::default_memory_pool());
+        ASSERT_TRUE(ts_builder.Append(micros).ok());
+        std::shared_ptr<arrow::Array> ts_array;
+        ASSERT_TRUE(ts_builder.Finish(&ts_array).ok());
+        auto made = arrow::StructArray::Make({meta_array, value_array, ts_array},
+                                             {"metadata", "value", "typed_value"});
+        ASSERT_TRUE(made.ok()) << made.status().ToString();
+        std::shared_ptr<arrow::StructArray> shredded = made.ValueOrDie();
+
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> assembled_array,
+                             VariantReassembler::AssembleVariantArray(
+                                 shredded, schema, pool_, arrow::default_memory_pool()));
+        auto assembled = std::static_pointer_cast<arrow::StructArray>(assembled_array);
+        auto value_column = std::static_pointer_cast<arrow::BinaryArray>(assembled->field(0));
+        auto metadata_column = std::static_pointer_cast<arrow::BinaryArray>(assembled->field(1));
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<GenericVariant> variant,
+            GenericVariant::Create(value_column->GetView(0), metadata_column->GetView(0), pool_));
+        ASSERT_OK_AND_ASSIGN(std::string actual_json, variant->ToJson());
+        ASSERT_EQ(actual_json, expected_json);
+    };
+
+    int64_t micros = 1700000000000000;
+    reassemble_one(arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"),
+                   BuildVariant([&](VariantBuilder& b) { return b.AppendTimestamp(micros); }),
+                   micros);
+    reassemble_one(arrow::timestamp(arrow::TimeUnit::MICRO),
+                   BuildVariant([&](VariantBuilder& b) { return b.AppendTimestampNtz(micros); }),
+                   micros);
+}
+
+TEST_F(VariantShreddingTest, ScalarSchemaToArrowType) {
+    using SK = VariantSchema::ScalarKind;
+    auto check = [](VariantSchema::ScalarType scalar,
+                    const std::shared_ptr<arrow::DataType>& expected) {
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::DataType> type,
+                             VariantShreddingUtils::ScalarSchemaToArrowType(scalar));
+        ASSERT_TRUE(type->Equals(*expected)) << type->ToString();
+    };
+    check({SK::kBoolean}, arrow::boolean());
+    check({SK::kByte}, arrow::int8());
+    check({SK::kShort}, arrow::int16());
+    check({SK::kInt}, arrow::int32());
+    check({SK::kLong}, arrow::int64());
+    check({SK::kFloat}, arrow::float32());
+    check({SK::kDouble}, arrow::float64());
+    check({SK::kString}, arrow::utf8());
+    check({SK::kBinary}, arrow::binary());
+    check({SK::kDecimal, 10, 2}, arrow::decimal128(10, 2));
+    check({SK::kDate}, arrow::date32());
+    check({SK::kTimestampLtz}, arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"));
+    check({SK::kTimestampNtz}, arrow::timestamp(arrow::TimeUnit::MICRO));
+    // kUuid has no shredded arrow representation.
+    ASSERT_NOK(VariantShreddingUtils::ScalarSchemaToArrowType({SK::kUuid}));
+}
+
+TEST_F(VariantShreddingTest, InvalidShreddingSchemas) {
+    // Not a struct.
+    ASSERT_NOK(VariantShreddingUtils::BuildVariantSchema(arrow::int32()));
+    // Empty struct.
+    ASSERT_NOK(VariantShreddingUtils::BuildVariantSchema(arrow::struct_({})));
+    // The "value" column must be binary.
+    ASSERT_NOK(VariantShreddingUtils::BuildVariantSchema(
+        arrow::struct_({arrow::field("value", arrow::int32())})));
+    // Unknown field name.
+    ASSERT_NOK(VariantShreddingUtils::BuildVariantSchema(
+        arrow::struct_({arrow::field("bogus", arrow::binary())})));
+    // A top-level schema must carry a metadata column.
+    ASSERT_NOK(VariantShreddingUtils::BuildVariantSchema(
+        arrow::struct_({arrow::field("value", arrow::binary())})));
+    // Unsupported typed_value type.
+    ASSERT_NOK(VariantShreddingUtils::BuildVariantSchema(arrow::struct_(
+        {arrow::field("metadata", arrow::binary()), arrow::field("value", arrow::binary()),
+         arrow::field("typed_value", arrow::map(arrow::utf8(), arrow::int32()))})));
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/common/logging/logging_test.cpp b/src/paimon/common/logging/logging_test.cpp
index fc0b10c28..78b2be860 100644
--- a/src/paimon/common/logging/logging_test.cpp
+++ b/src/paimon/common/logging/logging_test.cpp
@@ -15,14 +15,60 @@
  */
 #include "paimon/logging.h"
 
+#include <atomic>
+#include <cstdarg>
+#include <filesystem>
+#include <fstream>
 #include <future>
+#include <iostream>
+#include <memory>
 #include <random>
+#include <sstream>
+#include <string>
 #include <thread>
 
+#include "glog/log_severity.h"
+#include "glog/logging.h"
+#include "glog/raw_logging.h"
 #include "paimon/common/executor/future.h"
 #include "paimon/executor.h"
 #include "paimon/testing/utils/testharness.h"
 namespace paimon::test {
+namespace {
+
+std::atomic<int> g_creator_calls{0};
+
+// A Logger that forwards to glog exactly like the built-in adaptor. It is used to
+// restore behavior-preserving logging after exercising the custom-creator path,
+// because the registry cannot be reset to "unset" through the public API.
+class GlogForwardingLogger : public Logger {
+ public:
+    void LogV(PaimonLogLevel level, const char* fname, int lineno, const char* /*function*/,
+              const char* fmt, ...) override {
+        va_list args;
+        va_start(args, fmt);
+        google::RawLog__(ToGlog(level), fname, lineno, fmt, args);
+        va_end(args);
+    }
+
+    bool IsLevelEnabled(PaimonLogLevel /*level*/) const override {
+        return true;
+    }
+
+ private:
+    static google::LogSeverity ToGlog(PaimonLogLevel level) {
+        switch (level) {
+            case PAIMON_LOG_LEVEL_WARN:
+                return google::GLOG_WARNING;
+            case PAIMON_LOG_LEVEL_ERROR:
+                return google::GLOG_ERROR;
+            default:
+                return google::GLOG_INFO;
+        }
+    }
+};
+
+}  // namespace
 TEST(LoggerTest, TestMultiThreadGetLogger) {
     ASSERT_OK_AND_ASSIGN(auto executor, CreateDefaultExecutor(/*thread_count=*/4));
     auto get_logger = []() {
@@ -36,4 +82,99 @@ TEST(LoggerTest, TestMultiThreadGetLogger) {
     }
     Wait(futures);
 }
+
+TEST(LoggerTest, TestLogAllSeverities) {
+    // The default logger routes every PaimonLogLevel through the severity mapping.
+    auto logger = Logger::GetLogger("severity_test");
+    ASSERT_TRUE(logger);
+
+    logger->LogV(PAIMON_LOG_LEVEL_DEBUG, __FILE__, __LINE__, __FUNCTION__, "debug severity");
+    logger->LogV(PAIMON_LOG_LEVEL_INFO, __FILE__, __LINE__, __FUNCTION__, "info severity");
+    logger->LogV(PAIMON_LOG_LEVEL_WARN, __FILE__, __LINE__, __FUNCTION__, "warn severity");
+    logger->LogV(PAIMON_LOG_LEVEL_ERROR, __FILE__, __LINE__, __FUNCTION__, "error severity");
+    // NONE and MAX fall into the default branch of the mapping.
+    logger->LogV(PAIMON_LOG_LEVEL_NONE, __FILE__, __LINE__, __FUNCTION__, "none severity");
+    logger->LogV(PAIMON_LOG_LEVEL_MAX, __FILE__, __LINE__, __FUNCTION__, "max severity");
+}
+
+// Demonstrates that glog's normal LOG() path actually writes a log file to disk.
+// Note: Paimon's Logger/GlogAdaptor uses google::RawLog__, which only goes to
+// stderr and never touches disk, so this test drives glog's file sink directly.
+TEST(LoggerTest, TestGlogWritesLogFileToDisk) {
+    namespace fs = std::filesystem;
+
+    // Save the global glog flags we are about to change so other tests are unaffected.
+    const bool prev_logtostderr = FLAGS_logtostderr;
+    const bool prev_timestamp_in_name = FLAGS_timestamp_in_logfile_name;
+    const int32_t prev_minloglevel = FLAGS_minloglevel;
+
+    // A unique, empty directory so the only file inside is the one glog creates for us.
+    fs::path log_dir = fs::temp_directory_path() /
+                       ("paimon_glog_disk_" + std::to_string(RandomNumber(0, 1'000'000'000)));
+    fs::create_directories(log_dir);
+    const fs::path base = log_dir / "paimon_demo";
+
+    FLAGS_logtostderr = false;                // must be false, otherwise glog skips the file sink
+    FLAGS_timestamp_in_logfile_name = false;  // deterministic file name (no time/pid suffix)
+    FLAGS_minloglevel = google::GLOG_INFO;    // do not filter out INFO
+
+    if (!google::IsGoogleLoggingInitialized()) {
+        google::InitGoogleLogging("paimon-log-disk-test");
+    }
+    // Force INFO logs to a file under our unique directory.
+    google::SetLogDestination(google::GLOG_INFO, base.string().c_str());
+
+    const std::string token = "PAIMON_DISK_LOG_DEMO_" + std::to_string(RandomNumber(0, 1'000'000));
+    LOG(INFO) << "hello from disk logging, token=" << token;
+    google::FlushLogFiles(google::GLOG_INFO);
+
+    // Collect the content of whatever regular file glog created in our directory.
+    std::string on_disk_path;
+    std::string content;
+    for (const auto& entry : fs::directory_iterator(log_dir)) {
+        if (!entry.is_regular_file()) {
+            continue;
+        }
+        std::ifstream in(entry.path());
+        std::stringstream buffer;
+        buffer << in.rdbuf();
+        if (buffer.str().find(token) != std::string::npos) {
+            on_disk_path = entry.path().string();
+            content = buffer.str();
+            break;
+        }
+    }
+
+    // Show the real on-disk file and its raw content so it can be eyeballed in test output.
+    std::cout << "\n===== on-disk glog file: " << on_disk_path << " =====\n"
+              << content << "===== end of file =====\n";
+
+    ASSERT_FALSE(on_disk_path.empty())
+        << "no glog file containing the token was written to " << log_dir.string();
+    ASSERT_NE(content.find(token), std::string::npos);
+
+    // Stop writing INFO logs to the directory we are about to delete, then restore flags.
+    google::SetLogDestination(google::GLOG_INFO, "");
+    FLAGS_logtostderr = prev_logtostderr;
+    FLAGS_timestamp_in_logfile_name = prev_timestamp_in_name;
+    FLAGS_minloglevel = prev_minloglevel;
+    std::error_code ec;
+    fs::remove_all(log_dir, ec);
+}
+
+// Keep this test last: it installs a process-wide logger creator that cannot be
+// unset, so it must not run before tests that rely on the default logger.
+TEST(LoggerTest, TestRegisterCustomLoggerCreator) {
+    g_creator_calls.store(0);
+    Logger::RegisterLogger([](const std::string& /*path*/) -> std::unique_ptr<Logger> {
+        g_creator_calls.fetch_add(1);
+        return std::make_unique<GlogForwardingLogger>();
+    });
+
+    auto logger = Logger::GetLogger("custom_path");
+    ASSERT_TRUE(logger);
+    // GetLogger must have gone through the registered creator branch.
+    ASSERT_EQ(1, g_creator_calls.load());
+    ASSERT_TRUE(logger->IsLevelEnabled(PAIMON_LOG_LEVEL_INFO));
+}
 }  // namespace paimon::test
diff --git a/src/paimon/common/utils/arrow/mem_utils_test.cpp b/src/paimon/common/utils/arrow/mem_utils_test.cpp
index 2881d79bf..112268b2c 100644
--- a/src/paimon/common/utils/arrow/mem_utils_test.cpp
+++ b/src/paimon/common/utils/arrow/mem_utils_test.cpp
@@ -16,10 +16,53 @@
 
 #include "paimon/common/utils/arrow/mem_utils.h"
 
+#include <cstdint>
+#include <memory>
+#include <new>
+
+#include "arrow/status.h"
 #include "gtest/gtest.h"
 #include "paimon/memory/memory_pool.h"
 
 namespace paimon::test {
+namespace {
+
+// A MemoryPool whose allocations always fail, either by returning nullptr or by
+// throwing std::bad_alloc, so the adaptor's out-of-memory paths can be exercised.
+class FailingMemoryPool : public MemoryPool {
+ public:
+    enum class Mode { kReturnNull, kThrowBadAlloc };
+
+    explicit FailingMemoryPool(Mode mode) : mode_(mode) {}
+
+    void* Malloc(uint64_t /*size*/, uint64_t /*alignment*/ = 0) override {
+        if (mode_ == Mode::kThrowBadAlloc) {
+            throw std::bad_alloc();
+        }
+        return nullptr;
+    }
+
+    void* Realloc(void* /*p*/, size_t /*old_size*/, size_t /*new_size*/,
+                  uint64_t /*alignment*/ = 0) override {
+        if (mode_ == Mode::kThrowBadAlloc) {
+            throw std::bad_alloc();
+        }
+        return nullptr;
+    }
+
+    void Free(void* /*p*/, uint64_t /*size*/) override {}
+    uint64_t CurrentUsage() const override {
+        return 0;
+    }
+    uint64_t MaxMemoryUsage() const override {
+        return 0;
+    }
+
+ private:
+    Mode mode_;
+};
+
+}  // namespace
 
 TEST(MemUtilsTest, TestSimple) {
     const int64_t alignment = 64;
@@ -71,4 +114,32 @@ TEST(MemUtilsTest, TestSimple) {
     ASSERT_EQ(50, pool->max_memory());
 }
 
+TEST(MemUtilsTest, TestAllocateOutOfMemory) {
+    uint8_t* ptr = nullptr;
+
+    // Underlying pool returns nullptr for a positive size.
+    auto null_pool =
+        GetArrowPool(std::make_shared<FailingMemoryPool>(FailingMemoryPool::Mode::kReturnNull));
+    ASSERT_TRUE(null_pool->Allocate(16, 64, &ptr).IsOutOfMemory());
+
+    // Underlying pool throws std::bad_alloc.
+    auto throw_pool =
+        GetArrowPool(std::make_shared<FailingMemoryPool>(FailingMemoryPool::Mode::kThrowBadAlloc));
+    ASSERT_TRUE(throw_pool->Allocate(16, 64, &ptr).IsOutOfMemory());
+}
+
+TEST(MemUtilsTest, TestReallocateOutOfMemory) {
+    uint8_t* ptr = nullptr;
+
+    // Underlying pool returns nullptr for a positive new size.
+    auto null_pool =
+        GetArrowPool(std::make_shared<FailingMemoryPool>(FailingMemoryPool::Mode::kReturnNull));
+    ASSERT_TRUE(null_pool->Reallocate(/*old_size=*/0, /*new_size=*/16, 64, &ptr).IsOutOfMemory());
+
+    // Underlying pool throws std::bad_alloc.
+    auto throw_pool =
+        GetArrowPool(std::make_shared<FailingMemoryPool>(FailingMemoryPool::Mode::kThrowBadAlloc));
+    ASSERT_TRUE(throw_pool->Reallocate(/*old_size=*/0, /*new_size=*/16, 64, &ptr).IsOutOfMemory());
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/common/utils/file_type_test.cpp b/src/paimon/common/utils/file_type_test.cpp
index be77b6e16..21b5c14c1 100644
--- a/src/paimon/common/utils/file_type_test.cpp
+++ b/src/paimon/common/utils/file_type_test.cpp
@@ -28,6 +28,14 @@ TEST(FileTypeTest, TestIsIndex) {
     ASSERT_FALSE(FileTypeUtils::IsIndex(FileType::kData));
 }
 
+TEST(FileTypeTest, TestToString) {
+    ASSERT_EQ(FileTypeUtils::ToString(FileType::kMeta), "meta");
+    ASSERT_EQ(FileTypeUtils::ToString(FileType::kData), "data");
+    ASSERT_EQ(FileTypeUtils::ToString(FileType::kBucketIndex), "bucket_index");
+    ASSERT_EQ(FileTypeUtils::ToString(FileType::kGlobalIndex), "global_index");
+    ASSERT_EQ(FileTypeUtils::ToString(FileType::kFileIndex), "file_index");
+}
+
 TEST(FileTypeTest, TestMetaPrefix) {
     ASSERT_EQ(FileTypeUtils::Classify("dfs://cluster/db/snapshot/snapshot-1"), FileType::kMeta);
     ASSERT_EQ(FileTypeUtils::Classify("dfs://cluster/db/schema/schema-2"), FileType::kMeta);
@@ -173,6 +181,12 @@ TEST(FileTypeTest, TestInvalidTempWrapperFallsBackToOriginalName) {
 
     // Too short -> should not unwrap.
     ASSERT_EQ(FileTypeUtils::Classify("dfs://cluster/db/snapshot/.x.tmp"), FileType::kData);
+
+    // Long enough and ends with .tmp, but the char before the trailing 41-char suffix is not a
+    // dot -> should not unwrap.
+    ASSERT_EQ(FileTypeUtils::Classify(
+                  "dfs://cluster/db/snapshot/.snapshot-1234567890123456789012345678901234.tmp"),
+              FileType::kData);
 }
 
 }  // namespace paimon::test
diff --git a/src/paimon/common/utils/status_test.cpp b/src/paimon/common/utils/status_test.cpp
index 9fffb11d9..33be82668 100644
--- a/src/paimon/common/utils/status_test.cpp
+++ b/src/paimon/common/utils/status_test.cpp
@@ -71,6 +71,43 @@ TEST(StatusTest, TestToStringWithDetail) {
     ASSERT_EQ(status.ToString(), ss.str());
 }
 
+TEST(StatusTest, TestCodeAsString) {
+    ASSERT_EQ("OK", Status::CodeAsString(StatusCode::OK));
+    ASSERT_EQ("Out of memory", Status::CodeAsString(StatusCode::OutOfMemory));
+    ASSERT_EQ("Key error", Status::CodeAsString(StatusCode::KeyError));
+    ASSERT_EQ("Type error", Status::CodeAsString(StatusCode::TypeError));
+    ASSERT_EQ("Invalid", Status::CodeAsString(StatusCode::Invalid));
+    ASSERT_EQ("IOError", Status::CodeAsString(StatusCode::IOError));
+    ASSERT_EQ("Capacity error", Status::CodeAsString(StatusCode::CapacityError));
+    ASSERT_EQ("Index error", Status::CodeAsString(StatusCode::IndexError));
+    ASSERT_EQ("Cancelled", Status::CodeAsString(StatusCode::Cancelled));
+    ASSERT_EQ("Unknown error", Status::CodeAsString(StatusCode::UnknownError));
+    ASSERT_EQ("NotImplemented", Status::CodeAsString(StatusCode::NotImplemented));
+    ASSERT_EQ("Serialization error", Status::CodeAsString(StatusCode::SerializationError));
+    ASSERT_EQ("Not exist", Status::CodeAsString(StatusCode::NotExist));
+    ASSERT_EQ("Exist", Status::CodeAsString(StatusCode::Exist));
+
+    // An out-of-range code falls into the default branch. The cast is intentional to
+    // exercise the defensive default, so the enum-range analyzer check is suppressed.
+    // NOLINTNEXTLINE(clang-analyzer-optin.core.EnumCastOutOfRange)
+    ASSERT_EQ("Unknown", Status::CodeAsString(static_cast<StatusCode>(99)));
+
+    // The instance overload returns "OK" for a success status.
+    ASSERT_EQ("OK", Status::OK().CodeAsString());
+}
+
+TEST(StatusDeathTest, TestAbort) {
+    // Death tests fork(); with the default "fast" style, forking in a multi-threaded
+    // process is unsafe and under ThreadSanitizer the child aborts with a sanitizer
+    // message before Abort() runs. The "threadsafe" style re-execs the test binary in
+    // a clean process so Abort()'s own output is produced and can be matched.
+    const std::string prev_style = testing::GTEST_FLAG(death_test_style);
+    testing::GTEST_FLAG(death_test_style) = "threadsafe";
+    ASSERT_DEATH(Status::IOError("boom").Abort(), "Paimon Fatal Error");
+    ASSERT_DEATH(Status::IOError("boom").Abort("custom prefix"), "custom prefix");
+    testing::GTEST_FLAG(death_test_style) = prev_style;
+}
+
 TEST(StatusTest, TestWithDetail) {
     Status status(StatusCode::IOError, "summary");
     auto detail = std::make_shared<TestStatusDetail>();