Replication and scaling

A single database server has a ceiling: one machine's cores, memory and disk. Long before that ceiling there is a different limit, which is that one machine is one failure. Replication answers the second problem and helps with the first; partitioning keeps a big table manageable; sharding splits the database itself, and it is the last resort because it takes away the things a relational database is for. This lesson is the mechanisms, the consistency problem replication creates, and the order in which to reach for each.

#How replication works

A primary accepts writes and streams its change log to one or more replicas, which replay it. In PostgreSQL that is the write-ahead log, shipped as streaming replication, producing a byte-for-byte physical copy; logical replication (PostgreSQL 10+) publishes row changes for chosen tables instead, which is what you use to replicate across versions or into a different schema. MySQL ships its binary log, row-based by default since 5.7, with GTIDs to identify positions.

Whether the primary waits for the replica decides what a commit means:

Asynchronous is the default because the round trip is paid on every commit, and it means that a failover promotes a replica that may be a little behind. For a system where losing thirty seconds of orders is unacceptable, one synchronous replica is the price, and it must be in a different failure domain from the primary or it protects nothing.

#Replication lag, and read-your-writes

A replica is behind the primary by however long the log takes to arrive and apply: milliseconds normally, seconds under load, minutes during a large migration or a long-running transaction on the replica that blocks replay. Any read sent to a replica may be stale by that much.

The bug this produces has a name, read-your-writes: a user updates their profile, the write goes to the primary, the next page load reads from a replica that has not caught up, and the user sees their old name and hits save again. Three fixes:

1. Route by recency. After a write, send that user's reads to the primary for a few seconds (a cookie or a session flag with a timestamp). Simple, and what most applications do.
2. Wait for the position. The write returns the log position (pg_current_wal_lsn(), or the GTID); the read on the replica waits until pg_last_wal_replay_lsn() has passed it (MySQL: WAIT_FOR_EXECUTED_GTID_SET). Exact, and adds latency equal to the lag.
3. Do not read stale data where it matters. Checkout, balances and anything a user just changed read from the primary; catalogue, search and reports read from replicas. Most systems end up here, with the routing decision per query.

The framework-level version is Spring's AbstractRoutingDataSource choosing a datasource by whether the current transaction is read-only (@Transactional(readOnly = true)), with the recency rule layered on top. Whatever the mechanism, monitor lag (pg_stat_replication, SHOW REPLICA STATUS) and alert on it, because every fix above assumes it is small.

#Read replicas and failover

Replicas earn their keep twice. They take the read load, which is most of the load in most systems, off the primary: reporting queries, search, the read side of an API. And one of them becomes the new primary when the old one dies. Failover is not automatic on either engine out of the box; Patroni (PostgreSQL) or Orchestrator and MySQL's Group Replication manage it, and managed services (RDS Multi-AZ, Cloud SQL) do it for you at the cost of control. What the application needs is a connection string that survives the switch, a DNS name or a proxy, and a pool that recycles connections on error, which HikariCP does.

A replica is not a backup. It replays a DROP TABLE as faithfully as anything else. Backups are base backups plus archived log for point-in-time recovery, tested by actually restoring one.

#Partitioning: one table in pieces

A table with a billion rows of events is slow to index, slow to vacuum and impossible to delete from in bulk. Partitioning splits it into child tables by a key, and the engine routes queries to the children that can match:

CREATE TABLE events (
  id BIGINT GENERATED ALWAYS AS IDENTITY, occurred_at TIMESTAMPTZ NOT NULL, payload JSONB,
  PRIMARY KEY (id, occurred_at)
) PARTITION BY RANGE (occurred_at);
CREATE TABLE events_2026_09 PARTITION OF events FOR VALUES FROM ('2026-09-01') TO ('2026-10-01');
-- retention: a month of data goes in one statement, no DELETE, no bloat
DROP TABLE events_2025_09;

A query with WHERE occurred_at >= '2026-09-01' touches one partition (partition pruning), each partition has its own smaller indexes, and retention is DROP TABLE. The constraints: the partition key must be part of every unique constraint, so the primary key above includes occurred_at; a query without the key in its predicate scans every partition; and too many partitions (thousands) slow planning. Range by time for logs and events, list by tenant or region when data is cleanly separable, hash to spread a hot table evenly. MySQL supports the same three; both keep the table logically one, so the application does not change.

#Sharding: the last resort

Sharding splits the database into independent databases, each holding a slice of the rows by a shard key: customers 1–1M on shard A, 1M–2M on shard B. It is how a relational database scales writes past one machine, and it costs exactly what it looks like: there are no joins across shards, no transactions across shards, no unique constraints across shards, and every query must know its shard key or fan out to all of them. Resharding, when the key distribution or the number of shards changes, is a migration of live data. Vitess (MySQL) and Citus (PostgreSQL) make it operable, not free.

Before sharding, in order, because each is cheaper than the next and most systems never get past the third:

1. The query fixes from the Optimisation lesson, and the indexes from the Indexes lesson.
2. Caching the reads that dominate.
3. Read replicas for the rest of the read load.
4. Partitioning the tables that are large.
5. A bigger machine. It is boring and it works; one server today has 128 cores and terabytes of memory.
6. Splitting by service, so that orders and catalogue are different databases, before splitting one table by key.
7. Sharding, with the key chosen so that the queries you actually run stay within one shard, which usually means the tenant or the customer.

#The connection multiplication problem

Scaling out the application tier multiplies connections: twenty instances with a pool of ten each is two hundred connections, and the Connection pooling lesson explains why the database wants twenty. PgBouncer or ProxySQL in front of the primary, and one in front of each replica, is the piece that makes horizontal application scaling and a healthy database coexist. Put it in before the instance count grows, not after the incident.

#Under the hood: the log, the sender, and where lag comes from

PostgreSQL's replication is the write-ahead log leaving the building. Every change is already a WAL record for crash recovery; a WAL sender process on the primary streams those records over a replication connection to a WAL receiver on the replica, which writes them to its own log, and a startup process there replays them into data pages, exactly as crash recovery would. The replica is therefore a byte-identical copy that is always in "recovery", which is why it accepts no writes, why it must run the same major version, and why a DROP TABLE arrives as faithfully as an insert. Lag has three components, each visible in pg_stat_replication: write_lag (the record reached the replica's disk), flush_lag (it was fsynced) and replay_lag (it was applied and is now visible to queries). Replay is single-threaded and can fall behind a primary with many writers, and it stops while a replica query holds a snapshot that the incoming WAL would invalidate, which is the conflict the PRODUCTION callout below describes.

MySQL ships the binary log instead: a logical record of row changes (with binlog_format = ROW), not page images. The replica's I/O thread copies it into a relay log; SQL applier threads (several, with replica_parallel_workers, since ordering is tracked per transaction with GTIDs and write-sets) replay it as ordinary statements against the replica's own storage. Because it is logical, a MySQL replica can run a newer version, have different indexes, or skip tables, and it also means replay does real work: a statement that updated ten million rows on the primary updates ten million rows again on the replica, on fewer threads, and that is the lag spike after a bulk update. PostgreSQL's logical replication works the same way, decoding WAL into row changes through a publication/subscription, and shares the same trade.

Failover is a promotion: the replica exits recovery, starts a new timeline, and begins accepting writes. What makes it safe is not the promotion but everything around it: something must decide the old primary is dead and not merely slow (Patroni uses a distributed lock in etcd or Consul with a lease; a primary that loses the lease demotes itself), the old primary must not accept writes after the decision (fencing, or it becomes a split brain with two divergent histories), and clients must find the new primary, through a DNS change, a virtual IP, a proxy such as HAProxy or PgBouncer pointed by the orchestrator, or the JDBC driver's targetServerType=primary across a host list. Asynchronous replicas that were behind the promoted one must be re-cloned or rewound (pg_rewind) because their history now diverges from the new timeline.

Partition pruning is a planner feature: with the partition key in the WHERE, the planner excludes child tables whose bounds cannot match at plan time, and since PostgreSQL 11 also at execution time for parameters and joins. Every partition has its own indexes and statistics, and the parent has none of its own, which is why a query without the key touches every partition's index in turn and why thousands of partitions make planning slow: the planner must open each. Global unique constraints are impossible for the same reason; uniqueness is per partition, so the key must be in every unique index.

#Walkthrough: the replica that was six hours behind and nobody knew

A reporting replica served the analytics dashboard and, one Monday, the finance team reported that Friday's numbers had changed overnight.

SELECT date_trunc('day', paid_at), SUM(amount) FROM payments
WHERE paid_at >= now() - interval '7 days' GROUP BY 1;

1. On Friday evening a data-fix job on the primary updated 30 million payments rows in one transaction. The primary finished in eleven minutes. The replica's single-threaded replay took most of the night to apply the same change, and during that time replay_lag climbed to six hours.
2. Nobody was alerted, because the only replication monitor checked that pg_stat_replication had a row with state = 'streaming', and it did: the WAL was arriving fine. Lag was never graphed.
3. Meanwhile, a long analytics query on the replica held a snapshot; with hot_standby_feedback = off and max_standby_streaming_delay = 30s, the replica cancelled it after thirty seconds of conflict (ERROR: canceling statement due to conflict with recovery), which the dashboard retried, which conflicted again, for hours. The dashboard showed stale data and intermittent errors, and the on-call engineer looked at the dashboard's service, not the database.
4. now() on a replica is the replica's clock; the "last seven days" window was correct, and the data in it was six hours stale. Friday's figures shifted when replay finished. Nothing was wrong; nothing was observable.
5. Fixes: alert on replay_lag (and pg_last_xact_replay_timestamp() age) above a minute; set max_standby_streaming_delay = -1 on the reporting replica, so it delays replay rather than cancel queries, with the understanding that lag on it is by design; keep the user-facing replica at the default so it stays fresh; run bulk updates in batches so replay can interleave. And the dashboard now shows "data as of ", which is the honest fix.

A replica has two obligations, freshness and query stability, and one knob that trades between them. A replica used for both is the one that fails at both.

#Try it yourself

#Misconceptions

• "A replica is a copy of the data." It is a copy of the log, replayed; on PostgreSQL that makes it a physical twin that must match versions, on MySQL a logical twin that redoes the work.
• "Streaming means in sync." state = streaming says the log is arriving; replay_lag says whether it has been applied. Alert on the second.
• "Synchronous replication means no data loss, full stop." It means no loss if the acknowledging replica survives; with one candidate its maintenance is your outage.
• "Failover is just promoting a replica." Deciding the primary is dead, fencing it, repointing clients and rewinding stale replicas are the hard parts; promotion is one command.
• "Partitioning makes every query faster." Only queries that carry the partition key are pruned; the rest touch every partition and pay for the planning of each.

#Going deeper

• PostgreSQL manual, "High Availability, Load Balancing, and Replication", especially "Streaming Replication", "Synchronous Replication" and "Hot Standby" (conflicts and hot_standby_feedback).
• MySQL reference, "Replication" (binary log formats, GTIDs, multithreaded appliers) and "Group Replication".
• Patroni documentation, "Replication modes" and the failover walkthrough; pg_rewind reference.
• PostgreSQL manual, "Table Partitioning", the "Declarative Partitioning Best Practices" section on partition counts and pruning.
• Martin Kleppmann, Designing Data-Intensive Applications, chapters 5 (replication) and 6 (partitioning).