When One Server Isn't Enough
Everything in this book so far scales one machine. Reads can be spread across replicas, which Chapter 13 builds, but every write in a Postgres cluster goes to exactly one primary, and that is architectural rather than a tuning matter. Eventually a workload arrives that one primary cannot take, and the answer is to put the data on several machines that each own a slice of it.
This topic exists mostly so you can recognize that moment, and far more often recognize its absence. Sharding is sold early and constantly, usually to teams whose primary is running on defaults with no pooler in front of it. The ending is stated up front: Cartwheel does not shard, is not close to needing to, and the rest of this page is the reasoning that lets you say so with a number attached.
How Far One Server Goes
Further than the conversation usually assumes. A modern two-socket server with NVMe storage runs a Postgres primary at tens of thousands of transactions per second and holds tens of terabytes, and the constraint that bites first is almost never raw write capacity. It is a missing index, an unpooled connection count, a table that has to be maintained as one unit, or a report competing with checkout for the same buffers. Every one of those has a fix that costs a week and no architectural change.
So the ladder runs in a specific order, cheapest first, and each rung should end with a measurement rather than a feeling. Pool the connections, because 400 backends on 16 cores is a throughput loss before it is a capacity problem. Fix the queries and the indexes, because the top three statements by total execution time are usually most of the load. Partition the tables whose maintenance no longer fits the window, which is what this chapter has been doing. Add read replicas to move the analytics traffic off the write path. Then buy a bigger machine: doubling the cores costs a purchase order and changes no application code, and it is the rung teams skip because it sounds unsophisticated.
What Sharding Costs
Distribution is the only one of these that scales writes, and it is priced accordingly. Cross-shard joins either become expensive or become impossible, depending on the system, and the ones that survive do so because the tables were deliberately arranged to make them local. Transactions spanning shards need two-phase commit and get slower and more failure-prone. Global uniqueness weakens: a unique constraint can be enforced within a shard cheaply and across shards only with coordination, which is the same rule partitioning taught, one layer up and with a network in it. Foreign keys across shards are usually not available at all.
The costs that hurt most are the ones outside the database. Every query the application issues must carry the shard key or fan out to all workers, which is a rewrite of the data access layer rather than a configuration change. Rebalancing shards when one grows faster than the others is an operational project with a runbook and a rollback plan, not a maintenance task. And the schema is now pinned to the distribution decision: changing the shard key later means moving all of the data again. None of those costs is paid once. They are permanent constraints on the data model, priced in every feature written after the migration.
Citus, the Postgres Extension for Distribution
Citus is the mature way to do this while staying on Postgres. It is an extension, not a fork, so the SQL, the types and the drivers are the ones you already have. One node is designated the coordinator and holds the metadata; the others are workers. A distributed table is split into shards by the hash of its distribution column, each shard being an ordinary Postgres table on a worker, and the coordinator routes each incoming query — to a single worker when the query names one distribution-column value, or in parallel across many as per-shard tasks when it does not. Small tables can be declared reference tables, which are replicated in full onto every worker so that joins against them never leave the node.
SELECT create_distributed_table('orders', 'customer_id'); SELECT create_distributed_table('order_items', 'customer_id'); -- colocated SELECT create_distributed_table('delivery_events', 'customer_id'); -- colocated SELECT create_reference_table('product_categories'); -- on every worker
Tables distributed on the same column are colocated: rows with the same customer_id land in matching shards on the same worker, so a join between them runs locally and never crosses the network. Notice what those three lines require of Cartwheel's schema. Neither order_items nor delivery_events has a customer_id column today; they reach the customer through order_id. Distributing them by customer means adding that column to both and backfilling it, on a table of 1.2 billion rows, because the distribution column has to physically exist in the table it distributes.
Choosing a Distribution Column
This is the decision that determines whether the system works, and it is made once. A column that groups related data, a tenant id in a SaaS product or customer_id at Cartwheel, keeps almost every query on a single worker, because almost every query is about one tenant or one customer. A column chosen for even distribution instead, an event id or a random uuid, spreads the load beautifully and makes a large fraction of queries fan out to every node, where the slowest worker sets the latency and the coordinator has to combine partial results.
Work it out from the query log, not from the entity-relationship diagram. Take the twenty statements with the highest total execution time, and for each one ask which single value, if it were known, would confine the query to one shard. When the same column answers for most of them, that is the distribution column. When no column answers for most of them, and the workload is genuinely a mix of per-customer reads and cross-customer aggregates, there are two workloads on one server, and the second of them belongs in a warehouse rather than on a shard key.
The Alternatives
Citus has three competitors that deserve considering first. Read replicas scale reads and nothing else, at the cost of a read path that sees slightly stale data; Cartwheel's dashboard already accepts 30 seconds of lag, which is what makes pg-replica-a viable. Moving the analytical half of the workload out to a warehouse removes the queries that were driving the "we need more machines" conclusion, and often reveals that the OLTP half fits comfortably on one primary. Application-level sharding, where the application picks a database per customer bucket, gives complete control and hands you every cross-shard join, every rebalance and every schema rollout to write yourself; it remains common and rather more defensible than its reputation suggests, particularly when the shard boundary is a genuine business boundary.
The fourth option is a different engine. CockroachDB and YugabyteDB speak the PostgreSQL wire protocol and support much of its SQL, and they distribute writes natively. What they do not give you is Postgres: the plans, the transaction costs, the extension ecosystem and the operational behaviour are all their own. psql connecting is a statement about the wire protocol and about nothing underneath it.
The Honest Recommendation
Cartwheel peaks at 3,000 orders a minute. That is 50 write transactions a second on the checkout path, on a 16-core primary with 64 GB of RAM and NVMe storage, which now has a pooler in front of it, an event table partitioned by month, and a replica carrying the reporting load. There is at least an order of magnitude of headroom in that configuration before the primary is the constraint, and the next thing to break will be something specific and fixable, a query or a lock or a disk, rather than the fundamental capacity of one machine.
So the recommendation is to do nothing, and to know precisely what "nothing" is protecting. Write down the numbers that would change the answer: sustained write transactions per second, the p95 of the checkout statement, and the point at which replicas stop absorbing the read growth. Revisit them quarterly. Producing those three numbers on request costs one saved query and takes about a minute. Chapter 12 turns to the other thing one server has to get right, which is getting the data back when something removes it.
Vertical scaling buys more cores, more RAM and faster storage. It is the cheapest scaling anyone ever does, because the application does not change by a line, and the one people skip because it feels unsophisticated.
Read replicas — scale reads, and cost you a read path that can see stale data plus WAL shipping load on the primary. They do not scale writes by any amount, and adding more of them makes the primary do slightly more work.
Sharding — the only option that scales writes, and the only one that permanently changes the data model: a shard key in every query, cross-shard operations that are expensive or unavailable, and rebalancing as an ongoing operational discipline.
- Sharding before the primary is tuned — every query problem reappears on every worker, now with coordination overhead layered on top of it.
- Choosing a distribution column for even data spread rather than for the dominant query pattern, so most statements fan out to all workers and the slowest one sets the latency.
- Assuming a Postgres-compatible distributed engine behaves like Postgres, when the wire protocol matches and the plans, extensions and transaction costs do not.
- Counting read replicas as write capacity — the primary still takes every write, and each replica adds WAL shipping work to it.
- Building application-level sharding with no rebalancing plan, then discovering that shard 3 is out of disk and there is no procedure for moving customers off it.
- Distributing tables on a column that does not exist in them yet, without pricing the backfill and the model change that adding it requires.
- Exhaust pooling, query and index fixes, partitioning, replicas and bigger hardware in that order, and record the measurement that closes each step.
- Derive the distribution column from the twenty statements with the highest total execution time, not from the schema diagram.
- Colocate every set of tables that must be joined on the distribution column, and keep small lookup tables such as
product_categoriesas reference tables. - Move the analytical workload to a warehouse or a replica before concluding that the OLTP workload needs distribution.
- Write down the three numbers that would justify sharding, and review them quarterly instead of arguing from intuition.
- Benchmark any Postgres-compatible distributed engine against your own top queries before treating the compatibility claim as a migration plan.
Knowledge Check
Cartwheel's write load is rising. Which option is the only one that actually scales writes?
- Adding two more streaming replicas to spread the load off the primary
- Putting a second PgBouncer in front and enlarging the server-side pool
- Distributing the data so several nodes each own a slice of the writes
- Partitioning the remaining large tables so each write touches less data
What does the distribution column actually decide in a Citus cluster?
- How many replicas of each shard are kept across the worker nodes
- Which queries reach one worker and which have to fan out to all of them
- Which of the nodes in the cluster takes on the coordinator role
- The order in which rows are physically stored inside each shard
Why does Citus offer reference tables alongside distributed ones?
- They are read-only copies, so writes never have to be coordinated at all
- A small table copied to every worker keeps joins against it local
- They live only on the coordinator, which removes them from the shard count
- They cache frequently read rows and are refreshed on a configurable interval
Distributing order_items and delivery_events by customer_id requires what of Cartwheel's schema?
- A foreign key from each table to orders, which Citus then follows for routing
- Adding and backfilling a customer_id column on both, since neither has one
- Nothing, since both already carry customer_id inherited from the orders table
- A view exposing customer_id, which the coordinator uses to place the rows
At 3,000 orders a minute, what is the honest recommendation for Cartwheel?
- Migrate to a distributed Postgres-compatible engine before growth continues
- Do nothing, and write down the numbers that would change the answer
- Adopt Citus now, while the schema is small enough to distribute cheaply
- Shard in the application layer, which avoids the cost of the extension
You got correct