Skip to content

Core, Spark: Add clustered DV writer for ordered position delta writes#17287

Open
Neuw84 wants to merge 2 commits into
apache:mainfrom
Neuw84:clustered-dv-writer-main
Open

Core, Spark: Add clustered DV writer for ordered position delta writes#17287
Neuw84 wants to merge 2 commits into
apache:mainfrom
Neuw84:clustered-dv-writer-main

Conversation

@Neuw84

@Neuw84 Neuw84 commented Jul 18, 2026

Copy link
Copy Markdown
Contributor

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.

The claim of this PR is architectural bound: 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 files (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 withidentical sources; re-run on main pending CI).

Measured memory improvement (microbenchmark)

TestDVWriterMemoryBenchmark (included) writes deletes for N data files through both
writers and samples retained heap after forced GC just before close() - the point
where BaseDVFileWriter holds every touched file's position index while the clustered
writer has already flushed all but the last one. 2000 positions per file:

files with deletes BaseDVFileWriter ClusteredDVFileWriter reduction
1,000 4.7 MB 0.5 MB 8.9x
10,000 51.9 MB 5.3 MB 9.9x
50,000 258.5 MB 27.9 MB 9.3x

Base grows ~5.2KB per touched file (bitmap + map entries); clustered retains only the
per-file blob metadata needed to build DeleteFile entries on close (~0.55KB/file).
The ratio grows with positions-per-file (the bitmap term dominates) and the base
writer additionally loads every file's previous DV simultaneously inside close(),
which this measurement does not even include.

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.

The code can be easily backported to Spark 3.4.

AI Disclosure

Model: Claude Fable
Platform/Tool: Kiro
Human Oversight: reviewed

Relates to #17241

…on delta writes

BaseDVFileWriter buffers one in-memory position index per data file the
task deletes from and merges each file's previous deletes only on close,
so its peak resident state is proportional to the number of data files
the task touches. For hot-key MERGE workloads on tables with many files,
that working set grows with the table.

The data path has long offered a bounded alternative for ordered input
(ClusteredDataWriter vs FanoutDataWriter); the DV path had no such
variant. This adds ClusteredDVFileWriter, which requires the incoming
deletes to be clustered by data file: it keeps a single live position
index and, on the first delete of the next file, merges the previous
file's existing deletes, flushes the blob into the shared Puffin file
and releases the bitmap. Peak state becomes one position index plus
small per-file blob metadata, independent of how many files the task
deletes from. The writer fails fast if the input revisits a completed
data file, as that would produce multiple DVs for one file in a commit.

Spark position delta writes request ordering by spec, partition, file
and position when fanout writers are disabled, so SparkPositionDeltaWrite
now selects the clustered writer when the input is ordered and keeps
PartitioningDVWriter for unordered (fanout) input.
… heap just before close while writing deletes for N data files: BaseDVFileWriter holds one position index per file, the clustered writer only per-file blob metadata. At 2000 positions/file the buffered state is 258.5MB vs 27.9MB for 50k files (~9-10x across scales).
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