You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@datafusion.apache.org by "adriangb (via GitHub)" <gi...@apache.org> on 2026/09/03 18:49:40 UTC

[I] SingleDistinctToGroupBy fires where it costs far more memory: adding count(*) makes a query ~1000x cheaper [datafusion]

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: github-unsubscribe@datafusion.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org