diff --git a/cpp/velox/benchmarks/DeltaBitmapBenchmark.cc b/cpp/velox/benchmarks/DeltaBitmapBenchmark.cc index b84d103ef9e..ef1bb8af122 100644 --- a/cpp/velox/benchmarks/DeltaBitmapBenchmark.cc +++ b/cpp/velox/benchmarks/DeltaBitmapBenchmark.cc @@ -24,10 +24,14 @@ #include +#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(state.range(0)); + const uint64_t totalFileRows = 1000000; // 1M row file + const auto numDeleted = static_cast(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(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(); + 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 a48060d0606..5c00348273f 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(bytes[2]) << 16) | (static_cast(bytes[3]) << 24); } -uint64_t readUint64LittleEndian(const char* data) { - const auto* bytes = reinterpret_cast(data); - return static_cast(bytes[0]) | (static_cast(bytes[1]) << 8) | - (static_cast(bytes[2]) << 16) | (static_cast(bytes[3]) << 24) | - (static_cast(bytes[4]) << 32) | (static_cast(bytes[5]) << 40) | - (static_cast(bytes[6]) << 48) | (static_cast(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(); 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 f214e427e1a..7d6e03c034f 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& deletedRows) { + std::string createSerializedPayload(const std::vector& deletedRows) { RoaringBitmapArray bitmap; for (auto row : deletedRows) { bitmap.addSafe(row); @@ -184,8 +184,8 @@ TEST_F(DeltaDeletionVectorReaderTest, CardinalityValidationMismatchThrows) { } TEST_F(DeltaDeletionVectorReaderTest, LargeCardinalityValidation) { - std::vector deletedRows; - for (int64_t i = 0; i < 10000; i += 10) { + std::vector 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 deletedRows; - for (int64_t i = 45; i <= 55; ++i) { + std::vector 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 deletedRows; + for (uint64_t i = 0; i < 100; ++i) { + deletedRows.push_back(i); + } + + DeltaDeletionVectorReader reader; + reader.loadSerializedDeletionVector(createSerializedPayload(deletedRows)); + + auto deleteBitmap = AlignedBuffer::allocate(bits::nwords(100), pool_.get()); + reader.applyDeletionFilter(0, 100, deleteBitmap); + + auto* rawBitmap = deleteBitmap->as(); + 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(bits::nwords(100), pool_.get()); + reader.applyDeletionFilter(0, 100, deleteBitmap); + + auto* rawBitmap = deleteBitmap->as(); + 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(bits::nwords(100), pool_.get()); + reader.applyDeletionFilter(100, 100, deleteBitmap); + + auto* rawBitmap = deleteBitmap->as(); + 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(3) << 32; + std::vector deletedRows = {largeBase + 10, largeBase + 50, largeBase + 99}; + + DeltaDeletionVectorReader reader; + reader.loadSerializedDeletionVector(createSerializedPayload(deletedRows)); + + auto deleteBitmap = AlignedBuffer::allocate(bits::nwords(100), pool_.get()); + reader.applyDeletionFilter(largeBase, 100, deleteBitmap); + + auto* rawBitmap = deleteBitmap->as(); + 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(bits::nwords(1), pool_.get()); + reader.applyDeletionFilter(42, 1, deleteBitmap); + + auto* rawBitmap = deleteBitmap->as(); + 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(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 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(bits::nwords(200), pool_.get()); + reader.applyDeletionFilter(100, 200, deleteBitmap); + + auto* rawBitmap = deleteBitmap->as(); + // 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(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(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(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); +}