The DISTINCT in Your COUNT

Here is simply a query that shows up successful each analytics workload:

SELECT count(DISTINCT user_id) FROM events;

It looks for illustration the cheapest imaginable thing: count the chopped users. On a instrumentality pinch cores to spare you would expect Postgres to propulsion a fewer parallel workers astatine it, the measurement it does for almost immoderate ample scan. It does not. That 1 keyword, DISTINCT, switches disconnected parallel query for the full statement, and the larger your array the much it costs you. No mounting aliases scale changes that; the logic is successful really the aggregate has to execute.

The schema

Ten cardinal events, astir 50 1000 chopped users, a fistful of countries. Nothing unusual.

CREATE TABLE events ( id bigint GENERATED ALWAYS AS IDENTITY, user_id int NOT NULL, state text NOT NULL, magnitude 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 astatine its default of 2 connected caller cluster. For these examples I raised it to 4 and work_mem to 64MB, truthful there's nary assets starvation to blasted for the plans below.

Two counts, 2 different plans

Start pinch a plain count(*), which has thing 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 connected events (actual rows=2000000.00 loops=5)

Four workers positive the leader (loops=5) each scan their portion and support a moving count, and the leader adds the 5 partial counts together astatine the end.

Now adhd 1 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: outer merge Disk: 117448kB Buffers: shared hit=15915 read=47783, temp read=14681 written=14684 -> Seq Scan connected events (actual rows=10000000.00 loops=1) Buffers: shared hit=15912 read=47783

No Gather. No Partial Aggregate. No parallel scan. A azygous process sounds each 10 cardinal rows, sorts each 1 of them by user_id truthful duplicates beryllium adjacent to each other, past walks the sorted output counting the runs. The benignant does not fresh successful 64MB of work_mem, truthful it spills 115MB to a impermanent record connected disk. One core, the full table, positive disk IO that the parallel count(*) ne'er touched.

Why the planner can't divided it

The benignant is really Postgres computes DISTINCT wrong an aggregate: bid the values and adjacent adjacent ones collapse. A hash array is the different option, but the classical DISTINCT-aggregate way sorts. Either measurement it has to spot each worth successful 1 place, which is the full problem.

Parallel aggregation successful Postgres useful successful 2 halves. Each worker runs a Partial Aggregate that builds transition state, a mini moving summary of the rows it has seen. For count that authorities is conscionable a number. The leader past runs a Finalize Aggregate that merges those partial states pinch the aggregate's combine function, the point that knows really to fold 2 partial states into one. count's harvester usability adds the partial counts. sum, avg, min, max each person one. This split, scan successful parallel, harvester astatine the end, is the full ground of parallel query for aggregates.

count(DISTINCT user_id) has nary usable harvester step, and not because cipher wrote one. Think astir what a worker could manus back. To merge 2 workers' results into a correct world chopped count, the leader would request to cognize which users each worker saw, because a personification that appears successful worker 1's portion and again successful worker 2's portion must beryllium counted once, not twice. A partial count of chopped values cannot beryllium combined; you would person to vessel the full group of chopped values from each worker and national them. At that constituent you person moved each the information to 1 spot anyway, which is precisely what parallel aggregation exists to avoid.

An aggregate carrying DISTINCT (or an soul ORDER BY) truthful cannot tally successful partial mode, the planner cannot spot a Partial Aggregate nether a Gather, and pinch nary partial aggregate to feed, a parallel scan buys nothing. The full scheme collapses to serial.

I checked this against PostgreSQL 17.10, 18.4, and 19beta1: partial aggregation still does not screen chopped and ordered aggregates connected immoderate of them.

debug_parallel_query is simply a measurement to cheque this isn't a costs estimate that happened to favour serial execution. Set to on, it makes the planner scope for a parallel scheme wherever 1 is legal, moreover erstwhile 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 connected events

A Gather shows up, but pinch Workers Planned: 1 and Single Copy: true: 1 process runs the full plan, benignant included, and the Gather node only exists to way its output backmost done the executor's parallel machinery. Nothing astir the aggregate, the sort, aliases the scan really splits crossed workers. That's debug_parallel_query forcing parallel infrastructure onto a scheme that has nary partial aggregate to disagreement the activity with, and uncovering thing location for a 2nd worker to do.

FILTER clause does not person this problem.

EXPLAIN (COSTS OFF) SELECT count(*) FILTER (WHERE country='US') FROM events; Finalize Aggregate -> Gather Workers Planned: 4 -> Partial Aggregate -> Parallel Seq Scan connected events

Same parallel style arsenic plain count(*). FILTER conscionable decides which rows each worker folds into its partial count.

One DISTINCT poisons the full statement

The costs is not scoped to the chopped aggregate. It is scoped to the aggregation node it shares a query artifact with. Put a perfectly parallelizable aggregate adjacent to a chopped 1 successful the aforesaid SELECT and both suffer parallelism, acknowledgment to the truth that 1 Aggregate node computes some and it tin only tally 1 way. An aggregate successful a abstracted subquery aliases CTE is simply 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 connected events

sum(amount) connected its ain would person tally crossed 4 workers. Sharing a SELECT pinch 1 count(DISTINCT) drags it down to the aforesaid serial sort.

The rewrite: push the DISTINCT into a GROUP BY

Do the deduplication pinch the 1 cognition 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 precisely "the chopped 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 really galore groups came retired is past 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 connected events (actual rows=2000000.00 loops=5)

The rewrite is parallel again: 4 workers each hash their portion down to the section group of users, and the leader merges those into the last 50,001 groups successful memory, pinch nary benignant aliases disk spill.

The wall-clock quality connected this 10M-row table, identical hardware and settings, median of 3 runs:

QueryPlanTime

count(DISTINCT user_id)	serial sort, 115MB to disk	1211 ms
count(*) FROM (… GROUP BY user_id)	parallel hash, successful memory	360 ms

Both return 50001. The rewrite is astir 3.4x faster here, and the spread should widen pinch the table: the serial sort's costs grows pinch statement count, while the parallel hash keeps adding throughput pinch each worker.

If an approximate reply is acceptable, this is the problem HyperLogLog is meant to solve: its sketch is simply a mini authorities that, successful principle, has the harvester usability nonstop chopped counts lack, truthful partial sketches from workers should merge. The postgresql-hll hold builds connected that idea, and it is often suggested for dashboard-style chopped counts astatine the value of a bounded correction rate.

ORDER BY aggregates deed the aforesaid wall

The artifact is not circumstantial to DISTINCT. Any aggregate that needs its input successful a peculiar order, the ordered-set and ordered aggregates, fails to parallelize for the aforesaid reason, because a worker's locally-ordered partial consequence cannot beryllium merged without re-ordering crossed workers:

EXPLAIN (COSTS OFF) SELECT string_agg(country, ',' ORDER BY country) FROM events; Aggregate -> Sort Sort Key: country -> Seq Scan connected events

string_agg, array_agg, json_agg pinch an soul ORDER BY, and percentile_cont/percentile_disc each onshore here. If you person an aggregate that insists connected world bid aliases world distinctness, presume it runs connected 1 halfway until EXPLAIN tells you otherwise.

The harder case: per-group chopped counts

The cleanable rewrite supra is for a azygous chopped count complete the full table. The per-group type of the aforesaid query,

SELECT country, count(DISTINCT user_id) FROM events GROUP BY country;

is besides serial (a GroupAggregate complete a benignant connected country, user_id). The aforesaid thought applies, deduplicate first pinch a grouping the workers tin split, past aggregate:

SELECT country, count(*) FROM (SELECT country, user_id FROM events GROUP BY country, user_id) s GROUP BY country;

This makes the activity parallelizable, but whether the planner really picks the parallel way depends connected cardinalities and cost. In my testing the overall-count rewrite parallelized reliably, while this stacked-grouping shape sometimes stayed serial because the planner judged the 2 hash-aggregate layers inexpensive capable already. The norm is the same, push the distinctness into a GROUP BY the motor tin divide, but amended cheque EXPLAIN alternatively than assuming it took the parallel way only because you gave it the option.

When to really care

None of this matters connected a mini table. If the scan is simply a fewer 1000 rows, serial is instant and the rewrite only adds noise. The distinct-aggregate punishment is simply a usability of really galore rows the azygous halfway has to sort, truthful it shows up precisely wherever it hurts: ample truth tables, dashboards complete months of events, the nightly rollup that runs connected 1 halfway while the remainder beryllium idle. Those are the queries to inspect.

The show successful EXPLAIN (ANALYZE) is unmistakable erstwhile you cognize it: a top-level Aggregate pinch nary Gather beneath it, a large Sort pinch an outer merge ... Disk: line, and a azygous loops=1 scan of the full table. If you spot that style supra a chopped aliases ordered aggregate connected a array that matters, it's worthy the rewrite.