Reliability & Scale · Lesson 15

Scaling the database

Every other tier in your stack scales horizontally on day one — but the database is the one node that knows where everything lives. This lesson is about removing that bottleneck without trading away the reliability you need.

The library with one librarian

Imagine a library where a single librarian handles every request: shelving, retrieval, cataloguing, and reference questions — all on one pair of hands. Traffic is fine at first. But as the library grows, that librarian becomes the ceiling on every reader's experience. You can hire more librarians (replicas) for lookups while the head librarian stays in charge of updates. You can split the collection across wings (shards) so each librarian only knows their section. You can pre-build subject indexes (indexes) so nobody reads every shelf. And you can let people queue at a service desk (connection pool) instead of crowding the counter directly. Each of those moves has real costs — this lesson maps all of them.

Read replicas

The simplest lever: keep one primary that accepts writes, and run one or more read replicas that stream a copy of every change from the primary's write-ahead log. Your application routes all SELECT queries to the replicas and all INSERT / UPDATE / DELETE queries to the primary.

The cost is replication lag. The replica applies changes asynchronously, which means there is a window — typically a few milliseconds on a well-tuned setup, but up to seconds under write storms — where a replica returns data that the primary has already overwritten. A user who just submitted a form and immediately reloads the page can see the old data. This is not a bug you can fix in application code without re-routing that specific read to the primary. The section on read-after-write routing below covers the pattern.

Replicas help with read-heavy workloads — reporting queries, dashboards, full-text search — but they do nothing for write throughput. Once your primary can't absorb writes fast enough, you need sharding.

Sharding the data

A single machine has a ceiling: RAM, CPU, and disk IOPS are finite. Sharding horizontally partitions the data across multiple independent nodes so that each node owns a fraction of the rows. A shard is just a database node that holds a contiguous or algorithmically determined slice of the keyspace.

Choosing a shard key

The shard key is the field you use to decide which node owns a given row. Pick badly and you pay for it forever. Good shard keys are:

• High cardinality — enough distinct values to spread rows evenly.
• Stable — changing the shard key of an existing row requires moving it, which is expensive.
• Co-located with access patterns — most queries for a given user should hit one shard, not scatter across all of them (scatter-gather is slow and hard to paginate).

Hash sharding vs range sharding

Hash sharding puts the key through a hash function and maps the result to a shard (e.g. hash(user_id) % N). Distribution is even; range queries are expensive because adjacent IDs land on different shards. Range sharding assigns ordered ranges of keys to shards (e.g. users A–M on shard 1, N–Z on shard 2). Range queries are fast; hot ranges (think: all new users arriving in alphabetical order) create hot shards.

The hot-key / celebrity problem

Hash sharding distributes traffic evenly for uniform access patterns. Real workloads are never uniform. A social network where Beyoncé has 200 million followers will route every one of those followers' timeline reads to the single shard that owns Beyoncé's posts. No amount of horizontal scaling helps if every request still hits one shard. See the callout below for the mitigation patterns.

Why naïve resharding hurts

With modulo sharding (hash(key) % N), adding a fourth shard changes N from 3 to 4 and invalidates the placement of almost every key. If hash(key) % 3 = 2, there is no guarantee that hash(key) % 4 = 2 — in fact roughly 75% of keys land on a different shard after the change, forcing a massive data migration while the system must still serve traffic. Consistent hashing (covered in Lesson 08) solves this: nodes and keys are placed on a ring, and adding a node only moves the keys that sit between the new node and its predecessor — on average 1/N of the total keyspace.

Denormalization, indexing, and CQRS

Denormalization for read-heavy APIs

Normalization eliminates redundancy and makes writes clean, but a fully normalized schema often forces the database to join five tables to answer one API request. At high read volume those joins become the bottleneck. Denormalization strategically duplicates data in a pre-joined shape optimized for the most frequent read pattern. A posts table that redundantly stores author_name alongside author_id means the API can satisfy a timeline request with a single table scan, no join required. The cost: writes must maintain two copies. Acceptable when reads outnumber writes by 10:1 or more; dangerous when write consistency across copies is hard to enforce.

Indexing strategies

An index is a pre-sorted copy of a subset of your data that lets the database skip the full-table scan. Getting indexing wrong costs more than getting it right gains:

• B-tree indexes are the default and cover equality lookups and range queries (>, BETWEEN, ORDER BY). Almost every primary key and foreign key should have one.
• Composite indexes cover multi-column predicates. Column order matters — put the most selective column first, and the index can only be used left-to-right. An index on (user_id, created_at) accelerates WHERE user_id = 42 ORDER BY created_at DESC but not WHERE created_at > X alone.
• Partial indexes index only rows matching a condition (e.g. WHERE status = 'active'). Narrow, fast, cheap to maintain.
• Too many indexes slow writes — every INSERT or UPDATE must maintain each index. Audit unused indexes regularly (pg_stat_user_indexes in Postgres tracks actual usage).

CQRS: separating read and write models

Command Query Responsibility Segregation takes the denormalization idea to its logical conclusion: maintain a fully separate data model optimized for reads, updated asynchronously whenever a write (command) is processed. The write side uses a normalized schema tuned for consistency; the read side uses whatever pre-joined, pre-aggregated structure makes each query fast. An event — "post created" — flows from the write model to a background consumer that updates the read model. The tradeoff is explicit eventual consistency: the read model lags behind the write model by the consumer's processing latency. CQRS is overkill until you have genuinely incompatible read and write access patterns that no single schema can serve well.

Connection pooling

Opening a TCP connection to a database is not free: handshake, authentication, and session setup add 5–20 ms and consume a server-side slot. At high concurrency, thousands of threads each trying to open their own direct connection creates two problems: connection setup latency on the hot path, and the database process count ceiling (Postgres, for instance, spawns a process per connection and starts thrashing once you exceed a few hundred).

A connection pool maintains a fixed set of long-lived connections and lends them to application threads on demand. The application checks out a connection, executes a query, and returns the connection to the pool. If the pool is empty, the thread waits. The pool size should be tuned to the database's actual concurrency capacity, not to the application's thread count — a pool of 20 connections serving 500 application threads is normal; a pool of 500 connections to match thread count will overwhelm the database.

Common poolers: PgBouncer and pgpool-II for Postgres; HikariCP for JVM applications; the built-in pool in most ORM frameworks. Transaction-mode pooling (connections returned immediately after each transaction) gives the best multiplexing ratio but breaks session-level features like advisory locks and SET LOCAL.

Under the hood: how it actually works

Trace A — shard routing for user_id=42

Here is the exact arithmetic when your application looks up user_id=42 with three shards.

Trace B — replication lag timeline

A user updates their profile name. Here is what happens across the primary and replica at millisecond resolution.

How to debug & inspect it

Database performance problems rarely announce themselves. Most manifest as slow API response times with no obvious application code to blame. Start with the query planner.

The symptom–cause–fix table for the problems you will encounter most often:

By the numbers

Make it concrete. The platform is a social-media API that has grown to 200,000 write QPS on its primary database. Each shard is a managed Postgres node tuned to handle 25,000 write QPS before CPU and I/O become the bottleneck (Vitess operational guides use a similar per-shard ceiling). Reads run through replicas at a 4:1 read-to-write ratio.

How many shards do you need?

shards = ⌈total_write_QPS ÷ per_shard_QPS⌉
       = ⌈200,000 ÷ 25,000⌉
       = ⌈8.0⌉
       = 8 shards

Eight shards each at capacity. The moment write QPS creeps to 200,001 you need a ninth, so in practice you provision 10 and keep 80% utilisation as your ceiling — that headroom absorbs burst and gives you time to add shards without an emergency.

Replication-lag staleness window

The primary commits a write at t = 0. The WAL record travels over TCP and the replica applies it at roughly t = +40 ms under normal load (PostgreSQL streaming-replication docs show typical apply lag of 10–100 ms on a busy primary). Any read that arrives at the replica in the window (0, 40 ms) returns stale data.

Staleness probability: if inter-write interval for a given row is 500 ms, then P(stale read) ≈ 40 ms ÷ 500 ms = 8%. For a "post just published" read, the inter-write interval might be seconds, dropping staleness risk to under 1%.

Hot-key imbalance: the celebrity shard

Suppose one key — a celebrity account — attracts 15% of all traffic. With 8 shards and uniform distribution you'd expect each shard to receive 200,000 ÷ 8 = 25,000 QPS. But the hot key concentrates all 15% on shard 0:

Decision math: when to add shards and how many read replicas to provision

Add shards when: current_write_QPS > target_utilisation × per_shard_QPS × shard_count. At 80% target: add when current_write_QPS > 0.80 × 25,000 × 8 = 160,000 QPS. You crossed that threshold at 160,001 QPS — at that point you have time to add shards before hitting the ceiling at 200,000.

Read replicas per shard: reads run at 4× writes = 800,000 read QPS total across 8 shards = 100,000 read QPS per shard. A replica handles the same ~25,000 QPS as a primary, so each shard needs ⌈100,000 ÷ 25,000⌉ = 4 read replicas. Total fleet: 8 primaries + 32 replicas = 40 Postgres nodes. Caching even 50% of reads cuts that to 2 replicas per shard, halving the replica count.

Sources & further reading

• PostgreSQL documentation — Using EXPLAIN
• Vitess — MySQL horizontal sharding at scale
• AWS DynamoDB — Partitions and data distribution
• Designing Data-Intensive Applications — Martin Kleppmann
• pgpool-II — Connection pooling and load balancing for PostgreSQL