[
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()}} is meant to be a fixed total.
SPARK-12295 introduced it and normalised two implementations to that meaning in
the same commit: it gave {{UnsafeSorterSpillReader}} a stable {{numRecords}}
field alongside its existing {{numRecordsRemaining}}, and converted
{{UnsafeInMemorySorter.SortedIterator.numRecordsLeft()}} into a stable
{{getNumRecords()}}. {{UnsafeExternalSorter.SpillableIterator}} was the one
implementation left wired to a field that counts down on every {{loadNext()}}.
|| 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* |
h3. Consequences
*1. The spill merger under-reports.*
{{UnsafeSorterSpillMerger.addSpillIfNotEmpty}} reads each input's count after
calling {{loadNext()}}, so a merged iterator that includes the in-memory
{{SpillableIterator}} reports one record fewer than it will produce, per such
iterator. Two call sites feed it one:
{{UnsafeExternalSorter.getSortedIterator()}}, and since SPARK-56410 the final
round of {{UnsafeSorterBoundedSpillMerger.merge()}}.
Reproducer: two spills plus five records still in memory.
{{getSortedIterator().getNumRecords()}} returns 14, and the iterator then
yields 15 records.
*2. {{getSortedIterator()}} is incoherent with itself.* Its no-spill branch
returns a {{SpillableIterator}}, whose count falls to zero as it is consumed;
its spilled branch returns a merged iterator reporting a stable total. Same
public method, two different meanings.
h3. Impact
None today, and this should be stated plainly. No caller in the repository
reads a merged iterator's {{getNumRecords()}}. Iteration has always been
correct, because the merged iterator's {{hasNext()}} is driven by the priority
queue rather than the counter, and every other read of {{getNumRecords()}} is
of a fresh, unconsumed iterator.
Wrong since {{getNumRecords()}} was introduced in SPARK-12295 (2016-01-06);
first shipped in 2.0.0; never reported. In the ten years since, the only edits
to the relevant lines have been a line-length reflow and a comment typo.
The reason to fix it now rather than leave it is that the undocumented split
contract is actively propagating: the open PR for SPARK-57714 adds a third
merger, {{UnsafeLoserTreeSpillMerger}}, with the same {{loadNext()}}-then-count
ordering, and feeds a {{SpillableIterator}} into it.
h3. Proposed change
Give {{SpillableIterator}} a {{totalRecords}} field for {{getNumRecords()}},
leaving {{numRecords}} to count down for {{hasNext()}} and for sizing the spill
writer. Its {{getNumRecords()}} has exactly one caller, so this is contained;
it completes the normalisation SPARK-12295 started, resolves consequence 2 as
well, and makes the statement ordering in all three mergers immaterial.
Document the contract on the abstract method, including that the total does not
account for records an iterator was advanced past before being handed on.
h3. Related, not included
* {{UnsafeSorterSpillMerger.numRecords}} is an unguarded {{int}} summed across
all spills, so a sort of more than {{Integer.MAX_VALUE}} records reports a
negative total. {{UnsafeSorterBoundedSpillMerger}} already sums in a {{long}}
with an explicit check. Worth a separate ticket.
* {{ChainedIterator}} sums its children's totals, but
{{getIterator(startIndex)}} advances those children via {{moveOver()}} first,
so the sum includes skipped records. Consistent with the documented contract,
but the call site's use of it is misleading. Also unread today.
was:
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.
Summary: Make UnsafeSorterIterator.getNumRecords() a consistent total
(was: Fix record undercount in UnsafeSorterSpillMerger and document the
getNumRecords() contract)
> Make UnsafeSorterIterator.getNumRecords() a consistent total
> ------------------------------------------------------------
>
> 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()}} is meant to be a fixed total.
> SPARK-12295 introduced it and normalised two implementations to that meaning
> in the same commit: it gave {{UnsafeSorterSpillReader}} a stable
> {{numRecords}} field alongside its existing {{numRecordsRemaining}}, and
> converted {{UnsafeInMemorySorter.SortedIterator.numRecordsLeft()}} into a
> stable {{getNumRecords()}}. {{UnsafeExternalSorter.SpillableIterator}} was
> the one implementation left wired to a field that counts down on every
> {{loadNext()}}.
> || 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* |
> h3. Consequences
> *1. The spill merger under-reports.*
> {{UnsafeSorterSpillMerger.addSpillIfNotEmpty}} reads each input's count after
> calling {{loadNext()}}, so a merged iterator that includes the in-memory
> {{SpillableIterator}} reports one record fewer than it will produce, per such
> iterator. Two call sites feed it one:
> {{UnsafeExternalSorter.getSortedIterator()}}, and since SPARK-56410 the final
> round of {{UnsafeSorterBoundedSpillMerger.merge()}}.
> Reproducer: two spills plus five records still in memory.
> {{getSortedIterator().getNumRecords()}} returns 14, and the iterator then
> yields 15 records.
> *2. {{getSortedIterator()}} is incoherent with itself.* Its no-spill branch
> returns a {{SpillableIterator}}, whose count falls to zero as it is consumed;
> its spilled branch returns a merged iterator reporting a stable total. Same
> public method, two different meanings.
> h3. Impact
> None today, and this should be stated plainly. No caller in the repository
> reads a merged iterator's {{getNumRecords()}}. Iteration has always been
> correct, because the merged iterator's {{hasNext()}} is driven by the
> priority queue rather than the counter, and every other read of
> {{getNumRecords()}} is of a fresh, unconsumed iterator.
> Wrong since {{getNumRecords()}} was introduced in SPARK-12295 (2016-01-06);
> first shipped in 2.0.0; never reported. In the ten years since, the only
> edits to the relevant lines have been a line-length reflow and a comment typo.
> The reason to fix it now rather than leave it is that the undocumented split
> contract is actively propagating: the open PR for SPARK-57714 adds a third
> merger, {{UnsafeLoserTreeSpillMerger}}, with the same
> {{loadNext()}}-then-count ordering, and feeds a {{SpillableIterator}} into it.
> h3. Proposed change
> Give {{SpillableIterator}} a {{totalRecords}} field for {{getNumRecords()}},
> leaving {{numRecords}} to count down for {{hasNext()}} and for sizing the
> spill writer. Its {{getNumRecords()}} has exactly one caller, so this is
> contained; it completes the normalisation SPARK-12295 started, resolves
> consequence 2 as well, and makes the statement ordering in all three mergers
> immaterial. Document the contract on the abstract method, including that the
> total does not account for records an iterator was advanced past before being
> handed on.
> h3. Related, not included
> * {{UnsafeSorterSpillMerger.numRecords}} is an unguarded {{int}} summed
> across all spills, so a sort of more than {{Integer.MAX_VALUE}} records
> reports a negative total. {{UnsafeSorterBoundedSpillMerger}} already sums in
> a {{long}} with an explicit check. Worth a separate ticket.
> * {{ChainedIterator}} sums its children's totals, but
> {{getIterator(startIndex)}} advances those children via {{moveOver()}} first,
> so the sum includes skipped records. Consistent with the documented contract,
> but the call site's use of it is misleading. Also unread today.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
