[
https://issues.apache.org/jira/browse/SPARK-59509?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Wenchen Fan reassigned SPARK-59509:
-----------------------------------
Assignee: Qing Fu
> 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
> Assignee: 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]
