1fanwang opened a new pull request, #25739:
URL: https://github.com/apache/datafusion/pull/25739
## Which issue does this PR close?
- Part of #25248.
## Rationale for this change
Planning a `SELECT` with many aggregate expressions takes time that grows
with the square of the number of expressions. The report measured 2.7 s of SQL
planning for 2,500 items, and a profile shows almost all of it goes into
building the projection above the aggregate.
For each select item, the planner replaces sub-expressions that the
aggregate already computes with references to its output columns. It rebuilt
the lookup table of those expressions for every item, so N items built N tables
of N entries. Hashing an aggregate expression also hashes its function
signature, which makes each rebuild slow.
## What changes are included in this PR?
- The projection builder now builds that table once and reuses it for every
item, through a crate-private `Columnizer`. `columnize_expr` keeps its
signature and uses the same helper. This follows #25010, which reuses the
column normalization context in the same function.
- The `sql_planner` benchmark gains a case with 1,000 aggregates.
Very wide selects still do some quadratic work. Two SQL planner helpers,
`rebase_expr` and `find_aggregate_exprs`, compare each expression against a
list, and schema field lookups scan every field. I'd rather handle those in
separate PRs, so this one only references the issue.
## What is the testing strategy for this PR?
The rewrite gives the same result as before, so the existing expression, SQL
planner and optimizer tests and the sqllogictest suite cover it, and they pass
locally.
`sql_planner` benchmark against `main` on an 18-core M-series Mac,
`release-nonlto`:
| Benchmark | main | this PR | change |
| --- | --- | --- | --- |
| `logical_wide_aggregate_1000_exprs` (new) | 182.3 ms | 31.1 ms | -83% |
| `logical_wide_aggregate_100_exprs` | 2.43 ms | 0.89 ms | -63% |
| `physical_select_aggregates_from_200` | 6.98 ms | 4.69 ms | -33% |
The query shape from the issue, `CAST(sum(id + i) AS VARCHAR)` repeated N
times, planned with `EXPLAIN` in `datafusion-cli` (unoptimized `ci` build, so
absolute times are high):
| N | main | this PR |
| --- | --- | --- |
| 500 | 0.72 s | 0.66 s |
| 1000 | 1.56 s | 0.87 s |
| 2000 | 5.05 s | 1.83 s |
| 4000 | 18.83 s | 4.57 s |
<details>
<summary>Commands and raw output</summary>
Benchmarks. The new case is copied onto `main` so both runs measure the same
queries. The `sql_planner` bench checks that `benchmarks/data/hits_partitioned`
exists at startup; I used a one-row stand-in, since none of these three cases
read it.
```
PR=eb8f5ce9d4d0b50cffb97648188c84a4034ec92c
git fetch origin main "$PR"
git switch --detach origin/main
git checkout "$PR" -- datafusion/core/benches/sql_planner.rs
cargo bench -p datafusion --bench sql_planner --profile release-nonlto -- \
'logical_wide_aggregate|physical_select_aggregates_from_200'
--save-baseline main
git checkout -f "$PR"
cargo bench -p datafusion --bench sql_planner --profile release-nonlto -- \
'logical_wide_aggregate|physical_select_aggregates_from_200' --baseline
main
```
`main`:
```
physical_select_aggregates_from_200
time: [6.8908 ms 6.9799 ms 7.0789 ms]
logical_wide_aggregate_100_exprs
time: [2.3629 ms 2.4285 ms 2.4977 ms]
logical_wide_aggregate_1000_exprs
time: [180.60 ms 182.29 ms 184.28 ms]
```
This PR:
```
physical_select_aggregates_from_200
time: [4.6522 ms 4.6890 ms 4.7269 ms]
change: [−33.854% −32.821% −31.767%] (p = 0.00 <
0.05)
Performance has improved.
logical_wide_aggregate_100_exprs
time: [881.22 µs 888.79 µs 897.96 µs]
change: [−63.824% −62.709% −61.533%] (p = 0.00 <
0.05)
Performance has improved.
logical_wide_aggregate_1000_exprs
time: [30.705 ms 31.131 ms 31.628 ms]
change: [−83.222% −82.922% −82.577%] (p = 0.00 <
0.05)
Performance has improved.
```
`datafusion-cli` timing, with the CLI built by `cargo build --profile ci -p
datafusion-cli` on each side:
```bash
#!/usr/bin/env bash
# Usage: wide-timing.sh <datafusion-cli>
set -euo pipefail
python3 - <<'EOF'
for n in (500, 1000, 2000, 4000):
items = ', '.join(f'CAST(sum(id + {i}) AS VARCHAR) AS c{i}' for i in
range(n))
with open(f'wide_{n}.sql', 'w') as f:
f.write('CREATE TABLE t AS SELECT value AS id FROM range(1000);\n')
f.write(f'EXPLAIN SELECT {items} FROM t;\n')
EOF
for n in 500 1000 2000 4000; do
/usr/bin/time -p "$1" -q -f "wide_$n.sql" 2>&1 >/dev/null | awk -v n="$n"
'/^real/ {print "N=" n, $2 "s"}'
done
```
```
== main
N=500 0.72s
N=1000 1.56s
N=2000 5.05s
N=4000 18.83s
== this PR
N=500 0.66s
N=1000 0.87s
N=2000 1.83s
N=4000 4.57s
```
</details>
## Are there any user-facing changes?
No. Planning is faster, and `columnize_expr` keeps its signature and
behavior.
--
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]
