This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new cecd6cd0 feat(deletion-vectors): support bitmap64 deletion vectors
(#423)
cecd6cd0 is described below
commit cecd6cd05a5b72b2c71c5b311e0f8212aba0b2c0
Author: lszskye <[email protected]>
AuthorDate: Tue Oct 13 15:30:15 2026 +0800
feat(deletion-vectors): support bitmap64 deletion vectors (#423)
---
docs/source/user_guide/read.rst | 6 +-
include/paimon/defs.h | 1 -
src/paimon/CMakeLists.txt | 2 +
.../apply_deletion_vector_batch_reader_test.cpp | 17 +-
.../deletionvectors/bitmap64_deletion_vector.cpp | 86 ++++
.../deletionvectors/bitmap64_deletion_vector.h | 56 +++
.../bitmap64_deletion_vector_test.cpp | 286 +++++++++++
.../deletionvectors/bitmap_deletion_vector.cpp | 8 +-
.../core/deletionvectors/bitmap_deletion_vector.h | 7 +
.../bitmap_deletion_vector_test.cpp | 27 +-
.../core/deletionvectors/bucketed_dv_maintainer.h | 10 +-
.../bucketed_dv_maintainer_test.cpp | 41 +-
.../core/deletionvectors/deletion_vector.cpp | 41 +-
src/paimon/core/deletionvectors/deletion_vector.h | 11 +-
.../deletion_vector_index_file_writer_test.cpp | 4 +
.../core/deletionvectors/deletion_vector_test.cpp | 64 ++-
.../deletion_vectors_index_file.cpp | 1 +
.../deletion_vectors_index_file_test.cpp | 14 +
src/paimon/core/io/file_index_evaluator.cpp | 3 +-
src/paimon/core/io/file_index_evaluator_test.cpp | 26 +
src/paimon/core/operation/abstract_split_read.cpp | 18 +-
.../core/operation/data_evolution_split_read.cpp | 13 +-
.../operation/data_evolution_split_read_test.cpp | 23 +
.../core/operation/raw_file_split_read_test.cpp | 42 +-
.../testing/utils/deletion_vector_test_helper.h | 8 +-
test/inte/data_evolution_table_test.cpp | 52 +-
test/inte/pk_compaction_inte_test.cpp | 551 +++++++++++++++++----
27 files changed, 1202 insertions(+), 216 deletions(-)
diff --git a/docs/source/user_guide/read.rst b/docs/source/user_guide/read.rst
index adebaecb..a9477635 100644
--- a/docs/source/user_guide/read.rst
+++ b/docs/source/user_guide/read.rst
@@ -318,6 +318,10 @@ A data-evolution table (``data-evolution.enabled = true``)
may enable
``deletion-vectors.enabled``. Reading such a table is supported: a deleted row
disappears
from the result, including from the columns merged out of the other files that
cover it.
+Both Java-compatible 32-bit and 64-bit deletion vectors can be read, including
mixed
+formats in one index file. ``deletion-vectors.bitmap64 = true`` selects the
format of new
+deletion vectors on supported write paths; existing vectors retain their
format.
+
How the Deletion Vector Is Located
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -340,8 +344,6 @@ come back.
Limitations
~~~~~~~~~~~
-- Only the default 32-bit deletion vectors can be read.
``deletion-vectors.bitmap64`` is not
- supported yet, and a read fails when it actually encounters a 64-bit
deletion vector.
- Paimon C++ does not write deletion vectors for data-evolution tables, so the
deletes
themselves have to be issued by another engine.
- A commit that drops data files from such a table, an overwrite for instance,
is refused.
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index d6abd047..7b473240 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -408,7 +408,6 @@ struct PAIMON_EXPORT Options {
/// "deletion-vectors.bitmap64" - Enable 64 bit bitmap implementation.
Note that only 64 bit
/// bitmap implementation is compatible with Iceberg. Default value is
"false".
- /// @note: bitmap64 dv is not supported.
static const char DELETION_VECTOR_BITMAP64[];
/// "changelog-producer" - Whether to double write to a changelog file.
This changelog file
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index ddd0118e..79305ce4 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -263,6 +263,7 @@ set(PAIMON_CORE_SRCS
core/deletionvectors/deletion_vector.cpp
core/deletionvectors/deletion_file_writer.cpp
core/deletionvectors/bitmap_deletion_vector.cpp
+ core/deletionvectors/bitmap64_deletion_vector.cpp
core/deletionvectors/deletion_vectors_index_file.cpp
core/deletionvectors/deletion_vector_index_file_writer.cpp
core/global_index/global_index_evaluator_impl.cpp
@@ -795,6 +796,7 @@ if(PAIMON_BUILD_TESTS)
core/options/table_type_test.cpp
core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
core/deletionvectors/bitmap_deletion_vector_test.cpp
+ core/deletionvectors/bitmap64_deletion_vector_test.cpp
core/deletionvectors/bucketed_dv_maintainer_test.cpp
core/deletionvectors/deletion_file_writer_test.cpp
core/deletionvectors/deletion_vector_test.cpp
diff --git
a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
index 2e7b29b9..8384f485 100644
---
a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
+++
b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
@@ -28,6 +28,7 @@
#include "paimon/common/reader/prefetch_file_batch_reader_impl.h"
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/common/utils/read_ahead_cache.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/executor.h"
#include "paimon/testing/mock/mock_file_batch_reader.h"
#include "paimon/testing/mock/mock_file_system.h"
@@ -43,8 +44,9 @@ class FileSystem;
} // namespace paimon
namespace paimon::test {
-class ApplyDeletionVectorBatchReaderTest : public ::testing::Test,
- public
::testing::WithParamInterface<bool> {
+class ApplyDeletionVectorBatchReaderTest
+ : public ::testing::Test,
+ public ::testing::WithParamInterface<std::tuple<bool, bool>> {
public:
void SetUp() override {
int_type_ = arrow::int32();
@@ -77,9 +79,14 @@ class ApplyDeletionVectorBatchReaderTest : public
::testing::Test,
int32_t prefetch_batch_count = 3;
for (int32_t batch_size : {1, 2, 4, 10}) {
- auto dv = DeletionVector::FromPrimitiveArray(dv_data, pool.get());
+ std::shared_ptr<DeletionVector> dv =
+ DeletionVector::FromPrimitiveArray(dv_data, pool.get());
+ if (std::get<1>(GetParam())) {
+ dv = Bitmap64DeletionVector::FromBitmapDeletionVector(
+ *static_cast<BitmapDeletionVector*>(dv.get()));
+ }
std::unique_ptr<FileBatchReader> file_batch_reader;
- bool enable_prefetch = GetParam();
+ bool enable_prefetch = std::get<0>(GetParam());
if (enable_prefetch) {
MockFormatReaderBuilder reader_builder(data, target_type_,
batch_size);
ASSERT_OK_AND_ASSIGN(
@@ -182,5 +189,5 @@ TEST(ApplyDeletionVectorBatchReaderWarmupTest,
WarmupForwardsToInnerReader) {
}
INSTANTIATE_TEST_SUITE_P(EnablePrefetch, ApplyDeletionVectorBatchReaderTest,
- ::testing::Values(false, true));
+ ::testing::Combine(::testing::Bool(),
::testing::Bool()));
} // namespace paimon::test
diff --git a/src/paimon/core/deletionvectors/bitmap64_deletion_vector.cpp
b/src/paimon/core/deletionvectors/bitmap64_deletion_vector.cpp
new file mode 100644
index 00000000..61514470
--- /dev/null
+++ b/src/paimon/core/deletionvectors/bitmap64_deletion_vector.cpp
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
+
+#include <cstring>
+#include <limits>
+
+#include "arrow/util/crc32.h"
+#include "paimon/common/io/data_output_stream.h"
+#include "paimon/io/byte_order.h"
+
+namespace paimon {
+
+std::unique_ptr<Bitmap64DeletionVector>
Bitmap64DeletionVector::FromBitmapDeletionVector(
+ const BitmapDeletionVector& dv) {
+ auto result = std::make_unique<Bitmap64DeletionVector>();
+ result->roaring_bitmap_ =
OptimizedRoaringBitmap64::FromRoaringBitmap32(*dv.GetBitmap());
+ return result;
+}
+
+Status Bitmap64DeletionVector::Merge(const std::shared_ptr<DeletionVector>&
deletion_vector) {
+ auto* other = dynamic_cast<Bitmap64DeletionVector*>(deletion_vector.get());
+ if (!other) {
+ return Status::Invalid(
+ "Cannot merge a non-Bitmap64DeletionVector into a
Bitmap64DeletionVector");
+ }
+ roaring_bitmap_ |= other->roaring_bitmap_;
+ return Status::OK();
+}
+
+Result<PAIMON_UNIQUE_PTR<Bytes>> Bitmap64DeletionVector::SerializeToBytes(
+ const std::shared_ptr<MemoryPool>& pool) {
+ roaring_bitmap_.RunLengthEncode();
+ size_t bitmap_size = roaring_bitmap_.GetSizeInBytes();
+ constexpr int32_t overhead = MAGIC_NUMBER_SIZE_BYTES + LENGTH_SIZE_BYTES +
CRC_SIZE_BYTES;
+ if (bitmap_size > static_cast<size_t>(std::numeric_limits<int32_t>::max()
- overhead)) {
+ return Status::Invalid("Cannot serialize deletion vector > 2GB");
+ }
+ auto bitmap = roaring_bitmap_.Serialize(pool.get());
+ auto bytes = Bytes::AllocateBytes(MAGIC_NUMBER_SIZE_BYTES + bitmap_size,
pool.get());
+ int32_t magic = ToLittleEndian(MAGIC_NUMBER);
+ std::memcpy(bytes->data(), &magic, MAGIC_NUMBER_SIZE_BYTES);
+ std::memcpy(bytes->data() + MAGIC_NUMBER_SIZE_BYTES, bitmap->data(),
bitmap_size);
+ return bytes;
+}
+
+Result<int32_t> Bitmap64DeletionVector::SerializeTo(const
std::shared_ptr<MemoryPool>& pool,
+ DataOutputStream* out) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
SerializeToBytes(pool));
+ auto length = static_cast<int32_t>(bytes->size());
+ PAIMON_RETURN_NOT_OK(out->WriteValue<int32_t>(length));
+ PAIMON_RETURN_NOT_OK(out->WriteBytes(bytes));
+ uint32_t crc = arrow::internal::crc32(0, bytes->data(), bytes->size());
+ PAIMON_RETURN_NOT_OK(out->WriteValue<int32_t>(static_cast<int32_t>(crc)));
+ // Unlike DV32, Java DV64 metadata includes the length and CRC fields.
+ return length + LENGTH_SIZE_BYTES + CRC_SIZE_BYTES;
+}
+
+Result<PAIMON_UNIQUE_PTR<DeletionVector>>
Bitmap64DeletionVector::DeserializeWithoutMagicNumber(
+ const char* buffer, int32_t length, MemoryPool* pool) {
+ // Even an empty portable bitmap has an eight-byte bucket count.
+ if (length < static_cast<int32_t>(sizeof(int64_t))) {
+ return Status::Invalid("Invalid bitmap64 payload length");
+ }
+ auto result = pool->AllocateUnique<Bitmap64DeletionVector>();
+ PAIMON_RETURN_NOT_OK(result->roaring_bitmap_.Deserialize(buffer, length));
+ return result;
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/deletionvectors/bitmap64_deletion_vector.h
b/src/paimon/core/deletionvectors/bitmap64_deletion_vector.h
index 7ec827d5..061af459 100644
--- a/src/paimon/core/deletionvectors/bitmap64_deletion_vector.h
+++ b/src/paimon/core/deletionvectors/bitmap64_deletion_vector.h
@@ -18,13 +18,69 @@
*/
#pragma once
+#include "paimon/common/utils/optimized_roaring_bitmap64.h"
+#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/deletionvectors/deletion_vector.h"
namespace paimon {
+/// A `DeletionVector` based on `OptimizedRoaringBitmap64`, it only supports
files with
+/// row count not exceeding `OptimizedRoaringBitmap64::kMaxValue`.
+///
+/// Mostly copied from iceberg.
class Bitmap64DeletionVector : public DeletionVector {
public:
static constexpr int32_t MAGIC_NUMBER = 1681511377;
+ static constexpr int32_t MAGIC_NUMBER_SIZE_BYTES = 4;
+ static constexpr int32_t LENGTH_SIZE_BYTES = 4;
+ static constexpr int32_t CRC_SIZE_BYTES = 4;
+
+ static std::unique_ptr<Bitmap64DeletionVector> FromBitmapDeletionVector(
+ const BitmapDeletionVector& dv);
+
+ Status Delete(int64_t position) override {
+ return roaring_bitmap_.Add(position);
+ }
+
+ Result<bool> CheckedDelete(int64_t position) override {
+ PAIMON_ASSIGN_OR_RAISE(bool deleted, IsDeleted(position));
+ if (deleted) {
+ return false;
+ }
+ PAIMON_RETURN_NOT_OK(Delete(position));
+ return true;
+ }
+
+ Result<bool> IsDeleted(int64_t position) const override {
+ return roaring_bitmap_.Contains(position);
+ }
+
+ bool IsEmpty() const override {
+ return roaring_bitmap_.IsEmpty();
+ }
+
+ Result<int64_t> GetCardinality() const override {
+ return roaring_bitmap_.Cardinality();
+ }
+
+ Status ForEachDeletedPosition(const std::function<void(int64_t)>&
consumer) const override {
+ roaring_bitmap_.ForEach(consumer);
+ return Status::OK();
+ }
+
+ Status Merge(const std::shared_ptr<DeletionVector>& deletion_vector)
override;
+
+ Result<int32_t> SerializeTo(const std::shared_ptr<MemoryPool>& pool,
+ DataOutputStream* out) override;
+
+ Result<PAIMON_UNIQUE_PTR<Bytes>> SerializeToBytes(
+ const std::shared_ptr<MemoryPool>& pool) override;
+
+ static Result<PAIMON_UNIQUE_PTR<DeletionVector>>
DeserializeWithoutMagicNumber(
+ const char* buffer, int32_t length, MemoryPool* pool);
+
+ private:
+ OptimizedRoaringBitmap64 roaring_bitmap_;
};
} // namespace paimon
diff --git a/src/paimon/core/deletionvectors/bitmap64_deletion_vector_test.cpp
b/src/paimon/core/deletionvectors/bitmap64_deletion_vector_test.cpp
new file mode 100644
index 00000000..57c20912
--- /dev/null
+++ b/src/paimon/core/deletionvectors/bitmap64_deletion_vector_test.cpp
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
+
+#include <set>
+
+#include "gtest/gtest.h"
+#include "paimon/common/io/byte_array_output_stream.h"
+#include "paimon/common/io/data_output_stream.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/fs/file_system_factory.h"
+#include "paimon/io/byte_array_input_stream.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(Bitmap64DeletionVectorTest, TestBasicOperations) {
+ Bitmap64DeletionVector dv;
+ ASSERT_TRUE(dv.IsEmpty());
+ std::vector<int64_t> expected;
+ for (const auto& base : {0LL, 1LL << 31, 1LL << 32, 2LL << 32}) {
+ for (int64_t i = 0; i < 2000; i += 2) {
+ ASSERT_OK(dv.Delete(base + i));
+ expected.push_back(base + i);
+ }
+ for (int64_t i = 0; i < 2000; ++i) {
+ ASSERT_EQ(dv.IsDeleted(base + i).value(), i % 2 == 0);
+ }
+ }
+ ASSERT_FALSE(dv.IsEmpty());
+ ASSERT_EQ(dv.GetCardinality().value(), expected.size());
+ std::vector<int64_t> actual;
+ ASSERT_OK(dv.ForEachDeletedPosition([&](int64_t p) { actual.push_back(p);
}));
+ ASSERT_EQ(actual, expected);
+}
+
+TEST(Bitmap64DeletionVectorTest, TestCheckedDelete) {
+ Bitmap64DeletionVector dv;
+ std::vector<int64_t> positions = {
+ 0, 42, (1LL << 31) - 1, 1LL << 31, (1LL << 32) - 1, 1LL << 32, (2LL <<
32) + 42};
+ for (const auto& position : positions) {
+ ASSERT_TRUE(dv.CheckedDelete(position).value());
+ ASSERT_FALSE(dv.CheckedDelete(position).value());
+ ASSERT_TRUE(dv.IsDeleted(position).value());
+ }
+ ASSERT_EQ(dv.GetCardinality().value(), positions.size());
+}
+
+TEST(Bitmap64DeletionVectorTest, TestGetCardinality) {
+ for (const auto& step : {1, 10}) {
+ Bitmap64DeletionVector dv;
+ ASSERT_EQ(dv.GetCardinality().value(), 0);
+ for (const auto& base : {0LL, 1LL << 32}) {
+ for (int64_t i = 0; i < 100; ++i) {
+ ASSERT_OK(dv.Delete(base + i * step));
+ }
+ }
+ ASSERT_EQ(dv.GetCardinality().value(), 200);
+ ASSERT_OK(dv.Delete(0));
+ ASSERT_EQ(dv.GetCardinality().value(), 200);
+ ASSERT_OK(dv.Delete((2LL << 32) + 1));
+ ASSERT_EQ(dv.GetCardinality().value(), 201);
+ }
+}
+
+TEST(Bitmap64DeletionVectorTest, TestFromBitmapDeletionVector) {
+ for (const auto& positions :
+ {std::vector<int32_t>{}, std::vector<int32_t>{0, 1, 3,
RoaringBitmap32::MAX_VALUE}}) {
+ RoaringBitmap32 bitmap;
+ for (const auto& position : positions) {
+ bitmap.Add(position);
+ }
+ BitmapDeletionVector dv32(bitmap);
+ auto dv64 = Bitmap64DeletionVector::FromBitmapDeletionVector(dv32);
+ ASSERT_EQ(dv64->IsEmpty(), positions.empty());
+ ASSERT_EQ(dv64->GetCardinality().value(), positions.size());
+ std::vector<int64_t> actual;
+ ASSERT_OK(dv64->ForEachDeletedPosition([&](int64_t p) {
actual.push_back(p); }));
+ ASSERT_EQ(actual, std::vector<int64_t>(positions.begin(),
positions.end()));
+ ASSERT_OK(dv64->Delete(7));
+ ASSERT_OK(dv64->Delete((1LL << 32) + 7));
+ ASSERT_FALSE(dv32.IsDeleted(7).value());
+ ASSERT_EQ(dv32.GetCardinality().value(), positions.size());
+ ASSERT_OK(dv32.Delete(8));
+ ASSERT_FALSE(dv64->IsDeleted(8).value());
+ }
+}
+
+TEST(Bitmap64DeletionVectorTest, TestPositionOutOfRangeShouldFail) {
+ Bitmap64DeletionVector dv;
+ for (const auto& position : std::vector<int64_t>{-1,
OptimizedRoaringBitmap64::kMaxValue + 1}) {
+ ASSERT_NOK_WITH_MSG(dv.Delete(position),
+ "OptimizedRoaringBitmap64 supports positions that
are >= 0 and <=");
+ ASSERT_NOK_WITH_MSG(dv.CheckedDelete(position),
+ "OptimizedRoaringBitmap64 supports positions that
are >= 0 and <=");
+ ASSERT_NOK_WITH_MSG(dv.IsDeleted(position),
+ "OptimizedRoaringBitmap64 supports positions that
are >= 0 and <=");
+ }
+ ASSERT_TRUE(dv.IsEmpty());
+ ASSERT_EQ(dv.GetCardinality().value(), 0);
+ ASSERT_FALSE(dv.IsDeleted(OptimizedRoaringBitmap64::kMaxValue).value());
+}
+
+TEST(Bitmap64DeletionVectorTest, TestMerge) {
+ std::vector<int64_t> left = {1, 3, (1LL << 32) + 5};
+ std::vector<int64_t> right = {2, (1LL << 32) + 5, (2LL << 32) + 4};
+ for (const auto& left_empty : {false, true}) {
+ for (const auto& right_empty : {false, true}) {
+ auto dv = std::make_shared<Bitmap64DeletionVector>();
+ auto other = std::make_shared<Bitmap64DeletionVector>();
+ std::set<int64_t> expected;
+ if (!left_empty) {
+ for (const auto& p : left) {
+ ASSERT_OK(dv->Delete(p));
+ expected.insert(p);
+ }
+ }
+ if (!right_empty) {
+ for (const auto& p : right) {
+ ASSERT_OK(other->Delete(p));
+ expected.insert(p);
+ }
+ }
+ ASSERT_OK(dv->Merge(other));
+ ASSERT_NOK_WITH_MSG(dv->Merge(nullptr), "Cannot merge a
non-Bitmap64DeletionVector");
+ ASSERT_OK(dv->Merge(other));
+ for (const auto& positions : {std::vector<int32_t>{},
std::vector<int32_t>{7}}) {
+ RoaringBitmap32 bitmap;
+ for (const auto& position : positions) {
+ bitmap.Add(position);
+ }
+
ASSERT_NOK_WITH_MSG(dv->Merge(std::make_shared<BitmapDeletionVector>(bitmap)),
+ "Cannot merge a
non-Bitmap64DeletionVector");
+ }
+ ASSERT_EQ(dv->IsEmpty(), expected.empty());
+ ASSERT_EQ(dv->GetCardinality().value(), expected.size());
+ std::vector<int64_t> actual;
+ ASSERT_OK(dv->ForEachDeletedPosition([&](int64_t p) {
actual.push_back(p); }));
+ ASSERT_EQ(actual, std::vector<int64_t>(expected.begin(),
expected.end()));
+ for (const auto& p : expected) {
+ ASSERT_TRUE(dv->IsDeleted(p).value());
+ }
+ }
+ }
+}
+
+TEST(Bitmap64DeletionVectorTest, JavaCompatibleFramedRecord) {
+ // Generated with Java Paimon Bitmap64DeletionVector.serializeTo(),
including CRC.
+ std::vector<uint8_t> java_bytes = {
+ 0, 0, 0, 110, 209, 211, 57, 100, 3, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0,
+ 58, 48, 0, 0, 4, 0, 0, 0, 0, 0, 1, 0, 255, 127, 0, 0,
0, 128, 0, 0,
+ 255, 255, 0, 0, 40, 0, 0, 0, 44, 0, 0, 0, 46, 0, 0, 0,
48, 0, 0, 0,
+ 0, 0, 1, 0, 255, 255, 0, 0, 255, 255, 1, 0, 0, 0, 58,
48, 0, 0, 1, 0,
+ 0, 0, 0, 0, 0, 0, 16, 0, 0, 0, 0, 0, 2, 0, 0, 0,
58, 48, 0, 0,
+ 1, 0, 0, 0, 0, 0, 0, 0, 16, 0, 0, 0, 7, 0, 206,
177, 171, 208};
+ std::vector<int64_t> positions = {
+ 0, 1, 2147483647LL, 2147483648LL, 4294967295LL, 4294967296LL,
8589934599LL};
+ auto pool = GetDefaultPool();
+ for (const auto& with_length : {false, true}) {
+ auto input = std::make_shared<ByteArrayInputStream>(
+ reinterpret_cast<const char*>(java_bytes.data()),
java_bytes.size());
+ DataInputStream in(input);
+ ASSERT_OK_AND_ASSIGN(
+ auto decoded,
+ DeletionVector::Read(
+ &in, with_length ?
std::make_optional<int64_t>(java_bytes.size()) : std::nullopt,
+ pool.get()));
+ ASSERT_TRUE(dynamic_cast<Bitmap64DeletionVector*>(decoded.get()));
+ std::vector<int64_t> actual;
+ ASSERT_OK(decoded->ForEachDeletedPosition([&](int64_t p) {
actual.push_back(p); }));
+ ASSERT_EQ(actual, positions);
+ ASSERT_EQ(input->GetPos().value(), java_bytes.size());
+ }
+ Bitmap64DeletionVector dv;
+ for (const auto& p : positions) {
+ ASSERT_OK(dv.Delete(p));
+ }
+ auto output = std::make_shared<ByteArrayOutputStream>(
+ std::make_unique<MemorySegmentOutputStream>(1024, pool));
+ DataOutputStream out(output);
+ ASSERT_OK_AND_ASSIGN(int32_t metadata_length, dv.SerializeTo(pool, &out));
+ ASSERT_OK_AND_ASSIGN(auto bytes, output->Finish(pool.get()));
+ ASSERT_EQ(metadata_length, java_bytes.size());
+ ASSERT_EQ(std::vector<uint8_t>(bytes->data(), bytes->data() +
bytes->size()), java_bytes);
+}
+
+TEST(Bitmap64DeletionVectorTest, SerializeAndDeserialize) {
+ Bitmap64DeletionVector dv;
+ std::vector<int64_t> positions = {1, 7, 1024, (1LL << 32) + 7, (2LL << 32)
+ 1};
+ for (const auto& position : positions) {
+ ASSERT_OK(dv.Delete(position));
+ }
+ auto pool = GetDefaultPool();
+ ASSERT_OK_AND_ASSIGN(auto bytes, dv.SerializeToBytes(pool));
+ ASSERT_OK_AND_ASSIGN(
+ auto decoded,
+ Bitmap64DeletionVector::DeserializeWithoutMagicNumber(
+ bytes->data() + Bitmap64DeletionVector::MAGIC_NUMBER_SIZE_BYTES,
+ bytes->size() - Bitmap64DeletionVector::MAGIC_NUMBER_SIZE_BYTES,
pool.get()));
+ ASSERT_EQ(decoded->GetCardinality().value(), positions.size());
+ for (const auto& position : positions) {
+ ASSERT_TRUE(decoded->IsDeleted(position).value());
+ }
+ ASSERT_FALSE(decoded->IsDeleted(8).value());
+ ASSERT_FALSE(decoded->IsDeleted((1LL << 32) + 8).value());
+}
+
+TEST(Bitmap64DeletionVectorTest, SerializeToOutputStream) {
+ Bitmap64DeletionVector dv;
+ for (int64_t p = 0; p < 50; p += 2) {
+ ASSERT_OK(dv.Delete((1LL << 32) + p));
+ }
+ auto pool = GetDefaultPool();
+ auto dir = UniqueTestDirectory::Create();
+ auto path = PathUtil::JoinPath(dir->Str(), "dv64");
+ ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", path, {}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> output,
fs->Create(path, false));
+ DataOutputStream out(output);
+ ASSERT_OK_AND_ASSIGN(int32_t length, dv.SerializeTo(pool, &out));
+ ASSERT_OK(output->Flush());
+ ASSERT_OK(output->Close());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input, fs->Open(path));
+ DataInputStream in(input);
+ ASSERT_OK_AND_ASSIGN(int32_t bitmap_length, in.ReadValue<int32_t>());
+ ASSERT_EQ(length, bitmap_length +
Bitmap64DeletionVector::LENGTH_SIZE_BYTES +
+ Bitmap64DeletionVector::CRC_SIZE_BYTES);
+ ASSERT_EQ(in.Length().value(), length);
+ ASSERT_OK(in.Seek(0));
+ ASSERT_OK_AND_ASSIGN(auto decoded, DeletionVector::Read(&in, length,
pool.get()));
+ ASSERT_EQ(in.GetPos().value(), length);
+ ASSERT_EQ(decoded->GetCardinality().value(), 25);
+ for (int64_t p = 0; p <= 50; ++p) {
+ ASSERT_EQ(decoded->IsDeleted((1LL << 32) + p).value(), p < 50 && p % 2
== 0);
+ }
+}
+
+TEST(Bitmap64DeletionVectorTest, RejectMalformedRecords) {
+ auto pool = GetDefaultPool();
+ Bitmap64DeletionVector dv;
+ ASSERT_OK(dv.Delete(1LL << 32));
+ auto output = std::make_shared<ByteArrayOutputStream>(
+ std::make_unique<MemorySegmentOutputStream>(1024, pool));
+ DataOutputStream out(output);
+ ASSERT_OK_AND_ASSIGN(int32_t length, dv.SerializeTo(pool, &out));
+ ASSERT_OK_AND_ASSIGN(auto bytes, output->Finish(pool.get()));
+ auto input = std::make_shared<ByteArrayInputStream>(bytes->data(),
bytes->size());
+ DataInputStream in(input);
+ ASSERT_NOK_WITH_MSG(DeletionVector::Read(&in, length - 8, pool.get()),
"Size not match");
+ for (const auto& cut : {0, 1, 3, 4, 7, length - 5, length - 1}) {
+ auto truncated = std::make_shared<ByteArrayInputStream>(bytes->data(),
cut);
+ DataInputStream truncated_in(truncated);
+ ASSERT_NOK_WITH_MSG(DeletionVector::Read(&truncated_in, std::nullopt,
pool.get()),
+ "DataInputStream boundary check failed");
+ }
+ // Keep the frame intact and corrupt only its magic number.
+ bytes->data()[Bitmap64DeletionVector::LENGTH_SIZE_BYTES] ^= 1;
+ auto invalid_magic_input =
std::make_shared<ByteArrayInputStream>(bytes->data(), bytes->size());
+ DataInputStream invalid_magic_in(invalid_magic_input);
+ ASSERT_NOK_WITH_MSG(DeletionVector::Read(&invalid_magic_in, length,
pool.get()),
+ "Invalid magic number");
+ // A complete frame containing only magic and CRC has no bitmap payload.
+ char magic_only[] = {0, 0, 0, 4, static_cast<char>(0xD1),
static_cast<char>(0xD3), 0x39, 0x64,
+ 0, 0, 0, 0};
+ auto magic_only_input = std::make_shared<ByteArrayInputStream>(magic_only,
sizeof(magic_only));
+ DataInputStream magic_only_in(magic_only_input);
+ ASSERT_NOK_WITH_MSG(DeletionVector::Read(&magic_only_in,
sizeof(magic_only), pool.get()),
+ "Invalid bitmap64 payload length");
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/deletionvectors/bitmap_deletion_vector.cpp
b/src/paimon/core/deletionvectors/bitmap_deletion_vector.cpp
index ed8c3ba2..f789d86b 100644
--- a/src/paimon/core/deletionvectors/bitmap_deletion_vector.cpp
+++ b/src/paimon/core/deletionvectors/bitmap_deletion_vector.cpp
@@ -87,16 +87,12 @@ Result<PAIMON_UNIQUE_PTR<DeletionVector>>
BitmapDeletionVector::Deserialize(cons
}
Status BitmapDeletionVector::Merge(const std::shared_ptr<DeletionVector>&
deletion_vector) {
- if (!deletion_vector || deletion_vector->IsEmpty()) {
- return Status::OK();
- }
auto* other = dynamic_cast<BitmapDeletionVector*>(deletion_vector.get());
- if (other != nullptr) {
- roaring_bitmap_ |= other->roaring_bitmap_;
- } else {
+ if (!other) {
return Status::Invalid(
"Cannot merge a non-BitmapDeletionVector into a
BitmapDeletionVector");
}
+ roaring_bitmap_ |= other->roaring_bitmap_;
return Status::OK();
}
diff --git a/src/paimon/core/deletionvectors/bitmap_deletion_vector.h
b/src/paimon/core/deletionvectors/bitmap_deletion_vector.h
index d0803b71..7890e97d 100644
--- a/src/paimon/core/deletionvectors/bitmap_deletion_vector.h
+++ b/src/paimon/core/deletionvectors/bitmap_deletion_vector.h
@@ -68,6 +68,13 @@ class BitmapDeletionVector : public DeletionVector {
Status Merge(const std::shared_ptr<DeletionVector>& deletion_vector)
override;
+ Status ForEachDeletedPosition(const std::function<void(int64_t)>&
consumer) const override {
+ for (auto iter = roaring_bitmap_.Begin(); iter !=
roaring_bitmap_.End(); ++iter) {
+ consumer(*iter);
+ }
+ return Status::OK();
+ }
+
const RoaringBitmap32* GetBitmap() const {
return &roaring_bitmap_;
}
diff --git a/src/paimon/core/deletionvectors/bitmap_deletion_vector_test.cpp
b/src/paimon/core/deletionvectors/bitmap_deletion_vector_test.cpp
index 1aeba310..2ad00090 100644
--- a/src/paimon/core/deletionvectors/bitmap_deletion_vector_test.cpp
+++ b/src/paimon/core/deletionvectors/bitmap_deletion_vector_test.cpp
@@ -24,6 +24,7 @@
#include "gtest/gtest.h"
#include "paimon/common/io/memory_segment_output_stream.h"
#include "paimon/common/utils/path_util.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/fs/file_system_factory.h"
#include "paimon/testing/utils/testharness.h"
@@ -208,16 +209,38 @@ TEST(BitmapDeletionVectorTest, MergeEmptyDeletionVector) {
ASSERT_TRUE(dv1->IsDeleted(20).value());
}
-TEST(BitmapDeletionVectorTest, MergeNullDeletionVector) {
+TEST(BitmapDeletionVectorTest, MergeNullDeletionVectorShouldFail) {
RoaringBitmap32 roaring1;
roaring1.Add(7);
auto dv1 = std::make_shared<BitmapDeletionVector>(roaring1);
- ASSERT_OK(dv1->Merge(nullptr));
+ ASSERT_NOK_WITH_MSG(dv1->Merge(nullptr), "Cannot merge a
non-BitmapDeletionVector");
ASSERT_EQ(dv1->GetCardinality().value(), 1);
ASSERT_TRUE(dv1->IsDeleted(7).value());
}
+TEST(BitmapDeletionVectorTest, MergeDifferentTypeShouldFail) {
+ for (const auto& target_empty : {false, true}) {
+ for (const auto& source_empty : {false, true}) {
+ BitmapDeletionVector dv{RoaringBitmap32()};
+ if (!target_empty) {
+ ASSERT_OK(dv.Delete(7));
+ }
+ auto other = std::make_shared<Bitmap64DeletionVector>();
+ if (!source_empty) {
+ ASSERT_OK(other->Delete(8));
+ ASSERT_OK(other->Delete(1LL << 32));
+ }
+ ASSERT_NOK_WITH_MSG(dv.Merge(other), "Cannot merge a
non-BitmapDeletionVector");
+ ASSERT_EQ(dv.IsEmpty(), target_empty);
+ ASSERT_EQ(dv.GetCardinality().value(), target_empty ? 0 : 1);
+ ASSERT_EQ(dv.IsDeleted(7).value(), !target_empty);
+ ASSERT_FALSE(dv.IsDeleted(8).value());
+ ASSERT_EQ(other->GetCardinality().value(), source_empty ? 0 : 2);
+ }
+ }
+}
+
TEST(BitmapDeletionVectorTest, MergeIntoEmptyDeletionVector) {
RoaringBitmap32 empty_roaring;
auto dv1 = std::make_shared<BitmapDeletionVector>(empty_roaring);
diff --git a/src/paimon/core/deletionvectors/bucketed_dv_maintainer.h
b/src/paimon/core/deletionvectors/bucketed_dv_maintainer.h
index 4088c342..39cda8bb 100644
--- a/src/paimon/core/deletionvectors/bucketed_dv_maintainer.h
+++ b/src/paimon/core/deletionvectors/bucketed_dv_maintainer.h
@@ -23,6 +23,7 @@
#include <optional>
#include <string>
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/deletionvectors/deletion_vector.h"
#include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
@@ -45,12 +46,11 @@ class BucketedDvMaintainer {
std::shared_ptr<DeletionVector> dv;
if (auto it = deletion_vectors_.find(file_name); it ==
deletion_vectors_.end()) {
if (bitmap64_) {
- return Status::NotImplemented("not support bitmap 64 deletion
vectors");
+ dv = std::make_shared<Bitmap64DeletionVector>();
+ } else {
+ dv = std::make_shared<BitmapDeletionVector>(RoaringBitmap32());
}
- RoaringBitmap32 roaring_bitmap;
- auto inserted =
std::make_shared<BitmapDeletionVector>(roaring_bitmap);
- deletion_vectors_[file_name] = inserted;
- dv = std::move(inserted);
+ deletion_vectors_[file_name] = dv;
} else {
dv = it->second;
}
diff --git a/src/paimon/core/deletionvectors/bucketed_dv_maintainer_test.cpp
b/src/paimon/core/deletionvectors/bucketed_dv_maintainer_test.cpp
index f4ef95ae..9a0e80b5 100644
--- a/src/paimon/core/deletionvectors/bucketed_dv_maintainer_test.cpp
+++ b/src/paimon/core/deletionvectors/bucketed_dv_maintainer_test.cpp
@@ -137,20 +137,33 @@ TEST(BucketedDvMaintainerTest,
TestNotifyNewDeletionOnExistingVectorOnlyMarksWhe
ASSERT_TRUE(modified_write.has_value());
}
-TEST(BucketedDvMaintainerTest,
TestNotifyNewDeletionReturnsNotImplementedForBitmap64) {
- auto dir = UniqueTestDirectory::Create();
- ASSERT_TRUE(dir);
- auto index_file = CreateDvIndexFile(dir->Str(), /*bitmap64=*/true);
- BucketedDvMaintainer maintainer(index_file, /*deletion_vectors=*/{});
-
- Status status = maintainer.NotifyNewDeletion("file-new", /*position=*/1);
- ASSERT_TRUE(status.IsNotImplemented());
-
- auto lookup = maintainer.DeletionVectorOf("file-new");
- ASSERT_FALSE(lookup.has_value());
-
- ASSERT_OK_AND_ASSIGN(auto write_result,
maintainer.WriteDeletionVectorsIndex());
- ASSERT_FALSE(write_result.has_value());
+TEST(BucketedDvMaintainerTest, MixedFormatsAfterConfigurationChanges) {
+ for (bool initial_bitmap64 : {false, true}) {
+ auto dir = UniqueTestDirectory::Create();
+ auto index = CreateDvIndexFile(dir->Str(), initial_bitmap64);
+ BucketedDvMaintainer initial(index, {});
+ ASSERT_OK(initial.NotifyNewDeletion("old", 1));
+ ASSERT_OK_AND_ASSIGN(auto first, initial.WriteDeletionVectorsIndex());
+ ASSERT_TRUE(first.has_value());
+ auto changed_index = CreateDvIndexFile(dir->Str(), !initial_bitmap64);
+ ASSERT_OK_AND_ASSIGN(auto restored,
changed_index->ReadAllDeletionVectors(first.value()));
+ BucketedDvMaintainer changed(changed_index, restored);
+ ASSERT_OK(changed.NotifyNewDeletion("old", 7));
+ ASSERT_OK(changed.NotifyNewDeletion("new", 3));
+ const std::string high_file = initial_bitmap64 ? "old" : "new";
+ ASSERT_OK(changed.NotifyNewDeletion(high_file, (1LL << 32) + 9));
+ ASSERT_OK_AND_ASSIGN(auto second, changed.WriteDeletionVectorsIndex());
+ ASSERT_TRUE(second.has_value());
+ ASSERT_OK_AND_ASSIGN(auto mixed,
changed_index->ReadAllDeletionVectors(second.value()));
+ ASSERT_EQ(dynamic_cast<Bitmap64DeletionVector*>(mixed.at("old").get())
!= nullptr,
+ initial_bitmap64);
+ ASSERT_EQ(dynamic_cast<Bitmap64DeletionVector*>(mixed.at("new").get())
!= nullptr,
+ !initial_bitmap64);
+ ASSERT_TRUE(mixed.at("old")->IsDeleted(7).value());
+ ASSERT_TRUE(mixed.at(high_file)->IsDeleted((1LL << 32) + 9).value());
+ ASSERT_OK(changed.NotifyNewDeletion(high_file, (1LL << 32) + 9));
+ ASSERT_FALSE(changed.WriteDeletionVectorsIndex().value().has_value());
+ }
}
} // namespace paimon::test
diff --git a/src/paimon/core/deletionvectors/deletion_vector.cpp
b/src/paimon/core/deletionvectors/deletion_vector.cpp
index f981d19b..2241caec 100644
--- a/src/paimon/core/deletionvectors/deletion_vector.cpp
+++ b/src/paimon/core/deletionvectors/deletion_vector.cpp
@@ -20,6 +20,7 @@
#include <cassert>
#include <cstddef>
+#include <limits>
#include <string>
#include "fmt/format.h"
@@ -78,11 +79,6 @@ std::unordered_map<std::string, DeletionFile>
DeletionVector::CreateDeletionFile
return deletion_file_map;
}
-Result<PAIMON_UNIQUE_PTR<DeletionVector>>
DeletionVector::DeserializeFromBytes(const Bytes* bytes,
-
MemoryPool* pool) {
- return BitmapDeletionVector::Deserialize(bytes->data(), bytes->size(),
pool);
-}
-
Result<PAIMON_UNIQUE_PTR<DeletionVector>> DeletionVector::Read(const
FileSystem* file_system,
const
DeletionFile& deletion_file,
MemoryPool*
pool) {
@@ -90,15 +86,7 @@ Result<PAIMON_UNIQUE_PTR<DeletionVector>>
DeletionVector::Read(const FileSystem*
file_system->Open(deletion_file.path));
DataInputStream file_input_stream(input);
PAIMON_RETURN_NOT_OK(file_input_stream.Seek(deletion_file.offset));
- PAIMON_ASSIGN_OR_RAISE(int32_t actual_length,
file_input_stream.ReadValue<int32_t>());
- if (actual_length != deletion_file.length) {
- return Status::Invalid(
- fmt::format("Size not match, actual size: {}, expect size: {},
file path: {}",
- actual_length, deletion_file.length,
deletion_file.path));
- }
- auto bytes = Bytes::AllocateBytes(deletion_file.length, pool);
- PAIMON_RETURN_NOT_OK(file_input_stream.ReadBytes(bytes.get()));
- return DeserializeFromBytes(bytes.get(), pool);
+ return Read(&file_input_stream, deletion_file.length, pool);
}
Result<PAIMON_UNIQUE_PTR<DeletionVector>>
DeletionVector::Read(DataInputStream* input_stream,
@@ -127,10 +115,27 @@ Result<PAIMON_UNIQUE_PTR<DeletionVector>>
DeletionVector::Read(DataInputStream*
return
BitmapDeletionVector::DeserializeWithoutMagicNumber(bytes->data(),
bytes->size(),
pool);
} else if (EndianSwapValue(magic_number) ==
Bitmap64DeletionVector::MAGIC_NUMBER) {
- return Status::NotImplemented(
- "bitmap64 deletion vectors are not supported in this version, "
- "please use bitmap deletion vectors instead or upgrade to a
version "
- "that supports bitmap64.");
+ if (bitmap_length < Bitmap64DeletionVector::MAGIC_NUMBER_SIZE_BYTES ||
+ bitmap_length > std::numeric_limits<int32_t>::max() -
+ Bitmap64DeletionVector::LENGTH_SIZE_BYTES -
+ Bitmap64DeletionVector::CRC_SIZE_BYTES) {
+ return Status::Invalid(fmt::format("Invalid bitmap length: {}",
bitmap_length));
+ }
+ int64_t record_length = static_cast<int64_t>(bitmap_length) +
+ Bitmap64DeletionVector::LENGTH_SIZE_BYTES +
+ Bitmap64DeletionVector::CRC_SIZE_BYTES;
+ if (length.has_value() && record_length != length.value()) {
+ return Status::Invalid(fmt::format("Size not match, actual size:
{}, expected size: {}",
+ record_length, length.value()));
+ }
+ auto bytes = Bytes::AllocateBytes(
+ bitmap_length - Bitmap64DeletionVector::MAGIC_NUMBER_SIZE_BYTES,
pool);
+ PAIMON_RETURN_NOT_OK(input_stream->ReadBytes(bytes.get()));
+ // skip crc (4 bytes)
+ PAIMON_ASSIGN_OR_RAISE([[maybe_unused]] int32_t unused_crc,
+ input_stream->ReadValue<int32_t>());
+ return
Bitmap64DeletionVector::DeserializeWithoutMagicNumber(bytes->data(),
bytes->size(),
+ pool);
}
return Status::Invalid(fmt::format(
diff --git a/src/paimon/core/deletionvectors/deletion_vector.h
b/src/paimon/core/deletionvectors/deletion_vector.h
index f4f876ea..9cb8a6e6 100644
--- a/src/paimon/core/deletionvectors/deletion_vector.h
+++ b/src/paimon/core/deletionvectors/deletion_vector.h
@@ -18,6 +18,7 @@
*/
#pragma once
#include <cstdint>
+#include <functional>
#include <memory>
#include <unordered_map>
#include <utility>
@@ -106,6 +107,9 @@ class DeletionVector {
/// @return the number of distinct integers added to the DeletionVector.
virtual Result<int64_t> GetCardinality() const = 0;
+ /// Iterates over all deleted positions in this deletion vector.
+ virtual Status ForEachDeletedPosition(const std::function<void(int64_t)>&
consumer) const = 0;
+
/// Serializes the deletion vector.
virtual Result<int32_t> SerializeTo(const std::shared_ptr<MemoryPool>&
pool,
DataOutputStream* out) = 0;
@@ -116,13 +120,6 @@ class DeletionVector {
virtual Result<PAIMON_UNIQUE_PTR<Bytes>> SerializeToBytes(
const std::shared_ptr<MemoryPool>& pool) = 0;
- /// Deserializes a deletion vector from a byte array.
- ///
- /// @param bytes The byte array containing the serialized deletion vector.
- /// @return A DeletionVector instance that represents the deserialized
data.
- static Result<PAIMON_UNIQUE_PTR<DeletionVector>>
DeserializeFromBytes(const Bytes* bytes,
-
MemoryPool* pool);
-
static Result<PAIMON_UNIQUE_PTR<DeletionVector>> Read(const FileSystem*
file_system,
const DeletionFile&
deletion_file,
MemoryPool* pool);
diff --git
a/src/paimon/core/deletionvectors/deletion_vector_index_file_writer_test.cpp
b/src/paimon/core/deletionvectors/deletion_vector_index_file_writer_test.cpp
index a435f44e..7f18168e 100644
--- a/src/paimon/core/deletionvectors/deletion_vector_index_file_writer_test.cpp
+++ b/src/paimon/core/deletionvectors/deletion_vector_index_file_writer_test.cpp
@@ -47,6 +47,10 @@ class FailingDeletionVector : public DeletionVector {
return Status::Invalid("injected is-deleted failure");
}
+ Status ForEachDeletedPosition(const std::function<void(int64_t)>&) const
override {
+ return Status::OK();
+ }
+
bool IsEmpty() const override {
return true;
}
diff --git a/src/paimon/core/deletionvectors/deletion_vector_test.cpp
b/src/paimon/core/deletionvectors/deletion_vector_test.cpp
index bff4b46a..7aa0a756 100644
--- a/src/paimon/core/deletionvectors/deletion_vector_test.cpp
+++ b/src/paimon/core/deletionvectors/deletion_vector_test.cpp
@@ -27,12 +27,13 @@
#include <vector>
#include "gtest/gtest.h"
+#include "paimon/common/io/byte_array_output_stream.h"
+#include "paimon/common/io/data_output_stream.h"
#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/table/source/deletion_file.h"
#include "paimon/io/byte_array_input_stream.h"
-#include "paimon/io/byte_order.h"
#include "paimon/io/data_input_stream.h"
#include "paimon/testing/utils/testharness.h"
@@ -82,9 +83,15 @@ TEST(DeletionVectorTest, TestSimple) {
}
}
auto pool = GetDefaultPool();
- ASSERT_OK_AND_ASSIGN(auto bytes, deletion_vector->SerializeToBytes(pool));
- ASSERT_OK_AND_ASSIGN(auto de_deletion_vector,
- DeletionVector::DeserializeFromBytes(bytes.get(),
pool.get()));
+ auto output = std::make_shared<ByteArrayOutputStream>(
+ std::make_unique<MemorySegmentOutputStream>(1024, pool));
+ DataOutputStream out(output);
+ ASSERT_OK_AND_ASSIGN(int32_t length, deletion_vector->SerializeTo(pool,
&out));
+ ASSERT_OK_AND_ASSIGN(auto bytes, output->Finish(pool.get()));
+ auto input = std::make_shared<ByteArrayInputStream>(bytes->data(),
bytes->size());
+ DataInputStream in(input);
+ ASSERT_OK_AND_ASSIGN(auto de_deletion_vector, DeletionVector::Read(&in,
length, pool.get()));
+ ASSERT_EQ(input->GetPos().value(), bytes->size());
ASSERT_FALSE(deletion_vector->IsEmpty());
ASSERT_FALSE(de_deletion_vector->IsEmpty());
@@ -106,9 +113,16 @@ TEST(DeletionVectorTest, TestCompatibleWithJava) {
auto serialize_bytes = std::make_shared<Bytes>(data.size(), pool.get());
memcpy(serialize_bytes->data(), data.data(), data.size());
- // test deserialize
- ASSERT_OK_AND_ASSIGN(auto deletion_vector,
-
DeletionVector::DeserializeFromBytes(serialize_bytes.get(), pool.get()));
+ // Wrap the Java payload in a record with its length and CRC.
+ std::vector<uint8_t> record;
+ AppendInt32BigEndian(&record, data.size());
+ record.insert(record.end(), data.begin(), data.end());
+ AppendInt32BigEndian(&record, 0x5235C7CF);
+ auto input = std::make_shared<ByteArrayInputStream>(
+ reinterpret_cast<const char*>(record.data()), record.size());
+ DataInputStream in(input);
+ ASSERT_OK_AND_ASSIGN(auto deletion_vector, DeletionVector::Read(&in,
data.size(), pool.get()));
+ ASSERT_EQ(input->GetPos().value(), record.size());
std::vector<bool> expected = {false, true, true, false, true, false};
std::vector<bool> result;
for (size_t i = 0; i < expected.size(); i++) {
@@ -147,20 +161,30 @@ TEST(DeletionVectorTest,
ReadFromDataInputStreamInvalidBitmapLength) {
"Invalid bitmap length");
}
-TEST(DeletionVectorTest, ReadFromDataInputStreamBitmap64NotImplemented) {
- std::vector<uint8_t> data;
- AppendInt32BigEndian(&data, /*value=*/8);
- // Trigger: EndianSwapValue(magic_number) ==
Bitmap64DeletionVector::MAGIC_NUMBER.
- AppendInt32BigEndian(&data,
EndianSwapValue(Bitmap64DeletionVector::MAGIC_NUMBER));
-
- auto input_stream = std::make_shared<ByteArrayInputStream>(
- reinterpret_cast<const char*>(data.data()), data.size());
- DataInputStream in(input_stream);
+TEST(DeletionVectorTest, ReadFromDataInputStreamBitmap64) {
auto pool = GetDefaultPool();
-
- ASSERT_NOK_WITH_MSG(
- DeletionVector::Read(&in, std::nullopt, pool.get()),
- "NotImplemented: bitmap64 deletion vectors are not supported in this
version");
+ Bitmap64DeletionVector dv;
+ std::vector<int64_t> positions = {1, (1LL << 32) + 7, (2LL << 32) + 3};
+ for (const auto& position : positions) {
+ ASSERT_OK(dv.Delete(position));
+ }
+ auto output = std::make_shared<ByteArrayOutputStream>(
+ std::make_unique<MemorySegmentOutputStream>(1024, pool));
+ DataOutputStream out(output);
+ ASSERT_OK_AND_ASSIGN(int32_t length, dv.SerializeTo(pool, &out));
+ ASSERT_OK_AND_ASSIGN(auto bytes, output->Finish(pool.get()));
+ ASSERT_EQ(length, bytes->size());
+
+ auto input = std::make_shared<ByteArrayInputStream>(bytes->data(),
bytes->size());
+ DataInputStream in(input);
+ ASSERT_OK_AND_ASSIGN(auto decoded, DeletionVector::Read(&in, length,
pool.get()));
+ ASSERT_TRUE(dynamic_cast<Bitmap64DeletionVector*>(decoded.get()));
+ ASSERT_EQ(decoded->GetCardinality().value(), positions.size());
+ for (const auto& position : positions) {
+ ASSERT_TRUE(decoded->IsDeleted(position).value());
+ }
+ ASSERT_FALSE(decoded->IsDeleted((1LL << 32) + 8).value());
+ ASSERT_EQ(input->GetPos().value(), bytes->size());
}
TEST(DeletionVectorTest, ReadFromDataInputStreamInvalidMagicNumber) {
diff --git a/src/paimon/core/deletionvectors/deletion_vectors_index_file.cpp
b/src/paimon/core/deletionvectors/deletion_vectors_index_file.cpp
index 8743b43a..4467a701 100644
--- a/src/paimon/core/deletionvectors/deletion_vectors_index_file.cpp
+++ b/src/paimon/core/deletionvectors/deletion_vectors_index_file.cpp
@@ -55,6 +55,7 @@ DeletionVectorsIndexFile::ReadAllDeletionVectors(
auto data_input_stream = std::make_shared<DataInputStream>(input_stream);
PAIMON_RETURN_NOT_OK(CheckVersion(data_input_stream));
for (const auto& [_, deletion_vector_meta] :
deletion_vector_metas.value()) {
+
PAIMON_RETURN_NOT_OK(data_input_stream->Seek(deletion_vector_meta.GetOffset()));
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<DeletionVector> dv,
DeletionVector::Read(data_input_stream.get(),
diff --git
a/src/paimon/core/deletionvectors/deletion_vectors_index_file_test.cpp
b/src/paimon/core/deletionvectors/deletion_vectors_index_file_test.cpp
index 817c158b..9181d2bf 100644
--- a/src/paimon/core/deletionvectors/deletion_vectors_index_file_test.cpp
+++ b/src/paimon/core/deletionvectors/deletion_vectors_index_file_test.cpp
@@ -77,6 +77,20 @@ TEST(DeletionVectorsIndexFileTest, Basic) {
ASSERT_TRUE(is_deleted);
ASSERT_OK_AND_ASSIGN(is_deleted, read_back.at("dv2")->IsDeleted(99));
ASSERT_FALSE(is_deleted);
+
+ // Metadata order need not match the physical order of deletion vectors.
+ LinkedHashMap<std::string, DeletionVectorMeta> reversed;
+ for (const auto& name : {"dv2", "dv1"}) {
+ reversed.insert_or_assign(name,
meta->DvRanges().value().find(name)->second);
+ }
+ auto reversed_meta = std::make_shared<IndexFileMeta>(
+ meta->IndexType(), meta->FileName(), meta->FileSize(), 2, reversed,
std::nullopt);
+ ASSERT_OK_AND_ASSIGN(auto reversed_read,
index_file->ReadAllDeletionVectors(reversed_meta));
+ ASSERT_EQ(reversed_read.size(), 2);
+ ASSERT_TRUE(reversed_read.at("dv1")->IsDeleted(0).value());
+ ASSERT_FALSE(reversed_read.at("dv1")->IsDeleted(100).value());
+ ASSERT_TRUE(reversed_read.at("dv2")->IsDeleted(100).value());
+ ASSERT_FALSE(reversed_read.at("dv2")->IsDeleted(0).value());
}
TEST(DeletionVectorsIndexFileTest, ExternalPathAndIndexFileMeta) {
diff --git a/src/paimon/core/io/file_index_evaluator.cpp
b/src/paimon/core/io/file_index_evaluator.cpp
index 4405c099..59c20f47 100644
--- a/src/paimon/core/io/file_index_evaluator.cpp
+++ b/src/paimon/core/io/file_index_evaluator.cpp
@@ -46,6 +46,7 @@
#include "paimon/predicate/predicate_utils.h"
#include "paimon/predicate/vector_search.h"
#include "paimon/status.h"
+#include "paimon/utils/roaring_bitmap32.h"
namespace paimon {
class MemoryPool;
@@ -75,7 +76,7 @@ Result<std::shared_ptr<FileIndexResult>>
FileIndexEvaluator::Evaluate(
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<FileSystem>& file_system,
const std::shared_ptr<MemoryPool>& pool) {
- if (predicate == nullptr) {
+ if (predicate == nullptr || file_meta->row_count >
RoaringBitmap32::MAX_VALUE) {
return FileIndexResult::Remain();
}
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
diff --git a/src/paimon/core/io/file_index_evaluator_test.cpp
b/src/paimon/core/io/file_index_evaluator_test.cpp
index adeb5a39..1a0168ef 100644
--- a/src/paimon/core/io/file_index_evaluator_test.cpp
+++ b/src/paimon/core/io/file_index_evaluator_test.cpp
@@ -604,4 +604,30 @@ INSTANTIATE_TEST_SUITE_P(IndexStorage,
FileIndexSearchEvaluatorTest, ::testing::
return info.param ? "Embedded" : "External";
});
+TEST_F(FileIndexEvaluatorTest, TestLargeFileSkips32BitIndex) {
+ auto data_file_meta = std::make_shared<DataFileMeta>(
+ "data-414509f5-e40c-4245-b992-bbf486778ac9-0.orc", /*file_size=*/689,
+ /*row_count=*/(1LL << 32), /*min_key=*/BinaryRow::EmptyRow(),
+ /*max_key=*/BinaryRow::EmptyRow(),
/*key_stats=*/SimpleStats::EmptyStats(),
+ /*value_stats=*/SimpleStats::EmptyStats(), /*min_sequence_number=*/0,
+ /*max_sequence_number=*/7, /*schema_id=*/0,
+ /*level=*/0,
+ /*extra_files=*/
+ std::vector<std::optional<std::string>>(
+ {"data-414509f5-e40c-4245-b992-bbf486778ac9-0.orc.index"}),
+ /*creation_time=*/Timestamp(0ll, 0), /*delete_row_count=*/0,
+ /*embedded_index=*/nullptr, FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
/*write_cols=*/std::nullopt,
+ /*column_max_sequence_numbers=*/std::nullopt);
+ auto predicate =
+ PredicateBuilder::IsNull(/*field_index=*/2, /*field_name=*/"f2",
FieldType::INT);
+ ASSERT_OK_AND_ASSIGN(
+ auto result, FileIndexEvaluator::Evaluate(data_schema_,
core_options_.ToMap(), predicate,
+
/*data_file_path_factory=*/nullptr,
+ data_file_meta,
/*file_system=*/nullptr, pool_));
+ ASSERT_TRUE(result->IsRemain().value());
+ ASSERT_EQ(dynamic_cast<BitmapIndexResult*>(result.get()), nullptr);
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/operation/abstract_split_read.cpp
b/src/paimon/core/operation/abstract_split_read.cpp
index 690f7bbe..c6669023 100644
--- a/src/paimon/core/operation/abstract_split_read.cpp
+++ b/src/paimon/core/operation/abstract_split_read.cpp
@@ -50,6 +50,8 @@
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/object_utils.h"
+#include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/io/complete_row_tracking_fields_reader.h"
#include "paimon/core/io/data_file_meta.h"
@@ -185,8 +187,13 @@ Result<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::ApplyIndexAndDvReade
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
const std::map<std::string, std::string>& file_index_options,
std::vector<float>* index_scores) const {
+ std::shared_ptr<DeletionVector> deletion_vector;
+ if (dv_factory) {
+ PAIMON_ASSIGN_OR_RAISE(deletion_vector, dv_factory(file->file_name));
+ }
std::shared_ptr<FileIndexResult> file_index_result;
- if (options_.FileIndexReadEnabled()) {
+ if (options_.FileIndexReadEnabled() &&
+ dynamic_cast<Bitmap64DeletionVector*>(deletion_vector.get()) ==
nullptr) {
PAIMON_ASSIGN_OR_RAISE(file_index_result,
FileIndexEvaluator::Evaluate(data_schema,
file_index_options,
predicate,
data_file_path_factory, file,
@@ -217,10 +224,6 @@ Result<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::ApplyIndexAndDvReade
}
// prepare deletion bitmap for deletion vector
- std::shared_ptr<DeletionVector> deletion_vector;
- if (dv_factory) {
- PAIMON_ASSIGN_OR_RAISE(deletion_vector, dv_factory(file->file_name));
- }
const RoaringBitmap32* deletion = nullptr;
if (auto* bitmap_dv =
dynamic_cast<BitmapDeletionVector*>(deletion_vector.get())) {
deletion = bitmap_dv->GetBitmap();
@@ -247,9 +250,8 @@ Result<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::ApplyIndexAndDvReade
std::move(actual_selection)));
if (deletion_vector && !deletion && !deletion_vector->IsEmpty()) {
- // TODO(xinyu.lxy): if deletion vector is bitmap64, use
ApplyBitmapIndexBatchReader to
- // filter result
- return Status::NotImplemented("Only support BitmapDeletionVector");
+ reader =
+
std::make_unique<ApplyDeletionVectorBatchReader>(std::move(reader),
deletion_vector);
}
return reader;
}
diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp
b/src/paimon/core/operation/data_evolution_split_read.cpp
index 7a1a7f18..5c26d341 100644
--- a/src/paimon/core/operation/data_evolution_split_read.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read.cpp
@@ -97,8 +97,17 @@ class PositionShiftedDeletionVector : public DeletionVector {
/// IsEmpty; the row counting that does consult a vector's cardinality
builds its factory
/// from the split's deletion files, never from this view.
Result<int64_t> GetCardinality() const override {
- PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 valid, inner_->IsValid(offset_,
length_));
- return length_ - valid.Cardinality();
+ int64_t count = 0;
+ PAIMON_RETURN_NOT_OK(ForEachDeletedPosition([&count](int64_t) {
++count; }));
+ return count;
+ }
+
+ Status ForEachDeletedPosition(const std::function<void(int64_t)>&
consumer) const override {
+ return inner_->ForEachDeletedPosition([&](int64_t position) {
+ if (position >= offset_ && position - offset_ < length_) {
+ consumer(position - offset_);
+ }
+ });
}
Status Delete(int64_t) override {
diff --git a/src/paimon/core/operation/data_evolution_split_read_test.cpp
b/src/paimon/core/operation/data_evolution_split_read_test.cpp
index 6b4067b0..39fdb100 100644
--- a/src/paimon/core/operation/data_evolution_split_read_test.cpp
+++ b/src/paimon/core/operation/data_evolution_split_read_test.cpp
@@ -26,6 +26,7 @@
#include "gtest/gtest.h"
#include "paimon/common/data/binary_row.h"
#include "paimon/common/table/special_fields.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/manifest/file_source.h"
@@ -872,4 +873,26 @@ TEST_F(DataEvolutionSplitReadTest,
TestExcludeDeletedRowIds) {
ASSERT_TRUE(remaining.empty());
}
+TEST_F(DataEvolutionSplitReadTest, Bitmap64GroupWindowWithHighOffset) {
+ const int64_t offset = 1LL << 32;
+ auto anchor = CreateNormalFile("anchor.parquet", 100, offset + 30, 10);
+ auto blob = CreateDataFileMeta("blob0.blob", 100 + offset, 20, 20);
+ auto dv = std::make_shared<Bitmap64DeletionVector>();
+ for (int64_t p : {offset - 1, offset, offset + 5, offset + 20}) {
+ ASSERT_OK(dv->Delete(p));
+ }
+ DataEvolutionSplitRead::GroupDeletionVector group_dv{dv, Range(100, 100 +
offset + 29)};
+ ASSERT_OK_AND_ASSIGN(auto factory,
+ DataEvolutionSplitRead::CreateGroupDvFactory({anchor,
blob}, group_dv));
+ ASSERT_OK_AND_ASSIGN(auto view, factory("blob0.blob"));
+ ASSERT_TRUE(view);
+ ASSERT_TRUE(view->IsDeleted(0).value());
+ ASSERT_TRUE(view->IsDeleted(5).value());
+ ASSERT_FALSE(view->IsDeleted(6).value());
+ ASSERT_EQ(view->GetCardinality().value(), 2);
+ std::vector<int64_t> positions;
+ ASSERT_OK(view->ForEachDeletedPosition([&](int64_t p) {
positions.push_back(p); }));
+ ASSERT_EQ(positions, (std::vector<int64_t>{0, 5}));
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/operation/raw_file_split_read_test.cpp
b/src/paimon/core/operation/raw_file_split_read_test.cpp
index c6ad7f98..77f92109 100644
--- a/src/paimon/core/operation/raw_file_split_read_test.cpp
+++ b/src/paimon/core/operation/raw_file_split_read_test.cpp
@@ -35,6 +35,7 @@
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/core/core_options.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/global_index/indexed_split_impl.h"
#include "paimon/core/io/data_file_meta.h"
@@ -52,6 +53,7 @@
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/full_text_search.h"
+#include "paimon/predicate/predicate_builder.h"
#include "paimon/predicate/vector_search.h"
#include "paimon/read_context.h"
#include "paimon/status.h"
@@ -190,12 +192,17 @@ class RawFileSplitReadTest : public ::testing::Test {
}
void CheckReadResult(const std::shared_ptr<arrow::Schema>& read_schema,
- const std::shared_ptr<arrow::ChunkedArray>&
expected_array) const {
+ const std::shared_ptr<arrow::ChunkedArray>&
expected_array,
+ DeletionVector::Factory dv_factory = {}) const {
std::string path = paimon::test::GetDataDir() +
"/orc/multi_partition_append_table.db/"
"multi_partition_append_table";
ReadContextBuilder context_builder(path);
context_builder.SetReadFieldNames(read_schema->field_names());
+ if (dv_factory) {
+ context_builder.AddOption(Options::FILE_INDEX_READ_ENABLED,
"true");
+ context_builder.SetPredicate(PredicateBuilder::IsNotNull(0, "f0",
FieldType::STRING));
+ }
ASSERT_OK_AND_ASSIGN(std::shared_ptr<ReadContext> read_context,
context_builder.Finish());
SchemaManager schema_manager(std::make_shared<LocalFileSystem>(),
read_context->GetPath());
ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
@@ -227,8 +234,18 @@ class RawFileSplitReadTest : public ::testing::Test {
std::vector<std::unique_ptr<BatchReader>> batch_readers;
batch_readers.reserve(data_splits.size());
for (const auto& split : data_splits) {
- ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader,
- split_read->CreateReader(split));
+ auto impl = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ if (dv_factory) {
+ // A DV64 read must bypass file-index evaluation even for a
small file.
+ for (const auto& file : impl->DataFiles()) {
+ file->embedded_index = std::make_shared<Bytes>("invalid
index", pool_.get());
+ }
+ }
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<BatchReader> reader,
+ dv_factory ? split_read->CreateReader(impl->Partition(),
impl->Bucket(),
+ impl->DataFiles(),
dv_factory, std::nullopt)
+ : split_read->CreateReader(split));
batch_readers.emplace_back(std::move(reader));
}
auto batch_reader =
@@ -874,4 +891,23 @@ TEST_F(RawFileSplitReadTest, TestMatch) {
ASSERT_NOK(split_read->Match(nullptr, /*force_keep_delete=*/false));
}
+TEST_F(RawFileSplitReadTest, TestBitmap64DeletionVectorFallback) {
+ auto read_schema =
+ arrow::schema({arrow::field("f0", arrow::utf8()), arrow::field("f1",
arrow::int32()),
+ arrow::field("f2", arrow::int32()), arrow::field("f3",
arrow::float64())});
+ auto result_fields = read_schema->fields();
+ result_fields.insert(result_fields.begin(), arrow::field("_VALUE_KIND",
arrow::int8()));
+ auto expected =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_(result_fields), R"([[0, "Bob", 10, 0, 12.1], [0,
"Tony", 10, 0, 14.1]])")
+ .ValueOrDie();
+ DeletionVector::Factory factory =
+ [](const std::string& name) -> Result<std::shared_ptr<DeletionVector>>
{
+ auto dv = std::make_shared<Bitmap64DeletionVector>();
+ PAIMON_RETURN_NOT_OK(dv->Delete(name.find("01b6a930") !=
std::string::npos ? 1 : 0));
+ return dv;
+ };
+ CheckReadResult(read_schema,
std::make_shared<arrow::ChunkedArray>(expected), factory);
+}
+
} // namespace paimon::test
diff --git a/src/paimon/testing/utils/deletion_vector_test_helper.h
b/src/paimon/testing/utils/deletion_vector_test_helper.h
index ebca6e4d..35adf7f5 100644
--- a/src/paimon/testing/utils/deletion_vector_test_helper.h
+++ b/src/paimon/testing/utils/deletion_vector_test_helper.h
@@ -31,6 +31,7 @@
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/range_helper.h"
#include "paimon/core/core_options.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
#include "paimon/core/deletionvectors/deletion_vector.h"
#include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
@@ -117,7 +118,12 @@ class DeletionVectorTestHelper {
for (const auto& [file_name, deleted_positions] :
deleted_positions_by_file) {
// deleting through the vector itself applies the same position
bound a writer is
// held to, instead of narrowing to the bitmap's index type and
wrapping silently
- auto deletion_vector =
std::make_shared<BitmapDeletionVector>(RoaringBitmap32());
+ std::shared_ptr<DeletionVector> deletion_vector;
+ if (core_options.DeletionVectorsBitmap64()) {
+ deletion_vector = std::make_shared<Bitmap64DeletionVector>();
+ } else {
+ deletion_vector =
std::make_shared<BitmapDeletionVector>(RoaringBitmap32());
+ }
for (int64_t position : deleted_positions) {
if (position < 0) {
return Status::Invalid(fmt::format(
diff --git a/test/inte/data_evolution_table_test.cpp
b/test/inte/data_evolution_table_test.cpp
index 05da479d..804e7c26 100644
--- a/test/inte/data_evolution_table_test.cpp
+++ b/test/inte/data_evolution_table_test.cpp
@@ -48,11 +48,12 @@
#include "paimon/testing/utils/test_helper.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
-using DataEvolutionTableParam = std::tuple<std::string, bool>;
+using DataEvolutionTableParam = std::tuple<std::string, bool, bool>;
// This is a sdk end-to-end test for data evolution
class DataEvolutionTableTest : public ::testing::Test,
public
::testing::WithParamInterface<DataEvolutionTableParam> {
+ protected:
void SetUp() override {
dir_ = UniqueTestDirectory::Create("local");
int64_t seed = DateTimeUtils::GetCurrentUTCTimeUs();
@@ -168,6 +169,8 @@ class DataEvolutionTableTest : public ::testing::Test,
{Options::DATA_EVOLUTION_ENABLED, "true"}};
if (deletion_vectors_enabled) {
options.emplace(Options::DELETION_VECTORS_ENABLED, "true");
+ options.emplace(Options::DELETION_VECTOR_BITMAP64,
+ std::get<2>(GetParam()) ? "true" : "false");
}
options.insert(extra_options.begin(), extra_options.end());
CreateTable(/*partition_keys=*/{}, options);
@@ -682,7 +685,7 @@ class DataEvolutionTableTest : public ::testing::Test,
.ValueOrDie());
}
- private:
+ protected:
std::unique_ptr<UniqueTestDirectory> dir_;
mutable std::shared_ptr<Cache> snapshot_live_manifest_cache_;
arrow::FieldVector fields_ = {
@@ -692,6 +695,8 @@ class DataEvolutionTableTest : public ::testing::Test,
};
};
+class DataEvolutionDvTableTest : public DataEvolutionTableTest {};
+
TEST_P(DataEvolutionTableTest, TestBasic) {
CreateTable();
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
@@ -2935,15 +2940,15 @@ TEST_P(DataEvolutionTableTest, TestWithRowIds) {
}
}
-TEST_P(DataEvolutionTableTest, TestGlobalIndexScoresWithSingleFileFiltering) {
+TEST_P(DataEvolutionDvTableTest, TestGlobalIndexScoresWithSingleFileFiltering)
{
CheckGlobalIndexScoresWithFiltering(/*merge_files=*/false);
}
-TEST_P(DataEvolutionTableTest, TestGlobalIndexScoresWithMergedFileFiltering) {
+TEST_P(DataEvolutionDvTableTest, TestGlobalIndexScoresWithMergedFileFiltering)
{
CheckGlobalIndexScoresWithFiltering(/*merge_files=*/true);
}
-TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectors) {
+TEST_P(DataEvolutionDvTableTest, TestReadWithDeletionVectors) {
CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
@@ -3020,7 +3025,7 @@ TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectors) {
/*predicate=*/nullptr, /*row_ranges=*/{Range(1,
2)}));
}
-TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectorsAcrossReadBatches) {
+TEST_P(DataEvolutionDvTableTest, TestReadWithDeletionVectorsAcrossReadBatches)
{
// the 12 rows below span several read batches, so the deletion vector
empties a whole
// batch of every file of the group: each file reader then skips that
batch entirely and
// the column merge has to stay aligned on the surviving row count alone
@@ -3069,7 +3074,7 @@ TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectorsAcrossReadBatches) {
/*predicate=*/nullptr, /*row_ranges=*/{Range(3,
8)}));
}
-TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectorsOnPartOfRowRangeGroups) {
+TEST_P(DataEvolutionDvTableTest,
TestReadWithDeletionVectorsOnPartOfRowRangeGroups) {
// one split per row range group, so a group's deletion file must not
reach the other
CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
{{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
@@ -3139,7 +3144,7 @@ TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectorsOnPartOfRowRangeGroups
ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));
}
-TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectorsOnEveryRowRangeGroup) {
+TEST_P(DataEvolutionDvTableTest,
TestReadWithDeletionVectorsOnEveryRowRangeGroup) {
CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
@@ -3225,7 +3230,7 @@ TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectorsOnEveryRowRangeGroup)
/*predicate=*/nullptr, /*row_ranges=*/{Range(2,
5)}));
}
-TEST_P(DataEvolutionTableTest, TestReadWithFullyDeletedRowRangeGroup) {
+TEST_P(DataEvolutionDvTableTest, TestReadWithFullyDeletedRowRangeGroup) {
CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
@@ -3284,7 +3289,7 @@ TEST_P(DataEvolutionTableTest,
TestReadWithFullyDeletedRowRangeGroup) {
ASSERT_FALSE(only_deleted.rows);
}
-TEST_P(DataEvolutionTableTest, TestReadAfterUpdatingDeletionVectors) {
+TEST_P(DataEvolutionDvTableTest, TestReadAfterUpdatingDeletionVectors) {
CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
auto schema = arrow::schema(fields_);
@@ -3330,7 +3335,7 @@ TEST_P(DataEvolutionTableTest,
TestReadAfterUpdatingDeletionVectors) {
ASSERT_OK(ScanAndRead(table_path, schema->field_names(),
expected_after_update));
}
-TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectorsAfterAddingColumn) {
+TEST_P(DataEvolutionDvTableTest, TestReadWithDeletionVectorsAfterAddingColumn)
{
if (FileFormat() == "avro") {
GTEST_SKIP() << "Avro has no stats, which the added column's scan
pruning relies on";
}
@@ -3392,7 +3397,7 @@ TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectorsAfterAddingColumn) {
ASSERT_OK(ScanAndRead(table_path, {"f3"}, expected_f3));
}
-TEST_P(DataEvolutionTableTest,
TestLimitPushDownWithHeavilyDeletedFirstRowRangeGroup) {
+TEST_P(DataEvolutionDvTableTest,
TestLimitPushDownWithHeavilyDeletedFirstRowRangeGroup) {
// one split per row range group, so the limit has to span both to be
satisfied
CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
{{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
@@ -3541,23 +3546,36 @@ TEST_P(DataEvolutionTableTest,
TestLimitPushDownDisabledByRowRangeIndex) {
std::vector<DataEvolutionTableParam> GetTestValuesForDataEvolutionTableTest() {
std::vector<DataEvolutionTableParam> values;
for (bool enable_snapshot_live_manifest_cache : {false, true}) {
- values.emplace_back("parquet", enable_snapshot_live_manifest_cache);
+ values.emplace_back("parquet", enable_snapshot_live_manifest_cache,
/*bitmap64=*/false);
#ifdef PAIMON_ENABLE_MOSAIC
- values.emplace_back("mosaic", enable_snapshot_live_manifest_cache);
+ values.emplace_back("mosaic", enable_snapshot_live_manifest_cache,
/*bitmap64=*/false);
#endif
#ifdef PAIMON_ENABLE_LANCE
- values.emplace_back("lance", enable_snapshot_live_manifest_cache);
+ values.emplace_back("lance", enable_snapshot_live_manifest_cache,
/*bitmap64=*/false);
#endif
#ifdef PAIMON_ENABLE_ORC
- values.emplace_back("orc", enable_snapshot_live_manifest_cache);
+ values.emplace_back("orc", enable_snapshot_live_manifest_cache,
/*bitmap64=*/false);
#endif
#ifdef PAIMON_ENABLE_AVRO
- values.emplace_back("avro", enable_snapshot_live_manifest_cache);
+ values.emplace_back("avro", enable_snapshot_live_manifest_cache,
/*bitmap64=*/false);
#endif
}
return values;
}
+std::vector<DataEvolutionTableParam>
GetTestValuesForDataEvolutionDvTableTest() {
+ std::vector<DataEvolutionTableParam> values;
+ for (const auto& param : GetTestValuesForDataEvolutionTableTest()) {
+ for (const auto& bitmap64 : {false, true}) {
+ values.emplace_back(std::get<0>(param), std::get<1>(param),
bitmap64);
+ }
+ }
+ return values;
+}
+
+INSTANTIATE_TEST_SUITE_P(FileFormat, DataEvolutionDvTableTest,
+
::testing::ValuesIn(GetTestValuesForDataEvolutionDvTableTest()));
+
INSTANTIATE_TEST_SUITE_P(FileFormat, DataEvolutionTableTest,
::testing::ValuesIn(GetTestValuesForDataEvolutionTableTest()));
diff --git a/test/inte/pk_compaction_inte_test.cpp
b/test/inte/pk_compaction_inte_test.cpp
index 7e713b06..153a7926 100644
--- a/test/inte/pk_compaction_inte_test.cpp
+++ b/test/inte/pk_compaction_inte_test.cpp
@@ -20,6 +20,7 @@
#include <memory>
#include <optional>
#include <string>
+#include <tuple>
#include <utility>
#include <vector>
@@ -35,6 +36,7 @@
#include "paimon/common/table/special_fields.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/path_util.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
#include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
#include "paimon/core/io/data_file_meta.h"
#include "paimon/core/manifest/file_source.h"
@@ -45,12 +47,14 @@
#include "paimon/defs.h"
#include "paimon/file_store_commit.h"
#include "paimon/file_store_write.h"
+#include "paimon/format/file_format_factory.h"
#include "paimon/fs/file_system.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/predicate/literal.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/read_context.h"
+#include "paimon/reader/prefetch_file_batch_reader.h"
#include "paimon/record_batch.h"
#include "paimon/result.h"
#include "paimon/scan_context.h"
@@ -67,8 +71,188 @@
namespace paimon::test {
-class PkCompactionInteTest : public ::testing::Test,
- public ::testing::WithParamInterface<std::string>
{
+// Stores a few real Parquet rows and exposes sparse file positions to
exercise the full
+// lookup/DV path. This simulates large positions, not a physically large
Parquet file.
+class HighPositionFileBatchReader : public PrefetchFileBatchReader {
+ public:
+ explicit
HighPositionFileBatchReader(std::unique_ptr<PrefetchFileBatchReader> reader)
+ : reader_(std::move(reader)) {}
+
+ Result<ReadBatch> NextBatch() override {
+ return reader_->NextBatch();
+ }
+
+ Result<ReadBatchWithBitmap> NextBatchWithBitmap() override {
+ return reader_->NextBatchWithBitmap();
+ }
+
+ Result<std::unique_ptr<::ArrowSchema>> GetFileSchema() const override {
+ return reader_->GetFileSchema();
+ }
+
+ Status SetReadSchema(::ArrowSchema* schema, const
std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection)
override {
+ if (selection) {
+ return Status::Invalid("High position test format does not support
bitmap32 selection");
+ }
+ return reader_->SetReadSchema(schema, predicate, std::nullopt);
+ }
+
+ Result<uint64_t> GetPreviousBatchFileRowId(uint64_t batch_row_id) const
override {
+ PAIMON_ASSIGN_OR_RAISE(uint64_t position,
reader_->GetPreviousBatchFileRowId(batch_row_id));
+ return ToHighPosition(position);
+ }
+
+ Result<uint64_t> GetNumberOfRows() const override {
+ PAIMON_ASSIGN_OR_RAISE(uint64_t count, reader_->GetNumberOfRows());
+ return ToHighPosition(count);
+ }
+
+ Status SeekToRow(uint64_t row_number) override {
+ return reader_->SeekToRow(ToPhysicalPosition(row_number));
+ }
+
+ Result<uint64_t> GetNextRowToRead() const override {
+ PAIMON_ASSIGN_OR_RAISE(uint64_t position, reader_->GetNextRowToRead());
+ return ToHighPosition(position);
+ }
+
+ Result<std::vector<std::pair<uint64_t, uint64_t>>> GenReadRanges(
+ bool* need_prefetch) const override {
+ std::vector<std::pair<uint64_t, uint64_t>> ranges;
+ PAIMON_ASSIGN_OR_RAISE(ranges, reader_->GenReadRanges(need_prefetch));
+ for (auto& range : ranges) {
+ range = {ToHighPosition(range.first),
ToHighPosition(range.second)};
+ }
+ return ranges;
+ }
+
+ Status SetReadRanges(const std::vector<std::pair<uint64_t, uint64_t>>&
ranges) override {
+ std::vector<std::pair<uint64_t, uint64_t>> physical_ranges;
+ for (const auto& range : ranges) {
+ physical_ranges.emplace_back(ToPhysicalPosition(range.first),
+ ToPhysicalPosition(range.second));
+ }
+ return reader_->SetReadRanges(physical_ranges);
+ }
+
+ Result<std::vector<std::pair<uint64_t, uint64_t>>> PreBufferRange()
override {
+ return reader_->PreBufferRange();
+ }
+
+ void SetPreBufferRangeCallback(PreBufferRangeCallback callback) override {
+ reader_->SetPreBufferRangeCallback(std::move(callback));
+ }
+
+ bool SupportPreciseBitmapSelection() const override {
+ return false;
+ }
+
+ std::shared_ptr<Metrics> GetReaderMetrics() const override {
+ return reader_->GetReaderMetrics();
+ }
+
+ void Close() override {
+ reader_->Close();
+ }
+
+ void Warmup() override {
+ reader_->Warmup();
+ }
+
+ private:
+ static uint64_t ToHighPosition(uint64_t position) {
+ return (position << 32) + 7;
+ }
+
+ static uint64_t ToPhysicalPosition(uint64_t position) {
+ // Seek to the first stored row at or after a position in the sparse
coordinate space.
+ return position <= 7 ? 0 : ((position - 8) >> 32) + 1;
+ }
+
+ std::unique_ptr<PrefetchFileBatchReader> reader_;
+};
+
+class HighPositionReaderBuilder : public ReaderBuilder {
+ public:
+ explicit HighPositionReaderBuilder(std::unique_ptr<ReaderBuilder> builder)
+ : builder_(std::move(builder)) {}
+
+ ReaderBuilder* WithMemoryPool(const std::shared_ptr<MemoryPool>& pool)
override {
+ builder_->WithMemoryPool(pool);
+ return this;
+ }
+
+ ReaderBuilder* WithCache(const std::shared_ptr<Cache>& cache) override {
+ builder_->WithCache(cache);
+ return this;
+ }
+
+ ReaderBuilder* WithReadHints(const std::optional<ReadHints>& hints)
override {
+ builder_->WithReadHints(hints);
+ return this;
+ }
+
+ Result<std::unique_ptr<FileBatchReader>> Build(
+ const std::shared_ptr<InputStream>& input) const override {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileBatchReader> reader,
builder_->Build(input));
+ return
std::unique_ptr<FileBatchReader>(std::make_unique<HighPositionFileBatchReader>(
+ checked_pointer_cast<PrefetchFileBatchReader>(std::move(reader))));
+ }
+
+ private:
+ std::unique_ptr<ReaderBuilder> builder_;
+};
+
+class HighPositionFileFormat : public FileFormat {
+ public:
+ explicit HighPositionFileFormat(std::unique_ptr<FileFormat> parquet)
+ : parquet_(std::move(parquet)) {}
+
+ const std::string& Identifier() const override {
+ return identifier_;
+ }
+
+ Result<std::unique_ptr<ReaderBuilder>> CreateReaderBuilder(int32_t
batch_size) const override {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReaderBuilder> builder,
+ parquet_->CreateReaderBuilder(batch_size));
+ return std::unique_ptr<ReaderBuilder>(
+ std::make_unique<HighPositionReaderBuilder>(std::move(builder)));
+ }
+
+ Result<std::unique_ptr<WriterBuilder>> CreateWriterBuilder(::ArrowSchema*
schema,
+ int32_t
batch_size) const override {
+ return parquet_->CreateWriterBuilder(schema, batch_size);
+ }
+
+ Result<std::unique_ptr<FormatStatsExtractor>> CreateStatsExtractor(
+ ::ArrowSchema* schema) const override {
+ return parquet_->CreateStatsExtractor(schema);
+ }
+
+ private:
+ const std::string identifier_ = "high_position_parquet";
+ std::unique_ptr<FileFormat> parquet_;
+};
+
+class HighPositionFileFormatFactory : public FileFormatFactory {
+ public:
+ const char* Identifier() const override {
+ return "high_position_parquet";
+ }
+
+ Result<std::unique_ptr<FileFormat>> Create(
+ const std::map<std::string, std::string>& options) const override {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileFormat> parquet,
+ FileFormatFactory::Get("parquet", options));
+ return std::unique_ptr<FileFormat>(
+ std::make_unique<HighPositionFileFormat>(std::move(parquet)));
+ }
+};
+
+REGISTER_PAIMON_FACTORY(HighPositionFileFormatFactory);
+
+class PkCompactionInteTest : public ::testing::Test {
public:
void SetUp() override {
pool_ = GetDefaultPool();
@@ -211,8 +395,10 @@ class PkCompactionInteTest : public ::testing::Test,
Result<std::vector<std::shared_ptr<CommitMessage>>> CompactAndCommit(
const std::string& table_path, const std::map<std::string,
std::string>& partition,
- int32_t bucket, bool full_compaction, int64_t commit_identifier) {
+ int32_t bucket, bool full_compaction, int64_t commit_identifier,
+ const std::map<std::string, std::string>& options = {}) {
WriteContextBuilder write_builder(table_path, "commit_user_1");
+ write_builder.SetOptions(options);
write_builder.WithStreamingMode(true).WithTempDirectory(
PathUtil::JoinPath(dir_->Str(), "tmp"));
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriteContext> write_context,
write_builder.Finish());
@@ -378,12 +564,22 @@ class PkCompactionInteTest : public ::testing::Test,
return split.value();
}
- private:
+ protected:
std::shared_ptr<MemoryPool> pool_;
std::unique_ptr<UniqueTestDirectory> dir_;
arrow::FieldVector fields_;
};
+class PkCompactionFormatInteTest : public PkCompactionInteTest,
+ public
::testing::WithParamInterface<std::string> {};
+
+class PkCompactionDvInteTest : public PkCompactionInteTest,
+ public ::testing::WithParamInterface<bool> {};
+
+class PkCompactionFormatDvInteTest
+ : public PkCompactionInteTest,
+ public ::testing::WithParamInterface<std::tuple<std::string, bool>> {};
+
TEST_F(PkCompactionInteTest, TestMetadataOnlyLevelUpgradeKeepsValueStats) {
arrow::FieldVector fields = {arrow::field("id", arrow::int32()),
arrow::field("value", arrow::utf8())};
@@ -442,7 +638,7 @@ TEST_F(PkCompactionInteTest,
TestMetadataOnlyLevelUpgradeKeepsValueStats) {
}
// Verify shared-shredding MAP can be read correctly after PK full compaction.
-TEST_P(PkCompactionInteTest,
TestKeyValueTableFullCompactionWithMapSharedShredding) {
+TEST_P(PkCompactionFormatInteTest,
TestKeyValueTableFullCompactionWithMapSharedShredding) {
auto file_format = GetParam();
if (file_format == "avro" || file_format == "mosaic" || file_format ==
"lance") {
return;
@@ -514,8 +710,8 @@ TEST_P(PkCompactionInteTest,
TestKeyValueTableFullCompactionWithMapSharedShreddi
ASSERT_TRUE(success);
}
-TEST_P(PkCompactionInteTest,
TestKeyValueTableDvCompactionWithMapSharedShredding) {
- auto file_format = GetParam();
+TEST_P(PkCompactionFormatDvInteTest,
TestKeyValueTableDvCompactionWithMapSharedShredding) {
+ auto file_format = std::get<0>(GetParam());
if (file_format == "avro" || file_format == "mosaic" || file_format ==
"lance") {
return;
}
@@ -535,6 +731,7 @@ TEST_P(PkCompactionInteTest,
TestKeyValueTableDvCompactionWithMapSharedShredding
{Options::FILE_SYSTEM, "local"},
{Options::FILE_COMPRESSION, "none"},
{Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, std::get<1>(GetParam()) ? "true" :
"false"},
{"fields.tags.map.storage-layout", "shared-shredding"},
{"fields.tags.map.shared-shredding.max-columns", "2"},
{"parquet.page.size", "1"},
@@ -640,7 +837,7 @@ TEST_P(PkCompactionInteTest,
TestKeyValueTableDvCompactionWithMapSharedShredding
// 7. ScanAndVerify after DV compact (data read with DV applied)
// 8. Full compact to merge everything
// 9. ScanAndVerify after full compact (all data merged)
-TEST_F(PkCompactionInteTest, DeduplicateWithDeletionVectors) {
+TEST_P(PkCompactionDvInteTest, DeduplicateWithDeletionVectors) {
// f4 is a large padding field to make the initial file substantially
bigger than
// subsequent small level-0 files, preventing PickForSizeRatio from
merging them all.
arrow::FieldVector fields = {
@@ -649,11 +846,13 @@ TEST_F(PkCompactionInteTest,
DeduplicateWithDeletionVectors) {
arrow::field("f4", arrow::utf8())};
std::vector<std::string> primary_keys = {"f0", "f1", "f2"};
std::vector<std::string> partition_keys = {"f1"};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "2"},
- {Options::BUCKET_KEY, "f2"},
- {Options::FILE_SYSTEM,
"local"},
-
{Options::DELETION_VECTORS_ENABLED, "true"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "2"},
+ {Options::BUCKET_KEY, "f2"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -754,6 +953,111 @@ TEST_F(PkCompactionInteTest,
DeduplicateWithDeletionVectors) {
}
}
+TEST_F(PkCompactionInteTest, DeduplicateWithBitmap64HighPositions) {
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
+ arrow::field("f2", arrow::int32()), arrow::field("f3",
arrow::float64()),
+ arrow::field("f4", arrow::utf8())};
+ CreateTable(fields, /*partition_keys=*/{"f1"}, /*primary_keys=*/{"f0",
"f1", "f2"},
+ {{Options::FILE_FORMAT, "high_position_parquet"},
+ {Options::BUCKET, "2"},
+ {Options::BUCKET_KEY, "f2"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, "true"},
+ {Options::READ_BATCH_SIZE, "2"}});
+ std::string table_path = TablePath();
+ auto data_type = arrow::struct_(fields);
+ int64_t commit_id = 0;
+ std::string padding(2048, 'X');
+ auto initial = arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([
+ ["Alice", 10, 0, 1.0, ")" + padding + R"("],
+ ["Bob", 10, 0, 2.0, ")" + padding + R"("],
+ ["Carol", 10, 0, 3.0, ")" + padding + R"("],
+ ["Dave", 10, 0, 4.0, ")" + padding + R"("],
+ ["Eve", 10, 0, 5.0, ")" + padding + R"("]
+ ])")
+ .ValueOrDie();
+ ASSERT_OK(WriteAndCommit(table_path, {{"f1", "10"}}, 0, initial,
commit_id++));
+ ASSERT_OK_AND_ASSIGN(auto upgrade_msgs,
+ CompactAndCommit(table_path, {{"f1", "10"}}, 0,
+ /*full_compaction=*/true,
commit_id++));
+ ASSERT_FALSE(HasDeletionVectorIndexFiles(upgrade_msgs));
+
+ // Two small L0 files trigger lookup against the larger, upgraded file.
Bob and Carol
+ // occupy distinct positions above UINT32_MAX, both sharing Alice's low 32
bits (7).
+ for (const auto& update :
+ {R"([["Bob", 10, 0, 102.0, "u1"]])", R"([["Carol", 10, 0, 203.0,
"u2"]])"}) {
+ auto array = arrow::ipc::internal::json::ArrayFromJSON(data_type,
update).ValueOrDie();
+ ASSERT_OK(WriteAndCommit(table_path, {{"f1", "10"}}, 0, array,
commit_id++));
+ }
+ ASSERT_OK_AND_ASSIGN(auto compact_msgs,
+ CompactAndCommit(table_path, {{"f1", "10"}}, 0,
+ /*full_compaction=*/false,
commit_id++));
+ ASSERT_TRUE(HasDeletionVectorIndexFiles(compact_msgs));
+
+ // Reload committed metadata and DV bytes; no test-created DV or lookup
result is injected.
+ ScanContextBuilder scan_builder(table_path);
+ ASSERT_OK_AND_ASSIGN(auto scan_context, scan_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(auto scan,
TableScan::Create(std::move(scan_context)));
+ ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+ int32_t dv_count = 0;
+ for (const auto& split : plan->Splits()) {
+ auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+ ASSERT_TRUE(data_split);
+ auto deletion_files =
DeletionVector::CreateDeletionFileMap(data_split->DataFiles(),
+
data_split->DeletionFiles());
+ auto factory = DeletionVector::CreateFactory(dir_->GetFileSystem(),
deletion_files, pool_);
+ for (const auto& file : data_split->DataFiles()) {
+ if (deletion_files.count(file->file_name) == 0) {
+ continue;
+ }
+ ASSERT_EQ(file->row_count, 5);
+ ASSERT_OK_AND_ASSIGN(auto dv, factory(file->file_name));
+ ASSERT_TRUE(std::dynamic_pointer_cast<Bitmap64DeletionVector>(dv));
+ ASSERT_OK_AND_ASSIGN(int64_t cardinality, dv->GetCardinality());
+ ASSERT_EQ(cardinality, 2);
+ for (const auto& position : {(1LL << 32) + 7, (2LL << 32) + 7}) {
+ ASSERT_OK_AND_ASSIGN(bool deleted, dv->IsDeleted(position));
+ ASSERT_TRUE(deleted);
+ }
+ for (const auto& position : {7LL, (3LL << 32) + 7, (4LL << 32) +
7}) {
+ ASSERT_OK_AND_ASSIGN(bool deleted, dv->IsDeleted(position));
+ ASSERT_FALSE(deleted);
+ }
+ ++dv_count;
+ }
+ }
+ ASSERT_EQ(dv_count, 1);
+
+ std::map<std::pair<std::string, int32_t>, std::string> expected_data;
+ expected_data[{"f1=10/", 0}] = R"([
+ [0, "Alice", 10, 0, 1.0, ")" +
+ padding + R"("],
+ [0, "Dave", 10, 0, 4.0, ")" +
+ padding + R"("],
+ [0, "Eve", 10, 0, 5.0, ")" +
+ padding + R"("],
+ [0, "Bob", 10, 0, 102.0, "u1"],
+ [0, "Carol", 10, 0, 203.0, "u2"]
+ ])";
+ ScanAndVerify(table_path, fields, expected_data);
+
+ ASSERT_OK(CompactAndCommit(table_path, {{"f1", "10"}}, 0,
+ /*full_compaction=*/true, commit_id++));
+ expected_data[{"f1=10/", 0}] = R"([
+ [0, "Alice", 10, 0, 1.0, ")" +
+ padding + R"("],
+ [0, "Bob", 10, 0, 102.0, "u1"],
+ [0, "Carol", 10, 0, 203.0, "u2"],
+ [0, "Dave", 10, 0, 4.0, ")" +
+ padding + R"("],
+ [0, "Eve", 10, 0, 5.0, ")" +
+ padding + R"("]
+ ])";
+ ScanAndVerify(table_path, fields, expected_data);
+}
+
// Test: PK table compact writes output files to external path.
// Verifies that after configuring external paths with round-robin strategy,
// compact output files (compactAfter) have their external_path set, and the
@@ -863,7 +1167,7 @@ TEST_F(PkCompactionInteTest, CompactWithExternalPath) {
// 7. ScanAndVerify after DV compact
// 8. Full compact to merge everything
// 9. ScanAndVerify after full compact
-TEST_F(PkCompactionInteTest, AggMinWithAllNonNestedTypes) {
+TEST_P(PkCompactionDvInteTest, AggMinWithAllNonNestedTypes) {
// f15 is a large padding field to inflate the initial file size for DV
strategy.
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), // PK
arrow::field("f1", arrow::int32()), // PK
@@ -884,13 +1188,15 @@ TEST_F(PkCompactionInteTest,
AggMinWithAllNonNestedTypes) {
std::vector<std::string> primary_keys = {"f0", "f1"};
std::vector<std::string> partition_keys = {};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM,
"local"},
- {Options::MERGE_ENGINE,
"aggregation"},
-
{Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
-
{Options::DELETION_VECTORS_ENABLED, "true"},
-
{"fields.f7.aggregate-function", "bool_and"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"},
+ {"fields.f7.aggregate-function", "bool_and"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -1004,7 +1310,7 @@ TEST_F(PkCompactionInteTest, AggMinWithAllNonNestedTypes)
{
// 7. ScanAndVerify after DV compact
// 8. Full compact
// 9. ScanAndVerify after full compact
-TEST_F(PkCompactionInteTest, AggMinWithPkInMiddle) {
+TEST_P(PkCompactionDvInteTest, AggMinWithPkInMiddle) {
// f4 is a large padding field to inflate the initial file size for DV
strategy.
// PK fields f1(utf8) and f2(int32) are deliberately placed in the middle
of the schema,
// and the PK declaration order (f2, f1) differs from the schema order
(f1, f2).
@@ -1018,9 +1324,13 @@ TEST_F(PkCompactionInteTest, AggMinWithPkInMiddle) {
std::vector<std::string> partition_keys = {};
std::map<std::string, std::string> options = {
- {Options::FILE_FORMAT, "parquet"}, {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM, "local"}, {Options::MERGE_ENGINE,
"aggregation"},
- {Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
{Options::DELETION_VECTORS_ENABLED, "true"}};
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -1128,7 +1438,7 @@ TEST_F(PkCompactionInteTest, AggMinWithPkInMiddle) {
// 7. ScanAndVerify after DV compact
// 8. Full compact
// 9. ScanAndVerify after full compact
-TEST_F(PkCompactionInteTest, DeduplicateWithSequenceFieldAndPkInMiddle) {
+TEST_P(PkCompactionDvInteTest, DeduplicateWithSequenceFieldAndPkInMiddle) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), // value
arrow::field("f1", arrow::utf8()), // PK
@@ -1141,9 +1451,13 @@ TEST_F(PkCompactionInteTest,
DeduplicateWithSequenceFieldAndPkInMiddle) {
std::vector<std::string> partition_keys = {};
std::map<std::string, std::string> options = {
- {Options::FILE_FORMAT, "parquet"}, {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM, "local"}, {Options::MERGE_ENGINE,
"deduplicate"},
- {Options::SEQUENCE_FIELD, "s1,s0"},
{Options::DELETION_VECTORS_ENABLED, "true"}};
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::SEQUENCE_FIELD, "s1,s0"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -1252,7 +1566,7 @@ TEST_F(PkCompactionInteTest,
DeduplicateWithSequenceFieldAndPkInMiddle) {
// 7. ScanAndVerify after DV compact
// 8. Full compact
// 9. ScanAndVerify after full compact
-TEST_F(PkCompactionInteTest, DeduplicateNestedTypesWithSequenceField) {
+TEST_P(PkCompactionDvInteTest, DeduplicateNestedTypesWithSequenceField) {
auto map_type =
std::make_shared<arrow::MapType>(arrow::field("key", arrow::utf8(),
/*nullable=*/false),
arrow::field("value",
arrow::int32()));
@@ -1269,9 +1583,13 @@ TEST_F(PkCompactionInteTest,
DeduplicateNestedTypesWithSequenceField) {
std::vector<std::string> partition_keys = {};
std::map<std::string, std::string> options = {
- {Options::FILE_FORMAT, "parquet"}, {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM, "local"}, {Options::MERGE_ENGINE,
"deduplicate"},
- {Options::SEQUENCE_FIELD, "seq"}, {Options::DELETION_VECTORS_ENABLED,
"true"}};
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::SEQUENCE_FIELD, "seq"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -1503,7 +1821,7 @@ TEST_F(PkCompactionInteTest, CompactWithSchemaEvolution) {
// 6. ScanAndVerify after DV compact
// 7. Full compact → everything merges to level-5 orc
// 8. ScanAndVerify after full compact
-TEST_F(PkCompactionInteTest, FileFormatPerLevelWithDV) {
+TEST_P(PkCompactionDvInteTest, FileFormatPerLevelWithDV) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
arrow::field("f2", arrow::int32()), arrow::field("f3",
arrow::float64()),
@@ -1521,7 +1839,8 @@ TEST_F(PkCompactionInteTest, FileFormatPerLevelWithDV) {
{Options::BUCKET, "2"},
{Options::BUCKET_KEY, "f2"},
{Options::FILE_SYSTEM, "local"},
- {Options::DELETION_VECTORS_ENABLED, "true"}};
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -1778,17 +2097,19 @@ TEST_F(PkCompactionInteTest,
ContinuousWriteWithBackgroundCompact) {
// 4. Non-full compact → lookup against max-level file produces DV.
// 5. ScanAndVerify: "Carol" must be gone, "Bob" must have updated value.
// 6. Full compact → ScanAndVerify again, data must be identical.
-TEST_F(PkCompactionInteTest, DeduplicateWithRowKindAndDV) {
+TEST_P(PkCompactionDvInteTest, DeduplicateWithRowKindAndDV) {
// f3 is a large padding field to make the initial file substantially
bigger.
arrow::FieldVector fields = {
arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
arrow::field("f2", arrow::float64()), arrow::field("f3",
arrow::utf8())};
std::vector<std::string> primary_keys = {"f0"};
std::vector<std::string> partition_keys = {};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM,
"local"},
-
{Options::DELETION_VECTORS_ENABLED, "true"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -2162,7 +2483,7 @@ TEST_F(PkCompactionInteTest, TestFirstRowNoDV) {
}
}
-TEST_F(PkCompactionInteTest, TestPartialUpdateWithDV) {
+TEST_P(PkCompactionDvInteTest, TestPartialUpdateWithDV) {
// f2 and f3 are nullable value fields; PartialUpdate keeps the latest
non-null value
// per field. f4 is a large padding field to inflate the initial file size
so that
// PickForSizeRatio does not merge L0 files directly into Lmax.
@@ -2172,12 +2493,14 @@ TEST_F(PkCompactionInteTest, TestPartialUpdateWithDV) {
arrow::field("f3", arrow::float64(), /*nullable=*/true),
arrow::field("f4", arrow::utf8())};
std::vector<std::string> primary_keys = {"f0", "f1"};
std::vector<std::string> partition_keys = {"f1"};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::BUCKET_KEY, "f0"},
- {Options::FILE_SYSTEM,
"local"},
- {Options::MERGE_ENGINE,
"partial-update"},
-
{Options::DELETION_VECTORS_ENABLED, "true"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "f0"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "partial-update"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -2291,18 +2614,20 @@ TEST_F(PkCompactionInteTest, TestPartialUpdateWithDV) {
}
}
-TEST_F(PkCompactionInteTest, TestDeduplicateWithDvInAllLevels) {
+TEST_P(PkCompactionDvInteTest, TestDeduplicateWithDvInAllLevels) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
arrow::field("f2", arrow::int32()), arrow::field("f3",
arrow::float64()),
arrow::field("f4", arrow::utf8())};
std::vector<std::string> primary_keys = {"f0", "f1", "f2"};
std::vector<std::string> partition_keys = {"f1"};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::BUCKET_KEY, "f2"},
- {Options::FILE_SYSTEM,
"local"},
-
{Options::DELETION_VECTORS_ENABLED, "true"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "f2"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -2726,19 +3051,21 @@ TEST_F(PkCompactionInteTest,
TestAggregateWithNoDvAndOrphanDelete) {
}
}
-TEST_F(PkCompactionInteTest, TestDuplicateWithDvAndOrphanDelete) {
+TEST_P(PkCompactionDvInteTest, TestDuplicateWithDvAndOrphanDelete) {
arrow::FieldVector fields = {
arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
arrow::field("f2", arrow::int32()), arrow::field("f3",
arrow::float64()),
arrow::field("f4", arrow::utf8())};
std::vector<std::string> primary_keys = {"f0", "f1"};
std::vector<std::string> partition_keys = {"f1"};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::BUCKET_KEY, "f0"},
- {Options::FILE_SYSTEM,
"local"},
- {Options::MERGE_ENGINE,
"deduplicate"},
-
{Options::DELETION_VECTORS_ENABLED, "true"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "f0"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "deduplicate"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -2890,7 +3217,7 @@ TEST_F(PkCompactionInteTest,
TestKeyValueTableCompactionWithIOException) {
ASSERT_TRUE(compaction_run_complete);
}
-TEST_P(PkCompactionInteTest, TestKeyValueTableStreamWriteFullCompaction) {
+TEST_P(PkCompactionFormatInteTest, TestKeyValueTableStreamWriteFullCompaction)
{
arrow::FieldVector fields = {
arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()),
arrow::field("f2", arrow::int32()), arrow::field("f3",
arrow::float64())};
@@ -2983,7 +3310,7 @@ TEST_P(PkCompactionInteTest,
TestKeyValueTableStreamWriteFullCompaction) {
// 7. ScanAndVerify after DV compact
// 8. Full compact
// 9. ScanAndVerify after full compact
-TEST_F(PkCompactionInteTest, RemoteLookupFileWithExternalPath) {
+TEST_P(PkCompactionDvInteTest, RemoteLookupFileWithExternalPath) {
// f4 is a large padding field to inflate the initial file size for DV
strategy.
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), // value field (min agg)
@@ -3006,6 +3333,7 @@ TEST_F(PkCompactionInteTest,
RemoteLookupFileWithExternalPath) {
{Options::MERGE_ENGINE, "aggregation"},
{Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
{Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"},
{Options::LOOKUP_REMOTE_FILE_ENABLED, "true"},
{Options::LOOKUP_REMOTE_LEVEL_THRESHOLD, "1"},
{Options::DATA_FILE_EXTERNAL_PATHS, "FILE://" + external_path},
@@ -3121,7 +3449,7 @@ TEST_F(PkCompactionInteTest,
RemoteLookupFileWithExternalPath) {
// 8. ScanAndVerify after DV compact
// 9. Full compact
// 10. ScanAndVerify after full compact
-TEST_F(PkCompactionInteTest, RemoteLookupFileWithSchemaEvolution) {
+TEST_P(PkCompactionDvInteTest, RemoteLookupFileWithSchemaEvolution) {
// f4 is a large padding field to inflate the initial file size for DV
strategy.
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), // value field (min agg)
@@ -3132,14 +3460,16 @@ TEST_F(PkCompactionInteTest,
RemoteLookupFileWithSchemaEvolution) {
std::vector<std::string> primary_keys = {"f2", "f1"};
std::vector<std::string> partition_keys = {};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM,
"local"},
- {Options::MERGE_ENGINE,
"aggregation"},
-
{Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
-
{Options::DELETION_VECTORS_ENABLED, "true"},
-
{Options::LOOKUP_REMOTE_FILE_ENABLED, "true"},
-
{Options::LOOKUP_REMOTE_LEVEL_THRESHOLD, "1"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"},
+ {Options::LOOKUP_REMOTE_FILE_ENABLED, "true"},
+ {Options::LOOKUP_REMOTE_LEVEL_THRESHOLD, "1"}};
CreateTable(fields, partition_keys, primary_keys, options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -3275,8 +3605,8 @@ TEST_F(PkCompactionInteTest,
RemoteLookupFileWithSchemaEvolution) {
// 4. ScanAndVerify after DV compact
// 5. Full compact
// 6. ScanAndVerify after full compact
-TEST_P(PkCompactionInteTest, TestLookupCompatibility) {
- auto file_format = GetParam();
+TEST_P(PkCompactionFormatDvInteTest, TestLookupCompatibility) {
+ auto file_format = std::get<0>(GetParam());
if (file_format == "avro" || file_format == "mosaic" || file_format ==
"lance") {
return;
}
@@ -3292,11 +3622,13 @@ TEST_P(PkCompactionInteTest, TestLookupCompatibility) {
arrow::field("f3", arrow::float64()), // value field (min agg)
arrow::field("f4", arrow::utf8())}; // padding value field (min agg)
+ std::map<std::string, std::string> options = {
+ {Options::DELETION_VECTOR_BITMAP64, std::get<1>(GetParam()) ? "true" :
"false"}};
int64_t commit_id = 6;
// Step 2: Non-full compact → two level-0 files merge, lookup against
max-level produces DV.
ASSERT_OK_AND_ASSIGN(
auto dv_compact_msgs,
- CompactAndCommit(table_path, {}, 0, /*full_compaction=*/false,
commit_id++));
+ CompactAndCommit(table_path, {}, 0, /*full_compaction=*/false,
commit_id++, options));
// Step 2: Assert DV index files are present.
ASSERT_TRUE(HasDeletionVectorIndexFiles(dv_compact_msgs))
@@ -3322,7 +3654,7 @@ TEST_P(PkCompactionInteTest, TestLookupCompatibility) {
// Step 4: Full compact to merge everything.
ASSERT_OK_AND_ASSIGN(
auto full_compact_msgs,
- CompactAndCommit(table_path, {}, 0, /*full_compaction=*/true,
commit_id++));
+ CompactAndCommit(table_path, {}, 0, /*full_compaction=*/true,
commit_id++, options));
ASSERT_TRUE(HasExtraLookupFiles(full_compact_msgs));
// Step 5: ScanAndVerify after full compact.
@@ -3341,7 +3673,7 @@ TEST_P(PkCompactionInteTest, TestLookupCompatibility) {
}
}
-TEST_F(PkCompactionInteTest, PkDvAndAggWithIOException) {
+TEST_P(PkCompactionDvInteTest, PkDvAndAggWithIOException) {
// f4 is a large padding field to inflate the initial file size for DV
strategy.
arrow::FieldVector fields = {
arrow::field("f0", arrow::int32()), // value field (min agg)
@@ -3352,14 +3684,16 @@ TEST_F(PkCompactionInteTest, PkDvAndAggWithIOException)
{
std::vector<std::string> primary_keys = {"f2", "f1"};
std::vector<std::string> partition_keys = {};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM,
"local"},
- {Options::MERGE_ENGINE,
"aggregation"},
-
{Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
-
{Options::DELETION_VECTORS_ENABLED, "true"},
-
{Options::LOOKUP_REMOTE_FILE_ENABLED, "true"},
-
{Options::LOOKUP_REMOTE_LEVEL_THRESHOLD, "1"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {Options::FIELDS_DEFAULT_AGG_FUNC, "min"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"},
+ {Options::LOOKUP_REMOTE_FILE_ENABLED, "true"},
+ {Options::LOOKUP_REMOTE_LEVEL_THRESHOLD, "1"}};
auto data_type = arrow::struct_(fields);
int64_t commit_id = 0;
@@ -3450,7 +3784,7 @@ TEST_F(PkCompactionInteTest, PkDvAndAggWithIOException) {
// End-to-end coverage for the collect / merge_map / nested_update
aggregators: values are merged
// both on read (across level-0 files) and during compaction, then verified
from the written files.
-TEST_F(PkCompactionInteTest, AggCollectMergeMapAndNestedUpdate) {
+TEST_P(PkCompactionDvInteTest, AggCollectMergeMapAndNestedUpdate) {
auto map_type =
std::make_shared<arrow::MapType>(arrow::field("key", arrow::int32(),
/*nullable=*/false),
arrow::field("value",
arrow::int32()));
@@ -3459,16 +3793,18 @@ TEST_F(PkCompactionInteTest,
AggCollectMergeMapAndNestedUpdate) {
arrow::field("f1", arrow::list(arrow::int32())), arrow::field("f2",
map_type),
arrow::field("f3", arrow::list(arrow::struct_({arrow::field("id",
arrow::int32()),
arrow::field("name",
arrow::utf8())})))};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM,
"local"},
- {Options::MERGE_ENGINE,
"aggregation"},
-
{Options::DELETION_VECTORS_ENABLED, "true"},
-
{"fields.f1.aggregate-function", "collect"},
- {"fields.f1.distinct",
"true"},
-
{"fields.f2.aggregate-function", "merge_map"},
-
{"fields.f3.aggregate-function", "nested_update"},
- {"fields.f3.nested-key",
"id"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"},
+ {"fields.f1.aggregate-function", "collect"},
+ {"fields.f1.distinct", "true"},
+ {"fields.f2.aggregate-function", "merge_map"},
+ {"fields.f3.aggregate-function", "nested_update"},
+ {"fields.f3.nested-key", "id"}};
CreateTable(fields, /*partition_keys=*/{}, /*primary_keys=*/{"f0"},
options);
std::string table_path = TablePath();
auto data_type = arrow::struct_(fields);
@@ -3535,17 +3871,19 @@ TEST_F(PkCompactionInteTest,
AggCollectMergeMapAndNestedUpdate) {
// End-to-end coverage for the sketch aggregators. The last batch writes null
sketches, which is the
// path where the aggregator must hand back an owning copy of the accumulator
rather than a view
// into the row buffer that is about to be overwritten.
-TEST_F(PkCompactionInteTest, AggHllAndThetaSketches) {
+TEST_P(PkCompactionDvInteTest, AggHllAndThetaSketches) {
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()), // PK
arrow::field("f1", arrow::binary()),
arrow::field("f2", arrow::binary())};
- std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
"parquet"},
- {Options::BUCKET, "1"},
- {Options::FILE_SYSTEM,
"local"},
- {Options::MERGE_ENGINE,
"aggregation"},
-
{Options::DELETION_VECTORS_ENABLED, "true"},
-
{"fields.f1.aggregate-function", "hll_sketch"},
-
{"fields.f2.aggregate-function", "theta_sketch"}};
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::DELETION_VECTOR_BITMAP64, GetParam() ? "true" : "false"},
+ {"fields.f1.aggregate-function", "hll_sketch"},
+ {"fields.f2.aggregate-function", "theta_sketch"}};
CreateTable(fields, /*partition_keys=*/{}, /*primary_keys=*/{"f0"},
options);
std::string table_path = TablePath();
int64_t commit_id = 0;
@@ -3654,7 +3992,12 @@ std::vector<std::string>
GetTestValuesForCompactionInteTest() {
return values;
}
-INSTANTIATE_TEST_SUITE_P(FileFormat, PkCompactionInteTest,
+INSTANTIATE_TEST_SUITE_P(FileFormat, PkCompactionFormatInteTest,
::testing::ValuesIn(GetTestValuesForCompactionInteTest()));
+INSTANTIATE_TEST_SUITE_P(Bitmap64, PkCompactionDvInteTest, ::testing::Bool());
+INSTANTIATE_TEST_SUITE_P(
+ FileFormatAndBitmap64, PkCompactionFormatDvInteTest,
+
::testing::Combine(::testing::ValuesIn(GetTestValuesForCompactionInteTest()),
+ ::testing::Bool()));
} // namespace paimon::test
