refactor(data): hoist bulk scratch buffer to caller, switch AddManyForKey to span

Address review feedback on PR #645:

- AddManyForKey / BulkAddForKey now take std::span<const uint32_t>
  instead of pointer+length, matching the rest of the PR's style.
- ForEachPositionDelete takes a caller-owned scratch vector instead of
  a thread_local, removing the re-entrancy hazard documented on the
  prior API and giving the caller full control over buffer lifetime.

Author: Baunsgaard

src/iceberg/data/delete_loader.cc

@@ -115,6 +115,9 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde
 
   // Filter-path staging buffer; reused across batches via `clear()`.
   std::vector<int64_t> positions;
+  // Scratch buffer for `ForEachPositionDelete`'s bulk dispatch path;
+  // reused across batches and across both routing branches.
+  std::vector<uint32_t> bulk_scratch;
 
   while (true) {
     ICEBERG_ASSIGN_OR_RAISE(auto batch_opt, reader->Next());
@@ -147,7 +150,8 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde
     const int64_t* pos_data = Int64ValuesBuffer(pos_view);
 
     if (use_referenced_data_file_fast_path) {
-      ForEachPositionDelete(std::span<const int64_t>(pos_data, length), index);
+      ForEachPositionDelete(std::span<const int64_t>(pos_data, length), index,
+                            bulk_scratch);
       continue;
     }
 
@@ -160,7 +164,7 @@ Status DeleteLoader::LoadPositionDelete(const DataFile& file, PositionDeleteInde
         positions.push_back(pos_data[i]);
       }
     }
-    ForEachPositionDelete(positions, index);
+    ForEachPositionDelete(positions, index, bulk_scratch);
   }
 
   return reader->Close();

src/iceberg/deletes/position_delete_index.cc

@@ -39,9 +39,9 @@ void PositionDeleteIndex::Merge(const PositionDeleteIndex& other) {
   bitmap_.Or(other.bitmap_);
 }
 
-void PositionDeleteIndex::BulkAddForKey(int32_t key, const uint32_t* positions,
-                                        size_t n) {
-  bitmap_.AddManyForKey(key, positions, n);
+void PositionDeleteIndex::BulkAddForKey(int32_t key,
+                                        std::span<const uint32_t> positions) {
+  bitmap_.AddManyForKey(key, positions);
 }
 
 }  // namespace iceberg

src/iceberg/deletes/position_delete_index.h

@@ -25,6 +25,7 @@
 #include <cstdint>
 #include <memory>
 #include <span>
+#include <vector>
 
 #include "iceberg/deletes/roaring_position_bitmap.h"
 #include "iceberg/iceberg_data_export.h"
@@ -66,13 +67,14 @@ class ICEBERG_DATA_EXPORT PositionDeleteIndex {
   void Merge(const PositionDeleteIndex& other);
 
  private:
-  // Bulk-add `n` positions sharing high-32-bit `key`. Private hook for
+  // Bulk-add positions sharing high-32-bit `key`. Private hook for
   // `ForEachPositionDelete`'s bulk path; keeps `Delete` the sole public
   // mutation surface.
-  void BulkAddForKey(int32_t key, const uint32_t* positions, size_t n);
+  void BulkAddForKey(int32_t key, std::span<const uint32_t> positions);
 
   friend void ICEBERG_DATA_EXPORT
-  ForEachPositionDelete(std::span<const int64_t> positions, PositionDeleteIndex& target);
+  ForEachPositionDelete(std::span<const int64_t> positions, PositionDeleteIndex& target,
+                        std::vector<uint32_t>& scratch);
 
   RoaringPositionBitmap bitmap_;
 };

src/iceberg/deletes/position_delete_range_consumer.cc

@@ -90,7 +90,7 @@ void CoalesceIntoRanges(std::span<const int64_t> positions, PositionDeleteIndex&
 }  // namespace
 
 void ForEachPositionDelete(std::span<const int64_t> positions,
-                           PositionDeleteIndex& target) {
+                           PositionDeleteIndex& target, std::vector<uint32_t>& scratch) {
   if (positions.empty()) {
     return;
   }
@@ -116,10 +116,8 @@ void ForEachPositionDelete(std::span<const int64_t> positions,
   // boundaries / (sniff - 1) > kBulkThresholdPercent / 100, without FP.
   if (boundaries * 100 > (sniff - 1) * kBulkThresholdPercent) {
     // Bulk path: group by high-32-bit key, flush each group via CRoaring's
-    // `addMany` (through `BulkAddForKey`). The thread-local buffer is
-    // reused across calls; nested invocations on the same thread would
-    // corrupt it -- see `\warning` on `ForEachPositionDelete`.
-    thread_local std::vector<uint32_t> bulk_key_positions;
+    // `addMany` (through `BulkAddForKey`). Reuses the caller-owned `scratch`
+    // vector across key groups -- cleared between groups, capacity retained.
     const size_t n = positions.size();
     size_t i = 0;
     while (i < n) {
@@ -130,13 +128,13 @@ void ForEachPositionDelete(std::span<const int64_t> positions,
         break;
       }
       const int32_t key = HighKeyFromPosition(positions[i]);
-      bulk_key_positions.clear();
+      scratch.clear();
       while (i < n && IsValidPosition(positions[i]) &&
              HighKeyFromPosition(positions[i]) == key) {
-        bulk_key_positions.push_back(static_cast<uint32_t>(positions[i] & 0xFFFFFFFFu));
+        scratch.push_back(static_cast<uint32_t>(positions[i] & 0xFFFFFFFFu));
         ++i;
       }
-      target.BulkAddForKey(key, bulk_key_positions.data(), bulk_key_positions.size());
+      target.BulkAddForKey(key, scratch);
     }
     return;
   }

src/iceberg/deletes/position_delete_range_consumer.h

@@ -21,6 +21,7 @@
 
 #include <cstdint>
 #include <span>
+#include <vector>
 
 #include "iceberg/iceberg_data_export.h"
 
@@ -32,11 +33,12 @@ class PositionDeleteIndex;
 /// to calling `target.Delete(pos)` for each entry. Out-of-range positions
 /// are silently ignored. Sorted, mostly-contiguous input is fastest.
 ///
-/// \warning Not safe to call recursively or interleaved on the same thread:
-///     the bulk dispatch path uses a thread-local staging buffer that a
-///     nested invocation would corrupt. Concurrent calls on different
-///     threads are safe with disjoint `target`.
+/// \param scratch Caller-owned reusable buffer for the bulk dispatch path.
+///     Cleared and reused per key group; retain across calls to amortize
+///     allocations. Pass a distinct `scratch` per thread when calling
+///     concurrently with disjoint `target`.
 void ICEBERG_DATA_EXPORT ForEachPositionDelete(std::span<const int64_t> positions,
-                                               PositionDeleteIndex& target);
+                                               PositionDeleteIndex& target,
+                                               std::vector<uint32_t>& scratch);
 
 }  // namespace iceberg

src/iceberg/deletes/roaring_position_bitmap.cc

@@ -105,10 +105,10 @@ void RoaringPositionBitmap::Add(int64_t pos) {
   impl_->bitmaps[key].add(pos32);
 }
 
-void RoaringPositionBitmap::AddManyForKey(int32_t key, const uint32_t* positions,
-                                          size_t n) {
+void RoaringPositionBitmap::AddManyForKey(int32_t key,
+                                          std::span<const uint32_t> positions) {
   impl_->AllocateBitmapsIfNeeded(key + 1);
-  impl_->bitmaps[key].addMany(n, positions);
+  impl_->bitmaps[key].addMany(positions.size(), positions.data());
 }
 
 void RoaringPositionBitmap::AddRange(int64_t pos_start, int64_t pos_end) {

src/iceberg/deletes/roaring_position_bitmap.h

@@ -25,6 +25,7 @@
 #include <cstdint>
 #include <functional>
 #include <memory>
+#include <span>
 #include <string>
 #include <string_view>
 
@@ -116,7 +117,7 @@ class ICEBERG_DATA_EXPORT RoaringPositionBitmap {
   // Bulk-add positions sharing high-32-bit `key`. Internal hook for
   // `PositionDeleteIndex::BulkAddForKey`; per-key grouping is the caller's
   // job, keeping this a thin wrapper around CRoaring's `addMany`.
-  void AddManyForKey(int32_t key, const uint32_t* positions, size_t n);
+  void AddManyForKey(int32_t key, std::span<const uint32_t> positions);
   friend class PositionDeleteIndex;
 };
 

src/iceberg/test/position_delete_range_consumer_test.cc

@@ -49,7 +49,8 @@ std::set<int64_t> ExpectedValidSet(const std::vector<int64_t>& positions) {
 // Weaker checks would miss divergences at the 32-bit key boundary.
 void AssertMatchesBaseline(const std::vector<int64_t>& positions) {
   PositionDeleteIndex index;
-  ForEachPositionDelete(std::span<const int64_t>(positions), index);
+  std::vector<uint32_t> scratch;
+  ForEachPositionDelete(std::span<const int64_t>(positions), index, scratch);
   const auto expected = ExpectedValidSet(positions);
   ASSERT_EQ(index.Cardinality(), static_cast<int64_t>(expected.size()))
       << "input size=" << positions.size();