diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index ec0c6ea8c..061badf26 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -612,6 +612,7 @@ if(PAIMON_BUILD_TESTS)
                     common/data/columnar/columnar_utils_test.cpp
                     common/data/data_define_test.cpp
                     common/data/decimal_test.cpp
+                    common/data/generic_array_test.cpp
                     common/data/generic_row_test.cpp
                     common/data/internal_row_test.cpp
                     common/data/record_batch_test.cpp
@@ -827,6 +828,7 @@ if(PAIMON_BUILD_TESTS)
                     core/bucket/bucket_select_converter_test.cpp
                     core/bucket/default_bucket_function_test.cpp
                     core/bucket/hive_bucket_function_test.cpp
+                    core/bucket/hive_hasher_test.cpp
                     core/bucket/bucket_id_calculator_test.cpp
                     core/bucket/mod_bucket_function_test.cpp
                     core/casting/cast_executor_factory_test.cpp
diff --git a/src/paimon/common/data/generic_array_test.cpp b/src/paimon/common/data/generic_array_test.cpp
new file mode 100644
index 000000000..a75e0d068
--- /dev/null
+++ b/src/paimon/common/data/generic_array_test.cpp
@@ -0,0 +1,208 @@
+/*
+ * 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/data/generic_array.h"
+
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/generic_map.h"
+#include "paimon/common/data/generic_row.h"
+#include "paimon/data/decimal.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(GenericArrayTest, TestGetters) {
+    const Decimal decimal(/*precision=*/10, /*scale=*/2, 12345);
+    const Timestamp timestamp(/*millisecond=*/1234, /*nano_of_millisecond=*/567890);
+    std::shared_ptr<InternalArray> nested_array =
+        std::make_shared<GenericArray>(std::vector<VariantType>{int32_t{42}});
+    std::shared_ptr<InternalMap> map = std::make_shared<GenericMap>(nested_array, nested_array);
+    auto row = std::make_shared<GenericRow>(1);
+    row->SetField(0, int32_t{7});
+    GenericArray array({true, false, char{42}, int16_t{-1234}, int32_t{-56789},
+                        int64_t{12345678900}, -1.5F, 2.25, decimal, timestamp, nested_array, map,
+                        std::shared_ptr<InternalRow>(row), NullType{}});
+
+    ASSERT_EQ(array.Size(), 14);
+    for (int32_t i = 0; i < array.Size() - 1; ++i) {
+        ASSERT_FALSE(array.IsNullAt(i)) << i;
+    }
+    ASSERT_TRUE(array.IsNullAt(13));
+    ASSERT_TRUE(array.GetBoolean(0));
+    ASSERT_FALSE(array.GetBoolean(1));
+    ASSERT_EQ(array.GetByte(2), char{42});
+    ASSERT_EQ(array.GetShort(3), -1234);
+    ASSERT_EQ(array.GetInt(4), -56789);
+    ASSERT_EQ(array.GetDate(4), -56789);
+    ASSERT_EQ(array.GetLong(5), int64_t{12345678900});
+    ASSERT_FLOAT_EQ(array.GetFloat(6), -1.5F);
+    ASSERT_DOUBLE_EQ(array.GetDouble(7), 2.25);
+    ASSERT_EQ(array.GetDecimal(8, /*precision=*/10, /*scale=*/2), decimal);
+    ASSERT_EQ(array.GetTimestamp(9, /*precision=*/9), timestamp);
+    ASSERT_EQ(array.GetArray(10), nested_array);
+    ASSERT_EQ(array.GetArray(10)->GetInt(0), 42);
+    ASSERT_EQ(array.GetMap(11), map);
+    ASSERT_EQ(array.GetMap(11)->ValueArray()->GetInt(0), 42);
+    ASSERT_EQ(array.GetRow(12, /*num_fields=*/1), row);
+    ASSERT_EQ(array.GetRow(12, /*num_fields=*/1)->GetInt(0), 7);
+}
+
+TEST(GenericArrayTest, TestStringAndBinaryRepresentations) {
+    auto pool = GetDefaultPool();
+    for (const std::string& value :
+         {std::string(), std::string("hello world"), std::string("a\0b", 3)}) {
+        SCOPED_TRACE(value);
+        BinaryString string = BinaryString::FromString(value, pool.get());
+        std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(value, pool.get());
+        GenericArray array({string, bytes, std::string_view(value)});
+
+        ASSERT_EQ(array.Size(), 3);
+        for (int32_t i = 0; i < array.Size(); ++i) {
+            ASSERT_FALSE(array.IsNullAt(i));
+            ASSERT_EQ(array.GetString(i).ToString(), value) << i;
+            ASSERT_EQ(array.GetStringView(i), value) << i;
+            auto result = array.GetBinary(i);
+            ASSERT_TRUE(result);
+            ASSERT_EQ(*result, *bytes) << i;
+        }
+        ASSERT_EQ(array.GetBinary(1), bytes);
+        ASSERT_EQ(array.GetStringView(2).data(), value.data());
+    }
+}
+
+TEST(GenericArrayTest, TestHoldersKeepBorrowedValuesAlive) {
+    auto pool = GetDefaultPool();
+    const std::string string_value = "string owned by the source array";
+    const std::string binary_value("binary\0value", 12);
+    std::weak_ptr<InternalArray> source_ref;
+    std::unique_ptr<GenericArray> array;
+    {
+        std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(binary_value, pool.get());
+        auto source = std::make_shared<GenericArray>(
+            std::vector<VariantType>{BinaryString::FromString(string_value, pool.get()), bytes});
+        source_ref = source;
+        array = std::make_unique<GenericArray>(
+            std::vector<VariantType>{source->GetStringView(0), source->GetStringView(1)},
+            std::vector<std::shared_ptr<InternalArray>>{source});
+    }
+
+    ASSERT_FALSE(source_ref.expired());
+    ASSERT_EQ(array->GetStringView(0), string_value);
+    ASSERT_EQ(array->GetString(0).ToString(), string_value);
+    auto bytes = array->GetBinary(1);
+    ASSERT_TRUE(bytes);
+    ASSERT_EQ(std::string(bytes->data(), bytes->size()), binary_value);
+    array.reset();
+    ASSERT_TRUE(source_ref.expired());
+    ASSERT_EQ(std::string(bytes->data(), bytes->size()), binary_value);
+}
+
+TEST(GenericArrayTest, TestToPrimitiveArrays) {
+    {
+        GenericArray array({true, false, true});
+        ASSERT_OK_AND_ASSIGN(std::vector<char> result, array.ToBooleanArray());
+        ASSERT_EQ(result, (std::vector<char>{1, 0, 1}));
+    }
+    {
+        GenericArray array({char{0}, char{42}, std::numeric_limits<char>::max()});
+        ASSERT_OK_AND_ASSIGN(std::vector<char> result, array.ToByteArray());
+        ASSERT_EQ(result, (std::vector<char>{0, 42, std::numeric_limits<char>::max()}));
+    }
+    {
+        GenericArray array(
+            {std::numeric_limits<int16_t>::min(), int16_t{0}, std::numeric_limits<int16_t>::max()});
+        ASSERT_OK_AND_ASSIGN(std::vector<int16_t> result, array.ToShortArray());
+        ASSERT_EQ(result, (std::vector<int16_t>{std::numeric_limits<int16_t>::min(), 0,
+                                                std::numeric_limits<int16_t>::max()}));
+    }
+    {
+        GenericArray array(
+            {std::numeric_limits<int32_t>::min(), int32_t{0}, std::numeric_limits<int32_t>::max()});
+        ASSERT_OK_AND_ASSIGN(std::vector<int32_t> result, array.ToIntArray());
+        ASSERT_EQ(result, (std::vector<int32_t>{std::numeric_limits<int32_t>::min(), 0,
+                                                std::numeric_limits<int32_t>::max()}));
+    }
+    {
+        GenericArray array(
+            {std::numeric_limits<int64_t>::min(), int64_t{0}, std::numeric_limits<int64_t>::max()});
+        ASSERT_OK_AND_ASSIGN(std::vector<int64_t> result, array.ToLongArray());
+        ASSERT_EQ(result, (std::vector<int64_t>{std::numeric_limits<int64_t>::min(), 0,
+                                                std::numeric_limits<int64_t>::max()}));
+    }
+    {
+        GenericArray array({-1.5F, 0.0F, 2.25F});
+        ASSERT_OK_AND_ASSIGN(std::vector<float> result, array.ToFloatArray());
+        ASSERT_EQ(result, (std::vector<float>{-1.5F, 0.0F, 2.25F}));
+    }
+    {
+        GenericArray array({-1.5, 0.0, 2.25});
+        ASSERT_OK_AND_ASSIGN(std::vector<double> result, array.ToDoubleArray());
+        ASSERT_EQ(result, (std::vector<double>{-1.5, 0.0, 2.25}));
+    }
+}
+
+TEST(GenericArrayTest, TestEmptyArray) {
+    GenericArray array({});
+    ASSERT_EQ(array.Size(), 0);
+    ASSERT_OK_AND_ASSIGN(std::vector<char> booleans, array.ToBooleanArray());
+    ASSERT_TRUE(booleans.empty());
+    ASSERT_OK_AND_ASSIGN(std::vector<char> bytes, array.ToByteArray());
+    ASSERT_TRUE(bytes.empty());
+    ASSERT_OK_AND_ASSIGN(std::vector<int16_t> shorts, array.ToShortArray());
+    ASSERT_TRUE(shorts.empty());
+    ASSERT_OK_AND_ASSIGN(std::vector<int32_t> ints, array.ToIntArray());
+    ASSERT_TRUE(ints.empty());
+    ASSERT_OK_AND_ASSIGN(std::vector<int64_t> longs, array.ToLongArray());
+    ASSERT_TRUE(longs.empty());
+    ASSERT_OK_AND_ASSIGN(std::vector<float> floats, array.ToFloatArray());
+    ASSERT_TRUE(floats.empty());
+    ASSERT_OK_AND_ASSIGN(std::vector<double> doubles, array.ToDoubleArray());
+    ASSERT_TRUE(doubles.empty());
+}
+
+TEST(GenericArrayTest, TestPrimitiveArraysRejectNulls) {
+    GenericArray null_array({NullType{}});
+    ASSERT_TRUE(null_array.IsNullAt(0));
+    const std::string message = "Primitive array must not contain a null value.";
+    ASSERT_NOK_WITH_MSG(null_array.ToBooleanArray(), message);
+    ASSERT_NOK_WITH_MSG(null_array.ToByteArray(), message);
+    ASSERT_NOK_WITH_MSG(null_array.ToShortArray(), message);
+    ASSERT_NOK_WITH_MSG(null_array.ToIntArray(), message);
+    ASSERT_NOK_WITH_MSG(null_array.ToLongArray(), message);
+    ASSERT_NOK_WITH_MSG(null_array.ToFloatArray(), message);
+    ASSERT_NOK_WITH_MSG(null_array.ToDoubleArray(), message);
+
+    GenericArray booleans({true, NullType{}, false});
+    ASSERT_NOK_WITH_MSG(booleans.ToBooleanArray(), message);
+    GenericArray ints({int32_t{1}, NullType{}, int32_t{2}});
+    ASSERT_NOK_WITH_MSG(ints.ToIntArray(), message);
+}
+
+}  // namespace paimon::test
diff --git a/src/paimon/common/file_index/file_index_reader_test.cpp b/src/paimon/common/file_index/file_index_reader_test.cpp
index 40a255597..78002f718 100644
--- a/src/paimon/common/file_index/file_index_reader_test.cpp
+++ b/src/paimon/common/file_index/file_index_reader_test.cpp
@@ -70,6 +70,12 @@ TEST(FileIndexReaderTest, TestDefaultIndexReader) {
     ASSERT_TRUE(reader.VisitIn({lit0, lit1}).value()->IsRemain().value());
     ASSERT_TRUE(reader.VisitNotIn({lit0, lit1}).value()->IsRemain().value());
 
+    Literal text(FieldType::STRING, "abc", 3);
+    ASSERT_TRUE(reader.VisitStartsWith(text).value()->IsRemain().value());
+    ASSERT_TRUE(reader.VisitEndsWith(text).value()->IsRemain().value());
+    ASSERT_TRUE(reader.VisitContains(text).value()->IsRemain().value());
+    ASSERT_TRUE(reader.VisitLike(text).value()->IsRemain().value());
+
     std::shared_ptr<VectorSearch> vector_search =
         std::make_shared<VectorSearch>("embedding", 1, std::vector<float>{1.0f}, nullptr, nullptr,
                                        std::nullopt, std::map<std::string, std::string>{});
diff --git a/src/paimon/common/fs/object_store_file_system_test.cpp b/src/paimon/common/fs/object_store_file_system_test.cpp
index 6b8b56503..880d02bd2 100644
--- a/src/paimon/common/fs/object_store_file_system_test.cpp
+++ b/src/paimon/common/fs/object_store_file_system_test.cpp
@@ -25,6 +25,8 @@
 #include <cstring>
 #include <limits>
 #include <map>
+#include <memory>
+#include <string>
 #include <utility>
 #include <vector>
 
@@ -121,13 +123,30 @@ TEST(ObjectStoreFileSystemTest, TestObjectWinsOverPrefix) {
     ASSERT_FALSE(statuses[0].IsDir());
 }
 
+TEST(ObjectStoreFileSystemTest, TestExistsAndDirectoryStatus) {
+    auto client = std::make_shared<MockObjectStoreClient>();
+    ObjectStoreFileSystem fs("s3", client);
+    ASSERT_OK_AND_ASSIGN(bool root_exists, fs.Exists("s3://bucket/"));
+    ASSERT_TRUE(root_exists);
+    client->objects_["dir/file"] = "data";
+    ASSERT_OK_AND_ASSIGN(bool file_exists, fs.Exists("s3://bucket/dir/file"));
+    ASSERT_TRUE(file_exists);
+    ASSERT_OK_AND_ASSIGN(bool directory_exists, fs.Exists("s3://bucket/dir"));
+    ASSERT_TRUE(directory_exists);
+    ASSERT_OK_AND_ASSIGN(bool missing_exists, fs.Exists("s3://bucket/missing"));
+    ASSERT_FALSE(missing_exists);
+    ASSERT_OK_AND_ASSIGN(FileStatus status, fs.GetFileStatus("s3://bucket/dir"));
+    ASSERT_TRUE(status.IsDir());
+    ASSERT_EQ("s3://bucket/dir/", status.GetPath());
+    ASSERT_NOK_WITH_MSG(fs.Open("s3://bucket/dir"), "is a directory");
+}
+
 TEST(ObjectStoreFileSystemTest, TestHeadErrorIsNotMasked) {
     auto client = std::make_shared<MockObjectStoreClient>();
     client->head_error_ = Status::IOError("access denied");
     client->objects_["foo/bar"] = "child";
     ObjectStoreFileSystem fs("s3", client);
-    auto status = fs.GetFileStatus("s3://bucket/foo");
-    ASSERT_TRUE(status.status().IsIOError());
+    ASSERT_NOK_WITH_MSG(fs.GetFileStatus("s3://bucket/foo"), "access denied");
     ASSERT_EQ(client->list_calls_, 0);
 }
 
@@ -148,6 +167,14 @@ TEST(ObjectStoreFileSystemTest, TestPaginationAndDirectoryMarker) {
     ASSERT_EQ(statuses.size(), 2);
     ASSERT_EQ(statuses[0].GetPath(), "s3://bucket/dir/a");
     ASSERT_TRUE(statuses[1].IsDir());
+
+    std::vector<BasicFileStatus> basic_statuses;
+    ASSERT_OK(fs.ListDir("s3://bucket/dir/", &basic_statuses));
+    ASSERT_EQ(2u, basic_statuses.size());
+    ASSERT_EQ("s3://bucket/dir/a", basic_statuses[0].GetPath());
+    ASSERT_FALSE(basic_statuses[0].IsDir());
+    ASSERT_EQ("s3://bucket/dir/sub/", basic_statuses[1].GetPath());
+    ASSERT_TRUE(basic_statuses[1].IsDir());
 }
 
 TEST(ObjectStoreFileSystemTest, TestTruncatedPageRequiresContinuationToken) {
@@ -158,14 +185,15 @@ TEST(ObjectStoreFileSystemTest, TestTruncatedPageRequiresContinuationToken) {
     client->pages_[""] = page;
     ObjectStoreFileSystem fs("s3", client);
     std::vector<FileStatus> statuses;
-    ASSERT_TRUE(fs.ListFileStatus("s3://bucket/dir/", &statuses).IsIOError());
+    ASSERT_NOK_WITH_MSG(fs.ListFileStatus("s3://bucket/dir/", &statuses),
+                        "did not include a continuation token");
     ASSERT_TRUE(statuses.empty());
 }
 
 TEST(ObjectStoreFileSystemTest, TestOpenBucketRootIsDirectory) {
     auto client = std::make_shared<MockObjectStoreClient>();
     ObjectStoreFileSystem fs("s3", client);
-    ASSERT_TRUE(fs.Open("s3://bucket/").status().IsInvalid());
+    ASSERT_NOK_WITH_MSG(fs.Open("s3://bucket/"), "is a directory");
     ASSERT_EQ(client->head_calls_, 0);
     ASSERT_EQ(client->list_calls_, 0);
 }
@@ -187,6 +215,32 @@ TEST(ObjectStoreFileSystemTest, TestPathWithLeadingSlashes) {
     ASSERT_EQ(status.GetPath(), "s3://bucket/file");
 }
 
+TEST(ObjectStoreInputStreamTest, TestSeekAndEmptyRead) {
+    auto client = std::make_shared<MockObjectStoreClient>();
+    client->objects_["file"] = "data";
+    ObjectStoreFileSystem fs("s3", client);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> stream, fs.Open("s3://bucket/file"));
+    ASSERT_OK_AND_ASSIGN(std::string uri, stream->GetUri());
+    ASSERT_EQ("s3://bucket/file", uri);
+    ASSERT_OK(stream->Seek(1, FS_SEEK_SET));
+    ASSERT_OK(stream->Seek(1, FS_SEEK_CUR));
+    std::array<char, 2> buffer{};
+    ASSERT_OK_AND_ASSIGN(int64_t bytes_read, stream->Read(buffer.data(), buffer.size()));
+    ASSERT_EQ(2, bytes_read);
+    ASSERT_EQ("ta", std::string(buffer.data(), buffer.size()));
+    ASSERT_OK_AND_ASSIGN(int64_t position, stream->GetPos());
+    ASSERT_EQ(4, position);
+    ASSERT_OK_AND_ASSIGN(int64_t eof_read, stream->Read(buffer.data(), buffer.size()));
+    ASSERT_EQ(0, eof_read);
+    bool called = false;
+    stream->ReadAsync(nullptr, 0, position, [&called](Status status) {
+        called = true;
+        ASSERT_OK(status);
+    });
+    ASSERT_TRUE(called);
+    ASSERT_EQ(1u, client->ranges_.size());
+}
+
 TEST(ObjectStoreInputStreamTest, TestBoundsCloseAndSeekOverflow) {
     auto client = std::make_shared<MockObjectStoreClient>();
     client->objects_["file"] = std::string(128 * 1024, 'x');
@@ -196,16 +250,17 @@ TEST(ObjectStoreInputStreamTest, TestBoundsCloseAndSeekOverflow) {
     bool called = false;
     stream->ReadAsync(data, 4, 128 * 1024 - 2, [&called](Status status) {
         called = true;
-        ASSERT_TRUE(status.IsInvalid());
+        ASSERT_NOK_WITH_MSG(status, "exceeds length");
     });
     ASSERT_TRUE(called);
-    ASSERT_TRUE(stream->Seek(std::numeric_limits<int64_t>::max(), FS_SEEK_END).IsInvalid());
+    ASSERT_NOK_WITH_MSG(stream->Seek(std::numeric_limits<int64_t>::max(), FS_SEEK_END),
+                        "object store input position overflows");
     ASSERT_OK(stream->Close());
     called = false;
     stream->ReadAsync(data, 1, 0, [&called, &stream](Status status) {
         called = true;
-        ASSERT_TRUE(status.IsIOError());
-        ASSERT_TRUE(stream->GetPos().status().IsIOError());
+        ASSERT_NOK_WITH_MSG(status, "is closed");
+        ASSERT_NOK_WITH_MSG(stream->GetPos(), "is closed");
     });
     ASSERT_TRUE(called);
 }
@@ -217,7 +272,8 @@ TEST(ObjectStoreInputStreamTest, TestShortReadFails) {
     ObjectStoreFileSystem fs("s3", client);
     ASSERT_OK_AND_ASSIGN(auto stream, fs.Open("s3://bucket/file"));
     char data[4];
-    ASSERT_TRUE(stream->Read(data, sizeof(data)).status().IsIOError());
+    ASSERT_NOK_WITH_MSG(stream->Read(data, sizeof(data)),
+                        "range read for s3://bucket/file returned");
 }
 
 TEST(ObjectStoreInputStreamTest, TestSequentialReadAheadGrows) {
diff --git a/src/paimon/common/metrics/histogram_windowing.h b/src/paimon/common/metrics/histogram_windowing.h
index 658c38d39..27a36b585 100644
--- a/src/paimon/common/metrics/histogram_windowing.h
+++ b/src/paimon/common/metrics/histogram_windowing.h
@@ -36,6 +36,8 @@ namespace paimon {
 // Inspired by RocksDB's monitoring/histogram_windowing.h.
 // https://github.com/facebook/rocksdb/blob/main/monitoring/histogram_windowing.h
 // Maintains multiple time windows and aggregates stats across the recent ones.
+//
+// TODO(jinli.zjw): Refactor this implementation in the future.
 class HistogramWindowingImpl : public Histogram {
  public:
     // Default: 3 windows, 60s per window , 100 per window (paimon-java default)
diff --git a/src/paimon/common/metrics/histogram_windowing_test.cpp b/src/paimon/common/metrics/histogram_windowing_test.cpp
index 624f53d75..915b1518d 100644
--- a/src/paimon/common/metrics/histogram_windowing_test.cpp
+++ b/src/paimon/common/metrics/histogram_windowing_test.cpp
@@ -35,7 +35,7 @@ class HistogramWindowingImplTest : public ::testing::Test {
 };
 
 TEST_F(HistogramWindowingImplTest, TestAdvanceAndAggregateAcrossWindows) {
-    auto h = CreateHistogram(/*num_windows=*/4, /*micros_per_window=*/1000,
+    auto h = CreateHistogram(/*num_windows=*/2, /*micros_per_window=*/1000,
                              /*min_num_per_window=*/1);
     h.Add(1);
     now_micros_ += 1000;
@@ -46,25 +46,37 @@ TEST_F(HistogramWindowingImplTest, TestAdvanceAndAggregateAcrossWindows) {
     EXPECT_DOUBLE_EQ(s.min, 1);
     EXPECT_DOUBLE_EQ(s.max, 2);
     EXPECT_NEAR(s.average, 1.5, 1e-12);
-}
 
-TEST_F(HistogramWindowingImplTest, TestResetWhenBeyondMaxSpan) {
-    auto h = CreateHistogram(/*num_windows=*/2, /*micros_per_window=*/1000,
-                             /*min_num_per_window=*/1);
-    h.Add(1);
     now_micros_ += 1000;
-    h.Add(2);
-    ASSERT_EQ(h.GetStats().count, 2);
-
-    // Move beyond max_span relative to the current window start.
-    now_micros_ += 3000;
     h.Add(3);
+    ASSERT_EQ(h.GetStats().count, 2);
+    ASSERT_DOUBLE_EQ(h.GetStats().sum, 5);
+}
 
-    HistogramStats s = h.GetStats();
-    EXPECT_EQ(s.count, 1);
-    EXPECT_DOUBLE_EQ(s.min, 3);
-    EXPECT_DOUBLE_EQ(s.max, 3);
-    EXPECT_NEAR(s.average, 3.0, 1e-12);
+TEST_F(HistogramWindowingImplTest, TestResetAtOrBeyondMaxSpan) {
+    for (const uint64_t elapsed_micros : {2000, 3000}) {
+        SCOPED_TRACE(elapsed_micros);
+        auto h = CreateHistogram(/*num_windows=*/2, /*micros_per_window=*/1000,
+                                 /*min_num_per_window=*/1);
+        h.Add(1);
+        now_micros_ += 1000;
+        h.Add(2);
+        ASSERT_EQ(h.GetStats().count, 2);
+
+        // Reach or exceed max_span (2ms) after advancing to the second window.
+        now_micros_ += elapsed_micros;
+        h.Add(3);
+
+        HistogramStats s = h.GetStats();
+        EXPECT_EQ(s.count, 1);
+        EXPECT_DOUBLE_EQ(s.min, 3);
+        EXPECT_DOUBLE_EQ(s.max, 3);
+        EXPECT_NEAR(s.average, 3.0, 1e-12);
+
+        // The new sample expires relative to the reset time.
+        now_micros_ += 1000;
+        ASSERT_EQ(h.GetStats().count, 1);
+    }
 }
 
 TEST_F(HistogramWindowingImplTest, TestExpiresAtMaxSpan) {
@@ -84,20 +96,25 @@ TEST_F(HistogramWindowingImplTest, TestExpiresAtMaxSpan) {
 }
 
 TEST_F(HistogramWindowingImplTest, TestMergeSameParams) {
-    auto h1 = CreateHistogram(/*num_windows=*/4, /*micros_per_window=*/1000,
-                              /*min_num_per_window=*/1);
-    auto h2 = CreateHistogram(/*num_windows=*/4, /*micros_per_window=*/1000,
-                              /*min_num_per_window=*/1);
-    h1.Add(1);
-    h1.Add(2);
-    h2.Add(3);
+    auto source = CreateHistogram(/*num_windows=*/4, /*micros_per_window=*/1000,
+                                  /*min_num_per_window=*/1);
+    source.Add(3);
+    now_micros_ += 1000;
+    auto target = CreateHistogram(/*num_windows=*/4, /*micros_per_window=*/1000,
+                                  /*min_num_per_window=*/1);
+    target.Add(1);
+    target.Add(2);
 
-    h1.Merge(h2);
-    HistogramStats s = h1.GetStats();
+    target.Merge(source);
+    HistogramStats s = target.GetStats();
     EXPECT_EQ(s.count, 3);
     EXPECT_DOUBLE_EQ(s.sum, 6);
     EXPECT_DOUBLE_EQ(s.min, 1);
     EXPECT_DOUBLE_EQ(s.max, 3);
+
+    now_micros_ += 3000;
+    ASSERT_EQ(target.GetStats().count, 2);
+    ASSERT_DOUBLE_EQ(target.GetStats().sum, 3);
 }
 
 TEST_F(HistogramWindowingImplTest, TestMergeDifferentParamsFallback) {
@@ -131,6 +148,63 @@ TEST_F(HistogramWindowingImplTest, TestMergePlainHistogram) {
     EXPECT_DOUBLE_EQ(s.max, 20);
 }
 
+TEST_F(HistogramWindowingImplTest, TestClockRollback) {
+    auto h = CreateHistogram(/*num_windows=*/3, /*micros_per_window=*/1000,
+                             /*min_num_per_window=*/1);
+    h.Add(1);
+    HistogramImpl plain;
+    plain.Add(2);
+    auto same = CreateHistogram(/*num_windows=*/3, /*micros_per_window=*/1000,
+                                /*min_num_per_window=*/1);
+    same.Add(3);
+    auto different = CreateHistogram(/*num_windows=*/4, /*micros_per_window=*/1000,
+                                     /*min_num_per_window=*/1);
+    different.Add(4);
+
+    now_micros_ -= 1000;
+    h.Add(5);
+    h.Merge(plain);
+    h.Merge(same);
+    h.Merge(different);
+    now_micros_ += 1000;
+    ASSERT_EQ(h.GetStats().count, 1);
+    ASSERT_DOUBLE_EQ(h.GetStats().sum, 1);
+    h.Add(6);
+    ASSERT_EQ(h.GetStats().count, 2);
+    ASSERT_DOUBLE_EQ(h.GetStats().sum, 7);
+}
+
+TEST_F(HistogramWindowingImplTest, TestZeroWindows) {
+    auto h = CreateHistogram(/*num_windows=*/0, /*micros_per_window=*/1000,
+                             /*min_num_per_window=*/1);
+    h.Add(1);
+    HistogramImpl source;
+    source.Add(2);
+    h.Merge(source);
+    ASSERT_EQ(h.GetStats().count, 0);
+    ASSERT_DOUBLE_EQ(h.GetStats().sum, 0);
+    auto cloned = h.Clone();
+    ASSERT_TRUE(cloned);
+    cloned->Add(3);
+    ASSERT_EQ(cloned->GetStats().count, 0);
+}
+
+TEST_F(HistogramWindowingImplTest, TestZeroDuration) {
+    auto h = CreateHistogram(/*num_windows=*/3, /*micros_per_window=*/0,
+                             /*min_num_per_window=*/1);
+    h.Add(1);
+    now_micros_ += 1000000;
+    h.Add(2);
+    HistogramImpl source;
+    source.Add(3);
+    h.Merge(source);
+    ASSERT_EQ(h.GetStats().count, 3);
+    ASSERT_DOUBLE_EQ(h.GetStats().sum, 6);
+    now_micros_ += 1000000;
+    ASSERT_EQ(h.GetStats().count, 3);
+    ASSERT_DOUBLE_EQ(h.GetStats().sum, 6);
+}
+
 TEST_F(HistogramWindowingImplTest, TestMinNumPerWindow100Case) {
     // Validate min_num_per_window behavior:
     // - window advancement is gated by sample count
@@ -152,9 +226,14 @@ TEST_F(HistogramWindowingImplTest, TestMinNumPerWindow100Case) {
     EXPECT_DOUBLE_EQ(s1.min, 1);
     EXPECT_DOUBLE_EQ(s1.max, 2);
 
+    h.Add(3);
+    now_micros_ += 75000;
+    ASSERT_EQ(h.GetStats().count, 1);
+    ASSERT_DOUBLE_EQ(h.GetStats().sum, 3);
+
     // Cross enough time so that aligned_now - current_window_start >= max_span
     // (max_span = num_windows * micros_per_window = 150ms here).
-    now_micros_ += 175000;
+    now_micros_ += 100000;
     h.Add(1000);
 
     HistogramStats s2 = h.GetStats();
@@ -180,32 +259,12 @@ TEST_F(HistogramWindowingImplTest, TestLargeDatasetInSingleWindow) {
     EXPECT_NEAR(s.average, (static_cast<double>(n) + 1.0) / 2.0, 1e-12);
 }
 
-TEST_F(HistogramWindowingImplTest, TestCrossMaxSpanAfterAdvance) {
-    // Build multiple windows, then cross max_span so that all previous windows are dropped.
-    auto h = CreateHistogram(/*num_windows=*/2, /*micros_per_window=*/2000,
-                             /*min_num_per_window=*/1);
-
-    h.Add(1);
-    now_micros_ += 2000;
-    h.Add(2);  // should advance to next window.
-    ASSERT_EQ(h.GetStats().count, 2);
-
-    // Reach max_span (4ms) relative to the current window start.
-    now_micros_ += 4000;
-    h.Add(3);  // should reset.
-
-    HistogramStats s = h.GetStats();
-    EXPECT_EQ(s.count, 1);
-    EXPECT_DOUBLE_EQ(s.min, 3);
-    EXPECT_DOUBLE_EQ(s.max, 3);
-    EXPECT_DOUBLE_EQ(s.sum, 3);
-}
-
 TEST_F(HistogramWindowingImplTest, TestCloneConsistencyAndIndependence) {
     auto h = CreateHistogram(/*num_windows=*/4, /*micros_per_window=*/1000,
-                             /*min_num_per_window=*/1);
+                             /*min_num_per_window=*/2);
     h.Add(1);
     h.Add(2);
+    now_micros_ += 1000;
     h.Add(3);
     h.Add(4);
 
@@ -228,6 +287,7 @@ TEST_F(HistogramWindowingImplTest, TestCloneConsistencyAndIndependence) {
     EXPECT_DOUBLE_EQ(cloned_before.p99, before.p99);
     EXPECT_DOUBLE_EQ(cloned_before.p999, before.p999);
 
+    now_micros_ += 1000;
     // Mutating original should not affect cloned.
     h.Add(100);
     const HistogramStats after_original = h.GetStats();
@@ -242,8 +302,14 @@ TEST_F(HistogramWindowingImplTest, TestCloneConsistencyAndIndependence) {
     EXPECT_EQ(after_original2.count, after_original.count);
     EXPECT_EQ(after_clone2.count, cloned_before.count + 1);
 
+    now_micros_ += 3000;
+    ASSERT_EQ(h.GetStats().count, 1);
+    ASSERT_DOUBLE_EQ(h.GetStats().sum, 100);
+    ASSERT_EQ(cloned->GetStats().count, 1);
+    ASSERT_DOUBLE_EQ(cloned->GetStats().sum, 200);
+
     // Clones share the same time source, while keeping independent samples.
-    now_micros_ += 4000;
+    now_micros_ += 1000;
     EXPECT_EQ(h.GetStats().count, 0);
     EXPECT_EQ(cloned->GetStats().count, 0);
 }
diff --git a/src/paimon/core/bucket/hive_bucket_function_test.cpp b/src/paimon/core/bucket/hive_bucket_function_test.cpp
index d97a0294b..43d36da58 100644
--- a/src/paimon/core/bucket/hive_bucket_function_test.cpp
+++ b/src/paimon/core/bucket/hive_bucket_function_test.cpp
@@ -142,6 +142,28 @@ TEST_F(HiveBucketFunctionTest, TestHiveBucketFunction) {
     ASSERT_EQ(1, func->Bucket(row, 8));
 }
 
+TEST_F(HiveBucketFunctionTest, TestDecimalHighBitsCompatibleWithJava) {
+    std::vector<HiveFieldInfo> field_infos = {HiveFieldInfo(FieldType::DECIMAL, 10, 0)};
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<HiveBucketFunction> func,
+                         HiveBucketFunction::Create(field_infos));
+    auto pool = GetDefaultPool();
+    struct TestCase {
+        int64_t unscaled;
+        int32_t expected_bucket;
+    };
+    // Verified with Java's normalized decimal hash and (hash & INT32_MAX) % 1000.
+    const TestCase cases[] = {{1, 31},
+                              {4294967296LL, 961},
+                              {-4294967296LL, 687},
+                              {4294967297LL, 992},
+                              {8589934593LL, 953}};
+    for (const auto& test_case : cases) {
+        BinaryRow row = BinaryRowGenerator::GenerateRow(
+            {Decimal::FromUnscaledLong(test_case.unscaled, 10, 0)}, pool.get());
+        ASSERT_EQ(func->Bucket(row, 1000), test_case.expected_bucket) << test_case.unscaled;
+    }
+}
+
 /// Test matching Java: testHiveBucketFunctionWithNulls
 TEST_F(HiveBucketFunctionTest, TestHiveBucketFunctionWithNulls) {
     std::vector<FieldType> field_types = {FieldType::INT, FieldType::STRING};
diff --git a/src/paimon/core/bucket/hive_hasher.h b/src/paimon/core/bucket/hive_hasher.h
index e9dad0e44..6871848f9 100644
--- a/src/paimon/core/bucket/hive_hasher.h
+++ b/src/paimon/core/bucket/hive_hasher.h
@@ -52,7 +52,8 @@ class HiveHasher {
     }
 
     /// Normalize a Decimal value for Hive-compatible hashing.
-    /// This implements the same logic as HiveHasher.normalizeDecimal in Java.
+    /// This follows HiveHasher.normalizeDecimal in Java, except that integer-precision overflow
+    /// returns 0 here while Java normalization returns null.
     ///
     /// The normalization process:
     /// 1. Strip trailing zeros
@@ -61,12 +62,9 @@ class HiveHasher {
     /// 4. Round if necessary using HALF_UP
     ///
     /// @param decimal The decimal value to normalize.
-    /// @return The hash code of the normalized decimal, computed as Java BigDecimal.hashCode().
+    /// @return Java BigDecimal.hashCode() for the normalized decimal, or 0 if integer precision
+    /// exceeds Hive's 38-digit limit.
     static uint32_t HashDecimal(const Decimal& decimal) {
-        // Java BigDecimal.hashCode() = unscaledValue.intValue() * 31 + scale
-        // For compact decimals (precision <= 18), we can use the long value directly.
-        // For non-compact decimals, we need to handle the 128-bit value.
-
         // First normalize: strip trailing zeros and limit scale
         int32_t scale = decimal.Scale();
         auto value = decimal.Value();
@@ -106,10 +104,12 @@ class HiveHasher {
         int32_t int_digits = total_digits - scale;
 
         if (int_digits > HIVE_DECIMAL_MAX_PRECISION) {
-            // Overflow, return 0 (null equivalent)
+            // Preserve C++'s zero fallback when Java normalization would return null.
             return 0;
         }
 
+        // Valid production decimals never have scale > HIVE_DECIMAL_MAX_PRECISION (38).
+        // The scale limiting and rounding below only defend against unchecked, out-of-range inputs.
         int32_t max_scale = HIVE_DECIMAL_MAX_SCALE;
         if (HIVE_DECIMAL_MAX_PRECISION - int_digits < max_scale) {
             max_scale = HIVE_DECIMAL_MAX_PRECISION - int_digits;
@@ -119,17 +119,19 @@ class HiveHasher {
         }
 
         if (scale > max_scale) {
-            // Need to round: scale down with HALF_UP rounding
+            // Discard lower digits without rounding, then apply HALF_UP once at the
+            // final discarded digit. Rounding each digit can introduce a second carry.
             int32_t scale_diff = scale - max_scale;
-            for (int32_t i = 0; i < scale_diff; i++) {
-                auto quotient = value / 10;
-                auto remainder = value - quotient * 10;
-                if (remainder < 0) remainder = -remainder;
-                if (remainder >= 5) {
-                    value = quotient + (value < 0 ? -1 : 1);
-                } else {
-                    value = quotient;
-                }
+            for (int32_t i = 1; i < scale_diff && value != 0; i++) {
+                value /= 10;
+            }
+            auto quotient = value / 10;
+            auto remainder = value - quotient * 10;
+            if (remainder < 0) remainder = -remainder;
+            if (remainder >= 5) {
+                value = quotient + (value < 0 ? -1 : 1);
+            } else {
+                value = quotient;
             }
             scale = max_scale;
 
@@ -150,10 +152,24 @@ class HiveHasher {
         }
 
         // Compute Java BigDecimal.hashCode():
-        // hashCode = intValue(unscaledValue) * 31 + scale
-        // intValue() returns the low 32 bits of the value
-        auto int_value = static_cast<uint32_t>(value);
-        return int_value * 31U + static_cast<uint32_t>(scale);
+        // 1. Split the absolute unscaled value into 32-bit words, from most to least significant.
+        // 2. Starting at h = 0, fold each word with h = h * 31 + word (BigInteger hash).
+        // 3. Apply the original sign, then return h * 31 + the normalized scale.
+        // All arithmetic wraps modulo 2^32, preserving Java int hash bits.
+        // Leading zero words have no effect. For a 64-bit magnitude, the formula is
+        // sign * (high32 * 31 + low32) * 31 + scale.
+        // Example: unscaled = 2^32, scale = 0 -> (1 * 31 + 0) * 31 = 961.
+        abs_value = value < 0
+                        ? static_cast<unsigned __int128>(0) - static_cast<unsigned __int128>(value)
+                        : static_cast<unsigned __int128>(value);
+        uint32_t unscaled_hash = 0;
+        for (int32_t i = 3; i >= 0; i--) {
+            unscaled_hash = unscaled_hash * 31U + static_cast<uint32_t>(abs_value >> (i * 32));
+        }
+        if (value < 0) {
+            unscaled_hash = 0U - unscaled_hash;
+        }
+        return unscaled_hash * 31U + static_cast<uint32_t>(scale);
     }
 
  private:
diff --git a/src/paimon/core/bucket/hive_hasher_test.cpp b/src/paimon/core/bucket/hive_hasher_test.cpp
new file mode 100644
index 000000000..c5fde5214
--- /dev/null
+++ b/src/paimon/core/bucket/hive_hasher_test.cpp
@@ -0,0 +1,137 @@
+/*
+ * 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/bucket/hive_hasher.h"
+
+#include <cstdint>
+
+#include "gtest/gtest.h"
+#include "paimon/common/utils/decimal_utils.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(HiveHasherTest, TestHashIntPreservesBits) {
+    for (uint32_t value : {0U, 1U, 0x7fffffffU, 0x80000000U, 0xffffffffU}) {
+        ASSERT_EQ(HiveHasher::HashInt(value), value);
+    }
+}
+
+TEST(HiveHasherTest, TestHashLongCombinesBothHalves) {
+    struct TestCase {
+        uint64_t value;
+        uint32_t expected;
+    };
+    const TestCase cases[] = {{0, 0},
+                              {1, 1},
+                              {0x100000000ULL, 1},
+                              {0x0123456789abcdefULL, 0x88888888U},
+                              {0x7fffffffffffffffULL, 0x80000000U},
+                              {0x8000000000000000ULL, 0x80000000U},
+                              {0xffffffffffffffffULL, 0}};
+    for (const auto& test_case : cases) {
+        ASSERT_EQ(HiveHasher::HashLong(test_case.value), test_case.expected) << test_case.value;
+    }
+}
+
+TEST(HiveHasherTest, TestHashBytesUsesSignedBytesAndExplicitLength) {
+    ASSERT_EQ(HiveHasher::HashBytes("", 0), 0U);
+    ASSERT_EQ(HiveHasher::HashBytes("hello", 5), 99162322U);
+    ASSERT_EQ(HiveHasher::HashBytes("a\0b", 3), 93315U);
+    ASSERT_EQ(HiveHasher::HashBytes("a\0b", 1), 97U);
+    ASSERT_EQ(HiveHasher::HashBytes("\x7f", 1), 127U);
+    ASSERT_EQ(HiveHasher::HashBytes("\x80", 1), 4294967168U);
+    ASSERT_EQ(HiveHasher::HashBytes("\xff", 1), 4294967295U);
+    ASSERT_EQ(HiveHasher::HashBytes("\x7f\x80\xff\x00\x01", 5), 113472959U);
+    ASSERT_EQ(HiveHasher::HashBytes("\xff\x80\x7f\x00", 4), 4294818434U);
+}
+
+TEST(HiveHasherTest, TestHashZeroDecimalIgnoresScale) {
+    for (int32_t scale : {0, 1, 18, 38}) {
+        ASSERT_EQ(HiveHasher::HashDecimal(Decimal(38, scale, 0)), 0U) << scale;
+    }
+}
+
+TEST(HiveHasherTest, TestHashDecimalNormalizesTrailingZeros) {
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(10, 2, 1234)), 38256U);
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(10, 4, 123400)), 38256U);
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(10, 2, -1234)), 4294929044U);
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(10, 4, -123400)), 4294929044U);
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(10, 0, 1000)), 31000U);
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(10, 3, 1000000)), 31000U);
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(38, 38, 1)), 69U);
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(38, 38, -1)), 7U);
+}
+
+TEST(HiveHasherTest, TestHashNonCompactDecimalNormalizesTrailingZeros) {
+    ASSERT_OK_AND_ASSIGN(Decimal::int128_t unscaled,
+                         DecimalUtils::StrToInt128("12340000000000000000000000000000000000"));
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(38, 36, unscaled)), 38256U);
+    ASSERT_EQ(HiveHasher::HashDecimal(Decimal(38, 36, -unscaled)), 4294929044U);
+}
+
+TEST(HiveHasherTest, TestHashDecimalIncludesAllMagnitudeWords) {
+    struct TestCase {
+        const char* unscaled;
+        int32_t scale;
+        uint32_t expected;
+    };
+    // Verified with Java normalizeDecimal(...).hashCode(), including values with identical
+    // low 32 bits and values outside the signed long range.
+    const TestCase cases[] = {{"1", 0, 31U},
+                              {"4294967296", 0, 961U},
+                              {"-4294967296", 0, 4294966335U},
+                              {"4294967297", 0, 992U},
+                              {"8589934593", 0, 1953U},
+                              {"4294967296", 2, 963U},
+                              {"-4294967296", 2, 4294966337U},
+                              {"9223372036854775807", 0, 2147482656U},
+                              {"-9223372036854775808", 0, 2147483648U},
+                              {"9223372036854775808", 0, 2147483648U},
+                              {"18446744073709551616", 0, 29791U},
+                              {"-18446744073709551616", 0, 4294937505U},
+                              {"79228162514264337593543950336", 0, 923521U},
+                              {"-79228162514264337593543950336", 0, 4294043775U},
+                              {"99999999999999999999999999999999999999", 0, 3867768910U},
+                              {"-99999999999999999999999999999999999999", 0, 427198386U}};
+    for (const auto& test_case : cases) {
+        SCOPED_TRACE(test_case.unscaled);
+        SCOPED_TRACE(test_case.scale);
+        ASSERT_OK_AND_ASSIGN(Decimal::int128_t unscaled,
+                             DecimalUtils::StrToInt128(test_case.unscaled));
+        ASSERT_EQ(HiveHasher::HashDecimal(Decimal(38, test_case.scale, unscaled)),
+                  test_case.expected);
+    }
+}
+
+TEST(HiveHasherTest, TestHashDecimalRejectsIntegerPrecisionOverflow) {
+    // Decimal's raw constructor permits values beyond Hive's 38-digit limit.
+    // Java normalizeDecimal returns null here. Its bucket function then calls hashCode()
+    // without a null check and throws NullPointerException. These assertions document C++'s
+    // existing zero fallback, rather than matching Java hash values.
+    for (const char* value :
+         {"100000000000000000000000000000000000000", "-100000000000000000000000000000000000000",
+          "170141183460469231731687303715884105727", "-170141183460469231731687303715884105728"}) {
+        SCOPED_TRACE(value);
+        ASSERT_OK_AND_ASSIGN(Decimal::int128_t unscaled, DecimalUtils::StrToInt128(value));
+        ASSERT_EQ(HiveHasher::HashDecimal(Decimal(39, 0, unscaled)), 0U);
+    }
+}
+
+}  // namespace paimon::test
diff --git a/src/paimon/core/mergetree/merge_tree_writer_test.cpp b/src/paimon/core/mergetree/merge_tree_writer_test.cpp
index e557af325..b0bdce126 100644
--- a/src/paimon/core/mergetree/merge_tree_writer_test.cpp
+++ b/src/paimon/core/mergetree/merge_tree_writer_test.cpp
@@ -812,14 +812,15 @@ TEST_P(MergeTreeWriterTest, TestSortedReaderFailure) {
     ASSERT_TRUE(first_mixed_reader_closed);
     ASSERT_TRUE(second_mixed_reader_closed);
 
-    Status expected_status = Status::IOError("sorted reader failure");
     bool failing_reader_closed = false;
     std::vector<std::unique_ptr<KeyValueRecordReader>> failing_readers;
     failing_readers.push_back(std::make_unique<TrackingKeyValueRecordReader>(
-        CreateSingleReader(sorted_reader_array, /*batch_size=*/16, expected_status),
+        CreateSingleReader(sorted_reader_array, /*batch_size=*/16,
+                           Status::IOError("sorted reader failure")),
         &failing_reader_closed));
     Status failing_status = merge_writer->WriteSortedReadersToFiles(std::move(failing_readers));
-    ASSERT_EQ(expected_status, failing_status);
+    ASSERT_TRUE(failing_status.IsIOError());
+    ASSERT_NOK_WITH_MSG(failing_status, "sorted reader failure");
     ASSERT_TRUE(failing_reader_closed);
     ASSERT_OK(merge_writer->Close());
 }
diff --git a/src/paimon/format/parquet/file_reader_wrapper_test.cpp b/src/paimon/format/parquet/file_reader_wrapper_test.cpp
index 74b6bb8e1..e1ad5ac39 100644
--- a/src/paimon/format/parquet/file_reader_wrapper_test.cpp
+++ b/src/paimon/format/parquet/file_reader_wrapper_test.cpp
@@ -23,6 +23,7 @@
 #include <iterator>
 #include <map>
 #include <mutex>
+#include <stdexcept>
 #include <string>
 #include <vector>
 
@@ -65,8 +66,9 @@ namespace paimon::parquet::test {
 // Tracks positional reads (Read at offset / ReadAsync) issued through the stream.
 class ReadTrackingInputStream : public InputStream {
  public:
-    explicit ReadTrackingInputStream(std::shared_ptr<InputStream> input)
-        : input_(std::move(input)) {}
+    explicit ReadTrackingInputStream(std::shared_ptr<InputStream> input,
+                                     bool fail_first_async_read = false)
+        : input_(std::move(input)), fail_first_async_read_(fail_first_async_read) {}
 
     Status Seek(int64_t offset, SeekOrigin origin) override {
         return input_->Seek(offset, origin);
@@ -88,6 +90,13 @@ class ReadTrackingInputStream : public InputStream {
     void ReadAsync(char* buffer, int64_t size, int64_t offset,
                    std::function<void(Status)>&& callback) override {
         RecordPositionalRead(offset, size);
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            if (++async_read_calls_ == 1 && fail_first_async_read_) {
+                // Exercise the synchronous exception path in Parquet prebuffering.
+                throw std::runtime_error("injected prebuffer read failure");
+            }
+        }
         input_->ReadAsync(buffer, size, offset, std::move(callback));
     }
 
@@ -108,6 +117,11 @@ class ReadTrackingInputStream : public InputStream {
         return positional_read_bytes_;
     }
 
+    int64_t GetAsyncReadCalls() const {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return async_read_calls_;
+    }
+
  private:
     void RecordPositionalRead(int64_t offset, int64_t size) {
         (void)offset;
@@ -118,6 +132,8 @@ class ReadTrackingInputStream : public InputStream {
     std::shared_ptr<InputStream> input_;
     mutable std::mutex mutex_;
     int64_t positional_read_bytes_ = 0;
+    int64_t async_read_calls_ = 0;
+    bool fail_first_async_read_;
 };
 
 class FileReaderWrapperTest : public ::testing::Test {
@@ -295,6 +311,10 @@ TEST_F(FileReaderWrapperTest, PredicateReadsOnlyItsPageIndexesAndKeepsPayloadRea
     ASSERT_TRUE(index_ranges.column_index.has_value());
     ASSERT_TRUE(index_ranges.offset_index.has_value());
     int64_t bytes_before = tracking_input->GetPositionalReadBytes();
+    ASSERT_OK_AND_ASSIGN(auto all_ranges, reader->CalculateFilteredRowRanges(0, nullptr, {}));
+    ASSERT_EQ((std::vector<RowRanges::Range>{{0, 999}}), all_ranges.GetRanges());
+    ASSERT_EQ(bytes_before, tracking_input->GetPositionalReadBytes());
+
     auto predicate = PredicateBuilder::Equal(1, "col2", FieldType::INT, Literal(25));
     ASSERT_OK_AND_ASSIGN(auto ranges, reader->CalculateFilteredRowRanges(
                                           0, predicate, {{"col1", 0}, {"col2", 1}, {"col3", 2}}));
@@ -316,6 +336,30 @@ TEST_F(FileReaderWrapperTest, PredicateReadsOnlyItsPageIndexesAndKeepsPayloadRea
     ASSERT_FALSE(end);
 }
 
+TEST_F(FileReaderWrapperTest, PreBufferExceptionFallsBackToReadableData) {
+    std::string file_path = PathUtil::JoinPath(dir_->Str(), "prebuffer-failure.parquet");
+    PrepareParquetFile(file_path, /*row_count=*/200, /*enable_page_index=*/true);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input, fs_->Open(file_path));
+    auto failing_input =
+        std::make_shared<ReadTrackingInputStream>(input, /*fail_first_async_read=*/true);
+    ASSERT_OK_AND_ASSIGN(auto reader, PrepareReaderWrapperOnStream(failing_input));
+    ASSERT_EQ(0, failing_input->GetAsyncReadCalls());
+    RowRanges ranges(RowRanges::Range(20, 29));
+    ASSERT_OK(reader->PrepareForReading(
+        {TargetRowGroup(/*rg_index=*/0, /*is_partially_matched=*/true, ranges)}, {0, 1, 2}));
+    ASSERT_EQ(1, failing_input->GetAsyncReadCalls());
+
+    ASSERT_OK_AND_ASSIGN(auto batch, reader->Next());
+    ASSERT_TRUE(batch);
+    auto expected = PrepareArray(PrepareArrowSchema().second, /*record_batch_size=*/10,
+                                 /*offset=*/20);
+    auto expected_batch = arrow::RecordBatch::FromStructArray(expected);
+    ASSERT_TRUE(expected_batch.ok());
+    ASSERT_TRUE(batch->Equals(*expected_batch.ValueOrDie()));
+    ASSERT_OK_AND_ASSIGN(auto end, reader->Next());
+    ASSERT_FALSE(end);
+}
+
 TEST_F(FileReaderWrapperTest, ReusesParsedOffsetIndexesWithinFileReader) {
     std::string file_path = PathUtil::JoinPath(dir_->Str(), "parsed-index.parquet");
     PrepareParquetFile(file_path, /*row_count=*/2000, /*enable_page_index=*/true);
diff --git a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
index 4f80f3662..703541d19 100644
--- a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
+++ b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
@@ -200,14 +200,19 @@ class ParquetFileBatchReaderTest : public ::testing::Test,
     void WriteArray(const std::string& file_path, const std::shared_ptr<arrow::Array>& src_array,
                     const std::shared_ptr<arrow::Schema>& arrow_schema, int64_t write_batch_size,
                     bool enable_dictionary, int64_t max_row_group_length,
-                    int64_t max_page_size = 1024 * 1024 * 1024) const {
+                    int64_t max_page_size = 1024 * 1024 * 1024,
+                    bool enable_page_index = true) const {
         ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> out,
                              fs_->Create(file_path, /*overwrite=*/true));
         ::parquet::WriterProperties::Builder builder;
         builder.write_batch_size(write_batch_size);
         builder.max_row_group_length(max_row_group_length);
         builder.data_pagesize(max_page_size);
-        builder.enable_write_page_index();
+        if (enable_page_index) {
+            builder.enable_write_page_index();
+        } else {
+            builder.disable_write_page_index();
+        }
         enable_dictionary ? builder.enable_dictionary() : builder.disable_dictionary();
         auto writer_properties = builder.build();
         ASSERT_OK_AND_ASSIGN(auto format_writer, ParquetFormatWriter::Create(
@@ -1351,6 +1356,43 @@ TEST_F(ParquetFileBatchReaderTest, TestBitmapPagePushDownWithMultiRowGroups) {
     ASSERT_TRUE(result_array->Equals(expected_array)) << result_array->ToString();
 }
 
+TEST_F(ParquetFileBatchReaderTest, TestBitmapTrimWithoutPageIndexFallsBackToRowGroup) {
+    auto src_array = MakeSequentialIntData(8);
+    auto arrow_schema = arrow::schema(src_array->type()->fields());
+    WriteArray(file_path_, src_array, arrow_schema, /*write_batch_size=*/1,
+               /*enable_dictionary=*/false, /*max_row_group_length=*/4, /*max_page_size=*/1,
+               /*enable_page_index=*/false);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input, fs_->Open(file_path_));
+    ASSERT_OK_AND_ASSIGN(int64_t file_length, input->Length());
+    auto stream = std::make_unique<ArrowInputStreamAdapter>(std::move(input), file_length, pool_);
+    std::map<std::string, std::string> options = {
+        {PARQUET_READ_ENABLE_PAGE_INDEX_FILTER, "true"},
+        {PARQUET_READ_BITMAP_ROW_RANGE_REFINING_STRATEGY, "trim"}};
+    auto reader = PrepareParquetFileBatchReader(std::move(stream), options, arrow_schema,
+                                                /*predicate=*/nullptr, RoaringBitmap32::From({1}),
+                                                /*batch_size=*/8);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+                         paimon::test::ReadResultCollector::CollectResult(std::move(reader)));
+    ASSERT_TRUE(result);
+    // Without offset indexes, trim must retain the entire candidate row group.
+    ASSERT_TRUE(result->Equals(arrow::ChunkedArray(src_array->Slice(0, 4)))) << result->ToString();
+}
+
+TEST_F(ParquetFileBatchReaderTest, TestPredicateAndBitmapPageRangesDoNotOverlap) {
+    auto src_array = MakeSequentialIntData(4);
+    auto arrow_schema = arrow::schema(src_array->type()->fields());
+    WriteArray(file_path_, src_array, arrow_schema, /*write_batch_size=*/1,
+               /*enable_dictionary=*/false, /*max_row_group_length=*/4, /*max_page_size=*/1);
+    auto predicate =
+        PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::INT, Literal(3));
+    auto reader = PrepareParquetFileBatchReader(file_path_, arrow_schema, predicate,
+                                                RoaringBitmap32::From({0}), /*batch_size=*/4,
+                                                /*enable_page_level_filter=*/true);
+    // Both filters retain the row group, but their page ranges have no intersection.
+    ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch, reader->NextBatch());
+    ASSERT_TRUE(BatchReader::IsEofBatch(batch));
+}
+
 TEST_F(ParquetFileBatchReaderTest, TestBitmapPruningPrecedesPredicatePushDown) {
     class CountingPredicate : public LeafPredicate {
      public:
diff --git a/src/paimon/rest/rest_messages_test.cpp b/src/paimon/rest/rest_messages_test.cpp
index 627b394d0..9aba13bb9 100644
--- a/src/paimon/rest/rest_messages_test.cpp
+++ b/src/paimon/rest/rest_messages_test.cpp
@@ -21,6 +21,7 @@
 #include <map>
 #include <optional>
 #include <string>
+#include <vector>
 
 #include "gtest/gtest.h"
 #include "paimon/common/utils/rapidjson_util.h"
@@ -73,6 +74,18 @@ TEST(RestMessagesTest, ConfigResponseMerge) {
     ASSERT_EQ(0, merged.count("unset-absent"));
 }
 
+TEST(RestMessagesTest, ConfigResponseMissingOrNullMaps) {
+    const std::vector<std::string> inputs = {"{}", R"({"defaults": null, "overrides": null})"};
+    const std::map<std::string, std::string> client = {{"client", "client-value"}};
+    for (const auto& json : inputs) {
+        SCOPED_TRACE(json);
+        ASSERT_OK_AND_ASSIGN(ConfigResponse config, ConfigResponse::FromJsonString(json));
+        ASSERT_TRUE(config.GetDefaults().empty());
+        ASSERT_TRUE(config.GetOverrides().empty());
+        ASSERT_EQ(client, config.Merge(client));
+    }
+}
+
 TEST(RestMessagesTest, ListResponsesParse) {
     ASSERT_OK_AND_ASSIGN(ListDatabasesResponse databases,
                          ListDatabasesResponse::FromJsonString(
@@ -92,7 +105,7 @@ TEST(RestMessagesTest, ListResponsesParse) {
     ASSERT_FALSE(null_token.NextPageToken().has_value());
 }
 
-TEST(RestMessagesTest, ListSnapshotsResponseParse) {
+TEST(RestMessagesTest, ListSnapshotsResponseRoundTrip) {
     std::string json = R"({
         "snapshots": [{
             "version": 3,
@@ -137,9 +150,15 @@ TEST(RestMessagesTest, ListSnapshotsResponseParse) {
     SnapshotInfo compact_info = response.Data()[1].ToSnapshotInfo();
     ASSERT_EQ(SnapshotInfo::CommitKind::COMPACT, compact_info.commit_kind);
     ASSERT_FALSE(compact_info.watermark.has_value());
+
+    ASSERT_OK_AND_ASSIGN(std::string written, response.ToJsonString());
+    ASSERT_OK_AND_ASSIGN(ListSnapshotsResponse round_tripped,
+                         ListSnapshotsResponse::FromJsonString(written));
+    ASSERT_EQ(response.Data(), round_tripped.Data());
+    ASSERT_EQ(response.NextPageToken(), round_tripped.NextPageToken());
 }
 
-TEST(RestMessagesTest, GetDatabaseResponseParse) {
+TEST(RestMessagesTest, GetDatabaseResponseRoundTrip) {
     std::string json = R"({
         "id": "10",
         "name": "db1",
@@ -160,9 +179,18 @@ TEST(RestMessagesTest, GetDatabaseResponseParse) {
     ASSERT_EQ("owner1", options.at("owner"));
     ASSERT_EQ("100", options.at("createdAt"));
     ASSERT_EQ("updater", options.at("updatedBy"));
+
+    ASSERT_OK_AND_ASSIGN(std::string written, response.ToJsonString());
+    rapidjson::Document expected;
+    expected.Parse(json.c_str());
+    ASSERT_FALSE(expected.HasParseError());
+    rapidjson::Document doc;
+    doc.Parse(written.c_str());
+    ASSERT_FALSE(doc.HasParseError());
+    ASSERT_TRUE(expected == doc) << written;
 }
 
-TEST(RestMessagesTest, GetTableResponseParse) {
+TEST(RestMessagesTest, GetTableResponseRoundTrip) {
     std::string json = R"({
         "id": "42",
         "database": "db1",
@@ -195,6 +223,15 @@ TEST(RestMessagesTest, GetTableResponseParse) {
     ASSERT_FALSE(schema.HasParseError());
     ASSERT_EQ(2u, schema["fields"].Size());
     ASSERT_STREQ("f0", schema["fields"][0]["name"].GetString());
+
+    ASSERT_OK_AND_ASSIGN(std::string written, response.ToJsonString());
+    rapidjson::Document expected;
+    expected.Parse(json.c_str());
+    ASSERT_FALSE(expected.HasParseError());
+    rapidjson::Document doc;
+    doc.Parse(written.c_str());
+    ASSERT_FALSE(doc.HasParseError());
+    ASSERT_TRUE(expected == doc) << written;
 }
 
 TEST(RestMessagesTest, GetTableTokenResponseParse) {