Start practising

SYSTEM DESIGN · BUILDING BLOCKS · 5 OF 14

Relational Database at Scale: From One Box to Shards

The order of moves that takes a relational database from one server to a fleet: scale up and tune, pool connections, add read replicas and handle lag, partition big tables, split by feature, then shard with a routing layer. Plus failover without split brain, backups and point-in-time recovery, online schema changes, and how to size it.

  1. 1
  2. 2
  3. 3
  4. 4
  5. 5
  6. 6
  7. 7
  8. 8
  9. 9
  10. 10
  11. 11
  12. 12
  13. 13
  14. 14
1 / 15

Press play to hear the narration, or step through at your own pace.

Notes

The problem

With one primary, every read, write and connection lands on one machine. What runs out first is usually not what people expect. Connections: two hundred app servers with twenty connections each is four thousand, and in Postgres each is a process with its own memory. Then CPU from heavy queries, then disk once the working set no longer fits in RAM, and finally raw write throughput.

What runs out first
  • connections: each one costs
  • memory (Postgres: a process)
  • CPU: unindexed or heavy
  • queries, sorts, joins
  • disk IO: working set no
  • longer fits in RAM
  • write throughput: one WAL,
  • one machine

Scale up first

Move one is boring and powerful: fix the queries. Find the top statements by total time, add the right composite indexes, kill the N plus one loops. Then scale up. A single modern server with sixty-four cores, half a terabyte of RAM and NVMe drives handles more than most products ever need. It adds no new failure modes, and it buys you months.

Move 1: scale up and tune
  • find slow queries first
  • (pg_stat_statements, slow log)
  • add the right indexes
  • (Concepts episode 17)
  • bigger box: 64+ cores,
  • 512 GB-1 TB RAM, NVMe
  • handles more than most
  • products ever need
  • cheapest move, no new failure
  • modes, buys months

Connection pooling

Move two: a connection pooler like PgBouncer or ProxySQL. Four thousand app connections funnel into maybe a hundred real database connections, which is closer to what the server runs efficiently, a few per core. In transaction mode, a server connection is lent out for one transaction and returned. The catch: session state like SET commands or temporary tables doesn't survive between transactions.

Move 2: pool connections
  • app: 200 pods × 20 = 4,000
  • DB sweet spot: ~2-4 × cores
  • → pooler multiplexes them
  • transaction pooling: lend a
  • server connection for one
  • transaction, then return it
  • caveat: no session state
  • (SET, temp tables, advisory
  • locks) across transactions

Read replicas

Move three: read replicas. The primary streams its write-ahead log to replicas, which replay it. Reads now scale almost linearly with replica count; writes don't scale at all. Replication is usually asynchronous, so replicas lag, normally milliseconds but seconds under heavy writes. A user who just edited their profile might see the old one. Route that user's reads to the primary for a few seconds after a write.

Move 3: read replicas
  • primary streams its log
  • to N replicas (async)
  • reads scale ~linearly
  • writes do not scale at all
  • lag: usually ms, seconds
  • under heavy write load
  • read-your-writes: send the
  • user to primary for ~5 s
  • after they write

Partition tables

Move four: partition big tables inside the same database. A five-billion-row orders table becomes one partition per month. Queries that filter on the date only touch the relevant partition, indexes stay small enough to live in memory, and deleting old data is dropping a partition instantly instead of a delete that runs for hours. It's still one server, but it postpones a lot of pain.

Move 4: table partitioning
CREATE TABLE orders (...)
  PARTITION BY RANGE (created_at);
orders_2026_09, orders_2026_10 ...

queries with created_at only
  touch one partition (pruning)
drop old data: DROP PARTITION
  (instant, no huge DELETE)
indexes stay small and hot

still one server

Split by feature

Move five: split by feature, sometimes called vertical partitioning. Users, orders, catalogue and payments each get their own database, often owned by their own service. Each gets a whole machine and failures are isolated. You lose joins and transactions across them, so cross-domain workflows become sagas or the outbox pattern from Concepts episode thirty-three.

Move 5: split by feature
  • one DB per domain or service:
  • users, orders, catalogue,
  • payments
  • + each gets a whole machine
  • + failures are isolated
  • - no joins across them
  • - no single transaction across
  • them → sagas, outbox
  • (Concepts episode 33)

Sharding

Only when writes or data size exceed one machine do you shard: the same schema on many primaries, with rows divided by a shard key. Choose the key most queries filter on, like user id or tenant id, and keep a user's orders on the same shard as the user so joins stay local. A routing layer, like Vitess or Citus or a lookup table in your app, sends each query to the right shard. Queries across shards are slow and should be rare.

Move 6: shard
  • same schema on many primaries
  • rows split by a shard key
  • key: what most queries filter
  • on (user_id, tenant_id)
  • co-locate a user's orders
  • with the user → local joins
  • router: Vitess, Citus, or
  • app library + lookup table
  • cross-shard queries: avoid

Sharding costs

Sharding has costs you should name. A hot shard, when one big customer is thirty percent of traffic. Resharding, which is painful row by row, so start with many logical shards, say a thousand and twenty-four, on a few machines, and move whole shards later. Global uniqueness, so you need an ID generator. And global queries, like reports across all users, which move to a search index or an analytics store.

Sharding costs
  • hot shard: one tenant is 30%
  • → split it, or give it its own
  • resharding: start with many
  • logical shards (e.g. 1,024)
  • on few machines; move shards
  • not rows
  • unique constraints and IDs
  • across shards → ID generator
  • global queries → search/OLAP

Failover

Failover. If you promote an asynchronous replica, you lose whatever it hadn't received: your data loss window equals the lag. A synchronous standby makes each commit wait for its acknowledgement, a millisecond or two, so a failover loses nothing. A manager like Patroni or a managed service detects the failure, promotes the standby, and repoints clients in about thirty seconds. Crucially, it fences the old primary so two primaries never accept writes at once.

Surviving the primary dying
  • async replica promoted:
  • RPO = lag (lose last writes)
  • sync standby: commit waits
  • for its ack → RPO = 0,
  • +1-2 ms per write
  • manager (Patroni, Orchestrator,
  • RDS): detect, promote, repoint
  • RTO ~30 s
  • fence the old primary
  • (no split brain)

Backups

Replicas are not backups. A bad DROP TABLE reaches every replica in milliseconds. Take a base backup nightly and archive the write-ahead log continuously to object storage. Then you can restore to any second, like one minute before the mistake. And test restores on a schedule, because how long a restore takes is your real recovery time in a disaster.

Backups and recovery
  • replicas are NOT backups:
  • DROP TABLE replicates in ms
  • nightly base backup +
  • continuous WAL archive
  • → point-in-time recovery
  • ("restore to 14:31:59")
  • test restores regularly;
  • restore time = real RTO
  • for a disaster

Schema changes

At scale, schema changes become operations. Some migrations take locks that block a billion-row table for minutes, depending on engine and version. Use expand and contract: add a nullable column, write to both old and new, backfill in small batches, switch reads, and only then drop the old column. MySQL tools like gh-ost copy the table in the background and swap it in.

Online schema changes
  • adding a NOT NULL column with
  • a default can lock big tables
  • (engine/version dependent)
  • expand → migrate → contract:
  • 1 add nullable column
  • 2 dual-write, backfill in
  • small batches
  • 3 switch reads, then drop old
  • tools: gh-ost, pt-osc, pg_repack

Sizing

Sizing. Three thousand writes a second of one-kilobyte rows is three megabytes a second, about two hundred sixty gigabytes a day, and roughly triple that with indexes. One primary handles the write rate fine; the storage growth is the issue, so partition and archive old data. Thirty thousand reads a second at maybe ten thousand per replica means three replicas plus a spare. Shard when hot data passes several terabytes or writes saturate one box.

Sizing
writes: 3,000/s, rows ~1 KB
  → 3 MB/s, ~260 GB/day raw
  ×3 for indexes ≈ 0.8 TB/day

reads: 30,000/s; 1 replica
  ~10,000/s simple reads
  → 3 replicas + 1 spare

one primary OK for writes;
storage → partition + archive,
shard past ~5-10 TB hot data

Say it like this

In the interview, start concrete: Postgres with a synchronous standby, three async replicas for reads, a pooler in front, nightly backups plus WAL archiving. Then tell the story of ten times: partition the big table, split out a domain, and finally shard by user id with many logical shards. Mention that managed or distributed SQL, like Aurora, Spanner or CockroachDB, trades money or latency for less of this work.

In the interview
  • "Postgres, primary + sync
  • standby, 3 async replicas,
  • PgBouncer in front"
  • at 10x: partition orders by
  • month, split payments out,
  • then shard by user_id with
  • 1,024 logical shards
  • or managed: Aurora, Spanner,
  • CockroachDB, Vitess

Interview questions

What would you do first when a relational database becomes slow?

Find the most expensive queries by total time, fix them with better indexes or rewritten queries, add a connection pooler, and scale the machine up, since these are cheap and add no new failure modes before replicas or sharding.

Why use a connection pooler like PgBouncer?

Each database connection costs memory and scheduling overhead, and thousands of app connections degrade performance. A pooler multiplexes many client connections onto a small number of server connections, typically lending one per transaction.

How do you handle replication lag with read replicas?

Route reads that must reflect a user's own recent writes to the primary for a short window or until the replica has replayed past the write's log position, and send everything else to replicas.

Partitioning versus sharding?

Partitioning splits a table into pieces within one database server, which helps pruning and data retention. Sharding spreads rows across many database servers by a shard key, which scales writes and storage but adds routing and cross-shard complexity.

How do you choose a shard key?

Pick the column most queries filter on and that spreads load evenly, such as user_id or tenant_id, and co-locate related rows on the same shard so common joins and transactions stay on one shard.

How do you avoid painful resharding later?

Create many logical shards up front, for example 1,024, mapped onto a few physical servers, and rebalance by moving whole logical shards to new machines instead of re-hashing individual rows.

How does failover avoid losing data or causing split brain?

Use a synchronous standby so committed transactions exist on two machines, let a manager such as Patroni promote it, and fence the old primary, for example by revoking its lease or cutting it off, so it cannot keep accepting writes.

Why are replicas not a substitute for backups?

Replicas copy every change, including accidental deletes and bad migrations, within milliseconds. Base backups plus an archived write-ahead log allow point-in-time recovery to just before the mistake.

How do you change the schema of a huge table without downtime?

Use expand and contract: add the new column or table in a non-blocking way, dual-write and backfill in batches, switch reads, then remove the old structure, or use an online schema change tool that copies and swaps the table.