Shard last

On a small sharded cluster built for this chapter, the same join between projects and users ran in about 10 milliseconds one way and about 950 milliseconds the other. Same rows, same machines, same answer. The difference was one extra equality in the on clause, u.account_id = p.account_id. With it, every shard joined its own rows locally. Without it, the cluster had to reshuffle both tables across the network first, and by default it refused to try.

That's sharding. Every query is either kept local by the shard key or pays for the network, and you choose which on the day you pick the key.

We think most teams shard too early, and too few design for it before they do. Design for it from day one, then wait as long as you can. When the day comes, the tool matters less than the key.

Sharding multiplies everything

Sharding splits one logical database across several independent primaries. A shard is one subset of the rows; the shard key (or distribution column) decides which shard a row lives on: here, account_id, with a router in front that knows where each account lives.

Short of splitting the database by function, it's the only rung of the scaling ladder that raises the ceiling on writes and total data. The price is a boundary through everything Postgres did for you inside one server:

• Queries that don't name one shard key go to every shard and get merged; joins across shards ship rows over the network.
• Transactions across shards need two-phase commit to stay atomic.
• Constraints live on one shard: (account_id, email) can be unique, email alone can't.
• Migrations and operations happen N times: N sets of replicas, backups, vacuum backlogs, and places a 3 a.m. page can come from.

None of that is a reason never to shard. It's a reason to make sure the alternative is worse first.

Shard when one primary can no longer absorb writes or hold data

Three signals say one primary, on the biggest machine you're willing to run, can no longer absorb your writes or hold your data:

• Write throughput is pinned: writes, not reads, saturate the primary even after tuning WAL and checkpoints and batching.
• The data no longer fits the box, operationally: a restore longer than your recovery time objective, a vacuum that never catches up.
• A hard requirement splits tenants: data residency, or dedicated hardware by contract. The one case where sharding early is right.

Not signals: "we might be big one day", a slow query, a connection count, read load. Try the cheaper rungs first, including the underrated one: split by function, moving events and its writers to their own database, so each side still behaves like one Postgres.

The shard key is the whole design

Pick the shard key so your hottest queries, and the transactions you can't afford to split, stay on one shard. A good one is in almost every hot query's filter, is shared by rows joined or written together, and has many evenly spread values.

For multi-tenant SaaS, that's the tenant id: the app serves one account at a time, and an account's rows are joined together, never with another account's. Shard events by id and every "my account's events" query hits every shard; by created_at, all of today's writes land on one. Cross-tenant reports become fan-out: fine if rare.

A hash spreads tenants, not load. One customer's CI system adding 400,000 events turned its 13 MB shard into 61 MB, shared with every account that hashes there. Big tenants need their own shard, so moving one tenant must be routine.

Colocation keeps joins local. Shard users, projects, and events by account_id the same way and account 42's rows sit together, joined locally, if the join condition includes the shard key: the 10 ms against 950. Small tables every tenant reads are copied to every shard as reference tables.

 shardid | nodename | size
---------+----------+-------
  102147 | citus-w1 | 61 MB
  102166 | citus-w2 | 14 MB
  102169 | citus-w1 | 13 MB

prove "account 7's shard is now the largest, over 3 times the median events shard" \
  holds citus-coord "with s as (select shardid, shard_size from citus_shards where table_name = 'events'::regclass),
    seven as (select shardid from pg_dist_shard where logicalrelid = 'events'::regclass and hashint8(7) between shardminvalue::int and shardmaxvalue::int)
    select (select shardid from s order by shard_size desc limit 1) = (select shardid from seven)
       and (select max(shard_size) from s) > 3 * (select percentile_cont(0.5) within group (order by shard_size) from s)"

Keep the door open from day one

You can make sharding cheap without doing it. The same habits make row-level security simpler and cross-tenant bugs harder to write. Put account_id on every tenant-owned table, in every primary key, unique constraint, and foreign key, and in every query's where clause (Carry the tenant everywhere, Make cross-tenant references impossible). On one server those composite keys prevent cross-tenant references; sharded, they're the only kind that can be enforced.

Already on single-column keys? Migrate on one server, step by step: composite unique constraints, composite foreign keys (not valid, then validate), then queries. Each step pays for itself.

Route by tenant, never by id

Each shard has its own sequences, so two shards both hand out id = 1. If ids leave the database, use shard bits (the top bits of a bigint name the shard, Snowflake-style) or UUIDv7 (no coordination, uuidv7() in PostgreSQL 18); the id landscape compares the rest.

With 16 shard bits, each shard can issue 281 trillion ids. But an id says where a row was created, not where it lives after its tenant moves: route by tenant, never by id.

-- on citus-coord
   id    | node_bits
---------+-----------
 2000001 |         0
 
-- on citus-w1
       id        | node_bits
-----------------+-----------
 281474976710657 |         1
 
-- on citus-w2
       id        | node_bits
-----------------+-----------
 562949953421313 |         2

id_sql="insert into events (account_id, project_id, kind)
  select 42, id, 'login' from projects where account_id = 42 order by id limit 1
  returning id >> 48"
c_bits=$(coord -c "$id_sql"); w1_bits=$(on citus-w1 -c "$id_sql"); w2_bits=$(on citus-w2 -c "$id_sql")
prove "the same insert accepted on every node, ids carry the node's group in the top 16 bits" \
  [ "$c_bits $w1_bits $w2_bits" = "0 1 2" ]

Cross-shard queries fan out

A query without the shard key goes to every shard, and something merges the results: sums add, averages need sums and counts, limit 10 needs each shard's top 10, distinct counts need the values. Latency is the slowest shard's, and 100 such queries a second on 32 shards is 3,200 tasks a second. Tenant queries get cheaper per node as you add shards. Fan-out doesn't. Keep it to jobs and internal tools.

Cross-shard transactions need two-phase commit

Writes have the same boundary. A write to two shards is two transactions; if one fails, you have half a transfer. Two-phase commit has every shard prepare transaction before any commits. It holds locks everywhere until the second phase, and a coordinator that dies in between leaves prepared transactions holding locks and blocking vacuum. Plain PostgreSQL ships with it off (max_prepared_transactions = 0), and postgres_fdw doesn't use it. The best cross-shard transaction is the one your shard key made unnecessary.

out=$(coord -c "set citus.log_remote_commands = on" -c "set client_min_messages = notice" \
  -c "update events set kind = kind where kind = 'alert'" 2>&1)
prove "an update with no tenant filter commits with two-phase commit on both workers" \
  bash -c 'grep -q "PREPARE TRANSACTION" <<<"$1" && grep -q "COMMIT PREPARED" <<<"$1" && grep -q "citus-w1" <<<"$1" && grep -q "citus-w2" <<<"$1"' _ "$out"

Resharding means moving tenants

Sooner or later a tenant needs a machine of its own. Logical replication with a row filter (PostgreSQL 15+, what it's for) copies one tenant's rows, then streams changes.

With a tenant directory, a table mapping each tenant to its shard, the move takes four steps: publish where (account_id = 42) and subscribe; when lag is near zero, mark the tenant read-only; at zero, point the directory at the destination; clean up the source in batches. The read-only window lasts seconds. Logical replication doesn't carry sequences, so ids minted on the destination must not collide with the moved ones: one more reason for shard bits or UUIDv7.

The shard moves in Citus (an extension that shards inside Postgres, below) and the resharding in PgDog (a sharding proxy, below) use the same protocol, a good sign it's the right primitive.

-- on the source
create publication move_acct_42 for table events where (account_id = 42);
 
-- on the destination, which already has the table
create subscription move_acct_42
  connection 'host=citus-w1 dbname=fdwshard user=postgres'
  publication move_acct_42;

prove "the subscription copies account 42's 500 existing rows, and only those" \
  wait_for pg dest "select count(*) = 500 and min(account_id) = 42 and max(account_id) = 42 from events"

There are seven ways to shard a Postgres workload

The seven differ in where routing lives, who merges cross-shard results, and how much of Postgres you keep. Every fact about other projects in this section is from their own documentation as of October 2026; check before you commit. We ran postgres_fdw and Citus because both are open source and run anywhere: easy to demonstrate, not the default answer.

An application router makes every hard problem yours

You run ordinary primaries, and a router in your code looks up each tenant's shard. Any Postgres works, managed or not, and every hard problem (merges, moves, atomicity) is yours.

create table tenant_shards (
  account_id  bigint primary key,
  shard       text   not null,
  state       text   not null default 'active'
              check (state in ('active', 'read_only', 'moving'))
);

// Router sends every query for an account to the database that owns it.
type Router struct {
	mu        sync.RWMutex
	directory map[int64]string   // account_id -> shard, loaded from tenant_shards
	shards    map[string]*sql.DB // shard name -> connection pool
}
 
func (r *Router) For(accountID int64) (*sql.DB, error) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	name, ok := r.directory[accountID]
	if !ok {
		return nil, fmt.Errorf("account %d has no shard", accountID)
	}
	return r.shards[name], nil
}

postgres_fdw is a building block

Make each partition of a hash-partitioned table a foreign table on another server and the parent routes rows across machines. Tenant queries push their filter to one server; cross-tenant aggregates pull every row back. No two-phase commit, constraints, or rebalancing: a kit, not a car.

create server shard1 foreign data wrapper postgres_fdw
  options (host 'citus-w1', dbname 'fdwshard', async_capable 'true');
-- shard2 likewise, plus user mappings
 
create table events (
  account_id bigint not null,
  id         bigint not null,
  kind       text   not null
) partition by hash (account_id);
create foreign table events_s1 partition of events
  for values with (modulus 2, remainder 0)
  server shard1 options (table_name 'events');
-- events_s2: remainder 1, server shard2

prove "50,000 rows for 100 accounts route 26,000 to one server and 24,000 to the other" \
  [ "$(on citus-w1 -d fdwshard -c 'select count(*) from events') $(on citus-w2 -d fdwshard -c 'select count(*) from events')" = "26000 24000" ]
plan=$(on pg -d router -c "set enable_partitionwise_aggregate = on" -c "explain (verbose, costs off) select kind, count(*) from events group by kind")
prove "grouping by anything else pulls the rows back and counts locally" \
  bash -c 'grep -qF "Remote SQL: SELECT kind FROM public.events" <<<"$1" && ! grep -q "Relations: Aggregate on" <<<"$1"' _ "$plan"

Citus routes and merges inside Postgres

Citus is an open-source extension (AGPL-3.0) that makes a group of Postgres servers one distributed database: a coordinator you connect to, workers holding the shards, and routing, merges, two-phase commit, DDL, and rebalancing handled for you.

The latest release is 14.2.0 (14.0 added PostgreSQL 18). On our cluster (14.2.0 on PostgreSQL 18.4, the version its image ships; two workers, 32 colocated shards by account_id), a tenant query was one task on one worker, a cross-tenant one was 32 tasks merged on the coordinator, and the opening's join without account_id was refused.

Constraints must include the distribution column

A shard can only check its own rows, so Citus rejects any primary key, unique, or exclusion constraint without the distribution column: the original primary key (id) can't be distributed by account_id, nor unique (email). Foreign keys must join colocated tables on the distribution column, or point at a reference table. The day-one composite keys were all accepted.

select create_reference_table('accounts');
select create_distributed_table('users',    'account_id');
select create_distributed_table('projects', 'account_id',
                                colocate_with => 'users');
select create_distributed_table('events',   'account_id',
                                colocate_with => 'users');

ERROR:  cannot create constraint on "t1"
DETAIL:  Distributed relations cannot have UNIQUE, EXCLUDE, or PRIMARY KEY
constraints that do not include the partition column (with an equality
operator if EXCLUDE).

 Custom Scan (Citus Adaptive) (actual rows=5.00 loops=1)
   Task Count: 1
   Tuple data received from nodes: 68 bytes
   ->  Task
         Node: host=citus-w2 port=5432 dbname=postgres
         ->  GroupAggregate (actual rows=5.00 loops=1)
               Buffers: shared hit=1092 read=10
               ...
                     ->  Bitmap Heap Scan on events_102166 events
 Execution Time: 1.747 ms

prove "account 42's users, projects, and events sit on the same worker" \
  holds citus-coord "select count(distinct nodename) = 1 from citus_shards s join pg_dist_shard d using (shardid) where s.table_name in ('users'::regclass, 'projects'::regclass, 'events'::regclass) and hashint8(42) between d.shardminvalue::int and d.shardmaxvalue::int"
join_sql="select u.email, count(*) from projects p join users u on u.id = p.owner_id group by u.email order by 2 desc limit 3"
prove "a join without account_id is refused by default" \
  fails_with "the query contains a join that requires repartitioning" citus-coord "$join_sql"
slow=$(best_ms "select count(*) from projects p join users u on u.id = p.owner_id")
fast=$(best_ms "select count(*) from projects p join users u on u.account_id = p.account_id and u.id = p.owner_id")
prove "adding account_id to the join makes it at least 5 times faster" \
  awk -v s="$slow" -v f="$fast" 'BEGIN { exit !(s > 5 * f) }'

select isolate_tenant_to_new_shard(
  'events', 7,
  cascade_option      => 'CASCADE',
  shard_transfer_mode => 'block_writes');

new_shard=$(coord -c "select isolate_tenant_to_new_shard('events', 7, cascade_option => 'CASCADE', shard_transfer_mode => 'block_writes')")
prove "isolate_tenant_to_new_shard gives account 7 a shard whose hash range is a single value" \
  holds citus-coord "select shardminvalue = shardmaxvalue and shardminvalue::int = hashint8(7) from pg_dist_shard where shardid = $new_shard"

A proxy keeps the shards plain Postgres

A sharding proxy sits where a pooler would, parses each query, and sends it to the right plain-Postgres primary, so the shards need no extension. Three have public code: PgDog, SPQR, and Multigres. What one can merge across shards is whatever its parser supports today; test your own queries through it.

Managed sharding costs you features and an easy exit

Three providers run the cluster for you: Amazon Aurora PostgreSQL Limitless Database (generally available since October 31, 2024), Azure Database for PostgreSQL elastic clusters (managed Citus), and PlanetScale Neki (platform preview since September 10, 2026). You take the provider's versions and feature subset, and the exit is harder (the hosts themselves).

Distributed SQL speaks Postgres but isn't Postgres

CockroachDB and YugabyteDB speak the Postgres protocol and much of its SQL, but underneath they're different databases: small Raft-replicated units that split and rebalance themselves, with distributed transactions built in. What this book says about pages, vacuum, and the heap describes Postgres, not them. Choosing one is a migration to a different engine.

Schema-per-tenant costs grow with every tenant

The low-tech option is a schema, or a whole database, per tenant, placed on whichever server you like. Isolation is structural, and moving one tenant is one dump. The cost grows with tenant count: 1,000 tenants with three empty tables added 16,000 relations and over 40 MB of catalog before storing a row, and every migration runs 1,000 times, each with its own lock and its own chance to fail.

                          before    after 1,000 tenants
 relations in pg_class       415             16,415
 pg_attribute rows         3,161            100,161
 system catalog size      7.7 MB            56.0 MB
 database size, no rows   7.6 MB           134.0 MB

prove "1,000 tenants with three empty tables add exactly 16,000 relations" [ $((rel1 - rel0)) -eq 16000 ]
prove "the system catalogs grow by more than 40 MB before storing a row" [ $(( (cat1 - cat0) / 1048576 )) -gt 40 ]

Choose by where the hard parts live

Each option moves the hard parts somewhere else

How each handles the hard parts:

And what it costs to live with:

Notes on the managed row: (1) Limitless and elastic clusters make cross-shard writes atomic; Neki, in preview, commits them shard by shard. (2) Limitless splits shards but can't merge them; Neki reshards online. (3) High for Limitless and Neki, lower for elastic clusters, whose Citus you could run yourself. "Proxy" means PgDog and the others above; Citus "any node" is since Citus 11. Operational load is high for the do-it-yourself options because you run N primaries plus your own tooling; a proxy adds a service on the critical path; distributed SQL is a different database to learn.

Most teams don't need to shard yet

• You don't need to shard yet if a bigger box, replicas, partitioning, or splitting off one workload buys two more years. That's most teams.
• Tens of large tenants, or residency rules: database-per-tenant or an application router with a directory.
• Many tenants on shared tables, and you want Postgres to merge: Citus, self-hosted or managed (Azure elastic clusters), or a sharding proxy to keep the shards extension-free. Prototype both.
• Deep in AWS and fine with its feature subset: Aurora PostgreSQL Limitless Database. Read its unsupported-features list first.
• A managed router over plain Postgres, without atomic cross-shard writes yet: PlanetScale Neki, while in preview. Read its limitations first.
• Multi-region writes or hands-off resharding over full compatibility: a distributed SQL database, knowing it's a different engine.

Whichever you look at, run your ten most important queries on it and read the plans.

The short version

• Shard last. Short of splitting by function, it's the only rung that scales writes and storage past one machine, and it makes every cross-shard query, transaction, constraint, and migration harder.
• Real signals: writes saturating the biggest primary, data too big to restore within your RTO, or residency. Not read load, not "someday".
• Shard multi-tenant SaaS by tenant id, and start today: account_id on every tenant table, in every key, and in every query.
• Colocated joins on the shard key stay local: about 10 ms with account_id in the on clause, about 950 ms without.
• Anything globally unique, like email, needs its own unsharded table.
• Plan to move hot tenants: a directory plus row-filtered logical replication.
• Pick the approach by where routing, merging, and two-phase commit should live, and check your hottest queries' plans for fan-out.

Further reading

• postgres_fdw, PostgreSQL documentation. Foreign tables, pushdown, async_capable, and the note that it doesn't do two-phase commit.
• Citus, source repository. The extension used here, its license, and row- and schema-based sharding.
• Aurora PostgreSQL Limitless Database architecture, AWS documentation. Routers, shards, and table types; read the requirements page beside it.
• PgDog, source repository. A sharding proxy's view of routing, merges, two-phase commit, and resharding.
• Neki query planning, PlanetScale documentation. How a router picks one shard or all, and what cross-shard operations don't guarantee.
• CockroachDB PostgreSQL compatibility, CockroachDB documentation. The concrete list of what differs.