Sunninsky opened a new issue, #27400:
URL: https://github.com/apache/datafusion/issues/27400
### Is your feature request related to a problem or challenge?
Window aggregates and ranking functions cost far more per row than the work
they do, and with small partitions a fixed cost per
partition dominates. On `main` (`55de86155`), window aggregates
(`SUM/COUNT/AVG/MIN/MAX OVER (... ROWS/RANGE ...)`) cost **160 to 779 ns per
row**, and ranking functions and `lag` 28 to 87 ns per row, against a floor of
3.8 ns per row (scan + projection + final aggregate; 3,000,000 rows in
60,000-row partitions, x86-64 node).
Related: #23197 (epic), #26541 (per-partition cost in
`BoundedWindowAggExec`, assigned), #23273, #22355 / #22356, #15607.
#### Reproducer
datafusion-python 55.0.0 (DataFusion 55.2.0). The operator's own
`elapsed_compute` from `EXPLAIN ANALYZE`, one partition of the
plan, so the scan and the sort are not counted:
```python
from datafusion import SessionConfig, SessionContext
ctx = SessionContext(SessionConfig().with_target_partitions(1))
ctx.sql("CREATE TABLE t AS SELECT value / 60000 AS k, value AS ts, (value *
7919) % 2000 - 1000 AS v "
"FROM generate_series(0, 2999999)").collect() # 3M rows,
60,000-row partitions; use `value / 10` for 10-row ones
w = "SUM(v) OVER (PARTITION BY k ORDER BY ts ROWS BETWEEN 5 PRECEDING AND 5
FOLLOWING)"
ctx.sql(f"EXPLAIN ANALYZE SELECT sum(w) FROM (SELECT {w} AS w FROM
t)").show() # BoundedWindowAggExec ... elapsed_compute=
```
The script that runs seven shapes and prints ns per input row is in the
details block below. Output on a dedicated x86-64 node
(`r7i.8xlarge`; the process pinned to two cores, nothing else running):
```
$ python window_repro.py 3000000 60000
3,000,000 rows, 60,000 rows per partition, target_partitions = 1
row_number() BoundedWindowAggExec
25.8 ns/row
lag(v, 1) BoundedWindowAggExec
37.3 ns/row
SUM(v), default frame BoundedWindowAggExec
283.8 ns/row
SUM(v), ROWS 5 PRECEDING .. 5 FOLLOWING BoundedWindowAggExec
229.8 ns/row
MAX(v), ROWS 5 PRECEDING .. 5 FOLLOWING BoundedWindowAggExec
206.6 ns/row
SUM(v), RANGE 5 PRECEDING .. 5 FOLLOWING BoundedWindowAggExec
756.7 ns/row
SUM(v) OVER (PARTITION BY k) [whole partition] WindowAggExec
6.1 ns/row
$ python window_repro.py 3000000 10
3,000,000 rows, 10 rows per partition, target_partitions = 1
row_number() BoundedWindowAggExec
87.8 ns/row
lag(v, 1) BoundedWindowAggExec
135.7 ns/row
SUM(v), default frame BoundedWindowAggExec
403.3 ns/row
SUM(v), ROWS 5 PRECEDING .. 5 FOLLOWING BoundedWindowAggExec
263.5 ns/row
MAX(v), ROWS 5 PRECEDING .. 5 FOLLOWING BoundedWindowAggExec
253.2 ns/row
SUM(v), RANGE 5 PRECEDING .. 5 FOLLOWING BoundedWindowAggExec
783.3 ns/row
SUM(v) OVER (PARTITION BY k) [whole partition] WindowAggExec
55.6 ns/row
```
<details><summary>reproducer script</summary>
```python
"""Per-row cost of window evaluation in DataFusion, from the operator's own
`elapsed_compute` metric.
pip install datafusion # tested with datafusion-python 55.0.0
(DataFusion 55.2.0)
python window_repro.py [ROWS] [ROWS_PER_PARTITION]
The table is 3,000,000 rows in 60,000-row `PARTITION BY k` groups by
default, one thread, so the numbers are the cost of
the window operator alone (`EXPLAIN ANALYZE`, operator
`BoundedWindowAggExec`; the sort below it is not counted).
"""
import re
import sys
from datafusion import SessionConfig, SessionContext
ROWS = int(sys.argv[1]) if len(sys.argv) > 1 else 3_000_000
PART = int(sys.argv[2]) if len(sys.argv) > 2 else 60_000
OVER = "PARTITION BY k ORDER BY ts"
SHAPES = [
("row_number()", f"row_number() OVER ({OVER})"),
("lag(v, 1)", f"lag(v, 1) OVER ({OVER})"),
("SUM(v), default frame", f"SUM(v) OVER ({OVER})"),
("SUM(v), ROWS 5 PRECEDING .. 5 FOLLOWING", f"SUM(v) OVER ({OVER} ROWS
BETWEEN 5 PRECEDING AND 5 FOLLOWING)"),
("MAX(v), ROWS 5 PRECEDING .. 5 FOLLOWING", f"MAX(v) OVER ({OVER} ROWS
BETWEEN 5 PRECEDING AND 5 FOLLOWING)"),
("SUM(v), RANGE 5 PRECEDING .. 5 FOLLOWING", f"SUM(v) OVER ({OVER} RANGE
BETWEEN 5 PRECEDING AND 5 FOLLOWING)"),
("SUM(v) OVER (PARTITION BY k) [whole partition]", "SUM(v) OVER
(PARTITION BY k)"),
]
UNIT = {"ns": 1.0, "µs": 1e3, "us": 1e3, "ms": 1e6, "s": 1e9}
ctx = SessionContext(SessionConfig().with_target_partitions(1))
ctx.sql(
f"CREATE TABLE t AS SELECT value / {PART} AS k, value AS ts, (value *
7919) % 2000 - 1000 AS v "
f"FROM generate_series(0, {ROWS - 1})"
).collect()
def window_nanos(window):
"""elapsed_compute of the window operator, in ns, for SELECT sum(w) FROM
(SELECT <window> AS w FROM t)."""
batches = ctx.sql(f"EXPLAIN ANALYZE SELECT sum(w) FROM (SELECT {window}
AS w FROM t)").collect()
for b in batches:
for kind, plan in zip(b.column(0).to_pylist(),
b.column(1).to_pylist()):
if kind != "Plan with Metrics":
continue
for line in plan.splitlines():
if "WindowAggExec" in line:
m =
re.search(r"elapsed_compute=([0-9.]+)(ns|µs|us|ms|s)\b", line)
return float(m.group(1)) * UNIT[m.group(2)],
line.split(":")[0].strip()
raise RuntimeError("no window operator in the plan")
print(f"{ROWS:,} rows, {PART:,} rows per partition, target_partitions = 1")
for label, window in SHAPES:
best = min((window_nanos(window) for _ in range(3)), key=lambda r: r[0])
print(f"{label:52} {best[1]:22} {best[0] / ROWS:8.1f} ns/row")
```
</details>
#### Where the time goes on `main`
* Per row: the frame, two array slices and a `Vec<ArrayRef>`, up to three
dynamic calls (`update_batch`, `retract_batch`,
`evaluate`), a `ScalarValue` result and a final
`ScalarValue::iter_to_array` (`AggregateWindowExpr::get_result_column`).
`RANGE` with an offset also searches the peer group linearly, copying one
row into a `Vec<ScalarValue>` per step
(`WindowFrameStateRange::calculate_index_of_row`). The frame size does not
matter: `ROWS 5 PRECEDING .. 5 FOLLOWING` and `ROWS 500 PRECEDING .. 500
FOLLOWING` cost 261 and 259 ns per row on x86-64, 406 and 404 ns per row on
arm64.
* Per partition (`SortedSearch`): a `Vec<ScalarValue>` key hashed several
times per batch and per expression, a `WindowState` per
partition and expression, per-poll hash maps in `calculate_n_out_row`,
pruning. This is the cost #26541 describes.
### Describe the solution you'd like
A throwaway spike (no PR yet) with two steps. Both add provided methods
whose default keeps today's behaviour; nothing changes for
a user-defined function that does not opt in.
1. **Bulk frames.** The frames of a whole run of rows in one pass
(`WindowFrameContext::calculate_ranges`: `ROWS` closed form,
`RANGE` peers from an adjacent-inequality vector, `RANGE` offsets for
integer and timestamp keys by a typed two-pointer), an
accumulator hook that evaluates a run of frames in one call
(`Accumulator::evaluate_window_frames`; typed kernels for
`SUM/COUNT/AVG/MIN/MAX` that add and retract in the same order as today,
so floats stay bit-identical) and
`PartitionEvaluator::evaluate_rows` for causal frame-free functions
(`row_number`, `rank`, `dense_rank`, `lag`).
2. **Complete partitions evaluated directly.** In `Sorted` mode a run of
partitions that start and end inside one input batch is
evaluated with one call per window expression on the batch, without
per-partition key, state, hash map or pruning
(`WindowExpr::evaluate_complete_partitions`). Its provided implementation
runs the existing `evaluate_stateful` on each
partition, so any expression, user-defined ones included, behaves as
before; aggregates and the functions above have segmented
fast paths (`WindowFrameContext::calculate_ranges_segmented`,
`Accumulator::evaluate_window_segments`,
`PartitionEvaluator::evaluate_segments`). It is used when at least 4
complete partitions are in the batch and at most one open
partition is buffered, never with a `WindowStateObserver`. This is the
second direction of #26541.
New public items: `Accumulator::evaluate_window_frames`,
`Accumulator::evaluate_window_segments`,
`PartitionEvaluator::evaluate_rows`,
`PartitionEvaluator::evaluate_segments`,
`WindowExpr::evaluate_complete_partitions` (all provided
methods), `WindowFrameContext::calculate_ranges`,
`WindowFrameContext::calculate_ranges_segmented`, `CompletePartitions`,
`evaluate_complete_partitions_stateful`, `adjacent_distinct`, and a
`window_kernels` module in `datafusion-functions-aggregate-common`.
Names are provisional.
#### Results
CPU time of the whole query thread (scan, window operator, final aggregate).
The runs of `main` and of the prototype alternate on
one pinned core of a dedicated node; in-memory tables already in partition
order, 1M to 3M rows, one thread; the speedup is the median
over 3 rounds of the ratio taken inside a round:
**x86-64 (Intel Xeon Platinum 8488C, Sapphire Rapids)**: ns per input row,
upstream main -> prototype (speedup, median of paired rounds)
| shape | partitions of 60,000 rows | partitions of 240 rows | partitions of
10 rows | partitions of 1 row |
|---|---|---|---|---|
| row_number | 28 -> 5.7 (5x) | 31 -> 5.9 (5x) | 93 -> 6.1 (15x) | 675 -> 10
(65x) |
| rank (no ties) | 83 -> 8.2 (10x) | 87 -> 7.9 (11x) | 191 -> 7.9 (24x) |
1139 -> 12 (99x) |
| lag(v,1) | 39 -> 6.8 (6x) | 45 -> 7.9 (6x) | 141 -> 9.7 (15x) | 1116 -> 12
(93x) |
| first_value (default frame) | 134 -> 11 (12x) | 140 -> 13 (11x) | 243 ->
13 (19x) | 1229 -> 17 (74x) |
| SUM default frame (RANGE running) | 288 -> 15 (19x) | 292 -> 14 (21x) |
414 -> 14 (28x) | 1536 -> 19 (82x) |
| SUM ROWS 5..5 | 261 -> 13 (20x) | 262 -> 13 (20x) | 290 -> 15 (19x) | 1422
-> 20 (70x) |
| MAX ROWS 5..5 | 247 -> 14 (17x) | 252 -> 15 (17x) | 285 -> 14 (20x) | 1440
-> 21 (68x) |
| SUM RANGE 5..5 (offset) | 779 -> 22 (35x) | 795 -> 21 (37x) | 797 -> 23
(35x) | 1946 -> 42 (46x) |
| 4 aggs ROWS 5..5 (shared window) | 937 -> 33 (29x) | 947 -> 35 (27x) | 961
-> 38 (26x) | 4467 -> 52 (84x) |
| SUM whole partition | 13 -> 13 (1.0x) | 15 -> 15 (1.0x) | 63 -> 63 (1.0x)
| 554 -> 566 (1.0x) |
The last row is a control: `SUM(v) OVER (PARTITION BY k)` runs in
`WindowAggExec`, which the prototype does not touch. The query is a
scan, one window operator and a final aggregate (no sort, no I/O); a real
query pays for the rest as well, so its gain depends on how
much of its time the window operator is. Noise: the same binary run twice
(A/A, 10-row partitions, 5 rounds) differs by at most 1.05x on x86-64 and 1.12x
on arm64 in a single round; the control row, whose code the prototype does not
change, moves by 0.91x to 1.03x between the three different binaries (median
over the rounds, every partition size), so differences of about that size
between binaries are not evidence of a change in the code.
<details><summary>arm64 (Graviton3) and the other partition sizes</summary>
**arm64 (AWS Graviton3, Neoverse-V1)**: ns per input row, upstream main ->
prototype (speedup, median of paired rounds)
| shape | partitions of 60,000 rows | partitions of 240 rows | partitions of
10 rows | partitions of 1 row |
|---|---|---|---|---|
| row_number | 29 -> 8.6 (3x) | 33 -> 9.5 (4x) | 134 -> 9.3 (15x) | 951 ->
14 (66x) |
| rank (no ties) | 94 -> 13 (7x) | 100 -> 12 (8x) | 266 -> 12 (22x) | 1657
-> 15 (107x) |
| lag(v,1) | 49 -> 9.2 (5x) | 60 -> 12 (5x) | 209 -> 15 (14x) | 1618 -> 18
(91x) |
| first_value (default frame) | 213 -> 16 (13x) | 219 -> 19 (11x) | 380 ->
20 (19x) | 1739 -> 24 (74x) |
| SUM default frame (RANGE running) | 453 -> 20 (22x) | 460 -> 20 (23x) |
651 -> 21 (31x) | 2296 -> 26 (88x) |
| SUM ROWS 5..5 | 406 -> 18 (22x) | 435 -> 19 (23x) | 450 -> 24 (19x) | 2119
-> 29 (74x) |
| MAX ROWS 5..5 | 356 -> 20 (18x) | 388 -> 22 (18x) | 424 -> 21 (20x) | 2186
-> 29 (76x) |
| SUM RANGE 5..5 (offset) | 1267 -> 26 (49x) | 1286 -> 26 (49x) | 1245 -> 31
(40x) | 2903 -> 50 (58x) |
| 4 aggs ROWS 5..5 (shared window) | 1384 -> 45 (30x) | 1486 -> 48 (31x) |
1478 -> 52 (28x) | 6812 -> 66 (103x) |
| SUM whole partition | 7.9 -> 7.9 (1.0x) | 13 -> 13 (1.0x) | 90 -> 89
(1.0x) | 799 -> 790 (1.0x) |
Speedup of the prototype over `main` on dedicated nodes (one pinned core,
median over paired rounds of the ratio taken inside a round), range over the 11
shapes `row_number`, `rank`, `lag`, `first_value`, `SUM`/`COUNT`/`AVG` default
frame, `SUM`/`MAX` `ROWS 5..5`, `SUM RANGE 5..5`, four aggregates sharing a
window, by rows per partition: x86-64: 60,000 rows: 5-35x; 240 rows: 5-37x; 10
rows: 15-35x; 4 rows: 20-43x; 2 rows: 24-60x; 1 row: 46-99x | arm64: 60,000
rows: 3-49x; 240 rows: 4-49x; 10 rows: 14-40x; 4 rows: 20-42x; 2 rows: 26-65x;
1 row: 58-107x.
</details>
#### Correctness evidence
* A differential fuzzer runs random window queries (`ROWS` / `RANGE` /
`GROUPS`, offsets, nullable and extreme keys, floats with
`NaN` / `inf` / `-0.0`, timestamps, string and nullable PARTITION BY keys,
batch sizes 1 to 8192, `Sorted` and `Linear` modes) twice in
one process, once with the original per-row code and once with the
prototype, and compares the results (floats by bits, every `NaN` as
equal): 381,600 queries on the final tree in 10 configurations (the parts
of the prototype switched on and off, long partitions, standard functions only,
aggregates only), 0 mismatches and 0 queries that failed on one side only;
80,243 of the queries are rejected identically by both sides (for example
invalid frames). Earlier iterations ran 522,000 more queries; their only
mismatches (377, all in the first campaign) were `NaN` against `NaN` with a
different sign.
* The existing suites pass on the final tree: unit tests of the seven
touched crates (5,564 passed), sqllogictest `window` (7 files), the extended
`window` fuzz cases (4 passed), `core_integration window` (126 passed).
### Describe alternatives you've considered
* Cheaper per-partition keys (row format instead of `Vec<ScalarValue>`), as
#26541 notes, cut the hashing but keep the per-partition
batches and states, so the cost stays per partition.
* A new kernel API for window evaluation (#23273, #15607) is the longer-term
direction. The spike adds only provided methods that fall
back to today's protocol, so it does not preclude one.
* Intra-operator parallelism (#22355) targets one huge partition; the shapes
above (many partitions) are a different case.
### Additional context
#### Unknowns, limits
* Not covered: `GROUPS` frames, `FILTER`, aggregates without a kernel, `lag
... IGNORE NULLS`, float / decimal / date `RANGE` offsets,
user-defined functions (they keep today's cost until they implement a
hook). `WindowAggExec` and the `Linear` / `PartiallySorted`
modes keep their per-partition cost (the control row above).
* With 40 or 150 extra nullable columns carried through the window (500,000
rows, partitions of 10 rows) the only shape slower than step 1 alone is `GROUPS
2..2` in three expressions (no fast path): 0.90x (x86-64, 150 columns), 0.97x
(arm64, 150 columns), 0.97x (x86-64, 40 columns); every other shape and table
width is as fast or faster (median of 3 paired rounds).
* The sign of a `NaN` that arithmetic generates (`inf - inf`) is hardware
dependent and can differ between the kernel and the per-row
protocol; whether that is observable through SQL was not investigated.
Float sums keep the add / retract order of `main`, so a frame
that has held `Infinity` or `NaN` gives what it gives today (#26222): the
kernels reproduce it and a fix there would have to cover them.
* Not measured: `target_partitions > 1`, spilling and memory accounting
(#26707), I/O-bound queries.
* The timings were taken on a build that differs from the validated tree
only by lint edits (two `try_reserve` calls and `#[expect]` attributes); timed
against each other on x86-64 (9 rounds: 0.96-1.03x over 48 shape/partition-size
combinations, 0 with a different checksum) the two builds are within the noise
floor. The arm64 timings were not repeated.
* The spike toggles its behaviour with process-global flags read from the
environment; a PR would need instance-level configuration
or no opt-out.
#### Questions for maintainers
1. #26541 is assigned and its second bullet is step 2 above. Should that
part stay with its assignee (this design as a reference), or
would a PR built on it be welcome? This is raised as an issue first; no
PR has been opened.
2. Is a provided method on `Accumulator` / `PartitionEvaluator` the right
shape, or should the kernels live behind the kernel API
direction of #23273?
3. A possible order: benchmarks for these shapes
(`core/benches/window_query_sql.rs` has no moving-frame, `RANGE`-offset,
`rank`, `lag`
or small-partition shape), then bulk frame bounds, then the accumulator
kernels one aggregate at a time, then step 2.
The prototype can be shared on request.
**AI disclosure.** The prototype, the measurements and this text were
produced with Claude Code (an AI coding assistant) under the
supervision of the account owner, who directed the work and approved posting
this issue. Every number comes from a log of the runs
described above.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
