Sunninsky commented on issue #22355:
URL: https://github.com/apache/datafusion/issues/22355#issuecomment-6782329621
A data point from a downstream user (DataFusion Comet running Spark window
queries), for a different workload shape than the single-global-partition one
in the PoC: many partitions, one sort, and a frame that moves.
Setup: `datafusion-python` 55.0.0 (DataFusion 55.2.0), `target_partitions =
1`, 6M rows, 60,000 rows per `PARTITION BY k` partition, data already ordered
by `(k, t)`, `v` is `Int64`. Every query is `SELECT sum(w) FROM (SELECT
<window> AS w FROM t)` so the window cannot be pruned. The cost is the time
over `row_number() OVER (PARTITION BY k ORDER BY t)`, which pays for the same
sort and operator. Apple Silicon laptop, median of 3 runs, process CPU time
(the numbers match wall-clock time on an idle machine).
| window | DataFusion, ns/row |
|---|---|
| `SUM(v) OVER (PARTITION BY k)` | ~0 (bulk path) |
| `SUM(v)` default frame (`RANGE UNBOUNDED PRECEDING .. CURRENT ROW`) | 133 |
| `SUM(v)` `ROWS UNBOUNDED PRECEDING .. CURRENT ROW` | 64 |
| `SUM(v)` `ROWS 1 / 5 / 50 / 500 PRECEDING .. FOLLOWING` | 96 / 95 / 96 /
96 |
| `COUNT(v)` / `AVG(v)` `ROWS 5 .. 5` | 67 / 72 |
| `MAX(v)` / `MIN(v)` `ROWS 5 .. 5` | 93 / 94 |
Two things stand out. The cost per row is flat from a frame of 3 rows to a
frame of 1,001 rows, so it is the per-row evaluation and not the aggregation
itself (from reading 55.2: `get_result_column` makes two slices, calls
`update_batch` / `retract_batch` and `evaluate` into a `ScalarValue`, and
`iter_to_array` at the end, once per row). And a whole-partition aggregate is
free next to it. The same sliding `SUM` from prefix sums, column at a time
(NumPy, one thread, same answer, checked) costs 4-6 ns/row.
For context, DuckDB 2.0 on the same data with one thread (`PRAGMA
threads=1`): default frame 64, `ROWS 1..1` 59, `ROWS 5..5` 84, `ROWS 50..50`
102, `ROWS 500..500` 220 ns/row (it grows with the frame), `MAX` `ROWS 5..5`
51. So DataFusion is comparable today; the headroom is in not paying a per-row
cost at all.
This is the shape where intra-partition parallelism does not help (Spark has
already split the data by user, and the largest partition is 128k rows), so the
per-row cost is what we see. Against Databricks' incremental `WindowExec` it
made our native moving-frame aggregates 3-18% slower, and we keep them on the
JVM window by default.
One design input for the new window API, since it is about how values leave
a frame: `retract_batch` for `sum` / `avg` over floats gives wrong results once
`Infinity` or `NaN` has been in the frame (#26222: `inf - inf` is `NaN` and
never leaves the running sum). Spark and a recompute-per-frame approach return
the finite value. A vectorized or prefix-sum kernel for floats has the same
trap (`P[hi] - P[lo]` with an infinite prefix), so the interface may want a
per-type choice between "subtract" and "recompute" for float sums.
<details><summary>Script (needs <code>datafusion</code>,
<code>pyarrow</code>, <code>numpy</code>)</summary>
```python
import statistics, sys, time
import numpy as np, pyarrow as pa
from datafusion import SessionConfig, SessionContext
ROWS = int(sys.argv[1]) if len(sys.argv) > 1 else 6_000_000
PART = int(sys.argv[2]) if len(sys.argv) > 2 else 60_000
rng = np.random.default_rng(7)
k = (np.arange(ROWS) // PART).astype(np.int64)
t = np.arange(ROWS, dtype=np.int64)
v = rng.integers(-1000, 1000, ROWS).astype(np.int64)
tbl = pa.table({"k": k, "t": t, "v": v})
ctx = SessionContext(SessionConfig().with_target_partitions(1))
ctx.register_record_batches("t", [tbl.to_batches(max_chunksize=8192)])
OVER = "PARTITION BY k ORDER BY t"
def run(window):
ts = []
for _ in range(3):
t0 = time.process_time()
res = ctx.sql(f"SELECT sum(w) FROM (SELECT {window} AS w FROM
t)").collect()[0].column(0)[0].as_py()
ts.append(time.process_time() - t0)
return statistics.median(ts), res
base, _ = run(f"row_number() OVER ({OVER})")
for label, w in [("whole partition", "SUM(v) OVER (PARTITION BY k)"),
("default frame", f"SUM(v) OVER ({OVER})"),
("ROWS 5..5 SUM", f"SUM(v) OVER ({OVER} ROWS BETWEEN 5
PRECEDING AND 5 FOLLOWING)"),
("ROWS 500..500 SUM", f"SUM(v) OVER ({OVER} ROWS BETWEEN
500 PRECEDING AND 500 FOLLOWING)"),
("ROWS 5..5 MAX", f"MAX(v) OVER ({OVER} ROWS BETWEEN 5
PRECEDING AND 5 FOLLOWING)")]:
s, _ = run(w)
print(f"{label:20} {(s - base) / ROWS * 1e9:6.0f} ns/row over
row_number")
# reference: sliding SUM from prefix sums
P = np.concatenate(([0], np.cumsum(v)))
idx = np.arange(ROWS); start = (idx // PART) * PART; end = np.minimum(start
+ PART, ROWS)
w = 5
t0 = time.process_time()
out = P[np.minimum(idx + w + 1, end)] - P[np.maximum(idx - w, start)]
total = int(out.sum()); el = time.process_time() - t0
assert total == run(f"SUM(v) OVER ({OVER} ROWS BETWEEN 5 PRECEDING AND 5
FOLLOWING)")[1]
print(f"prefix sums ROWS 5..5: {el / ROWS * 1e9:.1f} ns/row (same answer)")
```
</details>
--
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]
