Postgres Sharding Doesn't Fix a Bad Cardinality Estimate

Sharding proposals tend to arrive with a latency graph and a connection-count chart. They rarely arrive with an EXPLAIN ANALYZE. The pitch has a familiar shape: p99 is climbing, CPU is pegged during peak hours, and the fix on offer is more nodes. Nobody has pulled up the plan for the query that's actually driving the saturation.

That's a costly omission, because a single misestimated query produces the same external symptoms as a workload that has outgrown one machine: high CPU, growing queue depth, timeouts under load. The two look identical on a dashboard and call for opposite responses. Sharding fixes one and multiplies the other.

The reasoning that gets teams there skips a step. The database is slow, the database is one machine, so make it more machines. Before any topology decision you need to know which query, which plan node, and which row estimate produced the slowness. A saturated CPU can mean the workload has outgrown the node. It can also mean the planner chose a nested loop against a scan it believed would return 1,700 rows, when the scan returned 16 million.

Sharding redistributes rows across nodes. It doesn't touch the estimate that chose the plan. That estimate comes from the statistics Postgres keeps on the columns involved, not from how many machines those columns are spread across. If the estimate was wrong on one node, the same query against the same schema gets the same wrong estimate on every node you add, each with a smaller slice of data to be wrong about.

1,700 estimated, 16 million actual

The planner turns a candidate plan plus a set of row-count estimates into a cost, then picks the cheapest plan it found. Every join order, join method, and index choice downstream of an estimate inherits whatever error is in it. A small error picks a slightly worse plan. A large error picks a plan whose entire justification was false.

Here is the shape of the failure. The plan below is constructed rather than copied from a real cluster (the numbers are chosen to be internally consistent), but the shape is the one you'll find when you go looking.

Nested Loop  (cost=0.99..8931.44 rows=1700 width=72)
             (actual time=0.071..57340.512 rows=16000000 loops=1)
  ->  Index Scan using events_region_type_idx on events e
        (cost=0.56..562.81 rows=1700 width=52)
        (actual time=0.044..5821.309 rows=16000000 loops=1)
        Index Cond: ((region = 'me-south'::text) AND (event_type = 'checkout'::text))
  ->  Index Scan using accounts_pkey on accounts a
        (cost=0.43..4.92 rows=1 width=20)
        (actual time=0.003..0.003 rows=1 loops=16000000)
        Index Cond: (id = e.account_id)
Planning Time: 0.388 ms
Execution Time: 58902.774 ms

Look at the two rows values on the outer scan. rows=1700 is what the planner expected the index scan on events to return. rows=16000000 next to actual time is what came back. On the strength of 1,700, a nested loop is the obvious choice: for each event, probe accounts by primary key, about 1,700 cheap lookups. What ran was 16 million lookups, one per outer row (loops=16000000 on the inner node), at three microseconds each. That's about 48 seconds of the 59, spent on work the planner priced at a few milliseconds. With an accurate estimate the planner would have built a hash on accounts and streamed the events through it. The index scan alone took under six seconds; the hash probe would have added a few more.

Nothing here is a capacity problem. The machine did exactly what it was asked to do, efficiently. It was asked to do the wrong thing.

Where the 1,700 came from

The misestimate on that scan is two ordinary mechanisms stacked.

The first is stale statistics. Say events has 640 million rows, and a new region, me-south, went live three weeks ago with a 16.2-million-row backfill from an acquired product. Autovacuum re-analyzes a table once the rows modified since the last analyze exceed autovacuum_analyze_threshold + autovacuum_analyze_scale_factor * reltuples. At the defaults that's 50 + 0.1 × 640M, roughly 64 million rows. A 16-million-row backfill doesn't get close, so the statistics still describe a table with six regions. When the planner meets region = 'me-south', a value that isn't in the column's most-common-values list, it assumes the value is one of the rare ones the sample didn't capture and estimates it from whatever frequency the MCV list left over. On this table that works out to about 68,000 rows, 0.01% of the table.

The second is the independence assumption. event_type = 'checkout' is 2.5% of the table overall, so the planner multiplies: 68,000 × 0.025 ≈ 1,700. But the backfill was almost entirely commerce events. 98% of me-south rows are checkouts, so the true count is 16.2M × 0.98 ≈ 16 million. Postgres assumes predicates on different columns are independent unless you tell it otherwise, and here they're nearly the same predicate.

A third mechanism is worth naming even though it isn't in this plan. A predicate wrapped in a function, lower(email) = $1 or date_trunc('day', created_at) = $1, can't use the column's histogram at all. The planner falls back to a default selectivity that has nothing to do with the data.

Two things follow. All of this lives in pg_statistic, and none of it is about volume: a 40x miss and a 9,000x miss are the same defect at different magnitudes. And because the estimate is a product of factors, error compounds. Each plan node that inherits a bad number multiplies its own error on top.

The fixes, cheapest first

The fix is a ladder. Each rung targets a specific blind spot in the estimator, and you stop climbing when the estimates converge.

Before any of it, find the query. pg_stat_statements ordered by total_exec_time names it, and that report belongs on every scaling proposal. Then run EXPLAIN (ANALYZE, BUFFERS) against production data skew and compare estimated to actual rows on every node. Plain EXPLAIN shows what the planner intends; EXPLAIN ANALYZE runs the query and reports what happened next to it. The gap between those two columns is the diagnostic.

Stale statistics: ANALYZE. Check last_autoanalyze in pg_stat_user_tables. If it predates the change that broke the estimate, ANALYZE events fixes the first layer on the spot: the planner now knows me-south is 2.5% of the table, not 0.01%. For the plan above, that alone moves the estimate to about 400,000 rows. Still 40x low, but high enough that the planner drops the nested loop for a hash join, and the catastrophe is over. Large append-mostly tables should run with a lower autovacuum_analyze_scale_factor (0.01 or less, set per table) so the next backfill doesn't repeat this. One command, no topology change.

Correlated columns: CREATE STATISTICS. The independence layer needs extended statistics:

CREATE STATISTICS events_region_type (dependencies, mcv)
  ON region, event_type FROM events;
ANALYZE events;

dependencies tells the planner that one column's value constrains the other; mcv gives it a joint most-common-values list, so ('me-south', 'checkout') is estimated as the combination it is rather than the product of two marginals. This is the durable fix for the pattern, and it costs almost nothing to keep.

Function-wrapped predicates: give the expression statistics. An index on lower(email) does double duty: it makes the lookup indexable, and Postgres gathers statistics on expression indexes, so the estimate becomes real. On PostgreSQL 14 and later you can also do CREATE STATISTICS ON (lower(email)) FROM users if you want the statistics without the index.

Cross-table correlation: materialize and re-analyze. Extended statistics are per table. When the correlation crosses a join (say enterprise accounts generate most failed checkouts, so a filter on accounts.plan changes the selectivity of a filter on events), nothing in pg_statistic can express it, and the row count coming out of the join is a guess multiplied by a guess. That's where a temp table earns its keep:

CREATE TEMP TABLE stage AS
SELECT e.id, e.account_id, e.product_id, e.occurred_at
FROM events e
JOIN accounts a ON a.id = e.account_id
WHERE e.region = 'me-south'
  AND e.event_type = 'checkout'
  AND a.plan = 'enterprise';

ANALYZE stage;

SELECT s.*, p.name
FROM stage s
JOIN products p ON p.id = s.product_id;

Be precise about what this buys. The join inside CREATE TEMP TABLE AS still runs with whatever plan the planner picks for it. If that's the bad nested loop, materializing doesn't rescue it; the rungs above do. What materialization fixes is everything downstream. ANALYZE stage replaces a chain of compounded guesses with a sample of the actual intermediate result, and the join to products is planned on a measured number. A WITH ... AS MATERIALIZED CTE doesn't do this: it forces the intermediate result to exist, but the planner still costs the rest of the query on the CTE's own estimate, which is the same compounded guess. The overhead of the temp table is one write of the intermediate rows and one ANALYZE pass, seconds against a query that was taking a minute, and it ships as a query change.

What sharding does to the estimate

Now suppose none of that was tried and the team shards. In Postgres that usually means Citus: tables are split into shards spread across worker nodes by a distribution column, typically the tenant, and a coordinator routes single-shard queries and fans out the rest. Take a 32-shard cluster distributed by account_id.

Each shard is an ordinary Postgres table on an ordinary Postgres node, with its own pg_statistic, maintained by that node's autovacuum, computed from the rows the distribution column sent there. Nothing about the estimator has changed. region and event_type are still assumed independent on every worker. The backfill that didn't trip the analyze threshold on one node is now spread across 32 nodes with 32 smaller thresholds, and it's still under every one of them. Extended statistics, if you created any, now need to exist on every shard. And the plan for each shard's fragment is chosen from that shard's local statistics, so the same query against the same schema yields the same misestimate everywhere.

It can get worse than "the same". Shards keyed by tenant are not uniform slices. One shard hosts a handful of large accounts, another hosts ten thousand small ones, and their MCV lists and histograms differ accordingly. The coordinator has no merged histogram for the table as a whole. For a query that fans out, each worker plans its fragment locally, and the step that combines them on the coordinator (an aggregate, a repartition join, an intermediate result pulled back for a final join) is planned without a global view of the data. postgres_fdw is explicit about the same problem: its use_remote_estimate option exists because local statistics for remote data are often wrong enough that it's better to ask the remote server to EXPLAIN.

So instead of one bad nested loop, there are 32 of them running in parallel, one per shard, with a combine step on top that has no real statistics to draw on. The temp-table fix stops composing, because there is no longer one plan to fix. And the diagnostic gets harder: the EXPLAIN ANALYZE that would have named the defect in a minute on one node now has to be read per shard and reconciled by hand.

When sharding is the right call

None of this means sharding is never warranted. It's warranted when the constraint is throughput rather than a plan, and that's testable before you commit.

Some ceilings are physical. A single primary's write rate is bounded by how fast it can generate and durably flush WAL. Checkpointing and autovacuum contend with a write-heavy workload for the same I/O, and no plan change touches that. A working set larger than memory pages to disk regardless of join order. If EXPLAIN ANALYZE on the production workload shows estimated and actual rows converging on every node, and the machine is still saturated under representative load, that's a real ceiling.

Even then, single-node partitioning usually comes first. Declarative partitioning lets the planner skip partitions that cannot match a filter, keeps maintenance operations proportional to partition size, and doesn't cost you cross-table joins or foreign keys. Partitioning and sharding compose: partition on one node now, shard the partitions across nodes later if you still need to. Retrofitting partitioning onto a live table is migration work, not a config flag, but it stays contained to one database in a way that sharding does not.

If none of that closes the gap, shard, and go in knowing the bill: cross-shard joins get expensive or disappear, foreign keys across the shard boundary stop being enforceable by the database, and transactional guarantees that used to be free become an application-level concern. That's worth paying once the plans are proven correct and the hardware is still the constraint. It isn't a hedge to pay because a dashboard looked bad.

Fragment the data before doing that proof, and a bad plan stops being a query problem fixable with one ANALYZE. It becomes a property of the architecture, replicated across every shard you added, running in parallel, and harder to undo than it was when you had one machine.

References

• PostgreSQL: Statistics Used by the Planner
• PostgreSQL: Row Estimation Examples
• PostgreSQL: Using EXPLAIN
• PostgreSQL: CREATE STATISTICS
• PostgreSQL: The Autovacuum Daemon
• PostgreSQL: pg_stat_statements
• PostgreSQL: postgres_fdw
• Citus: Understanding partitioning and sharding in Postgres and Citus