[VL] Optimize Delta DV applyDeletionFilter with iterator-based bulk lookup

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::RoaringBitmapArray;
+using gluten::delta::DeltaDeletionVectorReader;
+using namespace facebook::velox;
 
 namespace {
 
@@ -276,4 +280,78 @@ BENCHMARK_CAPTURE(
     ->Args({1 << 18, 64})
     ->Unit(benchmark::kMillisecond);
 
+// 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.
+  memory::MemoryManager::testingSetInstance(memory::MemoryManager::Options{});
+  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;
+  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.
+      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() * totalFileRows);
+  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);
+
 BENCHMARK_MAIN();

cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp

@@ -183,15 +183,29 @@ 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.
+  const uint64_t rangeEnd = baseReadOffset + size;
+  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);

cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp

@@ -217,3 +217,168 @@ 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<int64_t> deletedRows;
+  for (int64_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<int64_t> deletedRows = {
+      static_cast<int64_t>(largeBase + 10),
+      static_cast<int64_t>(largeBase + 50),
+      static_cast<int64_t>(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<int64_t> deletedRows;
+  // Dense: rows 0-49
+  for (int64_t i = 0; i < 50; ++i) {
+    deletedRows.push_back(i);
+  }
+  // Sparse: every 100th row from 100-900
+  for (int64_t i = 100; i < 1000; i += 100) {
+    deletedRows.push_back(i);
+  }
+  // Dense: rows 950-999
+  for (int64_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);
+}