The 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

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

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

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.

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

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

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:

Both 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.

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

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

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.