[
https://issues.apache.org/jira/browse/SPARK-59509?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-59509:
-----------------------------------
Labels: pull-request-available (was: )
> Fix record undercount in UnsafeSorterSpillMerger and document the
> getNumRecords() contract
> ------------------------------------------------------------------------------------------
>
> Key: SPARK-59509
> URL: https://issues.apache.org/jira/browse/SPARK-59509
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 5.0.0
> Reporter: Qing Fu
> Priority: Major
> Labels: pull-request-available
>
> h3. The contract
> {{UnsafeSorterIterator.getNumRecords()}} has two contracts, and the abstract
> method documents neither. Four of the five implementations report the *total*
> number of records and never change it as the iterator is consumed:
> || Implementation || {{getNumRecords()}} returns || Decrements on
> {{loadNext()}}? ||
> | {{UnsafeSorterSpillReader}} | total from the spill file header | no |
> | {{UnsafeInMemorySorter.SortedIterator}} | total | no |
> | {{UnsafeSorterSpillMerger}} (anonymous) | sum of the counts added | no |
> | {{UnsafeExternalSorter.ChainedIterator}} | sum taken at construction | no |
> | {{UnsafeExternalSorter.SpillableIterator}} | records *remaining* | *yes* |
> The two readings agree only while an iterator is untouched.
> h3. The bug
> {{UnsafeSorterSpillMerger.addSpillIfNotEmpty}} reads the count *after*
> calling {{loadNext()}}:
> {code:java}
> public void addSpillIfNotEmpty(UnsafeSorterIterator spillReader) throws
> IOException {
> if (spillReader.hasNext()) {
> spillReader.loadNext();
> priorityQueue.add(spillReader);
> numRecords += spillReader.getNumRecords();
> }
> }
> {code}
> {{UnsafeExternalSorter.getSortedIterator()}} adds its in-memory
> {{SpillableIterator}} to the merger alongside the spill readers, so the
> merged iterator reports one record fewer than it will actually produce, per
> in-memory iterator.
> Reproducer: two spills plus five records still in memory.
> {{getSortedIterator().getNumRecords()}} returns 14, and the iterator then
> yields 15 records.
> h3. Impact
> None today, and this should be stated plainly. A scan of the repository finds
> no caller that reads a merged iterator's {{getNumRecords()}} - the value is
> computed and never consumed. Iteration itself has always been correct,
> because the merged iterator's {{hasNext()}} is driven by the priority queue
> and not by the counter. The other three call sites that do read
> {{getNumRecords()}} all read fresh, unconsumed iterators, so none of them is
> affected.
> That is also why this has gone unnoticed since the line was introduced in
> SPARK-12295 (2016).
> It is still a wrong value, and a trap for the next caller who hands a merged
> iterator to {{UnsafeSorterSpillWriter}}, which enforces the declared count
> and fails with {{IllegalStateException: Number of records written exceeded
> numRecordsToWrite}}.
> h3. Proposed change
> Read the count before {{loadNext()}}, which is correct under either contract,
> and document the contract on {{UnsafeSorterIterator.getNumRecords()}} so the
> ambiguity that produced this does not produce the next one.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
