Neuw84 opened a new pull request, #17287:
URL: https://github.com/apache/iceberg/pull/17287
## What
Adds `ClusteredDVFileWriter` (core) and `ClusteredDVWriter` (io adapter), a
deletion vector writer for inputs clustered by data file, and wires Spark 3.5 /
4.0 / 4.1 position delta writes to use it when the incoming deletes are ordered
(`fanout-enabled=false` requests ordering by `_spec_id, _partition, _file,
_pos`).
## Why
`BaseDVFileWriter` buffers one in-memory position index per data file the
task deletes from, and at `close()` loads every touched file's previous DV
(blob bytes + deserialized bitmap) and merges them - all resident
simultaneously. Peak state is
O(files touched by the task). For hot-key MERGE workloads on bucketed
tables, a task can touch nearly every data file in its buckets every batch, so
this grows with the table.
The data path has had a bounded variant for ordered input since the
beginning (`ClusteredDataWriter` vs `FanoutDataWriter`), and the v2 position
delete path picks `ClusteredPositionDeleteWriter` when the input is ordered.
The DV path had no such option: `PartitioningDVWriter` is used regardless of
input ordering. This PR closes
that asymmetry.
With clustered input, the writer keeps one live position index. On the first
delete of the next data file it merges the previous file's existing deletes,
flushes the blob into the shared Puffin file and releases the bitmap - peak
resident state becomes a single position index plus small per-file blob
metadata, independent of how many files the task touches. Input that revisits a
completed file fails fast (same contract as `ClusteredWriter`), since that
would produce two DVs for one data file in a single commit.
## Benchmark honesty
We ran a three-arm A/B on EKS (Spark 4.0, Iceberg 1.11 line, 6x8-core
executors, 32g heap + 8g overhead, ~90k rows/s hot-key CDC MERGE into a
bucket(64) MoR mirror,no compaction; details in apache/iceberg#17241):
| arm | writer | fanout | first executor OOM wave |
|---|---|---|---|
| baseline | PartitioningDVWriter | true | t+75min |
| control | PartitioningDVWriter | false | t+90min |
| clustered | ClusteredDVWriter | false | t+93min |
At this scale the end-to-end OOM timing was dominated by an off-heap term
common to both writers (page cache / native buffers from the Puffin
read-modify-write churn), so the clustered writer did NOT measurably delay
container kills - the fanout=false ordering accounted for the improvement.
Throughput at matched batch was identical
across writers.
The claim of this PR is therefore the architectural bound, not an OOM fix:
heap-side
DV writer state becomes O(1 data file) instead of O(files touched) by
construction.
That matters for heap-constrained executors and for tasks that touch many
more files
than ours did (file-heavy tables, partition-scoped tasks, infrequent
compaction),
and it aligns the DV path with the existing clustered/fanout design on the
data path.
## Testing
- New tests in `TestDVWriters` (runs via `TestSparkDVWriters` in all three
Spark versions, 30 tests each, green): basic clustered write across two files,
previous-DV merge marking the old DV rewritten (commit replaces it), and
unclustered-input rejection.
- Existing `TestMergeOnReadDelete` / `TestMergeOnReadMerge` suites pass (the
parameter matrix includes v3 + fanout=false rows that route through the new
writer end to end): 780 tests, 0 failures (run on the 1.11-based branch with
identical sources; re-run on main pending CI).
## Notes for reviewers
- The gate is `context.inputOrdered()` - the same signal the v2 position
delete
path uses to pick `ClusteredPositionDeleteWriter`. Fanout (unordered)
input keeps
the existing `PartitioningDVWriter` behavior; nothing changes for those
plans.
- `ClusteredDVFileWriter` duplicates the blob/DV construction of
`BaseDVFileWriter`
rather than subclassing: the base class's `deleteIndexes` map and
close-time merge
loop are exactly the state this writer exists to avoid, so sharing would
mean
refactoring the base. Happy to extract a shared helper if preferred.
- Related: apache/iceberg#17281 (slim rewritable-deletes broadcast)
addresses the
driver-side term of the same workload; the two are independent.
---
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
