The DISTINCT in Your COUNT

# The DISTINCT in your COUNT Source: [https://boringsql.com/posts/distinct-in-your-count/](https://boringsql.com/posts/distinct-in-your-count/) Here is a query that shows up in every analytics workload: ``` SELECT count(DISTINCT user_id) FROM events; ``` It looks like the cheapest possible thing: count the distinct users\. On a machine with cores to spare you would expect Postgres to throw a few parallel workers at it, the way it does for almost any large scan\. It does not\. That one keyword,`DISTINCT`, switches off parallel query for the entire statement, and the larger your table the more it costs you\. No setting or index changes that; the reason is in how the aggregate has to execute\. ## The schema[https://boringsql.com/posts/distinct-in-your-count/#the-schema](https://boringsql.com/posts/distinct-in-your-count/#the-schema) Ten million events, about fifty thousand distinct users, a handful of countries\. Nothing unusual\. ``` CREATE TABLE events ( id bigint GENERATED ALWAYS AS IDENTITY, user_id int NOT NULL, country text NOT NULL, amount numeric(10,2) NOT NULL ); INSERT INTO events (user_id, country, amount) SELECT (random()*50000)::int + 1, (ARRAY['US','DE','GB','FR','JP','BR','IN','CA'])[(random()*7)::int + 1], (random()*500)::numeric(10,2) FROM generate_series(1, 10000000); ANALYZE events; ``` `max\_parallel\_workers\_per\_gather`is at its default of 2 on fresh cluster\. For these examples I raised it to 4 and`work\_mem`to 64MB, so there's no resource starvation to blame for the plans below\. ## Two counts, two different plans[https://boringsql.com/posts/distinct-in-your-count/#two-counts-two-different-plans](https://boringsql.com/posts/distinct-in-your-count/#two-counts-two-different-plans) Start with a plain`count\(\*\)`, which has nothing to deduplicate: ``` EXPLAIN (ANALYZE, COSTS OFF) SELECT count(*) FROM events; ``` ``` Finalize Aggregate (actual rows=1.00 loops=1) -> Gather (actual rows=5.00 loops=1) Workers Planned: 4 Workers Launched: 4 -> Partial Aggregate (actual rows=1.00 loops=5) -> Parallel Seq Scan on events (actual rows=2000000.00 loops=5) ``` Four workers plus the leader \(`loops=5`\) each scan their slice and keep a running count, and the leader adds the five partial counts together at the end\. Now add one word: ``` EXPLAIN (ANALYZE, COSTS OFF, BUFFERS) SELECT count(DISTINCT user_id) FROM events; ``` ``` Aggregate (actual rows=1.00 loops=1) Buffers: shared hit=15915 read=47783, temp read=14681 written=14684 -> Sort (actual rows=10000000.00 loops=1) Sort Key: user_id Sort Method: external merge Disk: 117448kB Buffers: shared hit=15915 read=47783, temp read=14681 written=14684 -> Seq Scan on events (actual rows=10000000.00 loops=1) Buffers: shared hit=15912 read=47783 ``` No`Gather`\. No`Partial Aggregate`\. No parallel scan\. A single process reads all ten million rows, sorts every one of them by`user\_id`so duplicates sit next to each other, then walks the sorted output counting the runs\. The sort does not fit in 64MB of`work\_mem`, so it spills 115MB to a temporary file on disk\. One core, the whole table, plus disk IO that the parallel`count\(\*\)`never touched\. ## Why the planner can't split it[https://boringsql.com/posts/distinct-in-your-count/#why-the-planner-can-t-split-it](https://boringsql.com/posts/distinct-in-your-count/#why-the-planner-can-t-split-it) The sort is how Postgres computes`DISTINCT`inside an aggregate: order the values and adjacent equal ones collapse\. A hash table is the other option, but the classic`DISTINCT`\-aggregate path sorts\. Either way it has to see every value in one place, which is the whole problem\. Parallel aggregation in Postgres works in two halves\. Each worker runs a**Partial Aggregate**that builds*transition state*, a small running summary of the rows it has seen\. For`count`that state is just a number\. The leader then runs a**Finalize Aggregate**that merges those partial states with the aggregate's*combine function*, the thing that knows how to fold two partial states into one\.`count`'s combine function adds the partial counts\.`sum`,`avg`,`min`,`max`all have one\. This split, scan in parallel, combine at the end, is the entire basis of parallel query for aggregates\. `count\(DISTINCT user\_id\)`has no usable combine step, and not because nobody wrote one\. Think about what a worker could hand back\. To merge two workers' results into a correct global distinct count, the leader would need to know*which*users each worker saw, because a user that appears in worker 1's slice and again in worker 2's slice must be counted once, not twice\. A partial count of distinct values cannot be combined; you would have to ship the entire set of distinct values from every worker and union them\. At that point you have moved all the data to one place anyway, which is exactly what parallel aggregation exists to avoid\. An aggregate carrying`DISTINCT`\(or an inner`ORDER BY`\) therefore cannot run in partial mode, the planner cannot place a Partial Aggregate under a Gather, and with no partial aggregate to feed, a parallel scan buys nothing\. The whole plan collapses to serial\. I checked this against PostgreSQL 17\.10, 18\.4, and 19beta1: partial aggregation still does not cover distinct and ordered aggregates on any of them\. `debug\_parallel\_query`is a way to check this isn't a cost estimate that happened to favor serial execution\. Set to`on`, it makes the planner reach for a parallel plan wherever one is legal, even when the optimizer thinks serial is cheaper: ``` SET debug_parallel_query = on; EXPLAIN (COSTS OFF) SELECT count(DISTINCT user_id) FROM events; ``` ``` Gather Workers Planned: 1 Single Copy: true -> Aggregate -> Sort Sort Key: user_id -> Seq Scan on events ``` A`Gather`shows up, but with`Workers Planned: 1`and`Single Copy: true`: one process runs the entire plan, sort included, and the Gather node only exists to route its output back through the executor's parallel machinery\. Nothing about the aggregate, the sort, or the scan actually splits across workers\. That's`debug\_parallel\_query`forcing parallel infrastructure onto a plan that has no partial aggregate to divide the work with, and finding nothing there for a second worker to do\. `FILTER`clause does not have this problem\. ``` EXPLAIN (COSTS OFF) SELECT count(*) FILTER (WHERE country='US') FROM events; ``` ``` Finalize Aggregate -> Gather Workers Planned: 4 -> Partial Aggregate -> Parallel Seq Scan on events ``` Same parallel shape as plain`count\(\*\)`\.`FILTER`just decides which rows each worker folds into its partial count\. ## One DISTINCT poisons the whole statement[https://boringsql.com/posts/distinct-in-your-count/#one-distinct-poisons-the-whole-statement](https://boringsql.com/posts/distinct-in-your-count/#one-distinct-poisons-the-whole-statement) The cost is not scoped to the distinct aggregate\. It is scoped to the aggregation node it shares a query block with\. Put a perfectly parallelizable aggregate next to a distinct one in the same`SELECT`and*both*lose parallelism, thanks to the fact that one Aggregate node computes both and it can only run one way\. An aggregate in a separate subquery or CTE is a different node and isn't affected: ``` EXPLAIN (COSTS OFF) SELECT sum(amount), count(DISTINCT user_id) FROM events; ``` ``` Aggregate -> Sort Sort Key: user_id -> Seq Scan on events ``` `sum\(amount\)`on its own would have run across four workers\. Sharing a`SELECT`with one`count\(DISTINCT\)`drags it down to the same serial sort\. ## The rewrite: push the DISTINCT into a GROUP BY[https://boringsql.com/posts/distinct-in-your-count/#the-rewrite-push-the-distinct-into-a-group-by](https://boringsql.com/posts/distinct-in-your-count/#the-rewrite-push-the-distinct-into-a-group-by) Do the deduplication with the one operation Postgres*can*parallelize, a`GROUP BY`, and count the groups afterward: ``` SELECT count(*) FROM (SELECT user_id FROM events GROUP BY user_id) s; ``` `GROUP BY user\_id`is exactly "the distinct user\_ids", and grouping has partial mode: each worker builds a partial hash of the groups it saw, and the leader merges those hashes\. Counting how many groups came out is then trivial\. ``` EXPLAIN (ANALYZE, COSTS OFF) SELECT count(*) FROM (SELECT user_id FROM events GROUP BY user_id) s; ``` ``` Aggregate (actual rows=1.00 loops=1) -> Finalize HashAggregate (actual rows=50001.00 loops=1) Group Key: events.user_id Batches: 1 Memory Usage: 3097kB -> Gather (actual rows=250005.00 loops=1) Workers Planned: 4 Workers Launched: 4 -> Partial HashAggregate (actual rows=50001.00 loops=5) Group Key: events.user_id Batches: 1 Memory Usage: 3097kB Worker 0: Batches: 1 Memory Usage: 3097kB Worker 1: Batches: 1 Memory Usage: 3097kB Worker 2: Batches: 1 Memory Usage: 3097kB Worker 3: Batches: 1 Memory Usage: 3097kB -> Parallel Seq Scan on events (actual rows=2000000.00 loops=5) ``` The rewrite is parallel again: four workers each hash their slice down to the local set of users, and the leader merges those into the final 50,001 groups in memory, with no sort or disk spill\. The wall\-clock difference on this 10M\-row table, identical hardware and settings, median of three runs: QueryPlanTime`count\(DISTINCT user\_id\)`serial sort, 115MB to disk1211 ms`count\(\*\) FROM \(… GROUP BY user\_id\)`parallel hash, in memory360 msBoth return`50001`\. The rewrite is about 3\.4x faster here, and the gap should widen with the table: the serial sort's cost grows with row count, while the parallel hash keeps adding throughput with each worker\. If an approximate answer is acceptable, this is the problem HyperLogLog is meant to solve: its sketch is a small state that, in principle, has the combine function exact distinct counts lack, so partial sketches from workers should merge\. The`postgresql\-hll`extension builds on that idea, and it is often suggested for dashboard\-style distinct counts at the price of a bounded error rate\. ## ORDER BY aggregates hit the same wall[https://boringsql.com/posts/distinct-in-your-count/#order-by-aggregates-hit-the-same-wall](https://boringsql.com/posts/distinct-in-your-count/#order-by-aggregates-hit-the-same-wall) The block is not specific to`DISTINCT`\. Any aggregate that needs its input in a particular order, the ordered\-set and ordered aggregates, fails to parallelize for the same reason, because a worker's locally\-ordered partial result cannot be merged without re\-ordering across workers: ``` EXPLAIN (COSTS OFF) SELECT string_agg(country, ',' ORDER BY country) FROM events; ``` ``` Aggregate -> Sort Sort Key: country -> Seq Scan on events ``` `string\_agg`,`array\_agg`,`json\_agg`with an inner`ORDER BY`, and`percentile\_cont`/`percentile\_disc`all land here\. If you have an aggregate that insists on global order or global distinctness, assume it runs on one core until`EXPLAIN`tells you otherwise\. ## The harder case: per\-group distinct counts[https://boringsql.com/posts/distinct-in-your-count/#the-harder-case-per-group-distinct-counts](https://boringsql.com/posts/distinct-in-your-count/#the-harder-case-per-group-distinct-counts) The clean rewrite above is for a single distinct count over the whole table\. The per\-group version of the same query, ``` SELECT country, count(DISTINCT user_id) FROM events GROUP BY country; ``` is also serial \(a`GroupAggregate`over a sort on`country, user\_id`\)\. The same idea applies, deduplicate first with a grouping the workers can split, then aggregate: ``` SELECT country, count(*) FROM (SELECT country, user_id FROM events GROUP BY country, user_id) s GROUP BY country; ``` This*makes*the work parallelizable, but whether the planner actually picks the parallel path depends on cardinalities and cost\. In my testing the overall\-count rewrite parallelized reliably, while this stacked\-grouping form sometimes stayed serial because the planner judged the two hash\-aggregate layers cheap enough already\. The rule is the same, push the distinctness into a`GROUP BY`the engine can divide, but better check`EXPLAIN`rather than assuming it took the parallel path only because you gave it the option\. ## When to actually care[https://boringsql.com/posts/distinct-in-your-count/#when-to-actually-care](https://boringsql.com/posts/distinct-in-your-count/#when-to-actually-care) None of this matters on a small table\. If the scan is a few thousand rows, serial is instant and the rewrite only adds noise\. The distinct\-aggregate penalty is a function of how many rows the single core has to sort, so it shows up exactly where it hurts: large fact tables, dashboards over months of events, the nightly rollup that runs on one core while the rest sit idle\. Those are the queries to inspect\. The tell in`EXPLAIN \(ANALYZE\)`is unmistakable once you know it: a top\-level`Aggregate`with no`Gather`beneath it, a big`Sort`with an`external merge \.\.\. Disk:`line, and a single`loops=1`scan of the whole table\. If you see that shape above a distinct or ordered aggregate on a table that matters, it's worth the rewrite\.