[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::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;
+}

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);

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);
+}