adriangb opened a new issue, #24929:
URL: https://github.com/apache/datafusion/issues/24929
### Describe the bug
`SingleDistinctToGroupBy` rewrites `AGG(DISTINCT x)` into an inner `GROUP BY
(group_keys, x)` plus an outer aggregate. The inner aggregate then holds one
row per distinct `(group_keys, x)` pair instead of one row per group, and every
other aggregate in the query moves down to that finer grain.
The rule decides whether to fire from **which aggregate functions appear**
in the query, not from whether the rewrite is actually cheaper.
`is_single_distinct_agg` accepts the plan when every non-distinct aggregate is
`sum`, `min` or `max`, and rejects it otherwise. That condition has nothing to
do with the cost of the rewrite, so the rule fires on plans where it is a pure
loss.
The clearest symptom is that **adding an aggregate to the select list makes
the query about 1000x cheaper in memory**, even though the extra aggregate
makes the query do strictly more work. `count` is not in the tolerated set, so
a `count(*)` disqualifies the plan from the rewrite and the query gets much
cheaper.
### To Reproduce
Using `datafusion-cli` on `main`:
```sql
CREATE TABLE t AS
SELECT v % 2000 AS g, (v * 48271) % 999983 AS x, v % 1000 AS y
FROM (SELECT unnest(generate_series(0, 3999999)) AS v);
-- 4,000,000 rows, 2,000 groups, ~4,000,000 distinct (g, x) pairs
```
The two plans differ only by the `count(*)`:
```
> EXPLAIN SELECT g, min(DISTINCT x) FROM t GROUP BY g;
Projection: t.g, min(alias1) AS min(DISTINCT t.x)
Aggregate: groupBy=[[t.g]], aggr=[[min(alias1)]]
Aggregate: groupBy=[[t.g, t.x AS alias1]], aggr=[[]]
TableScan: t projection=[g, x]
> EXPLAIN SELECT g, count(*), min(DISTINCT x) FROM t GROUP BY g;
Projection: t.g, count(Int64(1)) AS count(*), min(DISTINCT t.x)
Aggregate: groupBy=[[t.g]], aggr=[[count(Int64(1)), min(DISTINCT t.x)]]
TableScan: t projection=[g, x]
```
The rewritten plan builds a 4,000,000 row hash table. The plan that also
computes `count(*)` keeps 2,000 groups.
Under a bounded memory pool the smaller query is the one that fails. Run
each query with `datafusion-cli -m <limit> --mem-pool-type greedy`:
| query | rewritten | smallest limit that succeeds |
| --- | --- | --- |
| `SELECT g, min(DISTINCT x) FROM t GROUP BY g` | yes | 48M (fails at 40M) |
| `SELECT g, count(*), min(DISTINCT x) FROM t GROUP BY g` | no | 512K (fails
at 256K) |
At `-m 32M` the first query fails and the second, which computes strictly
more, succeeds:
```
Error: Resources exhausted: Additional allocation failed for
FinalHashAggregateStream[5] ...
greedy(used: 31.8 MB, pool_size: 32.0 MB)
```
Two more pairs behave the same way:
| query | rewritten | smallest limit that succeeds |
| --- | --- | --- |
| `SELECT g, sum(y), sum(DISTINCT x) FROM t GROUP BY g` | yes | 64M |
| `SELECT g, count(*), sum(y), sum(DISTINCT x) FROM t GROUP BY g` | no | 24M
|
### Expected behavior
Adding an aggregate to a query should not reduce its memory requirement by
three orders of magnitude. The rewrite should not fire on plans where it is
strictly more expensive than the plan it replaces.
### Additional context
#### Measured peak memory
To attribute the cost to the rule rather than to the memory limit, I ran
each query twice on identical data, once with the rule in the optimizer
pipeline and once with it removed, recording the high water mark of
`MemoryPool::reserved()` under an unbounded pool with `target_partitions = 1`.
Results are identical in both cases. 4,000,000 rows, `x` drawn from 1,000,000
values, `sum(y)` as the non-distinct companion.
| distinct aggregate | groups | type | rule off | rule on | |
| --- | --- | --- | --- | --- | --- |
| `count(DISTINCT x)` | 500,000 | BIGINT | 195.9 MiB | 223.4 MiB | 1.14x
worse |
| `count(DISTINCT x)` | 500,000 | VARCHAR | 4075.2 MiB | 271.4 MiB | 15x
better |
| `count(DISTINCT x)` | 2,000 | BIGINT | 168.2 MiB | 223.4 MiB | 1.33x worse
|
| `sum(DISTINCT x)` | 2,000 | BIGINT | 71.0 MiB | 223.4 MiB | 3.15x worse |
| `avg(DISTINCT x)` | 2,000 | BIGINT | 71.0 MiB | 223.4 MiB | 3.15x worse |
| `min(DISTINCT x)` | 2,000 | BIGINT | 0.2 MiB | 223.4 MiB | 1113x worse |
| `array_agg(DISTINCT x)` | 2,000 | BIGINT | 392.6 MiB | 223.4 MiB | 1.8x
better |
Process peak RSS for the `min(DISTINCT x)` pair is 1062 MiB against 784 MiB,
so this is real memory rather than an accounting artifact.
Using `sum`, `min` or `max` as the non-distinct companion gives
byte-identical numbers. The companion aggregate does not affect the mechanism,
only whether the rule fires.
#### Why the outcome varies
What decides the result is the storage cost per distinct value on each side.
The rewrite materializes one hash table row per distinct `(group_keys, x)`
pair, plus an accumulator slot per other aggregate at that grain. It wins when
the unrewritten accumulator costs more than that per value, such as a per-group
hash set of strings or an `array_agg`. It loses when the unrewritten
accumulator costs less.
`min` and `max` are the extreme case, because `min(DISTINCT x)` equals
`min(x)`. `min_max.rs` correctly never inspects `is_distinct`, so the
unrewritten plan costs one scalar per group while the rewrite builds a hash
table with one row per distinct pair.
#### Note on removing the rule
The numbers above also show that removing the rule outright would be a
regression. It is 15x better for `count(DISTINCT <string>)` at high group
cardinality and about 2x better for `array_agg(DISTINCT)`. A gate that reflects
when the rewrite actually helps looks more promising than either the current
function-name condition or a blanket removal.
#### Related work
- #11360 asks whether the rule is still needed now that distinct
accumulators exist, with a ClickBench comparison that found no clear advantage
from deleting it. That thread measures planning and runtime and asks whether
the rule is *unnecessary*. This report is about the rule being *actively
harmful* on a class of plans, on a trigger condition unrelated to its benefit,
and it also shows the rule is genuinely valuable on other plans.
- #8266, closing #8123, added the tolerance for non-distinct `sum`, `min`
and `max`. That tolerance is what lets the rule fire on the `min(DISTINCT)` and
`sum(DISTINCT)` cases above.
- #20782 proposed skipping the rewrite for `count(DISTINCT)` with no `GROUP
BY`, on the same grounds that the direct distinct accumulator is cheaper. It
was not merged.
- #11686 proposes eliminating `DISTINCT` on `min` and `max` early, which
would remove the worst case here at its source.
- #20942 and #21087 cover the multiple-distinct case, which this rule does
not handle at all.
- #24704 tracks blocked and chunked memory management in hash aggregation.
#### Version
Reproduced on `main` at `d2b626cc9` (`datafusion-cli 55.0.0`), macOS arm64.
`single_distinct_to_groupby.rs`, `min_max.rs` and `count.rs` are unchanged at
`20d1c5676`, the current `main` head at the time of writing.
--
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]
