[
https://issues.apache.org/jira/browse/SPARK-59509?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Qing Fu updated SPARK-59509:
----------------------------
Description:
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.
was:
{{UnsafeSorterSpillMerger.addSpillIfNotEmpty}} reads the incoming iterator's
record 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}
The two {{UnsafeSorterIterator}} implementations that reach this method
disagree about what {{getNumRecords()}} means:
* {{UnsafeSorterSpillReader.getNumRecords()}} returns the count stored in the
spill file header. It never changes, however many times {{loadNext()}} is
called.
* {{UnsafeExternalSorter.SpillableIterator.getNumRecords()}} returns the number
of records _remaining_, and decrements on every {{loadNext()}}.
{{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.
This is latent on master: the merged iterator is consumed with a plain {{while
(hasNext())}} loop, and no caller reads the declared count, so there is no
user-visible symptom today. It is still an incorrect value, and it is a trap
for any future caller that hands a merged iterator to
{{UnsafeSorterSpillWriter}}, which enforces the declared count and fails with
{{IllegalStateException: Number of records written exceeded numRecordsToWrite}}.
The fix is to read the count before {{loadNext()}}.
Reproducer: two spills plus five records still in memory.
{{getSortedIterator().getNumRecords()}} returns 14, and the iterator then
yields 15 records.
Summary: Fix record undercount in UnsafeSorterSpillMerger and document
the getNumRecords() contract (was: Fix record undercount in
UnsafeSorterSpillMerger when merging in-memory records)
> 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
>
> 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]
