Sunninsky opened a new pull request, #27410: URL: https://github.com/apache/datafusion/pull/27410
## Which issue does this PR close? - Part of #27400 (design, measurements and open questions). - Related to #26541: layers 3 and 4 below are one implementation of its two points. See [Relation to other work](#relation-to-other-work). > **Draft.** Opened so that the code can be read and measured while the questions in #27400 are discussed (shape of the API, coordination with #26541, order of PRs); #27400 has no answer yet. It is deliberately one PR so that the layers can be seen working together, as the [contributor guide](https://datafusion.apache.org/contributor-guide/#creating-pull-requests) suggests for a larger design; the [split plan](#review-guide-and-split-plan) at the end turns it into three smaller PRs. It will be split, reduced or closed to match what the maintainers prefer. ## Rationale for this change `BoundedWindowAggExec` evaluates a window function one row and one partition at a time. For every row it computes a frame, builds two array slices, makes up to three dynamic calls (`update_batch`, `retract_batch`, `evaluate`) and boxes the result in a `ScalarValue`; for every partition it builds a key, a `WindowState`, a buffer and several hash map entries; for every input column it slices or gathers one array per partition. The arithmetic is a small part of that. What a user sees on `main` (`4c7b30752`; one thread, dedicated x86-64 node; CPU time of the query thread per input row, of which the scan and the final aggregate around the window are 3.8 ns; 3M rows for 60,000-row partitions, 1M for 240 and 10, 300k for 1): | shape (ns per input row, `main`) | 60,000-row partitions | 240 rows | 10 rows | 1 row | |---|---:|---:|---:|---:| | `row_number()` | 28 | 31 | 92 | 659 | | `rank()` (no ties) | 83 | 86 | 189 | 1133 | | `lag(v, 1)` | 39 | 44 | 140 | 1129 | | `first_value`, default frame | 134 | 141 | 244 | 1142 | | `SUM`, default frame | 286 | 290 | 414 | 1529 | | `COUNT`, default frame | 231 | 240 | 359 | 1456 | | `AVG`, default frame | 238 | 246 | 378 | 1571 | | `SUM`, `ROWS 5 PRECEDING .. 5 FOLLOWING` | 260 | 263 | 285 | 1416 | | `MAX`, `ROWS 5 PRECEDING .. 5 FOLLOWING` | 246 | 253 | 280 | 1440 | | `SUM`, `RANGE 5 PRECEDING .. 5 FOLLOWING` | 779 | 796 | 796 | 1920 | | four aggregates sharing one window | 933 | 949 | 946 | 4336 | | control: `SUM(v) OVER (PARTITION BY k)` | 12 | 12 | 58 | 552 | The reproducer is in #27400. ## What changes are included in this PR? The protocol changes from *one row, one partition at a time* to *one run of rows, one run of partitions at a time*, in four layers. Each layer has a default that keeps today's behaviour, so every expression, user-defined ones included, stays correct without opting in. ``` input batch ─► partition ranges (PARTITION BY columns) Sorted: runs of equal keys · Linear/PartiallySorted: hash, then one `take` ─► [4] keep only the columns the expressions read → per-partition batches ─► per window expression [3] at least 4 complete partitions in the batch → one call on the whole run (Sorted mode only) otherwise per partition, `evaluate_stateful`: [1] the frames of all new rows in one pass [2] one call evaluates the run of frames (aggregate kernels, `evaluate_rows`) ─► emit the completed rows next to the full-width input ``` 1. **Frames of a run of rows in one pass** (`WindowFrameContext::calculate_ranges`, new file `window_state/bulk.rs`). `ROWS`: closed form. `RANGE` with `CURRENT ROW` / `UNBOUNDED` bounds: peer groups from one adjacent-inequality vector (`adjacent_distinct`, arrow `distinct`). `RANGE` with an offset on one integer or UTC/zone-less timestamp key: a typed two-pointer over the keys widened to `i128` (no overflow of `key ± offset`; `NULLS FIRST/LAST`, `DESC`, unsigned underflow as in the per-row path; interval offsets that are whole ticks of the key's unit). `GROUPS` and the remaining `RANGE` cases call the per-row code in a loop. 2. **A run of frames in one call.** Two optional hooks, `Accumulator::evaluate_window_frames` and `PartitionEvaluator::evaluate_rows`, both defaulting to "not supported". `SUM`, `COUNT`, `AVG`, `MIN`, `MAX`, `row_number`, `rank`, `dense_rank` and `lag` implement them with typed kernels (`functions-aggregate-common/src/window_kernels.rs`) that add and retract in the same order as the per-row protocol, so floating-point results are bit-identical. `StandardWindowExpr` and `AggregateWindowExpr` compute the frames of all new rows (layer 1), offer the run to the hook, and fall back to the per-row loop for the rows that need it. 3. **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 (`WindowExpr::evaluate_complete_partitions`), without per-partition key, state, buffer, hash map or pruning. The provided implementation runs the existing `evaluate_stateful` on each partition with a throw-away state, which is exactly the call the stream makes for such a partition today; aggregates and the functions above have segmented fast paths (`calculate_ranges_segmented`, `Accumulator::evaluate_window_segments`, `PartitionEvaluator::evaluate_segments`). Used when the batch has at least `MIN_PARTITIONS_FOR_DIRECT_EVALUATION = 4` complete partitions and at most one open partition is buffered, and never with a `WindowStateObserver`. The results wait in `SortedSearch::pending` until `calculate_out_columns` splices them in at the position of their rows. 4. **Column pruning** (`windows/pruned_input.rs`). The operator passes every input column through, but splitting the input by partition and buffering each partition needs only the columns the window expressions read. `PrunedInput` finds them (PARTITION BY, ORDER BY, arguments, the ORDER BY of an aggregate, `FILTER`), rewrites the expressions with remapped `Column` indices, and projects only the per-partition batches (slices, or the `take` of `Linear` mode), which also shrinks the buffers. Partition keys and ranges are still computed on the full batch, the output base (`input_buffer`) stays full width, and the observer gets the original expressions. Anything that cannot be rewritten faithfully keeps every column: a `WindowExpr` or window function of an unknown kind, lambda expressions, a rewritten expression whose type, nullability or `field()` differs, or a query that reads every column anyway. ### How a query takes the new paths | window expression | frames | evaluation | complete partitions (`Sorted`, ≥ 4 per batch) | |---|---|---|---| | `SUM` / `COUNT` / `AVG` / `MIN` / `MAX`, `ROWS` | closed form | kernel | segmented kernel | | the same, `RANGE` default frame or peers | peer groups | kernel | segmented kernel | | the same, `RANGE` offset on one integer / timestamp key | two-pointer | kernel | segmented kernel | | the same, `RANGE` offset on other keys or several ORDER BY columns, `GROUPS` | per row | kernel | per-partition stream code | | the same with `FILTER` | as above | per-row protocol | per-partition stream code | | `row_number`, `rank`, `dense_rank`, `lag` | not needed | `evaluate_rows` | `evaluate_segments` | | `lead` | not needed | per row | `evaluate_segments` | | `first_value`, `last_value`, `nth_value` | as the frame | `evaluate_ranges` (one call already on `main`) | segmented frames, then `evaluate_ranges` | | other built-in and user-defined functions | as before | as before | per-partition stream code | Layer 4 applies to every row of the table. `Linear` and `PartiallySorted` modes use layers 1, 2 and 4 (partitions are rarely complete in one batch there). ### Public API New public items, all provided methods or getters with a default that keeps today's behaviour: * `Accumulator::evaluate_window_frames`: evaluates a run of frames in one call; `Ok(None)` (the default) means "follow the per-row protocol". * `PartitionEvaluator::evaluate_rows`: the same for causal functions that ignore the frame. * `PlainAggregateWindowExpr::get_filter_expr`, `SlidingAggregateWindowExpr::get_filter_expr`: getters next to `get_aggregate_expr`. Public only for use inside the workspace (`#[doc(hidden)]` with the "Public Only for Internal Use" note of the [API health policy](https://datafusion.apache.org/contributor-guide/api-health.html#internal-public-apis)): `Accumulator::evaluate_window_segments`, `PartitionEvaluator::evaluate_segments`, `WindowExpr::evaluate_complete_partitions`, `CompletePartitions`, `WindowFrameContext::calculate_ranges`, `WindowFrameContext::calculate_ranges_segmented`, `datafusion_expr::window_state::adjacent_distinct` and the `datafusion_functions_aggregate_common::window_kernels` module. If the shape of the two public hooks is the open question, they can start `#[doc(hidden)]` as well. ### What does not change Results (bit-identical, floats included; see the `NaN` note below), plans and `EXPLAIN` output, SQL semantics, the `WindowAggExec` operator (unbounded frames; the expression code it calls is shared with the bounded operator, so a frame that is not constant also gets the one-pass frames and the kernels there), the planner, and the behaviour of any user-defined function that does not opt in. There is no configuration option and no environment flag. ## Performance **Method.** Dedicated nodes (`r7i.8xlarge` x86-64, Sapphire Rapids; `m7g.8xlarge` arm64, Neoverse-V1), one pod per node and nothing else scheduled on it; one timing process at a time, pinned to one core; `target_partitions = 1`; in-memory table that declares its sort order, so there is no sort. Time is the **CPU time of the query thread** (`CLOCK_THREAD_CPUTIME_ID`): scan, window operator and a final aggregate, so a real query gains in proportion to how much of its time the window operator is. Each shape is run three times in a process (median); `main` and the PR build alternate in rotating order for three rounds; the ratio is taken **inside a round** and the median over the rounds is shown (`[lowest-highest]` in the details). Both binaries are built from the same runner source, same toolchain (1.95.0), same profile (`release-nonlto`), separate target directories; every pair of runs also compares a checksum of the result. **Speedup by partition size** (`main` / PR; lowest and highest median over 11 shapes: `row_number`, `rank`, `lag`, `first_value` with the default frame, `SUM` / `COUNT` / `AVG` with the default frame, `SUM` and `MAX` over `ROWS 5 PRECEDING .. 5 FOLLOWING`, `SUM` over `RANGE 5 PRECEDING .. 5 FOLLOWING`, and four aggregates sharing one window; for 60,000 rows, all 59 shapes of the runner). The control is `SUM(v) OVER (PARTITION BY k)`, which runs in `WindowAggExec` and takes the constant-frame branch of `get_result_column`, the one part of that function this PR does not touch. | rows per partition | x86-64 | arm64 | control (x86-64 / arm64) | |---|---|---|---| | 60,000 | 5-38x | 3-50x | 0.97x / 1.00x | | 6,000 | 5-37x | 3-48x | 1.03x / 1.00x | | 3,000 | 5-37x | 3-47x | 1.03x / 1.00x | | 1,500 | 4-37x | 3-48x | 1.04x / 1.01x | | 600 | 4-39x | 3-48x | 1.00x / 1.00x | | 240 | 5-40x | 4-50x | 1.00x / 0.99x | | 10 | 15-35x | 15-41x | 1.00x / 1.02x | | 4 | 19-41x | 21-48x | 0.94x / 1.01x | | 2 | 23-60x | 27-68x | 1.00x / 1.00x | | 1 | 45-102x | 60-114x | 1.06x / 1.01x | | one partition of 3M | 5-38x | 3-50x | 1.04x / 1.00x | | 60,000, all 59 shapes of the runner | 1.0-39x | 1.0-52x | | | shape | `main` | PR | `main` / PR [lowest-highest round] | |---|---:|---:|---| | `row_number()` | 28 | 5.6 | 5x [5-5] | | `rank()` (no ties) | 83 | 7.9 | 10x [10-11] | | `lag(v, 1)` | 39 | 6.8 | 6x [6-6] | | `first_value`, default frame | 134 | 11 | 12x [12-12] | | `SUM`, default frame | 286 | 15 | 20x [20-20] | | `COUNT`, default frame | 231 | 12 | 19x [19-19] | | `AVG`, default frame | 238 | 15 | 16x [16-16] | | `SUM`, `ROWS 5 PRECEDING .. 5 FOLLOWING` | 260 | 13 | 21x [20-21] | | `MAX`, `ROWS 5 PRECEDING .. 5 FOLLOWING` | 246 | 14 | 17x [17-17] | | `SUM`, `RANGE 5 PRECEDING .. 5 FOLLOWING` | 779 | 21 | 38x [37-38] | | four aggregates sharing one window | 933 | 33 | 28x [28-28] | | control: `SUM(v) OVER (PARTITION BY k)` | 12 | 13 | 1.0x [0.9-1.0] | **Which layer produces the gain** (x86-64, 10-row partitions; `1-2` = layers 1 and 2 only, `1-3` = plus direct evaluation of complete partitions, `PR` = all four layers; each ratio compares two builds that ran back to back in a round): | shape | `main` | layers 1-2 | layers 1-3 | PR | `main` / 1-2 | 1-2 / 1-3 | 1-3 / PR | `main` / PR | |---|---:|---:|---:|---:|---|---|---|---| | `row_number()` | 92 | 59 | 6.1 | 6.0 | 1.6x | 10x | 1.0x | 15x | | `rank()` (no ties) | 189 | 113 | 7.9 | 7.9 | 1.7x | 14x | 1.0x | 24x | | `lag(v, 1)` | 140 | 136 | 9.5 | 9.6 | 1.0x | 14x | 1.0x | 15x | | `first_value`, default frame | 244 | 140 | 12 | 12 | 1.7x | 11x | 1.0x | 20x | | `SUM`, default frame | 414 | 154 | 14 | 14 | 2.7x | 11x | 1.0x | 29x | | `COUNT`, default frame | 359 | 147 | 12 | 12 | 2.4x | 13x | 1.0x | 31x | | `AVG`, default frame | 378 | 163 | 13 | 13 | 2.3x | 12x | 1.0x | 28x | | `SUM`, `ROWS 5 PRECEDING .. 5 FOLLOWING` | 285 | 124 | 15 | 15 | 2.3x | 8x | 1.0x | 19x | | `MAX`, `ROWS 5 PRECEDING .. 5 FOLLOWING` | 280 | 132 | 14 | 14 | 2.1x | 9x | 1.0x | 20x | | `SUM`, `RANGE 5 PRECEDING .. 5 FOLLOWING` | 796 | 133 | 23 | 23 | 6x | 6x | 1.0x | 35x | | four aggregates sharing one window | 946 | 396 | 38 | 38 | 2.4x | 10x | 1.0x | 25x | | control: `SUM(v) OVER (PARTITION BY k)` | 58 | 59 | 59 | 58 | 1.0x | 1.0x | 1.0x | 1.0x | Layers 1 and 2 change the cost of a row, layer 3 the cost of a partition, layer 4 the cost of a column. So the gain of layers 1-2 grows with the partition size (1.0-6x at 10 rows, 3.6-36x at 240 to 1,500 rows), the gain of layer 3 shrinks with it (5.8-105x at 1 to 10 rows, 1.3-1.6x at 240 rows, none from about 3,000 rows, where fewer than four partitions end in a batch), and the gain of layer 4 grows with the number of columns (next table). The tables for 240 rows and for 1 row are in the details. **Wide inputs** (500,000 rows, 10-row partitions, columns that no window function reads carried through the operator): | shape (ns per row, `main` -> PR, speedup) | 11 columns | 51 columns | 161 columns | gain of the pruning at 161 columns (1-3 -> PR) | |---|---|---|---|---| | row_number | 93 -> 6.8 (14x) | 401 -> 12 (34x) | 1229 -> 25 (48x) | 1.3x | | rank (no ties) | 189 -> 8.6 (22x) | 490 -> 14 (36x) | 1326 -> 27 (49x) | 1.2x | | lag(v,1) | 141 -> 10 (14x) | 452 -> 15 (29x) | 1293 -> 29 (44x) | 1.2x | | SUM ROWS 5..5 | 285 -> 16 (18x) | 621 -> 39 (16x) | 1498 -> 94 (16x) | 1.1x | | GROUPS 2..2 x3 (generic) | 835 -> 552 (1.5x) | 1181 -> 573 (2.1x) | 2063 -> 633 (3x) | 3x | | SUM FILTER default frame (generic) | 664 -> 553 (1.2x) | 1002 -> 578 (1.7x) | 1869 -> 633 (3.0x) | 2.7x | | STDDEV default frame (generic) | 399 -> 304 (1.3x) | 735 -> 327 (2.3x) | 1625 -> 384 (4x) | 4x | | mixed: SUM ROWS 5..5 + GROUPS + row_number | 616 -> 218 (2.8x) | 953 -> 242 (4x) | 1831 -> 299 (6x) | 5x | **Noise and controls.** x86-64: the PR binary run twice (A/A, 10-row partitions, 5 rounds) differs by at most 1.06x in a single round; the control row, which takes a code path the PR does not change, is 0.94-1.06x between `main` and the PR (median over the rounds, every partition size); arm64: the PR binary run twice (A/A, 10-row partitions, 5 rounds) differs by at most 1.11x in a single round; the control row, which takes a code path the PR does not change, is 0.99-1.02x between `main` and the PR (median over the rounds, every partition size). **Criterion** (the `window_query_sql` bench of `datafusion`, built like the runner above, with `release-nonlto` and not with the LTO profile that `cargo bench` uses; x86-64 node, one pinned core; `--sample-size 10`, because the existing benchmarks take one to four seconds per iteration; `main` and the PR alternate for three rounds and the ratio is taken inside a round). `window empty over` and `window partition by` run in `WindowAggExec` (whole-partition frames; that operator is not changed). Their aggregate rows run code this PR does not change (a constant frame: one `update_batch` and one `evaluate`), so they are the controls; their `first_value` / `last_value` / `nth_value` rows share `StandardWindowExpr::evaluate`, which now computes the frames in one pass. `window order by` and `window partition and order by` run in `BoundedWindowAggExec`. The `window frames` group is new: `main` has no moving-frame, `RANGE`-offset, `rank`, `lag` or small-partition shape, and its `sum, whole partition` rows are controls too. | benchmark | `main` | PR | `main` / PR [lowest-highest round] | |---|---:|---:|---| | window empty over, aggregate functions | 11.71 ms | 14.91 ms | 0.79x [0.76-0.85] | | window empty over, built-in functions | 44.47 ms | 39.22 ms | 1.14x [1.11-1.23] | | window order by, aggregate functions | 919.60 ms | 377.60 ms | 2.43x [2.43-2.45] | | window order by, built-in functions | 726.81 ms | 135.79 ms | 5.4x [5.3-5.4] | | window partition by, u64_wide, aggregate functions | 3.60 s | 3.69 s | 1.00x [0.95-1.00] | | window partition by, u64_narrow, aggregate functions | 78.55 ms | 80.59 ms | 0.97x [0.97-1.00] | | window partition by, u64_wide, built-in functions | 4.73 s | 4.79 s | 0.99x [0.98-0.99] | | window partition by, u64_narrow, built-in functions | 100.14 ms | 94.33 ms | 1.06x [0.82-1.12] | | window partition and order by, u64_wide, aggregate functions | 5.38 s | 3.74 s | 1.45x [1.43-1.45] | | window partition and order by, u64_narrow, aggregate functions | 954.73 ms | 533.19 ms | 1.79x [1.79-1.79] | | window partition and order by, u64_wide, built-in functions | 4.56 s | 231.96 ms | 19.6x [19.6-19.7] | | window partition and order by, u64_narrow, built-in functions | 668.74 ms | 226.17 ms | 2.95x [2.94-2.96] | | window frames/partitions of 10 rows, row_number | 121.90 ms | 2.96 ms | 41.2x [41.1-41.3] | | window frames/partitions of 10 rows, rank | 200.35 ms | 5.26 ms | 38.1x [38.1-38.3] | | window frames/partitions of 10 rows, lag | 188.75 ms | 6.54 ms | 28.8x [28.3-28.9] | | window frames/partitions of 10 rows, sum, default frame | 475.26 ms | 11.20 ms | 42.5x [42.2-42.7] | | window frames/partitions of 10 rows, sum, rows 5 preceding to 5 following | 339.21 ms | 12.62 ms | 26.8x [26.7-27.5] | | window frames/partitions of 10 rows, max, rows 5 preceding to 5 following | 333.36 ms | 12.15 ms | 27.4x [27.0-27.7] | | window frames/partitions of 10 rows, sum, range 5 preceding to 5 following | 1.01 s | 20.83 ms | 48.7x [48.1-48.8] | | window frames/partitions of 10 rows, sum, whole partition | 64.03 ms | 64.22 ms | 1.00x [0.97-1.00] | | window frames/partitions of 1000 rows, row_number | 27.04 ms | 2.64 ms | 10.2x [10.2-10.4] | | window frames/partitions of 1000 rows, rank | 94.07 ms | 5.13 ms | 18.3x [18.2-18.5] | | window frames/partitions of 1000 rows, lag | 40.34 ms | 6.19 ms | 6.5x [6.5-6.5] | | window frames/partitions of 1000 rows, sum, default frame | 315.33 ms | 10.57 ms | 29.8x [29.7-29.9] | | window frames/partitions of 1000 rows, sum, rows 5 preceding to 5 following | 292.51 ms | 10.27 ms | 28.5x [28.4-28.6] | | window frames/partitions of 1000 rows, max, rows 5 preceding to 5 following | 268.98 ms | 11.55 ms | 23.3x [23.3-23.6] | | window frames/partitions of 1000 rows, sum, range 5 preceding to 5 following | 941.96 ms | 18.41 ms | 51.2x [50.9-51.2] | | window frames/partitions of 1000 rows, sum, whole partition | 4.48 ms | 4.46 ms | 1.00x [1.00-1.01] | | window frames/100 more columns, sum, rows 5 preceding to 5 following | 1.51 s | 122.66 ms | 12.3x [12.2-12.3] | | window frames/100 more columns, three groups frames | 3.13 s | 1.36 s | 2.30x [2.29-2.32] | The controls read 0.97-1.00x, except `window empty over, aggregate functions` at 0.79x, which is noise: for one and the same binary it takes either about 11.5 ms or about 15 ms, from one process to the next. Six more alternating runs of `main`, the PR and a byte-identical copy of the PR binary (`--sample-size 20`) gave 11.5 to 15.0 ms for each of the three, `main` / PR 1.06x [0.90-1.30] and PR / copy 0.99x [0.78-1.02]; the code that benchmark runs is the same on both sides. A difference of 20% between two single runs of one benchmark is therefore not evidence of a change, in either direction. The logs can be shared. <details><summary>All shapes, all partition sizes, both architectures</summary> The full tables of the arm64 node (same shapes, same method) are not repeated here for length; the arm64 columns of the headline table above come from them. **x86-64** #### Summary | section | shapes | lowest / highest median base / PR | control `SUM whole partition` (same code in both binaries) | |---|---:|---|---| | 3M rows, partitions of 60,000 rows | 59 | 1.0x / 39x | 1.0x | | 1M rows, partitions of 240 rows | 11 | 5x / 40x | 1.0x | | 1M rows, partitions of 10 rows | 11 | 15x / 35x | 1.0x | | 500k rows, partitions of 4 rows | 11 | 19x / 41x | 0.9x | | 300k rows, partitions of 2 rows | 11 | 23x / 60x | 1.0x | | 300k rows, partitions of 1 row | 11 | 45x / 102x | 1.1x | | 3M rows, one partition | 11 | 5x / 38x | 1.0x | | 1M rows, partitions of 600 rows (about 13 complete partitions per input batch) | 11 | 4x / 39x | 1.0x | | 1M rows, partitions of 1,500 rows (about 5 per batch) | 11 | 4x / 37x | 1.0x | | 1M rows, partitions of 3,000 rows (about 2 per batch) | 11 | 5x / 37x | 1.0x | | 1M rows, partitions of 6,000 rows (about 1 per batch) | 11 | 5x / 37x | 1.0x | ##### 3M rows, partitions of 60,000 rows 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). | shape | base | PR | base / PR | same result | |---|---:|---:|---|---| | row_number | 28 | 5.6 | 5x [5-5] | yes | | rank (no ties) | 83 | 7.9 | 10x [10-11] | yes | | rank (ties of 4) | 86 | 8.3 | 10x [10-11] | yes | | dense_rank (ties of 4) | 85 | 8.1 | 11x [10-11] | yes | | percent_rank | 14 | 14 | 1.0x [1.0-1.0] | yes | | cume_dist | 13 | 13 | 1.0x [1.0-1.0] | yes | | ntile(4) | 12 | 11 | 1.1x [1.0-1.2] | yes | | lag(v,1) | 39 | 6.8 | 6x [6-6] | yes | | lead(v,1) | 40 | 39 | 1.0x [1.0-1.0] | yes | | lag(v,5) | 39 | 6.1 | 6x [6-6] | yes | | first_value (default frame) | 134 | 11 | 12x [12-12] | yes | | last_value (default frame) | 137 | 13 | 10x [10-10] | yes | | nth_value(v,3) (default frame) | 135 | 11 | 12x [12-12] | yes | | last_value ROWS 5..5 | 15 | 11 | 1.3x [1.3-1.3] | yes | | first_value ROWS UNB..UNB | 22 | 16 | 1.4x [1.2-1.4] | yes | | last_value ROWS UNB..UNB | 19 | 16 | 1.2x [1.2-1.2] | yes | | SUM whole partition | 12 | 13 | 1.0x [0.9-1.0] | yes | | MAX whole partition | 13 | 13 | 1.0x [1.0-1.0] | yes | | SUM ROWS UNB..UNB | 12 | 12 | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 286 | 15 | 20x [20-20] | yes | | SUM default frame, ties of 4 | 234 | 22 | 11x [11-11] | yes | | SUM ROWS UNB..CURRENT | 159 | 11 | 14x [14-14] | yes | | SUM(vn nullable) default frame | 321 | 16 | 20x [20-20] | yes | | SUM(float) default frame | 291 | 14 | 20x [20-20] | yes | | MAX default frame (running max) | 259 | 14 | 19x [19-19] | yes | | MIN default frame (running min) | 260 | 13 | 20x [20-20] | yes | | COUNT default frame | 231 | 12 | 19x [19-19] | yes | | COUNT(vn nullable) default frame | 283 | 13 | 22x [22-22] | yes | | AVG default frame | 238 | 15 | 16x [16-16] | yes | | SUM ROWS 5..5 | 260 | 13 | 21x [20-21] | yes | | SUM ROWS 500..500 | 258 | 13 | 20x [20-21] | yes | | SUM ROWS 2 PRECEDING..CURRENT | 257 | 12 | 22x [22-22] | yes | | SUM ROWS 3 PRECEDING..1 PRECEDING | 259 | 12 | 22x [22-22] | yes | | SUM ROWS CURRENT..UNB FOLLOWING | 181 | 17 | 11x [10-11] | yes | | COUNT ROWS 5..5 | 207 | 11 | 18x [18-18] | yes | | AVG ROWS 5..5 | 216 | 13 | 16x [16-16] | yes | | AVG(float) ROWS 5..5 | 216 | 13 | 16x [16-16] | yes | | SUM(float) ROWS 5..5 | 271 | 13 | 21x [21-21] | yes | | SUM(vn nullable) ROWS 5..5 | 329 | 16 | 21x [21-21] | yes | | MAX ROWS 5..5 | 246 | 14 | 17x [17-17] | yes | | MIN ROWS 5..5 | 246 | 14 | 18x [18-18] | yes | | MAX(vn nullable) ROWS 5..5 | 315 | 18 | 18x [17-18] | yes | | MAX(float) ROWS 5..5 | 247 | 21 | 12x [12-12] | yes | | SUM RANGE 5..5 (offset) | 779 | 21 | 38x [37-38] | yes | | SUM RANGE 10 PRECEDING..CURRENT | 754 | 20 | 38x [37-38] | yes | | SUM RANGE UNB..CURRENT (explicit) | 285 | 14 | 20x [20-21] | yes | | MAX RANGE 5..5 (offset) | 770 | 22 | 35x [35-35] | yes | | SUM GROUPS 2..2 | 315 | 213 | 1.5x [1.5-1.5] | yes | | SUM default frame (timestamp key) | 289 | 16 | 19x [18-19] | yes | | SUM RANGE 5 s PRECEDING..CURRENT (timestamp key) | 859 | 22 | 39x [38-40] | yes | | MAX RANGE 1 min PRECEDING..CURRENT (timestamp key) | 843 | 23 | 36x [36-37] | yes | | SUM default frame (float key) | 285 | 15 | 19x [19-19] | yes | | SUM RANGE 5..5 (unsigned key) | 779 | 23 | 34x [34-35] | yes | | rank (timestamp key) | 85 | 8.4 | 10x [10-10] | yes | | lag(vn,3,default) | 42 | 6.6 | 6x [6-6] | yes | | SUM default frame, ties of 8192 | 237 | 50 | 5x [5-5] | yes | | MAX default frame, ties of 8192 | 191 | 49 | 4x [4-4] | yes | | rank (ties of 8192) | 81 | 8.2 | 10x [10-10] | yes | | 4 aggs ROWS 5..5 (shared window) | 933 | 33 | 28x [28-28] | yes | | 4 aggs default frame (shared window) | 1006 | 34 | 29x [29-29] | yes | ##### 1M rows, partitions of 240 rows 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 31 | 8.0 | 5.8 | 5.7 | 5x [5-5] | 4x [4-4] | 1.4x [1.4-1.4] | 1.0x [1.0-1.0] | yes | | rank (no ties) | 86 | 12 | 7.8 | 7.8 | 11x [11-11] | 7x [7-7] | 1.6x [1.2-1.6] | 1.0x [1.0-1.3] | yes | | lag(v,1) | 44 | 12 | 9.5 | 9.4 | 5x [5-5] | 4x [4-4] | 1.3x [1.3-1.3] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 141 | 18 | 12 | 12 | 12x [12-12] | 8x [8-8] | 1.5x [1.5-1.5] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 12 | 12 | 12 | 12 | 1.0x [1.0-1.0] | 1.0x [0.9-1.0] | 1.0x [1.0-1.1] | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 290 | 20 | 13 | 13 | 22x [22-22] | 15x [14-15] | 1.5x [1.5-1.5] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 240 | 17 | 11 | 11 | 21x [21-21] | 14x [14-14] | 1.5x [1.5-1.5] | 1.0x [1.0-1.0] | yes | | AVG default frame | 246 | 21 | 13 | 13 | 19x [19-19] | 12x [12-12] | 1.6x [1.6-1.6] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 263 | 17 | 13 | 13 | 21x [21-21] | 15x [15-16] | 1.4x [1.3-1.4] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 253 | 19 | 14 | 14 | 18x [18-19] | 13x [13-13] | 1.4x [1.4-1.4] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 796 | 25 | 20 | 20 | 40x [39-40] | 31x [31-32] | 1.3x [1.3-1.3] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 949 | 50 | 34 | 34 | 28x [27-28] | 19x [19-19] | 1.5x [1.5-1.5] | 1.0x [1.0-1.0] | yes | ##### 1M rows, partitions of 10 rows 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 92 | 59 | 6.1 | 6.0 | 15x [15-16] | 1.6x [1.6-1.6] | 10x [10-10] | 1.0x [1.0-1.0] | yes | | rank (no ties) | 189 | 113 | 7.9 | 7.9 | 24x [24-24] | 1.7x [1.7-1.7] | 14x [14-14] | 1.0x [1.0-1.1] | yes | | lag(v,1) | 140 | 136 | 9.5 | 9.6 | 15x [15-15] | 1.0x [1.0-1.0] | 14x [14-14] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 244 | 140 | 12 | 12 | 20x [20-20] | 1.7x [1.7-1.8] | 11x [11-11] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 58 | 59 | 59 | 58 | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 414 | 154 | 14 | 14 | 29x [29-29] | 2.7x [2.7-2.7] | 11x [11-11] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 359 | 147 | 12 | 12 | 31x [31-31] | 2.4x [2.4-2.4] | 13x [13-13] | 1.0x [1.0-1.0] | yes | | AVG default frame | 378 | 163 | 13 | 13 | 28x [28-28] | 2.3x [2.3-2.3] | 12x [12-12] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 285 | 124 | 15 | 15 | 19x [19-19] | 2.3x [2.3-2.4] | 8x [8-8] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 280 | 132 | 14 | 14 | 20x [19-20] | 2.1x [2.1-2.1] | 9x [9-9] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 796 | 133 | 23 | 23 | 35x [35-35] | 6x [6-6] | 6x [6-6] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 946 | 396 | 38 | 38 | 25x [24-25] | 2.4x [2.4-2.5] | 10x [10-10] | 1.0x [1.0-1.0] | yes | ##### 500k rows, partitions of 4 rows 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 190 | 145 | 7.3 | 7.3 | 26x [26-26] | 1.3x [1.3-1.3] | 20x [20-20] | 1.0x [1.0-1.0] | yes | | rank (no ties) | 351 | 279 | 8.8 | 8.9 | 39x [39-40] | 1.3x [1.2-1.3] | 32x [31-32] | 1.0x [1.0-1.0] | yes | | lag(v,1) | 294 | 342 | 10 | 10 | 28x [28-29] | 0.9x [0.9-0.9] | 33x [33-33] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 401 | 332 | 13 | 13 | 30x [30-30] | 1.2x [1.2-1.2] | 25x [25-25] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 138 | 131 | 145 | 146 | 0.9x [0.9-0.9] | 1.0x [1.0-1.1] | 0.9x [0.9-0.9] | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 586 | 367 | 15 | 15 | 38x [37-38] | 1.6x [1.6-1.6] | 25x [24-25] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 533 | 354 | 13 | 13 | 41x [41-42] | 1.5x [1.5-1.5] | 27x [27-28] | 1.0x [1.0-1.0] | yes | | AVG default frame | 565 | 388 | 15 | 15 | 38x [38-38] | 1.5x [1.5-1.5] | 26x [26-27] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 406 | 293 | 21 | 21 | 19x [19-19] | 1.4x [1.4-1.4] | 14x [14-14] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 405 | 296 | 14 | 14 | 28x [28-29] | 1.4x [1.4-1.4] | 21x [20-21] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 888 | 308 | 30 | 30 | 29x [29-29] | 2.9x [2.9-2.9] | 10x [10-10] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 1247 | 889 | 44 | 44 | 28x [28-29] | 1.4x [1.4-1.4] | 20x [20-21] | 1.0x [1.0-1.0] | yes | ##### 300k rows, partitions of 2 rows 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 362 | 296 | 8.9 | 8.9 | 40x [40-41] | 1.2x [1.2-1.2] | 33x [32-34] | 1.0x [1.0-1.1] | yes | | rank (no ties) | 623 | 563 | 10 | 10 | 60x [60-61] | 1.1x [1.1-1.1] | 55x [55-55] | 1.0x [1.0-1.0] | yes | | lag(v,1) | 561 | 691 | 12 | 12 | 48x [47-48] | 0.8x [0.8-0.8] | 59x [58-59] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 678 | 662 | 16 | 16 | 43x [43-43] | 1.0x [1.0-1.0] | 42x [42-42] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 251 | 255 | 261 | 250 | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.1] | yes | | SUM default frame (RANGE running) | 896 | 729 | 17 | 17 | 53x [53-54] | 1.2x [1.2-1.2] | 42x [42-42] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 837 | 711 | 15 | 15 | 56x [56-56] | 1.2x [1.2-1.2] | 48x [48-48] | 1.0x [1.0-1.0] | yes | | AVG default frame | 900 | 773 | 16 | 16 | 55x [55-55] | 1.2x [1.2-1.2] | 47x [47-47] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 745 | 589 | 32 | 32 | 23x [23-23] | 1.3x [1.3-1.3] | 19x [18-19] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 747 | 593 | 18 | 18 | 41x [41-41] | 1.3x [1.3-1.3] | 33x [33-34] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 1233 | 609 | 47 | 46 | 27x [27-27] | 2.0x [2.0-2.0] | 13x [13-13] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 2279 | 1763 | 59 | 60 | 38x [38-38] | 1.3x [1.3-1.3] | 30x [30-30] | 1.0x [1.0-1.0] | yes | ##### 300k rows, partitions of 1 row 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 659 | 605 | 10 | 10 | 64x [62-66] | 1.1x [1.1-1.1] | 60x [60-62] | 1.0x [1.0-1.0] | yes | | rank (no ties) | 1133 | 1156 | 11 | 11 | 102x [94-103] | 1.0x [1.0-1.0] | 105x [104-106] | 1.0x [0.9-1.0] | yes | | lag(v,1) | 1129 | 1055 | 12 | 12 | 95x [94-95] | 1.1x [1.1-1.1] | 90x [89-90] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 1142 | 1100 | 16 | 16 | 70x [69-72] | 1.0x [1.0-1.1] | 68x [68-68] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 552 | 535 | 529 | 523 | 1.1x [1.0-1.1] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.1] | yes | | SUM default frame (RANGE running) | 1529 | 1397 | 19 | 19 | 81x [80-82] | 1.1x [1.1-1.1] | 74x [74-76] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 1456 | 1382 | 16 | 16 | 90x [88-91] | 1.1x [1.0-1.1] | 86x [85-86] | 1.0x [1.0-1.0] | yes | | AVG default frame | 1571 | 1499 | 18 | 18 | 88x [85-90] | 1.0x [1.0-1.1] | 84x [83-87] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 1416 | 1145 | 20 | 20 | 71x [70-71] | 1.2x [1.2-1.3] | 58x [57-58] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 1440 | 1191 | 20 | 20 | 72x [69-72] | 1.2x [1.2-1.2] | 60x [58-62] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 1920 | 1176 | 43 | 43 | 45x [45-46] | 1.6x [1.6-1.7] | 28x [27-28] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 4336 | 3497 | 51 | 52 | 84x [83-84] | 1.2x [1.2-1.3] | 69x [66-70] | 1.0x [1.0-1.0] | yes | ##### 3M rows, one partition 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). | shape | base | PR | base / PR | same result | |---|---:|---:|---|---| | row_number | 27 | 5.7 | 5x [5-5] | yes | | rank (no ties) | 83 | 7.7 | 11x [11-11] | yes | | lag(v,1) | 37 | 5.8 | 7x [7-7] | yes | | first_value (default frame) | 135 | 10 | 13x [13-13] | yes | | SUM whole partition | 12 | 11 | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 283 | 14 | 20x [20-20] | yes | | COUNT default frame | 231 | 12 | 19x [19-20] | yes | | AVG default frame | 238 | 14 | 17x [17-17] | yes | | SUM ROWS 5..5 | 259 | 12 | 21x [21-21] | yes | | MAX ROWS 5..5 | 245 | 14 | 18x [18-18] | yes | | SUM RANGE 5..5 (offset) | 779 | 21 | 38x [38-39] | yes | | 4 aggs ROWS 5..5 (shared window) | 933 | 33 | 28x [28-29] | yes | ##### 1M rows, partitions of 600 rows (about 13 complete partitions per input batch) 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 30 | 6.7 | 5.8 | 5.8 | 5x [5-5] | 4x [4-4] | 1.2x [1.1-1.2] | 1.0x [1.0-1.0] | yes | | rank (no ties) | 85 | 9.9 | 7.8 | 7.8 | 11x [11-11] | 9x [9-9] | 1.3x [1.3-1.3] | 1.0x [1.0-1.0] | yes | | lag(v,1) | 42 | 8.7 | 9.4 | 9.3 | 4x [4-4] | 5x [5-5] | 0.9x [0.9-0.9] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 139 | 15 | 12 | 12 | 12x [11-12] | 9x [9-9] | 1.2x [1.2-1.2] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 10 | 10 | 10 | 10 | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 287 | 17 | 13 | 13 | 22x [21-22] | 17x [17-17] | 1.3x [1.3-1.3] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 236 | 14 | 11 | 11 | 21x [20-21] | 17x [17-17] | 1.2x [1.2-1.2] | 1.0x [1.0-1.0] | yes | | AVG default frame | 241 | 17 | 13 | 13 | 18x [18-18] | 14x [14-14] | 1.3x [1.3-1.3] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 261 | 15 | 13 | 13 | 20x [20-21] | 18x [18-18] | 1.1x [1.1-1.2] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 250 | 16 | 14 | 14 | 18x [18-18] | 16x [16-16] | 1.2x [1.2-1.2] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 783 | 23 | 20 | 20 | 39x [39-39] | 35x [35-35] | 1.1x [1.1-1.1] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 958 | 41 | 34 | 34 | 28x [28-28] | 24x [24-24] | 1.2x [1.2-1.2] | 1.0x [1.0-1.0] | yes | ##### 1M rows, partitions of 1,500 rows (about 5 per batch) 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 29 | 6.1 | 5.9 | 5.9 | 5x [5-5] | 5x [4-5] | 1.0x [1.0-1.1] | 1.0x [1.0-1.0] | yes | | rank (no ties) | 84 | 8.9 | 7.9 | 8.0 | 10x [10-11] | 9x [9-10] | 1.1x [1.1-1.1] | 1.0x [0.9-1.0] | yes | | lag(v,1) | 40 | 7.5 | 9.2 | 9.2 | 4x [4-4] | 5x [5-5] | 0.8x [0.8-0.8] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 139 | 14 | 13 | 13 | 11x [11-11] | 10x [10-10] | 1.1x [1.0-1.1] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 9.7 | 9.8 | 9.5 | 9.4 | 1.0x [1.0-1.1] | 1.0x [1.0-1.0] | 1.0x [1.0-1.1] | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 287 | 15 | 14 | 14 | 20x [20-21] | 19x [18-19] | 1.1x [1.1-1.1] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 235 | 13 | 12 | 12 | 19x [19-19] | 18x [18-18] | 1.1x [1.0-1.1] | 1.0x [1.0-1.0] | yes | | AVG default frame | 239 | 15 | 13 | 14 | 18x [18-18] | 16x [16-16] | 1.1x [1.1-1.1] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 260 | 14 | 13 | 13 | 20x [20-20] | 19x [19-19] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 249 | 15 | 14 | 14 | 17x [17-18] | 17x [17-17] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 781 | 22 | 21 | 21 | 37x [37-37] | 36x [36-37] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 945 | 37 | 36 | 35 | 27x [27-27] | 26x [25-26] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | ##### 1M rows, partitions of 3,000 rows (about 2 per batch) 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 27 | 6.3 | 6.3 | 6.1 | 5x [4-5] | 4x [4-4] | 1.0x [1.0-1.0] | 1.0x [1.0-1.1] | yes | | rank (no ties) | 83 | 8.6 | 8.6 | 8.6 | 10x [10-10] | 10x [10-10] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | lag(v,1) | 39 | 7.2 | 7.2 | 7.1 | 6x [5-6] | 6x [5-6] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 138 | 13 | 13 | 13 | 11x [10-11] | 11x [11-11] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 9.4 | 9.5 | 9.2 | 9.1 | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 285 | 15 | 15 | 15 | 20x [19-20] | 19x [19-19] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 233 | 13 | 13 | 12 | 19x [19-19] | 18x [18-19] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | AVG default frame | 237 | 15 | 15 | 15 | 16x [16-16] | 16x [16-16] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 259 | 13 | 13 | 13 | 20x [20-20] | 20x [19-20] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 247 | 14 | 15 | 15 | 17x [17-17] | 17x [17-17] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 780 | 21 | 21 | 21 | 37x [37-38] | 37x [36-37] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 940 | 36 | 35 | 35 | 27x [26-27] | 26x [26-26] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | ##### 1M rows, partitions of 6,000 rows (about 1 per batch) 3 rounds. ns per row; `base` = upstream main, `PR` = this branch; ratio base / PR (above 1 = PR faster). Layers: `1-2` = bulk frames, accumulator / evaluator hooks (direct evaluation of complete partitions and column pruning off); `1-3` = plus direct evaluation of complete partitions (column pruning off); `PR` = all four layers. Each ratio compares two builds that ran back to back in the same round. | shape | base | 1-2 | 1-3 | PR | base / PR | base / 1-2 | 1-2 / 1-3 | 1-3 / PR | same result | |---|---:|---:|---:|---:|---|---|---|---|---| | row_number | 28 | 6.2 | 6.2 | 6.1 | 5x [4-5] | 5x [4-5] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | rank (no ties) | 83 | 8.6 | 8.7 | 8.7 | 10x [10-10] | 10x [10-10] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | lag(v,1) | 39 | 7.2 | 7.2 | 7.1 | 6x [6-6] | 6x [5-6] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | first_value (default frame) | 137 | 13 | 13 | 13 | 11x [11-11] | 11x [11-11] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM whole partition | 9.6 | 9.2 | 9.5 | 9.2 | 1.0x [1.0-1.0] | 1.0x [1.0-1.1] | 1.0x [0.9-1.0] | 1.0x [1.0-1.0] | yes | | SUM default frame (RANGE running) | 287 | 15 | 15 | 15 | 19x [19-20] | 19x [19-19] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | COUNT default frame | 234 | 13 | 13 | 13 | 18x [18-18] | 18x [18-18] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | AVG default frame | 239 | 15 | 15 | 15 | 16x [16-16] | 16x [16-16] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 261 | 13 | 13 | 13 | 20x [20-20] | 20x [19-20] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | MAX ROWS 5..5 | 249 | 14 | 15 | 15 | 17x [17-17] | 17x [17-17] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | SUM RANGE 5..5 (offset) | 782 | 21 | 21 | 21 | 37x [37-37] | 37x [37-37] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | | 4 aggs ROWS 5..5 (shared window) | 942 | 35 | 35 | 35 | 27x [27-27] | 27x [27-27] | 1.0x [1.0-1.0] | 1.0x [1.0-1.0] | yes | #### Columns that only pass through the window ##### 500k rows, partitions of 10 rows, 11 columns (no extra column) 3 rounds. `PR` = this branch, `PR, no pruning` = the same tree with `PrunedInput` always keeping every column. | shape | base | PR, no pruning | PR | base / PR | no pruning / PR (gain of the column pruning) | same result | |---|---:|---:|---:|---|---|---| | row_number | 93 | 6.8 | 6.8 | 14x [14-14] | 1.0x [1.0-1.0] | yes | | rank (no ties) | 189 | 8.6 | 8.6 | 22x [22-22] | 1.0x [1.0-1.0] | yes | | lag(v,1) | 141 | 10 | 10 | 14x [13-14] | 1.0x [1.0-1.0] | yes | | SUM ROWS 5..5 | 285 | 16 | 16 | 18x [18-18] | 1.0x [1.0-1.0] | yes | | GROUPS 2..2 x3 (generic) | 835 | 563 | 552 | 1.5x [1.5-1.5] | 1.0x [1.0-1.0] | yes | | SUM FILTER default frame (generic) | 664 | 547 | 553 | 1.2x [1.2-1.2] | 1.0x [1.0-1.0] | yes | | STDDEV default frame (generic) | 399 | 303 | 304 | 1.3x [1.3-1.3] | 1.0x [1.0-1.0] | yes | | mixed: SUM ROWS 5..5 + GROUPS + row_number | 616 | 226 | 218 | 2.8x [2.8-2.8] | 1.0x [1.0-1.0] | yes | ##### 500k rows, partitions of 10 rows, 51 columns (40 extra columns carried through the window) 3 rounds. `PR` = this branch, `PR, no pruning` = the same tree with `PrunedInput` always keeping every column. | shape | base | PR, no pruning | PR | base / PR | no pruning / PR (gain of the column pruning) | same result | |---|---:|---:|---:|---|---|---| | row_number | 401 | 14 | 12 | 34x [33-35] | 1.1x [1.1-1.2] | yes | | rank (no ties) | 490 | 15 | 14 | 36x [36-36] | 1.1x [1.1-1.1] | yes | | lag(v,1) | 452 | 17 | 15 | 29x [29-29] | 1.1x [1.1-1.1] | yes | | SUM ROWS 5..5 | 621 | 40 | 39 | 16x [16-16] | 1.0x [1.0-1.0] | yes | | GROUPS 2..2 x3 (generic) | 1181 | 972 | 573 | 2.1x [2.1-2.1] | 1.7x [1.7-1.7] | yes | | SUM FILTER default frame (generic) | 1002 | 855 | 578 | 1.7x [1.7-1.7] | 1.5x [1.5-1.5] | yes | | STDDEV default frame (generic) | 735 | 615 | 327 | 2.3x [2.3-2.3] | 1.9x [1.9-1.9] | yes | | mixed: SUM ROWS 5..5 + GROUPS + row_number | 953 | 535 | 242 | 4x [4-4] | 2.2x [2.2-2.2] | yes | ##### 500k rows, partitions of 10 rows, 161 columns (150 extra columns carried through the window) 3 rounds. `PR` = this branch, `PR, no pruning` = the same tree with `PrunedInput` always keeping every column. | shape | base | PR, no pruning | PR | base / PR | no pruning / PR (gain of the column pruning) | same result | |---|---:|---:|---:|---|---|---| | row_number | 1229 | 32 | 25 | 48x [48-49] | 1.3x [1.3-1.3] | yes | | rank (no ties) | 1326 | 34 | 27 | 49x [49-49] | 1.2x [1.2-1.2] | yes | | lag(v,1) | 1293 | 36 | 29 | 44x [44-45] | 1.2x [1.2-1.2] | yes | | SUM ROWS 5..5 | 1498 | 101 | 94 | 16x [16-16] | 1.1x [1.1-1.1] | yes | | GROUPS 2..2 x3 (generic) | 2063 | 2036 | 633 | 3x [3-3] | 3x [3-3] | yes | | SUM FILTER default frame (generic) | 1869 | 1683 | 633 | 3.0x [2.9-3.0] | 2.7x [2.6-2.7] | yes | | STDDEV default frame (generic) | 1625 | 1434 | 384 | 4x [4-4] | 4x [4-4] | yes | | mixed: SUM ROWS 5..5 + GROUPS + row_number | 1831 | 1352 | 299 | 6x [6-6] | 5x [4-5] | yes | #### Noise floor: the PR binary twice (A/A), 1M rows, partitions of 10 rows 5 rounds; ratio a / b. | shape | a / b median [lowest-highest] | |---|---| | row_number | 1.0x [1.0-1.0] | | rank (no ties) | 1.0x [1.0-1.0] | | lag(v,1) | 1.0x [1.0-1.0] | | first_value (default frame) | 1.0x [1.0-1.0] | | SUM whole partition | 1.0x [0.9-1.0] | | SUM default frame (RANGE running) | 1.0x [1.0-1.0] | | COUNT default frame | 1.0x [1.0-1.0] | | AVG default frame | 1.0x [1.0-1.0] | | SUM ROWS 5..5 | 1.0x [1.0-1.0] | | MAX ROWS 5..5 | 1.0x [1.0-1.0] | | SUM RANGE 5..5 (offset) | 1.0x [1.0-1.0] | | 4 aggs ROWS 5..5 (shared window) | 1.0x [1.0-1.0] | Largest single-round deviation from 1.0: 1.06x. </details> ## What is the testing strategy for this PR? * **sqllogictest `window_bulk_evaluation.slt`** (new, 84 records). A table of 600 rows in 35 partitions of 1 to 35 rows; twenty-one families (moving `ROWS` frames, following and preceding frames, running frames, `RANGE` with offsets, `RANGE ... DESC`, `RANGE` offsets over unsigned, signed, timestamp (seconds; nanoseconds in UTC) and nullable keys (`ASC` / `DESC` with `NULLS FIRST` / `LAST`: four windows, each in a query of its own, because the planner evaluates a window backwards when its ordering is the reverse of the sort below it), peers, default frame, ranks, `lag` / `lead` over numbers and over text and dictionary-encoded columns with defaults, `first_value` / `last_value`, `FILTER`, `GROUPS`, mixed expressions, string and null keys). Every window result is compared with an *oracle* that computes the same value with a self-join and an ordinary aggregate, so the test does not depend on the code under test: 21 families × 600 rows = 12,600 compared rows per run, asserted at bat ch sizes 8192, 1, 7, 50, 70, 100 and 599 (at 70 and 100 a batch starts inside a partition and holds four or more complete partitions after it, which is where a directly evaluated run is spliced in after the rows of the open partition). A second part runs five window functions over a table of 13 columns of different types (lists, structs, decimals, timestamps, strings), six of which they read, and compares the result with the same windows over the narrow table (7 pass-through columns must come out unchanged), at batch sizes 8192, 7 and 1; and a window that reads no column at all. The file passes unchanged on `main` (4c7b30752): it checks behaviour, not the new code, so it can be reviewed and merged first. * **Unit tests** for `PrunedInput` (kept columns and remapped indices, projection with the row count, identity when every column is read, zero columns). The frame and kernel code has no unit tests of its own: following the [testing guide](https://datafusion.apache.org/contributor-guide/testing.html#choosing-what-kind-of-test-to-write) it is tested through SQL, by the file above and by the existing extended fuzz cases (`datafusion/core/tests/fuzz_cases/window_fuzz.rs`, which compare `BoundedWindowAggExec` with `WindowAggExec` on random data in `Sorted`, `Linear` and `PartiallySorted` modes). * **Existing suites**, run on this branch with the project's `ci` profile (debug assertions and overflow checks on): the whole sqllogictest suite (305 files), the unit tests of `datafusion-physical-plan`, `-physical-expr`, `-expr`, `-expr-common`, `-functions-window`, `-functions-aggregate` and `-functions-aggregate-common`, the 126 `window` tests of `core_integration` and the 4 `window` cases of `extended_tests`: all pass. `cargo fmt`, `cargo clippy -D warnings` (both CI feature sets), `cargo doc -D warnings` and `typos` are clean. * **Differential fuzzing against `main`** (not part of the PR; the harness and the logs can be shared): the same random window queries run on a build of `main` and a build of this branch, with a digest of every result (floats by bits, every `NaN` as equal). Random frames (`ROWS` / `RANGE` / `GROUPS`, offsets), functions (aggregates with and without `FILTER`, ranking, `lag`/`lead`, `first`/`last`/`nth_value`), ORDER BY keys of 15 kinds (signed and unsigned integers of every width with extreme values, floats, decimals, dates, three timestamp units, strings, nullable keys; each with `ASC` / `DESC` and `NULLS FIRST` / `LAST` in all four combinations), string / nullable / two-column PARTITION BY keys, batch sizes 1 to 8192, `Sorted` and `Linear` modes, long partitions, and every other column of the table carried through the window. Result: **0 differences** in 314,000 queries on x86-64 and in 308,531 on arm64 (every digest identical). The configurations: all functions (120,000 queries) , aggregates only (80,000), standard (non-aggregate) window functions only (80,000), all functions and standard functions only, each with every other column of the table returned through the window (12,000 queries each; on arm64 the first of these two was stopped after 6,531), and long partitions with big batches (6,000, and 4,000 with the columns carried). About one query in five is an error (an invalid frame for the key type); both builds report it identically, because the digest of an error is the hash of its text. The runs were made on the branch before its last edits, which only removed comments, added a lint attribute and turned an `else if let ... else { return None }` into an early return with `?`. * **Mutation check**: fourteen one-line faults (not part of the PR; the harness and the logs can be shared), applied one at a time to a copy of the tree and followed by this sqllogictest file (by the unit tests of `PrunedInput` for the last one). A fault counts as detected only if the tests fail after a successful build. The faults: the splice position of a directly evaluated run of partitions; `COUNT` and the sliding `MIN` / `MAX` queue not restarting at a partition boundary; a `ROWS` frame ending one row early; a `RANGE` offset search that compares in ascending order for a `DESC` key; four ways of ordering NULL keys wrongly (`NULLS FIRST` and `NULLS LAST`, in the start search and in the end search); a frame ending one row before the end of its peer group; the unit of an interval offset on a `Timestamp(Second)` key; `lag` reading the previous partition; the columns of a pruned batch in reverse order; a column index off by one in a rewritten expression. **All fourteen are detected .** Two of them survived earlier versions of the file, and each led to a change of it: a directly evaluated run spliced before the rows that complete the open partition is visible only when a batch starts inside a partition and holds four or more complete partitions after it (batch sizes 70 and 100 were added); and a `NULLS FIRST` ascending key was never evaluated as written, because the planner runs a window backwards when its ordering is the reverse of the sort below it (`ASC NULLS FIRST` and `DESC NULLS LAST` shared one sort), so each combination of direction and NULL placement is now a window in a query of its own. ## Are there any user-facing changes? No change to SQL, plans, `EXPLAIN` output or results. New public API (provided methods and getters with a default that keeps today's behaviour, so nothing breaks): `Accumulator::evaluate_window_frames`, `PartitionEvaluator::evaluate_rows`, `PlainAggregateWindowExpr::get_filter_expr`, `SlidingAggregateWindowExpr::get_filter_expr`. Documentation for UDAF / UDWF authors can follow once the shape of the hooks is agreed. ## Limits, risks and open questions * **API shape.** Is a provided method on `Accumulator` / `PartitionEvaluator` the right way, or should the kernels sit behind the kernel direction of #23273? The hooks are small, documented and default to today's behaviour; they can be hidden or moved without touching the layers around them. * **A partition that continues across batches** still pays for the carried tail (`PartitionBatchState::extend`, and the full-width `concat_batches` of `input_buffer`). Frames that look ahead (`FOLLOWING`) and `Linear` mode with interleaved partitions are where it shows; pruning shrinks the first, the second needs a different buffer structure. * **No fast path** for `GROUPS` frames, `FILTER`, `IGNORE NULLS`, `lead` in the stream, `percent_rank` / `cume_dist` / `ntile`, float / decimal / date `RANGE` offsets, named-timezone timestamp keys, and user-defined window functions: they run the generic protocol and benefit from layers 3 and 4 only. Measured on x86-64 with 10-row partitions, the shapes without a fast path (`GROUPS 2..2`, `SUM ... FILTER`, `STDDEV`) are 1.2-1.5x faster at 11 columns, 1.7-2.3x at 51 columns and 3-4x at 161 columns, and a window that mixes a moving `SUM` with a `GROUPS` frame and `row_number` is 2.8x, 4x and 6x faster; at 60,000-row partitions `percent_rank`, `cume_dist` and `lead` are unchanged (1.0x), `ntile` is 1.1x and `SUM` over `GROUPS 2..2` is 1.5x. * **`MIN_PARTITIONS_FOR_DIRECT_EVALUATION = 4`** is a heuristic. The 600 to 6,000-row sections of the details above show both sides of it (about 13 and 5 complete partitions per batch use the direct path, about 2 and 1 do not). Where it engages, it is neutral to 1.3x faster than layers 1 and 2 alone for the aggregates and ranking functions, and 7-19% slower for `lag` (the segmented form gathers by index where the per-partition form slices; not profiled); the PR as a whole is still 4x faster than `main` there. It is a constant on purpose and the first thing to change if the maintainers prefer another trade-off. * **User-defined window functions** receive `PartitionEvaluatorArgs::input_exprs` that refer to the projected schema (column *indices* change; names, types, nullability and literals do not). A function that depends on the index of its input column would see a difference. * **`NaN` sign.** The sign of a `NaN` that arithmetic generates (`inf - inf`) is hardware dependent and can differ between a 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). * **Not measured**: `target_partitions > 1`, spilling and memory accounting (#26707), I/O-bound queries, CI platforms other than x86-64 and arm64 Linux, and the `release` profile that `cargo bench` uses (LTO, one codegen unit): every timing here is from `release-nonlto` builds of both sides. * **Size.** 24 files, +4,251 / -152 lines: 2,023 lines are the three new engine files (`bulk.rs`, `window_kernels.rs`, `pruned_input.rs`, the last with its unit tests), 726 the new sqllogictest file and 129 the benchmark; the other 1,373 are the hooks, their implementations for the built-in functions, and the code that drives them. The layers are independent enough to be split (below). **AI assistance.** This change, its benchmarks and this description were prepared with an AI coding assistant, at the direction of the account owner. Every number above comes from a log of the runs described, and the unknowns are listed in the section above. The PR stays a draft until its author has read all of it. ## Relation to other work * **#26541** (assigned): its first point (keep only the needed columns in the partition batches) is layer 4; its second (the fixed cost per partition) is layer 3. #27400 asks how to coordinate; if the assignee prefers to take those parts, the layers are independent, and this PR can shrink to layers 1 and 2. * **#23273** (`WindowCall` / `WindowKernel`) and **#15607** (vectorized evaluation): a longer-term direction; the hooks here are provided methods with fallbacks, so they do not preclude it. * **#22355 / #22356** (intra-operator parallelism of one large partition) targets a different shape; the two compose. * **#23197** (epic) and **#26222** (`inf - inf` in a sliding sum): see the notes above. * **#24508** (draft; where the arguments of a window aggregate with `FILTER` are evaluated, in `window_expr.rs`) and **#27207** (protobuf of window expressions, `windows/mod.rs`) touch files that this PR also changes. I expect textual conflicts only; whichever lands second rebases. `FILTER` keeps the per-row protocol here. ## Review guide and split plan Suggested reading order: `datafusion/expr/src/window_state/bulk.rs` (frames) and the two new trait methods; `window_kernels.rs`; `physical-expr/src/window/{window_expr,standard}.rs` (how the hooks are driven); `bounded_window_agg_exec.rs` (`SortedSearch::update_partition_batch`, `get_aggregate_result_out_column`); `pruned_input.rs`; the slt file. If smaller PRs are preferred, the split is: **(A)** layers 1 and 2 (frames, hooks, kernels, benchmarks, the oracle slt), about 60% of the lines; **(B)** layer 3 (direct evaluation of complete partitions), about a fifth; **(C)** layer 4 (column pruning), about a fifth. They apply in that order of dependency, and the table in [Which layer produces the gain](#performance) gives each one's share of the speedup: (A) alone is 3.6-36x faster at 240 to 1,500 rows per partition, 1.0-6x at 10 rows and 0.8-2.9x at 1 to 4 rows (`lag` is the 0.8-1.0x case at 2 to 10 rows); (B) is what makes partitions of 1 to 10 rows fast (another 5.8-105x on top of (A)); (C) matters for inputs with many columns. -- 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]
