comphead opened a new issue, #26541: URL: https://github.com/apache/datafusion/issues/26541
### Is your feature request related to a problem or challenge? In sorted mode, `BoundedWindowAggExec` keeps state for every PARTITION BY group it sees: - `SortedSearch::evaluate_partition_batches` builds a `Vec<ScalarValue>` key and a slice of every input column for each group in a batch ([bounded_window_agg_exec.rs#L1063-L1092](https://github.com/apache/datafusion/blob/3ee3e394c19c97ccc7cca7ed3becb75cf364746a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs#L1063-L1092)). - `update_partition_batch` validates the slice into a `RecordBatch` and inserts it into an `IndexMap` ([#L667-L700](https://github.com/apache/datafusion/blob/3ee3e394c19c97ccc7cca7ed3becb75cf364746a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs#L667-L700)). - Each window expression creates an evaluator and a `WindowAggState` under a clone of the key, and `prune_partition_batches` drops all of it again ([#L1417-L1470](https://github.com/apache/datafusion/blob/3ee3e394c19c97ccc7cca7ed3becb75cf364746a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs#L1417-L1470)). When groups hold one or a few rows, this bookkeeping dominates. On 1M one-row partitions, `max(v) OVER (PARTITION BY k ORDER BY i)` takes 3x as long as the same function over `ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING`, which runs in `WindowAggExec` and returns the same values when every partition has one row. Columns that only pass through the window widen the gap, because every partition batch slices every input column although the window expressions read only their arguments and keys. ```sql SET datafusion.execution.target_partitions = 1; CREATE TABLE t AS SELECT value AS k, value AS i, value AS v, concat('carried-0-', value) AS c0, concat('carried-1-', value) AS c1, concat('carried-2-', value) AS c2, concat('carried-3-', value) AS c3, concat('carried-4-', value) AS c4, concat('carried-5-', value) AS c5, concat('carried-6-', value) AS c6, concat('carried-7-', value) AS c7, concat('carried-8-', value) AS c8, concat('carried-9-', value) AS c9 FROM generate_series(1, 1000000); SELECT count(m) FROM (SELECT max(v) OVER (PARTITION BY k ORDER BY i) AS m FROM t); ``` | Query | `BoundedWindowAggExec` | `WindowAggExec` (`UNBOUNDED FOLLOWING`) | |---|---:|---:| | `max(v)` | 0.63 s | 0.21 s | | `max(v)` and `count` of 10 carried string columns | 0.96 s | 0.44 s | ### Describe the solution you'd like - Keep only the columns the window expressions read in the partition batches. The output columns already come from `input_buffer`. - In sorted mode, evaluate a partition that starts and ends inside one input batch directly on its range of the batch, with no key, partition batch or per-partition state. Only the partition still open at the end of the batch needs to be carried into the next one. ### Describe alternatives you've considered Cheaper keys, for example row-format keys instead of `Vec<ScalarValue>`, would reduce the hashing and allocation but keep the per-partition batches and state. ### Additional context The timings are from `datafusion-cli` built from 3ee3e394c with the `release-nonlto` profile on an Apple M5 Max. In a profile of Comet's window benchmark at one row per partition with 18 carried columns, slicing the per-partition batches took 20% of the time and dropping them during pruning another 18%. -- 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]
