Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 75 additions & 1 deletion cpp/velox/benchmarks/DeltaBitmapBenchmark.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand Down Expand Up @@ -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();
Comment on lines +304 to +305

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch — moved testingSetInstance out of the benchmark body into a custom main() that initializes once before running benchmarks. This mirrors the pattern in GenericBenchmark.cc:642.

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);
Comment on lines +308 to +333

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch on both points.

Tail rows: Fixed — SetItemsProcessed now uses numBatches * batchSize (the rows actually processed) instead of totalFileRows. For the default batch size of 4096, the tail is 576 rows out of 1M (0.06%), so the benchmark results are unaffected, but the counter is now accurate.

Padding bits in popcount: This is actually safe — applyDeletionFilter calls memset(rawBitmap, 0, nbytes(size)) at the top, which zeros the entire bitmap including padding bits in the last word. Added an inline comment clarifying this.

}

// 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;
}
35 changes: 21 additions & 14 deletions cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down Expand Up @@ -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);
Expand Down
193 changes: 188 additions & 5 deletions cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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);
}

Expand All @@ -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);
}

Expand All @@ -217,3 +217,186 @@ 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.
const uint64_t nearMax = UINT64_MAX - 50;
auto payload = createSerializedPayload({nearMax + 10, nearMax + 20});

DeltaDeletionVectorReader reader;
reader.loadSerializedDeletionVector(payload);

// Request a batch of 100 starting at nearMax — this would overflow without
// the saturation guard (nearMax + 100 > UINT64_MAX).
auto deleteBitmap = AlignedBuffer::allocate<uint64_t>(bits::nwords(100), pool_.get());
reader.applyDeletionFilter(nearMax, 100, deleteBitmap);

auto* rawBitmap = deleteBitmap->as<uint64_t>();
// Rows at relative positions 10 and 20 should be marked as deleted.
EXPECT_TRUE(bits::isBitSet(rawBitmap, 10));
EXPECT_TRUE(bits::isBitSet(rawBitmap, 20));
EXPECT_FALSE(bits::isBitSet(rawBitmap, 0));
EXPECT_FALSE(bits::isBitSet(rawBitmap, 30));
}
Comment on lines +383 to +403

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch. Fixed in c566737 — the test now uses low-value bitmap entries ({10, 20, 30}) that are valid under addSafe()'s kMaxRepresentableValue limit, and exercises the overflow guard purely through a high baseReadOffset (UINT64_MAX - 50 with size = 100). This verifies the saturating arithmetic in rangeEnd without needing to bypass the serialization validation.

Constructing a raw Roaring64Map payload with near-UINT64_MAX positions would let us test deletions actually present in the overflow range, but Delta's spec caps row indices well below that, so such a DV would be invalid in practice.

Loading