Storage and data systems

Compute is easy to scale because it forgets. Storage is hard because it must remember, and every trick in this article is a way of remembering more, faster, on more machines, without losing anything. The trade is always the same four things: read latency, write volume, schema flexibility and how strict the guarantees are.

The map

Read this first when short on time. Every branch is a section below.

Choosing a storage model

A storage model is a bet on how the data will be read and written. Each one is fast at the shape it was built for and slow or awkward at the rest, so the first question is never "which database is best" but "what are the access patterns".

flowchart TD
  A{Transactions and joins?} -- yes --> R[Relational]
  A -- no --> B{Shape of the data?}
  B -- large binary files --> O[Object storage]
  B -- relationships are the point --> G[Graph]
  B -- text to search --> S[Search index]
  B -- records --> C{Access pattern?}
  C -- lookup by key, tiny latency --> K[Key-value]
  C -- nested records, evolving fields --> D[Document]
  C -- huge write rate, time-ordered --> W[Wide-column]

Two things the table hides. First, a relational database is the right default: it is the most flexible, the tooling is mature, and a single Postgres carries most products a long way. Second, real systems are polyglot: orders in Postgres, sessions in Redis, product search in Elasticsearch, images in S3, and an event log in Kafka, with the relational store as the source of truth and the rest derived from it.

Sharding: splitting one table across machines

Sharding partitions a dataset horizontally: each shard holds a subset of the rows and runs on its own server, so both storage and write throughput grow with the number of shards. The one decision that matters is the shard key, because it decides which queries stay on one machine.

flowchart TD
  Q["query: orders for user 8213"] --> R{"router: hash(user_id) mod 4"}
  R -- 0 --> S0[(shard 0: users 0..)]
  R -- 1 --> S1[(shard 1)]
  R -- 2 --> S2[(shard 2)]
  R -- 3 --> S3[(shard 3)]
  Q2["query: all orders over $100 today"] -.-> SG[scatter to all shards, gather results]
  SG -.-> S0 & S1 & S2 & S3

Choosing the shard key

The key should appear in almost every query (user id for a consumer app, tenant id for SaaS), spread load evenly, and rarely change. Hotspots come from keys that break the second rule: one celebrity account, one tenant that is a hundred times bigger than the rest, or a timestamp key where every write lands on "now". Fixes include adding a random suffix to hot keys, splitting hot shards, or caching the hot entity.

What sharding costs

• Cross-shard queries become scatter-gather: ask every shard, merge the results. Joins across shards are effectively gone; denormalize or keep a copy of what you need on the same shard.
• Cross-shard transactions need two-phase commit or sagas. Design so that a transaction's rows share a shard key.
• Resharding is the operation everyone fears: moving data while serving traffic. Plan for it on day one with more logical shards than physical servers (Instagram started with thousands of logical shards across a few machines) or a system that reshards online.
• Unique ids can no longer come from one auto-increment; use UUIDs, Snowflake-style ids, or per-shard ranges.

Replication: keeping copies in sync

Replication keeps the same data on several nodes so that reads can be spread out and a failed node does not lose anything. The design question is who accepts writes and how long a copy may lag behind.

sequenceDiagram
  autonumber
  participant C as Client
  participant L as Leader
  participant F1 as Follower (sync)
  participant F2 as Follower (async)
  C->>L: UPDATE balance
  L->>L: append to WAL, apply
  L->>F1: ship WAL record
  F1-->>L: acknowledged
  L-->>C: committed
  L->>F2: ship WAL record (later)
  Note over F2: lag: a read here may return the old balance

Three shapes

• Single leader. One node takes all writes and streams its log to followers, which serve reads. Simple and the default in Postgres, MySQL and most managed databases.
• Multi-leader. Several nodes accept writes, usually one per region, and exchange changes. Faster local writes, but concurrent writes to the same row must be reconciled: last-writer-wins, merge rules, or conflict-free data types.
• Leaderless. Any node accepts writes; a write is successful when W nodes confirm and a read asks R nodes, with W + R greater than N to guarantee overlap. This is the Dynamo model used by Cassandra and Riak; the reliability article covers quorums in depth.

Synchronous or asynchronous

A synchronous follower must acknowledge before the leader confirms the write: no data is lost on failover, at the cost of write latency and the risk that a slow follower stalls everything. An asynchronous follower is told about the write later: fast, but if the leader dies before shipping, those writes are gone. Most deployments use one synchronous follower (or a quorum of them) plus asynchronous read replicas.

Living with lag

An asynchronous replica is always slightly behind, from milliseconds to, under load, minutes. Three anomalies follow and each has a standard fix:

• Read your own writes. A user updates their profile, refreshes, and sees the old one. Fix: route that user's reads to the leader for a short window after a write, or track the write's log position and require a replica to have reached it.
• Monotonic reads. Two refreshes hit two replicas with different lag and time appears to go backwards. Fix: pin a user to one replica.
• Consistent prefix. An answer arrives before the question it replies to. Fix: write causally related data to the same partition.

Failover

When the leader dies, a follower is promoted and clients are repointed, either by the database's own election, an orchestrator (Patroni, Orchestrator) or the managed service. The dangers are losing unshipped asynchronous writes, and split brain: the old leader comes back and keeps accepting writes. Fencing (revoking the old leader's ability to write, for example with a lease or a fencing token) is what makes failover safe.

Replication lag

How far a follower is behind the leader, in time or in log position. The number that turns into "I saved it but it is not there".

Read replica

An asynchronous follower used only for reads. Cheap read scaling as long as the application tolerates lag.

Fencing

Making sure a demoted leader cannot write. Without it, failover can produce two diverging histories.

Indexing: B-trees and LSM-trees

An index is a second data structure that trades write cost and space for fast lookups on a column. The two families behind every database differ in what they optimise: B-trees for reads, LSM-trees for writes.

flowchart TB
  subgraph BT["B-tree: update in place"]
    direction TB
    W1[write] --> P[find the page, modify it, WAL it]
    P --> Rd[read: walk root to leaf, 3 or 4 page reads]
  end
  subgraph LSM["LSM-tree: append, then merge"]
    direction TB
    W2[write] --> WAL[WAL] --> MT[memtable, sorted in memory]
    MT -- full --> L0[flush: SSTable level 0]
    L0 -- compaction --> L1[level 1, larger sorted files]
    L1 -- compaction --> L2[level 2 ...]
    Rd2[read] --> MT
    Rd2 -. bloom filter says maybe .-> L0 & L1 & L2
  end
  BT ~~~ LSM

Using indexes well

• Every index is a write. A table with six indexes does seven writes per insert. Index what queries filter and sort on, and nothing else.
• Column order matters in a composite index. (user_id, created_at) serves "this user's recent orders"; it does not serve "all orders on a date".
• Covering indexes include the columns the query returns, so the table itself is never touched.
• Partial indexes cover only rows that match a condition (unshipped orders), which keeps them small and hot.
• Secondary indexes in distributed stores are either local to a partition (cheap, only useful with the partition key) or global (a separate sharded index, eventually consistent, costs extra writes). DynamoDB's LSI and GSI are the canonical pair.

-- Serves: WHERE user_id = ? ORDER BY created_at DESC LIMIT 20
CREATE INDEX orders_by_user_recent ON orders (user_id, created_at DESC);

-- Partial index: only the rows a worker actually polls for
CREATE INDEX orders_unshipped ON orders (created_at) WHERE shipped_at IS NULL;

-- Always confirm the plan uses it before shipping
EXPLAIN ANALYZE SELECT * FROM orders WHERE user_id = 8213 ORDER BY created_at DESC LIMIT 20;

SSTable

Sorted string table. An immutable file of key-value pairs in key order, the unit an LSM-tree flushes and compacts.

Bloom filter

A tiny probabilistic set that answers "definitely not here" or "maybe here", letting a read skip files that cannot contain the key.

Write amplification

Bytes actually written to disk divided by bytes the application wrote. Compaction rewrites data several times, so LSM engines can reach 10× or more.

Caching: keeping hot data close

A cache keeps copies of frequently read data in a faster, smaller store so that most reads never reach the database. The patterns differ in who fills the cache and when writes go through it, and every one of them faces the same enemy: staleness.

flowchart TB
  subgraph CA["Cache-aside"]
    direction LR
    A1[app] -- 1 GET key --> C1[(cache)]
    C1 -- 2 miss --> A1
    A1 -- 3 read --> D1[(database)]
    A1 -- 4 SET key, TTL --> C1
  end
  subgraph WT["Write-through"]
    direction LR
    A2[app] -- write --> C2[(cache)] -- write synchronously --> D2[(database)]
  end
  subgraph WB["Write-back"]
    direction LR
    A3[app] -- write --> C3[(cache)] -. flush in batches .-> D3[(database)]
  end
  CA ~~~ WT ~~~ WB

Expiry, eviction and invalidation

Every entry needs a way to leave. A TTL bounds staleness. Eviction (LRU or LFU when memory is full) bounds size. Invalidation on write removes an entry the moment it is wrong, and it is the hard part: the write path must know every key the change affects, and a delete that races with a concurrent read can leave the old value back in the cache. Deleting rather than updating on write, and keeping TTLs even when you invalidate, limits the damage.

Stampedes

When a popular key expires, every request that misses goes to the database at once. Three defences: a lock or single-flight so one request recomputes while the others wait or serve stale; early refresh, where an entry is refreshed in the background shortly before it expires; and jittered TTLs, so that a thousand keys written together do not expire together.

suspend fun product(id: Long): Product {
    cache.get("product:$id")?.let { return it }               // 1. hit
    return loader.singleFlight("product:$id") {                // 2. one loader per key
        val p = db.findProduct(id)                             // 3. read the source
        val ttl = 10.minutes + Random.nextLong(0, 60).seconds  // 4. jitter the expiry
        cache.set("product:$id", p, ttl)
        p
    }
}

suspend fun updateProduct(p: Product) {
    db.save(p)
    cache.delete("product:${p.id}")   // invalidate by deleting, never by writing the new value
}

Consistent hashing: adding a node without moving everything

Consistent hashing places both nodes and keys on the same circular hash space so that when a node is added or removed, only the keys next to it change owner. Plain hash(key) mod N remaps almost every key when N changes; consistent hashing remaps about 1/N of them.

The plain version has two flaws: with few nodes the arcs are uneven, and when a node leaves, its whole arc lands on one neighbour. Virtual nodes fix both by giving each physical server many positions. Replication falls out naturally: store each key on the next N distinct nodes clockwise, and a node's data is automatically spread across several successors.

Denormalization: copies instead of joins

Denormalization stores the same fact in more than one place so that reads need no join. Normalized data is easy to keep correct because each fact lives once; denormalized data is fast to read because each query finds everything it needs in one row, document or partition.

flowchart TB
  subgraph N["Normalized"]
    direction LR
    O1[orders: id, user_id, total] --- U1[users: id, name, city]
    O1 --- I1[order_items: order_id, product_id, qty] --- P1[products: id, name, price]
    Q1["order page = 4 tables, 3 joins"]
  end
  subgraph DN["Denormalized"]
    direction LR
    O2["order document: id, user{name, city}, items[{product name, price, qty}], total"]
    Q2["order page = 1 read"]
    T2["timeline: precomputed list per user"]
  end
  N -- events or CDC keep the copies fresh --> DN

Forms it takes

• Embedding. A document holds the customer's name and the product names as they were at order time. Also correct historically: the order should not change when the product is renamed.
• Precomputed aggregates. A like_count column instead of COUNT(*) on every read.
• Materialized views. A stored query result refreshed on a schedule or incrementally.
• Read models. A separate store shaped for one screen (a timeline, a search index, a dashboard) fed by events from the source of truth. This is the CQRS idea: one model for writes, others for reads.
• Single-table design. In DynamoDB, several entity types share one table with keys chosen so that each screen is one query.

Keeping copies honest

Every copy is a promise to update it. Dual writes from application code (write the order, then update the timeline) fail halfway and drift. Better: let the source of truth emit the change (an outbox table, change data capture) and have consumers maintain the copies, accepting that they lag by a moment. And keep a way to rebuild every derived store from the source, because one day you will need it.

Write-ahead logging: durability before speed

A write-ahead log is an append-only file that a database writes every change to, and forces to disk, before it touches its real data structures. If the process dies at any point, replaying the log rebuilds what was in memory. It is how a database can be both durable and fast: the expensive random writes happen later, in batches.

sequenceDiagram
  autonumber
  participant C as Client
  participant D as Database
  participant W as WAL on disk
  participant P as Data pages
  C->>D: COMMIT (update row)
  D->>W: append record, fsync
  W-->>D: durable
  D-->>C: committed
  D->>D: modify the page in memory
  D-->>P: checkpoint later: write dirty pages, trim the log
  Note over D,P: crash before the checkpoint? replay the WAL from the last checkpoint

• Sequential wins. Appending to one file is the fastest thing a disk can do; updating pages scattered across a table is the slowest. The log takes the fast path on the critical path and defers the slow one.
• Group commit. Many transactions committing at once share a single fsync, which is batching applied to durability.
• Checkpoints. Periodically the dirty pages are written out and a checkpoint record marks how far the log can be truncated. Recovery replays only what came after.
• The same log powers replication. Followers are just processes replaying the leader's WAL, and change data capture reads it too.

fsync

The system call that forces written data from the OS cache to the physical disk. Durability starts only when it returns.

Checkpoint

A point at which all changes before it are on disk in the data files, so the log before it can be discarded.

Transactions and isolation levels

A transaction groups several operations so that they succeed or fail as one and cannot see each other half done. ACID names the four promises, and the isolation level is the dial that trades how much concurrent transactions can see of each other against how much they block.

Atomic

All of the transaction's writes happen, or none do. Implemented with the log: uncommitted changes are rolled back on recovery.

Consistent

Constraints (unique keys, foreign keys, checks) hold before and after. This one is a promise about your rules, kept by the database.

Isolated

Concurrent transactions behave as if they ran one at a time, to the degree the isolation level says.

Durable

Once committed, the change survives crashes. The write-ahead log is what makes this true.

Modern engines implement snapshot levels with MVCC: each write creates a new version of the row and readers see the version that was current when they started, so readers never block writers. The anomaly worth remembering is write skew: two transactions each read a condition (two doctors on call), each decide based on it (I can go off duty), and both commit, leaving nobody on call. Only serializable isolation, or an explicit lock, prevents it.

Recap