This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 370582dc feat(file-index): support BSI and Bloom filter writers (#234)
370582dc is described below
commit 370582dc1e3ec414bd83b739681e02686db86464
Author: Zhang Jiawei <[email protected]>
AuthorDate: Tue Aug 25 17:59:46 2026 +0800
feat(file-index): support BSI and Bloom filter writers (#234)
---
.../common/file_index/bitmap/bitmap_file_index.cpp | 24 +----
.../common/file_index/bitmap/bitmap_file_index.h | 2 +-
.../bloomfilter/bloom_filter_file_index.cpp | 88 +++++++++++++--
.../bloomfilter/bloom_filter_file_index.h | 38 +++++--
.../bloomfilter/bloom_filter_file_index_test.cpp | 55 +++++++++-
.../common/file_index/bloomfilter/fast_hash.cpp | 16 ++-
.../file_index/bloomfilter/fast_hash_test.cpp | 31 ++++++
.../bsi/bit_slice_index_bitmap_file_index.cpp | 105 ++++++++++++++++++
.../bsi/bit_slice_index_bitmap_file_index.h | 34 ++++--
.../bsi/bit_slice_index_bitmap_file_index_test.cpp | 119 +++++++++++++++------
.../rangebitmap/range_bitmap_file_index.cpp | 33 ++----
.../rangebitmap/range_bitmap_file_index.h | 6 +-
.../rangebitmap/range_bitmap_file_index_test.cpp | 10 +-
.../rangebitmap/range_bitmap_io_test.cpp | 8 +-
src/paimon/common/lookup/lookup_store_factory.cpp | 3 +-
src/paimon/common/sst/sst_file_io_test.cpp | 4 +-
src/paimon/common/utils/bloom_filter.cpp | 24 +++--
src/paimon/common/utils/bloom_filter.h | 8 +-
src/paimon/common/utils/bloom_filter64.cpp | 52 +++++++--
src/paimon/common/utils/bloom_filter64.h | 11 +-
src/paimon/common/utils/bloom_filter64_test.cpp | 42 +++++++-
src/paimon/common/utils/bloom_filter_test.cpp | 35 ++++--
src/paimon/common/utils/math.h | 27 +++++
src/paimon/common/utils/math_test.cpp | 18 ++++
src/paimon/core/io/data_file_index_writer_test.cpp | 31 +++++-
25 files changed, 658 insertions(+), 166 deletions(-)
diff --git a/src/paimon/common/file_index/bitmap/bitmap_file_index.cpp
b/src/paimon/common/file_index/bitmap/bitmap_file_index.cpp
index 6df16143..43bb71b0 100644
--- a/src/paimon/common/file_index/bitmap/bitmap_file_index.cpp
+++ b/src/paimon/common/file_index/bitmap/bitmap_file_index.cpp
@@ -103,24 +103,18 @@ Result<std::shared_ptr<FileIndexWriter>>
BitmapFileIndex::CreateWriter(
"invalid schema for BitmapFileIndexWriter, supposed to have single
"
"field.");
}
- auto arrow_field = arrow_schema->field(0);
- return BitmapFileIndexWriter::Create(arrow_schema, arrow_field->name(),
options_, pool);
+ return BitmapFileIndexWriter::Create(arrow_schema->field(0), options_,
pool);
}
Result<std::shared_ptr<BitmapFileIndexWriter>> BitmapFileIndexWriter::Create(
- const std::shared_ptr<arrow::Schema>& arrow_schema, const std::string&
field_name,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<MemoryPool>& pool) {
+ const std::shared_ptr<arrow::Field>& field, const std::map<std::string,
std::string>& options,
+ const std::shared_ptr<MemoryPool>& pool) {
PAIMON_ASSIGN_OR_RAISE(int8_t version,
OptionsUtils::GetValueFromMap<int8_t>(options,
BitmapFileIndex::VERSION,
BitmapFileIndex::VERSION_2));
- auto arrow_field = arrow_schema->GetFieldByName(field_name);
- if (!arrow_field) {
- return Status::Invalid(
- fmt::format("field {} not in arrow_schema for
BitmapFileIndexWriter", field_name));
- }
- auto struct_type = arrow::struct_({arrow_field});
+ std::shared_ptr<arrow::DataType> struct_type = arrow::struct_({field});
return std::shared_ptr<BitmapFileIndexWriter>(
- new BitmapFileIndexWriter(version, struct_type, arrow_field->type(),
options, pool));
+ new BitmapFileIndexWriter(version, struct_type, field->type(),
options, pool));
}
BitmapFileIndexWriter::BitmapFileIndexWriter(int8_t version,
@@ -137,15 +131,7 @@ BitmapFileIndexWriter::BitmapFileIndexWriter(int8_t
version,
Status BitmapFileIndexWriter::AddBatch(::ArrowArray* batch) {
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array>
arrow_array,
arrow::ImportArray(batch, struct_type_));
- if (!arrow_array || arrow_array->type_id() != arrow::Type::STRUCT) {
- return Status::Invalid("invalid batch for BitmapFileIndexWriter,
expected a struct array");
- }
auto struct_array = checked_pointer_cast<arrow::StructArray>(arrow_array);
- if (struct_array->num_fields() != 1) {
- return Status::Invalid(
- "invalid batch for BitmapFileIndexWriter, expected a struct array
with exactly one "
- "field");
- }
PAIMON_ASSIGN_OR_RAISE(
std::vector<Literal> array_values,
LiteralConverter::ConvertLiteralsFromArray(*(struct_array->field(0)),
/*own_data=*/true));
diff --git a/src/paimon/common/file_index/bitmap/bitmap_file_index.h
b/src/paimon/common/file_index/bitmap/bitmap_file_index.h
index 9cf15ddc..6aa0f416 100644
--- a/src/paimon/common/file_index/bitmap/bitmap_file_index.h
+++ b/src/paimon/common/file_index/bitmap/bitmap_file_index.h
@@ -71,7 +71,7 @@ class PAIMON_EXPORT BitmapFileIndex : public FileIndexer {
class BitmapFileIndexWriter : public FileIndexWriter {
public:
static Result<std::shared_ptr<BitmapFileIndexWriter>> Create(
- const std::shared_ptr<arrow::Schema>& arrow_schema, const std::string&
field_name,
+ const std::shared_ptr<arrow::Field>& field,
const std::map<std::string, std::string>& options, const
std::shared_ptr<MemoryPool>& pool);
Status AddBatch(::ArrowArray* batch) override;
diff --git
a/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp
b/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp
index ca33c696..86d07d63 100644
--- a/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp
+++ b/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp
@@ -19,10 +19,19 @@
#include "paimon/common/file_index/bloomfilter/bloom_filter_file_index.h"
#include <cstddef>
+#include <cstdint>
+#include <cstring>
#include <functional>
#include <utility>
+#include <vector>
+#include "arrow/c/bridge.h"
#include "fmt/format.h"
+#include "paimon/common/predicate/literal_converter.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/common/utils/options_utils.h"
#include "paimon/fs/file_system.h"
#include "paimon/memory/bytes.h"
#include "paimon/predicate/literal.h"
@@ -31,7 +40,8 @@
namespace paimon {
class MemoryPool;
-BloomFilterFileIndex::BloomFilterFileIndex(const std::map<std::string,
std::string>& options) {}
+BloomFilterFileIndex::BloomFilterFileIndex(const std::map<std::string,
std::string>& options)
+ : options_(options) {}
Result<std::shared_ptr<FileIndexReader>> BloomFilterFileIndex::CreateReader(
::ArrowSchema* c_arrow_schema, int32_t start, int32_t length,
const std::shared_ptr<InputStream>& input_stream,
@@ -58,15 +68,77 @@ Result<std::shared_ptr<FileIndexReader>>
BloomFilterFileIndex::CreateReader(
return BloomFilterFileIndexReader::Create(arrow_type, bytes);
}
+Result<std::shared_ptr<FileIndexWriter>> BloomFilterFileIndex::CreateWriter(
+ ::ArrowSchema* c_arrow_schema, const std::shared_ptr<MemoryPool>& pool)
const {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema>
arrow_schema,
+ arrow::ImportSchema(c_arrow_schema));
+ if (arrow_schema->num_fields() != 1) {
+ return Status::Invalid(
+ "invalid schema for BloomFilterFileIndexWriter, supposed to have
single field.");
+ }
+ return BloomFilterFileIndexWriter::Create(arrow_schema->field(0),
options_, pool);
+}
+
+Result<std::shared_ptr<BloomFilterFileIndexWriter>>
BloomFilterFileIndexWriter::Create(
+ const std::shared_ptr<arrow::Field>& field, const std::map<std::string,
std::string>& options,
+ const std::shared_ptr<MemoryPool>& pool) {
+ PAIMON_ASSIGN_OR_RAISE(FastHash::HashFunction hash_function,
+ FastHash::GetHashFunction(field->type()));
+ PAIMON_ASSIGN_OR_RAISE(
+ int32_t items, OptionsUtils::GetValueFromMap<int32_t>(options,
BloomFilterFileIndex::kItems,
+
BloomFilterFileIndex::kDefaultItems));
+ PAIMON_ASSIGN_OR_RAISE(
+ double fpp, OptionsUtils::GetValueFromMap<double>(options,
BloomFilterFileIndex::kFpp,
+
BloomFilterFileIndex::kDefaultFpp));
+ std::shared_ptr<arrow::DataType> struct_type = arrow::struct_({field});
+ PAIMON_ASSIGN_OR_RAISE(BloomFilter64 filter, BloomFilter64::Create(items,
fpp, pool));
+ return std::shared_ptr<BloomFilterFileIndexWriter>(
+ new BloomFilterFileIndexWriter(struct_type, hash_function,
std::move(filter), pool));
+}
+
+BloomFilterFileIndexWriter::BloomFilterFileIndexWriter(
+ const std::shared_ptr<arrow::DataType>& struct_type,
+ const FastHash::HashFunction& hash_function, BloomFilter64&& filter,
+ const std::shared_ptr<MemoryPool>& pool)
+ : struct_type_(struct_type),
+ hash_function_(hash_function),
+ filter_(std::move(filter)),
+ pool_(pool) {}
+
+Status BloomFilterFileIndexWriter::AddBatch(::ArrowArray* batch) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
+ arrow::ImportArray(batch, struct_type_));
+ std::shared_ptr<arrow::StructArray> struct_array =
+ checked_pointer_cast<arrow::StructArray>(array);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<Literal> values,
+ LiteralConverter::ConvertLiteralsFromArray(*struct_array->field(0),
/*own_data=*/false));
+ for (const Literal& value : values) {
+ if (!value.IsNull()) {
+ filter_.AddHash(hash_function_(value));
+ }
+ }
+ return Status::OK();
+}
+
+Result<PAIMON_UNIQUE_PTR<Bytes>> BloomFilterFileIndexWriter::SerializedBytes()
const {
+ constexpr int32_t kHeaderLength = sizeof(int32_t);
+ const int32_t bit_set_length = filter_.GetBitSet().ByteLength();
+ PAIMON_UNIQUE_PTR<Bytes> bytes =
+ Bytes::AllocateBytes(kHeaderLength + bit_set_length, pool_.get());
+ const int32_t num_hash_functions =
ToBigEndian(filter_.GetNumHashFunctions());
+ std::memcpy(bytes->data(), &num_hash_functions,
sizeof(num_hash_functions));
+ filter_.GetBitSet().ToByteArray(kHeaderLength, bit_set_length,
bytes->data());
+ return bytes;
+}
+
Result<std::shared_ptr<BloomFilterFileIndexReader>>
BloomFilterFileIndexReader::Create(
const std::shared_ptr<arrow::DataType>& arrow_type, const
std::shared_ptr<Bytes>& bytes) {
- // compatible with java, little endian
- const char* data = bytes->data();
- auto num_hash_functions =
-
static_cast<int32_t>((static_cast<uint32_t>(static_cast<uint8_t>(data[0])) <<
24) |
-
(static_cast<uint32_t>(static_cast<uint8_t>(data[1])) << 16) |
-
(static_cast<uint32_t>(static_cast<uint8_t>(data[2])) << 8) |
-
static_cast<uint32_t>(static_cast<uint8_t>(data[3])));
+ // Compatible with Java's big-endian numHashFunctions header.
+ int32_t big_endian_num_hash_functions;
+ std::memcpy(&big_endian_num_hash_functions, bytes->data(),
+ sizeof(big_endian_num_hash_functions));
+ const int32_t num_hash_functions =
FromBigEndian(big_endian_num_hash_functions);
PAIMON_ASSIGN_OR_RAISE(FastHash::HashFunction hash_function,
FastHash::GetHashFunction(arrow_type));
auto bit_set = std::make_unique<BloomFilter64::BitSet>(bytes,
/*offset=*/sizeof(int32_t));
diff --git a/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.h
b/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.h
index 8152c220..62d082c7 100644
--- a/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.h
+++ b/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.h
@@ -25,10 +25,10 @@
#include "arrow/c/bridge.h"
#include "paimon/common/file_index/bloomfilter/fast_hash.h"
-#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/bloom_filter64.h"
#include "paimon/file_index/file_index_reader.h"
#include "paimon/file_index/file_index_result.h"
+#include "paimon/file_index/file_index_writer.h"
#include "paimon/file_index/file_indexer.h"
#include "paimon/result.h"
namespace paimon {
@@ -55,11 +55,37 @@ class BloomFilterFileIndex : public FileIndexer {
const std::shared_ptr<MemoryPool>& pool) const override;
Result<std::shared_ptr<FileIndexWriter>> CreateWriter(
- ::ArrowSchema* arrow_schema, const std::shared_ptr<MemoryPool>& pool)
const override {
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::DataType>
arrow_type,
- arrow::ImportType(arrow_schema));
- return Status::NotImplemented("do not support index writer in bloom
filter");
- }
+ ::ArrowSchema* arrow_schema, const std::shared_ptr<MemoryPool>& pool)
const override;
+
+ static constexpr int32_t kDefaultItems = 1000000;
+ static constexpr double kDefaultFpp = 0.1;
+ static constexpr char kItems[] = "items";
+ static constexpr char kFpp[] = "fpp";
+
+ private:
+ std::map<std::string, std::string> options_;
+};
+
+class BloomFilterFileIndexWriter : public FileIndexWriter {
+ public:
+ static Result<std::shared_ptr<BloomFilterFileIndexWriter>> Create(
+ const std::shared_ptr<arrow::Field>& field,
+ const std::map<std::string, std::string>& options, const
std::shared_ptr<MemoryPool>& pool);
+
+ Status AddBatch(::ArrowArray* batch) override;
+
+ Result<PAIMON_UNIQUE_PTR<Bytes>> SerializedBytes() const override;
+
+ private:
+ BloomFilterFileIndexWriter(const std::shared_ptr<arrow::DataType>&
struct_type,
+ const FastHash::HashFunction& hash_function,
BloomFilter64&& filter,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ private:
+ std::shared_ptr<arrow::DataType> struct_type_;
+ FastHash::HashFunction hash_function_;
+ BloomFilter64 filter_;
+ std::shared_ptr<MemoryPool> pool_;
};
class BloomFilterFileIndexReader : public FileIndexReader {
diff --git
a/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index_test.cpp
b/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index_test.cpp
index 209f0b40..23fadce2 100644
--- a/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index_test.cpp
+++ b/src/paimon/common/file_index/bloomfilter/bloom_filter_file_index_test.cpp
@@ -18,10 +18,16 @@
#include "paimon/common/file_index/bloomfilter/bloom_filter_file_index.h"
+#include <cstring>
+#include <map>
+#include <string>
#include <utility>
#include <vector>
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/json_simple.h"
#include "gtest/gtest.h"
+#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/field_type_utils.h"
#include "paimon/data/timestamp.h"
#include "paimon/defs.h"
@@ -49,21 +55,60 @@ class BloomFilterIndexReaderTest : public ::testing::Test {
return c_schema;
}
+ Result<PAIMON_UNIQUE_PTR<Bytes>> WriteIndex(
+ const std::shared_ptr<arrow::DataType>& data_type, const std::string&
json,
+ const std::map<std::string, std::string>& options) const {
+ const std::shared_ptr<arrow::Schema> schema =
+ arrow::schema({arrow::field("f0", data_type)});
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(schema->fields()),
json)
+ .ValueOrDie();
+ ::ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema,
&c_schema));
+ BloomFilterFileIndex file_index(options);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileIndexWriter> writer,
+ file_index.CreateWriter(&c_schema, pool_));
+ ::ArrowArray c_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array));
+ PAIMON_RETURN_NOT_OK(writer->AddBatch(&c_array));
+ return writer->SerializedBytes();
+ }
+
private:
std::shared_ptr<MemoryPool> pool_;
};
+TEST_F(BloomFilterIndexReaderTest, TestWriterRejectsInvalidOptionsAndType) {
+ auto create_writer = [&](const std::shared_ptr<arrow::DataType>& type,
+ const std::map<std::string, std::string>&
options) {
+ BloomFilterFileIndex file_index(options);
+ return file_index.CreateWriter(CreateArrowSchema(type).get(), pool_);
+ };
+ ASSERT_NOK_WITH_MSG(create_writer(arrow::int32(), {{"items", "0"}}),
+ "items must be greater than 0");
+ ASSERT_NOK_WITH_MSG(create_writer(arrow::int32(), {{"fpp", "1"}}),
+ "fpp must be greater than 0 and less than 1");
+ ASSERT_NOK_WITH_MSG(create_writer(arrow::boolean(), {}),
+ "bloom filter index does not support BOOLEAN");
+}
+
TEST_F(BloomFilterIndexReaderTest, TestStringType) {
- // data: "a", "b", ""
- std::vector<uint8_t> index_bytes = {0, 0, 0, 6, 0, 32, 32, 3, 208, 32, 0,
64, 73, 16, 201};
- auto input_stream = std::make_shared<ByteArrayInputStream>(
- reinterpret_cast<char*>(index_bytes.data()), index_bytes.size());
+ // Java writer output for data: "a", "b", "" with items=10 and fpp=0.02.
+ const std::vector<uint8_t> expected = {0, 0, 0, 6, 0, 32, 32, 3, 208, 32,
0, 64, 73, 16, 201};
+ ASSERT_OK_AND_ASSIGN(
+ PAIMON_UNIQUE_PTR<Bytes> bytes,
+ WriteIndex(arrow::utf8(), R"([["a"], ["b"], [""]])", {{"items", "10"},
{"fpp", "0.02"}}));
+ ASSERT_EQ(expected.size(), bytes->size());
+ ASSERT_EQ(0, std::memcmp(expected.data(), bytes->data(), expected.size()));
+
+ std::shared_ptr<ByteArrayInputStream> input_stream =
+ std::make_shared<ByteArrayInputStream>(bytes->data(), bytes->size());
BloomFilterFileIndex file_index({});
ASSERT_OK_AND_ASSIGN(
auto reader,
file_index.CreateReader(CreateArrowSchema(arrow::utf8()).get(),
- /*start=*/0, /*length=*/index_bytes.size(),
input_stream, pool_));
+ /*start=*/0, /*length=*/bytes->size(),
input_stream, pool_));
ASSERT_TRUE(reader);
ASSERT_TRUE(reader->VisitEqual(Literal(FieldType::STRING, "a",
1)).value()->IsRemain().value());
ASSERT_TRUE(reader->VisitEqual(Literal(FieldType::STRING, "b",
1)).value()->IsRemain().value());
diff --git a/src/paimon/common/file_index/bloomfilter/fast_hash.cpp
b/src/paimon/common/file_index/bloomfilter/fast_hash.cpp
index d98dc476..b1d8784f 100644
--- a/src/paimon/common/file_index/bloomfilter/fast_hash.cpp
+++ b/src/paimon/common/file_index/bloomfilter/fast_hash.cpp
@@ -19,6 +19,7 @@
#include "paimon/common/file_index/bloomfilter/fast_hash.h"
#include <cassert>
+#include <cmath>
#include <cstring>
#include <string>
#include <utility>
@@ -34,6 +35,11 @@
#include "xxhash.h" // NOLINT(build/include_subdir)
namespace paimon {
+namespace {
+constexpr int32_t kCanonicalFloatNaNBits = 0x7fc00000;
+constexpr int64_t kCanonicalDoubleNaNBits = 0x7ff8000000000000L;
+} // namespace
+
Result<FastHash::HashFunction> FastHash::GetHashFunction(
const std::shared_ptr<arrow::DataType>& arrow_type) {
PAIMON_ASSIGN_OR_RAISE(FieldType field_type,
@@ -58,14 +64,20 @@ Result<FastHash::HashFunction> FastHash::GetHashFunction(
});
case FieldType::FLOAT:
return HashFunction([](const Literal& literal) -> int64_t {
- auto raw_value = literal.GetValue<float>();
+ const auto raw_value = literal.GetValue<float>();
+ if (std::isnan(raw_value)) {
+ return GetLongHash(kCanonicalFloatNaNBits);
+ }
int32_t bits = 0;
std::memcpy(&bits, &raw_value, sizeof(raw_value));
return GetLongHash(bits);
});
case FieldType::DOUBLE:
return HashFunction([](const Literal& literal) -> int64_t {
- auto raw_value = literal.GetValue<double>();
+ const auto raw_value = literal.GetValue<double>();
+ if (std::isnan(raw_value)) {
+ return GetLongHash(kCanonicalDoubleNaNBits);
+ }
int64_t bits;
std::memcpy(&bits, &raw_value, sizeof(raw_value));
return GetLongHash(bits);
diff --git a/src/paimon/common/file_index/bloomfilter/fast_hash_test.cpp
b/src/paimon/common/file_index/bloomfilter/fast_hash_test.cpp
index 680d591b..8a528e44 100644
--- a/src/paimon/common/file_index/bloomfilter/fast_hash_test.cpp
+++ b/src/paimon/common/file_index/bloomfilter/fast_hash_test.cpp
@@ -18,6 +18,9 @@
#include "paimon/common/file_index/bloomfilter/fast_hash.h"
+#include <cmath>
+#include <cstdint>
+#include <cstring>
#include <limits>
#include <string>
#include <vector>
@@ -164,4 +167,32 @@ TEST_F(FastHashTest, TestCompatibleWithJava) {
}
}
+TEST_F(FastHashTest, TestNaNCompatibleWithJava) {
+ auto float_from_bits = [](uint32_t bits) {
+ float value;
+ std::memcpy(&value, &bits, sizeof(value));
+ return value;
+ };
+ const float float_nan = float_from_bits(0x7fc12345);
+ const float negative_float_nan = float_from_bits(0xffc54321);
+ ASSERT_TRUE(std::isnan(float_nan));
+ ASSERT_TRUE(std::isnan(negative_float_nan));
+ ASSERT_OK_AND_ASSIGN(auto float_hash_function,
FastHash::GetHashFunction(arrow::float32()));
+ CheckResult(float_hash_function, {Literal(float_nan),
Literal(negative_float_nan)},
+ {0x67c27c6d9936ae63, 0x67c27c6d9936ae63});
+
+ auto double_from_bits = [](uint64_t bits) {
+ double value;
+ std::memcpy(&value, &bits, sizeof(value));
+ return value;
+ };
+ const double double_nan = double_from_bits(0x7ff8123456789abc);
+ const double negative_double_nan = double_from_bits(0xfff8abcdef012345);
+ ASSERT_TRUE(std::isnan(double_nan));
+ ASSERT_TRUE(std::isnan(negative_double_nan));
+ ASSERT_OK_AND_ASSIGN(auto double_hash_function,
FastHash::GetHashFunction(arrow::float64()));
+ CheckResult(double_hash_function, {Literal(double_nan),
Literal(negative_double_nan)},
+ {0x13d2d3f2cc0e846e, 0x13d2d3f2cc0e846e});
+}
+
} // namespace paimon::test
diff --git
a/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.cpp
b/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.cpp
index 7ce53e9a..4cb98cdc 100644
--- a/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.cpp
+++ b/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.cpp
@@ -18,13 +18,23 @@
#include "paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h"
+#include <algorithm>
#include <cassert>
#include <climits>
#include <cstddef>
#include <cstdint>
+#include <limits>
+#include <optional>
+#include <utility>
+#include <vector>
+#include "arrow/c/bridge.h"
#include "fmt/format.h"
#include "paimon/common/file_index/bsi/bit_slice_index_roaring_bitmap.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/common/memory/memory_segment_utils.h"
+#include "paimon/common/predicate/literal_converter.h"
+#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/date_time_utils.h"
#include "paimon/common/utils/field_type_utils.h"
@@ -53,6 +63,101 @@ class MemoryPool;
BitSliceIndexBitmapFileIndex::BitSliceIndexBitmapFileIndex(
const std::map<std::string, std::string>& options) {}
+Result<std::shared_ptr<FileIndexWriter>>
BitSliceIndexBitmapFileIndex::CreateWriter(
+ ::ArrowSchema* c_arrow_schema, const std::shared_ptr<MemoryPool>& pool)
const {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema>
arrow_schema,
+ arrow::ImportSchema(c_arrow_schema));
+ if (arrow_schema->num_fields() != 1) {
+ return Status::Invalid(
+ "invalid schema for BitSliceIndexBitmapFileIndexWriter, supposed
to have single "
+ "field.");
+ }
+ const std::shared_ptr<arrow::Field> field = arrow_schema->field(0);
+ PAIMON_ASSIGN_OR_RAISE(ValueMapperType value_mapper,
GetValueMapper(field->type()));
+ return std::make_shared<BitSliceIndexBitmapFileIndexWriter>(field,
value_mapper, pool);
+}
+
+BitSliceIndexBitmapFileIndexWriter::BitSliceIndexBitmapFileIndexWriter(
+ const std::shared_ptr<arrow::Field>& field,
+ const BitSliceIndexBitmapFileIndex::ValueMapperType& value_mapper,
+ const std::shared_ptr<MemoryPool>& pool)
+ : struct_type_(arrow::struct_({field})),
+ field_name_(field->name()),
+ value_mapper_(value_mapper),
+ pool_(pool) {}
+
+Status BitSliceIndexBitmapFileIndexWriter::AddBatch(::ArrowArray* batch) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
+ arrow::ImportArray(batch, struct_type_));
+ auto struct_array = checked_pointer_cast<arrow::StructArray>(array);
+ if (struct_array->length() >
static_cast<int64_t>(std::numeric_limits<int32_t>::max()) -
+ static_cast<int64_t>(values_.size())) {
+ return Status::Invalid("bsi index row count exceeds the supported
int32 range");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<Literal> literals,
+ LiteralConverter::ConvertLiteralsFromArray(*struct_array->field(0),
/*own_data=*/false));
+ values_.reserve(values_.size() + literals.size());
+ for (const Literal& literal : literals) {
+ if (literal.IsNull()) {
+ values_.emplace_back(std::nullopt);
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(int64_t value, value_mapper_(literal));
+ if (value == std::numeric_limits<int64_t>::min()) {
+ return Status::Invalid(
+ fmt::format("bsi index does not support INT64_MIN for field
'{}'", field_name_));
+ }
+ values_.emplace_back(value);
+ if (value < 0) {
+ const int64_t absolute_value = SafeAbs(value);
+ negative_min_ = std::min(negative_min_, absolute_value);
+ negative_max_ = std::max(negative_max_, absolute_value);
+ } else {
+ positive_min_ = std::min(positive_min_, value);
+ positive_max_ = std::max(positive_max_, value);
+ }
+ }
+ return Status::OK();
+}
+
+Result<PAIMON_UNIQUE_PTR<Bytes>>
BitSliceIndexBitmapFileIndexWriter::SerializedBytes() const {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<BitSliceIndexRoaringBitmap::Appender> positive,
+ BitSliceIndexRoaringBitmap::Appender::Create(positive_min_,
positive_max_));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<BitSliceIndexRoaringBitmap::Appender> negative,
+ BitSliceIndexRoaringBitmap::Appender::Create(negative_min_,
negative_max_));
+ for (size_t i = 0; i < values_.size(); ++i) {
+ if (!values_[i]) {
+ continue;
+ }
+ const int64_t value = values_[i].value();
+ if (value < 0) {
+ PAIMON_RETURN_NOT_OK(negative->Append(static_cast<int32_t>(i),
SafeAbs(value)));
+ } else {
+ PAIMON_RETURN_NOT_OK(positive->Append(static_cast<int32_t>(i),
value));
+ }
+ }
+
+ MemorySegmentOutputStream
output_stream(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool_);
+ output_stream.SetOrder(ByteOrder::PAIMON_BIG_ENDIAN);
+ output_stream.WriteValue<int8_t>(BitSliceIndexBitmapFileIndex::VERSION_1);
+ output_stream.WriteValue<int32_t>(static_cast<int32_t>(values_.size()));
+ const bool has_positive = positive->IsNotEmpty();
+ output_stream.WriteValue<bool>(has_positive);
+ if (has_positive) {
+ output_stream.WriteBytes(positive->Serialize(pool_));
+ }
+ const bool has_negative = negative->IsNotEmpty();
+ output_stream.WriteValue<bool>(has_negative);
+ if (has_negative) {
+ output_stream.WriteBytes(negative->Serialize(pool_));
+ }
+ return MemorySegmentUtils::CopyToBytes(output_stream.Segments(),
/*offset=*/0,
+
/*num_bytes=*/output_stream.CurrentSize(), pool_.get());
+}
+
Result<std::shared_ptr<FileIndexReader>>
BitSliceIndexBitmapFileIndex::CreateReader(
::ArrowSchema* c_arrow_schema, int32_t start, int32_t length,
const std::shared_ptr<InputStream>& input_stream,
diff --git
a/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h
b/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h
index e661ea2a..d7dfe092 100644
--- a/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h
+++ b/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h
@@ -22,15 +22,15 @@
#include <functional>
#include <map>
#include <memory>
+#include <optional>
#include <string>
#include <utility>
#include <vector>
-#include "arrow/c/bridge.h"
#include "arrow/type.h"
-#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/file_index/file_index_reader.h"
#include "paimon/file_index/file_index_result.h"
+#include "paimon/file_index/file_index_writer.h"
#include "paimon/file_index/file_indexer.h"
#include "paimon/predicate/literal.h"
#include "paimon/result.h"
@@ -54,14 +54,12 @@ class BitSliceIndexBitmapFileIndex : public FileIndexer {
const std::shared_ptr<MemoryPool>& pool) const override;
Result<std::shared_ptr<FileIndexWriter>> CreateWriter(
- ::ArrowSchema* arrow_schema, const std::shared_ptr<MemoryPool>& pool)
const override {
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::DataType>
arrow_type,
- arrow::ImportType(arrow_schema));
- return Status::NotImplemented("do not support index writer in bsi");
- }
+ ::ArrowSchema* arrow_schema, const std::shared_ptr<MemoryPool>& pool)
const override;
using ValueMapperType = std::function<Result<int64_t>(const Literal&
literal)>;
+ static constexpr int8_t VERSION_1 = 1;
+
private:
static Result<ValueMapperType> GetValueMapper(
const std::shared_ptr<arrow::DataType>& arrow_type);
@@ -74,9 +72,29 @@ class BitSliceIndexBitmapFileIndex : public FileIndexer {
}
return static_cast<int64_t>(literal.GetValue<T>());
}
+};
+
+class BitSliceIndexBitmapFileIndexWriter : public FileIndexWriter {
+ public:
+ BitSliceIndexBitmapFileIndexWriter(
+ const std::shared_ptr<arrow::Field>& field,
+ const BitSliceIndexBitmapFileIndex::ValueMapperType& value_mapper,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ Status AddBatch(::ArrowArray* batch) override;
+
+ Result<PAIMON_UNIQUE_PTR<Bytes>> SerializedBytes() const override;
private:
- static constexpr int8_t VERSION_1 = 1;
+ std::shared_ptr<arrow::DataType> struct_type_;
+ std::string field_name_;
+ BitSliceIndexBitmapFileIndex::ValueMapperType value_mapper_;
+ std::vector<std::optional<int64_t>> values_;
+ int64_t positive_min_ = 0;
+ int64_t positive_max_ = 0;
+ int64_t negative_min_ = 0;
+ int64_t negative_max_ = 0;
+ std::shared_ptr<MemoryPool> pool_;
};
class BitSliceIndexBitmapFileIndexReader
diff --git
a/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index_test.cpp
b/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index_test.cpp
index 128c1a69..760434e1 100644
---
a/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index_test.cpp
+++
b/src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index_test.cpp
@@ -19,9 +19,13 @@
#include "paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h"
#include <cstdint>
+#include <string>
#include <utility>
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/json_simple.h"
#include "gtest/gtest.h"
+#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/field_type_utils.h"
#include "paimon/data/timestamp.h"
#include "paimon/defs.h"
@@ -60,6 +64,24 @@ class BitSliceIndexBitmapIndexReaderTest : public
::testing::Test {
<< ", expected=" << RoaringBitmap32::From(expected).ToString();
}
+ Result<PAIMON_UNIQUE_PTR<Bytes>> WriteIndex(const
std::shared_ptr<arrow::DataType>& data_type,
+ const std::string& json) const
{
+ const std::shared_ptr<arrow::Schema> schema =
+ arrow::schema({arrow::field("f0", data_type)});
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(schema->fields()),
json)
+ .ValueOrDie();
+ ::ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema,
&c_schema));
+ BitSliceIndexBitmapFileIndex file_index({});
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileIndexWriter> writer,
+ file_index.CreateWriter(&c_schema, pool_));
+ ::ArrowArray c_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array));
+ PAIMON_RETURN_NOT_OK(writer->AddBatch(&c_array));
+ return writer->SerializedBytes();
+ }
+
private:
std::shared_ptr<MemoryPool> pool_;
};
@@ -75,48 +97,75 @@ TEST_F(BitSliceIndexBitmapIndexReaderTest, TestMix) {
0, 0, 0, 2, 58, 48, 0, 0, 1, 0, 0, 0, 0, 0, 2, 0, 16, 0, 0, 0,
3, 0, 4, 0, 5,
0, 0, 0, 0, 2, 58, 48, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 16, 0, 0,
0, 5, 0, 58, 48,
0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 16, 0, 0, 0, 3, 0, 4, 0};
+ const auto check_mix_reader = [this](const
std::shared_ptr<FileIndexReader>& reader) {
+ // test equal
+ CheckResult(reader->VisitEqual(Literal(2)).value(), {1, 7});
+ CheckResult(reader->VisitEqual(Literal(-2)).value(), {3, 4});
+ CheckResult(reader->VisitEqual(Literal(100)).value(), {});
+
+ // test not equal
+ CheckResult(reader->VisitNotEqual(Literal(2)).value(), {0, 3, 4, 5, 8,
9});
+ CheckResult(reader->VisitNotEqual(Literal(-2)).value(), {0, 1, 5, 7,
8, 9});
+ CheckResult(reader->VisitNotEqual(Literal(100)).value(), {0, 1, 3, 4,
5, 7, 8, 9});
+
+ // test in
+ CheckResult(reader->VisitIn({Literal(-1), Literal(1), Literal(2),
Literal(3)}).value(),
+ {0, 1, 5, 7});
+
+ // test not in
+ CheckResult(reader->VisitNotIn({Literal(-1), Literal(1), Literal(2),
Literal(3)}).value(),
+ {3, 4, 8, 9});
+
+ // test null
+ CheckResult(reader->VisitIsNull().value(), {2, 6, 10});
+
+ // test not null
+ CheckResult(reader->VisitIsNotNull().value(), {0, 1, 3, 4, 5, 7, 8,
9});
+
+ // test less than
+ CheckResult(reader->VisitLessThan(Literal(2)).value(), {0, 3, 4, 5,
8});
+ CheckResult(reader->VisitLessOrEqual(Literal(2)).value(), {0, 1, 3, 4,
5, 7, 8});
+ CheckResult(reader->VisitLessThan(Literal(-1)).value(), {3, 4});
+ CheckResult(reader->VisitLessOrEqual(Literal(-1)).value(), {3, 4, 5});
+
+ // test greater than
+ CheckResult(reader->VisitGreaterThan(Literal(-2)).value(), {0, 1, 5,
7, 8, 9});
+ CheckResult(reader->VisitGreaterOrEqual(Literal(-2)).value(), {0, 1,
3, 4, 5, 7, 8, 9});
+ CheckResult(reader->VisitGreaterThan(Literal(2)).value(), {9});
+ CheckResult(reader->VisitGreaterOrEqual(Literal(2)).value(), {1, 7,
9});
+ };
+
+ ASSERT_OK_AND_ASSIGN(
+ PAIMON_UNIQUE_PTR<Bytes> written_bytes,
+ WriteIndex(arrow::int32(),
+ R"([[1], [2], [null], [-2], [-2], [-1], [null], [2], [0],
[5], [null]])"));
+ auto written_stream =
+ std::make_shared<ByteArrayInputStream>(written_bytes->data(),
written_bytes->size());
+ BitSliceIndexBitmapFileIndex file_index({});
+ ASSERT_OK_AND_ASSIGN(auto written_reader,
+
file_index.CreateReader(CreateArrowSchema(arrow::int32()).get(),
+ /*start=*/0,
/*length=*/written_bytes->size(),
+ written_stream, pool_));
+ check_mix_reader(written_reader);
+
+ // Reading the Java-produced fixture below. C++ and Java may choose
different valid portable
+ // roaring containers for the same bitmap, so their complete byte streams
need not be identical.
auto input_stream =
std::make_shared<ByteArrayInputStream>(index_bytes.data(),
index_bytes.size());
- BitSliceIndexBitmapFileIndex file_index({});
ASSERT_OK_AND_ASSIGN(
- auto reader,
+ auto java_bytes_reader,
file_index.CreateReader(CreateArrowSchema(arrow::int32()).get(),
/*start=*/0, /*length=*/index_bytes.size(),
input_stream, pool_));
- // test equal
- CheckResult(reader->VisitEqual(Literal(2)).value(), {1, 7});
- CheckResult(reader->VisitEqual(Literal(-2)).value(), {3, 4});
- CheckResult(reader->VisitEqual(Literal(100)).value(), {});
-
- // test not equal
- CheckResult(reader->VisitNotEqual(Literal(2)).value(), {0, 3, 4, 5, 8, 9});
- CheckResult(reader->VisitNotEqual(Literal(-2)).value(), {0, 1, 5, 7, 8,
9});
- CheckResult(reader->VisitNotEqual(Literal(100)).value(), {0, 1, 3, 4, 5,
7, 8, 9});
-
- // test in
- CheckResult(reader->VisitIn({Literal(-1), Literal(1), Literal(2),
Literal(3)}).value(),
- {0, 1, 5, 7});
-
- // test not in
- CheckResult(reader->VisitNotIn({Literal(-1), Literal(1), Literal(2),
Literal(3)}).value(),
- {3, 4, 8, 9});
-
- // test null
- CheckResult(reader->VisitIsNull().value(), {2, 6, 10});
-
- // test not null
- CheckResult(reader->VisitIsNotNull().value(), {0, 1, 3, 4, 5, 7, 8, 9});
+ check_mix_reader(java_bytes_reader);
+}
- // test less than
- CheckResult(reader->VisitLessThan(Literal(2)).value(), {0, 3, 4, 5, 8});
- CheckResult(reader->VisitLessOrEqual(Literal(2)).value(), {0, 1, 3, 4, 5,
7, 8});
- CheckResult(reader->VisitLessThan(Literal(-1)).value(), {3, 4});
- CheckResult(reader->VisitLessOrEqual(Literal(-1)).value(), {3, 4, 5});
+TEST_F(BitSliceIndexBitmapIndexReaderTest,
TestWriterRejectsInt64MinAndUnsupportedType) {
+ ASSERT_NOK_WITH_MSG(WriteIndex(arrow::int64(),
R"([[-9223372036854775808]])"),
+ "bsi index does not support INT64_MIN for field 'f0'");
- // test greater than
- CheckResult(reader->VisitGreaterThan(Literal(-2)).value(), {0, 1, 5, 7, 8,
9});
- CheckResult(reader->VisitGreaterOrEqual(Literal(-2)).value(), {0, 1, 3, 4,
5, 7, 8, 9});
- CheckResult(reader->VisitGreaterThan(Literal(2)).value(), {9});
- CheckResult(reader->VisitGreaterOrEqual(Literal(2)).value(), {1, 7, 9});
+ BitSliceIndexBitmapFileIndex file_index({});
+
ASSERT_NOK_WITH_MSG(file_index.CreateWriter(CreateArrowSchema(arrow::boolean()).get(),
pool_),
+ "BitSliceIndexBitmapFileIndex only support");
}
TEST_F(BitSliceIndexBitmapIndexReaderTest, TestPositiveOnly) {
diff --git
a/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.cpp
b/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.cpp
index 5d0543be..88e081bd 100644
--- a/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.cpp
+++ b/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.cpp
@@ -60,18 +60,12 @@ Result<std::shared_ptr<FileIndexWriter>>
RangeBitmapFileIndex::CreateWriter(
return Status::Invalid(
"invalid schema for RangeBitmapFileIndexWriter, supposed to have
single field.");
}
- const auto arrow_field = arrow_schema_ptr->field(0);
- return RangeBitmapFileIndexWriter::Create(arrow_schema_ptr,
arrow_field->name(), options_,
- pool);
+ return RangeBitmapFileIndexWriter::Create(arrow_schema_ptr->field(0),
options_, pool);
}
Result<std::shared_ptr<RangeBitmapFileIndexWriter>>
RangeBitmapFileIndexWriter::Create(
- const std::shared_ptr<arrow::Schema>& arrow_schema, const std::string&
field_name,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<MemoryPool>& pool) {
- const auto field = arrow_schema->GetFieldByName(field_name);
- if (!field) {
- return Status::Invalid(fmt::format("Field not found in schema: {}",
field_name));
- }
+ const std::shared_ptr<arrow::Field>& field, const std::map<std::string,
std::string>& options,
+ const std::shared_ptr<MemoryPool>& pool) {
PAIMON_ASSIGN_OR_RAISE(FieldType field_type,
FieldTypeUtils::ConvertToFieldType(field->type()->id()));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeyFactory> shared_key_factory,
@@ -82,27 +76,18 @@ Result<std::shared_ptr<RangeBitmapFileIndexWriter>>
RangeBitmapFileIndexWriter::
chunk_size_it != options.end()) {
PAIMON_ASSIGN_OR_RAISE(parsed_chunk_size,
MemorySize::ParseBytes(chunk_size_it->second));
}
- auto struct_type = arrow::struct_({field});
+ std::shared_ptr<arrow::DataType> struct_type = arrow::struct_({field});
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<RangeBitmap::Appender> appender_ptr,
RangeBitmap::Appender::Create(shared_key_factory, parsed_chunk_size,
pool));
- return std::make_shared<RangeBitmapFileIndexWriter>(
- struct_type, field->type(), options, pool, shared_key_factory,
std::move(appender_ptr));
+ return std::make_shared<RangeBitmapFileIndexWriter>(struct_type, pool,
shared_key_factory,
+
std::move(appender_ptr));
}
Status RangeBitmapFileIndexWriter::AddBatch(::ArrowArray* batch) {
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
arrow::ImportArray(batch, struct_type_));
- if (!array || array->type_id() != arrow::Type::STRUCT) {
- return Status::Invalid(
- "invalid batch for RangeBitmapFileIndexWriter, expected a struct
array");
- }
auto struct_array = checked_pointer_cast<arrow::StructArray>(array);
- if (struct_array->num_fields() != 1) {
- return Status::Invalid(
- "invalid batch for RangeBitmapFileIndexWriter, expected a struct
array with exactly "
- "one field");
- }
PAIMON_ASSIGN_OR_RAISE(std::vector<Literal> array_values,
LiteralConverter::ConvertLiteralsFromArray(*(struct_array->field(0)),
/*own_data=*/true));
@@ -117,13 +102,9 @@ Result<PAIMON_UNIQUE_PTR<Bytes>>
RangeBitmapFileIndexWriter::SerializedBytes() c
}
RangeBitmapFileIndexWriter::RangeBitmapFileIndexWriter(
- const std::shared_ptr<arrow::DataType>& struct_type,
- const std::shared_ptr<arrow::DataType>& arrow_type,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<MemoryPool>& pool,
+ const std::shared_ptr<arrow::DataType>& struct_type, const
std::shared_ptr<MemoryPool>& pool,
const std::shared_ptr<KeyFactory>& key_factory,
std::unique_ptr<RangeBitmap::Appender> appender)
: struct_type_(struct_type),
- arrow_type_(arrow_type),
- options_(options),
pool_(pool),
key_factory_(key_factory),
appender_(std::move(appender)) {}
diff --git a/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.h
b/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.h
index fc99cac3..64289a53 100644
--- a/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.h
+++ b/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.h
@@ -61,15 +61,13 @@ class PAIMON_EXPORT RangeBitmapFileIndex final : public
FileIndexer {
class RangeBitmapFileIndexWriter final : public FileIndexWriter {
public:
static Result<std::shared_ptr<RangeBitmapFileIndexWriter>> Create(
- const std::shared_ptr<arrow::Schema>& arrow_schema, const std::string&
field_name,
+ const std::shared_ptr<arrow::Field>& field,
const std::map<std::string, std::string>& options, const
std::shared_ptr<MemoryPool>& pool);
Status AddBatch(::ArrowArray* batch) override;
Result<PAIMON_UNIQUE_PTR<Bytes>> SerializedBytes() const override;
RangeBitmapFileIndexWriter(const std::shared_ptr<arrow::DataType>&
struct_type,
- const std::shared_ptr<arrow::DataType>&
arrow_type,
- const std::map<std::string, std::string>&
options,
const std::shared_ptr<MemoryPool>& pool,
const std::shared_ptr<KeyFactory>& key_factory,
std::unique_ptr<RangeBitmap::Appender>
appender);
@@ -78,8 +76,6 @@ class RangeBitmapFileIndexWriter final : public
FileIndexWriter {
/// @note struct_type_ contains only one field with arrow_type_, used for
import from C
/// interface.
std::shared_ptr<arrow::DataType> struct_type_;
- std::shared_ptr<arrow::DataType> arrow_type_;
- std::map<std::string, std::string> options_;
std::shared_ptr<MemoryPool> pool_;
std::shared_ptr<KeyFactory> key_factory_;
std::unique_ptr<RangeBitmap::Appender> appender_;
diff --git
a/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index_test.cpp
b/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index_test.cpp
index e1479784..6157ee02 100644
--- a/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index_test.cpp
+++ b/src/paimon/common/file_index/rangebitmap/range_bitmap_file_index_test.cpp
@@ -123,17 +123,15 @@ Result<std::shared_ptr<RangeBitmapFileIndexReader>>
RangeBitmapFileIndexTest::Cr
std::shared_ptr<arrow::Array> arrow_array;
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Finish(&arrow_array));
// Wrap in StructArray (single field) as required by
RangeBitmapFileIndexWriter
- arrow::FieldVector fields = {arrow::field("test_field", arrow_type)};
+ auto field = arrow::field("test_field", arrow_type);
+ arrow::FieldVector fields = {field};
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray>
struct_array,
arrow::StructArray::Make({arrow_array},
fields));
auto c_array = std::make_unique<::ArrowArray>();
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*struct_array,
c_array.get()));
- // Create schema for the field
- const auto schema = arrow::schema({arrow::field("test_field",
arrow_type)});
// Create writer
- PAIMON_ASSIGN_OR_RAISE(
- std::shared_ptr<RangeBitmapFileIndexWriter> writer,
- RangeBitmapFileIndexWriter::Create(schema, "test_field", options,
pool_));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RangeBitmapFileIndexWriter> writer,
+ RangeBitmapFileIndexWriter::Create(field, options,
pool_));
// Add the batch
PAIMON_RETURN_NOT_OK(writer->AddBatch(c_array.get()));
// Get serialized payload
diff --git a/src/paimon/common/file_index/rangebitmap/range_bitmap_io_test.cpp
b/src/paimon/common/file_index/rangebitmap/range_bitmap_io_test.cpp
index e2bb867e..fd8e0c88 100644
--- a/src/paimon/common/file_index/rangebitmap/range_bitmap_io_test.cpp
+++ b/src/paimon/common/file_index/rangebitmap/range_bitmap_io_test.cpp
@@ -65,18 +65,16 @@ class RangeBitmapIoTest : public ::testing::Test {
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Finish(&arrow_array));
// Wrap in StructArray
- arrow::FieldVector fields = {arrow::field("test_field", arrow_type)};
+ const std::shared_ptr<arrow::Field> field = arrow::field("test_field",
arrow_type);
+ arrow::FieldVector fields = {field};
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray>
struct_array,
arrow::StructArray::Make({arrow_array}, fields));
auto c_array = std::make_unique<::ArrowArray>();
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*struct_array,
c_array.get()));
- // Create schema
- const auto schema = arrow::schema({arrow::field("test_field",
arrow_type)});
-
// Create writer and write data
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RangeBitmapFileIndexWriter>
writer,
- RangeBitmapFileIndexWriter::Create(schema,
"test_field", {}, pool_));
+ RangeBitmapFileIndexWriter::Create(field, {},
pool_));
PAIMON_RETURN_NOT_OK(writer->AddBatch(c_array.get()));
return writer->SerializedBytes();
diff --git a/src/paimon/common/lookup/lookup_store_factory.cpp
b/src/paimon/common/lookup/lookup_store_factory.cpp
index c5742e12..d6cb8d40 100644
--- a/src/paimon/common/lookup/lookup_store_factory.cpp
+++ b/src/paimon/common/lookup/lookup_store_factory.cpp
@@ -36,7 +36,8 @@ Result<std::shared_ptr<BloomFilter>>
LookupStoreFactory::BfGenerator(int64_t row
if (row_count <= 0 || !options.LookupCacheBloomFilterEnabled()) {
return std::shared_ptr<BloomFilter>();
}
- auto bloom_filter = BloomFilter::Create(row_count,
options.GetLookupCacheBloomFilterFpp());
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<BloomFilter> bloom_filter,
+ BloomFilter::Create(row_count,
options.GetLookupCacheBloomFilterFpp()));
MemorySegment memory_segment =
MemorySegment::AllocateHeapMemory(bloom_filter->ByteLength(), pool);
PAIMON_RETURN_NOT_OK(bloom_filter->SetMemorySegment(memory_segment));
diff --git a/src/paimon/common/sst/sst_file_io_test.cpp
b/src/paimon/common/sst/sst_file_io_test.cpp
index eac9a8a8..54d4b1b1 100644
--- a/src/paimon/common/sst/sst_file_io_test.cpp
+++ b/src/paimon/common/sst/sst_file_io_test.cpp
@@ -93,7 +93,7 @@ TEST_P(SstFileIOTest, TestSimple) {
fs_->Create(index_path, /*overwrite=*/false));
// write data
- auto bf = BloomFilter::Create(30, 0.01);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BloomFilter> bf,
BloomFilter::Create(30, 0.01));
auto seg_for_bf = MemorySegment::AllocateHeapMemory(bf->ByteLength(),
pool_.get());
ASSERT_OK(bf->SetMemorySegment(seg_for_bf));
auto writer = std::make_shared<SstFileWriter>(out, bf, 50, factory, pool_);
@@ -234,7 +234,7 @@ TEST_F(SstFileIOTest, TestIOException) {
CHECK_HOOK_STATUS(out_result.status(), i);
std::shared_ptr<OutputStream> out = std::move(out_result).value();
- auto bf = BloomFilter::Create(30, 0.01);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BloomFilter> bf,
BloomFilter::Create(30, 0.01));
MemorySegment seg_for_bf =
MemorySegment::AllocateHeapMemory(bf->ByteLength(), pool_.get());
ASSERT_OK(bf->SetMemorySegment(seg_for_bf));
auto writer = std::make_shared<SstFileWriter>(out, bf, 50, factory,
pool_);
diff --git a/src/paimon/common/utils/bloom_filter.cpp
b/src/paimon/common/utils/bloom_filter.cpp
index d66420f4..ba9fa4b4 100644
--- a/src/paimon/common/utils/bloom_filter.cpp
+++ b/src/paimon/common/utils/bloom_filter.cpp
@@ -25,29 +25,35 @@
namespace paimon {
-int32_t BloomFilter::OptimalNumOfBits(int64_t expect_entries, double fpp) {
- if (expect_entries <= 0 || fpp <= 0.0 || fpp >= 1.0) {
+int32_t BloomFilter::OptimalNumOfBits(int64_t expected_entries, double fpp) {
+ if (expected_entries <= 0 || fpp <= 0.0 || fpp >= 1.0) {
return 0;
}
- double result = -static_cast<double>(expect_entries) * log(fpp) / (log(2)
* log(2));
+ double result = -static_cast<double>(expected_entries) * log(fpp) /
(log(2) * log(2));
if (result > INT32_MAX) return INT32_MAX;
if (result < 0) return 0;
return static_cast<int32_t>(result);
}
-int32_t BloomFilter::OptimalNumOfHashFunctions(int64_t expect_entries, int64_t
bit_size) {
- if (expect_entries <= 0) {
+int32_t BloomFilter::OptimalNumOfHashFunctions(int64_t expected_entries,
int64_t bit_size) {
+ if (expected_entries <= 0) {
return 1;
}
- double ratio = static_cast<double>(bit_size) /
static_cast<double>(expect_entries);
+ double ratio = static_cast<double>(bit_size) /
static_cast<double>(expected_entries);
double result = ratio * std::log(2.0);
return std::max(1, static_cast<int32_t>(std::round(result)));
}
-std::shared_ptr<BloomFilter> BloomFilter::Create(int64_t expect_entries,
double fpp) {
+Result<std::shared_ptr<BloomFilter>> BloomFilter::Create(int64_t
expected_entries, double fpp) {
+ if (expected_entries <= 0) {
+ return Status::Invalid("expected entries must be greater than 0 for
bloom filter");
+ }
+ if (!std::isfinite(fpp) || fpp <= 0.0 || fpp >= 1.0) {
+ return Status::Invalid("fpp must be greater than 0 and less than 1 for
bloom filter");
+ }
auto bytes =
-
static_cast<int32_t>(ceil(BloomFilter::OptimalNumOfBits(expect_entries, fpp) /
8.0));
- return std::make_shared<BloomFilter>(expect_entries, bytes);
+
static_cast<int32_t>(ceil(BloomFilter::OptimalNumOfBits(expected_entries, fpp)
/ 8.0));
+ return std::make_shared<BloomFilter>(expected_entries, bytes);
}
BloomFilter::BloomFilter(int64_t expected_entries, int32_t byte_length)
diff --git a/src/paimon/common/utils/bloom_filter.h
b/src/paimon/common/utils/bloom_filter.h
index 8f1b3271..61e27af5 100644
--- a/src/paimon/common/utils/bloom_filter.h
+++ b/src/paimon/common/utils/bloom_filter.h
@@ -22,7 +22,7 @@
#include <memory>
#include "paimon/common/utils/bit_set.h"
-#include "paimon/memory/bytes.h"
+#include "paimon/result.h"
#include "paimon/visibility.h"
namespace paimon {
@@ -30,9 +30,9 @@ namespace paimon {
/// Bloom filter based on MemorySegment.
class PAIMON_EXPORT BloomFilter {
public:
- static int32_t OptimalNumOfBits(int64_t expect_entries, double fpp);
- static int32_t OptimalNumOfHashFunctions(int64_t expect_entries, int64_t
bit_size);
- static std::shared_ptr<BloomFilter> Create(int64_t expect_entries, double
fpp);
+ static int32_t OptimalNumOfBits(int64_t expected_entries, double fpp);
+ static int32_t OptimalNumOfHashFunctions(int64_t expected_entries, int64_t
bit_size);
+ static Result<std::shared_ptr<BloomFilter>> Create(int64_t
expected_entries, double fpp);
public:
BloomFilter(int64_t expected_entries, int32_t byte_length);
diff --git a/src/paimon/common/utils/bloom_filter64.cpp
b/src/paimon/common/utils/bloom_filter64.cpp
index f2685a13..7a975161 100644
--- a/src/paimon/common/utils/bloom_filter64.cpp
+++ b/src/paimon/common/utils/bloom_filter64.cpp
@@ -21,6 +21,8 @@
#include <algorithm>
#include <cassert>
#include <cmath>
+#include <cstring>
+#include <limits>
#include <utility>
#include "paimon/memory/bytes.h"
@@ -47,17 +49,42 @@ bool BloomFilter64::BitSet::Get(int32_t index) const {
}
int32_t BloomFilter64::BitSet::BitSize() const {
- return (bytes_->size() - offset_) * BloomFilter64::BYTE_SIZE;
+ return ByteLength() * BloomFilter64::BYTE_SIZE;
}
-BloomFilter64::BloomFilter64(int64_t items, double fpp, const
std::shared_ptr<MemoryPool>& pool)
- : pool_(pool) {
- auto nb = static_cast<int32_t>(-items * std::log(fpp) / (std::log(2) *
std::log(2)));
- num_bits_ = nb + (BloomFilter64::BYTE_SIZE - (nb %
BloomFilter64::BYTE_SIZE));
- num_hash_functions_ = std::max(
- 1, static_cast<int32_t>(std::round(static_cast<double>(num_bits_) /
items * std::log(2))));
- auto bytes = std::make_shared<Bytes>(num_bits_ / BloomFilter64::BYTE_SIZE,
pool_.get());
- bit_set_ = std::make_unique<BitSet>(bytes, /*offset=*/0);
+int32_t BloomFilter64::BitSet::ByteLength() const {
+ return static_cast<int32_t>(bytes_->size() - offset_);
+}
+
+void BloomFilter64::BitSet::ToByteArray(int32_t offset, int32_t length, char*
bytes) const {
+ assert(bytes);
+ assert(offset >= 0);
+ assert(length >= 0);
+ assert(static_cast<size_t>(offset_ + length) <= bytes_->size());
+ std::memcpy(bytes + offset, bytes_->data() + offset_, length);
+}
+
+Result<BloomFilter64> BloomFilter64::Create(int64_t items, double fpp,
+ const std::shared_ptr<MemoryPool>&
pool) {
+ if (items <= 0) {
+ return Status::Invalid("items must be greater than 0 for bloom
filter");
+ }
+ if (!std::isfinite(fpp) || fpp <= 0.0 || fpp >= 1.0) {
+ return Status::Invalid("fpp must be greater than 0 and less than 1 for
bloom filter");
+ }
+ const double log_two = std::log(2);
+ const double estimated_bits = -static_cast<double>(items) * std::log(fpp)
/ (log_two * log_two);
+ if (estimated_bits > std::numeric_limits<int32_t>::max() - BYTE_SIZE) {
+ return Status::Invalid("bloom filter size exceeds the supported
range");
+ }
+ const auto num_bits_without_padding = static_cast<int32_t>(estimated_bits);
+ const int32_t num_bits =
+ num_bits_without_padding + (BYTE_SIZE - (num_bits_without_padding %
BYTE_SIZE));
+ const int32_t num_hash_functions = std::max(
+ 1, static_cast<int32_t>(std::round(static_cast<double>(num_bits) /
items * log_two)));
+ auto bytes = std::make_shared<Bytes>(num_bits / BYTE_SIZE, pool.get());
+ auto bit_set = std::make_unique<BitSet>(bytes, /*offset=*/0);
+ return BloomFilter64(num_hash_functions, std::move(bit_set), pool);
}
BloomFilter64::BloomFilter64(int32_t num_hash_functions,
std::unique_ptr<BitSet>&& bit_set)
@@ -65,6 +92,13 @@ BloomFilter64::BloomFilter64(int32_t num_hash_functions,
std::unique_ptr<BitSet>
num_hash_functions_(num_hash_functions),
bit_set_(std::move(bit_set)) {}
+BloomFilter64::BloomFilter64(int32_t num_hash_functions,
std::unique_ptr<BitSet>&& bit_set,
+ const std::shared_ptr<MemoryPool>& pool)
+ : num_bits_(bit_set->BitSize()),
+ num_hash_functions_(num_hash_functions),
+ pool_(pool),
+ bit_set_(std::move(bit_set)) {}
+
void BloomFilter64::AddHash(int64_t hash64) {
auto hash1 = static_cast<int32_t>(hash64);
auto hash2 = static_cast<int32_t>(static_cast<uint64_t>(hash64) >> 32);
diff --git a/src/paimon/common/utils/bloom_filter64.h
b/src/paimon/common/utils/bloom_filter64.h
index 3ba4e86f..2440b303 100644
--- a/src/paimon/common/utils/bloom_filter64.h
+++ b/src/paimon/common/utils/bloom_filter64.h
@@ -22,6 +22,7 @@
#include <memory>
#include "paimon/memory/bytes.h"
+#include "paimon/result.h"
#include "paimon/visibility.h"
namespace paimon {
@@ -31,7 +32,9 @@ class MemoryPool;
/// Bloom filter 64 handle 64 bits hash.
class PAIMON_EXPORT BloomFilter64 {
public:
- BloomFilter64(int64_t items, double fpp, const
std::shared_ptr<MemoryPool>& pool);
+ static Result<BloomFilter64> Create(int64_t items, double fpp,
+ const std::shared_ptr<MemoryPool>&
pool);
+
class BitSet;
BloomFilter64(int32_t num_hash_functions, std::unique_ptr<BitSet>&&
bit_set);
@@ -54,6 +57,8 @@ class PAIMON_EXPORT BloomFilter64 {
void Set(int32_t index);
bool Get(int32_t index) const;
int32_t BitSize() const;
+ int32_t ByteLength() const;
+ void ToByteArray(int32_t offset, int32_t length, char* bytes) const;
private:
static constexpr int8_t MASK = 0x07;
@@ -64,9 +69,11 @@ class PAIMON_EXPORT BloomFilter64 {
};
private:
+ BloomFilter64(int32_t num_hash_functions, std::unique_ptr<BitSet>&&
bit_set,
+ const std::shared_ptr<MemoryPool>& pool);
+
static constexpr int32_t BYTE_SIZE = 8;
- private:
int32_t num_bits_ = -1;
int32_t num_hash_functions_ = -1;
std::shared_ptr<MemoryPool> pool_;
diff --git a/src/paimon/common/utils/bloom_filter64_test.cpp
b/src/paimon/common/utils/bloom_filter64_test.cpp
index aa548fb7..6f60fd01 100644
--- a/src/paimon/common/utils/bloom_filter64_test.cpp
+++ b/src/paimon/common/utils/bloom_filter64_test.cpp
@@ -28,13 +28,14 @@
#include "gtest/gtest.h"
#include "paimon/memory/bytes.h"
#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
TEST(BloomFilter64Test, TestSimple) {
int32_t items = 10000;
auto pool = GetDefaultPool();
- BloomFilter64 bloom_filter(items, 0.02, pool);
+ ASSERT_OK_AND_ASSIGN(BloomFilter64 bloom_filter,
BloomFilter64::Create(items, 0.02, pool));
std::mt19937_64 engine(std::random_device{}()); //
NOLINT(whitespace/braces)
std::uniform_int_distribution<int64_t>
distribution(std::numeric_limits<int64_t>::min(),
std::numeric_limits<int64_t>::max());
@@ -61,6 +62,39 @@ TEST(BloomFilter64Test, TestSimple) {
ASSERT_TRUE(static_cast<double>(false_positives) / num < 0.03);
}
+TEST(BloomFilter64Test, TestInvalidItemsAndFpp) {
+ std::shared_ptr<MemoryPool> pool = GetDefaultPool();
+
+ ASSERT_NOK_WITH_MSG(BloomFilter64::Create(/*items=*/0, /*fpp=*/0.1, pool),
+ "items must be greater than 0");
+ ASSERT_NOK_WITH_MSG(BloomFilter64::Create(/*items=*/-1, /*fpp=*/0.1, pool),
+ "items must be greater than 0");
+ ASSERT_NOK_WITH_MSG(BloomFilter64::Create(/*items=*/100, /*fpp=*/0.0,
pool),
+ "fpp must be greater than 0 and less than 1");
+ ASSERT_NOK_WITH_MSG(BloomFilter64::Create(/*items=*/100, /*fpp=*/-0.1,
pool),
+ "fpp must be greater than 0 and less than 1");
+ ASSERT_NOK_WITH_MSG(BloomFilter64::Create(/*items=*/100, /*fpp=*/1.0,
pool),
+ "fpp must be greater than 0 and less than 1");
+ ASSERT_NOK_WITH_MSG(BloomFilter64::Create(/*items=*/100, /*fpp=*/1.1,
pool),
+ "fpp must be greater than 0 and less than 1");
+ ASSERT_NOK_WITH_MSG(
+ BloomFilter64::Create(/*items=*/std::numeric_limits<int64_t>::max(),
/*fpp=*/0.1, pool),
+ "bloom filter size exceeds the supported range");
+}
+
+TEST(BloomFilter64Test, TestKeepMemoryPoolAlive) {
+ std::weak_ptr<MemoryPool> weak_pool;
+ {
+ std::shared_ptr<MemoryPool> pool(GetMemoryPool());
+ weak_pool = pool;
+ ASSERT_OK_AND_ASSIGN(BloomFilter64 bloom_filter,
+ BloomFilter64::Create(/*items=*/100, /*fpp=*/0.1,
pool));
+ pool.reset();
+ ASSERT_FALSE(weak_pool.expired());
+ }
+ ASSERT_TRUE(weak_pool.expired());
+}
+
TEST(BloomFilter64Test, TestCompatibleWithJava) {
// data: -10, -5, 0, 13, 100, 200, 500
std::vector<uint8_t> se_bytes = {241, 255, 17, 0, 0, 0, 0, 0, 0, 0, 0, 0};
@@ -74,10 +108,10 @@ TEST(BloomFilter64Test, TestCompatibleWithJava) {
ASSERT_TRUE(bloom_filter.TestHash(value));
}
- BloomFilter64 bloom_filter2(10, 0.01, pool);
+ ASSERT_OK_AND_ASSIGN(BloomFilter64 bloom_filter2,
BloomFilter64::Create(10, 0.01, pool));
ASSERT_EQ(7, bloom_filter2.GetNumHashFunctions());
- ASSERT_EQ(se_bytes.size() * BloomFilter64::BYTE_SIZE,
bloom_filter2.num_bits_);
- ASSERT_EQ(se_bytes.size(), bloom_filter2.GetBitSet().bytes_->size());
+ ASSERT_EQ(se_bytes.size() * 8, bloom_filter2.GetBitSet().BitSize());
+ ASSERT_EQ(se_bytes.size(), bloom_filter2.GetBitSet().ByteLength());
}
} // namespace paimon::test
diff --git a/src/paimon/common/utils/bloom_filter_test.cpp
b/src/paimon/common/utils/bloom_filter_test.cpp
index fdd6abb3..0832fdf5 100644
--- a/src/paimon/common/utils/bloom_filter_test.cpp
+++ b/src/paimon/common/utils/bloom_filter_test.cpp
@@ -35,7 +35,8 @@ namespace paimon::test {
TEST(BloomFilterTest, TestOneSegmentBuilder) {
int32_t items = 100;
auto pool = GetDefaultPool();
- auto bloom_filter = BloomFilter::Create(items, 0.01);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BloomFilter> bloom_filter,
+ BloomFilter::Create(items, 0.01));
auto seg = MemorySegment::AllocateHeapMemory(1024, pool.get());
ASSERT_OK(bloom_filter->SetMemorySegment(seg));
@@ -54,12 +55,32 @@ TEST(BloomFilterTest, TestOneSegmentBuilder) {
}
TEST(BloomFilterTest, TestEstimatedHashFunctions) {
- ASSERT_EQ(7, BloomFilter::Create(1000, 0.01)->GetNumHashFunctions());
- ASSERT_EQ(7, BloomFilter::Create(10000, 0.01)->GetNumHashFunctions());
- ASSERT_EQ(7, BloomFilter::Create(100000, 0.01)->GetNumHashFunctions());
- ASSERT_EQ(4, BloomFilter::Create(100000, 0.05)->GetNumHashFunctions());
- ASSERT_EQ(7, BloomFilter::Create(1000000, 0.01)->GetNumHashFunctions());
- ASSERT_EQ(4, BloomFilter::Create(1000000, 0.05)->GetNumHashFunctions());
+ auto get_num_hash_functions = [](int64_t expected_entries, double fpp) {
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<BloomFilter> bloom_filter,
+ BloomFilter::Create(expected_entries, fpp));
+ return bloom_filter->GetNumHashFunctions();
+ };
+ ASSERT_EQ(7, get_num_hash_functions(1000, 0.01));
+ ASSERT_EQ(7, get_num_hash_functions(10000, 0.01));
+ ASSERT_EQ(7, get_num_hash_functions(100000, 0.01));
+ ASSERT_EQ(4, get_num_hash_functions(100000, 0.05));
+ ASSERT_EQ(7, get_num_hash_functions(1000000, 0.01));
+ ASSERT_EQ(4, get_num_hash_functions(1000000, 0.05));
+}
+
+TEST(BloomFilterTest, TestInvalidExpectedEntriesAndFpp) {
+ ASSERT_NOK_WITH_MSG(BloomFilter::Create(/*expected_entries=*/0,
/*fpp=*/0.1),
+ "expected entries must be greater than 0");
+ ASSERT_NOK_WITH_MSG(BloomFilter::Create(/*expected_entries=*/-1,
/*fpp=*/0.1),
+ "expected entries must be greater than 0");
+ ASSERT_NOK_WITH_MSG(BloomFilter::Create(/*expected_entries=*/100,
/*fpp=*/0.0),
+ "fpp must be greater than 0 and less than 1");
+ ASSERT_NOK_WITH_MSG(BloomFilter::Create(/*expected_entries=*/100,
/*fpp=*/-0.1),
+ "fpp must be greater than 0 and less than 1");
+ ASSERT_NOK_WITH_MSG(BloomFilter::Create(/*expected_entries=*/100,
/*fpp=*/1.0),
+ "fpp must be greater than 0 and less than 1");
+ ASSERT_NOK_WITH_MSG(BloomFilter::Create(/*expected_entries=*/100,
/*fpp=*/1.1),
+ "fpp must be greater than 0 and less than 1");
}
TEST(BloomFilterTest, TestBloomNumBits) {
diff --git a/src/paimon/common/utils/math.h b/src/paimon/common/utils/math.h
index 54ad6cf7..6aba6523 100644
--- a/src/paimon/common/utils/math.h
+++ b/src/paimon/common/utils/math.h
@@ -36,6 +36,7 @@
#include "fmt/format.h"
#include "paimon/common/utils/options_utils.h"
+#include "paimon/io/byte_order.h"
#include "paimon/status.h"
namespace paimon {
@@ -136,4 +137,30 @@ inline T EndianSwapValue(T v) {
}
}
+template <typename T>
+inline T ToBigEndian(T value) {
+ if constexpr (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
+ return EndianSwapValue(value);
+ }
+ return value;
+}
+
+template <typename T>
+inline T ToLittleEndian(T value) {
+ if constexpr (SystemByteOrder() == ByteOrder::PAIMON_BIG_ENDIAN) {
+ return EndianSwapValue(value);
+ }
+ return value;
+}
+
+template <typename T>
+inline T FromBigEndian(T value) {
+ return ToBigEndian(value);
+}
+
+template <typename T>
+inline T FromLittleEndian(T value) {
+ return ToLittleEndian(value);
+}
+
} // namespace paimon
diff --git a/src/paimon/common/utils/math_test.cpp
b/src/paimon/common/utils/math_test.cpp
index 36df1cba..49d31d47 100644
--- a/src/paimon/common/utils/math_test.cpp
+++ b/src/paimon/common/utils/math_test.cpp
@@ -19,6 +19,8 @@
#include "paimon/common/utils/math.h"
+#include <array>
+#include <cstring>
#include <limits>
#include "gtest/gtest.h"
@@ -44,6 +46,22 @@ TEST(MathTest, EndianSwapValue) {
ASSERT_EQ(swapped64, 0xF0DEBC9A78563412);
}
+TEST(MathTest, ToEndian) {
+ constexpr uint32_t kValue = 0x12345678;
+
+ const uint32_t big_endian = ToBigEndian(kValue);
+ std::array<uint8_t, sizeof(big_endian)> big_endian_bytes{};
+ std::memcpy(big_endian_bytes.data(), &big_endian, sizeof(big_endian));
+ ASSERT_EQ((std::array<uint8_t, 4>{0x12, 0x34, 0x56, 0x78}),
big_endian_bytes);
+ ASSERT_EQ(kValue, FromBigEndian(big_endian));
+
+ const uint32_t little_endian = ToLittleEndian(kValue);
+ std::array<uint8_t, sizeof(little_endian)> little_endian_bytes{};
+ std::memcpy(little_endian_bytes.data(), &little_endian,
sizeof(little_endian));
+ ASSERT_EQ((std::array<uint8_t, 4>{0x78, 0x56, 0x34, 0x12}),
little_endian_bytes);
+ ASSERT_EQ(kValue, FromLittleEndian(little_endian));
+}
+
TEST(MathTest, InRange) {
// signed -> unsigned: negative values out of range, boundary values in
range
ASSERT_TRUE(InRange<uint32_t>(0));
diff --git a/src/paimon/core/io/data_file_index_writer_test.cpp
b/src/paimon/core/io/data_file_index_writer_test.cpp
index e9ab7940..1e2c9593 100644
--- a/src/paimon/core/io/data_file_index_writer_test.cpp
+++ b/src/paimon/core/io/data_file_index_writer_test.cpp
@@ -197,11 +197,38 @@ TEST_F(DataFileIndexWriterTest,
TestExternalIndexAndAbortCleanup) {
ASSERT_FALSE(exists);
}
+TEST_F(DataFileIndexWriterTest, TestBsiAndBloomFilterEmbeddedRoundTrip) {
+ ASSERT_OK_AND_ASSIGN(auto writer,
+ CreateWriter({{"file-index.bsi.columns", "f0"},
+ {"file-index.bloom-filter.columns",
"f1"},
+ {"file-index.bloom-filter.f1.items",
"100"},
+ {"file-index.bloom-filter.f1.fpp",
"0.01"},
+
{Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1MB"}}));
+ ASSERT_OK(writer->AddBatch(CreateBatch(R"([{"f0": 1, "f1": 10},
+ {"f0": -2, "f1": 20}])")));
+ ASSERT_OK(writer->AddBatch(CreateBatch(R"([{"f0": null, "f1": 30},
+ {"f0": 5, "f1": 40}])")));
+
+ ASSERT_OK_AND_ASSIGN(FileIndexWriteResult result,
writer->Finish("unused.orc"));
+ ASSERT_TRUE(result.embedded_index);
+ ASSERT_OK_AND_ASSIGN(auto reader, CreateReader(result.embedded_index));
+
+ ASSERT_OK_AND_ASSIGN(auto bsi_readers, ReadColumn(reader.get(), "f0"));
+ ASSERT_EQ(1, bsi_readers.size());
+ ASSERT_OK_AND_ASSIGN(auto greater_result,
bsi_readers[0]->VisitGreaterThan(Literal(1)));
+ ASSERT_EQ("{3}", greater_result->ToString());
+ ASSERT_OK_AND_ASSIGN(auto null_result, bsi_readers[0]->VisitIsNull());
+ ASSERT_EQ("{2}", null_result->ToString());
+
+ ASSERT_OK_AND_ASSIGN(auto bloom_readers, ReadColumn(reader.get(), "f1"));
+ ASSERT_EQ(1, bloom_readers.size());
+ ASSERT_OK_AND_ASSIGN(auto present_result,
bloom_readers[0]->VisitEqual(Literal(30)));
+ ASSERT_TRUE(present_result->IsRemain().value());
+}
+
TEST_F(DataFileIndexWriterTest, TestUnavailableWriterFailsCreation) {
ASSERT_NOK_WITH_MSG(CreateWriter({{"file-index.unknown.columns", "f0"}}),
"File index type 'unknown' is not registered");
- ASSERT_NOK_WITH_MSG(CreateWriter({{"file-index.bloom-filter.columns",
"f0"}}),
- "do not support index writer in bloom filter");
}
TEST_F(DataFileIndexWriterTest, TestRejectSystemFieldIndex) {
