markj-db opened a new pull request, #59190:
URL: https://github.com/apache/spark/pull/59190
### What changes were proposed in this pull request?
`org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray` counted
the buffered rows of a
window partition in a 32-bit `Int` (`numRows`), exposed via `def length:
Int`. When a single
window partition exceeds `Int.MaxValue` (2,147,483,647) rows, the counter
overflows and wraps
negative, which silently corrupts the window-frame logic that reads it.
This PR:
- Widens the row counter and its accessor to `Long` (`numRows`/`length` in
`ExternalAppendOnlyUnsafeRowArray`), and widens the offset-frame cursor
`OffsetWindowFunctionFrameBase.inputIndex` to `Long`. This makes
`lag`/`lead`
(`FrameLessOffsetWindowFunctionFrame`) and the unbounded-offset `nth_value`
(`UnboundedOffsetWindowFunctionFrame`) produce **correct** results over
partitions larger than
`Int.MaxValue` rows. These frames ignore the driver row index and track
their cursor as a
`Long`.
- Adds `WindowFunctionFrame.supportsLargePartition` (defaults to `false`;
overridden to `true` on
the two offset frames above). The remaining frames still track the per-row
index and window
bounds in 32-bit `Int` -- including the driver's `rowIndex` in
`WindowEvaluatorFactory` and
`ArrowWindowPythonEvaluatorFactory`, and the bounds in the
sliding/ranking/segment-tree frames
-- so they cannot handle partitions larger than `Int.MaxValue` rows.
- Adds a fail-fast guard
(`WindowEvaluatorFactoryBase.checkPartitionSizeLimit`, called by both
window drivers after buffering a partition) that throws a new user-facing
error
`WINDOW_FUNCTION_PARTITION_SIZE_EXCEEDS_LIMIT` when a partition exceeds
`Int.MaxValue` rows and
any active frame does not support it, instead of silently producing wrong
results (or throwing
an `ArrayIndexOutOfBoundsException`). The 32-bit narrowing sites in those
frames now use
`Math.toIntExact` as a defensive backstop that fails loudly rather than
truncating silently.
Full `Long` propagation for the non-offset frames (frame-bound API,
`AggregateProcessor`, and the
core `UnsafeExternalSorter.getInsertionOrderIterator(int startIndex)`) is a
larger change and is
intentionally left as a follow-up; this PR converts their silent corruption
into a clear error.
This is the read-path counterpart of the write/insertion-path overflow fixed
by
[SPARK-51757](https://issues.apache.org/jira/browse/SPARK-51757), which did
not widen
`numRows`/`length`.
### Why are the changes needed?
Window queries over a single partition larger than `Int.MaxValue` rows can
return silently
incorrect results. The clearest case, confirmed by reproduction, is
`lag`/`lead` returning `NULL`
for every row:
```sql
SELECT count(*) AS n_rows, max(prev) AS max_prev, max(id) AS max_id
FROM (
SELECT id, lag(id, 1) OVER (PARTITION BY CAST(0 AS INT) ORDER BY id) AS
prev
FROM range(2200000000)
);
```
```
n_rows = 2200000000 (all rows funnel to ONE partition via PARTITION BY
a constant)
max_prev = NULL <-- BUG. Correct value is 2199999998
max_id = 2199999999 (sanity: the range itself is intact)
```
`lag(id, 1)` over ids `0..2199999999` must yield `NULL, 0, 1, ...,
2199999998`, so `max(prev)`
must be `2199999998`; it returned `NULL` for every row. For `2^31 < N <
2^32` the wrapped count is
negative and routes to the "offset outside partition -> default" branch; an
`ArrayIndexOutOfBoundsException` is the sibling symptom at larger `N`. A
query should either
produce a correct result or fail with a clear error, never silently return
wrong data.
### Does this PR introduce _any_ user-facing change?
Yes.
- `lag`/`lead` and unbounded-offset `nth_value` over a partition larger than
`Int.MaxValue` rows
now return the correct result. Previously they returned `NULL` (or threw
`ArrayIndexOutOfBoundsException`).
- Other window frames (e.g. sliding, running aggregates, ranking) over such
a partition now fail
with the new error condition
`WINDOW_FUNCTION_PARTITION_SIZE_EXCEEDS_LIMIT` (SQLSTATE `54000`)
instead of silently returning wrong results. Partitions up to and
including `Int.MaxValue` rows
are unaffected.
### How was this patch tested?
- New `WindowFunctionFrameSuite`: covers the `checkPartitionSizeLimit` guard
at the `Int.MaxValue`
boundary (allowed) and above it (rejected for unsafe frames, allowed when
all frames support
large partitions, rejected when any frame does not), the thrown error
condition, and that
`FrameLessOffsetWindowFunctionFrame` declares large-partition support
while the base defaults to
`false`.
- New test in `ExternalAppendOnlyUnsafeRowArraySuite` pinning `length` as a
`Long`.
- Regression: `DataFrameWindowFunctionsSuite` and `WindowSegmentTreeSuite`
pass unchanged.
Materializing a 2^31+ row partition in a unit test is infeasible (the manual
reproduction above
took ~30 minutes over 2.2B rows on a dedicated core), so the overflow
boundary is covered by the
targeted unit tests above rather than an end-to-end run.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac
This pull request and its description were written by Isaac.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
