This is an automated email from the ASF dual-hosted git repository.
SteNicholas 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 d688d399 feat(global-index): add bitmap index format (#352)
d688d399 is described below
commit d688d3992a2df2e592843beaf533d587b2d11ff9
Author: lxy <[email protected]>
AuthorDate: Thu Sep 24 18:17:46 2026 +0800
feat(global-index): add bitmap index format (#352)
---
src/paimon/CMakeLists.txt | 2 +
src/paimon/common/global_index/CMakeLists.txt | 3 +
.../bitmap/bitmap_global_index_format.cpp | 583 +++++++++++++++++++++
.../bitmap/bitmap_global_index_format.h | 303 +++++++++++
.../bitmap/bitmap_global_index_format_test.cpp | 143 +++++
.../bitmap/bitmap_global_index_writer.cpp | 182 +++++++
.../bitmap/bitmap_global_index_writer.h | 97 ++++
.../global_index/bitmap/bitmap_index_reader.cpp | 579 ++++++++++++++++++++
.../global_index/bitmap/bitmap_index_reader.h | 160 ++++++
.../bitmap/bitmap_index_reader_test.cpp | 172 ++++++
src/paimon/common/io/data_output_stream.cpp | 6 +-
src/paimon/common/io/data_output_stream.h | 4 +-
src/paimon/common/utils/var_length_int_utils.h | 37 ++
.../common/utils/var_length_int_utils_test.cpp | 50 ++
14 files changed, 2319 insertions(+), 2 deletions(-)
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index 6fc99343..eb17538d 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -617,6 +617,8 @@ if(PAIMON_BUILD_TESTS)
common/global_index/bitmap_global_index_result_test.cpp
common/global_index/bitmap_scored_global_index_result_test.cpp
common/global_index/bitmap/bitmap_global_index_test.cpp
+
common/global_index/bitmap/bitmap_global_index_format_test.cpp
+ common/global_index/bitmap/bitmap_index_reader_test.cpp
common/global_index/sorted_index_file_meta_test.cpp
common/global_index/btree/btree_file_footer_test.cpp
common/global_index/key_serializer_test.cpp
diff --git a/src/paimon/common/global_index/CMakeLists.txt
b/src/paimon/common/global_index/CMakeLists.txt
index 18806d58..4a0e92fa 100644
--- a/src/paimon/common/global_index/CMakeLists.txt
+++ b/src/paimon/common/global_index/CMakeLists.txt
@@ -16,6 +16,9 @@
set(PAIMON_GLOBAL_INDEX_SRC
bitmap/bitmap_global_index.cpp
+ bitmap/bitmap_global_index_format.cpp
+ bitmap/bitmap_global_index_writer.cpp
+ bitmap/bitmap_index_reader.cpp
bitmap/bitmap_global_index_factory.cpp
btree/btree_file_footer.cpp
btree/btree_global_index_factory.cpp
diff --git
a/src/paimon/common/global_index/bitmap/bitmap_global_index_format.cpp
b/src/paimon/common/global_index/bitmap/bitmap_global_index_format.cpp
new file mode 100644
index 00000000..279ac27a
--- /dev/null
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_format.cpp
@@ -0,0 +1,583 @@
+/*
+ * 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/global_index/bitmap/bitmap_global_index_format.h"
+
+#include <algorithm>
+#include <limits>
+
+#include "fmt/format.h"
+#include "paimon/common/global_index/key_serializer.h"
+#include "paimon/common/io/data_output_stream.h"
+#include "paimon/common/memory/memory_slice_input.h"
+#include "paimon/common/memory/memory_slice_output.h"
+#include "paimon/common/sst/block_trailer.h"
+#include "paimon/common/sst/sst_file_utils.h"
+#include "paimon/common/utils/crc32c.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/common/utils/var_length_int_utils.h"
+#include "paimon/io/byte_order.h"
+#include "paimon/predicate/literal.h"
+
+namespace paimon {
+namespace {
+
+Status WriteAll(const char* data, int64_t length, OutputStream* output_stream)
{
+ PAIMON_ASSIGN_OR_RAISE(int64_t written, output_stream->Write(data,
length));
+ if (written != length) {
+ return Status::IOError(
+ fmt::format("Failed to write bitmap global index block: expected
{} bytes, wrote {}.",
+ length, written));
+ }
+ return Status::OK();
+}
+
+Result<std::shared_ptr<Bytes>> ReadKey(MemorySliceInput* input, MemoryPool*
pool) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t key_length,
VarLengthIntUtils::ReadVarLenInt(input));
+ if (key_length > input->Available()) {
+ return Status::Invalid(
+ fmt::format("Bitmap dictionary key length {} exceeds remaining
block size {}.",
+ key_length, input->Available()));
+ }
+ return input->ReadSliceView(key_length).CopyBytes(pool);
+}
+
+Status ValidateBlockInfo(const BitmapGlobalIndexFormat::BlockInfo& block) {
+ if (block.Offset() < 0) {
+ return Status::Invalid("Invalid negative bitmap block offset.");
+ }
+ if (block.Length() < 0) {
+ return Status::Invalid("Invalid negative bitmap block length.");
+ }
+ return Status::OK();
+}
+
+} // namespace
+
+Result<BitmapGlobalIndexFormat::SerializedKey>
BitmapGlobalIndexFormat::SerializedKey::FromLiteral(
+ const std::shared_ptr<KeySerializer>& serializer, const Literal& literal) {
+ if (serializer == nullptr) {
+ return Status::Invalid("Cannot serialize a bitmap dictionary key
without KeySerializer.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
serializer->Serialize(literal));
+ return SerializedKey(std::move(bytes));
+}
+
+int32_t BitmapGlobalIndexFormat::SerializedKey::CompareTo(const SerializedKey&
other) const {
+ if (bytes_ == nullptr || other.bytes_ == nullptr) {
+ if (bytes_ == other.bytes_) {
+ return 0;
+ }
+ return bytes_ == nullptr ? -1 : 1;
+ }
+ size_t compare_length = std::min(bytes_->size(), other.bytes_->size());
+ for (size_t i = 0; i < compare_length; ++i) {
+ int32_t left = static_cast<uint8_t>(bytes_->data()[i]);
+ int32_t right = static_cast<uint8_t>(other.bytes_->data()[i]);
+ if (left != right) {
+ return left < right ? -1 : 1;
+ }
+ }
+ if (bytes_->size() == other.bytes_->size()) {
+ return 0;
+ }
+ return bytes_->size() < other.bytes_->size() ? -1 : 1;
+}
+
+Result<std::shared_ptr<Bytes>> BitmapGlobalIndexFormat::SeekableReader::Read(
+ const BlockInfo& block) {
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(block));
+ return Read(block.Offset(), block.Length());
+}
+
+Result<int32_t> BitmapGlobalIndexFormat::DictionaryEntry::EstimatedSize()
const {
+ if (key_.GetBytes() == nullptr) {
+ return Status::Invalid("Bitmap dictionary key is null.");
+ }
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(key_.GetBytes()->size(), "bitmap
dictionary key length"));
+ PAIMON_ASSIGN_OR_RAISE(int32_t key_length_size,
+
EstimatedVarLenIntSize(static_cast<int32_t>(key_.GetBytes()->size())));
+ PAIMON_ASSIGN_OR_RAISE(int32_t offset_size,
EstimatedVarLenLongSize(bitmap_block_.Offset()));
+ PAIMON_ASSIGN_OR_RAISE(int32_t length_size,
EstimatedVarLenIntSize(bitmap_block_.Length()));
+ int64_t size =
+ static_cast<int64_t>(key_length_size) + key_.GetBytes()->size() +
offset_size + length_size;
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(size, "bitmap
dictionary entry size"));
+ return static_cast<int32_t>(size);
+}
+
+Result<std::unique_ptr<BitmapGlobalIndexFormat::StreamingWriter>>
+BitmapGlobalIndexFormat::StreamingWriter::Create(
+ const std::shared_ptr<OutputStream>& output_stream, int32_t
dictionary_block_size,
+ const std::shared_ptr<BlockCompressionFactory>& compression_factory,
+ const std::shared_ptr<MemoryPool>& pool) {
+ if (output_stream == nullptr) {
+ return Status::Invalid("Cannot create bitmap StreamingWriter without
an output stream.");
+ }
+ if (dictionary_block_size <= 0) {
+ return Status::Invalid("Bitmap dictionary block size must be greater
than 0.");
+ }
+ if (pool == nullptr) {
+ return Status::Invalid("Cannot create bitmap StreamingWriter without a
memory pool.");
+ }
+ return std::unique_ptr<StreamingWriter>(
+ new StreamingWriter(output_stream, dictionary_block_size,
compression_factory, pool));
+}
+
+Status BitmapGlobalIndexFormat::StreamingWriter::Write(SerializedKey key,
+ const RoaringBitmap64&
bitmap) {
+ if (finished_) {
+ return Status::Invalid("Cannot write to a finished bitmap
StreamingWriter.");
+ }
+ if (key.GetBytes() == nullptr) {
+ return Status::Invalid("Cannot write a null serialized bitmap
dictionary key.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(BlockInfo bitmap_block,
+ WriteBitmapBlock(bitmap, output_stream_.get(),
pool_.get()));
+ DictionaryEntry entry(std::move(key), std::move(bitmap_block));
+ PAIMON_ASSIGN_OR_RAISE(int32_t estimated_size,
EstimatedDictionaryBlockSizeAfter(entry));
+ if (!current_dictionary_entries_.empty() && estimated_size >
dictionary_block_size_) {
+ PAIMON_RETURN_NOT_OK(FlushDictionaryBlock());
+ }
+ PAIMON_ASSIGN_OR_RAISE(int32_t entry_size, entry.EstimatedSize());
+ if (value_count_ == std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid("Bitmap global index value count exceeds
INT32_MAX.");
+ }
+ current_dictionary_entries_size_ += entry_size;
+ current_dictionary_entries_.push_back(std::move(entry));
+ ++value_count_;
+ return Status::OK();
+}
+
+Status BitmapGlobalIndexFormat::StreamingWriter::Finish(const RoaringBitmap64&
null_rows,
+ const RoaringBitmap64&
non_null_rows) {
+ if (finished_) {
+ return Status::Invalid("Bitmap StreamingWriter has already been
finished.");
+ }
+ PAIMON_RETURN_NOT_OK(FlushDictionaryBlock());
+ PAIMON_ASSIGN_OR_RAISE(BlockInfo null_rows_block,
+ WriteBitmapBlock(null_rows, output_stream_.get(),
pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(BlockInfo non_null_rows_block,
+ WriteBitmapBlock(non_null_rows,
output_stream_.get(), pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(BlockInfo index_block,
+ WriteIndexBlock(dictionary_block_metas_,
compression_factory_.get(),
+ output_stream_.get(), pool_.get()));
+ PAIMON_RETURN_NOT_OK(WriteFooter(null_rows_block, non_null_rows_block,
index_block,
+ value_count_, output_stream_.get()));
+ finished_ = true;
+ return Status::OK();
+}
+
+Status BitmapGlobalIndexFormat::StreamingWriter::FlushDictionaryBlock() {
+ if (current_dictionary_entries_.empty()) {
+ return Status::OK();
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ DictionaryBlockMeta block,
+ WriteDictionaryBlock(current_dictionary_entries_,
compression_factory_.get(),
+ output_stream_.get(), pool_.get()));
+ dictionary_block_metas_.push_back(std::move(block));
+ current_dictionary_entries_.clear();
+ current_dictionary_entries_size_ = 0;
+ return Status::OK();
+}
+
+Result<int32_t>
BitmapGlobalIndexFormat::StreamingWriter::EstimatedDictionaryBlockSizeAfter(
+ const DictionaryEntry& entry) const {
+ size_t entry_count = current_dictionary_entries_.size() + 1;
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(entry_count, "bitmap dictionary block
entry count"));
+ PAIMON_ASSIGN_OR_RAISE(int32_t count_size,
+
EstimatedVarLenIntSize(static_cast<int32_t>(entry_count)));
+ PAIMON_ASSIGN_OR_RAISE(int32_t entry_size, entry.EstimatedSize());
+ int64_t total =
+ static_cast<int64_t>(count_size) + current_dictionary_entries_size_ +
entry_size;
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(total, "bitmap
dictionary block size"));
+ return static_cast<int32_t>(total);
+}
+
+Status BitmapGlobalIndexFormat::WriteFooter(const BlockInfo& null_rows_block,
+ const BlockInfo&
non_null_rows_block,
+ const BlockInfo& index_block,
int32_t value_count,
+ OutputStream* output_stream) {
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(null_rows_block));
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(non_null_rows_block));
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(index_block));
+ if (value_count < 0) {
+ return Status::Invalid("Invalid negative bitmap value count.");
+ }
+ DataOutputStream output(output_stream);
+ output.SetOrder(ByteOrder::PAIMON_BIG_ENDIAN);
+ PAIMON_RETURN_NOT_OK(output.WriteValue<int64_t>(null_rows_block.Offset()));
+ PAIMON_RETURN_NOT_OK(output.WriteValue<int32_t>(null_rows_block.Length()));
+
PAIMON_RETURN_NOT_OK(output.WriteValue<int64_t>(non_null_rows_block.Offset()));
+
PAIMON_RETURN_NOT_OK(output.WriteValue<int32_t>(non_null_rows_block.Length()));
+ PAIMON_RETURN_NOT_OK(output.WriteValue<int64_t>(index_block.Offset()));
+ PAIMON_RETURN_NOT_OK(output.WriteValue<int32_t>(index_block.Length()));
+ PAIMON_RETURN_NOT_OK(output.WriteValue<int32_t>(value_count));
+ PAIMON_RETURN_NOT_OK(output.WriteValue<int32_t>(kVersion));
+ PAIMON_RETURN_NOT_OK(output.WriteValue<int32_t>(kMagic));
+ return output_stream->Flush();
+}
+
+Result<BitmapGlobalIndexFormat::BlockInfo>
BitmapGlobalIndexFormat::WriteBitmapBlock(
+ const RoaringBitmap64& bitmap, OutputStream* output_stream, MemoryPool*
pool) {
+ std::shared_ptr<Bytes> bytes = bitmap.Serialize(pool);
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(bytes->size(), "serialized bitmap block
size"));
+ PAIMON_ASSIGN_OR_RAISE(int64_t offset, output_stream->GetPos());
+ PAIMON_RETURN_NOT_OK(WriteAll(bytes->data(), bytes->size(),
output_stream));
+ return BlockInfo(offset, static_cast<int32_t>(bytes->size()));
+}
+
+Result<BitmapGlobalIndexFormat::DictionaryBlockMeta>
BitmapGlobalIndexFormat::WriteDictionaryBlock(
+ const std::vector<DictionaryEntry>& entries, BlockCompressionFactory*
compression_factory,
+ OutputStream* output_stream, MemoryPool* pool) {
+ if (entries.empty()) {
+ return Status::Invalid("Cannot write an empty bitmap dictionary
block.");
+ }
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(entries.size(), "bitmap dictionary block
entry count"));
+ PAIMON_ASSIGN_OR_RAISE(int32_t entry_count_size,
+
EstimatedVarLenIntSize(static_cast<int32_t>(entries.size())));
+ int64_t estimated_size = entry_count_size;
+ for (const DictionaryEntry& entry : entries) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t entry_size, entry.EstimatedSize());
+ estimated_size += entry_size;
+ }
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(estimated_size, "bitmap dictionary block
size"));
+ MemorySliceOutput output(static_cast<int32_t>(estimated_size), pool);
+
PAIMON_RETURN_NOT_OK(output.WriteVarLenInt(static_cast<int32_t>(entries.size())));
+ for (const DictionaryEntry& entry : entries) {
+ const std::shared_ptr<Bytes>& key_bytes = entry.Key().GetBytes();
+
PAIMON_RETURN_NOT_OK(output.WriteVarLenInt(static_cast<int32_t>(key_bytes->size())));
+ output.WriteBytes(key_bytes);
+
PAIMON_RETURN_NOT_OK(output.WriteVarLenLong(entry.BitmapBlock().Offset()));
+
PAIMON_RETURN_NOT_OK(output.WriteVarLenInt(entry.BitmapBlock().Length()));
+ }
+ std::shared_ptr<Bytes> bytes = output.ToSlice().CopyBytes(pool);
+ PAIMON_ASSIGN_OR_RAISE(BlockInfo block,
+ WriteCompressibleBlock(bytes, compression_factory,
output_stream, pool));
+ return DictionaryBlockMeta(entries.front().Key(), block.Offset(),
block.Length());
+}
+
+Result<BitmapGlobalIndexFormat::BlockInfo>
BitmapGlobalIndexFormat::WriteIndexBlock(
+ const std::vector<DictionaryBlockMeta>& blocks, BlockCompressionFactory*
compression_factory,
+ OutputStream* output_stream, MemoryPool* pool) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t estimated_size,
EstimatedIndexBlockSize(blocks));
+ MemorySliceOutput output(estimated_size, pool);
+
PAIMON_RETURN_NOT_OK(output.WriteVarLenInt(static_cast<int32_t>(blocks.size())));
+ for (const DictionaryBlockMeta& block : blocks) {
+ const std::shared_ptr<Bytes>& key_bytes = block.FirstKey().GetBytes();
+
PAIMON_RETURN_NOT_OK(output.WriteVarLenInt(static_cast<int32_t>(key_bytes->size())));
+ output.WriteBytes(key_bytes);
+ PAIMON_RETURN_NOT_OK(output.WriteVarLenLong(block.Offset()));
+ PAIMON_RETURN_NOT_OK(output.WriteVarLenInt(block.Length()));
+ }
+ std::shared_ptr<Bytes> bytes = output.ToSlice().CopyBytes(pool);
+ return WriteCompressibleBlock(bytes, compression_factory, output_stream,
pool);
+}
+
+Result<BitmapGlobalIndexFormat::BlockInfo>
BitmapGlobalIndexFormat::WriteCompressibleBlock(
+ const std::shared_ptr<Bytes>& uncompressed, BlockCompressionFactory*
compression_factory,
+ OutputStream* output_stream, MemoryPool* pool) {
+ PAIMON_ASSIGN_OR_RAISE(BlockEncoding encoding,
+ EncodeBlock(uncompressed, compression_factory,
pool));
+ PAIMON_ASSIGN_OR_RAISE(int64_t offset, output_stream->GetPos());
+ PAIMON_RETURN_NOT_OK(WriteAll(encoding.bytes->data(), encoding.length,
output_stream));
+
+ uint32_t crc = CRC32C::calculate(encoding.bytes->data(), encoding.length);
+ char compression_value =
+ static_cast<char>(static_cast<int32_t>(encoding.compression_type) &
0xFF);
+ crc = CRC32C::calculate(&compression_value, sizeof(compression_value),
crc);
+ BlockTrailer trailer(static_cast<int8_t>(encoding.compression_type),
static_cast<int32_t>(crc));
+ MemorySlice trailer_slice = trailer.WriteBlockTrailer(pool);
+ PAIMON_RETURN_NOT_OK(WriteAll(trailer_slice.Data(),
trailer_slice.Length(), output_stream));
+ return BlockInfo(offset, encoding.length);
+}
+
+Result<BitmapGlobalIndexFormat::BlockEncoding>
BitmapGlobalIndexFormat::EncodeBlock(
+ const std::shared_ptr<Bytes>& uncompressed, BlockCompressionFactory*
compression_factory,
+ MemoryPool* pool) {
+ if (uncompressed == nullptr) {
+ return Status::Invalid("Uncompressed bitmap index block is null.");
+ }
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(uncompressed->size(),
+ "uncompressed bitmap
index block size"));
+ auto uncompressed_length = static_cast<int32_t>(uncompressed->size());
+ BlockEncoding result{uncompressed, uncompressed_length,
BlockCompressionType::NONE};
+ if (compression_factory == nullptr ||
+ compression_factory->GetCompressionType() ==
BlockCompressionType::NONE) {
+ return result;
+ }
+
+ std::shared_ptr<BlockCompressor> compressor =
compression_factory->GetCompressor();
+ if (compressor == nullptr) {
+ return Status::Invalid("Bitmap block compression factory returned a
null compressor.");
+ }
+ int32_t maximum_compressed_size =
compressor->GetMaxCompressedSize(uncompressed_length);
+ if (maximum_compressed_size < 0 ||
+ maximum_compressed_size >
+ std::numeric_limits<int32_t>::max() -
VarLengthIntUtils::kMaxVarIntSize) {
+ return Status::Invalid("Invalid maximum compressed bitmap block
size.");
+ }
+ std::shared_ptr<Bytes> compressed =
+ Bytes::AllocateBytes(maximum_compressed_size +
VarLengthIntUtils::kMaxVarIntSize, pool);
+ PAIMON_ASSIGN_OR_RAISE(int32_t prefix_length,
+ VarLengthIntUtils::EncodeInt(uncompressed_length,
compressed->data()));
+ PAIMON_ASSIGN_OR_RAISE(
+ int32_t compressed_length,
+ compressor->Compress(
+ uncompressed->data(), uncompressed_length, compressed->data() +
prefix_length,
+ maximum_compressed_size + VarLengthIntUtils::kMaxVarIntSize -
prefix_length));
+ if (compressed_length < 0 ||
+ compressed_length > std::numeric_limits<int32_t>::max() -
prefix_length) {
+ return Status::Invalid("Invalid compressed bitmap block size.");
+ }
+ int32_t encoded_length = prefix_length + compressed_length;
+ if (encoded_length < uncompressed_length - (uncompressed_length / 8)) {
+ result.bytes = std::move(compressed);
+ result.length = encoded_length;
+ result.compression_type = compression_factory->GetCompressionType();
+ }
+ return result;
+}
+
+Result<BitmapGlobalIndexFormat::Footer> BitmapGlobalIndexFormat::ReadFooter(
+ int64_t file_size, SeekableReader* reader) {
+ if (reader == nullptr) {
+ return Status::Invalid("Cannot read bitmap footer without a reader.");
+ }
+ if (file_size < kFooterLength) {
+ return Status::Invalid("Invalid bitmap global index file size.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
+ reader->Read(file_size - kFooterLength,
kFooterLength));
+ if (bytes == nullptr || bytes->size() != kFooterLength) {
+ return Status::Invalid("Truncated bitmap global index footer.");
+ }
+ MemorySliceInput input(MemorySlice::Wrap(bytes));
+ input.SetOrder(ByteOrder::PAIMON_BIG_ENDIAN);
+ int64_t null_rows_offset = input.ReadLong();
+ int32_t null_rows_length = input.ReadInt();
+ BlockInfo null_rows_block(null_rows_offset, null_rows_length);
+ int64_t non_null_rows_offset = input.ReadLong();
+ int32_t non_null_rows_length = input.ReadInt();
+ BlockInfo non_null_rows_block(non_null_rows_offset, non_null_rows_length);
+ int64_t index_offset = input.ReadLong();
+ int32_t index_length = input.ReadInt();
+ BlockInfo index_block(index_offset, index_length);
+ int32_t value_count = input.ReadInt();
+ int32_t version = input.ReadInt();
+ int32_t magic = input.ReadInt();
+ if (magic != kMagic) {
+ return Status::Invalid("File is not a bitmap global index file (bad
footer magic).");
+ }
+ if (version != kVersion) {
+ return Status::Invalid(
+ fmt::format("Unsupported bitmap global index file version: {}.",
version));
+ }
+ if (value_count < 0) {
+ return Status::Invalid("Invalid bitmap value count.");
+ }
+ for (const BlockInfo* block : {&null_rows_block, &non_null_rows_block,
&index_block}) {
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(*block));
+ if (block->Offset() > file_size - kFooterLength ||
+ block->Length() > file_size - kFooterLength - block->Offset()) {
+ return Status::Invalid("Bitmap footer references a block outside
the file payload.");
+ }
+ }
+ return Footer(std::move(null_rows_block), std::move(non_null_rows_block),
+ std::move(index_block));
+}
+
+Result<std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>>
+BitmapGlobalIndexFormat::ReadIndexBlock(const BlockInfo& index_block,
SeekableReader* reader,
+ MemoryPool* pool) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
+ ReadCompressibleBlock(index_block, reader, pool));
+ MemorySliceInput input(MemorySlice::Wrap(bytes));
+ PAIMON_ASSIGN_OR_RAISE(int32_t block_count,
VarLengthIntUtils::ReadVarLenInt(&input));
+ if (block_count > input.Available()) {
+ return Status::Invalid("Bitmap dictionary block count exceeds the
encoded block size.");
+ }
+ std::vector<DictionaryBlockMeta> blocks;
+ blocks.reserve(block_count);
+ for (int32_t i = 0; i < block_count; ++i) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> key_bytes,
ReadKey(&input, pool));
+ PAIMON_ASSIGN_OR_RAISE(int64_t offset,
VarLengthIntUtils::ReadVarLenLong(&input));
+ PAIMON_ASSIGN_OR_RAISE(int32_t length,
VarLengthIntUtils::ReadVarLenInt(&input));
+ BlockInfo block(offset, length);
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(block));
+ blocks.emplace_back(SerializedKey(std::move(key_bytes)), offset,
length);
+ }
+ if (input.Available() != 0) {
+ return Status::Invalid("Bitmap dictionary block index has trailing
bytes.");
+ }
+ return blocks;
+}
+
+Result<BitmapGlobalIndexFormat::DictionaryBlock>
BitmapGlobalIndexFormat::ReadDictionaryBlock(
+ const DictionaryBlockMeta& block, SeekableReader* reader, MemoryPool*
pool) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
+ ReadCompressibleBlock(block, reader, pool));
+ MemorySliceInput input(MemorySlice::Wrap(bytes));
+ PAIMON_ASSIGN_OR_RAISE(int32_t entry_count,
VarLengthIntUtils::ReadVarLenInt(&input));
+ if (entry_count > input.Available()) {
+ return Status::Invalid("Bitmap dictionary entry count exceeds the
encoded block size.");
+ }
+ std::vector<DictionaryEntry> entries;
+ entries.reserve(entry_count);
+ for (int32_t i = 0; i < entry_count; ++i) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> key_bytes,
ReadKey(&input, pool));
+ PAIMON_ASSIGN_OR_RAISE(int64_t bitmap_offset,
VarLengthIntUtils::ReadVarLenLong(&input));
+ PAIMON_ASSIGN_OR_RAISE(int32_t bitmap_length,
VarLengthIntUtils::ReadVarLenInt(&input));
+ BlockInfo bitmap_block(bitmap_offset, bitmap_length);
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(bitmap_block));
+ entries.emplace_back(SerializedKey(std::move(key_bytes)),
std::move(bitmap_block));
+ }
+ if (input.Available() != 0) {
+ return Status::Invalid("Bitmap dictionary block has trailing bytes.");
+ }
+ return DictionaryBlock(std::move(entries));
+}
+
+Result<RoaringBitmap64> BitmapGlobalIndexFormat::ReadBitmap(const BlockInfo&
block,
+ SeekableReader*
reader) {
+ if (reader == nullptr) {
+ return Status::Invalid("Cannot read bitmap block without a reader.");
+ }
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(block));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes, reader->Read(block));
+ RoaringBitmap64 bitmap;
+ PAIMON_RETURN_NOT_OK(bitmap.Deserialize(bytes->data(), bytes->size()));
+ return bitmap;
+}
+
+Result<std::shared_ptr<Bytes>> BitmapGlobalIndexFormat::ReadCompressibleBlock(
+ const BlockInfo& block, SeekableReader* reader, MemoryPool* pool) {
+ if (reader == nullptr || pool == nullptr) {
+ return Status::Invalid("Cannot read compressed bitmap block without
reader and pool.");
+ }
+ PAIMON_RETURN_NOT_OK(ValidateBlockInfo(block));
+ if (block.Length() > std::numeric_limits<int32_t>::max() -
BlockTrailer::ENCODED_LENGTH) {
+ return Status::Invalid("Bitmap block is too large.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<Bytes> block_and_trailer,
+ reader->Read(block.Offset(), block.Length() +
BlockTrailer::ENCODED_LENGTH));
+ if (block_and_trailer == nullptr ||
+ block_and_trailer->size() !=
+ static_cast<size_t>(block.Length() +
BlockTrailer::ENCODED_LENGTH)) {
+ return Status::Invalid("Truncated compressed bitmap index block.");
+ }
+
+ MemorySlice all = MemorySlice::Wrap(block_and_trailer);
+ MemorySlice block_slice = all.Slice(0, block.Length());
+ MemorySlice trailer_slice = all.Slice(block.Length(),
BlockTrailer::ENCODED_LENGTH);
+ MemorySliceInput trailer_input(trailer_slice);
+ std::unique_ptr<BlockTrailer> trailer =
BlockTrailer::ReadBlockTrailer(&trailer_input);
+ PAIMON_ASSIGN_OR_RAISE(BlockCompressionType compression_type,
+ SstFileUtils::From(trailer->CompressionType()));
+ uint32_t crc = CRC32C::calculate(block_slice.Data(), block_slice.Length());
+ auto compression_value =
static_cast<char>(static_cast<int32_t>(compression_type) & 0xFF);
+ crc = CRC32C::calculate(&compression_value, sizeof(compression_value),
crc);
+ if (trailer->Crc32c() != static_cast<int32_t>(crc)) {
+ return Status::Invalid(
+ fmt::format("Expected CRC32C({:#x}) but found CRC32C({:#x}) for
bitmap index block.",
+ static_cast<uint32_t>(trailer->Crc32c()), crc));
+ }
+ if (compression_type == BlockCompressionType::NONE) {
+ return block_slice.CopyBytes(pool);
+ }
+
+ MemorySliceInput compressed_input(block_slice);
+ PAIMON_ASSIGN_OR_RAISE(int32_t uncompressed_length,
+
VarLengthIntUtils::ReadVarLenInt(&compressed_input));
+ std::shared_ptr<Bytes> uncompressed =
Bytes::AllocateBytes(uncompressed_length, pool);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<BlockCompressionFactory>
compression_factory,
+ BlockCompressionFactory::Create(compression_type));
+ std::shared_ptr<BlockDecompressor> decompressor =
compression_factory->GetDecompressor();
+ if (decompressor == nullptr) {
+ return Status::Invalid("Bitmap block compression factory returned a
null decompressor.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ int32_t actual_length,
+ decompressor->Decompress(block_slice.Data() +
compressed_input.Position(),
+ compressed_input.Available(),
uncompressed->data(),
+ uncompressed_length));
+ if (actual_length != uncompressed_length) {
+ return Status::Invalid(
+ fmt::format("Invalid bitmap block: expected uncompressed size {},
actual size {}.",
+ uncompressed_length, actual_length));
+ }
+ return uncompressed;
+}
+
+Result<int32_t> BitmapGlobalIndexFormat::EstimatedVarLenIntSize(int32_t value)
{
+ if (value < 0) {
+ return Status::Invalid(fmt::format("Invalid negative var length int:
{}.", value));
+ }
+ int32_t size = 1;
+ while ((value & ~0x7F) != 0) {
+ value >>= 7;
+ ++size;
+ }
+ return size;
+}
+
+Result<int32_t> BitmapGlobalIndexFormat::EstimatedVarLenLongSize(int64_t
value) {
+ if (value < 0) {
+ return Status::Invalid(fmt::format("Invalid negative var length long:
{}.", value));
+ }
+ int32_t size = 1;
+ while ((value & ~0x7FLL) != 0) {
+ value >>= 7;
+ ++size;
+ }
+ return size;
+}
+
+Result<int32_t> BitmapGlobalIndexFormat::EstimatedIndexBlockSize(
+ const std::vector<DictionaryBlockMeta>& blocks) {
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(blocks.size(), "bitmap dictionary block
count"));
+ PAIMON_ASSIGN_OR_RAISE(int32_t count_size,
+
EstimatedVarLenIntSize(static_cast<int32_t>(blocks.size())));
+ int64_t size = count_size;
+ for (const DictionaryBlockMeta& block : blocks) {
+ const std::shared_ptr<Bytes>& key = block.FirstKey().GetBytes();
+ if (key == nullptr) {
+ return Status::Invalid("Bitmap dictionary index key is null.");
+ }
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(key->size(), "bitmap dictionary
index key length"));
+ PAIMON_ASSIGN_OR_RAISE(int32_t key_length_size,
+
EstimatedVarLenIntSize(static_cast<int32_t>(key->size())));
+ PAIMON_ASSIGN_OR_RAISE(int32_t offset_size,
EstimatedVarLenLongSize(block.Offset()));
+ PAIMON_ASSIGN_OR_RAISE(int32_t length_size,
EstimatedVarLenIntSize(block.Length()));
+ size += key_length_size + key->size() + offset_size + length_size;
+ PAIMON_RETURN_NOT_OK(
+ ValidateValueInRange<int32_t>(size, "bitmap dictionary index block
size"));
+ }
+ return static_cast<int32_t>(size);
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/global_index/bitmap/bitmap_global_index_format.h
b/src/paimon/common/global_index/bitmap/bitmap_global_index_format.h
new file mode 100644
index 00000000..967ce267
--- /dev/null
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_format.h
@@ -0,0 +1,303 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "paimon/common/compression/block_compression_factory.h"
+#include "paimon/common/memory/memory_slice.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+
+class KeySerializer;
+class Literal;
+
+/// Shared file format helpers for bitmap global index.
+///
+/// Bitmap blocks for non-null keys are written as keys arrive. Dictionary
blocks are flushed
+/// periodically, so these two kinds of blocks may be interleaved. The blocks
at the end of the
+/// file have a fixed order:
+///
+/// +--------------------------------------------------+
+/// | Bitmap Block for key 1 (RoaringBitmap64) |
+/// +--------------------------------------------------+
+/// | ... Bitmap / Dictionary Blocks may interleave ...|
+/// +--------------------------------------------------+
+/// | Dictionary Block N |
+/// +--------------------------------------------------+
+/// | Null Rows Bitmap Block |
+/// +--------------------------------------------------+
+/// | Non-null Rows Bitmap Block |
+/// +--------------------------------------------------+
+/// | Dictionary Block Index |
+/// +--------------------------------------------------+
+/// | Footer (48 bytes) |
+/// +--------------------------------------------------+
+///
+/// Each dictionary entry maps one serialized key to its bitmap block:
+///
+/// Dictionary Block:
+/// [entry count]
+/// [key length | key bytes | bitmap offset | bitmap length] ...
+///
+/// The dictionary block index maps the first key of each dictionary block to
that block:
+///
+/// Dictionary Block Index:
+/// [block count]
+/// [first key length | first key bytes | dictionary offset | dictionary
length] ...
+///
+/// The footer stores the offset and length of the null rows bitmap, non-null
rows bitmap and
+/// dictionary block index, followed by value count, version and magic.
Dictionary blocks and the
+/// dictionary block index use the compressible-block encoding and are
followed by a block trailer;
+/// bitmap blocks contain the serialized RoaringBitmap64 bytes directly.
+class BitmapGlobalIndexFormat {
+ public:
+ BitmapGlobalIndexFormat() = delete;
+ ~BitmapGlobalIndexFormat() = delete;
+
+ class BlockInfo;
+
+ /// Serialized bitmap dictionary key.
+ class SerializedKey {
+ public:
+ explicit SerializedKey(std::shared_ptr<Bytes> bytes) :
bytes_(std::move(bytes)) {}
+
+ static Result<SerializedKey> FromLiteral(const
std::shared_ptr<KeySerializer>& serializer,
+ const Literal& literal);
+
+ const std::shared_ptr<Bytes>& GetBytes() const {
+ return bytes_;
+ }
+
+ int32_t CompareTo(const SerializedKey& other) const;
+
+ bool operator==(const SerializedKey& other) const {
+ return CompareTo(other) == 0;
+ }
+
+ private:
+ std::shared_ptr<Bytes> bytes_;
+ };
+
+ /// Minimal random-access reader used by bitmap index format decoders.
+ class SeekableReader {
+ public:
+ virtual ~SeekableReader() = default;
+
+ virtual Result<std::shared_ptr<Bytes>> Read(int64_t offset, int32_t
length) = 0;
+
+ Result<std::shared_ptr<Bytes>> Read(const BlockInfo& block);
+ };
+
+ /// Encoded block location within a bitmap index file.
+ class BlockInfo {
+ public:
+ BlockInfo(int64_t offset, int32_t length) : offset_(offset),
length_(length) {}
+
+ int64_t Offset() const {
+ return offset_;
+ }
+
+ int32_t Length() const {
+ return length_;
+ }
+
+ private:
+ int64_t offset_;
+ int32_t length_;
+ };
+
+ /// Dictionary block metadata read from the block index.
+ class DictionaryBlockMeta : public BlockInfo {
+ public:
+ DictionaryBlockMeta(SerializedKey first_key, int64_t offset, int32_t
length)
+ : BlockInfo(offset, length), first_key_(std::move(first_key)) {}
+
+ const SerializedKey& FirstKey() const {
+ return first_key_;
+ }
+
+ private:
+ SerializedKey first_key_;
+ };
+
+ /// One encoded dictionary key and its bitmap block.
+ class DictionaryEntry {
+ public:
+ DictionaryEntry(SerializedKey key, BlockInfo bitmap_block)
+ : key_(std::move(key)), bitmap_block_(std::move(bitmap_block)) {}
+
+ const SerializedKey& Key() const {
+ return key_;
+ }
+
+ const BlockInfo& BitmapBlock() const {
+ return bitmap_block_;
+ }
+
+ Result<int32_t> EstimatedSize() const;
+
+ private:
+ SerializedKey key_;
+ BlockInfo bitmap_block_;
+ };
+
+ /// Decoded dictionary block.
+ class DictionaryBlock {
+ public:
+ explicit DictionaryBlock(std::vector<DictionaryEntry> entries)
+ : entries_(std::move(entries)) {}
+
+ const std::vector<DictionaryEntry>& Entries() const {
+ return entries_;
+ }
+
+ private:
+ std::vector<DictionaryEntry> entries_;
+ };
+
+ /// Bitmap index footer block references.
+ class Footer {
+ public:
+ Footer(BlockInfo null_rows_block, BlockInfo non_null_rows_block,
BlockInfo index_block)
+ : null_rows_block_(std::move(null_rows_block)),
+ non_null_rows_block_(std::move(non_null_rows_block)),
+ index_block_(std::move(index_block)) {}
+
+ const BlockInfo& NullRowsBlock() const {
+ return null_rows_block_;
+ }
+
+ const BlockInfo& NonNullRowsBlock() const {
+ return non_null_rows_block_;
+ }
+
+ const BlockInfo& IndexBlock() const {
+ return index_block_;
+ }
+
+ private:
+ BlockInfo null_rows_block_;
+ BlockInfo non_null_rows_block_;
+ BlockInfo index_block_;
+ };
+
+ /// Streaming writer for encoded bitmap dictionary entries.
+ class StreamingWriter {
+ public:
+ static Result<std::unique_ptr<StreamingWriter>> Create(
+ const std::shared_ptr<OutputStream>& output_stream, int32_t
dictionary_block_size,
+ const std::shared_ptr<BlockCompressionFactory>&
compression_factory,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ Status Write(SerializedKey key, const RoaringBitmap64& bitmap);
+
+ Status Finish(const RoaringBitmap64& null_rows, const RoaringBitmap64&
non_null_rows);
+
+ private:
+ StreamingWriter(const std::shared_ptr<OutputStream>& output_stream,
+ int32_t dictionary_block_size,
+ const std::shared_ptr<BlockCompressionFactory>&
compression_factory,
+ const std::shared_ptr<MemoryPool>& pool)
+ : output_stream_(output_stream),
+ dictionary_block_size_(dictionary_block_size),
+ compression_factory_(compression_factory),
+ pool_(pool) {}
+
+ Status FlushDictionaryBlock();
+ Result<int32_t> EstimatedDictionaryBlockSizeAfter(const
DictionaryEntry& entry) const;
+
+ std::shared_ptr<OutputStream> output_stream_;
+ int32_t dictionary_block_size_;
+ std::shared_ptr<BlockCompressionFactory> compression_factory_;
+ std::shared_ptr<MemoryPool> pool_;
+ std::vector<DictionaryBlockMeta> dictionary_block_metas_;
+ std::vector<DictionaryEntry> current_dictionary_entries_;
+ int32_t current_dictionary_entries_size_ = 0;
+ int32_t value_count_ = 0;
+ bool finished_ = false;
+ };
+
+ static Result<Footer> ReadFooter(int64_t file_size, SeekableReader*
reader);
+
+ static Result<std::vector<DictionaryBlockMeta>> ReadIndexBlock(const
BlockInfo& index_block,
+
SeekableReader* reader,
+ MemoryPool*
pool);
+
+ static Result<DictionaryBlock> ReadDictionaryBlock(const
DictionaryBlockMeta& block,
+ SeekableReader* reader,
MemoryPool* pool);
+
+ static Result<RoaringBitmap64> ReadBitmap(const BlockInfo& block,
SeekableReader* reader);
+
+ static constexpr int32_t kMagic = 0x42474958;
+ static constexpr int32_t kVersion = 1;
+ static constexpr int32_t kFooterLength = 48;
+
+ private:
+ friend class StreamingWriter;
+
+ struct BlockEncoding {
+ std::shared_ptr<Bytes> bytes;
+ int32_t length;
+ BlockCompressionType compression_type;
+ };
+
+ static Status WriteFooter(const BlockInfo& null_rows_block,
+ const BlockInfo& non_null_rows_block, const
BlockInfo& index_block,
+ int32_t value_count, OutputStream*
output_stream);
+
+ static Result<BlockInfo> WriteBitmapBlock(const RoaringBitmap64& bitmap,
+ OutputStream* output_stream,
MemoryPool* pool);
+
+ static Result<DictionaryBlockMeta> WriteDictionaryBlock(
+ const std::vector<DictionaryEntry>& entries, BlockCompressionFactory*
compression_factory,
+ OutputStream* output_stream, MemoryPool* pool);
+
+ static Result<BlockInfo> WriteIndexBlock(const
std::vector<DictionaryBlockMeta>& blocks,
+ BlockCompressionFactory*
compression_factory,
+ OutputStream* output_stream,
MemoryPool* pool);
+
+ static Result<BlockInfo> WriteCompressibleBlock(const
std::shared_ptr<Bytes>& uncompressed,
+ BlockCompressionFactory*
compression_factory,
+ OutputStream*
output_stream, MemoryPool* pool);
+
+ static Result<BlockEncoding> EncodeBlock(const std::shared_ptr<Bytes>&
uncompressed,
+ BlockCompressionFactory*
compression_factory,
+ MemoryPool* pool);
+
+ static Result<std::shared_ptr<Bytes>> ReadCompressibleBlock(const
BlockInfo& block,
+
SeekableReader* reader,
+ MemoryPool*
pool);
+
+ static Result<int32_t> EstimatedVarLenIntSize(int32_t value);
+ static Result<int32_t> EstimatedVarLenLongSize(int64_t value);
+ static Result<int32_t> EstimatedIndexBlockSize(const
std::vector<DictionaryBlockMeta>& blocks);
+};
+
+} // namespace paimon
diff --git
a/src/paimon/common/global_index/bitmap/bitmap_global_index_format_test.cpp
b/src/paimon/common/global_index/bitmap/bitmap_global_index_format_test.cpp
new file mode 100644
index 00000000..2ad25cf0
--- /dev/null
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_format_test.cpp
@@ -0,0 +1,143 @@
+/*
+ * 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/global_index/bitmap/bitmap_global_index_format.h"
+
+#include <cstring>
+#include <memory>
+#include <string>
+
+#include "gtest/gtest.h"
+#include "paimon/common/io/byte_array_output_stream.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+
+class BytesSeekableReader : public BitmapGlobalIndexFormat::SeekableReader {
+ public:
+ BytesSeekableReader(std::shared_ptr<Bytes> bytes,
std::shared_ptr<MemoryPool> pool)
+ : bytes_(std::move(bytes)), pool_(std::move(pool)) {}
+
+ Result<std::shared_ptr<Bytes>> Read(int64_t offset, int32_t length)
override {
+ if (offset < 0 || length < 0 || offset >
static_cast<int64_t>(bytes_->size()) ||
+ length > static_cast<int64_t>(bytes_->size()) - offset) {
+ return Status::Invalid("Read exceeds in-memory bitmap index
data.");
+ }
+ std::shared_ptr<Bytes> result = Bytes::AllocateBytes(length,
pool_.get());
+ std::memcpy(result->data(), bytes_->data() + offset, length);
+ return result;
+ }
+
+ private:
+ std::shared_ptr<Bytes> bytes_;
+ std::shared_ptr<MemoryPool> pool_;
+};
+
+BitmapGlobalIndexFormat::SerializedKey SerializedString(const std::string&
value,
+ MemoryPool* pool) {
+ return BitmapGlobalIndexFormat::SerializedKey(Bytes::AllocateBytes(value,
pool));
+}
+
+std::shared_ptr<ByteArrayOutputStream> CreateOutput(const
std::shared_ptr<MemoryPool>& pool) {
+ std::unique_ptr<MemorySegmentOutputStream> segmented =
+ std::make_unique<MemorySegmentOutputStream>(/*segment_size=*/16, pool);
+ return std::make_shared<ByteArrayOutputStream>(std::move(segmented));
+}
+
+} // namespace
+
+TEST(BitmapGlobalIndexFormatTest, RoundTripMultipleDictionaryBlocks) {
+ std::shared_ptr<MemoryPool> pool = GetDefaultPool();
+ std::shared_ptr<ByteArrayOutputStream> output = CreateOutput(pool);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlockCompressionFactory>
compression_factory,
+
BlockCompressionFactory::Create(BlockCompressionType::NONE));
+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<BitmapGlobalIndexFormat::StreamingWriter>
writer,
+ BitmapGlobalIndexFormat::StreamingWriter::Create(
+ output, /*dictionary_block_size=*/1,
compression_factory, pool));
+
+ ASSERT_OK(writer->Write(SerializedString("apple", pool.get()),
RoaringBitmap64::From({0, 2})));
+ ASSERT_OK(writer->Write(SerializedString("banana", pool.get()),
RoaringBitmap64::From({3})));
+ ASSERT_OK(writer->Finish(RoaringBitmap64::From({1}),
RoaringBitmap64::From({0, 2, 3})));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Bytes> bytes,
output->Finish(pool.get()));
+
+ BytesSeekableReader reader(bytes, pool);
+ ASSERT_OK_AND_ASSIGN(BitmapGlobalIndexFormat::Footer footer,
+ BitmapGlobalIndexFormat::ReadFooter(bytes->size(),
&reader));
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta> block_metas,
+ BitmapGlobalIndexFormat::ReadIndexBlock(footer.IndexBlock(), &reader,
pool.get()));
+ ASSERT_EQ(2, block_metas.size());
+
+ ASSERT_OK_AND_ASSIGN(
+ BitmapGlobalIndexFormat::DictionaryBlock first_block,
+ BitmapGlobalIndexFormat::ReadDictionaryBlock(block_metas.front(),
&reader, pool.get()));
+ ASSERT_EQ(1, first_block.Entries().size());
+ ASSERT_EQ("apple",
std::string(first_block.Entries()[0].Key().GetBytes()->data(),
+
first_block.Entries()[0].Key().GetBytes()->size()));
+ ASSERT_OK_AND_ASSIGN(
+ RoaringBitmap64 first_bitmap,
+
BitmapGlobalIndexFormat::ReadBitmap(first_block.Entries()[0].BitmapBlock(),
&reader));
+ ASSERT_EQ(RoaringBitmap64::From({0, 2}), first_bitmap);
+
+ ASSERT_OK_AND_ASSIGN(
+ BitmapGlobalIndexFormat::DictionaryBlock second_block,
+ BitmapGlobalIndexFormat::ReadDictionaryBlock(block_metas.back(),
&reader, pool.get()));
+ ASSERT_EQ(1, second_block.Entries().size());
+ ASSERT_EQ("banana",
std::string(second_block.Entries()[0].Key().GetBytes()->data(),
+
second_block.Entries()[0].Key().GetBytes()->size()));
+ ASSERT_OK_AND_ASSIGN(
+ RoaringBitmap64 second_bitmap,
+
BitmapGlobalIndexFormat::ReadBitmap(second_block.Entries()[0].BitmapBlock(),
&reader));
+ ASSERT_EQ(RoaringBitmap64::From({3}), second_bitmap);
+
+ ASSERT_OK_AND_ASSIGN(RoaringBitmap64 null_rows,
+
BitmapGlobalIndexFormat::ReadBitmap(footer.NullRowsBlock(), &reader));
+ ASSERT_EQ(RoaringBitmap64::From({1}), null_rows);
+ ASSERT_OK_AND_ASSIGN(RoaringBitmap64 non_null_rows,
+
BitmapGlobalIndexFormat::ReadBitmap(footer.NonNullRowsBlock(), &reader));
+ ASSERT_EQ(RoaringBitmap64::From({0, 2, 3}), non_null_rows);
+}
+
+TEST(BitmapGlobalIndexFormatTest, RoundTripEmptyNullBitmap) {
+ std::shared_ptr<MemoryPool> pool = GetDefaultPool();
+ std::shared_ptr<ByteArrayOutputStream> output = CreateOutput(pool);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BlockCompressionFactory>
compression_factory,
+
BlockCompressionFactory::Create(BlockCompressionType::NONE));
+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<BitmapGlobalIndexFormat::StreamingWriter>
writer,
+ BitmapGlobalIndexFormat::StreamingWriter::Create(
+ output, /*dictionary_block_size=*/4096,
compression_factory, pool));
+
+ ASSERT_OK(writer->Write(SerializedString("apple", pool.get()),
RoaringBitmap64::From({0})));
+ ASSERT_OK(writer->Finish(RoaringBitmap64(), RoaringBitmap64::From({0})));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Bytes> bytes,
output->Finish(pool.get()));
+
+ BytesSeekableReader reader(bytes, pool);
+ ASSERT_OK_AND_ASSIGN(BitmapGlobalIndexFormat::Footer footer,
+ BitmapGlobalIndexFormat::ReadFooter(bytes->size(),
&reader));
+ ASSERT_OK_AND_ASSIGN(RoaringBitmap64 null_rows,
+
BitmapGlobalIndexFormat::ReadBitmap(footer.NullRowsBlock(), &reader));
+ ASSERT_TRUE(null_rows.IsEmpty());
+ ASSERT_OK_AND_ASSIGN(RoaringBitmap64 non_null_rows,
+
BitmapGlobalIndexFormat::ReadBitmap(footer.NonNullRowsBlock(), &reader));
+ ASSERT_EQ(RoaringBitmap64::From({0}), non_null_rows);
+}
+
+} // namespace paimon::test
diff --git
a/src/paimon/common/global_index/bitmap/bitmap_global_index_writer.cpp
b/src/paimon/common/global_index/bitmap/bitmap_global_index_writer.cpp
new file mode 100644
index 00000000..f7f5865d
--- /dev/null
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_writer.cpp
@@ -0,0 +1,182 @@
+/*
+ * 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/global_index/bitmap/bitmap_global_index_writer.h"
+
+#include <limits>
+#include <utility>
+
+#include "arrow/c/bridge.h"
+#include "fmt/format.h"
+#include "paimon/common/global_index/global_index_utils.h"
+#include "paimon/common/global_index/key_serializer.h"
+#include "paimon/common/global_index/sorted_index_file_meta.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/preconditions.h"
+
+namespace paimon {
+namespace {
+
+constexpr char kBitmapIdentifier[] = "bitmap";
+
+} // namespace
+
+Result<std::shared_ptr<BitmapGlobalIndexWriter>>
BitmapGlobalIndexWriter::Create(
+ const std::string& field_name, const std::shared_ptr<arrow::StructType>&
arrow_type,
+ const std::shared_ptr<GlobalIndexFileWriter>& file_writer, int32_t
dictionary_block_size,
+ const std::shared_ptr<BlockCompressionFactory>& compression_factory,
+ const std::shared_ptr<MemoryPool>& pool) {
+ if (arrow_type == nullptr || file_writer == nullptr || pool == nullptr) {
+ return Status::Invalid(
+ "Cannot create BitmapGlobalIndexWriter without schema, file
writer, and memory pool.");
+ }
+ if (dictionary_block_size <= 0) {
+ return Status::Invalid("Bitmap dictionary block size must be greater
than 0.");
+ }
+ std::shared_ptr<arrow::Field> key_field =
arrow_type->GetFieldByName(field_name);
+ PAIMON_RETURN_NOT_OK(Preconditions::CheckNotNull(
+ key_field, fmt::format("field {} not in arrow_array when Create
BitmapGlobalIndexWriter",
+ field_name)));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeySerializer> key_serializer,
+ KeySerializer::Create(key_field->type(), pool));
+ return std::shared_ptr<BitmapGlobalIndexWriter>(
+ new BitmapGlobalIndexWriter(field_name, arrow_type,
std::move(key_serializer), file_writer,
+ dictionary_block_size,
compression_factory, pool));
+}
+
+Status BitmapGlobalIndexWriter::AddBatch(::ArrowArray* arrow_array,
+ std::vector<int64_t>&&
relative_row_ids) {
+ if (finished_) {
+ return Status::Invalid("Cannot add a batch to a finished
BitmapGlobalIndexWriter.");
+ }
+ PAIMON_RETURN_NOT_OK(GlobalIndexUtils::CheckRelativeRowIds(
+ arrow_array, relative_row_ids, /*expected_next_row_id=*/std::nullopt));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
+ arrow::ImportArray(arrow_array,
arrow_type_));
+ if (array == nullptr || array->type_id() != arrow::Type::STRUCT) {
+ return Status::Invalid(
+ "arrow array must be struct array when AddBatch to
BitmapGlobalIndexWriter");
+ }
+ std::shared_ptr<arrow::StructArray> struct_array =
+ checked_pointer_cast<arrow::StructArray>(array);
+ std::shared_ptr<arrow::Array> value_array =
struct_array->GetFieldByName(field_name_);
+ PAIMON_RETURN_NOT_OK(Preconditions::CheckNotNull(
+ value_array,
+ fmt::format("field {} not in arrow_array when AddBatch to
BitmapGlobalIndexWriter",
+ field_name_)));
+ PAIMON_ASSIGN_OR_RAISE(std::vector<Literal> literals,
+
LiteralConverter::ConvertLiteralsFromArray(*value_array,
+
/*own_data=*/true));
+ for (size_t i = 0; i < literals.size(); ++i) {
+ if (row_count_ == std::numeric_limits<int64_t>::max()) {
+ return Status::Invalid("Bitmap global index row count exceeds
INT64_MAX.");
+ }
+ ++row_count_;
+ int64_t row_id = relative_row_ids[i];
+ const Literal& literal = literals[i];
+ if (literal.IsNull()) {
+ null_rows_.Add(row_id);
+ continue;
+ }
+
+ non_null_rows_.Add(row_id);
+ if (last_key_.has_value()) {
+ PAIMON_ASSIGN_OR_RAISE(int32_t comparison,
literal.CompareTo(last_key_.value()));
+ if (comparison < 0) {
+ return Status::Invalid(
+ "Bitmap index keys must be written in monotonically
increasing order.");
+ }
+ if (comparison > 0) {
+ PAIMON_RETURN_NOT_OK(FlushCurrentBitmap());
+ }
+ }
+ if (!first_key_.has_value()) {
+ first_key_ = literal;
+ }
+ last_key_ = literal;
+ current_bitmap_.Add(row_id);
+ }
+ return Status::OK();
+}
+
+Result<std::vector<GlobalIndexIOMeta>> BitmapGlobalIndexWriter::Finish() {
+ if (finished_) {
+ return Status::Invalid("BitmapGlobalIndexWriter has already been
finished.");
+ }
+ finished_ = true;
+ if (row_count_ == 0) {
+ return std::vector<GlobalIndexIOMeta>();
+ }
+
+ PAIMON_RETURN_NOT_OK(FlushCurrentBitmap());
+ PAIMON_ASSIGN_OR_RAISE(BitmapGlobalIndexFormat::StreamingWriter *
streaming_writer,
+ GetOrCreateStreamingWriter());
+ PAIMON_RETURN_NOT_OK(streaming_writer->Finish(null_rows_, non_null_rows_));
+ PAIMON_RETURN_NOT_OK(output_stream_->Close());
+
+ std::shared_ptr<Bytes> first_key_bytes;
+ std::shared_ptr<Bytes> last_key_bytes;
+ if (first_key_.has_value()) {
+ PAIMON_ASSIGN_OR_RAISE(first_key_bytes,
key_serializer_->Serialize(first_key_.value()));
+ }
+ if (last_key_.has_value()) {
+ PAIMON_ASSIGN_OR_RAISE(last_key_bytes,
key_serializer_->Serialize(last_key_.value()));
+ }
+ SortedIndexFileMeta index_meta(first_key_bytes, last_key_bytes,
!null_rows_.IsEmpty());
+ std::shared_ptr<Bytes> metadata = index_meta.Serialize(pool_.get());
+ PAIMON_ASSIGN_OR_RAISE(int64_t file_size,
file_writer_->GetFileSize(file_name_));
+ return std::vector<GlobalIndexIOMeta>{
+ GlobalIndexIOMeta(file_writer_->ToPath(file_name_), file_size,
std::move(metadata))};
+}
+
+Status BitmapGlobalIndexWriter::FlushCurrentBitmap() {
+ if (current_bitmap_.IsEmpty()) {
+ return Status::OK();
+ }
+ if (!last_key_.has_value()) {
+ return Status::Invalid("BitmapGlobalIndexWriter has a bitmap without a
dictionary key.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(BitmapGlobalIndexFormat::StreamingWriter *
streaming_writer,
+ GetOrCreateStreamingWriter());
+ PAIMON_ASSIGN_OR_RAISE(
+ BitmapGlobalIndexFormat::SerializedKey key,
+ BitmapGlobalIndexFormat::SerializedKey::FromLiteral(key_serializer_,
last_key_.value()));
+ PAIMON_RETURN_NOT_OK(streaming_writer->Write(std::move(key),
current_bitmap_));
+ current_bitmap_ = RoaringBitmap64();
+ return Status::OK();
+}
+
+Result<BitmapGlobalIndexFormat::StreamingWriter*>
+BitmapGlobalIndexWriter::GetOrCreateStreamingWriter() {
+ if (streaming_writer_ != nullptr) {
+ return streaming_writer_.get();
+ }
+ PAIMON_ASSIGN_OR_RAISE(file_name_,
file_writer_->NewFileName(kBitmapIdentifier));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> output_stream,
+ file_writer_->NewOutputStream(file_name_));
+ output_stream_ = std::move(output_stream);
+ PAIMON_ASSIGN_OR_RAISE(streaming_writer_,
BitmapGlobalIndexFormat::StreamingWriter::Create(
+ output_stream_,
dictionary_block_size_,
+ compression_factory_,
pool_));
+ return streaming_writer_.get();
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/global_index/bitmap/bitmap_global_index_writer.h
b/src/paimon/common/global_index/bitmap/bitmap_global_index_writer.h
new file mode 100644
index 00000000..3f7e8a2b
--- /dev/null
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_writer.h
@@ -0,0 +1,97 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "arrow/api.h"
+#include "paimon/common/compression/block_compression_factory.h"
+#include "paimon/common/global_index/bitmap/bitmap_global_index_format.h"
+#include "paimon/global_index/global_index_writer.h"
+#include "paimon/global_index/io/global_index_file_writer.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+
+class KeySerializer;
+
+/// Streaming global index writer for the Java-compatible bitmap format.
+///
+/// Non-null keys must be written in monotonically increasing order so
completed bitmaps can be
+/// streamed to the output file instead of retained until Finish().
+class BitmapGlobalIndexWriter : public GlobalIndexWriter {
+ public:
+ static Result<std::shared_ptr<BitmapGlobalIndexWriter>> Create(
+ const std::string& field_name, const
std::shared_ptr<arrow::StructType>& arrow_type,
+ const std::shared_ptr<GlobalIndexFileWriter>& file_writer, int32_t
dictionary_block_size,
+ const std::shared_ptr<BlockCompressionFactory>& compression_factory,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ ~BitmapGlobalIndexWriter() override = default;
+
+ Status AddBatch(::ArrowArray* arrow_array, std::vector<int64_t>&&
relative_row_ids) override;
+
+ Result<std::vector<GlobalIndexIOMeta>> Finish() override;
+
+ private:
+ BitmapGlobalIndexWriter(std::string field_name,
std::shared_ptr<arrow::DataType> arrow_type,
+ std::shared_ptr<KeySerializer> key_serializer,
+ std::shared_ptr<GlobalIndexFileWriter> file_writer,
+ int32_t dictionary_block_size,
+ std::shared_ptr<BlockCompressionFactory>
compression_factory,
+ std::shared_ptr<MemoryPool> pool)
+ : field_name_(std::move(field_name)),
+ arrow_type_(std::move(arrow_type)),
+ key_serializer_(std::move(key_serializer)),
+ file_writer_(std::move(file_writer)),
+ dictionary_block_size_(dictionary_block_size),
+ compression_factory_(std::move(compression_factory)),
+ pool_(std::move(pool)) {}
+
+ Status FlushCurrentBitmap();
+
+ Result<BitmapGlobalIndexFormat::StreamingWriter*>
GetOrCreateStreamingWriter();
+
+ std::string field_name_;
+ std::shared_ptr<arrow::DataType> arrow_type_;
+ std::shared_ptr<KeySerializer> key_serializer_;
+ std::shared_ptr<GlobalIndexFileWriter> file_writer_;
+ int32_t dictionary_block_size_;
+ std::shared_ptr<BlockCompressionFactory> compression_factory_;
+ std::shared_ptr<MemoryPool> pool_;
+
+ std::string file_name_;
+ std::shared_ptr<OutputStream> output_stream_;
+ std::unique_ptr<BitmapGlobalIndexFormat::StreamingWriter>
streaming_writer_;
+ int64_t row_count_ = 0;
+ std::optional<Literal> first_key_;
+ std::optional<Literal> last_key_;
+ RoaringBitmap64 current_bitmap_;
+ RoaringBitmap64 null_rows_;
+ RoaringBitmap64 non_null_rows_;
+ bool finished_ = false;
+};
+
+} // namespace paimon
diff --git a/src/paimon/common/global_index/bitmap/bitmap_index_reader.cpp
b/src/paimon/common/global_index/bitmap/bitmap_index_reader.cpp
new file mode 100644
index 00000000..8a38a67a
--- /dev/null
+++ b/src/paimon/common/global_index/bitmap/bitmap_index_reader.cpp
@@ -0,0 +1,579 @@
+/*
+ * 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/global_index/bitmap/bitmap_index_reader.h"
+
+#include <algorithm>
+#include <cstring>
+#include <set>
+#include <string_view>
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/common/global_index/key_serializer.h"
+#include "paimon/common/predicate/like.h"
+#include "paimon/global_index/bitmap_global_index_result.h"
+#include "paimon/predicate/literal.h"
+
+namespace paimon {
+
+Result<std::shared_ptr<BitmapIndexReader>> BitmapIndexReader::Create(
+ const std::shared_ptr<KeySerializer>& key_serializer,
+ const std::shared_ptr<GlobalIndexFileReader>& file_reader, const
GlobalIndexIOMeta& meta,
+ const std::shared_ptr<MemoryPool>& pool) {
+ if (key_serializer == nullptr || file_reader == nullptr || pool ==
nullptr) {
+ return Status::Invalid(
+ "Cannot create BitmapIndexReader without serializer, file reader,
and memory pool.");
+ }
+ if (meta.file_size < 0) {
+ return Status::Invalid("Cannot create BitmapIndexReader with a
negative file size.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input,
+ file_reader->GetInputStream(meta.file_path));
+ std::shared_ptr<BitmapIndexReader> reader(
+ new BitmapIndexReader(key_serializer, std::move(input),
meta.file_size, pool));
+ Result<BitmapGlobalIndexFormat::Footer> footer =
+ BitmapGlobalIndexFormat::ReadFooter(meta.file_size, reader.get());
+ if (!footer.ok()) {
+ [[maybe_unused]] Status close_status = reader->Close();
+ return footer.status();
+ }
+ reader->footer_ = std::move(footer).value();
+ return reader;
+}
+
+BitmapIndexReader::~BitmapIndexReader() {
+ [[maybe_unused]] Status close_status = Close();
+}
+
+std::shared_ptr<GlobalIndexResult> BitmapIndexReader::CreateResult(
+ std::function<Result<RoaringBitmap64>()> supplier) {
+ return std::make_shared<BitmapGlobalIndexResult>(std::move(supplier));
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitIsNotNull()
{
+ return CreateResult([reader = shared_from_this()]() { return
reader->IsNotNull(); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitIsNull() {
+ return CreateResult([reader = shared_from_this()]() { return
reader->IsNull(); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitEqual(const
Literal& literal) {
+ return CreateResult(
+ [reader = shared_from_this(), literal]() { return
reader->Equal(literal); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitNotEqual(
+ const Literal& literal) {
+ return CreateResult(
+ [reader = shared_from_this(), literal]() { return
reader->NotEqual(literal); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitLessThan(
+ const Literal& literal) {
+ return CreateResult(
+ [reader = shared_from_this(), literal]() { return
reader->LessThan(literal); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitLessOrEqual(
+ const Literal& literal) {
+ return CreateResult(
+ [reader = shared_from_this(), literal]() { return
reader->LessOrEqual(literal); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitGreaterThan(
+ const Literal& literal) {
+ return CreateResult(
+ [reader = shared_from_this(), literal]() { return
reader->GreaterThan(literal); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
BitmapIndexReader::VisitGreaterOrEqual(
+ const Literal& literal) {
+ return CreateResult(
+ [reader = shared_from_this(), literal]() { return
reader->GreaterOrEqual(literal); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitIn(
+ const std::vector<Literal>& literals) {
+ return CreateResult([reader = shared_from_this(), literals]() { return
reader->In(literals); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitNotIn(
+ const std::vector<Literal>& literals) {
+ return CreateResult(
+ [reader = shared_from_this(), literals]() { return
reader->NotIn(literals); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitStartsWith(
+ const Literal& prefix) {
+ return CreateResult(
+ [reader = shared_from_this(), prefix]() { return
reader->StartsWith(prefix); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
BitmapIndexReader::VisitEndsWith(const Literal& suffix) {
+ return CreateResult(
+ [reader = shared_from_this(), suffix]() { return
reader->EndsWith(suffix); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitContains(
+ const Literal& literal) {
+ return CreateResult(
+ [reader = shared_from_this(), literal]() { return
reader->Contains(literal); });
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> BitmapIndexReader::VisitLike(const
Literal& literal) {
+ return CreateResult([reader = shared_from_this(), literal]() { return
reader->Like(literal); });
+}
+
+Result<std::shared_ptr<ScoredGlobalIndexResult>>
BitmapIndexReader::VisitVectorSearch(
+ const std::shared_ptr<VectorSearch>& vector_search) {
+ return Status::Invalid("Vector search is not supported in
BitmapIndexReader.");
+}
+
+Result<std::shared_ptr<GlobalIndexResult>>
BitmapIndexReader::VisitFullTextSearch(
+ const std::shared_ptr<FullTextSearch>& full_text_search) {
+ return Status::Invalid("Full text search is not supported in
BitmapIndexReader.");
+}
+
+Result<std::shared_ptr<Bytes>> BitmapIndexReader::Read(int64_t offset, int32_t
length) {
+ if (closed_ || input_ == nullptr) {
+ return Status::Invalid("Cannot read from a closed BitmapIndexReader.");
+ }
+ if (offset < 0 || length < 0 || offset > file_size_ || length > file_size_
- offset) {
+ return Status::Invalid(fmt::format("Bitmap index read range [{}, {})
exceeds file size {}.",
+ offset, offset + length,
file_size_));
+ }
+ std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(length, pool_.get());
+ if (length == 0) {
+ return bytes;
+ }
+ PAIMON_ASSIGN_OR_RAISE(int64_t actual_length, input_->Read(bytes->data(),
length, offset));
+ if (actual_length != length) {
+ return Status::IOError(
+ fmt::format("Truncated bitmap index read: expected {} bytes at {},
but read {}.",
+ length, offset, actual_length));
+ }
+ return bytes;
+}
+
+Status BitmapIndexReader::Close() {
+ if (closed_) {
+ return Status::OK();
+ }
+ closed_ = true;
+ if (input_ == nullptr) {
+ return Status::OK();
+ }
+ return input_->Close();
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::IsNull() {
+ PAIMON_ASSIGN_OR_RAISE(const RoaringBitmap64* bitmap, GetNullRows());
+ return *bitmap;
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::IsNotNull() {
+ PAIMON_ASSIGN_OR_RAISE(const RoaringBitmap64* bitmap, GetNonNullRows());
+ return *bitmap;
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::Equal(const Literal& literal) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::optional<BitmapGlobalIndexFormat::BlockInfo>
bitmap_block,
+ FindBitmapBlock(literal));
+ if (!bitmap_block.has_value()) {
+ return RoaringBitmap64();
+ }
+ return BitmapGlobalIndexFormat::ReadBitmap(bitmap_block.value(), this);
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::In(const std::vector<Literal>&
literals) {
+ RoaringBitmap64 result;
+ std::set<std::string> serialized_keys;
+ for (const Literal& literal : literals) {
+ if (literal.IsNull()) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ BitmapGlobalIndexFormat::SerializedKey serialized_key,
+
BitmapGlobalIndexFormat::SerializedKey::FromLiteral(key_serializer_, literal));
+ const std::shared_ptr<Bytes>& key_bytes = serialized_key.GetBytes();
+ if (!serialized_keys.emplace(key_bytes->data(),
key_bytes->size()).second) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 bitmap, Equal(literal));
+ result |= bitmap;
+ }
+ return result;
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::StartsWith(const Literal& literal) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ if (literal.GetType() != FieldType::STRING) {
+ return Status::Invalid("StartsWith requires a string literal in
BitmapIndexReader.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ BitmapGlobalIndexFormat::SerializedKey prefix,
+ BitmapGlobalIndexFormat::SerializedKey::FromLiteral(key_serializer_,
literal));
+ if (prefix.GetBytes()->size() == 0) {
+ return IsNotNull();
+ }
+ std::optional<BitmapGlobalIndexFormat::SerializedKey> upper_bound =
+ PrefixUpperBound(prefix, pool_.get());
+ PAIMON_ASSIGN_OR_RAISE(const
std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>* blocks,
+ GetDictionaryBlocks());
+ int32_t index = std::max(FindSerializedDictionaryBlockIndex(*blocks,
prefix), 0);
+ RoaringBitmap64 result;
+ while (index < static_cast<int32_t>(blocks->size())) {
+ const BitmapGlobalIndexFormat::DictionaryBlockMeta& block_meta =
(*blocks)[index];
+ if (upper_bound.has_value() &&
block_meta.FirstKey().CompareTo(upper_bound.value()) >= 0) {
+ return result;
+ }
+ PAIMON_ASSIGN_OR_RAISE(const BitmapGlobalIndexFormat::DictionaryBlock*
block,
+ GetDictionaryBlock(block_meta));
+ for (const BitmapGlobalIndexFormat::DictionaryEntry& entry :
block->Entries()) {
+ if (entry.Key().CompareTo(prefix) < 0) {
+ continue;
+ }
+ if (!StartsWith(entry.Key(), prefix)) {
+ return result;
+ }
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 bitmap,
+
BitmapGlobalIndexFormat::ReadBitmap(entry.BitmapBlock(), this));
+ result |= bitmap;
+ }
+ ++index;
+ }
+ return result;
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::EndsWith(const Literal& literal) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ if (literal.GetType() != FieldType::STRING) {
+ return Status::Invalid("EndsWith requires a string literal in
BitmapIndexReader.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ BitmapGlobalIndexFormat::SerializedKey suffix,
+ BitmapGlobalIndexFormat::SerializedKey::FromLiteral(key_serializer_,
literal));
+ if (suffix.GetBytes()->size() == 0) {
+ return IsNotNull();
+ }
+ return ScanSerializedDictionary(
+ [&suffix](const BitmapGlobalIndexFormat::SerializedKey& key) ->
Result<bool> {
+ return EndsWith(key, suffix);
+ });
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::Contains(const Literal& literal) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ if (literal.GetType() != FieldType::STRING) {
+ return Status::Invalid("Contains requires a string literal in
BitmapIndexReader.");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ BitmapGlobalIndexFormat::SerializedKey infix,
+ BitmapGlobalIndexFormat::SerializedKey::FromLiteral(key_serializer_,
literal));
+ if (infix.GetBytes()->size() == 0) {
+ return IsNotNull();
+ }
+ return ScanSerializedDictionary(
+ [&infix](const BitmapGlobalIndexFormat::SerializedKey& key) ->
Result<bool> {
+ return Contains(key, infix);
+ });
+}
+
+// TODO(xinyu.lxy): Optimize range predicates by locating the dictionary
lower/upper bound and
+// scanning only the matching side instead of traversing the entire dictionary.
+Result<RoaringBitmap64> BitmapIndexReader::LessThan(const Literal& literal) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ return ScanDictionary([&literal](const Literal& key) -> Result<bool> {
+ PAIMON_ASSIGN_OR_RAISE(int32_t comparison, key.CompareTo(literal));
+ return comparison < 0;
+ });
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::LessOrEqual(const Literal& literal)
{
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ return ScanDictionary([&literal](const Literal& key) -> Result<bool> {
+ PAIMON_ASSIGN_OR_RAISE(int32_t comparison, key.CompareTo(literal));
+ return comparison <= 0;
+ });
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::GreaterThan(const Literal& literal)
{
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ return ScanDictionary([&literal](const Literal& key) -> Result<bool> {
+ PAIMON_ASSIGN_OR_RAISE(int32_t comparison, key.CompareTo(literal));
+ return comparison > 0;
+ });
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::GreaterOrEqual(const Literal&
literal) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ return ScanDictionary([&literal](const Literal& key) -> Result<bool> {
+ PAIMON_ASSIGN_OR_RAISE(int32_t comparison, key.CompareTo(literal));
+ return comparison >= 0;
+ });
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::NotEqual(const Literal& literal) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 result, IsNotNull());
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 equal, Equal(literal));
+ result -= equal;
+ return result;
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::NotIn(const std::vector<Literal>&
literals) {
+ for (const Literal& literal : literals) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 result, IsNotNull());
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 in, In(literals));
+ result -= in;
+ return result;
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::Like(const Literal& literal) {
+ if (literal.IsNull()) {
+ return RoaringBitmap64();
+ }
+ if (literal.GetType() != FieldType::STRING) {
+ return Status::Invalid("LIKE requires a string literal in
BitmapIndexReader.");
+ }
+ auto pattern = literal.GetValue<std::string>();
+ return ScanDictionary([pattern = std::move(pattern)](const Literal& key)
-> Result<bool> {
+ if (key.GetType() != FieldType::STRING) {
+ return false;
+ }
+ return
paimon::Like::Instance().TestString(key.GetValue<std::string>(), pattern);
+ });
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::ScanDictionary(const
LiteralPredicate& predicate) {
+ PAIMON_ASSIGN_OR_RAISE(
+ const std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>*
block_metas,
+ GetDictionaryBlocks());
+ RoaringBitmap64 result;
+ for (const BitmapGlobalIndexFormat::DictionaryBlockMeta& block_meta :
*block_metas) {
+ PAIMON_ASSIGN_OR_RAISE(const BitmapGlobalIndexFormat::DictionaryBlock*
block,
+ GetDictionaryBlock(block_meta));
+ for (const BitmapGlobalIndexFormat::DictionaryEntry& entry :
block->Entries()) {
+ PAIMON_ASSIGN_OR_RAISE(Literal key, key_serializer_->Deserialize(
+
MemorySlice::Wrap(entry.Key().GetBytes())));
+ PAIMON_ASSIGN_OR_RAISE(bool matches, predicate(key));
+ if (matches) {
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 bitmap,
BitmapGlobalIndexFormat::ReadBitmap(
+
entry.BitmapBlock(), this));
+ result |= bitmap;
+ }
+ }
+ }
+ return result;
+}
+
+Result<RoaringBitmap64> BitmapIndexReader::ScanSerializedDictionary(
+ const DictionaryPredicate& predicate) {
+ PAIMON_ASSIGN_OR_RAISE(
+ const std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>*
block_metas,
+ GetDictionaryBlocks());
+ RoaringBitmap64 result;
+ for (const BitmapGlobalIndexFormat::DictionaryBlockMeta& block_meta :
*block_metas) {
+ PAIMON_ASSIGN_OR_RAISE(const BitmapGlobalIndexFormat::DictionaryBlock*
block,
+ GetDictionaryBlock(block_meta));
+ for (const BitmapGlobalIndexFormat::DictionaryEntry& entry :
block->Entries()) {
+ PAIMON_ASSIGN_OR_RAISE(bool matches, predicate(entry.Key()));
+ if (matches) {
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 bitmap,
BitmapGlobalIndexFormat::ReadBitmap(
+
entry.BitmapBlock(), this));
+ result |= bitmap;
+ }
+ }
+ }
+ return result;
+}
+
+Result<std::optional<BitmapGlobalIndexFormat::BlockInfo>>
BitmapIndexReader::FindBitmapBlock(
+ const Literal& literal) {
+ PAIMON_ASSIGN_OR_RAISE(const
std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>* blocks,
+ GetDictionaryBlocks());
+ if (blocks->empty()) {
+ return std::optional<BitmapGlobalIndexFormat::BlockInfo>();
+ }
+ PAIMON_ASSIGN_OR_RAISE(int32_t index,
FindLogicalDictionaryBlockIndex(*blocks, literal));
+ if (index < 0) {
+ return std::optional<BitmapGlobalIndexFormat::BlockInfo>();
+ }
+ PAIMON_ASSIGN_OR_RAISE(const BitmapGlobalIndexFormat::DictionaryBlock*
block,
+ GetDictionaryBlock((*blocks)[index]));
+ for (const BitmapGlobalIndexFormat::DictionaryEntry& entry :
block->Entries()) {
+ PAIMON_ASSIGN_OR_RAISE(
+ Literal key,
key_serializer_->Deserialize(MemorySlice::Wrap(entry.Key().GetBytes())));
+ PAIMON_ASSIGN_OR_RAISE(int32_t comparison, key.CompareTo(literal));
+ if (comparison == 0) {
+ return
std::optional<BitmapGlobalIndexFormat::BlockInfo>(entry.BitmapBlock());
+ }
+ if (comparison > 0) {
+ break;
+ }
+ }
+ return std::optional<BitmapGlobalIndexFormat::BlockInfo>();
+}
+
+Result<const std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>*>
+BitmapIndexReader::GetDictionaryBlocks() {
+ if (!dictionary_blocks_.has_value()) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta> blocks,
+ BitmapGlobalIndexFormat::ReadIndexBlock(footer_.IndexBlock(),
this, pool_.get()));
+ dictionary_blocks_ = std::move(blocks);
+ }
+ return &dictionary_blocks_.value();
+}
+
+Result<const BitmapGlobalIndexFormat::DictionaryBlock*>
BitmapIndexReader::GetDictionaryBlock(
+ const BitmapGlobalIndexFormat::DictionaryBlockMeta& block_meta) {
+ auto iterator = dictionary_block_cache_.find(block_meta.Offset());
+ if (iterator != dictionary_block_cache_.end()) {
+ return iterator->second.get();
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ BitmapGlobalIndexFormat::DictionaryBlock block,
+ BitmapGlobalIndexFormat::ReadDictionaryBlock(block_meta, this,
pool_.get()));
+ auto inserted = dictionary_block_cache_.emplace(
+ block_meta.Offset(),
+
std::make_shared<BitmapGlobalIndexFormat::DictionaryBlock>(std::move(block)));
+ return inserted.first->second.get();
+}
+
+Result<const RoaringBitmap64*> BitmapIndexReader::GetNullRows() {
+ if (!null_rows_.has_value()) {
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 bitmap,
+
BitmapGlobalIndexFormat::ReadBitmap(footer_.NullRowsBlock(), this));
+ null_rows_ = std::move(bitmap);
+ }
+ return &null_rows_.value();
+}
+
+Result<const RoaringBitmap64*> BitmapIndexReader::GetNonNullRows() {
+ if (!non_null_rows_.has_value()) {
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap64 bitmap,
BitmapGlobalIndexFormat::ReadBitmap(
+
footer_.NonNullRowsBlock(), this));
+ non_null_rows_ = std::move(bitmap);
+ }
+ return &non_null_rows_.value();
+}
+
+Result<int32_t> BitmapIndexReader::FindLogicalDictionaryBlockIndex(
+ const std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>& blocks,
+ const Literal& literal) {
+ int32_t low = 0;
+ auto high = static_cast<int32_t>(blocks.size()) - 1;
+ while (low <= high) {
+ int32_t middle = low + ((high - low) / 2);
+ PAIMON_ASSIGN_OR_RAISE(
+ Literal first_key,
+
key_serializer_->Deserialize(MemorySlice::Wrap(blocks[middle].FirstKey().GetBytes())));
+ PAIMON_ASSIGN_OR_RAISE(int32_t comparison,
first_key.CompareTo(literal));
+ if (comparison <= 0) {
+ low = middle + 1;
+ } else {
+ high = middle - 1;
+ }
+ }
+ return high;
+}
+
+int32_t BitmapIndexReader::FindSerializedDictionaryBlockIndex(
+ const std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>& blocks,
+ const BitmapGlobalIndexFormat::SerializedKey& key) {
+ int32_t low = 0;
+ auto high = static_cast<int32_t>(blocks.size()) - 1;
+ while (low <= high) {
+ int32_t middle = low + ((high - low) / 2);
+ int32_t comparison = blocks[middle].FirstKey().CompareTo(key);
+ if (comparison <= 0) {
+ low = middle + 1;
+ } else {
+ high = middle - 1;
+ }
+ }
+ return high;
+}
+
+bool BitmapIndexReader::StartsWith(const
BitmapGlobalIndexFormat::SerializedKey& key,
+ const
BitmapGlobalIndexFormat::SerializedKey& prefix) {
+ const std::shared_ptr<Bytes>& key_bytes = key.GetBytes();
+ const std::shared_ptr<Bytes>& prefix_bytes = prefix.GetBytes();
+ return key_bytes->size() >= prefix_bytes->size() &&
+ std::memcmp(key_bytes->data(), prefix_bytes->data(),
prefix_bytes->size()) == 0;
+}
+
+bool BitmapIndexReader::EndsWith(const BitmapGlobalIndexFormat::SerializedKey&
key,
+ const BitmapGlobalIndexFormat::SerializedKey&
suffix) {
+ const std::shared_ptr<Bytes>& key_bytes = key.GetBytes();
+ const std::shared_ptr<Bytes>& suffix_bytes = suffix.GetBytes();
+ return key_bytes->size() >= suffix_bytes->size() &&
+ std::memcmp(key_bytes->data() + key_bytes->size() -
suffix_bytes->size(),
+ suffix_bytes->data(), suffix_bytes->size()) == 0;
+}
+
+bool BitmapIndexReader::Contains(const BitmapGlobalIndexFormat::SerializedKey&
key,
+ const BitmapGlobalIndexFormat::SerializedKey&
infix) {
+ std::string_view key_bytes(key.GetBytes()->data(), key.GetBytes()->size());
+ std::string_view infix_bytes(infix.GetBytes()->data(),
infix.GetBytes()->size());
+ return key_bytes.find(infix_bytes) != std::string_view::npos;
+}
+
+std::optional<BitmapGlobalIndexFormat::SerializedKey>
BitmapIndexReader::PrefixUpperBound(
+ const BitmapGlobalIndexFormat::SerializedKey& prefix, MemoryPool* pool) {
+ const std::shared_ptr<Bytes>& prefix_bytes = prefix.GetBytes();
+ for (auto i = static_cast<int64_t>(prefix_bytes->size()) - 1; i >= 0; --i)
{
+ auto value = static_cast<uint8_t>(prefix_bytes->data()[i]);
+ if (value != 0xFF) {
+ std::shared_ptr<Bytes> upper_bound = Bytes::AllocateBytes(i + 1,
pool);
+ std::memcpy(upper_bound->data(), prefix_bytes->data(), i + 1);
+ upper_bound->data()[i] = static_cast<char>(value + 1);
+ return
BitmapGlobalIndexFormat::SerializedKey(std::move(upper_bound));
+ }
+ }
+ return std::nullopt;
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/global_index/bitmap/bitmap_index_reader.h
b/src/paimon/common/global_index/bitmap/bitmap_index_reader.h
new file mode 100644
index 00000000..41890050
--- /dev/null
+++ b/src/paimon/common/global_index/bitmap/bitmap_index_reader.h
@@ -0,0 +1,160 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "paimon/common/global_index/bitmap/bitmap_global_index_format.h"
+#include "paimon/global_index/global_index_io_meta.h"
+#include "paimon/global_index/global_index_reader.h"
+#include "paimon/global_index/io/global_index_file_reader.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+
+class KeySerializer;
+
+/// Reader for one Java-compatible bitmap global index file.
+class BitmapIndexReader : public GlobalIndexReader,
+ public BitmapGlobalIndexFormat::SeekableReader,
+ public
std::enable_shared_from_this<BitmapIndexReader> {
+ public:
+ static Result<std::shared_ptr<BitmapIndexReader>> Create(
+ const std::shared_ptr<KeySerializer>& key_serializer,
+ const std::shared_ptr<GlobalIndexFileReader>& file_reader, const
GlobalIndexIOMeta& meta,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ ~BitmapIndexReader() override;
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIsNotNull() override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIsNull() override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitEqual(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitNotEqual(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLessThan(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLessOrEqual(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitGreaterThan(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitGreaterOrEqual(const
Literal& literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIn(
+ const std::vector<Literal>& literals) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitNotIn(
+ const std::vector<Literal>& literals) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitStartsWith(const Literal&
prefix) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitEndsWith(const Literal&
suffix) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitContains(const Literal&
literal) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLike(const Literal&
literal) override;
+
+ Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitVectorSearch(
+ const std::shared_ptr<VectorSearch>& vector_search) override;
+ Result<std::shared_ptr<GlobalIndexResult>> VisitFullTextSearch(
+ const std::shared_ptr<FullTextSearch>& full_text_search) override;
+
+ bool IsThreadSafe() const override {
+ return false;
+ }
+
+ std::string GetIndexType() const override {
+ return "bitmap";
+ }
+
+ using BitmapGlobalIndexFormat::SeekableReader::Read;
+ Result<std::shared_ptr<Bytes>> Read(int64_t offset, int32_t length)
override;
+
+ Status Close();
+
+ private:
+ using DictionaryPredicate =
+ std::function<Result<bool>(const
BitmapGlobalIndexFormat::SerializedKey&)>;
+ using LiteralPredicate = std::function<Result<bool>(const Literal&)>;
+
+ BitmapIndexReader(std::shared_ptr<KeySerializer> key_serializer,
+ std::shared_ptr<InputStream> input, int64_t file_size,
+ std::shared_ptr<MemoryPool> pool)
+ : key_serializer_(std::move(key_serializer)),
+ input_(std::move(input)),
+ file_size_(file_size),
+ pool_(std::move(pool)) {}
+
+ std::shared_ptr<GlobalIndexResult> CreateResult(
+ std::function<Result<RoaringBitmap64>()> supplier);
+
+ Result<RoaringBitmap64> IsNull();
+ Result<RoaringBitmap64> IsNotNull();
+ Result<RoaringBitmap64> Equal(const Literal& literal);
+ Result<RoaringBitmap64> In(const std::vector<Literal>& literals);
+ Result<RoaringBitmap64> StartsWith(const Literal& literal);
+ Result<RoaringBitmap64> EndsWith(const Literal& literal);
+ Result<RoaringBitmap64> Contains(const Literal& literal);
+ Result<RoaringBitmap64> LessThan(const Literal& literal);
+ Result<RoaringBitmap64> LessOrEqual(const Literal& literal);
+ Result<RoaringBitmap64> GreaterThan(const Literal& literal);
+ Result<RoaringBitmap64> GreaterOrEqual(const Literal& literal);
+ Result<RoaringBitmap64> NotEqual(const Literal& literal);
+ Result<RoaringBitmap64> NotIn(const std::vector<Literal>& literals);
+ Result<RoaringBitmap64> Like(const Literal& literal);
+
+ Result<RoaringBitmap64> ScanDictionary(const LiteralPredicate& predicate);
+ Result<RoaringBitmap64> ScanSerializedDictionary(const
DictionaryPredicate& predicate);
+ Result<std::optional<BitmapGlobalIndexFormat::BlockInfo>> FindBitmapBlock(
+ const Literal& literal);
+
+ Result<const std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>*>
GetDictionaryBlocks();
+ Result<const BitmapGlobalIndexFormat::DictionaryBlock*> GetDictionaryBlock(
+ const BitmapGlobalIndexFormat::DictionaryBlockMeta& block_meta);
+ Result<const RoaringBitmap64*> GetNullRows();
+ Result<const RoaringBitmap64*> GetNonNullRows();
+
+ Result<int32_t> FindLogicalDictionaryBlockIndex(
+ const std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>&
blocks,
+ const Literal& literal);
+ static int32_t FindSerializedDictionaryBlockIndex(
+ const std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>&
blocks,
+ const BitmapGlobalIndexFormat::SerializedKey& key);
+
+ static bool StartsWith(const BitmapGlobalIndexFormat::SerializedKey& key,
+ const BitmapGlobalIndexFormat::SerializedKey&
prefix);
+ static bool EndsWith(const BitmapGlobalIndexFormat::SerializedKey& key,
+ const BitmapGlobalIndexFormat::SerializedKey& suffix);
+ static bool Contains(const BitmapGlobalIndexFormat::SerializedKey& key,
+ const BitmapGlobalIndexFormat::SerializedKey& infix);
+ static std::optional<BitmapGlobalIndexFormat::SerializedKey>
PrefixUpperBound(
+ const BitmapGlobalIndexFormat::SerializedKey& prefix, MemoryPool*
pool);
+
+ std::shared_ptr<KeySerializer> key_serializer_;
+ std::shared_ptr<InputStream> input_;
+ int64_t file_size_;
+ BitmapGlobalIndexFormat::Footer footer_{{0, 0}, {0, 0}, {0, 0}};
+ std::shared_ptr<MemoryPool> pool_;
+
+ std::optional<RoaringBitmap64> null_rows_;
+ std::optional<RoaringBitmap64> non_null_rows_;
+ std::optional<std::vector<BitmapGlobalIndexFormat::DictionaryBlockMeta>>
dictionary_blocks_;
+ std::unordered_map<int64_t,
std::shared_ptr<BitmapGlobalIndexFormat::DictionaryBlock>>
+ dictionary_block_cache_;
+ bool closed_ = false;
+};
+
+} // namespace paimon
diff --git a/src/paimon/common/global_index/bitmap/bitmap_index_reader_test.cpp
b/src/paimon/common/global_index/bitmap/bitmap_index_reader_test.cpp
new file mode 100644
index 00000000..cb6f58bd
--- /dev/null
+++ b/src/paimon/common/global_index/bitmap/bitmap_index_reader_test.cpp
@@ -0,0 +1,172 @@
+/*
+ * 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/global_index/bitmap/bitmap_index_reader.h"
+
+#include <memory>
+#include <numeric>
+#include <string>
+#include <vector>
+
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/global_index/bitmap/bitmap_global_index_writer.h"
+#include "paimon/common/global_index/key_serializer.h"
+#include "paimon/core/global_index/global_index_file_manager.h"
+#include "paimon/global_index/bitmap_global_index_result.h"
+#include "paimon/testing/mock/mock_index_path_factory.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class BitmapIndexReaderTest : public ::testing::Test {
+ protected:
+ // Use a small threshold to exercise dictionary block rollover and
cross-block reads.
+ static constexpr int32_t kBitmapDictionaryBlockSize = 24;
+
+ static Literal StringLiteral(const std::string& value) {
+ return Literal(FieldType::STRING, value.data(), value.size());
+ }
+
+ static void CheckResult(const Result<std::shared_ptr<GlobalIndexResult>>&
result,
+ const std::vector<int64_t>& expected) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> index_result,
result);
+ std::shared_ptr<BitmapGlobalIndexResult> bitmap_result =
+ std::dynamic_pointer_cast<BitmapGlobalIndexResult>(index_result);
+ ASSERT_TRUE(bitmap_result);
+ ASSERT_OK_AND_ASSIGN(const RoaringBitmap64* bitmap,
bitmap_result->GetBitmap());
+ ASSERT_EQ(RoaringBitmap64::From(expected), *bitmap);
+ }
+
+ void SetUp() override {
+ pool_ = GetDefaultPool();
+ test_dir_ = UniqueTestDirectory::Create("local");
+ ASSERT_TRUE(test_dir_);
+ file_manager_ = std::make_shared<GlobalIndexFileManager>(
+ test_dir_->GetFileSystem(),
std::make_shared<MockIndexPathFactory>(test_dir_->Str()),
+ /*checkpoint_path_factory=*/nullptr);
+ ASSERT_OK_AND_ASSIGN(compression_factory_,
+
BlockCompressionFactory::Create(BlockCompressionType::NONE));
+ }
+
+ Result<std::shared_ptr<BitmapGlobalIndexWriter>> CreateWriter(
+ const std::shared_ptr<arrow::Field>& field,
+ int32_t dictionary_block_size = kBitmapDictionaryBlockSize) const {
+ std::shared_ptr<arrow::StructType> struct_type =
+
std::static_pointer_cast<arrow::StructType>(arrow::struct_({field}));
+ return BitmapGlobalIndexWriter::Create(field->name(), struct_type,
file_manager_,
+ dictionary_block_size,
compression_factory_, pool_);
+ }
+
+ Result<std::shared_ptr<BitmapIndexReader>> CreateReader(
+ const std::shared_ptr<arrow::DataType>& type, const GlobalIndexIOMeta&
meta) const {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeySerializer> serializer,
+ KeySerializer::Create(type, pool_));
+ return BitmapIndexReader::Create(serializer, file_manager_, meta,
pool_);
+ }
+
+ std::shared_ptr<MemoryPool> pool_;
+ std::unique_ptr<UniqueTestDirectory> test_dir_;
+ std::shared_ptr<GlobalIndexFileManager> file_manager_;
+ std::shared_ptr<BlockCompressionFactory> compression_factory_;
+};
+
+TEST_F(BitmapIndexReaderTest, WriteAndReadStringPredicates) {
+ std::shared_ptr<arrow::Field> field = arrow::field("f0", arrow::utf8());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapGlobalIndexWriter> writer,
CreateWriter(field));
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({field}),
+ R"([["apple"], ["apple"], [null], ["banana"], ["band"], ["cab"],
[null]])")
+ .ValueOrDie();
+ ArrowArray c_array;
+ ASSERT_TRUE(arrow::ExportArray(*array, &c_array).ok());
+ std::vector<int64_t> row_ids(array->length());
+ std::iota(row_ids.begin(), row_ids.end(), 0);
+ ASSERT_OK(writer->AddBatch(&c_array, std::move(row_ids)));
+ ASSERT_OK_AND_ASSIGN(std::vector<GlobalIndexIOMeta> metas,
writer->Finish());
+ ASSERT_EQ(1, metas.size());
+ ASSERT_TRUE(metas[0].metadata);
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapIndexReader> reader,
+ CreateReader(field->type(), metas[0]));
+ CheckResult(reader->VisitIsNull(), {2, 6});
+ CheckResult(reader->VisitIsNotNull(), {0, 1, 3, 4, 5});
+ CheckResult(reader->VisitEqual(StringLiteral("apple")), {0, 1});
+ CheckResult(reader->VisitEqual(StringLiteral("missing")), {});
+ CheckResult(
+ reader->VisitIn({StringLiteral("apple"), StringLiteral("apple"),
StringLiteral("cab")}),
+ {0, 1, 5});
+ CheckResult(reader->VisitNotEqual(StringLiteral("banana")), {0, 1, 4, 5});
+ CheckResult(reader->VisitNotIn({StringLiteral("apple"),
StringLiteral("cab")}), {3, 4});
+
+ CheckResult(reader->VisitLessThan(StringLiteral("band")), {0, 1, 3});
+ CheckResult(reader->VisitLessOrEqual(StringLiteral("band")), {0, 1, 3, 4});
+ CheckResult(reader->VisitGreaterThan(StringLiteral("banana")), {4, 5});
+ CheckResult(reader->VisitGreaterOrEqual(StringLiteral("banana")), {3, 4,
5});
+
+ CheckResult(reader->VisitStartsWith(StringLiteral("ban")), {3, 4});
+ CheckResult(reader->VisitEndsWith(StringLiteral("le")), {0, 1});
+ CheckResult(reader->VisitContains(StringLiteral("an")), {3, 4});
+ CheckResult(reader->VisitLike(StringLiteral("ban_n_")), {3});
+ ASSERT_OK(reader->Close());
+}
+
+TEST_F(BitmapIndexReaderTest, WriteAndReadLogicalIntegerOrder) {
+ std::shared_ptr<arrow::Field> field = arrow::field("f0", arrow::int32());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapGlobalIndexWriter> writer,
+ CreateWriter(field, /*dictionary_block_size=*/8));
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({field}),
+ R"([[-257], [-1], [0], [1],
[256], [1000]])")
+ .ValueOrDie();
+ ArrowArray c_array;
+ ASSERT_TRUE(arrow::ExportArray(*array, &c_array).ok());
+ std::vector<int64_t> row_ids(array->length());
+ std::iota(row_ids.begin(), row_ids.end(), 0);
+ ASSERT_OK(writer->AddBatch(&c_array, std::move(row_ids)));
+ ASSERT_OK_AND_ASSIGN(std::vector<GlobalIndexIOMeta> metas,
writer->Finish());
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapIndexReader> reader,
+ CreateReader(field->type(), metas[0]));
+ CheckResult(reader->VisitEqual(Literal(static_cast<int32_t>(256))), {4});
+ CheckResult(reader->VisitLessThan(Literal(static_cast<int32_t>(1))), {0,
1, 2});
+ CheckResult(reader->VisitGreaterOrEqual(Literal(static_cast<int32_t>(0))),
{2, 3, 4, 5});
+}
+
+TEST_F(BitmapIndexReaderTest, RejectsDescendingKeys) {
+ std::shared_ptr<arrow::Field> field = arrow::field("f0", arrow::int32());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapGlobalIndexWriter> writer,
CreateWriter(field));
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({field}),
R"([[2], [1]])")
+ .ValueOrDie();
+ ArrowArray c_array;
+ ASSERT_TRUE(arrow::ExportArray(*array, &c_array).ok());
+ ASSERT_NOK_WITH_MSG(writer->AddBatch(&c_array, {0, 1}), "monotonically
increasing");
+}
+
+TEST_F(BitmapIndexReaderTest, EmptyWriterCreatesNoFile) {
+ std::shared_ptr<arrow::Field> field = arrow::field("f0", arrow::utf8());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<BitmapGlobalIndexWriter> writer,
CreateWriter(field));
+ ASSERT_OK_AND_ASSIGN(std::vector<GlobalIndexIOMeta> metas,
writer->Finish());
+ ASSERT_TRUE(metas.empty());
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/io/data_output_stream.cpp
b/src/paimon/common/io/data_output_stream.cpp
index 53825a05..12e46c11 100644
--- a/src/paimon/common/io/data_output_stream.cpp
+++ b/src/paimon/common/io/data_output_stream.cpp
@@ -25,7 +25,11 @@
namespace paimon {
DataOutputStream::DataOutputStream(const std::shared_ptr<OutputStream>&
output_stream)
- : output_stream_(output_stream) {
+ : owned_output_stream_(output_stream),
output_stream_(owned_output_stream_.get()) {
+ assert(output_stream_);
+}
+
+DataOutputStream::DataOutputStream(OutputStream* output_stream) :
output_stream_(output_stream) {
assert(output_stream_);
}
diff --git a/src/paimon/common/io/data_output_stream.h
b/src/paimon/common/io/data_output_stream.h
index dfb6456b..a1ce5f01 100644
--- a/src/paimon/common/io/data_output_stream.h
+++ b/src/paimon/common/io/data_output_stream.h
@@ -40,6 +40,7 @@ class OutputStream;
class PAIMON_EXPORT DataOutputStream {
public:
explicit DataOutputStream(const std::shared_ptr<OutputStream>&
output_stream);
+ explicit DataOutputStream(OutputStream* output_stream);
template <typename T>
Status WriteValue(const T& value) {
@@ -71,7 +72,8 @@ class PAIMON_EXPORT DataOutputStream {
bool NeedSwap() const;
private:
- std::shared_ptr<OutputStream> output_stream_;
+ std::shared_ptr<OutputStream> owned_output_stream_;
+ OutputStream* output_stream_;
ByteOrder byte_order_ = ByteOrder::PAIMON_BIG_ENDIAN;
};
diff --git a/src/paimon/common/utils/var_length_int_utils.h
b/src/paimon/common/utils/var_length_int_utils.h
index ec76460c..a4c67fb2 100644
--- a/src/paimon/common/utils/var_length_int_utils.h
+++ b/src/paimon/common/utils/var_length_int_utils.h
@@ -22,6 +22,7 @@
#include <cstring>
#include "fmt/format.h"
+#include "paimon/common/utils/math.h"
#include "paimon/macros.h"
#include "paimon/result.h"
namespace paimon {
@@ -134,6 +135,42 @@ class VarLengthIntUtils {
}
return Status::Invalid("Malformed varint64: too many continuation
bytes");
}
+
+ /// Reads a non-negative varint32 from a bounded input.
+ template <typename Input>
+ static Result<int32_t> ReadVarLenInt(Input* input) {
+ uint32_t result = 0;
+ for (int32_t shift = 0; shift < 32; shift += 7) {
+ if (input->Available() == 0) {
+ return Status::Invalid("Truncated varint32 input.");
+ }
+ auto value = static_cast<uint8_t>(input->ReadByte());
+ result |= static_cast<uint32_t>(value & 0x7F) << shift;
+ if ((value & 0x80) == 0) {
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(result,
"varint32"));
+ return static_cast<int32_t>(result);
+ }
+ }
+ return Status::Invalid("Malformed varint32 input.");
+ }
+
+ /// Reads a non-negative varint64 from a bounded input.
+ template <typename Input>
+ static Result<int64_t> ReadVarLenLong(Input* input) {
+ uint64_t result = 0;
+ for (int32_t shift = 0; shift <= 56; shift += 7) {
+ if (input->Available() == 0) {
+ return Status::Invalid("Truncated varint64 input.");
+ }
+ auto value = static_cast<uint8_t>(input->ReadByte());
+ result |= static_cast<uint64_t>(value & 0x7F) << shift;
+ if ((value & 0x80) == 0) {
+ PAIMON_RETURN_NOT_OK(ValidateValueInRange<int64_t>(result,
"varint64"));
+ return static_cast<int64_t>(result);
+ }
+ }
+ return Status::Invalid("Malformed varint64 input.");
+ }
};
} // namespace paimon
diff --git a/src/paimon/common/utils/var_length_int_utils_test.cpp
b/src/paimon/common/utils/var_length_int_utils_test.cpp
index c2404e2f..5c414b26 100644
--- a/src/paimon/common/utils/var_length_int_utils_test.cpp
+++ b/src/paimon/common/utils/var_length_int_utils_test.cpp
@@ -23,6 +23,7 @@
#include <vector>
#include "gtest/gtest.h"
+#include "paimon/common/memory/memory_slice_input.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
@@ -141,4 +142,53 @@ TEST(VarLengthIntUtilsTest, TestEncodeLongBytesNumber) {
ASSERT_EQ(encoded_length, i + 1) << values[i];
}
}
+
+TEST(VarLengthIntUtilsTest, TestReadVarLenInt) {
+ const std::vector<int32_t> values = {0, 127, 128, 16384,
std::numeric_limits<int32_t>::max()};
+ for (int32_t value : values) {
+ char buffer[VarLengthIntUtils::kMaxVarIntSize];
+ ASSERT_OK_AND_ASSIGN(int32_t encoded_length,
VarLengthIntUtils::EncodeInt(value, buffer));
+ MemorySliceInput
input{MemorySlice::Wrap(MemorySegment::WrapView(buffer, encoded_length))};
+ ASSERT_OK_AND_ASSIGN(int32_t actual,
VarLengthIntUtils::ReadVarLenInt(&input));
+ ASSERT_EQ(value, actual);
+ ASSERT_EQ(0, input.Available());
+ }
+
+ const char truncated[] = {static_cast<char>(0x80)};
+ MemorySliceInput truncated_input{MemorySlice::Wrap(
+ MemorySegment::WrapView(truncated,
static_cast<int32_t>(sizeof(truncated))))};
+ ASSERT_NOK(VarLengthIntUtils::ReadVarLenInt(&truncated_input));
+
+ const char malformed[] = {static_cast<char>(0x80), static_cast<char>(0x80),
+ static_cast<char>(0x80), static_cast<char>(0x80),
+ static_cast<char>(0x80)};
+ MemorySliceInput malformed_input{MemorySlice::Wrap(
+ MemorySegment::WrapView(malformed,
static_cast<int32_t>(sizeof(malformed))))};
+ ASSERT_NOK(VarLengthIntUtils::ReadVarLenInt(&malformed_input));
+}
+
+TEST(VarLengthIntUtilsTest, TestReadVarLenLong) {
+ const std::vector<int64_t> values = {0, 127, 128, 16384,
std::numeric_limits<int64_t>::max()};
+ for (int64_t value : values) {
+ char buffer[VarLengthIntUtils::kMaxVarLongSize];
+ ASSERT_OK_AND_ASSIGN(int32_t encoded_length,
VarLengthIntUtils::EncodeLong(value, buffer));
+ MemorySliceInput
input{MemorySlice::Wrap(MemorySegment::WrapView(buffer, encoded_length))};
+ ASSERT_OK_AND_ASSIGN(int64_t actual,
VarLengthIntUtils::ReadVarLenLong(&input));
+ ASSERT_EQ(value, actual);
+ ASSERT_EQ(0, input.Available());
+ }
+
+ const char truncated[] = {static_cast<char>(0x80)};
+ MemorySliceInput truncated_input{MemorySlice::Wrap(
+ MemorySegment::WrapView(truncated,
static_cast<int32_t>(sizeof(truncated))))};
+ ASSERT_NOK(VarLengthIntUtils::ReadVarLenLong(&truncated_input));
+
+ const char malformed[] = {
+ static_cast<char>(0x80), static_cast<char>(0x80),
static_cast<char>(0x80),
+ static_cast<char>(0x80), static_cast<char>(0x80),
static_cast<char>(0x80),
+ static_cast<char>(0x80), static_cast<char>(0x80),
static_cast<char>(0x80)};
+ MemorySliceInput malformed_input{MemorySlice::Wrap(
+ MemorySegment::WrapView(malformed,
static_cast<int32_t>(sizeof(malformed))))};
+ ASSERT_NOK(VarLengthIntUtils::ReadVarLenLong(&malformed_input));
+}
} // namespace paimon::test
