Repository navigation
Grouped count(DISTINCT x) on numbers takes a slower path #25936
Description
Describe the bug
SELECT g, count(DISTINCT x) FROM t GROUP BY g can run in two ways:
- Rewrite. The
SingleDistinctToGroupByrule first builds a list of every different(g, x)pair, then counts per group. - Count directly. One pass over the rows. For numbers,there is a fast way to do this.
For numbers, counting directly is about 20% faster and needs the same memory. But currently DataFusion always uses the rewrite.
To Reproduce
Use datafusion-cli.
SET datafusion.execution.target_partitions = 1;
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);
SELECT g, count(DISTINCT x) FROM t GROUP BY g;datafusion-cli -m <limit> -d 0 --mem-pool-type greedy -f <file>
The table has 4 million rows, 2,000 groups, and 4 million different (g, x) pairs.
| rewrite | memory limit needed | time |
|---|---|---|
| yes | 192M | 0.152 s |
| no | 192M | 0.124 s |
The rule must be programmatically switched off.
EXPLAIN with the rule on. The inner aggregate builds one row per (g, x) pair (4 million), and the outer one counts them per g:
> EXPLAIN FORMAT indent SELECT g, count(DISTINCT x) FROM t GROUP BY g;
logical_plan
Projection: t.g, count(alias1) AS count(DISTINCT t.x)
Aggregate: groupBy=[[t.g]], aggr=[[count(alias1)]]
Aggregate: groupBy=[[t.g, t.x AS alias1]], aggr=[[]]
TableScan: t projection=[g, x]
physical_plan
ProjectionExec: expr=[g@0 as g, count(alias1)@1 as count(DISTINCT t.x)]
AggregateExec: mode=Single, gby=[g@0 as g], aggr=[count(alias1)]
AggregateExec: mode=Single, gby=[g@0 as g, x@1 as alias1], aggr=[]
DataSourceExec: partitions=1, partition_sizes=[489]With the rule off there is one aggregate, which counts directly:
> EXPLAIN FORMAT indent SELECT g, count(DISTINCT x) FROM t GROUP BY g;
logical_plan
Aggregate: groupBy=[[t.g]], aggr=[[count(DISTINCT t.x)]]
TableScan: t projection=[g, x]
physical_plan
AggregateExec: mode=Single, gby=[g@0 as g], aggr=[count(DISTINCT t.x)]
DataSourceExec: partitions=1, partition_sizes=[489]For text (VARCHAR), the rewrite is more than twice as fast and needs less memory (224M vs 384M), so the rewrite is the right choice there.
Expected behavior
For number columns, count(DISTINCT x) should count directly instead of using the rewrite.
Activity
QinXi-ai commentedon Oct 2, 2026
take
perf: add opt-in native grouped integer DISTINCT counts (apache#25988)
QinXi-ai commentedon Nov 8, 2026
Following up on @alamb's comment on #25988: I agree this issue should be reopened. The default still takes the original path. I should have removed "Closes #25936" when I changed the PR to opt-in.
I've traced the spill path and have a plan for getting this enabled by default:
COUNT(DISTINCT)that writes and merges(group keys, distinct value)pairs. The current path spills lists, sorts by group, then rebuilds each group's distinct set during replay. Sorting by the value too would let replay deduplicate and count incrementally, including when one group's values exceed memory. The existing spill-file and merge code should be reusable; draining values in bounded chunks and accounting for the conversion buffers need new work.falseretaining the rewrite as a fallback.So far I've checked the code and run a standalone Python model through 315 differential cases, including duplicates across spill runs, null-only groups and a group larger than the model's buffer. The Rust implementation and performance measurements are still ahead. I'll start with the accumulator change; the spill work is the larger part of making the default safe under a memory limit.
QinXi-ai commentedon Nov 10, 2026
I've opened #27299 for the first step of the plan above, caching each group's first non-null value to improve the duplicate-heavy case. It's a draft, and I'd welcome early feedback on the approach.