fix: align Hive decimal hashing and extend unit coverage

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

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

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>{});

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) {

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)