[jira] [Created] (SPARK-59941) Widen UnsafeExternalSorter row accounting to Long for window partitions larger than Int.MaxValue rows

["Mark Jarvin (Jira)" <ji...@apache.org>]

Mark Jarvin created SPARK-59941:
-----------------------------------

             Summary: Widen UnsafeExternalSorter row accounting to Long for window partitions larger than Int.MaxValue rows
                 Key: SPARK-59941
                 URL: https://issues.apache.org/jira/browse/SPARK-59941
             Project: Spark
          Issue Type: Improvement
          Components: Spark Core
    Affects Versions: 5.0.0
            Reporter: Mark Jarvin


h2. Summary

{{org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter}} accounts for buffered rows
in 32-bit {{int}}s on the read path:

* {{UnsafeSorterSpillWriter.recordsSpilled()}} returns {{int}}.
* {{UnsafeSorterIterator.getNumRecords()}} returns {{int}}.
* {{UnsafeInMemorySorter.numRecords()}} returns {{int}}.
* {{getIterator(int startIndex)}} accumulates spilled counts into an {{int i}} and skips a spill
  file unless {{i + spillWriter.recordsSpilled() > startIndex}}; {{ChainedIterator.numRecords}} is
  also {{int}}.

When a single buffered partition exceeds {{Int.MaxValue}} rows and has spilled, the running {{int}}
total overflows negative. Later spill files are then skipped (rows silently dropped), or
{{moveOver}} throws a raw {{ArrayIndexOutOfBoundsException}}. This affects window execution even on
the {{getIterator(0)}} path used by LEAD/LAG and unbounded-offset {{nth_value}}.

h2. Context

Follow-up to SPARK-59929, which widened the window row counter
({{ExternalAppendOnlyUnsafeRowArray.numRows}}/{{length}}) to {{Long}} and added a fail-fast guard
({{WINDOW_FUNCTION_PARTITION_SIZE_EXCEEDS_LIMIT}}) that currently rejects *every* window frame over
a partition larger than {{Int.MaxValue}} rows. Those frames cannot be made correct over such
partitions until this sorter accounting is widened.

h2. Proposed work

Widen the sorter's row accounting to {{long}} (the {{getIterator}} accumulator and skip condition,
{{ChainedIterator.numRecords}}, and the {{recordsSpilled()}} / {{getNumRecords()}} /
{{numRecords()}} surface and their consumers), then relax the window guard so the offset frames
(LEAD/LAG, unbounded {{nth_value}}) produce correct results for partitions larger than
{{Int.MaxValue}} rows instead of failing.

Related: SPARK-59929, SPARK-51757.




--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org