What changes were proposed in this pull request?

UnsafeSorterIterator.getNumRecords() is meant to be a fixed total. SPARK-12295 (2016) 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().

This PR finishes that normalisation by splitting SpillableIterator's counter the same way:

+    private final int numRecords;
+    private int numRecordsRemaining;

     SpillableIterator(UnsafeSorterIterator inMemIterator) {
       this.upstream = inMemIterator;
       this.numRecords = inMemIterator.getNumRecords();
+      this.numRecordsRemaining = this.numRecords;
     }

numRecordsRemaining keeps driving hasNext() and the spill-writer sizing; getNumRecords() returns the fixed numRecords. The field names deliberately mirror UnsafeSorterSpillReader, which already models the same pair of concepts under exactly these names.

All five implementations now agree, and the contract is documented on the abstract method.

Why are the changes needed?

The inconsistency has two consequences.

1. UnsafeSorterSpillMerger under-reports. It 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().

2. getSortedIterator() is incoherent with itself. Its no-spill branch returns a SpillableIterator, whose count falls to zero as it is consumed, while its spilled branch returns a merged iterator reporting a stable total. Same public method, two different meanings.

To be clear about impact: there is none today. 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. Every other read of getNumRecords() is of a fresh, unconsumed iterator. That is why this has gone unnoticed since 2016; it first shipped in 2.0.0.

The reason to fix it now rather than leave it is that the undocumented split contract is actively propagating. The open PR #56804 (SPARK-57714) adds a third merger, UnsafeLoserTreeSpillMerger, with the same loadNext()-then-count ordering, and feeds a SpillableIterator into it; its new tests pass only because they use non-decrementing test iterators.

Fixing SpillableIterator rather than the merger's statement order was deliberate: it is the implementation that diverges, its getNumRecords() has exactly one caller, and fixing it there also resolves consequence 2 and makes the ordering in all three mergers immaterial. The merger is therefore left untouched by this PR.

Two related items are deliberately not included, to keep this focused:

• 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. Happy to file that separately.
• ChainedIterator sums its children's totals, but getIterator(startIndex) advances those children via moveOver() first, so the sum includes skipped records. That is consistent with the documented contract — the total does not track advancement — but the call site's use of it is misleading. Also unread today.

Does this PR introduce any user-facing change?

No. No public API, behaviour or output changes. SpillableIterator.getNumRecords() has a single in-tree caller, and nothing reads the value it ultimately feeds.

How was this patch tested?

Three tests in UnsafeExternalSorterSuite:

• testGetNumRecordsCountsUnspilledRecords — two spills plus five records still in memory, so the merger receives a SpillableIterator; asserts the merged count is 15 and that 15 records are produced.
• testGetNumRecordsIsATotalNotARemainingCount — the no-spill branch, asserting the count does not fall as the iterator is consumed.
• testGetNumRecordsWithoutUnspilledRecords — the all-spilled case, already correct, pinned against regression.

The first two assert spillFilesCreated.size() so they fail rather than silently pass if they stop exercising the branch they target.

Verified each half is load-bearing. Reverting only the SpillableIterator change:

[ERROR] UnsafeExternalSorterSuite.testGetNumRecordsIsATotalNotARemainingCount
        getNumRecords() must not count down ==> expected: <15> but was: <0>

And on master before this change, the merged count was expected: <15> but was: <14>.

UnsafeExternalSorterSuite (29), UnsafeExternalSorterRadixSortSuite (29) and UnsafeInMemorySorterSuite (3) pass, and dev/lint-java is clean.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5)