Mark Jarvin created SPARK-59929:
-----------------------------------
Summary: lag/lead return NULL and other window frames silently
corrupt results for partitions larger than Int.MaxValue rows
Key: SPARK-59929
URL: https://issues.apache.org/jira/browse/SPARK-59929
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Mark Jarvin
h2. Summary
{{org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray}} counts the
buffered rows
of a window partition in a 32-bit {{Int}} ({{numRows}}). When a single window
partition exceeds
{{Int.MaxValue}} (2,147,483,647) rows, {{numRows}} overflows and wraps
negative. Downstream
window-frame logic that reads this count (and the per-row index/bounds, also
tracked as 32-bit
{{Int}}) then misbehaves, producing *silently wrong results* with no exception.
This is the read-path counterpart of the write/insertion-path overflow fixed by
SPARK-51757,
which did not widen {{numRows}}/{{length}}.
h2. Symptom
For {{lag}}/{{lead}}, the offset frame ({{FrameLessOffsetWindowFunctionFrame}})
guards with
{{absOffset > rows.length}}. With a negative (wrapped) {{length}}, the guard
misfires and the
frame emits the function's default value (NULL) for *every* row.
Confirmed reproduction (classic non-Photon {{WindowExec}}; ~30 min over 2.2B
rows):
{code: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)
);
{code}
Observed vs. correct:
{code}
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)
{code}
{{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, routing to the "offset
outside
partition -> default" branch (silent NULLs). An
{{ArrayIndexOutOfBoundsException}} is the sibling
symptom when the wrapped count/index lands at an out-of-bounds value (needs
{{N}} past
{{2^32}}). Same root cause, two symptoms.
h2. Root cause
File {{sql/core/.../execution/ExternalAppendOnlyUnsafeRowArray.scala}}:
* {{private var numRows = 0}} -- 32-bit {{Int}}, incremented per row in
{{add()}}.
* {{def length: Int = numRows}} -- returns the overflowed count.
File {{sql/core/.../execution/window/WindowFunctionFrame.scala}}:
* {{OffsetWindowFunctionFrameBase.inputIndex}} -- 32-bit {{Int}} cursor into
the partition.
* The {{absOffset > rows.length}} / {{inputIndex < input.length}} comparisons
are wrong once
{{length}} is negative.
In addition, the window drivers ({{WindowEvaluatorFactory}} and
{{ArrowWindowPythonEvaluatorFactory}}) track the per-partition row index as a
32-bit {{Int}}
({{rowIndex}}), and the sliding/ranking/segment-tree frames track their window
bounds as
{{Int}}, so those frame types also cannot handle partitions larger than
{{Int.MaxValue}} rows.
h2. Impact
Window queries over a single partition larger than {{Int.MaxValue}} rows can
return silently
incorrect results (most visibly, {{lag}}/{{lead}} returning NULL for every
row), or throw an
{{ArrayIndexOutOfBoundsException}}, depending on the exact partition size and
frame type.
h2. Fix direction
* Widen the row counter ({{numRows}}/{{length}}) and the offset frame's cursor
to {{Long}} so
{{lag}}/{{lead}} (and the unbounded-offset {{nth_value}} frame) produce
correct results for
large partitions.
* For the remaining frame types, which still rely on 32-bit row indices/bounds,
fail the query
with a clear user-facing error instead of silently producing wrong results.
Full {{Long}}
propagation for those frames (frame-bound API, aggregate processor, and the
core
{{UnsafeExternalSorter}} start index) is a larger follow-up.
Related: SPARK-51757 (write/insertion-path overflow in the same area).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
