Skip to content

Commit becc48a

Browse files
zjw1111zjw1111
authored
fix: align Hive decimal hashing and extend unit coverage (#510)
1 parent eb0dd42 commit becc48a

13 files changed

Lines changed: 728 additions & 89 deletions

‎src/paimon/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -613,6 +613,7 @@ if(PAIMON_BUILD_TESTS)
613613
common/data/columnar/columnar_utils_test.cpp
614614
common/data/data_define_test.cpp
615615
common/data/decimal_test.cpp
616+
common/data/generic_array_test.cpp
616617
common/data/generic_row_test.cpp
617618
common/data/internal_row_test.cpp
618619
common/data/record_batch_test.cpp
@@ -829,6 +830,7 @@ if(PAIMON_BUILD_TESTS)
829830
core/bucket/bucket_select_converter_test.cpp
830831
core/bucket/default_bucket_function_test.cpp
831832
core/bucket/hive_bucket_function_test.cpp
833+
core/bucket/hive_hasher_test.cpp
832834
core/bucket/bucket_id_calculator_test.cpp
833835
core/bucket/mod_bucket_function_test.cpp
834836
core/casting/cast_executor_factory_test.cpp

‎src/paimon/common/data/generic_array_test.cpp‎

Lines changed: 208 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,208 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
#include "paimon/common/data/generic_array.h"
21+
22+
#include <cstdint>
23+
#include <limits>
24+
#include <memory>
25+
#include <string>
26+
#include <string_view>
27+
#include <vector>
28+
29+
#include "gtest/gtest.h"
30+
#include "paimon/common/data/binary_string.h"
31+
#include "paimon/common/data/generic_map.h"
32+
#include "paimon/common/data/generic_row.h"
33+
#include "paimon/data/decimal.h"
34+
#include "paimon/data/timestamp.h"
35+
#include "paimon/memory/bytes.h"
36+
#include "paimon/memory/memory_pool.h"
37+
#include "paimon/testing/utils/testharness.h"
38+
39+
namespace paimon::test {
40+
41+
TEST(GenericArrayTest, TestGetters) {
42+
const Decimal decimal(/*precision=*/10, /*scale=*/2, 12345);
43+
const Timestamp timestamp(/*millisecond=*/1234, /*nano_of_millisecond=*/567890);
44+
std::shared_ptr<InternalArray> nested_array =
45+
std::make_shared<GenericArray>(std::vector<VariantType>{int32_t{42}});
46+
std::shared_ptr<InternalMap> map = std::make_shared<GenericMap>(nested_array, nested_array);
47+
auto row = std::make_shared<GenericRow>(1);
48+
row->SetField(0, int32_t{7});
49+
GenericArray array({true, false, char{42}, int16_t{-1234}, int32_t{-56789},
50+
int64_t{12345678900}, -1.5F, 2.25, decimal, timestamp, nested_array, map,
51+
std::shared_ptr<InternalRow>(row), NullType{}});
52+
53+
ASSERT_EQ(array.Size(), 14);
54+
for (int32_t i = 0; i < array.Size() - 1; ++i) {
55+
ASSERT_FALSE(array.IsNullAt(i)) << i;
56+
}
57+
ASSERT_TRUE(array.IsNullAt(13));
58+
ASSERT_TRUE(array.GetBoolean(0));
59+
ASSERT_FALSE(array.GetBoolean(1));
60+
ASSERT_EQ(array.GetByte(2), char{42});
61+
ASSERT_EQ(array.GetShort(3), -1234);
62+
ASSERT_EQ(array.GetInt(4), -56789);
63+
ASSERT_EQ(array.GetDate(4), -56789);
64+
ASSERT_EQ(array.GetLong(5), int64_t{12345678900});
65+
ASSERT_FLOAT_EQ(array.GetFloat(6), -1.5F);
66+
ASSERT_DOUBLE_EQ(array.GetDouble(7), 2.25);
67+
ASSERT_EQ(array.GetDecimal(8, /*precision=*/10, /*scale=*/2), decimal);
68+
ASSERT_EQ(array.GetTimestamp(9, /*precision=*/9), timestamp);
69+
ASSERT_EQ(array.GetArray(10), nested_array);
70+
ASSERT_EQ(array.GetArray(10)->GetInt(0), 42);
71+
ASSERT_EQ(array.GetMap(11), map);
72+
ASSERT_EQ(array.GetMap(11)->ValueArray()->GetInt(0), 42);
73+
ASSERT_EQ(array.GetRow(12, /*num_fields=*/1), row);
74+
ASSERT_EQ(array.GetRow(12, /*num_fields=*/1)->GetInt(0), 7);
75+
}
76+
77+
TEST(GenericArrayTest, TestStringAndBinaryRepresentations) {
78+
auto pool = GetDefaultPool();
79+
for (const std::string& value :
80+
{std::string(), std::string("hello world"), std::string("a\0b", 3)}) {
81+
SCOPED_TRACE(value);
82+
BinaryString string = BinaryString::FromString(value, pool.get());
83+
std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(value, pool.get());
84+
GenericArray array({string, bytes, std::string_view(value)});
85+
86+
ASSERT_EQ(array.Size(), 3);
87+
for (int32_t i = 0; i < array.Size(); ++i) {
88+
ASSERT_FALSE(array.IsNullAt(i));
89+
ASSERT_EQ(array.GetString(i).ToString(), value) << i;
90+
ASSERT_EQ(array.GetStringView(i), value) << i;
91+
auto result = array.GetBinary(i);
92+
ASSERT_TRUE(result);
93+
ASSERT_EQ(*result, *bytes) << i;
94+
}
95+
ASSERT_EQ(array.GetBinary(1), bytes);
96+
ASSERT_EQ(array.GetStringView(2).data(), value.data());
97+
}
98+
}
99+
100+
TEST(GenericArrayTest, TestHoldersKeepBorrowedValuesAlive) {
101+
auto pool = GetDefaultPool();
102+
const std::string string_value = "string owned by the source array";
103+
const std::string binary_value("binary\0value", 12);
104+
std::weak_ptr<InternalArray> source_ref;
105+
std::unique_ptr<GenericArray> array;
106+
{
107+
std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(binary_value, pool.get());
108+
auto source = std::make_shared<GenericArray>(
109+
std::vector<VariantType>{BinaryString::FromString(string_value, pool.get()), bytes});
110+
source_ref = source;
111+
array = std::make_unique<GenericArray>(
112+
std::vector<VariantType>{source->GetStringView(0), source->GetStringView(1)},
113+
std::vector<std::shared_ptr<InternalArray>>{source});
114+
}
115+
116+
ASSERT_FALSE(source_ref.expired());
117+
ASSERT_EQ(array->GetStringView(0), string_value);
118+
ASSERT_EQ(array->GetString(0).ToString(), string_value);
119+
auto bytes = array->GetBinary(1);
120+
ASSERT_TRUE(bytes);
121+
ASSERT_EQ(std::string(bytes->data(), bytes->size()), binary_value);
122+
array.reset();
123+
ASSERT_TRUE(source_ref.expired());
124+
ASSERT_EQ(std::string(bytes->data(), bytes->size()), binary_value);
125+
}
126+
127+
TEST(GenericArrayTest, TestToPrimitiveArrays) {
128+
{
129+
GenericArray array({true, false, true});
130+
ASSERT_OK_AND_ASSIGN(std::vector<char> result, array.ToBooleanArray());
131+
ASSERT_EQ(result, (std::vector<char>{1, 0, 1}));
132+
}
133+
{
134+
GenericArray array({char{0}, char{42}, std::numeric_limits<char>::max()});
135+
ASSERT_OK_AND_ASSIGN(std::vector<char> result, array.ToByteArray());
136+
ASSERT_EQ(result, (std::vector<char>{0, 42, std::numeric_limits<char>::max()}));
137+
}
138+
{
139+
GenericArray array(
140+
{std::numeric_limits<int16_t>::min(), int16_t{0}, std::numeric_limits<int16_t>::max()});
141+
ASSERT_OK_AND_ASSIGN(std::vector<int16_t> result, array.ToShortArray());
142+
ASSERT_EQ(result, (std::vector<int16_t>{std::numeric_limits<int16_t>::min(), 0,
143+
std::numeric_limits<int16_t>::max()}));
144+
}
145+
{
146+
GenericArray array(
147+
{std::numeric_limits<int32_t>::min(), int32_t{0}, std::numeric_limits<int32_t>::max()});
148+
ASSERT_OK_AND_ASSIGN(std::vector<int32_t> result, array.ToIntArray());
149+
ASSERT_EQ(result, (std::vector<int32_t>{std::numeric_limits<int32_t>::min(), 0,
150+
std::numeric_limits<int32_t>::max()}));
151+
}
152+
{
153+
GenericArray array(
154+
{std::numeric_limits<int64_t>::min(), int64_t{0}, std::numeric_limits<int64_t>::max()});
155+
ASSERT_OK_AND_ASSIGN(std::vector<int64_t> result, array.ToLongArray());
156+
ASSERT_EQ(result, (std::vector<int64_t>{std::numeric_limits<int64_t>::min(), 0,
157+
std::numeric_limits<int64_t>::max()}));
158+
}
159+
{
160+
GenericArray array({-1.5F, 0.0F, 2.25F});
161+
ASSERT_OK_AND_ASSIGN(std::vector<float> result, array.ToFloatArray());
162+
ASSERT_EQ(result, (std::vector<float>{-1.5F, 0.0F, 2.25F}));
163+
}
164+
{
165+
GenericArray array({-1.5, 0.0, 2.25});
166+
ASSERT_OK_AND_ASSIGN(std::vector<double> result, array.ToDoubleArray());
167+
ASSERT_EQ(result, (std::vector<double>{-1.5, 0.0, 2.25}));
168+
}
169+
}
170+
171+
TEST(GenericArrayTest, TestEmptyArray) {
172+
GenericArray array({});
173+
ASSERT_EQ(array.Size(), 0);
174+
ASSERT_OK_AND_ASSIGN(std::vector<char> booleans, array.ToBooleanArray());
175+
ASSERT_TRUE(booleans.empty());
176+
ASSERT_OK_AND_ASSIGN(std::vector<char> bytes, array.ToByteArray());
177+
ASSERT_TRUE(bytes.empty());
178+
ASSERT_OK_AND_ASSIGN(std::vector<int16_t> shorts, array.ToShortArray());
179+
ASSERT_TRUE(shorts.empty());
180+
ASSERT_OK_AND_ASSIGN(std::vector<int32_t> ints, array.ToIntArray());
181+
ASSERT_TRUE(ints.empty());
182+
ASSERT_OK_AND_ASSIGN(std::vector<int64_t> longs, array.ToLongArray());
183+
ASSERT_TRUE(longs.empty());
184+
ASSERT_OK_AND_ASSIGN(std::vector<float> floats, array.ToFloatArray());
185+
ASSERT_TRUE(floats.empty());
186+
ASSERT_OK_AND_ASSIGN(std::vector<double> doubles, array.ToDoubleArray());
187+
ASSERT_TRUE(doubles.empty());
188+
}
189+
190+
TEST(GenericArrayTest, TestPrimitiveArraysRejectNulls) {
191+
GenericArray null_array({NullType{}});
192+
ASSERT_TRUE(null_array.IsNullAt(0));
193+
const std::string message = "Primitive array must not contain a null value.";
194+
ASSERT_NOK_WITH_MSG(null_array.ToBooleanArray(), message);
195+
ASSERT_NOK_WITH_MSG(null_array.ToByteArray(), message);
196+
ASSERT_NOK_WITH_MSG(null_array.ToShortArray(), message);
197+
ASSERT_NOK_WITH_MSG(null_array.ToIntArray(), message);
198+
ASSERT_NOK_WITH_MSG(null_array.ToLongArray(), message);
199+
ASSERT_NOK_WITH_MSG(null_array.ToFloatArray(), message);
200+
ASSERT_NOK_WITH_MSG(null_array.ToDoubleArray(), message);
201+
202+
GenericArray booleans({true, NullType{}, false});
203+
ASSERT_NOK_WITH_MSG(booleans.ToBooleanArray(), message);
204+
GenericArray ints({int32_t{1}, NullType{}, int32_t{2}});
205+
ASSERT_NOK_WITH_MSG(ints.ToIntArray(), message);
206+
}
207+
208+
} // namespace paimon::test

‎src/paimon/common/file_index/file_index_reader_test.cpp‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,12 @@ TEST(FileIndexReaderTest, TestDefaultIndexReader) {
7070
ASSERT_TRUE(reader.VisitIn({lit0, lit1}).value()->IsRemain().value());
7171
ASSERT_TRUE(reader.VisitNotIn({lit0, lit1}).value()->IsRemain().value());
7272

73+
Literal text(FieldType::STRING, "abc", 3);
74+
ASSERT_TRUE(reader.VisitStartsWith(text).value()->IsRemain().value());
75+
ASSERT_TRUE(reader.VisitEndsWith(text).value()->IsRemain().value());
76+
ASSERT_TRUE(reader.VisitContains(text).value()->IsRemain().value());
77+
ASSERT_TRUE(reader.VisitLike(text).value()->IsRemain().value());
78+
7379
std::shared_ptr<VectorSearch> vector_search =
7480
std::make_shared<VectorSearch>("embedding", 1, std::vector<float>{1.0f}, nullptr, nullptr,
7581
std::nullopt, std::map<std::string, std::string>{});

‎src/paimon/common/fs/object_store_file_system_test.cpp‎

Lines changed: 65 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
#include <cstring>
2626
#include <limits>
2727
#include <map>
28+
#include <memory>
29+
#include <string>
2830
#include <utility>
2931
#include <vector>
3032

@@ -121,13 +123,30 @@ TEST(ObjectStoreFileSystemTest, TestObjectWinsOverPrefix) {
121123
ASSERT_FALSE(statuses[0].IsDir());
122124
}
123125

126+
TEST(ObjectStoreFileSystemTest, TestExistsAndDirectoryStatus) {
127+
auto client = std::make_shared<MockObjectStoreClient>();
128+
ObjectStoreFileSystem fs("s3", client);
129+
ASSERT_OK_AND_ASSIGN(bool root_exists, fs.Exists("s3://bucket/"));
130+
ASSERT_TRUE(root_exists);
131+
client->objects_["dir/file"] = "data";
132+
ASSERT_OK_AND_ASSIGN(bool file_exists, fs.Exists("s3://bucket/dir/file"));
133+
ASSERT_TRUE(file_exists);
134+
ASSERT_OK_AND_ASSIGN(bool directory_exists, fs.Exists("s3://bucket/dir"));
135+
ASSERT_TRUE(directory_exists);
136+
ASSERT_OK_AND_ASSIGN(bool missing_exists, fs.Exists("s3://bucket/missing"));
137+
ASSERT_FALSE(missing_exists);
138+
ASSERT_OK_AND_ASSIGN(FileStatus status, fs.GetFileStatus("s3://bucket/dir"));
139+
ASSERT_TRUE(status.IsDir());
140+
ASSERT_EQ("s3://bucket/dir/", status.GetPath());
141+
ASSERT_NOK_WITH_MSG(fs.Open("s3://bucket/dir"), "is a directory");
142+
}
143+
124144
TEST(ObjectStoreFileSystemTest, TestHeadErrorIsNotMasked) {
125145
auto client = std::make_shared<MockObjectStoreClient>();
126146
client->head_error_ = Status::IOError("access denied");
127147
client->objects_["foo/bar"] = "child";
128148
ObjectStoreFileSystem fs("s3", client);
129-
auto status = fs.GetFileStatus("s3://bucket/foo");
130-
ASSERT_TRUE(status.status().IsIOError());
149+
ASSERT_NOK_WITH_MSG(fs.GetFileStatus("s3://bucket/foo"), "access denied");
131150
ASSERT_EQ(client->list_calls_, 0);
132151
}
133152

@@ -148,6 +167,14 @@ TEST(ObjectStoreFileSystemTest, TestPaginationAndDirectoryMarker) {
148167
ASSERT_EQ(statuses.size(), 2);
149168
ASSERT_EQ(statuses[0].GetPath(), "s3://bucket/dir/a");
150169
ASSERT_TRUE(statuses[1].IsDir());
170+
171+
std::vector<BasicFileStatus> basic_statuses;
172+
ASSERT_OK(fs.ListDir("s3://bucket/dir/", &basic_statuses));
173+
ASSERT_EQ(2u, basic_statuses.size());
174+
ASSERT_EQ("s3://bucket/dir/a", basic_statuses[0].GetPath());
175+
ASSERT_FALSE(basic_statuses[0].IsDir());
176+
ASSERT_EQ("s3://bucket/dir/sub/", basic_statuses[1].GetPath());
177+
ASSERT_TRUE(basic_statuses[1].IsDir());
151178
}
152179

153180
TEST(ObjectStoreFileSystemTest, TestTruncatedPageRequiresContinuationToken) {
@@ -158,14 +185,15 @@ TEST(ObjectStoreFileSystemTest, TestTruncatedPageRequiresContinuationToken) {
158185
client->pages_[""] = page;
159186
ObjectStoreFileSystem fs("s3", client);
160187
std::vector<FileStatus> statuses;
161-
ASSERT_TRUE(fs.ListFileStatus("s3://bucket/dir/", &statuses).IsIOError());
188+
ASSERT_NOK_WITH_MSG(fs.ListFileStatus("s3://bucket/dir/", &statuses),
189+
"did not include a continuation token");
162190
ASSERT_TRUE(statuses.empty());
163191
}
164192

165193
TEST(ObjectStoreFileSystemTest, TestOpenBucketRootIsDirectory) {
166194
auto client = std::make_shared<MockObjectStoreClient>();
167195
ObjectStoreFileSystem fs("s3", client);
168-
ASSERT_TRUE(fs.Open("s3://bucket/").status().IsInvalid());
196+
ASSERT_NOK_WITH_MSG(fs.Open("s3://bucket/"), "is a directory");
169197
ASSERT_EQ(client->head_calls_, 0);
170198
ASSERT_EQ(client->list_calls_, 0);
171199
}
@@ -187,6 +215,32 @@ TEST(ObjectStoreFileSystemTest, TestPathWithLeadingSlashes) {
187215
ASSERT_EQ(status.GetPath(), "s3://bucket/file");
188216
}
189217

218+
TEST(ObjectStoreInputStreamTest, TestSeekAndEmptyRead) {
219+
auto client = std::make_shared<MockObjectStoreClient>();
220+
client->objects_["file"] = "data";
221+
ObjectStoreFileSystem fs("s3", client);
222+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> stream, fs.Open("s3://bucket/file"));
223+
ASSERT_OK_AND_ASSIGN(std::string uri, stream->GetUri());
224+
ASSERT_EQ("s3://bucket/file", uri);
225+
ASSERT_OK(stream->Seek(1, FS_SEEK_SET));
226+
ASSERT_OK(stream->Seek(1, FS_SEEK_CUR));
227+
std::array<char, 2> buffer{};
228+
ASSERT_OK_AND_ASSIGN(int64_t bytes_read, stream->Read(buffer.data(), buffer.size()));
229+
ASSERT_EQ(2, bytes_read);
230+
ASSERT_EQ("ta", std::string(buffer.data(), buffer.size()));
231+
ASSERT_OK_AND_ASSIGN(int64_t position, stream->GetPos());
232+
ASSERT_EQ(4, position);
233+
ASSERT_OK_AND_ASSIGN(int64_t eof_read, stream->Read(buffer.data(), buffer.size()));
234+
ASSERT_EQ(0, eof_read);
235+
bool called = false;
236+
stream->ReadAsync(nullptr, 0, position, [&called](Status status) {
237+
called = true;
238+
ASSERT_OK(status);
239+
});
240+
ASSERT_TRUE(called);
241+
ASSERT_EQ(1u, client->ranges_.size());
242+
}
243+
190244
TEST(ObjectStoreInputStreamTest, TestBoundsCloseAndSeekOverflow) {
191245
auto client = std::make_shared<MockObjectStoreClient>();
192246
client->objects_["file"] = std::string(128 * 1024, 'x');
@@ -196,16 +250,17 @@ TEST(ObjectStoreInputStreamTest, TestBoundsCloseAndSeekOverflow) {
196250
bool called = false;
197251
stream->ReadAsync(data, 4, 128 * 1024 - 2, [&called](Status status) {
198252
called = true;
199-
ASSERT_TRUE(status.IsInvalid());
253+
ASSERT_NOK_WITH_MSG(status, "exceeds length");
200254
});
201255
ASSERT_TRUE(called);
202-
ASSERT_TRUE(stream->Seek(std::numeric_limits<int64_t>::max(), FS_SEEK_END).IsInvalid());
256+
ASSERT_NOK_WITH_MSG(stream->Seek(std::numeric_limits<int64_t>::max(), FS_SEEK_END),
257+
"object store input position overflows");
203258
ASSERT_OK(stream->Close());
204259
called = false;
205260
stream->ReadAsync(data, 1, 0, [&called, &stream](Status status) {
206261
called = true;
207-
ASSERT_TRUE(status.IsIOError());
208-
ASSERT_TRUE(stream->GetPos().status().IsIOError());
262+
ASSERT_NOK_WITH_MSG(status, "is closed");
263+
ASSERT_NOK_WITH_MSG(stream->GetPos(), "is closed");
209264
});
210265
ASSERT_TRUE(called);
211266
}
@@ -217,7 +272,8 @@ TEST(ObjectStoreInputStreamTest, TestShortReadFails) {
217272
ObjectStoreFileSystem fs("s3", client);
218273
ASSERT_OK_AND_ASSIGN(auto stream, fs.Open("s3://bucket/file"));
219274
char data[4];
220-
ASSERT_TRUE(stream->Read(data, sizeof(data)).status().IsIOError());
275+
ASSERT_NOK_WITH_MSG(stream->Read(data, sizeof(data)),
276+
"range read for s3://bucket/file returned");
221277
}
222278

223279
TEST(ObjectStoreInputStreamTest, TestSequentialReadAheadGrows) {

‎src/paimon/common/metrics/histogram_windowing.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,8 @@ namespace paimon {
3636
// Inspired by RocksDB's monitoring/histogram_windowing.h.
3737
// https://github.com/facebook/rocksdb/blob/main/monitoring/histogram_windowing.h
3838
// Maintains multiple time windows and aggregates stats across the recent ones.
39+
//
40+
// TODO(jinli.zjw): Refactor this implementation in the future.
3941
class HistogramWindowingImpl : public Histogram {
4042
public:
4143
// Default: 3 windows, 60s per window , 100 per window (paimon-java default)

0 commit comments

Comments
 (0)