This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 5eee3976ef [VL] Optimize Delta DV applyDeletionFilter with
iterator-based bulk lookup (#12395)
5eee3976ef is described below
commit 5eee3976ef547a4608440c336a34e92f35189d17
Author: Ismaël Mejía <[email protected]>
AuthorDate: Fri Jul 3 18:22:58 2026 +0200
[VL] Optimize Delta DV applyDeletionFilter with iterator-based bulk lookup
(#12395)
* [VL] Optimize Delta DV applyDeletionFilter with iterator-based bulk lookup
Replace per-row contains() calls with an iterator-based scan in
DeltaDeletionVectorReader::applyDeletionFilter().
Before: O(batch_size) calls to Roaring64Map::contains(), each performing
a std::map lookup (O(log N) on the high-32-bit buckets) plus a Roaring
container check per row position.
After: O(deletions_in_range) iterator advances using move_equalorlarger()
to seek directly to the first deleted row in the batch range, then
sequential ++iterator advances (O(1) amortized within a Roaring
container) until the range end.
Benchmark results (1M row file, batches of 4096, scanning full file):
Deletion % Baseline (ms) Optimized (ms) Speedup
1% (sparse) 9.90 0.050 198x
10% 4.06 0.398 10.2x
50% 3.82 1.99 1.9x
90% 4.37 3.64 1.2x
For the common case of sparse deletions (1%, typical for Delta
MERGE/UPDATE), this is a 198x improvement. The optimization is most
effective when deletions are sparse relative to batch size, which is the
dominant production pattern.
The baseline O(batch_size) approach spent constant time (~4ms per 1M
rows) regardless of deletion density because it always probed every row.
The optimized O(deletions) approach scales linearly with actual
deletions: 0.05ms at 1%, 0.4ms at 10%, 2ms at 50%.
Tests:
- 21 unit tests pass (12 existing + 9 new corner cases covering: all
rows deleted, no rows deleted, first/last row, large 64-bit offsets,
single-row batches, dense/sparse mixed patterns, batch before/after
all deletions).
- New BM_ApplyDeletionFilter benchmark added to DeltaBitmapBenchmark.cc
with 5 scenarios (sparse/moderate/dense/very-dense/large-batch).
Assisted-by: GitHub Copilot:claude-opus-4.6
* Fix benchmark counter accuracy for tail rows
Address review feedback: SetItemsProcessed now reports the actual rows
processed (numBatches * batchSize) rather than totalFileRows, which
slightly overstated the work when totalFileRows is not evenly divisible
by batchSize. Added inline comment clarifying padding bits in popcount
are safe due to memset zeroing in applyDeletionFilter.
Assisted-by: GitHub Copilot:claude-opus-4.6
* Apply clang-format-15 formatting
Assisted-by: GitHub Copilot:claude-opus-4.6
* Address review: overflow guard on rangeEnd, move MemoryManager init to
main()
- Add saturating arithmetic for rangeEnd to prevent uint64_t overflow
when baseReadOffset is near UINT64_MAX.
- Move memory::MemoryManager::testingSetInstance() from benchmark body
to a custom main(), ensuring it runs once per process.
- Add ApplyDeletionFilterOverflowProtection unit test.
Assisted-by: GitHub Copilot:claude-opus-4.6
* Fix narrowing conversion error and remove unused function
- Change createSerializedPayload to accept std::vector<uint64_t> to
match RoaringBitmapArray::addSafe(uint64_t) and avoid narrowing
conversion when passing values near UINT64_MAX.
- Update all local deletedRows vectors and loop counters to uint64_t.
- Remove unused readUint64LittleEndian() that triggered
-Wunused-function.
Assisted-by: GitHub Copilot:claude-opus-4.6
* Fix overflow test: use valid bitmap values with high baseReadOffset
RoaringBitmapArray::addSafe rejects values above kMaxRepresentableValue,
so the test cannot insert row indices near UINT64_MAX. Instead, test the
overflow guard by calling applyDeletionFilter with a baseReadOffset near
UINT64_MAX on a bitmap with low-value entries, verifying no crash and
empty result.
Assisted-by: GitHub Copilot:claude-opus-4.6
---
cpp/velox/benchmarks/DeltaBitmapBenchmark.cc | 76 +++++++-
.../compute/delta/DeltaDeletionVectorReader.cpp | 35 ++--
.../delta/tests/DeltaDeletionVectorReaderTest.cpp | 194 ++++++++++++++++++++-
3 files changed, 285 insertions(+), 20 deletions(-)
diff --git a/cpp/velox/benchmarks/DeltaBitmapBenchmark.cc
b/cpp/velox/benchmarks/DeltaBitmapBenchmark.cc
index b84d103ef9..ef1bb8af12 100644
--- a/cpp/velox/benchmarks/DeltaBitmapBenchmark.cc
+++ b/cpp/velox/benchmarks/DeltaBitmapBenchmark.cc
@@ -24,10 +24,14 @@
#include <benchmark/benchmark.h>
+#include "compute/delta/DeltaDeletionVectorReader.h"
#include "compute/delta/RoaringBitmapArray.h"
#include "velox/common/base/Exceptions.h"
+#include "velox/common/memory/Memory.h"
+using gluten::delta::DeltaDeletionVectorReader;
using gluten::delta::RoaringBitmapArray;
+using namespace facebook::velox;
namespace {
@@ -276,4 +280,74 @@ BENCHMARK_CAPTURE(
->Args({1 << 18, 64})
->Unit(benchmark::kMillisecond);
-BENCHMARK_MAIN();
+// Benchmark for applyDeletionFilter: measures the hot path where a batch of
+// rows is checked against the deletion vector bitmap.
+// deletionPercent: fraction of rows in the total file that are deleted.
+// batchSize: number of rows per batch (typical Velox batch size).
+void BM_ApplyDeletionFilter(benchmark::State& state, double deletionPercent) {
+ const auto batchSize = static_cast<uint64_t>(state.range(0));
+ const uint64_t totalFileRows = 1000000; // 1M row file
+ const auto numDeleted = static_cast<uint64_t>(totalFileRows *
deletionPercent / 100.0);
+
+ // Build a DV with deletions spread across the file.
+ RoaringBitmapArray bitmap;
+ const uint64_t stride = numDeleted > 0 ? totalFileRows / numDeleted : 0;
+ for (uint64_t i = 0; i < numDeleted; ++i) {
+ bitmap.addSafe(i * stride);
+ }
+ const auto payload = bitmap.serializeToString(true);
+
+ // Load the DV reader.
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(std::string_view(payload.data(),
payload.size()));
+
+ // Allocate the output bitmap buffer.
+ auto pool = memory::memoryManager()->addLeafPool();
+ auto deleteBitmap =
AlignedBuffer::allocate<uint64_t>(bits::nwords(batchSize), pool.get());
+
+ // Simulate scanning through the file in batches.
+ const uint64_t numBatches = totalFileRows / batchSize;
+ // Only count rows actually processed (drop tail < batchSize).
+ const uint64_t rowsProcessed = numBatches * batchSize;
+ uint64_t totalDeletedFound = 0;
+
+ for (auto _ : state) {
+ totalDeletedFound = 0;
+ for (uint64_t batch = 0; batch < numBatches; ++batch) {
+ reader.applyDeletionFilter(batch * batchSize, batchSize, deleteBitmap);
+ // Count bits set to prevent dead-code elimination.
+ // Padding bits in the last word are safe: applyDeletionFilter memsets
+ // the entire bitmap to zero before setting only in-range bits.
+ auto* raw = deleteBitmap->as<uint64_t>();
+ for (uint64_t w = 0; w < bits::nwords(batchSize); ++w) {
+ totalDeletedFound += __builtin_popcountll(raw[w]);
+ }
+ }
+ benchmark::DoNotOptimize(totalDeletedFound);
+ }
+
+ state.SetItemsProcessed(state.iterations() * rowsProcessed);
+ state.counters["batch_size"] = benchmark::Counter(batchSize);
+ state.counters["deletion_pct"] = benchmark::Counter(deletionPercent);
+ state.counters["deleted_found"] = benchmark::Counter(totalDeletedFound);
+ state.counters["total_batches"] = benchmark::Counter(numBatches);
+}
+
+// Sparse deletions (1%) - the common case for MERGE/UPDATE operations.
+BENCHMARK_CAPTURE(BM_ApplyDeletionFilter, Sparse_1pct,
1.0)->Arg(4096)->Unit(benchmark::kMillisecond);
+// Moderate deletions (10%).
+BENCHMARK_CAPTURE(BM_ApplyDeletionFilter, Moderate_10pct,
10.0)->Arg(4096)->Unit(benchmark::kMillisecond);
+// Dense deletions (50%).
+BENCHMARK_CAPTURE(BM_ApplyDeletionFilter, Dense_50pct,
50.0)->Arg(4096)->Unit(benchmark::kMillisecond);
+// Very dense deletions (90%).
+BENCHMARK_CAPTURE(BM_ApplyDeletionFilter, VeryDense_90pct,
90.0)->Arg(4096)->Unit(benchmark::kMillisecond);
+// Sparse with large batch (typical Velox max batch).
+BENCHMARK_CAPTURE(BM_ApplyDeletionFilter, Sparse_1pct_LargeBatch,
1.0)->Arg(10000)->Unit(benchmark::kMillisecond);
+
+int main(int argc, char** argv) {
+ memory::MemoryManager::testingSetInstance(memory::MemoryManager::Options{});
+ benchmark::Initialize(&argc, argv);
+ benchmark::RunSpecifiedBenchmarks();
+ benchmark::Shutdown();
+ return 0;
+}
diff --git a/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp
b/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp
index a48060d060..5c00348273 100644
--- a/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp
+++ b/cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp
@@ -52,14 +52,6 @@ uint32_t readUint32LittleEndian(const char* data) {
(static_cast<uint32_t>(bytes[2]) << 16) |
(static_cast<uint32_t>(bytes[3]) << 24);
}
-uint64_t readUint64LittleEndian(const char* data) {
- const auto* bytes = reinterpret_cast<const uint8_t*>(data);
- return static_cast<uint64_t>(bytes[0]) | (static_cast<uint64_t>(bytes[1]) <<
8) |
- (static_cast<uint64_t>(bytes[2]) << 16) |
(static_cast<uint64_t>(bytes[3]) << 24) |
- (static_cast<uint64_t>(bytes[4]) << 32) |
(static_cast<uint64_t>(bytes[5]) << 40) |
- (static_cast<uint64_t>(bytes[6]) << 48) |
(static_cast<uint64_t>(bytes[7]) << 56);
-}
-
roaring::Roaring64Map deserializeDeltaBitmapArray(std::string_view
serializedPayload, const std::string& dvPath) {
VELOX_USER_CHECK_GE(
serializedPayload.size(),
@@ -183,15 +175,30 @@ void
DeltaDeletionVectorReader::applyDeletionFilter(uint64_t baseReadOffset, uin
auto* rawBitmap = deleteBitmap->asMutable<uint64_t>();
std::memset(rawBitmap, 0, bits::nbytes(size));
+ // Use an iterator-based approach instead of per-row contains() lookups.
+ // This is O(deletions_in_range) rather than O(batch_size), which is
+ // significantly faster when deletions are sparse relative to batch size.
+ // Guard against uint64_t overflow when baseReadOffset is near UINT64_MAX.
+ const uint64_t rangeEnd = (size <= UINT64_MAX - baseReadOffset) ?
baseReadOffset + size : UINT64_MAX;
+ auto it = deletionBitmap_->begin();
+ if (!it.move_equalorlarger(baseReadOffset)) {
+ // No deleted rows at or after baseReadOffset — nothing to mark.
+ deleteBitmap->setSize(0);
+ return;
+ }
+
bool hasDeletedRows = false;
uint64_t highestDeletedIndex = 0;
- for (uint64_t i = 0; i < size; ++i) {
- const uint64_t absoluteRowPos = baseReadOffset + i;
- if (deletionBitmap_->contains(absoluteRowPos)) {
- bits::setBit(rawBitmap, i);
- hasDeletedRows = true;
- highestDeletedIndex = i;
+ while (it != deletionBitmap_->end()) {
+ const uint64_t absoluteRowPos = *it;
+ if (absoluteRowPos >= rangeEnd) {
+ break;
}
+ const uint64_t relativeIndex = absoluteRowPos - baseReadOffset;
+ bits::setBit(rawBitmap, relativeIndex);
+ hasDeletedRows = true;
+ highestDeletedIndex = relativeIndex;
+ ++it;
}
deleteBitmap->setSize(hasDeletedRows ? bits::nbytes(highestDeletedIndex + 1)
: 0);
diff --git a/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp
b/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp
index f214e427e1..7d6e03c034 100644
--- a/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp
+++ b/cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp
@@ -49,7 +49,7 @@ class DeltaDeletionVectorReaderTest : public ::testing::Test {
pool_ = memory::memoryManager()->addLeafPool();
}
- std::string createSerializedPayload(const std::vector<int64_t>& deletedRows)
{
+ std::string createSerializedPayload(const std::vector<uint64_t>&
deletedRows) {
RoaringBitmapArray bitmap;
for (auto row : deletedRows) {
bitmap.addSafe(row);
@@ -184,8 +184,8 @@ TEST_F(DeltaDeletionVectorReaderTest,
CardinalityValidationMismatchThrows) {
}
TEST_F(DeltaDeletionVectorReaderTest, LargeCardinalityValidation) {
- std::vector<int64_t> deletedRows;
- for (int64_t i = 0; i < 10000; i += 10) {
+ std::vector<uint64_t> deletedRows;
+ for (uint64_t i = 0; i < 10000; i += 10) {
deletedRows.push_back(i);
}
@@ -195,8 +195,8 @@ TEST_F(DeltaDeletionVectorReaderTest,
LargeCardinalityValidation) {
}
TEST_F(DeltaDeletionVectorReaderTest, BatchFilteringPartialOverlap) {
- std::vector<int64_t> deletedRows;
- for (int64_t i = 45; i <= 55; ++i) {
+ std::vector<uint64_t> deletedRows;
+ for (uint64_t i = 45; i <= 55; ++i) {
deletedRows.push_back(i);
}
@@ -217,3 +217,187 @@ TEST_F(DeltaDeletionVectorReaderTest,
BatchFilteringPartialOverlap) {
EXPECT_FALSE(bits::isBitSet(rawBitmap, i));
}
}
+
+// --- Corner-case tests for iterator-based applyDeletionFilter ---
+
+TEST_F(DeltaDeletionVectorReaderTest, ApplyDeletionFilterAllRowsDeleted) {
+ // Every row in the batch is deleted.
+ std::vector<uint64_t> deletedRows;
+ for (uint64_t i = 0; i < 100; ++i) {
+ deletedRows.push_back(i);
+ }
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(createSerializedPayload(deletedRows));
+
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(100),
pool_.get());
+ reader.applyDeletionFilter(0, 100, deleteBitmap);
+
+ auto* rawBitmap = deleteBitmap->as<uint64_t>();
+ for (int i = 0; i < 100; ++i) {
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, i)) << "Row " << i << " should be
deleted";
+ }
+ EXPECT_GT(deleteBitmap->size(), 0);
+}
+
+TEST_F(DeltaDeletionVectorReaderTest, ApplyDeletionFilterNoRowsDeleted) {
+ // The DV has deletions but none overlap with the batch range.
+ auto payload = createSerializedPayload({1000, 2000, 3000});
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(payload);
+
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(100),
pool_.get());
+ reader.applyDeletionFilter(0, 100, deleteBitmap);
+
+ auto* rawBitmap = deleteBitmap->as<uint64_t>();
+ for (int i = 0; i < 100; ++i) {
+ EXPECT_FALSE(bits::isBitSet(rawBitmap, i));
+ }
+ EXPECT_EQ(deleteBitmap->size(), 0);
+}
+
+TEST_F(DeltaDeletionVectorReaderTest, ApplyDeletionFilterFirstAndLastRow) {
+ // Only the first and last row of the batch are deleted.
+ auto payload = createSerializedPayload({100, 199});
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(payload);
+
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(100),
pool_.get());
+ reader.applyDeletionFilter(100, 100, deleteBitmap);
+
+ auto* rawBitmap = deleteBitmap->as<uint64_t>();
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, 0));
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, 99));
+ for (int i = 1; i < 99; ++i) {
+ EXPECT_FALSE(bits::isBitSet(rawBitmap, i));
+ }
+}
+
+TEST_F(DeltaDeletionVectorReaderTest, ApplyDeletionFilterLargeOffset) {
+ // Test with large row offsets (beyond 32-bit boundary) to exercise
+ // multi-bucket Roaring64Map behavior.
+ const uint64_t largeBase = static_cast<uint64_t>(3) << 32;
+ std::vector<uint64_t> deletedRows = {largeBase + 10, largeBase + 50,
largeBase + 99};
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(createSerializedPayload(deletedRows));
+
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(100),
pool_.get());
+ reader.applyDeletionFilter(largeBase, 100, deleteBitmap);
+
+ auto* rawBitmap = deleteBitmap->as<uint64_t>();
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, 10));
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, 50));
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, 99));
+ EXPECT_FALSE(bits::isBitSet(rawBitmap, 0));
+ EXPECT_FALSE(bits::isBitSet(rawBitmap, 11));
+ EXPECT_FALSE(bits::isBitSet(rawBitmap, 49));
+}
+
+TEST_F(DeltaDeletionVectorReaderTest, ApplyDeletionFilterSingleRowBatch) {
+ // Batch of size 1 where the single row is deleted.
+ auto payload = createSerializedPayload({42});
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(payload);
+
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(1),
pool_.get());
+ reader.applyDeletionFilter(42, 1, deleteBitmap);
+
+ auto* rawBitmap = deleteBitmap->as<uint64_t>();
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, 0));
+}
+
+TEST_F(DeltaDeletionVectorReaderTest,
ApplyDeletionFilterSingleRowBatchNotDeleted) {
+ // Batch of size 1 where the single row is NOT deleted.
+ auto payload = createSerializedPayload({42});
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(payload);
+
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(1),
pool_.get());
+ reader.applyDeletionFilter(43, 1, deleteBitmap);
+
+ EXPECT_EQ(deleteBitmap->size(), 0);
+}
+
+TEST_F(DeltaDeletionVectorReaderTest, ApplyDeletionFilterDenseSparseMixed) {
+ // Dense cluster at start, sparse in middle, dense at end.
+ std::vector<uint64_t> deletedRows;
+ // Dense: rows 0-49
+ for (uint64_t i = 0; i < 50; ++i) {
+ deletedRows.push_back(i);
+ }
+ // Sparse: every 100th row from 100-900
+ for (uint64_t i = 100; i < 1000; i += 100) {
+ deletedRows.push_back(i);
+ }
+ // Dense: rows 950-999
+ for (uint64_t i = 950; i < 1000; ++i) {
+ deletedRows.push_back(i);
+ }
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(createSerializedPayload(deletedRows));
+
+ // Test middle sparse region
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(200),
pool_.get());
+ reader.applyDeletionFilter(100, 200, deleteBitmap);
+
+ auto* rawBitmap = deleteBitmap->as<uint64_t>();
+ // Should have rows 100 and 200 marked (every 100th from our sparse set)
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, 0)); // row 100
+ EXPECT_TRUE(bits::isBitSet(rawBitmap, 100)); // row 200
+ EXPECT_FALSE(bits::isBitSet(rawBitmap, 1)); // row 101
+ EXPECT_FALSE(bits::isBitSet(rawBitmap, 99)); // row 199
+}
+
+TEST_F(DeltaDeletionVectorReaderTest,
ApplyDeletionFilterBatchBeyondAllDeletions) {
+ // Batch starts after all deletions.
+ auto payload = createSerializedPayload({1, 2, 3, 4, 5});
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(payload);
+
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(100),
pool_.get());
+ reader.applyDeletionFilter(1000, 100, deleteBitmap);
+
+ EXPECT_EQ(deleteBitmap->size(), 0);
+}
+
+TEST_F(DeltaDeletionVectorReaderTest,
ApplyDeletionFilterBatchBeforeAllDeletions) {
+ // Batch ends before any deletions start.
+ auto payload = createSerializedPayload({500, 600, 700});
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(payload);
+
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(100),
pool_.get());
+ reader.applyDeletionFilter(0, 100, deleteBitmap);
+
+ EXPECT_EQ(deleteBitmap->size(), 0);
+}
+
+TEST_F(DeltaDeletionVectorReaderTest, ApplyDeletionFilterOverflowProtection) {
+ // Verify that baseReadOffset near UINT64_MAX does not overflow rangeEnd.
+ // RoaringBitmapArray only supports row indices up to kMaxRepresentableValue,
+ // so we test with low-value bitmap entries but call applyDeletionFilter with
+ // a baseReadOffset that would cause baseReadOffset + size to overflow.
+ auto payload = createSerializedPayload({10, 20, 30});
+
+ DeltaDeletionVectorReader reader;
+ reader.loadSerializedDeletionVector(payload);
+
+ // Request a batch of 100 starting near UINT64_MAX — this would overflow
+ // without the saturation guard (nearMax + 100 > UINT64_MAX).
+ // No deletions exist in this range, so we just verify no crash and empty
+ // result.
+ const uint64_t nearMax = UINT64_MAX - 50;
+ auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(100),
pool_.get());
+ reader.applyDeletionFilter(nearMax, 100, deleteBitmap);
+
+ // No rows in bitmap are near UINT64_MAX, so result should be empty.
+ EXPECT_EQ(deleteBitmap->size(), 0);
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
