feat(data): coalesce position deletes into range inserts

src/iceberg/CMakeLists.txt

@@ -165,6 +165,7 @@ set(ICEBERG_DATA_SOURCES
     data/position_delete_writer.cc
     data/writer.cc
     deletes/position_delete_index.cc
+    deletes/position_delete_range_consumer.cc
     deletes/roaring_position_bitmap.cc
     puffin/file_metadata.cc
     puffin/json_serde.cc

src/iceberg/data/delete_loader.cc

@@ -1,3 +1,3 @@
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -19,14 +19,21 @@
 
 #include "iceberg/data/delete_loader.h"
 
+#include <cstring>
+#include <span>
 #include <string>
 #include <vector>
 
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
 #include "iceberg/arrow_c_data_guard_internal.h"
 #include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/deletes/position_delete_range_consumer.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/metadata_columns.h"
+#include "iceberg/result.h"
 #include "iceberg/row/arrow_array_wrapper.h"
 #include "iceberg/schema.h"
 #include "iceberg/util/macros.h"
@@ -57,6 +64,24 @@
   return ReaderFactoryRegistry::Open(file.file_format, options);
 }
 
+/// Raw `int64` values buffer (offset-adjusted). Skips the validity bitmap:
+/// `kDeleteFilePos` is required by the V2 spec.
+const int64_t* Int64ValuesBuffer(const ArrowArrayView* view) {
+  return view->buffer_views[1].data.as_int64 + view->offset;
+}
+
+/// String-equals at `row_idx` via nanoarrow's unsafe direct-buffer access.
+/// Skips the validity bitmap: `kDeleteFilePath` is required by the V2 spec.
+bool StringEquals(const ArrowArrayView* view, int64_t row_idx,
+                  std::string_view target) {
+  ArrowStringView sv = ArrowArrayViewGetStringUnsafe(view, row_idx);
+  if (static_cast<size_t>(sv.size_bytes) != target.size()) {
+    return false;
+  }
+  return target.empty() ||
+         std::memcmp(sv.data, target.data(), target.size()) == 0;
+}
+
 }  // namespace
 
 DeleteLoader::DeleteLoader(std::shared_ptr<FileIO> io) : io_(std::move(io)) {}
@@ -71,30 +96,60 @@
   ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, reader->Schema());
   internal::ArrowSchemaGuard schema_guard(&arrow_schema);
 
+  // Reused across batches; reads child buffers directly to avoid the
+  // per-row `Scalar` dispatch in `ArrowArrayStructLike`.
+  ArrowArrayView array_view;
+  internal::ArrowArrayViewGuard view_guard(&array_view);
+  ArrowError error;
+  ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+      ArrowArrayViewInitFromSchema(&array_view, &arrow_schema, &error), error);
+
+  // Fast path when the writer's `referenced_data_file` hint matches our
+  // target: skip the path column, hand `pos_data` straight to
+  // `ForEachPositionDelete`. Trusts the hint -- spec-compliant writers
+  // only set it when all rows share one data file.
+  const bool use_referenced_data_file_fast_path =
+      file.referenced_data_file.has_value() &&
+      file.referenced_data_file.value() == data_file_path;
+
+  // Filter-path staging buffer; reused across batches via `clear()`.
+  std::vector<int64_t> positions;
+
   while (true) {
     ICEBERG_ASSIGN_OR_RAISE(auto batch_opt, reader->Next());
     if (!batch_opt.has_value()) break;
 
     auto& batch = batch_opt.value();
     internal::ArrowArrayGuard batch_guard(&batch);
 
-    ICEBERG_ASSIGN_OR_RAISE(
-        auto row, ArrowArrayStructLike::Make(arrow_schema, batch, /*row_index=*/0));
+    ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+        ArrowArrayViewSetArray(&array_view, &batch, &error), error);
 
-    for (int64_t i = 0; i < batch.length; ++i) {
-      if (i > 0) {
-        ICEBERG_RETURN_UNEXPECTED(row->Reset(i));
-      }
-      // Field 0: file_path
-      ICEBERG_ASSIGN_OR_RAISE(auto path_scalar, row->GetField(0));
-      auto path = std::get<std::string_view>(path_scalar);
-
-      if (path == data_file_path) {
-        // Field 1: pos
-        ICEBERG_ASSIGN_OR_RAISE(auto pos_scalar, row->GetField(1));
-        index.Delete(std::get<int64_t>(pos_scalar));
+    const int64_t length = batch.length;
+    if (length <= 0) {
+      continue;
+    }
+
+    // Child indices must match `PosDeleteSchema()`: 0 = file_path, 1 = pos.
+    const ArrowArrayView* pos_view = array_view.children[1];
+    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);
+      continue;
+    }
+
+    const ArrowArrayView* path_view = array_view.children[0];
+    positions.clear();
+    if (positions.capacity() < static_cast<size_t>(length)) {
+      positions.reserve(static_cast<size_t>(length));
+    }
+    for (int64_t i = 0; i < length; ++i) {
+      if (StringEquals(path_view, i, data_file_path)) {
+        positions.push_back(pos_data[i]);
       }
     }
+    ForEachPositionDelete(positions, index);
   }
 
   return reader->Close();

src/iceberg/deletes/position_delete_index.cc

@@ -39,4 +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);
+}
+
 }  // namespace iceberg

src/iceberg/deletes/position_delete_index.h

@@ -24,17 +24,19 @@
 
 #include <cstdint>
 #include <memory>
+#include <span>
 
 #include "iceberg/deletes/roaring_position_bitmap.h"
 #include "iceberg/iceberg_data_export.h"
 
 namespace iceberg {
 
-/// \brief Tracks deleted row positions using a bitmap.
+/// \brief Tracks deleted row positions for an Iceberg MOR data file.
+/// Positions are 0-based row indices.
 ///
-/// This class provides a domain-specific API for position deletes
-/// in Iceberg MOR (merge-on-read) tables. Positions are 0-based
-/// row indices within a data file.
+/// \note Not thread-safe. Callers must externally serialize every access
+///     -- including read-only methods -- whenever any thread might mutate
+///     the instance. Distinct instances are independent.
 class ICEBERG_DATA_EXPORT PositionDeleteIndex {
  public:
   PositionDeleteIndex() = default;
@@ -65,6 +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
+  // `ForEachPositionDelete`'s bulk path; keeps `Delete` the sole public
+  // mutation surface.
+  void BulkAddForKey(int32_t key, const uint32_t* positions, size_t n);
+
+  friend void ForEachPositionDelete(std::span<const int64_t> positions,
+                                    PositionDeleteIndex& target);
+
   RoaringPositionBitmap bitmap_;
 };
 

src/iceberg/deletes/position_delete_range_consumer.cc

@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/deletes/position_delete_range_consumer.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <span>
+#include <vector>
+
+#include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/deletes/roaring_position_bitmap.h"
+
+namespace iceberg {
+
+namespace {
+
+bool IsValidPosition(int64_t pos) {
+  return pos >= 0 && pos <= RoaringPositionBitmap::kMaxPosition;
+}
+
+// Unsigned subtraction so negative or wrap-around input can't
+// false-positive via signed overflow.
+bool IsAdjacent(int64_t prev, int64_t next) {
+  return (static_cast<uint64_t>(next) - static_cast<uint64_t>(prev)) == 1;
+}
+
+// `RoaringPositionBitmap` shards positions by their high 32 bits; the
+// bulk path groups by this key before flushing via `BulkAddForKey`.
+int32_t HighKeyFromPosition(int64_t pos) {
+  return static_cast<int32_t>(pos >> 32);
+}
+
+// Emit `[range_start, last_position]`, collapsing singletons. Callers
+// pre-filter via `IsValidPosition`, so `last_position + 1` cannot overflow.
+void EmitRange(PositionDeleteIndex& target, int64_t range_start,
+               int64_t last_position) {
+  if (range_start == last_position) {
+    target.Delete(range_start);
+  } else {
+    target.Delete(range_start, last_position + 1);
+  }
+}
+
+// Emit closed-interval runs; out-of-range positions are silently skipped
+// to match `Delete(pos)`.
+void CoalesceIntoRanges(std::span<const int64_t> positions,
+                        PositionDeleteIndex& target) {
+  const size_t n = positions.size();
+
+  size_t i = 0;
+  while (i < n && !IsValidPosition(positions[i])) {
+    ++i;
+  }
+  if (i == n) {
+    return;
+  }
+
+  int64_t range_start = positions[i];
+  int64_t last_position = range_start;
+  ++i;
+
+  for (; i < n; ++i) {
+    const int64_t pos = positions[i];
+    if (!IsValidPosition(pos)) {
+      continue;
+    }
+    if (!IsAdjacent(last_position, pos)) {
+      EmitRange(target, range_start, last_position);
+      range_start = pos;
+    }
+    last_position = pos;
+  }
+
+  EmitRange(target, range_start, last_position);
+}
+
+}  // namespace
+
+void ForEachPositionDelete(std::span<const int64_t> positions,
+                           PositionDeleteIndex& target) {
+  if (positions.empty()) {
+    return;
+  }
+
+  // Below this size the bulk path's fixed overhead exceeds coalescing
+  // even on fully scattered input; skip the sniff.
+  constexpr size_t kMinSniffSize = 64;
+  if (positions.size() < kMinSniffSize) {
+    CoalesceIntoRanges(positions, target);
+    return;
+  }
+
+  // Estimate boundary density (fraction of adjacent pairs where
+  // `pos[i] != pos[i-1] + 1`) over a bounded prefix. Misclassification is
+  // performance-only -- both paths produce identical contents.
+  constexpr size_t kSniffSize = 1024;
+  // 10% threshold: boundary-heavy inputs go to bulk addMany; run-heavy
+  // inputs stay on coalesce where Roaring's addRange collapses runs.
+  constexpr size_t kBulkThresholdPercent = 10;
+
+  const size_t sniff = std::min(positions.size(), kSniffSize);
+  size_t boundaries = 0;
+  for (size_t i = 1; i < sniff; ++i) {
+    boundaries += static_cast<size_t>(!IsAdjacent(positions[i - 1], positions[i]));
+  }
+
+  // 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;
+    const size_t n = positions.size();
+    size_t i = 0;
+    while (i < n) {
+      while (i < n && !IsValidPosition(positions[i])) {
+        ++i;
+      }
+      if (i == n) {
+        break;
+      }
+      const int32_t key = HighKeyFromPosition(positions[i]);
+      bulk_key_positions.clear();
+      while (i < n && IsValidPosition(positions[i]) &&
+             HighKeyFromPosition(positions[i]) == key) {
+        bulk_key_positions.push_back(
+            static_cast<uint32_t>(positions[i] & 0xFFFFFFFFu));
+        ++i;
+      }
+      target.BulkAddForKey(key, bulk_key_positions.data(),
+                           bulk_key_positions.size());
+    }
+    return;
+  }
+
+  CoalesceIntoRanges(positions, target);
+}
+
+}  // namespace iceberg

src/iceberg/deletes/position_delete_range_consumer.h

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <span>
+
+#include "iceberg/iceberg_data_export.h"
+
+namespace iceberg {
+
+class PositionDeleteIndex;
+
+/// \brief Apply `positions` to `target` as deletes; semantically equivalent
+/// to calling `target.Delete(pos)` for each entry. Out-of-range positions
+/// are silently ignored. Sorted, mostly-contiguous input is fastest.
+/// Mirrors Java's `PositionDeleteRangeConsumer.forEach`.
+///
+/// \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` (see `PositionDeleteIndex`).
+void ICEBERG_DATA_EXPORT ForEachPositionDelete(std::span<const int64_t> positions,
+                                               PositionDeleteIndex& target);
+
+}  // namespace iceberg

src/iceberg/deletes/roaring_position_bitmap.cc

@@ -105,6 +105,12 @@ void RoaringPositionBitmap::Add(int64_t pos) {
   impl_->bitmaps[key].add(pos32);
 }
 
+void RoaringPositionBitmap::AddManyForKey(int32_t key, const uint32_t* positions,
+                                          size_t n) {
+  impl_->AllocateBitmapsIfNeeded(key + 1);
+  impl_->bitmaps[key].addMany(n, positions);
+}
+
 void RoaringPositionBitmap::AddRange(int64_t pos_start, int64_t pos_end) {
   pos_start = std::max(pos_start, int64_t{0});
   pos_end = std::min(pos_end, kMaxPosition + 1);