parthchandra opened a new issue, #7270:
URL: https://github.com/apache/datafusion-comet/issues/7270
Part of #7267.
## Problem
Comet picks the window operator in
`native/core/src/execution/planner/window.rs:133`:
if every window expression in the Spark Window node can run with bounded
memory
(`uses_bounded_memory()`), it builds `BoundedWindowAggExec`; otherwise it
builds
DataFusion's `WindowAggExec` (`window.rs:142`) for the whole node.
`WindowAggExec` is used when any expression in the node:
- is `percent_rank`, `cume_dist` or `ntile` (never bounded); or
- is an aggregate (sum, avg, count, min, max, first, last) or `nth_value`
whose frame
ends at UNBOUNDED FOLLOWING (`physical-expr window/aggregate.rs:218`,
`standard.rs:277`, DataFusion 55.2.0). That includes Spark's default frame
when
there is no ORDER BY, so a plain `sum(x) OVER (PARTITION BY k)` lands here.
`WindowAggExec::compute_aggregates` buffers every input batch of the task,
then
runs `concat_batches` over all of them (`window_agg_exec.rs:539`). The result
columns are concatenated the same way (`:563`), and the operator emits one
batch
that is never sliced to the batch size (`:573`).
If the task's input holds more than 2 GiB in any string/binary column,
including
columns that are only passed through, the concat fails with
`OffsetOverflowError`.
A string result such as `max(blob) OVER (PARTITION BY k)` can also overflow
in the
result concat. With no PARTITION BY, Spark sends every row to one task.
No batch size setting helps, because the operator always builds one batch
from the
whole input.
## Options
1. **Emit per partition (Comet or upstream).** The input is already sorted
by the
partition keys, so the operator can compute and emit at each partition
boundary
instead of concatenating the whole task input. This bounds the concat to
one
window partition.
2. **Project pass-through columns out before the window** and join them back
by row
position afterwards. Only the partition/order keys and the function
inputs need
to go through the concat. Large payload columns usually aren't in those.
3. **Fall back to Spark** when the window input has string/binary columns
and the
plan can't bound their size. Simple, but over-broad: most string columns
are
small. Could be gated on a config.
Main already routes to `BoundedWindowAggExec` whenever DataFusion allows it,
so
there is no cheap rerouting option.
Recommendation: option 1, with option 3 as a stopgap if needed.
## Tests
- Plan-level test that `sum(x) OVER (PARTITION BY k)` and `ntile` plan as
`WindowAggExec`, then that the chosen fix applies.
- SQL-file tests of `percent_rank`, `ntile` and `sum(x) OVER (PARTITION BY
k)` with a
binary pass-through column, compared with Spark.
## Notes
- Risk is high: the default-frame aggregate case is common.
--
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]
