Core, Spark: Add clustered DV writer for ordered position delta writes#17287
Open
Neuw84 wants to merge 2 commits into
Open
Core, Spark: Add clustered DV writer for ordered position delta writes#17287Neuw84 wants to merge 2 commits into
Neuw84 wants to merge 2 commits into
Conversation
…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.
3 tasks
… 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).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Adds
ClusteredDVFileWriter(core) andClusteredDVWriter(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=falserequests ordering by_spec_id, _partition, _file, _pos).Why
BaseDVFileWriterbuffers one in-memory position index per data file the task deletes from, and atclose()loads every touched file's previous DV (blob bytes + deserialized bitmap) and merges them - all resident simultaneously. Peak state isO(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 (
ClusteredDataWritervsFanoutDataWriter), and the v2 position delete path picksClusteredPositionDeleteWriterwhen the input is ordered. The DV path had no such option:PartitioningDVWriteris used regardless of input ordering. This PR closesthat 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 viaTestSparkDVWritersin 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), andunclustered-input rejection.
Existing
TestMergeOnReadDelete/TestMergeOnReadMergesuites 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 bothwriters and samples retained heap after forced GC just before
close()- the pointwhere
BaseDVFileWriterholds every touched file's position index while the clusteredwriter has already flushed all but the last one. 2000 positions per file:
Base grows ~5.2KB per touched file (bitmap + map entries); clustered retains only the
per-file blob metadata needed to build
DeleteFileentries 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 pickClusteredPositionDeleteWriter. Fanout (unordered) input keeps the existingPartitioningDVWriterbehavior; nothing changes for those plans.ClusteredDVFileWriterduplicates the blob/DV construction ofBaseDVFileWriterrather than subclassing: the base class'sdeleteIndexesmap 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