Skip to content

Core: Arrow position delete reader V2 - #16440

Closed
Baunsgaard wants to merge 14 commits into
apache:mainfrom
Baunsgaard:arrow-position-delete-reader
Closed

Baunsgaard wants to merge 14 commits into
apache:mainfrom
Baunsgaard:arrow-position-delete-reader

Conversation

@Baunsgaard

@Baunsgaard Baunsgaard commented May 20, 2026 •

Copy link
Copy Markdown
Contributor

Add VectorizedPositionDeleteReader, an Arrow-vectorized reader for V2 position delete files. Reads (file_path, pos) directly from Arrow VarChar / BigInt buffers and feeds the shared RangeAccumulator from #16052, so consecutive positions become PositionDeleteIndex.delete(start, end) range inserts. No per-row Java allocation on the hot path.

Stacked on #16052, that PR adds the coalescing primitive in iceberg-core; this PR wires it into the Arrow read path.

BaseDeleteLoader.readPosDeletes now dispatches through a new PositionDeleteIndexReader SPI on FormatModelRegistry, so the Arrow path is picked up automatically when iceberg-arrow is on the classpath.

Like #16052, this primarily benefits Iceberg V2 tables; V3 DVs deserialize directly and bypass both paths.

Single-file read (PositionDeleteReaderBenchmark, single-shot)

Mode Distribution Baseline (s) Vectorized (s) Speedup
No filter dense 0.130 0.047 2.77x
No filter sparse 0.147 0.069 2.13x
Filtered by path dense 0.306 0.323 0.95x
Filtered by path sparse 0.338 0.356 0.95x

Filtered cases are bounded by per-row `file_path` comparison cost; both readers pay the same floor there.

Multi-file via BaseDeleteLoader (BaseDeleteLoaderBenchmark, average time)

Distribution v2-legacy (ms) v2-arrow (ms) v3-dv (ms) v2-arrow vs v2-legacy v3-dv vs v2-arrow
dense 152.99 ± 3.72 120.72 ± 20.00 0.012 ± 0.001 1.27x ~10,000x
sparse 163.42 ± 3.31 122.50 ± 2.84 1.025 ± 0.006 1.33x ~120x

V3 DVs (Puffin + RoaringBitmap) are on a different order of magnitude: this PR brings the V2 path faster. Recommended takeaway for reviewers: this is the right reader for tables that remain on V2, not a reason to stay on V2.

Add PositionDeleteRangeConsumer that coalesces runs of consecutive
positions into a single delete(start, end) call on a PositionDeleteIndex,
and route Deletes.toPositionIndex() and PositionDeleteIndex.merge through
it so sorted position delete files are inserted into the bitmap as ranges
instead of one position at a time.

PositionDeleteRangeConsumer is itself the stateful coalescing accumulator
(no inner class). External callers see three operations:

- new PositionDeleteRangeConsumer(target) -- bind to a target index.
- acceptAll(long[] positions, int from, int to) -- the bulk entry point;
  safe to call many times across batches.
- flush() -- emit the pending run.

Internally the consumer runs a sniff window (SNIFF_SIZE positions, drop
to per-position if more than BOUNDARY_THRESHOLD_PERCENT gaps), but those
constants and the per-element accept(long) entry are package-private,
so the public surface is independent of the strategy. The two static
forEach(...) helpers that adapt Iterable<Long> and PositionDeleteIndex
sources to the buffered acceptAll path are also package-private; all
current callers (Deletes, PositionDeleteIndex, BitmapPositionDeleteIndex)
live in the same package.

PositionDeleteIndex.merge (default impl) and BitmapPositionDeleteIndex's
non-bitmap fallback branch now delegate to PositionDeleteRangeConsumer
.forEach(source, target), so any merge of a non-bitmap PositionDeleteIndex
into a bitmap gets the range-coalescing fast path instead of a
per-position delete loop. Sources at or below SMALL_SOURCE_THRESHOLD = 8
positions skip the batched path entirely and delete directly, avoiding
the half-kilobyte buffer allocation for negligible inputs.

Benchmarks (1024-position dense input, BenchmarkPositionDeleteRangeConsumer)
show the bulk acceptAll path is ~12% faster than per-element accept and
~10x lower run-to-run variance. The boxed forEach helper buffers into a
small primitive long[64] -- 512 bytes, comfortably L1-resident -- and
captures the full bulk-path win at 1/16 the allocation cost of larger
drain buffers; the inner-loop throughput of acceptAll is independent of
slice length, so a larger buffer would just allocate more without buying
anything.
Coalesce Deletes.toPositionIndexes by path via a peeking iterator that
streams the run of positions for each file into
PositionDeleteRangeConsumer.forEach. This addresses the sibling code path
raised in PR review: the per-path cached loader was building bitmaps one
position at a time even though the same coalescing logic already exists
for the single-file path.

Drop the constructor null-check and per-call validation Preconditions on
PositionDeleteRangeConsumer. The class surface stays useful with plain
Java semantics (NPE on null target, IOOBE on bad slice) and the JIT
inlines the hot path more aggressively without the extra checks.

Remove the package-private accept(long) method; only the static forEach
helpers and tests used it. Route the equivalent test cases through
acceptAll via a feed() helper, drop the Preconditions-coupled tests, and
rename the interleaved test to acceptAllPersistsStateAcrossMixedSliceSizes
so the intent is explicit.

Add TestDeletesToPositionIndexes covering empty input, single-path runs,
multi-path independence, mixed CharSequence/Utf8 keys, sparse positions,
and out-of-order path revisits.
Baunsgaard added 11 commits May 27, 2026 10:45
Address P1 review findings from the full-review pass:

- Fix BOUNDARY_THRESHOLD_PERCENT Javadoc, which described the threshold
  backwards. Crossing the threshold drops the consumer to the
  per-position fallback, not the bulk path.
- Tighten the class Javadoc so correctness ("any input is correct") is
  cleanly separated from the performance benefit ("scales with monotone
  runs"), and add an @APinote stating this class is a cross-module
  helper consumed by VectorizedPositionDeleteReader in iceberg-arrow.
- Document the UnsupportedOperationException catch in isSmallSource as
  a contract: implementations that don't report cardinality are routed
  through the batched path so bitmap-backed sources never regress.
- Replace the silent if (from >= to) return; in acceptAll with
  Objects.checkFromToIndex so negative / past-end slices fail fast
  instead of being silently dropped.
- Rename Deletes.positionsForPath to drainPositionsForPath and lead its
  Javadoc with the side effect; the returned Iterable consumes the
  peeking iterator on traversal.
- Replace the inline Schema definition in Deletes with
  DeleteSchemaUtil.pathPosSchema(), aligning with every other reader
  and writer in the codebase and removing schema drift risk.
Address P1 and P2 review findings on test coverage:

- Extract a toIndexes(Iterable<StructLike>) helper in
  TestDeletesToPositionIndexes so the six existing call sites stop
  duplicating the CloseableIterable.withNoopClose(...) boilerplate.
- Add deleteFilePropagatesToEachIndex covering the
  toPositionIndexes(source, DeleteFile) overload. The file passed to the
  per-path BitmapPositionDeleteIndex constructor is now asserted to be
  reachable via deleteFiles() on every emitted index.
- Add closeFailureSurfacesAsUncheckedIOException using an inline
  CloseableIterable that throws on close(). Verifies the canonical
  rethrow message and IOException cause.
- Add testMergeNonBitmapSourcePropagatesPositionsAndDeleteFiles in
  TestBitmapPositionDeleteIndex covering the else arm of
  BitmapPositionDeleteIndex.merge(PositionDeleteIndex). Uses an
  anonymous non-bitmap source to exercise the
  PositionDeleteRangeConsumer.forEach path and confirms the source's
  deleteFiles() collection is copied into the target.
Address P1 perf finding from the full-review pass: the cached
multi-file path through toPositionIndexes was boxing every position
into a Long before unboxing it back into the consumer's slice buffer,
costing one autobox per position for sparse / many-path inputs.

- Add a forEach(PrimitiveIterator.OfLong, PositionDeleteIndex) overload
  on PositionDeleteRangeConsumer that drains primitive longs directly
  into the FOREACH_BATCH_SIZE buffer.
- Convert Deletes.drainPositionsForPath to return
  PrimitiveIterator.OfLong so positions stay unboxed all the way from
  the StructLike accessor into the consumer's slice.
- Demote the three static forEach overloads from public to
  package-private. All callers live in org.apache.iceberg.deletes
  (Deletes, BitmapPositionDeleteIndex, the PositionDeleteIndex default
  merge); shrinking the surface matches the @APinote added in the
  previous commit and keeps the buffering policy private.
Add PositionDeleteIndex.create() and create(DeleteFile) so callers
outside the deletes package can build a mutable index without
referencing the package-private BitmapPositionDeleteIndex constructor
or routing through Deletes.toPositionIndex(CloseableIterable<Long>).
Both factories are documented as mutable and not safe for concurrent
mutation, with @see cross-references between empty(), create(), and
create(DeleteFile).

Pairs naturally with the merge-default rewiring earlier in this PR:
external callers that construct a target index now have a clean entry
point that benefits from the range-coalescing merge fast path.

Tests cover the empty mutable index without provenance, a non-null
DeleteFile recorded as provenance, and a null DeleteFile yielding no
provenance.
Introduce a new SPI in FormatModelRegistry so engines that can decode
position-delete files into bitmap indexes more efficiently than the
generic record reader can register a fast path per FileFormat:

- Add the PositionDeleteIndexReader interface with two operations:
  read(file, dataLocation, deleteFile) for a single data-file filter,
  and readAll(file, deleteFile) for the cache path that wants every
  data file's bitmap.
- Add registerPositionDeleteIndexReader(FileFormat, reader) and
  positionDeleteIndexReader(FileFormat) lookup on FormatModelRegistry.
  Registration is once-per-format with the same exclusivity contract
  as existing FormatModel registration; lookup returns Optional so
  callers can fall back when no reader is registered.

This commit only adds the registry capability. Subsequent commits
register an Arrow implementation and have BaseDeleteLoader consult
the registry before opening the per-row record reader.
…te files

Adds a public VectorizedPositionDeleteReader in iceberg-arrow that
materializes a PositionDeleteIndex from a Parquet position-delete file
without going through Record / boxed-Long iteration:
  - reads pos as a primitive long via ArrowBuf.getLong (pos is required
    by the Iceberg position-delete spec, so the per-row validity check
    is intentionally skipped; Javadoc documents the invariant);
  - reads file_path via VarCharVector#getArrowVector so dictionary-
    encoded paths decode once, then compares UTF-8 bytes directly
    against a single byte[] target built per read;
  - coalesces consecutive matching positions into delete(start, end+1)
    range inserts via an internal RangeCoalescer shared by the
    no-filter and filtered paths;
  - projects pos only when no path filter is requested, the typical
    deletion-vector case.

Uses the PositionDeleteIndex.create() / create(DeleteFile) factories
introduced earlier in the stack to construct the mutable index without
referencing the package-private BitmapPositionDeleteIndex constructor.

Tests cover filter-by-path with two data files, dense coalescing on
50k contiguous positions, no-filter mode unioning positions across two
data files, null-input rejection, batch-size equivalence, and
DeleteFile provenance through the reader (7 test methods, 9
invocations including a parameterized invalid-batch-size case).

No new dependencies and no engine wiring: engines opt in by calling
VectorizedPositionDeleteReader.read(...) directly; ArrowReader's
behavior on delete files is unchanged.
Reuse existing primitives and broaden test coverage based on alignment
review:

- Reuse DeleteSchemaUtil.pathPosSchema() instead of re-declaring
  the (file_path, pos) schema locally; this is the same schema used
  by the rest of the delete-file pipeline.
- Replace the hard-coded `setArrowValidityVector= true` with
  NullCheckingForGet.NULL_CHECKING_ENABLED, matching the convention
  in ArrowFormatModels.
- Document why DEFAULT_BATCH_SIZE (8192) diverges from the
  VectorizedArrowReader default (5000): position-delete files
  project at most two narrow columns, so a larger batch amortises
  per-batch decoding cost.
- Make TestVectorizedPositionDeleteReader public to match the rest
  of the iceberg-arrow test suite.
- Add three tests: dictionary-encoded file_path filtering, an
  unknown-path filter that returns an empty index, and a check that
  unreadable input surfaces a RuntimeException with the file
  location preserved (the Parquet stack wraps IOExceptions as
  RuntimeIOException, so the read() catch block fires only on close
  failures).
- Tighten an assertion label in filtersByDataFilePath, drop a
  redundant exists/delete in the test helper, and break a 105-char
  line in honorsExplicitBatchSize.
Wire the zero-copy position-delete reader into the new SPI so delete
loaders can find it through FormatModelRegistry instead of via a hard
iceberg-arrow dependency:

- Add VectorizedPositionDeleteReader.readAllByDataFile(...) which
  decodes an entire delete file into one PositionDeleteIndex per data
  file path. The implementation tracks the active path's bytes and
  reuses the existing range coalescer; it works on both sorted and
  unsorted delete files (the latter by reusing an existing index in
  the result map when a path reappears).
- Have ArrowFormatModels.register() also register an Arrow-backed
  PositionDeleteIndexReader for FileFormat.PARQUET.
- Cover the new entry point with tests for the grouped read,
  contiguous-run coalescing, null input rejection, and a registry
  round-trip that proves ArrowFormatModels.register() wires the SPI
  when iceberg-arrow is on the classpath.
Have BaseDeleteLoader consult FormatModelRegistry for a registered
PositionDeleteIndexReader before opening the per-row record reader.
When iceberg-arrow is on the classpath, parquet position-delete files
go through the Arrow zero-copy reader; when no reader is registered
for the format, the existing generic-record path is used unchanged.

The change is contained to readPosDeletes(...) in both the cached and
uncached branches, so estimateSize and merge semantics are unaffected.

Add a focused test that registers a stub reader and verifies the
loader forwards the data-file path to it, plus a control test that
asserts the unchanged fallback behavior when no reader is registered.
Replace the per-row full byte comparison in appendGroupedByPath with a
length-plus-first-byte fast filter. The common case for position delete
files (sorted by file_path, single data file dominating a batch) now pays
roughly three buffer reads per row instead of one read per byte of the
path, with a single full-path verification at the end of the run.

A divergent middle byte at the same length and first byte (only possible
for unsorted files) is still handled correctly: when the end-of-run
verification disagrees, the method falls back to the per-row full byte
comparison so the grouping stays exact.
Replace the package-local RangeCoalescer with the public
PositionDeleteRangeConsumer primitive introduced in iceberg-core, and
switch each of the three hot loops to the bulk acceptAll(long[], int,
int) entry point so the per-row work stays inside the consumer's tight
loop.

- appendAll: copies each batch's positions into a scratch long[] once
  and bulk-accepts the whole batch.
- appendFiltered: packs the matching positions of each batch into the
  scratch buffer and bulk-accepts per matching run; a non-matching row
  drains the buffer and flushes the consumer so a gap caused by
  another data file never coalesces.
- appendGroupedByPath: bulk-accepts each path run (already bounded by
  endOfPathRun) into the active per-path consumer.

The breakRun() call sites switch to flush() (same semantics: emit the
active run, allow the next accept to start a new one). The scratch
buffer is grown lazily to the largest batch size seen and reused
across batches.
@Baunsgaard
Baunsgaard force-pushed the arrow-position-delete-reader branch from 4017207 to 29eb214 Compare May 27, 2026 21:47
…ract

- Reject null dataLocation in the registered PositionDeleteIndexReader
  adapter; the direct VectorizedPositionDeleteReader#read API still
  accepts null for the union-all case.
- Narrow the grouped read contract to Iceberg v2 (file_path, pos) sort
  order; document undefined behavior on unsorted input.
- Cover the BaseDeleteLoader caching path that dispatches to
  PositionDeleteIndexReader#readAll.
- Mirror error-path tests (non-positive batch size, parquet read
  failures) and add dictionary-encoded and explicit-batch-size grouped
  tests for readAllByDataFile.
- Extract BaseDeleteLoader.fastReaderFor to deduplicate registry lookup.
@github-actions

Copy link
Copy Markdown

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the [email protected] list. Thank you for your contributions.

@github-actions github-actions Bot added the stale label Jun 29, 2026
@Baunsgaard

Copy link
Copy Markdown
Contributor Author

This PR needs further discussion like its parent PR. Therefore closing.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant