4. Data Storage and Database Design

Storage is where most systems ultimately succeed or fail: compute is easy to add and easy to lose, but data must survive, stay correct, and remain queryable as it grows by orders of magnitude. This chapter works up from how bytes are physically organized, through relational and non-relational stores, to the replication, partitioning, and lifecycle machinery that keeps data available at scale.

<a id="41-storage-fundamentals"></a>

4.1 Storage Fundamentals

Every database is a program built on top of files and disks; understanding that layer explains why databases behave the way they do under load.

Block, File, and Object Storage

Before choosing a database, you choose how bytes are addressed on the infrastructure beneath it. Three abstractions dominate, and they differ in what the smallest addressable unit is — which in turn determines latency, mutability, scalability, and price.

Block storage exposes a raw device as a sequence of fixed-size blocks with no notion of files. It is what a disk actually is, and it is what a database wants: the engine imposes its own structure and can rewrite any block in place. File storage adds a hierarchy of directories and files with shared access semantics — the familiar filesystem, usable by many machines at once via NFS or SMB. Object storage discards the hierarchy entirely: you PUT and GET whole immutable blobs by key over HTTP, and the system scales essentially without limit.

The analogy: block storage is a blank warehouse floor where you install your own shelving. File storage is a filing cabinet with labelled drawers. Object storage is a valet service — hand over a box, get a ticket, and ask for it back later; you never reach in and change part of it.

 BLOCK                    FILE                     OBJECT

 [ block 0 ][ block 1 ]   /reports/2026/q1.pdf     PUT /bucket/reports/q1.pdf
 [ block 2 ][ block 3 ]   /reports/2026/q2.pdf     GET /bucket/reports/q1.pdf
 addressed by offset      addressed by path        addressed by key
 mutable in place         mutable in place         immutable (replace whole object)
 one writer (usually)     many readers/writers     massively concurrent
 ~0.1-1 ms                ~1-10 ms                 ~20-200 ms
 EBS, SAN, local NVMe     NFS, EFS, SMB            S3, GCS, Azure Blob

# The access pattern each abstraction is built for.

# BLOCK: the database seeks to an offset and rewrites 8 KB in place.
with open("/dev/nvme0n1", "r+b") as dev:
    dev.seek(page_no * 8192)          # random access by offset
    dev.write(page_bytes)             # in-place mutation of one page

# FILE: shared, hierarchical, POSIX semantics (locks, permissions, append)
with open("/mnt/shared/uploads/report.csv", "a") as f:
    f.write(row)                      # many machines can mount the same tree

# OBJECT: whole-blob replace over HTTP; there is no "write 8 bytes at offset 4096"
s3.put_object(Bucket="media", Key="videos/42/master.mp4", Body=blob)
# Changing one byte means re-uploading the object. That constraint is exactly
# what lets the service scale to exabytes with 11 nines of durability.

Property	Block	File	Object
Unit	Fixed-size block	File in a directory	Whole object with a key
Mutation	In place, byte-level	In place, byte-level	Replace entire object
Latency	Lowest	Low	Highest (HTTP round trip)
Scalability	One volume, one host	Limited by the server	Effectively unlimited
Cost per GB	Highest	High	Lowest (with cold tiers)
Metadata	None	Path, permissions	Rich, user-defined
Typical use	Database volumes	Shared app files, home directories	Media, backups, logs, data lakes

The design rule that follows: large binary objects belong in object storage, with only a pointer in the database. Storing a 5 MB image as a database blob inflates backups, evicts useful pages from the buffer pool, and makes replication expensive — for a file that a CDN could have served directly from object storage.

Key Takeaways

• The three abstractions differ in addressable unit: block, file, and whole object.
• Block storage is what databases want — low latency and in-place mutation.
• Object storage trades mutability and latency for unlimited scale, durability, and low cost.
• Store blobs in object storage and keep only the key in your database.

🧪 Practice

1. Choose a storage type for each: a PostgreSQL data volume, user-uploaded profile photos, a shared build cache across CI runners, seven years of audit logs.
2. Explain why appending one line to a 1 GB object in S3 is expensive and what pattern avoids it.
3. Interview: A team stores 2 MB PDFs as BYTEA columns in Postgres and backups now take six hours. What do you change and how do you migrate? (Hint: think about what the database is uniquely good at, and what it is merely capable of.)

Storage Engines Overview

A database has two largely separable halves: the part that understands queries (parser, planner, executor) and the part that actually puts bytes on disk and finds them again. That second half is the storage engine, and it determines almost every performance characteristic you will observe — write throughput, read latency, space usage, and compaction behaviour.

The engine's job is harder than "write it to a file" because of three constraints. Disks favour sequential access (Chapter 2.1). Memory is far faster than disk but volatile. And a crash can happen between any two instructions, so committed data must still be there afterwards.

Every engine therefore has the same anatomy:

 ANATOMY OF A STORAGE ENGINE

   query executor
        |
        v
   +----------------------+   buffer pool / page cache: hot pages in RAM,
   |   in-memory layer    |   so most reads never touch the disk
   +----------------------+
        |            \
        | flush        \ every write first appends here, synchronously:
        v               v
   +-----------+     +--------------------+
   | data files|     | write-ahead log    |  sequential, crash recovery
   +-----------+     +--------------------+
        |
        v
   background work: compaction, vacuum, checkpointing, index maintenance

The fundamental split between engine families is what a write costs. Update-in-place engines (B-tree family) find the right page and modify it, which makes reads cheap and predictable but writes random. Log-structured engines (LSM family) only ever append, which makes writes sequential and fast but pushes work into background compaction and makes reads consult several places.

# The same logical update, seen from each engine family.

# UPDATE-IN-PLACE (B-tree): locate, modify, write back
def update_in_place(key, value, tree, wal):
    wal.append(("SET", key, value))       # durability first: sequential append
    page = tree.find_leaf_page(key)       # ~3-4 random reads to walk the tree
    page.set(key, value)                  # mutate the page in memory
    buffer_pool.mark_dirty(page)          # flushed later at a checkpoint
    # Reads afterwards: one tree walk. Simple and predictable.

# LOG-STRUCTURED (LSM): never modify, always append
def append_only(key, value, memtable, wal):
    wal.append(("SET", key, value))       # same durability step
    memtable[key] = value                 # sorted in-memory structure
    if memtable.size > THRESHOLD:
        flush_as_new_sorted_file(memtable)  # one big sequential write
        memtable.clear()
    # Reads afterwards: check memtable, then each sorted file, newest first.
    # Old versions of the key still exist on disk until compaction removes them.

Engine	Family	Used by
InnoDB	B-tree	MySQL (default)
PostgreSQL heap + btree	B-tree (with MVCC heap)	PostgreSQL
WiredTiger	B-tree/LSM	MongoDB (default)
RocksDB / LevelDB	LSM	Many embedded and custom systems
Cassandra SSTables	LSM	Cassandra, ScyllaDB

Two properties worth naming now because they recur throughout the chapter. Write amplification is the ratio of bytes actually written to disk versus bytes logically written by the application — compaction and page rewrites both inflate it. Read amplification is how many disk reads one logical lookup costs. Every engine trades one against the other, plus space amplification (how much disk the same logical data occupies).

Key Takeaways

• The storage engine, not the query language, determines performance characteristics.
• All engines share an anatomy: in-memory buffer, write-ahead log, data files, and background maintenance.
• The core split is update-in-place (B-tree) versus append-only (log-structured).
• Engines trade write, read, and space amplification against each other; there is no free configuration.

🧪 Practice

1. For a workload of 90% reads on recent data, sketch which parts of the engine anatomy carry the load.
2. Explain why every engine writes to a log before modifying data files.
3. Interview: Two databases expose identical SQL but behave very differently under a write-heavy load. What explains it? (Hint: the difference is below the query layer, and it shows up in what a single write physically does.)

B-Trees vs. LSM Trees

This is the most consequential storage decision in the chapter, and it is essentially a question about your write pattern. Both structures keep data sorted so range queries work; they differ entirely in how they absorb changes.

A B-tree (specifically B+tree) is a balanced tree of fixed-size pages. All records live in leaf pages linked together for range scans, and internal pages hold only keys for navigation. Because the fan-out is huge — a 4 KB page holds hundreds of keys — even a billion rows sit about four levels deep, so any lookup is roughly four page reads, and typically fewer because upper levels stay cached. Updates modify the target page in place.

An LSM tree (log-structured merge tree) never modifies anything. Writes land in a sorted in-memory structure (the memtable); when it fills, it is flushed as an immutable sorted file (an SSTable). Reads check the memtable, then each SSTable from newest to oldest. Background compaction merges SSTables, discarding superseded values and tombstones.

 B+TREE                                LSM TREE

        [ 50 | 100 ]                    writes -> [ memtable (sorted, RAM) ]
       /      |      \                                     | flush
  [10,30] [60,80] [120,150]             L0: [SST][SST][SST]  (may overlap)
     |       |        |                       | compaction
   leaves linked --> for range scans   L1: [ SST ][ SST ]     (non-overlapping)
                                             | compaction
  read : ~4 page reads, predictable    L2: [   bigger SSTs   ]
  write: find page, modify IN PLACE
         (random I/O)                  read : check memtable + N levels
                                       write: append only (sequential I/O)

# Why LSM writes are fast and LSM reads need help.
def lsm_get(key, memtable, sstables, bloom_filters):
    if key in memtable:
        return memtable[key]                    # newest data, no disk I/O
    for sst in sstables:                        # newest to oldest
        # A Bloom filter answers "definitely not present" in O(1) from memory,
        # skipping the disk read for most files. Without it, a lookup for a
        # missing key would touch EVERY SSTable.
        if not bloom_filters[sst].might_contain(key):
            continue
        value = sst.lookup(key)                 # binary search inside one file
        if value is TOMBSTONE:
            return None                         # deleted; do not look further
        if value is not None:
            return value                        # first hit wins: it is the newest
    return None

Dimension	B-tree	LSM tree
Write path	Random in-place page writes	Sequential appends
Write throughput	Lower	Much higher
Write amplification	Moderate (page + WAL)	Higher over time (repeated compaction)
Read (point lookup)	Predictable, ~4 reads	Variable: memtable + several levels
Read amplification	Low	Higher; mitigated by Bloom filters
Space amplification	Fragmentation, partly empty pages	Old versions until compaction
Range scans	Excellent (linked leaves)	Good (merge across sorted files)
Latency predictability	High	Compaction causes periodic spikes
Compression	Page-level, moderate	Better (large immutable sorted blocks)
Best for	Read-heavy, transactional, mixed	Write-heavy, time-series, logs

A crucial operational note: deletes in an LSM are writes. Removing a key appends a tombstone, so a delete-heavy workload grows the dataset until compaction runs — and a range scan across many tombstones can be dramatically slower than the row count suggests.

The rule of thumb: B-tree unless writes dominate. Transactional systems, mixed workloads, and anything needing predictable p99 latency want a B-tree. Time series, event logs, metrics, and ingest-heavy pipelines want an LSM.

Key Takeaways

• B-trees update pages in place: predictable reads, random writes.
• LSM trees only append and compact in the background: fast sequential writes, variable reads.
• Bloom filters are what make LSM point lookups viable by skipping most files.
• LSM deletes are tombstone writes, and compaction causes periodic latency spikes.

🧪 Practice

1. For each workload, pick an engine and justify it: banking ledger, IoT sensor ingest, product catalogue, application logs.
2. Explain why an LSM range scan over a heavily deleted key range can be slow despite returning few rows.
3. Interview: Your LSM-backed service has a good p50 and a terrible p99. What is likely happening and what would you tune? (Hint: think about what runs in the background and what it competes for.)

Write-Ahead Logging

A database must survive a crash at any instant, including halfway through updating a page. But writing data pages to disk on every commit would be unbearably slow — they are scattered across the file, so each commit becomes several random writes. Write-ahead logging resolves the conflict: make the durable write cheap and sequential, and let the organized write happen lazily.

The rule is in the name: before any change is applied to a data page, a description of that change is appended to a log and flushed to durable storage. Because the log is append-only, the flush is one sequential write, which is orders of magnitude cheaper than scattered page writes.

The analogy is a chef's order ticket rail. Tickets are written immediately in arrival order — cheap and sequential. The actual dishes are prepared in whatever order is efficient. If the kitchen catches fire, the tickets tell you exactly what was owed.

 COMMIT PATH WITH A WAL

  1. change pages in the buffer pool (RAM, fast, not yet durable)
  2. append log records describing the change
  3. fsync the log            <- the ONLY synchronous disk write per commit
  4. acknowledge the commit to the client
  ...later...
  5. checkpoint: write dirty pages to the data files, sequentially and batched
  6. log segments before the checkpoint can be recycled

 CRASH RECOVERY

  find last checkpoint -> REDO every logged change after it
                       -> UNDO changes from transactions that never committed
  Result: exactly the committed state, nothing more, nothing less.

def commit(txn, wal, buffer_pool):
    for change in txn.changes:
        page = buffer_pool.get(change.page_id)
        wal.append(LogRecord(lsn=next_lsn(), txn=txn.id, page=change.page_id,
                             before=page.snapshot(change.offset),   # for UNDO
                             after=change.new_bytes))               # for REDO
        page.apply(change)
        buffer_pool.mark_dirty(page)

    wal.append(LogRecord(txn=txn.id, type="COMMIT"))
    wal.fsync()          # THE durability barrier. Returning before this means
                         # acknowledging a commit that a power cut can erase.
    return "committed"   # data pages may still be dirty in RAM — that is fine,
                         # the log can reconstruct them.

The WAL is also the foundation for several features that appear later in this chapter, which is why it is worth understanding well:

Feature	How the WAL enables it
Crash recovery	Replay committed changes, roll back uncommitted ones
Replication	Ship log records to replicas and replay them
Point-in-time recovery	Restore a backup, then replay the log to an instant
Change data capture	Read the log to emit a stream of row changes
Group commit	Batch many transactions into one fsync

That last row is where most database write tuning happens. fsync costs milliseconds, so engines batch concurrent commits into a single flush — raising throughput enormously while adding a small latency floor. The dangerous knob is the one that relaxes durability (innodb_flush_log_at_trx_commit=2, synchronous_commit=off): it makes commits much faster by acknowledging before the log is durable, which means a crash silently loses recently "committed" transactions.

Key Takeaways

• Log the change before applying it; the log flush is the only synchronous write per commit.
• Sequential log appends are far cheaper than scattered data-page writes.
• Recovery replays committed changes after the last checkpoint and undoes uncommitted ones.
• The WAL also powers replication, PITR, and CDC — and relaxing its fsync trades durability for speed.

🧪 Practice

1. Trace what happens on recovery if a crash occurs after step 3 but before step 5 in the commit path above.
2. Explain group commit and why it improves throughput but not single-commit latency.
3. Interview: A team sets synchronous_commit=off to double write throughput. What exactly have they given up? (Hint: name the window of time and what a client was told during it.)

Indexes and Index Types

Without an index, answering "find the row where email = x" means reading every row — O(n) work that grows with the table. An index is a separate, ordered structure that maps values to row locations, turning that scan into a handful of page reads. It is the single highest-leverage performance tool in any database, and also the most commonly misapplied.

The cost side matters as much as the benefit. Every index must be updated on every insert, update, and delete of the indexed column, so indexes make writes slower and consume disk and memory. An unused index is pure overhead.

-- The order of columns in a composite index is not cosmetic. Think of it as
-- a phone book sorted by (last_name, first_name).
CREATE INDEX idx_orders_customer_created
    ON orders (customer_id, created_at DESC);

-- USES the index fully: leftmost prefix present, sorted order matches
SELECT * FROM orders WHERE customer_id = 7 ORDER BY created_at DESC LIMIT 20;

-- USES the index partially: only the first column narrows the search
SELECT * FROM orders WHERE customer_id = 7;

-- CANNOT use it: skips the leftmost column, like searching a phone book by
-- first name only
SELECT * FROM orders WHERE created_at > '2026-01-01';

-- A covering index answers the query entirely from the index, never touching
-- the table ("index-only scan").
CREATE INDEX idx_orders_cover ON orders (customer_id, created_at) INCLUDE (total);
SELECT created_at, total FROM orders WHERE customer_id = 7;

-- Common mistakes that silently disable an index
SELECT * FROM users WHERE lower(email) = 'a@b.com';   -- function on the column
CREATE INDEX idx_users_email_lower ON users (lower(email));  -- fix: index the expression

SELECT * FROM events WHERE user_id::text = '7';        -- implicit cast
SELECT * FROM logs  WHERE message LIKE '%timeout%';    -- leading wildcard: no B-tree help
                                                       -- fix: a full-text or trigram index

Index type	Structure	Good for
B-tree	Balanced tree	Equality, ranges, sorting, prefix matches — the default
Hash	Hash table	Equality only; no ranges, no ordering
Bitmap	Bit per row per value	Low-cardinality columns in analytics
GiST / R-tree	Spatial tree	Geometry, ranges, nearest-neighbour
GIN / inverted	Term to row list	Full text, arrays, JSON containment
Trigram	3-character grams	Fuzzy and substring matching
Partial	B-tree on a subset	WHERE status = 'active' — smaller and cheaper
Clustered	Table stored in index order	Range scans on the clustering key

Two structural distinctions worth knowing. A clustered index determines the physical row order, so there can be only one, and range scans on it are sequential (InnoDB clusters on the primary key; PostgreSQL does not cluster by default). A secondary index stores a pointer to the row, so using it costs an extra lookup — unless the index covers the query.

Practical guidance: index the columns in WHERE, JOIN, and ORDER BY clauses of your actual slow queries; prefer a few well-ordered composite indexes over many single-column ones; and periodically drop indexes the database reports as unused.

Key Takeaways

• Indexes turn scans into lookups but slow every write and consume space.
• Composite index column order matters: only a leftmost prefix can be used.
• Covering indexes answer a query without touching the table.
• Functions, casts, and leading wildcards on an indexed column silently disable the index.

🧪 Practice

1. Design indexes for a query filtering on tenant_id and status, sorted by created_at, returning id and title.
2. Explain why WHERE lower(email) = ? misses an index on email and give two fixes.
3. Interview: A table has 12 indexes and writes are slow. How do you decide which to drop? (Hint: the database itself records something that answers this — and you also need to know which queries are actually hot.)

Data Serialization Formats

Every time data leaves a process — over a network, into a queue, onto disk — it must be converted to bytes and back. That choice affects payload size, CPU cost, schema evolution, and whether a human can debug it with cat. Because stored data outlives the code that wrote it, serialization is a long-term commitment.

The formats divide along two axes: text versus binary, and schemaless versus schema-driven.

                    schemaless                 schema-driven
                +---------------------+---------------------------+
      text      |  JSON, XML, CSV     |  JSON + JSON Schema       |
                |  human-readable     |  validated, still verbose |
                +---------------------+---------------------------+
      binary    |  MessagePack, BSON  |  Protobuf, Avro, Thrift   |
                |  compact, opaque    |  compact, evolvable, typed|
                +---------------------+---------------------------+

# The same record in four formats — note size and what each carries.
record = {"id": 42, "name": "Ada", "active": True, "score": 99.5}

# JSON: 49 bytes, self-describing, universally readable
b'{"id":42,"name":"Ada","active":true,"score":99.5}'

# MessagePack: 37 bytes, binary but still self-describing (keys included)
b'\x84\xa2id*\xa4name\xa3Ada\xa6active\xc3\xa5score\xcb@X\xe0\x00\x00\x00\x00\x00'

# Protobuf: 18 bytes, field NUMBERS instead of names; needs the schema to read
b'\x08*\x12\x03Ada\x18\x01!\x00\x00\x00\x00\x00\xe0X@'

# Avro: similar size to Protobuf, but the schema travels with the file (once
# per file, not per record) — ideal for large batches in a data lake.

Format	Size	Speed	Human-readable	Schema	Evolution
JSON	Large	Slow	Yes	Optional	Convention only
XML	Largest	Slowest	Yes	XSD	Verbose but formal
CSV	Small	Fast	Yes	None	Fragile (column order)
MessagePack	Medium	Fast	No	None	Same as JSON
Protobuf	Small	Fastest	No	Required	Field numbers (Chapter 3)
Avro	Small	Fast	No	Required	Reader/writer resolution
Parquet	Smallest (columnar)	Fast for scans	No	Required	Column-level

Parquet deserves separate mention because it is columnar: values of the same column are stored together, so an analytical query reading 3 of 50 columns reads only those 3, and similar values compress far better. That is why analytics stores use it and transactional databases do not — row-oriented storage wins when you need whole records.

 ROW-ORIENTED (OLTP)              COLUMNAR (OLAP)

 [id|name|score][id|name|score]   [id,id,id,...][name,name,...][score,...]

 "give me row 42"    -> 1 read    "average score of 10M rows"
 "average of scores" -> read all      -> read ONE column, highly compressed

Practical guidance: JSON for public APIs and anything a human debugs; Protobuf or Avro for internal high-volume traffic and long-lived stored events; Parquet for analytical data at rest. And whichever you choose, decide the evolution strategy before the first record is written (Chapter 3.3).

Key Takeaways

• Serialization choice governs payload size, CPU, evolvability, and debuggability.
• Text formats are readable and verbose; binary schema-driven formats are compact and evolvable.
• Columnar formats like Parquet win for analytics because queries read only the columns they need.
• Stored data outlives its writer — decide schema evolution before writing the first record.

🧪 Practice

1. Estimate the storage difference between 1 billion events at 300 bytes of JSON versus 90 bytes of Protobuf.
2. Explain why Parquet compresses better than row-oriented storage for the same data.
3. Interview: You are choosing a format for events retained seven years. What matters most? (Hint: the code that reads them in year six does not exist yet — what must be true for it to succeed?)

<a id="42-relational-databases"></a>

4.2 Relational Databases

Relational databases remain the default for transactional data because they combine a declarative query language with correctness guarantees that are expensive to rebuild by hand.

Schema Design and Normalization

Store the same fact in two places and, sooner or later, the two copies will disagree. Normalization is the systematic prevention of that outcome: organize data so every fact lives in exactly one place, and inconsistency becomes structurally impossible rather than something you police with application code.

The problems normalization eliminates have names. An update anomaly is changing a customer's address in three of four rows. An insertion anomaly is being unable to record a product because no order references it yet. A deletion anomaly is losing a supplier's phone number because the last order from them was deleted.

 UNNORMALIZED: one wide table, facts repeated

 order_id | customer_name | customer_email | product   | price | qty
 ---------|---------------|----------------|-----------|-------|----
 1        | Ada Lovelace  | ada@ex.com     | Lamp      | 29.00 |  2
 2        | Ada Lovelace  | ada@ex.com     | Desk      | 199.00|  1
 3        | Ada Lovelace  | ada@exam.com   | Chair     | 89.00 |  1   <- drifted!

 NORMALIZED: each fact in exactly one place

 customers            orders                    order_items
 id | name  | email   id | customer_id | date   order_id | product_id | qty | price
 7  | Ada   | ada@... 1  | 7           | ...    1        | 12         | 2   | 29.00
                                                ^ price captured at purchase time:
                                                  a genuinely different fact from
                                                  the product's CURRENT price

The normal forms are a progression, and the first three cover almost all practical cases:

Form	Requirement	Plain-language rule
1NF	Atomic values; no repeating groups	No comma-separated lists in a column
2NF	1NF + no partial dependency on part of a composite key	Every column depends on the whole key
3NF	2NF + no transitive dependencies	Non-key columns must not depend on each other
BCNF	Stricter 3NF for overlapping candidate keys	Every determinant is a candidate key

-- 1NF violation: a list stuffed into one column
CREATE TABLE users (id INT, tags TEXT);          -- tags = 'admin,beta,eu'
-- Cannot index, cannot join, cannot count reliably.

-- 3NF violation: zip determines city, so city depends on a non-key column
CREATE TABLE addresses (id INT PRIMARY KEY, zip TEXT, city TEXT);

-- Normalized, with the constraints that make correctness the database's job
CREATE TABLE customers (
    id    BIGSERIAL PRIMARY KEY,
    email CITEXT NOT NULL UNIQUE,                 -- uniqueness enforced here,
    name  TEXT NOT NULL                           -- not by a racy app-side check
);

CREATE TABLE orders (
    id          BIGSERIAL PRIMARY KEY,
    customer_id BIGINT NOT NULL REFERENCES customers(id) ON DELETE RESTRICT,
    status      TEXT NOT NULL CHECK (status IN ('pending','paid','cancelled')),
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE order_items (
    order_id   BIGINT NOT NULL REFERENCES orders(id) ON DELETE CASCADE,
    product_id BIGINT NOT NULL REFERENCES products(id),
    qty        INT NOT NULL CHECK (qty > 0),
    unit_price NUMERIC(12,2) NOT NULL,            -- historical price, not a copy
    PRIMARY KEY (order_id, product_id)            -- natural composite key
);

Constraints deserve emphasis. A UNIQUE constraint is not a duplicate of an application check — it is the only version that holds under concurrency, because two simultaneous requests can both pass a SELECT ... WHERE email = ? check and then both insert. Foreign keys, CHECK constraints, and NOT NULL work the same way: they are correctness guarantees the database enforces even when application code has bugs.

Also decide money and time representation once, at schema design time: use exact NUMERIC/DECIMAL for money (never FLOAT, which cannot represent 0.10 exactly) and TIMESTAMPTZ in UTC for instants.

Key Takeaways

• Normalization puts every fact in one place, making update, insert, and delete anomalies impossible.
• 3NF covers nearly all practical cases: every non-key column depends on the key, the whole key, and nothing but the key.
• Constraints (UNIQUE, foreign keys, CHECK) are concurrency-safe guarantees that application checks are not.
• Historical values like purchase price are distinct facts, not denormalized copies.

🧪 Practice

1. Normalize a spreadsheet-style table of employee, department, dept_manager, project, hours to 3NF and name the anomalies you removed.
2. Explain why an application-level uniqueness check is unsafe and write the constraint that fixes it.
3. Interview: When is storing a product's price on the order line not denormalization? (Hint: ask whether the two values can ever legitimately differ, and what each one means.)

Denormalization Trade-offs

Normalization optimizes for write correctness; it can be hostile to read performance. A fully normalized schema may require joining six tables to render one screen, and at high read volume those joins become the bottleneck. Denormalization deliberately reintroduces redundancy to make reads cheaper — trading write complexity and consistency risk for read speed.

The essential framing: denormalization is not "bad normalization", it is a cache stored inside your database, and it has all the problems caches have. Every redundant copy must be kept in sync, and the cost of that synchronization is what you are buying read speed with.

-- Normalized: correct, but a hot query has to aggregate on every read
SELECT p.id, p.title,
       count(c.id)   AS comment_count,      -- recomputed for every page view
       avg(r.rating) AS avg_rating
FROM posts p
LEFT JOIN comments c ON c.post_id = p.id
LEFT JOIN ratings  r ON r.post_id = p.id
GROUP BY p.id
ORDER BY p.created_at DESC
LIMIT 20;                                   -- scans all comments and ratings

-- Denormalized: counters maintained on write, so reads are a single-table scan
ALTER TABLE posts
  ADD COLUMN comment_count INT     NOT NULL DEFAULT 0,
  ADD COLUMN rating_sum    INT     NOT NULL DEFAULT 0,
  ADD COLUMN rating_count  INT     NOT NULL DEFAULT 0;

SELECT id, title, comment_count,
       CASE WHEN rating_count > 0
            THEN rating_sum::numeric / rating_count END AS avg_rating
FROM posts ORDER BY created_at DESC LIMIT 20;   -- no joins, no aggregation

-- Keeping the copy correct is the entire cost. Update it in the SAME
-- transaction as the source of truth, or the two will drift.
BEGIN;
  INSERT INTO comments (post_id, author_id, body) VALUES (42, 7, 'Nice');
  UPDATE posts SET comment_count = comment_count + 1 WHERE id = 42;
COMMIT;
-- Two consequences to accept knowingly:
--   1. Every comment insert now contends on the posts row (a hotspot for a
--      viral post — see 4.5 on hotspots).
--   2. Any path that deletes comments without this update silently corrupts
--      the counter, so a periodic reconciliation job is mandatory.

Technique	What it duplicates	Sync mechanism
Counter column	An aggregate	Same-transaction update; periodic reconciliation
Duplicated attribute	A parent's column on the child	Trigger or application write path
Precomputed join table	A whole join result	Batch or event-driven rebuild
Materialized view	A query result	Database-managed refresh
Read model (CQRS)	An entire query-optimized store	Event stream (Chapter 10.3)

The decision checklist:

• Measure first. Denormalize a query you have proven is slow, not one you suspect might be.
• Prefer reversible forms. A materialized view can be dropped and rebuilt; a duplicated column scattered through application code cannot.
• Always have a reconciliation job. Redundant data drifts — the question is only whether you detect it before your users do.
• Keep the source of truth unambiguous. Every derived value must have exactly one authoritative origin.

Key Takeaways

• Denormalization buys read speed with write complexity and consistency risk.
• It is a cache inside the database, with all the invalidation problems that implies.
• Update derived values in the same transaction, and run periodic reconciliation regardless.
• Prefer database-managed forms (materialized views) over hand-maintained copies.

🧪 Practice

1. Add a denormalized last_message_at to a conversations table and list every write path that must maintain it.
2. Write a reconciliation query that detects drift in a comment_count column.
3. Interview: A counter column on a viral post becomes a write hotspot. How do you fix it without giving up fast reads? (Hint: the contention is on one row — consider spreading the write and merging on read.)

ACID Properties

A transaction is a group of operations that must succeed or fail as a unit. Without that guarantee, a transfer between two accounts can debit one and never credit the other, and no amount of careful application code fully closes the gap, because crashes and concurrency do not respect your control flow. ACID names the four guarantees that make transactions trustworthy.

Atomicity — all operations commit or none do. Consistency — the database moves from one valid state to another, respecting every declared constraint. Isolation — concurrent transactions do not observe each other's partial work. Durability — once committed, it survives a crash.

Note that these are different guarantees with different implementations: atomicity and durability come from the write-ahead log, isolation from locking or MVCC, and consistency from constraints plus the other three.

 A TRANSFER WITHOUT ATOMICITY

  UPDATE accounts SET balance = balance - 100 WHERE id = 1;   -- succeeds
  *** crash ***
  UPDATE accounts SET balance = balance + 100 WHERE id = 2;   -- never runs

  Result: 100 units destroyed. No retry can detect this reliably, because the
  process that knew what it was doing is gone.

 WITH ATOMICITY

  BEGIN; ...both updates...; COMMIT;
  A crash before COMMIT -> recovery rolls both back. A crash after COMMIT ->
  recovery replays both. There is no in-between state to observe.

def transfer(conn, from_id, to_id, amount):
    with conn.transaction():                       # BEGIN ... COMMIT/ROLLBACK
        # A: both updates or neither
        rows = conn.execute(
            "UPDATE accounts SET balance = balance - %s "
            "WHERE id = %s AND balance >= %s",     # C: the CHECK is in the WHERE
            (amount, from_id, amount))
        if rows.rowcount == 0:
            raise InsufficientFunds()              # rollback: nothing happened
        conn.execute("UPDATE accounts SET balance = balance + %s WHERE id = %s",
                     (amount, to_id))
        # I: another transaction reading these rows sees either the pre- or
        #    post-state, never one update without the other.
    # D: the commit returns only after the WAL is fsynced.

# The classic mistake: doing the arithmetic in application code.
#   bal = SELECT balance ...        <- read
#   UPDATE ... SET balance = bal-100 <- write based on a stale read
# Two concurrent transfers both read 500, both write 400, and 100 units appear
# from nowhere. Let the database do the arithmetic, or lock the row explicitly.

Property	Guarantees	Implemented by	Failure if absent
Atomicity	All or nothing	WAL undo, rollback	Partial updates, orphaned rows
Consistency	Constraints always hold	Constraints + A, I, D	Invalid states (negative stock)
Isolation	Concurrent transactions do not interfere	Locking or MVCC	Dirty/phantom reads, lost updates
Durability	Commits survive crashes	WAL fsync, replication	Acknowledged data disappears

Two caveats that matter in system design. Isolation is the property most often weakened deliberately for performance — the next topic covers exactly how much you give up. And ACID is a guarantee about one database; spanning several services or shards requires the distributed techniques in Chapter 8.5 (two-phase commit, sagas, outbox), which are strictly weaker and much more complex.

Key Takeaways

• Atomicity and durability come from the WAL; isolation from locking or MVCC; consistency from constraints plus the others.
• Perform arithmetic inside the database or under an explicit lock — never on a stale application-side read.
• Isolation is the property routinely traded away for throughput.
• ACID applies within one database; crossing service or shard boundaries needs different, weaker machinery.

🧪 Practice

1. Show a concrete interleaving where two concurrent read-modify-write transfers lose an update, and fix it two different ways.
2. Which ACID property does each prevent: a NOT NULL violation, reading uncommitted data, losing a commit to a power cut, a half-completed transfer?
3. Interview: Why can't you get full ACID across two microservices with separate databases? (Hint: consider what a commit would have to coordinate and what happens when one participant becomes unreachable mid-decision.)

Transaction Isolation Levels

Perfect isolation means transactions behave as if they ran one at a time. That is correct and slow — serial execution wastes the concurrency modern hardware offers. So databases provide a dial, and the levels on that dial are defined by which specific anomalies they permit.

Knowing the anomalies by name is what makes the dial usable:

Anomaly	What happens
Dirty read	You read data another transaction wrote but has not committed
Non-repeatable read	You read the same row twice and get different values
Phantom read	You run the same query twice and new rows appear
Lost update	Two read-modify-write cycles overlap; one update vanishes
Write skew	Both transactions read a shared state, then each writes based on it, breaking an invariant neither violated alone

 THE FOUR STANDARD LEVELS

                       dirty   non-repeatable  phantom   write
                       read        read         read     skew
 READ UNCOMMITTED       yes         yes          yes      yes
 READ COMMITTED         no          yes          yes      yes
 REPEATABLE READ        no          no           yes*     yes
 SERIALIZABLE           no          no           no       no

 * PostgreSQL's REPEATABLE READ (snapshot isolation) also prevents phantoms,
   but still allows write skew. MySQL InnoDB prevents phantoms with gap locks.
   Defaults differ: PostgreSQL and Oracle -> READ COMMITTED; MySQL -> REPEATABLE READ.

Write skew is the one that surprises people, because every transaction looks individually correct:

-- Invariant: at least one doctor must remain on call.
-- Two doctors, both on call, both try to go off call simultaneously.

-- Transaction A                         -- Transaction B
BEGIN;                                   BEGIN;
SELECT count(*) FROM oncall              SELECT count(*) FROM oncall
  WHERE on_call = true;   -- sees 2        WHERE on_call = true;   -- sees 2
-- "2 > 1, safe to leave"                -- "2 > 1, safe to leave"
UPDATE oncall SET on_call = false        UPDATE oncall SET on_call = false
  WHERE doctor_id = 1;                     WHERE doctor_id = 2;
COMMIT;                                  COMMIT;
-- Result: zero doctors on call. Neither transaction saw the other's write,
-- and each preserved the invariant against the state it saw.

-- FIX 1: SERIALIZABLE — the database detects the conflict and aborts one
BEGIN ISOLATION LEVEL SERIALIZABLE; ...   -- application must retry on failure

-- FIX 2: materialize the conflict with an explicit lock
SELECT count(*) FROM oncall WHERE on_call = true FOR UPDATE;  -- serializes them

# SERIALIZABLE moves the burden to the application: transactions can now fail
# with a serialization error, and retrying is mandatory, not optional.
def run_serializable(conn, fn, attempts=3):
    for attempt in range(attempts):
        try:
            with conn.transaction(isolation="SERIALIZABLE"):
                return fn(conn)
        except SerializationFailure:          # e.g. PostgreSQL SQLSTATE 40001
            if attempt == attempts - 1:
                raise
            time.sleep(0.05 * 2 ** attempt)   # backoff; the retry usually succeeds

Practical guidance: READ COMMITTED for most workloads, with explicit SELECT ... FOR UPDATE on the specific rows whose invariants matter. Reach for SERIALIZABLE when correctness rules span multiple rows in ways that are hard to lock precisely — and only if every transaction path has a retry loop.

Key Takeaways

• Isolation levels are defined by which anomalies they permit, not by how they are implemented.
• Defaults differ by database — know yours; READ COMMITTED and REPEATABLE READ are both common.
• Write skew survives snapshot isolation and is invisible in single-transaction review.
• SERIALIZABLE requires application retry logic; without it you have swapped silent corruption for visible failures nobody handles.

🧪 Practice

1. Construct an interleaving that produces a non-repeatable read, then show which level prevents it.
2. Write the booking logic for "a meeting room cannot be double-booked" at READ COMMITTED, using explicit locking.
3. Interview: Your service intermittently violates a business rule that every code path appears to enforce. What do you suspect? (Hint: consider what two correct transactions can do together that neither does alone.)

Locking and MVCC

Isolation levels describe what you get; locking and MVCC are how databases deliver it. The two approaches embody opposite philosophies, and knowing which one your database uses explains most of its concurrency behaviour.

Locking (pessimistic) assumes conflict: acquire a lock before touching data, and make everyone else wait. It is simple and correct, but readers block writers and writers block readers, so throughput collapses under contention and deadlocks become possible.

MVCC (Multi-Version Concurrency Control, optimistic) keeps multiple versions of each row. A writer creates a new version rather than overwriting, and each transaction reads the version that was current when it started. The headline consequence: readers never block writers, and writers never block readers. This is why PostgreSQL, Oracle, MySQL InnoDB, and most modern engines use it.

 MVCC: EACH ROW IS A CHAIN OF VERSIONS

  row id=42   [v1 balance=500, xmin=100, xmax=105]   <- superseded by txn 105
              [v2 balance=400, xmin=105, xmax=null]  <- current

  Txn 103 (started before 105 committed) reads v1: 500
  Txn 107 (started after)                reads v2: 400
  Neither waits for the other. The cost: old versions accumulate and must be
  cleaned up (PostgreSQL VACUUM, InnoDB purge). Neglect that and you get
  table bloat and, in Postgres, transaction-id wraparound danger.

-- PESSIMISTIC: take the lock, then act. Correct under contention, but holds
-- the lock for the whole transaction — keep it short and never do I/O inside.
BEGIN;
SELECT stock FROM products WHERE id = 42 FOR UPDATE;   -- row lock acquired
UPDATE products SET stock = stock - 1 WHERE id = 42;
COMMIT;                                                 -- lock released

-- OPTIMISTIC: no lock; detect conflict at write time using a version column.
-- Best when conflicts are rare — no waiting in the common case.
UPDATE products
   SET stock = stock - 1, version = version + 1
 WHERE id = 42 AND version = 7;        -- 0 rows affected => someone else won
-- The application checks rowcount and retries with a fresh read.

-- Avoiding deadlocks: two transactions locking the same rows in OPPOSITE order
-- will deadlock. The fix is a global ordering rule.
SELECT * FROM accounts WHERE id IN (1, 2) ORDER BY id FOR UPDATE;
--                                          ^ always lock in ascending id order

Aspect	Pessimistic locking	MVCC / optimistic
Conflict handling	Prevent by waiting	Detect at commit, retry
Readers vs. writers	Block each other	Never block each other
Throughput, low contention	Lower (lock overhead)	Higher
Throughput, high contention	Predictable	Retry storms possible
Deadlocks	Yes; needs lock ordering	No deadlocks, but aborts
Storage cost	None	Old versions until cleanup
Application burden	None	Must handle retries

Two operational consequences follow directly from MVCC. Long-running transactions are dangerous: an idle transaction left open holds back cleanup of every version newer than its snapshot, bloating tables across the whole database. And SELECT ... FOR UPDATE still exists in MVCC systems — you need it whenever a read informs a write, because MVCC's snapshot alone does not prevent lost updates.

Key Takeaways

• Locking prevents conflicts by waiting; MVCC allows concurrency and detects conflicts.
• Under MVCC, readers never block writers and writers never block readers.
• MVCC's cost is version accumulation — vacuum/purge must keep up, and long idle transactions block it.
• Use FOR UPDATE when a read decides a write, and always acquire multiple locks in a consistent order.

🧪 Practice

1. Show a two-transaction interleaving that deadlocks, then rewrite it with a lock-ordering rule that cannot.
2. Implement optimistic concurrency with a version column, including the retry path.
3. Interview: A monitoring dashboard opens a transaction and leaves it idle for hours. What breaks and why? (Hint: think about what the database must keep around in case that transaction reads again.)

Query Planning and Optimization

SQL is declarative: you state the result you want, not how to compute it. The query planner turns that statement into an execution strategy, and for any non-trivial query there are many possible strategies whose costs differ by orders of magnitude. Understanding the planner is what turns "the database is slow" into a specific, fixable problem.

The planner estimates cost from statistics — row counts, value distributions, and index selectivity — collected by background sampling. Its estimates are usually good, and when they are wrong it is almost always because statistics are stale or the data is skewed.

-- ALWAYS read the actual plan rather than guessing.
EXPLAIN (ANALYZE, BUFFERS)
SELECT o.id, o.total, c.name
FROM orders o
JOIN customers c ON c.id = o.customer_id
WHERE o.created_at >= '2026-01-01' AND o.status = 'paid'
ORDER BY o.created_at DESC
LIMIT 20;

 READING A PLAN (innermost/most-indented runs first)

 Limit  (cost=0.85..42.1 rows=20)  (actual time=0.05..1.20 rows=20 loops=1)
   -> Nested Loop  (cost=0.85..2104 rows=1000)  (actual rows=20)
        -> Index Scan Backward using idx_orders_created on orders o
             Index Cond: (created_at >= '2026-01-01')
             Filter: (status = 'paid')
             Rows Removed by Filter: 4820          <- WARNING: reading and
        -> Index Scan using customers_pkey on customers c    discarding 4820 rows
             Index Cond: (id = o.customer_id)

 What to look for:
   estimated rows vs. ACTUAL rows      large gap  -> stale statistics or skew
   Seq Scan on a large table           -> missing index (fine on small tables)
   "Rows Removed by Filter" large      -> the index does not cover the filter
   Nested Loop with many loops         -> should probably be a hash join
   external merge Disk: 24MB           -> sort spilled; raise work_mem or index

-- The fix here: an index matching BOTH the filter and the sort order.
CREATE INDEX idx_orders_status_created
    ON orders (status, created_at DESC);
-- Now: equality column first, then the range/sort column. The planner can seek
-- directly to status='paid' and walk created_at in order, so LIMIT 20 stops
-- after 20 rows instead of filtering thousands.

-- Keep statistics fresh; after a bulk load they are always wrong.
ANALYZE orders;

-- For a skewed column the default sample may be too small
ALTER TABLE orders ALTER COLUMN status SET STATISTICS 1000;

Join strategy	How it works	Planner picks it when
Nested loop	For each outer row, look up inner	Outer side is small; inner indexed
Hash join	Build a hash table, probe it	Large unsorted inputs, equality join
Merge join	Walk two sorted inputs together	Both inputs already sorted on the key

The most common real-world planner problems, in order of frequency: stale statistics after a bulk load; non-sargable predicates (a function or cast on the indexed column, from the previous subchapter); LIMIT with an unsupported sort, forcing a full sort before returning 20 rows; and parameter-sensitive plans, where a plan cached for a selective value is disastrous for a common one.

Key Takeaways

• The planner chooses among strategies using table statistics; stale statistics are the top cause of bad plans.
• Read EXPLAIN ANALYZE and compare estimated with actual rows — a large gap points to the problem.
• Match indexes to both the filter and the sort so LIMIT can stop early.
• Watch for sequential scans on large tables, high "rows removed by filter", and sorts spilling to disk.

🧪 Practice

1. Take a slow query you know, run EXPLAIN ANALYZE, and identify the single most expensive node.
2. Explain why ORDER BY created_at LIMIT 10 can be slow without a matching index even though it returns 10 rows.
3. Interview: A query is fast in staging and slow in production with identical schemas. Name three causes. (Hint: what differs between the two environments besides the SQL text?)

Connection Pooling

Opening a database connection is expensive: a TCP handshake, TLS negotiation, authentication, and — in PostgreSQL — forking a backend process that allocates several megabytes. Doing that per request adds tens of milliseconds and can exhaust the database's connection limit long before its CPU is busy. A connection pool keeps a set of established connections and lends them out.

The counterintuitive part is that more connections do not mean more throughput. Past a point set by cores and disk parallelism, additional connections only add context switching, lock contention, and memory pressure — throughput flattens and latency rises. The right pool is usually far smaller than teams expect.

 WITHOUT POOLING                      WITH POOLING

 request -> connect (30 ms)           request -> borrow (0.1 ms)
         -> query   (2 ms)                    -> query   (2 ms)
         -> close                             -> return to pool

 93% of the time is setup.            The connection is established once and
 At 500 RPS you also create 500       reused thousands of times.
 backends/second on the database.

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql://app@db/orders",
    pool_size=20,          # persistent connections per application instance
    max_overflow=10,       # temporary extras under burst (total ceiling: 30)
    pool_timeout=3,        # FAIL FAST if none is free; do not queue forever
    pool_recycle=1800,     # recycle before the DB or a proxy times them out
    pool_pre_ping=True,    # cheap liveness check; avoids handing out dead sockets
)

# Sizing is arithmetic, not intuition. Little's Law (Chapter 1.2):
#   connections needed = query_rate x average_query_duration
qps, avg_query_s = 800, 0.004
print(qps * avg_query_s)                     # 3.2 -> ~4 busy at any instant
# A pool of 20 leaves generous headroom for latency spikes. A pool of 200 would
# just queue work inside the database instead of inside the pool.

# The ceiling that actually matters is the DATABASE's, shared by every client:
instances, pool_per_instance = 40, 30
print(instances * pool_per_instance)         # 1200 -> far above a typical
                                             # max_connections of 100-500

That last calculation is the classic outage. Each service instance looks reasonable in isolation; multiplied by autoscaling, the fleet exhausts the database's connection limit and every service fails at once, including the ones that were healthy.

Problem	Fix
Pool exhausted under load	Shorten queries; fail fast rather than queueing forever
Fleet exceeds database max_connections	An external pooler (PgBouncer) in front of the database
Connections killed by an idle firewall	pool_recycle shorter than the idle timeout
Leaked connections	Always use context managers; monitor checked-out count
Long transactions holding connections	Never do HTTP calls or sleeps inside a transaction

For large fleets, an external pooler such as PgBouncer in transaction mode is the standard answer: thousands of client connections multiplex onto a few dozen real database connections. The caveat is that transaction-mode pooling breaks session-scoped features — prepared statements, advisory locks, SET state, and temporary tables — so it changes what your application may assume.

Key Takeaways

• Connection setup is expensive; pools amortize it across many requests.
• More connections do not mean more throughput — size from Little's Law, not optimism.
• The real limit is the database's total, shared across every instance; multiply before you autoscale.
• Fail fast on pool exhaustion, recycle connections, and never hold one across an external call.

🧪 Practice

1. Size a pool for a service handling 1,200 QPS with 6 ms average query time, and state your headroom assumption.
2. Compute the fleet-wide connection total for 60 instances with pool_size=25 and compare it with a max_connections of 400.
3. Interview: Adding application servers made database latency worse. Explain. (Hint: consider what each new server brings with it, and what the database does when it has more concurrent work than cores.)

<a id="43-nosql-and-specialized-stores"></a>

4.3 NoSQL and Specialized Stores

Relational databases are general-purpose; each store in this subchapter gives up some generality to be dramatically better at one access pattern.

Key-Value Stores

The simplest possible database: a distributed hash map. You GET and PUT values by key, and the store makes no attempt to understand what the value contains. That restriction is the source of its power — with no query planner, no joins, and no secondary indexes to maintain, lookups are O(1) and the data partitions trivially across any number of nodes by hashing the key.

The analogy is a coat check. Hand over a coat, get a numbered ticket, retrieve it instantly by number. You cannot ask "which coats are blue?" — that is precisely the query the design refuses to support.

 KEY                          VALUE (opaque bytes)
 ----------------------------------------------------
 session:a3f9c2               {"user_id":7,"exp":...}
 cart:user:7                  {"items":[...]}
 ratelimit:client:88:minute   47
 feature_flags:v3             {"new_checkout":true}

 Supported:      GET, PUT, DELETE, TTL, atomic counters
 NOT supported:  "find all sessions for user 7"  <- needs a second key

# Redis: the value type is where the expressiveness lives.
r.setex("session:a3f9c2", 3600, json.dumps(session))   # string + TTL
r.incr("views:post:42")                                # atomic counter
r.hset("user:7", mapping={"name": "Ada", "tier": "pro"})  # hash: partial updates
r.zadd("leaderboard", {"ada": 4820})                   # sorted set: ranked queries
r.zrevrange("leaderboard", 0, 9)                       # top 10 in O(log n + 10)

# The design discipline: you must know the key before you can read.
# Any other access path is a key you maintain YOURSELF, in the same transaction.
pipe = r.pipeline(transaction=True)
pipe.set(f"session:{sid}", payload, ex=3600)
pipe.sadd(f"user_sessions:{user_id}", sid)     # the reverse index, hand-built
pipe.expire(f"user_sessions:{user_id}", 3600)
pipe.execute()
# Now "log this user out everywhere" is possible — because you planned for it.

Store	Persistence	Distinguishing feature
Redis	Optional (RDB/AOF)	Rich value types, single-threaded atomicity
Memcached	None	Pure cache, multi-threaded, very simple
DynamoDB	Durable, managed	Serverless scaling, secondary indexes
etcd / ZooKeeper	Durable, consensus	Strong consistency for coordination
RocksDB	Embedded	LSM engine used inside other systems

Key-value stores are the right answer for caching, sessions, rate limiting, feature flags, leaderboards, and any lookup where the key is known. They are the wrong answer whenever you need to query by value, and the failure mode is predictable: someone eventually writes a SCAN over the keyspace in production and blocks the server.

Key Takeaways

• A key-value store trades query capability for O(1) lookups and trivial partitioning.
• Every access path must be a key you deliberately maintain, including reverse indexes.
• Redis's value types (hashes, sorted sets) provide atomic operations beyond simple get/put.
• Never scan the keyspace in production — if you need that query, you chose the wrong store.

🧪 Practice

1. Design the key schema for a shopping cart supporting "get cart", "add item", and "expire after 30 days".
2. Implement "log out all sessions for a user" and name what must happen atomically.
3. Interview: When is Redis the primary store rather than a cache? (Hint: ask what happens to the business if the dataset is lost, and what durability settings actually guarantee.)

Document Stores

Relational schemas require you to know your fields in advance and to spread one logical entity across several tables. That is exactly right when data is shared between many use cases, and it is friction when an entity is naturally self-contained and its shape varies. Document stores keep each entity as one nested JSON-like document, retrievable in a single read.

The key property is locality: everything about one entity sits together on disk, so fetching it costs one read rather than a five-table join. The trade is that data shared across documents gets duplicated, and keeping duplicates in sync becomes your job.

// One document holds what would be four normalized tables.
{
  "_id": ObjectId("..."),
  "orderNumber": "ORD-2026-4821",
  "status": "paid",
  "customer": {                        // embedded: fetched with the order,
    "id": 7,                           // duplicated if the name changes
    "name": "Ada Lovelace",
    "shippingAddress": { "city": "Turin", "country": "IT" }
  },
  "items": [                           // an array, not a join table
    { "productId": 12, "name": "Lamp", "qty": 2, "unitPrice": "29.00" }
  ],
  "total": "58.00",
  "createdAt": ISODate("2026-03-11T10:04:22Z")
}

// The central modelling decision: embed or reference.
db.orders.findOne({ orderNumber: "ORD-2026-4821" });   // 1 read, whole order

// EMBED when: the child is owned by the parent, read together, bounded in size,
//             and rarely changes independently.       (order items)
// REFERENCE when: the child is shared, unbounded, or updated on its own.
{ "_id": 42, "authorId": 7 }                            // reference
db.posts.aggregate([                                    // then $lookup to join
  { $match: { _id: 42 } },
  { $lookup: { from: "users", localField: "authorId",
               foreignField: "_id", as: "author" } }
]);

// The hard limit that decides many cases: a document has a maximum size
// (16 MB in MongoDB). An unbounded array — comments on a viral post, events on
// a device — will eventually exceed it. Never embed an unbounded collection.

Aspect	Document store	Relational
Entity storage	One nested document	Rows across several tables
Schema	Flexible per document	Enforced, uniform
Read of one entity	Single read	Join across tables
Shared data	Duplicated; must be synced	Referenced once
Multi-entity queries	Weaker ($lookup is limited)	Native and optimized
Schema changes	No migration needed to add fields	ALTER TABLE, planned migration

"Schemaless" is the most misleading word in this area. The schema does not disappear; it moves into application code, where every reader must handle every historical shape a document might have. Mature teams therefore apply schema validation at the database level and version their documents explicitly — which is most of the discipline of a relational schema, applied voluntarily.

Key Takeaways

• Documents give locality: one entity, one read, nested structure included.
• Embed owned, bounded, co-read data; reference shared, unbounded, or independently updated data.
• Never embed an unbounded array — document size limits are hard limits.
• Schemaless moves the schema into application code; use validation and document versions deliberately.

🧪 Practice

1. Model a blog with posts, authors, comments, and tags. Justify each embed-versus-reference decision.
2. Explain what breaks when a user changes their display name in a design that embeds the author in every post, and give two fixes.
3. Interview: A team chose a document store "so we don't need migrations". What have they actually deferred? (Hint: the old data does not update itself — who deals with it, and when?)

Wide-Column Stores

Some workloads write far more than they read, need to run across many datacenters without a single leader, and must scale to petabytes with predictable latency. Relational databases struggle at that shape. Wide-column stores (Cassandra, ScyllaDB, HBase, Bigtable) are built for it, with an LSM engine, leaderless replication, and a data model designed around partitions.

The model looks tabular but behaves very differently. A partition key determines which node holds the data; clustering keys determine the sort order within that partition. A query that specifies the partition key is a single-node sorted lookup — extremely fast. A query that does not is a scatter-gather across the whole cluster, and the database will usually refuse to run it.

 PARTITION KEY = which node.  CLUSTERING KEYS = sort order inside the node.

 PRIMARY KEY ((sensor_id), reading_time DESC)
                ^^^^^^^^^   ^^^^^^^^^^^^
 node A:  sensor_1 -> [t9|22.1][t8|22.3][t7|22.0] ...   one contiguous, sorted row
 node B:  sensor_2 -> [t9|18.4][t8|18.6] ...
 node C:  sensor_3 -> ...

 FAST:  WHERE sensor_id = 1 AND reading_time > ...   (one partition, sorted scan)
 SLOW:  WHERE reading_time > ...                     (every partition; refused
                                                      without ALLOW FILTERING)

-- The defining discipline: model tables per QUERY, not per entity. Duplicating
-- data across tables is normal and expected here, not a design smell.
CREATE TABLE readings_by_sensor (
    sensor_id    UUID,
    reading_time TIMESTAMP,
    value        DOUBLE,
    PRIMARY KEY ((sensor_id), reading_time)
) WITH CLUSTERING ORDER BY (reading_time DESC);

CREATE TABLE readings_by_site_day (       -- the SAME data, keyed for a second query
    site_id      UUID,
    day          DATE,                    -- bucketing keeps partitions bounded
    reading_time TIMESTAMP,
    sensor_id    UUID,
    value        DOUBLE,
    PRIMARY KEY ((site_id, day), reading_time)
);
-- The application writes both. There are no joins to fall back on.

-- Tunable consistency per query (see 4.4 on quorums):
SELECT * FROM readings_by_sensor WHERE sensor_id = ? LIMIT 100;   -- often ONE
INSERT INTO readings_by_sensor (...) VALUES (...);                -- often QUORUM

Property	Wide-column	Relational
Write path	LSM, append-only, very fast	B-tree, in-place
Scaling	Linear, add nodes	Vertical, then sharding
Topology	Leaderless, multi-datacenter native	Leader-follower
Joins	None	Native
Query planning	None — the key decides everything	Cost-based planner
Consistency	Tunable per operation	ACID by default
Modelled around	Queries	Entities

Two failure modes account for most wide-column trouble. Unbounded partitions: a partition key like site_id alone grows forever, eventually making one row gigabytes in size — hence the day bucket above. And tombstone accumulation from deletes or TTLs, which makes range scans read enormous numbers of deleted rows before returning anything.

Key Takeaways

• Wide-column stores optimize for write throughput, linear scaling, and multi-datacenter operation.
• The partition key picks the node; clustering keys sort within it; queries must supply the partition key.
• Model one table per query and duplicate data freely — there are no joins.
• Bound partition sizes with bucketing, and watch tombstones on delete-heavy tables.

🧪 Practice

1. Design tables for "messages in a chat room, newest first" and "all messages by one user", and state what the write path must do.
2. Explain why PRIMARY KEY ((user_id), created_at) becomes a problem for a user with 50 million events, and fix it.
3. Interview: Why does Cassandra have no joins, and what does that force onto the application? (Hint: think about what a join would require across nodes, and when the alternative work has to happen instead.)

Graph Databases

Some questions are about relationships rather than records: "who are the friends of my friends who work at this company?", "what is the shortest path of ownership between these two companies?", "which accounts are within three transfers of a known fraudster?". In SQL, each additional hop is another self-join, and cost grows exponentially. Graph databases make traversal the primitive operation instead.

The reason they are fast is index-free adjacency: each node stores direct pointers to its neighbours, so following an edge is a pointer dereference rather than an index lookup. Traversal cost is proportional to the part of the graph you actually visit, not to the total number of nodes.

 THE SAME QUESTION, TWO SHAPES

 RELATIONAL: friends-of-friends-of-friends
   friendships JOIN friendships JOIN friendships
   -> each join multiplies rows; at depth 5 the query is unusable

 GRAPH:
        (Ada)---FRIEND---(Bob)---FRIEND---(Cleo)
          |                                  |
       WORKS_AT                           WORKS_AT
          v                                  v
       (AcmeCo)                          (AcmeCo)

   MATCH (a)-[:FRIEND*2..3]-(f)-[:WORKS_AT]->(c)
   -> walk pointers outward; cost depends on the neighbourhood visited

// Cypher reads like the picture it describes.
MATCH (me:Person {email: $email})-[:FRIEND*2..3]-(fof:Person)
      -[:WORKS_AT]->(c:Company {name: 'AcmeCo'})
WHERE NOT (me)-[:FRIEND]-(fof)        // exclude direct friends
  AND me <> fof
RETURN DISTINCT fof.name, length(shortestPath((me)-[:FRIEND*]-(fof))) AS distance
ORDER BY distance
LIMIT 10;

-- The relational equivalent, for comparison. It works, and it degrades fast
-- as depth grows because each level fans out.
WITH RECURSIVE reachable(person_id, depth) AS (
    SELECT friend_id, 1 FROM friendships WHERE person_id = $1
  UNION
    SELECT f.friend_id, r.depth + 1
    FROM reachable r JOIN friendships f ON f.person_id = r.person_id
    WHERE r.depth < 3                    -- the depth cap is doing heavy lifting
)
SELECT DISTINCT person_id FROM reachable;

Use case	Why a graph fits
Social networks	Friend-of-friend, mutual connections
Fraud detection	Rings and paths between accounts
Recommendations	"People who bought X also connected to Y"
Network and IT topology	Dependency and impact analysis
Identity and access	Transitive permission resolution
Knowledge graphs	Entities and typed relationships

Graph databases are a poor fit for the workloads relational databases excel at — bulk aggregations, reporting, and high-volume simple CRUD — and they are usually harder to shard, because partitioning a graph inevitably cuts edges and turns local traversals into network calls. The common architecture is therefore a graph store alongside a primary store, holding the relationship data only.

Key Takeaways

• Graph databases make traversal the primitive via index-free adjacency.
• Traversal cost scales with the neighbourhood visited, not with total dataset size.
• Multi-hop questions that need recursive self-joins in SQL are the signal to consider one.
• They shard poorly and are weak at aggregation — usually a companion store, not the primary.

🧪 Practice

1. Model a movie recommendation domain as nodes and typed edges, then write the traversal for "movies liked by people who liked what I liked".
2. Write the SQL for a 4-hop friend query and explain where the cost comes from.
3. Interview: Why are graph databases hard to shard? (Hint: think about what a traversal does when the next node lives on another machine.)

Time-Series Databases

Metrics, sensor readings, financial ticks, and application telemetry share an unusual shape: writes are almost exclusively appends of recent timestamps, records are small and numerous, data is queried in time ranges with aggregation, and old data becomes less valuable with age. A general-purpose database handles this badly — indexes bloat, tables grow without bound, and avg() over a billion rows is slow. Time-series databases exploit every one of those properties.

Three optimizations do most of the work. Time-based partitioning (chunks per hour or day) means a query for the last hour touches one chunk and dropping old data is a chunk drop rather than a mass delete. Columnar, delta-encoded storage compresses ruthlessly, because consecutive timestamps differ by a constant and consecutive values differ slightly. And automatic downsampling replaces old raw points with pre-aggregated rollups.

 WHY TIME SERIES COMPRESSES SO WELL

  timestamps  1710000000, 1710000010, 1710000020, 1710000030
  delta                +10,        +10,        +10       -> store "10" once
  values      22.14, 22.15, 22.13, 22.16
  XOR of floats -> few changed bits -> a handful of bits per point

  Real-world result: 10-20x smaller than row-oriented storage for the same data.

 RETENTION AS A LIFECYCLE, NOT A DELETE JOB

  0-7 days     raw, 1-second resolution      (fast queries, high volume)
  7-90 days    1-minute rollups              (100x smaller)
  90d-2 years  1-hour rollups                (another 60x smaller)
  > 2 years    dropped, or archived to object storage

-- TimescaleDB: a hypertable is time-partitioned automatically.
CREATE TABLE readings (
    time     TIMESTAMPTZ NOT NULL,
    sensor_id INT        NOT NULL,
    value    DOUBLE PRECISION
);
SELECT create_hypertable('readings', 'time', chunk_time_interval => INTERVAL '1 day');

-- A continuous aggregate: the rollup is maintained incrementally, not
-- recomputed on every query.
CREATE MATERIALIZED VIEW readings_hourly
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 hour', time) AS bucket,
       sensor_id,
       avg(value) AS avg_value,
       max(value) AS max_value
FROM readings
GROUP BY bucket, sensor_id;

-- Retention becomes policy, not a nightly DELETE that bloats the table.
SELECT add_retention_policy('readings', INTERVAL '7 days');
SELECT add_compression_policy('readings', INTERVAL '1 day');

Capability	Time-series DB	General-purpose DB
Write pattern	Optimized for timestamp appends	General; index maintenance cost
Compression	10-20x via delta/XOR encoding	Modest, page-level
Time-range queries	Partition pruning, near-constant	Index scan over a growing table
Downsampling	Built-in continuous aggregates	Manual cron jobs
Retention	Drop a chunk (instant)	DELETE (slow, bloats, vacuums)
Late-arriving data	Supported but costly	Same as any write

The main gotchas are cardinality and out-of-order writes. Every distinct combination of tag values creates a separate series, so putting a user id or a request id in a tag can create millions of series and exhaust memory — this is the single most common way teams take down a metrics system. And data arriving long after its timestamp forces old chunks to be rewritten, which is expensive.

Key Takeaways

• Time-series stores exploit append-mostly writes, time-range queries, and declining data value.
• Time partitioning makes range queries fast and retention a chunk drop rather than a delete.
• Delta and XOR encoding give 10-20x compression on timestamps and slowly changing values.
• Guard tag cardinality — high-cardinality labels are the classic way to kill a metrics system.

🧪 Practice

1. Design a retention and downsampling policy for 50,000 sensors reporting every 10 seconds, kept usefully for two years.
2. Compute the raw storage for one year at that rate at 16 bytes per point, then estimate it with 10x compression.
3. Interview: A team adds request_id as a metric label and the system falls over. Explain precisely why. (Hint: think about what a unique label value creates, and how many of them exist.)

Search Engines and Inverted Indexes

WHERE description LIKE '%wireless headphones%' cannot use a B-tree index — the leading wildcard makes ordering useless — so it scans every row. Beyond that, it does not do what users mean: it will not match "headphone", will not rank results, and will not tolerate a typo. Full-text search is a different problem requiring a different index.

The core structure is the inverted index: instead of mapping documents to their words, it maps each word to the documents containing it. Finding documents with two terms becomes an intersection of two lists — fast regardless of corpus size.

 FORWARD INDEX (what a database has)   INVERTED INDEX (what search needs)

 doc1 -> "wireless bluetooth headphones"  wireless   -> [doc1, doc3]
 doc2 -> "wired headphones"               bluetooth  -> [doc1]
 doc3 -> "wireless mouse"                 headphones -> [doc1, doc2]
                                          wired      -> [doc2]
                                          mouse      -> [doc3]

 Query "wireless headphones":
   intersect [doc1,doc3] with [doc1,doc2] -> doc1
   then RANK by relevance (BM25: term rarity x frequency x field length)

Getting from raw text to that index is the analysis pipeline, and it is where search quality is actually determined:

 "Wireless Bluetooth Headphones!"
    -> tokenize      ["Wireless","Bluetooth","Headphones"]
    -> lowercase     ["wireless","bluetooth","headphones"]
    -> stop words    (remove "the","a","and" — not applicable here)
    -> stemming      ["wireless","bluetooth","headphon"]   <- so "headphone"
    -> synonyms      ["wireless","bt","bluetooth",...]        also matches
    -> index

 The SAME pipeline must run on the query. A mismatch between index-time and
 query-time analysis is the most common cause of "why does nothing match?".

{
  "query": {
    "bool": {
      "must":   [{ "multi_match": {
                     "query": "wireless headphones",
                     "fields": ["title^3", "description"],
                     "fuzziness": "AUTO" }}],
      "filter": [{ "term":  { "in_stock": true }},
                 { "range": { "price": { "lte": 200 }}}]
    }
  },
  "aggs": { "by_brand": { "terms": { "field": "brand.keyword" }}},
  "sort": ["_score"]
}

Three things in that query are what a relational database cannot easily do: title^3 boosts title matches over description matches, fuzziness tolerates typos, and aggs computes faceted counts (the "Brand (42)" filters in a storefront sidebar) in the same pass.

Capability	Search engine	Relational database
Full-text matching	Native, analyzed	LIKE scan or a bolt-on index
Relevance ranking	BM25, tunable boosts	None
Typo tolerance	Fuzzy matching	None
Faceted aggregation	Built in	GROUP BY per facet
Transactions	None	ACID
Source of truth	No — a derived index	Yes

That final row is the architectural rule: a search engine is a derived index, never the system of record. The database owns the data; a pipeline (often CDC, covered in 4.6) keeps the index in sync, and the index must be rebuildable from scratch, because eventually you will need to reindex after a mapping change.

Key Takeaways

• Inverted indexes map terms to documents, making multi-term search an intersection.
• The analysis pipeline (tokenize, lowercase, stem, synonyms) determines match quality and must be identical at index and query time.
• Search engines add ranking, fuzziness, and faceting that databases lack.
• Treat the index as derived and disposable; the database remains the source of truth.

🧪 Practice

1. Build the inverted index for three short sentences and show the intersection for a two-word query.
2. Explain why searching for "running" fails to match "run" without stemming, and what could go wrong if stemming is too aggressive.
3. Interview: How do you keep a search index in sync with a database, and what happens when they diverge? (Hint: think about the sync mechanism, how you detect drift, and what "rebuild from scratch" costs.)

Vector Databases

Keyword search matches strings. It cannot tell that "car" and "automobile" mean the same thing, or find "documents about the same topic as this one". Semantic similarity requires representing meaning numerically: an embedding model maps text, images, or audio into a vector of hundreds or thousands of dimensions, such that similar meanings land near each other. A vector database stores those vectors and answers "find the nearest ones".

The naive approach — compute the distance to every stored vector — is O(n) per query and hopeless at millions of vectors. Vector databases use approximate nearest neighbour (ANN) indexes, which accept a small loss of recall in exchange for sub-linear search.

 EMBEDDING SPACE (shown in 2D; really 384-3072 dimensions)

        "dog"  "puppy"
           *   *                      Similarity = cosine of the angle
              "cat"                   between vectors, or Euclidean distance.
                 *
                            "car"     A query is embedded with the SAME model,
                              *       then the index finds its nearest points.
                          "automobile"
                              *

 HNSW INDEX: a navigable small-world graph in layers
   layer 2   o--------------o        coarse hops across the space
   layer 1   o----o----o----o        medium hops
   layer 0   o-o-o-o-o-o-o-o-o       every vector; fine-grained
   Search descends layer by layer: O(log n) instead of O(n).

# The standard retrieval-augmented pattern: embed, search, filter, use.
query_vec = embed_model.encode("How do I reset my password?")   # SAME model
                                                                # used at index time

results = index.search(
    vector=query_vec,
    top_k=5,
    filter={"tenant_id": 42, "doc_type": "help_article"},  # pre-filter: correctness
    metric="cosine",                                        # AND it shrinks the search
)
for r in results:
    print(r.score, r.metadata["title"])

# Two rules that decide whether this works in production:
#  1. Query and documents MUST use the same embedding model and version.
#     Changing models means re-embedding the entire corpus — plan for it.
#  2. Filters must be applied by the index, not after retrieval. Fetching
#     top-5 then filtering by tenant can return zero results for a tenant
#     whose documents were all ranked below the global top 5 — and, worse,
#     leaks nothing but also finds nothing.

Index type	Build cost	Query speed	Recall	Memory	Notes
Flat (exact)	None	O(n)	100%	Low	Correct; fine below ~100k
IVF	Medium	Fast	Tunable	Medium	Cluster, then search a few
HNSW	High	Very fast	High	High	The common default
PQ / IVF-PQ	High	Very fast	Lower	Very low	Compresses vectors; large scale

Vectors are large: 1 million vectors at 1,536 dimensions of float32 is about 6 GB before index overhead, so memory is usually the binding constraint. And because pure semantic search can miss exact identifiers ("error code E4021"), the strongest production setups use hybrid search — combining BM25 keyword scores with vector similarity — rather than either alone.

Key Takeaways

• Embeddings turn meaning into coordinates; vector databases find nearest neighbours in that space.
• ANN indexes such as HNSW trade a little recall for sub-linear search.
• Query and corpus must share an embedding model; changing it means re-embedding everything.
• Filter inside the index, mind the memory footprint, and prefer hybrid keyword plus vector search.

🧪 Practice

1. Compute the memory for 5 million 768-dimension float32 vectors, then with 8-bit quantization.
2. Explain why post-filtering search results by tenant is both a correctness and a relevance problem.
3. Interview: When does keyword search beat vector search? (Hint: think about exact identifiers, rare tokens, and what an embedding blurs together.)

Polyglot Persistence

Each store in this subchapter is excellent at one thing and mediocre or hopeless at others. Polyglot persistence is the deliberate use of several stores in one system, each holding the data whose access pattern it serves best — rather than forcing everything into one database and accepting the worst fit everywhere.

The benefit is obvious; the cost is what teams underestimate. Every additional store is another thing to operate, back up, secure, monitor, upgrade, and staff expertise for — and, most importantly, another copy of data that can drift out of sync with the others.

 ONE SYSTEM, SEVERAL STORES, ONE SOURCE OF TRUTH

                        writes
                          |
                          v
                   [ PostgreSQL ]  <- SOURCE OF TRUTH: orders, users, payments
                          |
                    change stream (CDC, section 4.6)
              +-----------+-----------+-------------+
              v           v           v             v
        [Elasticsearch] [Redis]  [ClickHouse]  [vector DB]
         product search  cache    analytics     semantic search
              |           |           |             |
         derived      derived     derived       derived
       (rebuildable) (expendable) (rebuildable) (rebuildable)

 The rule that makes this safe: exactly ONE store is authoritative for any
 given fact. Everything else is derived and must be rebuildable from it.

# The anti-pattern: dual writes. Two writes, no transaction between them.
def create_product_bad(product, db, search):
    db.insert(product)              # succeeds
    search.index(product)           # network error -> permanently inconsistent
    # There is no rollback for the first write once the second fails, and a
    # retry can double-apply. Dual writes always drift, eventually.

# The pattern: write once, derive asynchronously from the change stream.
def create_product(product, db):
    with db.transaction():
        db.insert("products", product)
        db.insert("outbox", {                       # same transaction: atomic
            "event": "product.created",
            "payload": product.to_json(),
        })
    # A separate relay publishes outbox rows to the stream; consumers update
    # the search index, the cache, and the analytics store. Each consumer is
    # idempotent and can be replayed from scratch if it falls behind or breaks.

Store	Holds	Rebuildable from source?
Relational	Transactional entities	Source of truth
Key-value cache	Hot lookups, sessions	Yes (or expendable)
Search index	Text and facets	Yes
Analytics/columnar	Aggregations, reporting	Yes
Time-series	Metrics and telemetry	Usually its own source
Vector	Embeddings	Yes (re-embed)
Object storage	Blobs	Source of truth for blobs

Practical discipline: start with one database. A single relational store with JSON columns, full-text search, and array types covers a surprising amount of ground, and PostgreSQL in particular can serve as document store, queue, search engine, and time-series database well past the point most teams assume. Add a specialized store when you have a measured problem the current one cannot solve — and when you do, be explicit about which store owns the truth and how the derived one is rebuilt.

Key Takeaways

• Polyglot persistence matches each access pattern to the store that serves it best.
• Every extra store adds operational burden and another copy that can drift.
• Exactly one store owns each fact; all others are derived and must be rebuildable.
• Avoid dual writes — derive from a change stream using the outbox pattern.
• Start with one database and add stores only for measured problems.

🧪 Practice

1. For an e-commerce platform, list which store you would use for each of: orders, product search, session state, recommendation embeddings, dashboards.
2. Explain the dual-write problem with a concrete interleaving and show how the outbox pattern removes it.
3. Interview: A team proposes six databases for a new product. What questions do you ask? (Hint: ask what each is measured to solve, who operates it at 3 a.m., and which one owns the truth.)

<a id="44-replication"></a>

4.4 Replication

Replication keeps copies of the same data on several machines, which is how systems survive hardware failure, serve reads at scale, and place data near users — and every scheme is a different answer to "what happens when the copies disagree?".

Leader-Follower Replication

One machine holding your data is one machine away from losing it, and one machine serving reads is a hard ceiling on read capacity. Replication solves both, and the simplest scheme — one leader, many followers — solves them with the least conceptual difficulty, which is why it is the default in PostgreSQL, MySQL, MongoDB, and most managed database services.

The rules are minimal: all writes go to the leader, the leader streams its change log to followers, and followers apply those changes in the same order and serve reads. Because there is exactly one writer, there are no write conflicts to resolve — the entire class of problems that makes the next two topics hard simply does not arise.

                       writes
                          |
                          v
                     [  LEADER  ]
                      /    |    \        replication stream (the WAL, 4.1)
                     v     v     v
              [follower][follower][follower]
                  ^         ^         ^
                  |         |         |
                       reads (scaled horizontally)

 Writes: capped by ONE machine.       Reads: scale by adding followers.
 Followers are always slightly behind — that lag has its own topic below.

# Read/write splitting: the application must route deliberately, because the
# two connections have different capabilities and different guarantees.
class Database:
    def __init__(self, leader_dsn, follower_dsns):
        self.leader    = connect(leader_dsn)
        self.followers = [connect(d) for d in follower_dsns]

    def write(self, sql, params):
        return self.leader.execute(sql, params)        # ALL writes, one place

    def read(self, sql, params, consistent=False):
        # `consistent=True` forces the leader: use it when the caller cannot
        # tolerate stale data (right after its own write, or for a balance check).
        conn = self.leader if consistent else random.choice(self.followers)
        return conn.execute(sql, params)

# The failure mode to design for: a follower that has fallen far behind is
# still "up" and still answering. Route on lag, not just on liveness.
def healthy_followers(followers, max_lag_seconds=5):
    return [f for f in followers
            if f.replication_lag_seconds() < max_lag_seconds] or [LEADER]

Replication can be shipped at three levels, and the choice has real consequences:

Method	What is shipped	Pros	Cons
Statement-based	The SQL text	Compact	Non-deterministic functions (now(), random()) diverge
Write-ahead log	Physical page changes	Exact, efficient	Followers must match the version and layout byte-for-byte
Logical/row-based	Logical row changes	Version-tolerant; enables CDC and selective replication	Larger stream

Property	What leader-follower gives you
Read scaling	Linear with follower count
Write scaling	None — one leader is the ceiling (that is what sharding is for)
Consistency	Strong on the leader; eventual on followers
Failure of a follower	Harmless; route around it
Failure of the leader	Writes stop until failover completes (see Failover and Split-Brain)

Key Takeaways

• One leader takes all writes; followers replay the log and serve reads.
• A single writer eliminates write conflicts entirely — the scheme's main virtue.
• Read capacity scales with followers; write capacity does not scale at all.
• Route reads on replication lag, not just on whether the follower responds.

🧪 Practice

1. Design read/write routing for an app where the checkout path must never read stale data but the catalogue may.
2. Explain why statement-based replication corrupts a table containing created_at DEFAULT now().
3. Interview: You add five followers and write latency gets worse. What could explain that? (Hint: consider what the leader now does on every commit, and what a struggling follower does to it.)

Multi-Leader Replication

A single leader has two limitations that no amount of follower-adding fixes: all writes must reach one machine, and if users are spread across continents, half of them pay a cross-ocean round trip on every write. Multi-leader replication allows several nodes to accept writes and replicate to each other — typically one leader per region, or one per datacenter.

The benefit is local write latency and survival of a whole region. The cost is severe and unavoidable: two leaders can accept conflicting writes to the same row before either learns of the other, so conflict resolution stops being a theoretical concern and becomes a feature you must design.

 MULTI-LEADER ACROSS REGIONS

  [EU users] -> [EU leader] <====async replication====> [US leader] <- [US users]
                     |                                        |
                [EU followers]                          [US followers]

  Local writes: ~5 ms instead of ~90 ms. Region loss: survivable.

 THE CONFLICT

  t=0  EU: UPDATE users SET name='Ada L.'   |  US: UPDATE users SET name='A. Lovelace'
  t=1  both commit locally, both succeed    |
  t=2  replication crosses; each node now sees a change it did not expect.
       Which one is correct? The database cannot know. YOU must decide.

# Conflict resolution strategies, weakest to strongest.

# 1. LAST WRITE WINS: simple, and silently destroys data. Also depends on
#    clocks agreeing across regions, which they do not (Chapter 8.3).
def lww(a, b):
    return a if a.timestamp > b.timestamp else b       # the loser is just gone

# 2. APPLICATION-DEFINED MERGE: correct when the domain has a merge rule.
def merge_cart(a, b):
    items = {}
    for item in a.items + b.items:
        # Union with max quantity: adding from two devices keeps both additions.
        items[item.product_id] = max(items.get(item.product_id, 0), item.qty)
    return Cart(items)

# 3. CONFLICT-FREE DATA TYPES (CRDTs, Chapter 8.5): structure the data so
#    concurrent updates merge deterministically with no coordination.
#    A counter becomes per-node counters that are summed:
counter = {"eu": 12, "us": 7}                 # each region increments its own key
total = sum(counter.values())                 # 19 — no conflict is possible

# 4. AVOID CONFLICTS ENTIRELY: partition write ownership so two leaders never
#    touch the same row. Route every user's writes to their home region.
def leader_for(user):
    return LEADERS[user.home_region]           # the strategy most systems should use

Topology	Structure	Trade-off
Circular	Each node replicates to the next	Cheap; one node down breaks the ring
Star	One hub relays to all	Simple; the hub is a single point of failure
All-to-all	Every node to every other	Resilient; messages can arrive out of order

Multi-leader is genuinely necessary in a few situations: multi-region write locality, offline-capable clients (each device is effectively a leader that syncs later), and collaborative editing. Outside those, the complexity rarely pays — and the most common successful pattern is the fourth one above: partition ownership so conflicts cannot occur, which gives local writes without conflict resolution.

Key Takeaways

• Multiple leaders give local write latency and regional fault tolerance.
• Concurrent conflicting writes become unavoidable and must be resolved explicitly.
• Last-write-wins is simple and silently loses data; prefer domain merges or CRDTs.
• The best strategy is usually to partition write ownership so conflicts never arise.

🧪 Practice

1. Design conflict resolution for a shopping cart edited on two devices while offline.
2. Explain why last-write-wins is unsafe across regions even with NTP running.
3. Interview: When is multi-leader replication the right choice? (Hint: name a requirement that a single leader cannot satisfy at all, not merely one it satisfies slowly.)

Leaderless Replication and Quorums

Both previous schemes have a designated writer, so a leader failure means writes stop until failover finishes. Leaderless replication removes the role entirely: a client (or a coordinator on its behalf) writes to several replicas directly and reads from several replicas, using overlap to guarantee it sees current data. Dynamo, Cassandra, and Riak popularized this design.

The mechanism is the quorum condition. With N replicas, W acknowledgements required for a write, and R responses required for a read, if W + R > N then any read set and any write set must share at least one replica — so a read is guaranteed to see the latest acknowledged write.

 N = 3, W = 2, R = 2   (W + R = 4 > 3)

 WRITE v2 to replicas A and B (2 acks: done)      A[v2]  B[v2]  C[v1]
 READ from B and C                                        ^^^^^^^^^^^
   -> responses: v2 (from B) and v1 (from C)
   -> the overlap guarantees v2 is present; newest version wins
   -> read repair: C is updated to v2 in the background

 Tuning the same N for different goals:
   W=1, R=1  fastest, no overlap guarantee -> eventual consistency
   W=3, R=1  fast reads, writes fail if ANY replica is down
   W=1, R=3  fast writes, slow reads, writes survive 2 failures
   W=2, R=2  balanced: survives one node down for both reads and writes

def quorum_write(key, value, replicas, W):
    version = hlc_now()                        # a version, not a wall clock
    acks, errors = 0, []
    for r in replicas:                         # in practice: send in parallel
        try:
            r.put(key, value, version); acks += 1
        except Exception as e:
            errors.append(e)
        if acks >= W:
            return "ok"                        # return as soon as W agree
    raise QuorumNotMet(f"{acks}/{W} acks", errors)
    # Note: replicas that DID succeed keep the write. A failed quorum write is
    # not rolled back — the value may still surface later. That is the biggest
    # semantic difference from a transactional database.

def quorum_read(key, replicas, R):
    responses = [r.get(key) for r in replicas[:R]]
    newest = max(responses, key=lambda x: x.version)
    for r, resp in zip(replicas[:R], responses):
        if resp.version < newest.version:
            r.put(key, newest.value, newest.version)   # read repair, in background
    return newest.value

Because failed writes are not rolled back and replicas can miss updates while briefly unavailable, leaderless systems need two background mechanisms: read repair (fix stale replicas when a read notices) and anti-entropy (a background process comparing replicas via Merkle trees and reconciling differences). Some systems also use hinted handoff, where a healthy node temporarily stores writes destined for an unavailable peer.

Property	Leaderless
Write availability	High — no failover; any W replicas suffice
Consistency	Tunable per operation via W and R
Conflicts	Possible; resolved by version vectors or LWW
Operational model	Symmetric nodes, easy to add capacity
Weakness	No transactions; quorum reads cost latency; sloppy quorums weaken the guarantee

The important caveat: W + R > N guarantees overlap only under a strict quorum. Many systems default to sloppy quorums, where writes may be accepted by any W reachable nodes rather than the W that actually own the key — which raises availability and quietly removes the overlap guarantee.

Key Takeaways

• Leaderless replication removes failover entirely: any W replicas can accept a write.
• W + R > N guarantees a read overlaps the latest write — the core quorum condition.
• Tune W and R per operation to trade latency, availability, and consistency.
• Failed quorum writes are not rolled back, and sloppy quorums break the overlap guarantee.

🧪 Practice

1. For N = 5, list every (W, R) pair satisfying the quorum condition and rank them by write latency.
2. Explain what a client can and cannot conclude when a quorum write returns an error.
3. Interview: With N=3, W=2, R=2, how many node failures can the system tolerate for writes, and for reads? (Hint: count how many must respond and how many remain.)

Synchronous vs. Asynchronous Replication

Every replication scheme faces one question at commit time: does the leader wait for replicas before telling the client "committed"? The answer sets the system's durability guarantee, its write latency, and its behaviour when a replica is slow — and it is the single most consequential replication setting.

Asynchronous: the leader commits locally and acknowledges immediately; replicas catch up afterwards. Fast, and a replica being slow or down does not affect writes. But if the leader dies before shipping recent commits, those acknowledged writes are permanently lost.

Synchronous: the leader waits for at least one replica to confirm before acknowledging. No acknowledged write is lost if the leader dies. But write latency now includes a network round trip, and a stalled replica stalls all writes.

 ASYNCHRONOUS                          SYNCHRONOUS

  client -> leader: write               client -> leader: write
  leader: fsync local WAL               leader: fsync local WAL
  leader -> client: OK      (~2 ms)     leader -> replica: ship, wait for ack
  leader -> replica: ship  (later)      replica: fsync, ack        (+1 RTT)
                                        leader -> client: OK       (~4 ms same AZ,
  Leader dies now -> recent commits                       ~90 ms cross-region)
  are gone. RPO > 0.
                                        Leader dies now -> replica has everything.
                                        Replica hangs  -> writes hang too.

-- PostgreSQL: the durability dial, from fastest to safest.
SET synchronous_commit = off;          -- not even local fsync: fast, lossy
SET synchronous_commit = local;        -- local fsync only: async replication
SET synchronous_commit = on;           -- wait for the synchronous standby

-- Semi-synchronous: the practical middle ground for multi-replica setups.
-- Wait for ANY 1 of 3 named standbys — durability without depending on a
-- specific replica being healthy.
ALTER SYSTEM SET synchronous_standby_names = 'ANY 1 (replica_a, replica_b, replica_c)';

-- Per-transaction control: pay for durability only where it matters.
BEGIN;
SET LOCAL synchronous_commit = on;     -- this payment must not be lost
INSERT INTO payments ...;
COMMIT;

Aspect	Asynchronous	Synchronous
Write latency	Local commit only	Plus one replica round trip
Data loss on leader failure	Possible (RPO > 0)	None (RPO = 0)
Slow/failed replica	No impact on writes	Writes block or degrade
Cross-region cost	Free	Adds full inter-region latency
Typical use	Read replicas, analytics, DR copies	Financial and regulated data

The pattern most production systems settle on is semi-synchronous: one synchronous replica in a different availability zone (cheap — sub-millisecond, so the latency cost is small) plus asynchronous replicas elsewhere for read scaling and cross-region disaster recovery. That buys RPO = 0 for facility failure without paying cross-region latency on every commit.

It is also worth being explicit about what synchronous replication does not guarantee: it means the replica has the data, not that the replica has applied it, and not that a failover will actually happen correctly. That last part is the next topic.

Key Takeaways

• Asynchronous replication is fast and can lose acknowledged writes when the leader fails.
• Synchronous replication guarantees RPO = 0 at the cost of a round trip per commit and coupling to replica health.
• Semi-synchronous (wait for any one of several) gives durability without depending on one replica.
• Set the durability level per data class — payments and analytics do not need the same guarantee.

🧪 Practice

1. Compute added write latency for a synchronous replica in the same AZ (0.5 ms RTT) versus another region (80 ms RTT), at 200 writes per transaction.
2. Explain what "RPO = 0" means concretely to a user who just clicked Pay.
3. Interview: Your synchronous replica has a disk problem and writes have stopped cluster-wide. What are your options right then? (Hint: consider what you can change quickly and what you are giving up by changing it.)

Replication Lag and Read-Your-Writes

Asynchronous replication means followers are always slightly behind — usually milliseconds, sometimes minutes under load or during a bulk operation. That delay is replication lag, and it produces a family of user-visible bugs that are notoriously hard to reproduce, because they depend on which replica happened to answer.

The three classic anomalies each break a different intuition:

 1. READ-YOUR-WRITES violated
    User edits their profile -> write goes to the leader
    User reloads             -> read hits a lagging follower
    "My change didn't save!"  (it did; they just cannot see it)

 2. MONOTONIC READS violated
    Read 1 -> follower A (up to date)   sees the new comment
    Read 2 -> follower B (behind)       comment is gone
    Time appears to run backwards.

 3. CONSISTENT PREFIX violated
    Q: "How far is the restaurant?"  written at t=1
    A: "About 2 kilometres"          written at t=2
    A replica applies them out of order -> the answer appears before the question.

# Fix 1: route the user's own reads to the leader for a short window after a write.
def read_profile(user_id, session, leader, followers):
    if session.wrote_recently(user_id, within_ms=1000):
        return leader.get_profile(user_id)          # read-your-writes guaranteed
    return pick(followers).get_profile(user_id)

# Fix 2: LSN pinning — precise instead of time-based. Capture the write position,
# then require any replica serving this user to have replayed at least that far.
def write_profile(user_id, data, leader, session):
    lsn = leader.execute_returning_lsn("UPDATE profiles ...", data)
    session.set_min_lsn(user_id, lsn)               # remember the exact position
    return lsn

def read_profile_pinned(user_id, session, leader, followers):
    need = session.get_min_lsn(user_id)
    caught_up = [f for f in followers if f.replay_lsn() >= need]
    return (pick(caught_up) if caught_up else leader).get_profile(user_id)

# Fix 3: sticky routing — pin one user's session to one replica for its duration.
#        Gives monotonic reads cheaply, but concentrates load unevenly.

Anomaly	Guarantee needed	Cheapest practical fix
Missing own write	Read-your-writes	Route to leader after write, or LSN pinning
Data disappearing	Monotonic reads	Sticky routing to one replica per session
Effect before cause	Consistent prefix	Keep causally related writes in one partition

Two operational habits matter as much as the code. Alert on lag, not just on liveness — a follower 90 seconds behind is a correctness problem while reporting perfect health. And remove lagging replicas from the read pool automatically, since the common cause of a lag spike (a long transaction, a bulk UPDATE, a vacuum storm) also makes the stale data most damaging.

Key Takeaways

• Replication lag causes read-your-writes, monotonic read, and consistent prefix violations.
• Route a user's reads to the leader briefly after their write, or pin to a replication position.
• Sticky per-session routing gives monotonic reads at the cost of uneven load.
• Monitor lag as a first-class metric and eject lagging replicas from the read pool.

🧪 Practice

1. Reproduce a read-your-writes violation on paper with a 200 ms lag and a client that reads 50 ms after writing.
2. Implement LSN-based routing and describe its behaviour when every follower is behind.
3. Interview: Users report intermittently missing data after saving, but it is always there on a later refresh. Diagnose. (Hint: ask what differs between the failing and succeeding request — it is not the code path.)

Failover and Split-Brain

Replication protects data; failover is the process that turns a replica into the new leader when the old one fails. It is the part of the system that runs only during incidents, which is exactly why it is the least tested and most dangerous component — and why the worst database outages are usually failover going wrong rather than the original hardware failure.

Failover has three steps and each has a failure mode:

 1. DETECT   Is the leader really dead, or just slow, or is the network broken
             between us and it? This is undecidable in general (Chapter 8.1).
             Too eager -> unnecessary failover. Too slow -> long outage.

 2. ELECT    Choose a new leader — ideally the most up-to-date replica.
             Choosing a lagging replica silently discards recent commits.

 3. REDIRECT Point clients, proxies, and the old leader at the new topology.
             Miss any of them and you have two leaders.

Split-brain is the catastrophe: the old leader was never actually dead (just partitioned or paused), a new leader was promoted, and now two nodes accept writes on the same data. Both sets of writes are "committed", they diverge, and reconciliation afterwards is manual and lossy.

 SPLIT-BRAIN

   before:          [L] <- clients
                   /   \
                 [F1]  [F2]

   network splits: [L] <- some clients still reach it and keep writing
                   ~~~~~~~~ partition ~~~~~~~~
                   [F1 -> promoted to leader] <- other clients write here
                   [F2]

   Both accept writes. Both are internally consistent. Together they are
   irreconcilable: two different "true" histories of the same rows.

 THE THREE DEFENCES

 1. QUORUM-BASED ELECTION
    Only a majority partition may elect a leader. With 3 nodes, a 1-node
    partition can never win — which is why an ODD number of nodes matters,
    and why 2-node clusters cannot safely fail over at all.

 2. FENCING (STONITH)
    Before promoting, forcibly disable the old leader: revoke its storage
    lease, block it at the network layer, or power it off. "Shoot The Other
    Node In The Head" is a real and deliberately blunt term.

 3. LEASES / EPOCH NUMBERS
    A leader holds a time-bounded lease and must renew it. Each promotion
    increments an epoch; storage and clients reject writes carrying an old
    epoch, so a revived zombie leader is refused rather than obeyed.

# Epoch fencing: the storage layer, not the application, enforces single-writer.
class FencedStorage:
    def __init__(self):
        self.epoch = 0

    def promote(self, node):
        self.epoch += 1                       # every promotion invalidates the past
        return self.epoch

    def write(self, node, epoch, data):
        if epoch < self.epoch:
            # The old leader woke up and tried to write with a stale epoch.
            raise FencedOut(f"epoch {epoch} < current {self.epoch}")
        return self._apply(data)
# This is why cloud block storage and consensus systems expose fencing tokens:
# correctness cannot rest on the old leader noticing it was replaced.

Decision	Trade-off
Detection timeout	Short: false failovers. Long: extended outages
Automatic vs. manual	Automatic is fast and can be wrong; manual is slow and deliberate
Promote most-current replica	Minimizes data loss; may not be available
Number of nodes	Odd numbers only; three minimum for a real quorum
Old leader's unreplicated writes	Usually discarded — and clients were told they committed

That last row is the honest summary of asynchronous failover: some acknowledged writes are thrown away. If that is unacceptable, you need synchronous replication, and you should rehearse failover regularly (Chapter 9.4) rather than discovering its behaviour during an incident.

Key Takeaways

• Failover is detect, elect, redirect — and each step can fail in its own way.
• Split-brain means two leaders accepting divergent writes; reconciliation is manual and lossy.
• Defend with majority-quorum election, fencing of the old leader, and epoch/lease tokens enforced by storage.
• Asynchronous failover discards unreplicated commits that clients were told had succeeded — rehearse it before it happens for real.

🧪 Practice

1. Explain why a two-node cluster cannot safely perform automatic failover, and what minimum change fixes it.
2. Design a fencing mechanism for a service whose leader writes to shared object storage.
3. Interview: Your failover promoted a replica that was 30 seconds behind. What are the consequences and how do you prevent a repeat? (Hint: name what those 30 seconds contained and who was told it was safe.)

<a id="45-partitioning-and-sharding"></a>

4.5 Partitioning and Sharding

Replication copies the same data everywhere; partitioning splits different data across machines, which is the only way to scale writes and datasets past what one machine can hold.

Vertical vs. Horizontal Partitioning

When one database can no longer hold the data or absorb the write rate, you split it. There are exactly two directions to cut, and they solve different problems — confusing them is a common source of wasted migrations.

Vertical partitioning splits by column or by table: move some columns to a separate table, or some tables to a separate database. It is the natural first step, it often follows service boundaries, and it is straightforward — but it has a ceiling, because eventually one table alone is too big.

Horizontal partitioning (sharding) splits by row: the same schema lives on many machines, each holding a disjoint subset of rows. This is the one that scales without limit, and it is also the one that costs you joins, transactions, and operational simplicity.

 ORIGINAL TABLE
   users(id, email, name, bio, avatar_blob, last_login, preferences_json)

 VERTICAL (split columns by access pattern)
   users_core(id, email, name, last_login)        <- hot: read constantly, small
   users_profile(id, bio, avatar_url, prefs)      <- cold: read rarely, large
   Effect: the hot table's rows are smaller, so more fit per page and the
   buffer pool holds more of it. Same machine, better cache behaviour.

 HORIZONTAL (split rows across machines)
   shard 0: users where id % 4 = 0     shard 2: users where id % 4 = 2
   shard 1: users where id % 4 = 1     shard 3: users where id % 4 = 3
   Effect: 4x the write capacity, 4x the storage — and no cross-shard joins.

-- Native table partitioning: horizontal splitting WITHIN one database.
-- Not distribution, but it delivers partition pruning and instant drops.
CREATE TABLE events (
    id         BIGSERIAL,
    occurred_at TIMESTAMPTZ NOT NULL,
    payload    JSONB
) PARTITION BY RANGE (occurred_at);

CREATE TABLE events_2026_03 PARTITION OF events
    FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');

-- The query planner touches only the relevant partition:
SELECT * FROM events WHERE occurred_at >= '2026-03-15';   -- one partition scanned

-- Retention becomes instant instead of an hours-long DELETE:
DROP TABLE events_2026_01;                                 -- metadata operation

Dimension	Vertical partitioning	Horizontal sharding
Splits by	Columns or tables	Rows
Scales	Until one table is too big	Effectively without limit
Joins	Preserved within each side	Lost across shards
Transactions	Preserved within each side	Lost across shards
Operational cost	Low	High
Typical trigger	Mixed hot/cold columns; service extraction	Write throughput or dataset size

The order to attempt these matters, because each step is cheaper than the next: optimize queries and indexes, then add read replicas, then cache, then partition tables within one database, then vertically split, and only then shard. Sharding is the last resort not because it does not work, but because it permanently changes what your application may assume — no cross-shard joins, no global transactions, no global AUTO_INCREMENT, no simple ORDER BY ... LIMIT.

Key Takeaways

• Vertical partitioning splits columns or tables; horizontal sharding splits rows across machines.
• Only horizontal sharding scales writes and dataset size without limit.
• Native table partitioning gives pruning and instant retention drops without distribution.
• Exhaust indexing, replicas, caching, and partitioning before sharding — it is a one-way door.

🧪 Practice

1. Identify hot and cold columns in a users table and design the vertical split.
2. Write a monthly range partition scheme for an events table with a 13-month retention policy.
3. Interview: A team wants to shard because "the database is slow". What do you check first? (Hint: sharding fixes exactly two problems — name them, and confirm the team has one.)

Range-Based Sharding

Once you have decided to shard, you must decide how rows map to shards. Range-based sharding assigns contiguous key ranges to shards: A-F on shard 1, G-M on shard 2, and so on. It is the scheme used by HBase, Bigtable, and MongoDB's ranged sharding, and its defining property is that sorted order is preserved.

That property is worth a lot. Range queries — "all orders between these dates", "all usernames starting with 'Sm'" — touch only the shards covering that range, often just one. No other scheme gives you that.

 RANGE SHARDING BY DATE

  shard 0: 2026-01-01 .. 2026-03-31
  shard 1: 2026-04-01 .. 2026-06-30
  shard 2: 2026-07-01 .. 2026-09-30
  shard 3: 2026-10-01 .. open                   <- ALL current writes land here

  Range query "last 7 days"  -> 1 shard. Excellent.
  Write distribution         -> 1 shard doing 100% of the work. Catastrophic.

That diagram is the whole trade. Range sharding on a monotonically increasing key (a timestamp, an auto-increment id) creates a permanent hotspot: every write goes to the last shard while the others sit idle. The classic mitigation is a compound key that puts a high-cardinality attribute first:

# BAD: pure time ordering -> all writes hit the newest shard
shard_key = event.timestamp

# BETTER: distribute by entity, keep time ordering WITHIN each entity
shard_key = (event.device_id, event.timestamp)
# Writes spread across devices; "last hour for device X" is still one shard's
# contiguous range. This is the same idea as Cassandra's partition + clustering
# key from 4.3.

# The routing function must handle boundaries carefully.
RANGES = [                       # (inclusive_start, exclusive_end, shard)
    (None,        "f",  0),      # None = unbounded below
    ("f",         "m",  1),
    ("m",         "t",  2),
    ("t",         None, 3),      # None = unbounded above
]

def shard_for(key):
    for start, end, shard in RANGES:
        if (start is None or key >= start) and (end is None or key < end):
            return shard
    raise KeyError(key)          # gaps in the range map are a real outage source

Property	Range sharding
Range queries	Excellent — contiguous keys stay together
Write distribution	Poor for sequential keys; excellent for random ones
Rebalancing	Natural: split a hot range into two
Adding a shard	Split an existing range; move only that data
Metadata	Requires a range map that must stay authoritative
Hotspot risk	High, and structural rather than accidental

The rebalancing story is range sharding's second real advantage: when a range gets too large or too hot, the system splits it in half and moves one half elsewhere. Only that range's data moves, and it happens automatically in systems like HBase and CockroachDB.

Key Takeaways

• Range sharding assigns contiguous key ranges, preserving sort order.
• Range queries touch only the relevant shards — its decisive advantage.
• Sequential keys create a permanent write hotspot on the last shard.
• Prefix the key with a high-cardinality attribute to distribute writes while keeping local ordering.

🧪 Practice

1. Design a range shard key for IoT readings that supports "last 24 hours for one device" without hotspotting.
2. Explain what happens to a range-sharded cluster when a range map has a gap or an overlap.
3. Interview: Why does range sharding on created_at fail for a write-heavy system, and how do you fix it while keeping time-range queries fast? (Hint: the fix does not remove time from the key — it changes what comes first.)

Hash-Based Sharding

If the problem with range sharding is uneven distribution, the direct fix is to destroy the ordering deliberately. Hash-based sharding applies a hash function to the key and uses the result to select a shard, which spreads even perfectly sequential keys uniformly across the cluster.

This is the most common scheme for OLTP workloads that are dominated by single-key lookups, because it gives excellent write distribution with almost no metadata — you compute the shard rather than looking it up.

import hashlib

def shard_for(key, num_shards):
    # Use a stable hash, not Python's built-in hash(): the built-in is
    # randomized per process, so the same key would route differently on every
    # restart. This bug corrupts data silently.
    digest = hashlib.md5(str(key).encode()).digest()
    return int.from_bytes(digest[:8], "big") % num_shards

for user_id in (1, 2, 3, 1000, 1001):
    print(user_id, shard_for(user_id, 4))
# Sequential ids scatter across all 4 shards — exactly what range sharding
# could not do.

 THE COST: ORDER IS GONE

  Range sharding                     Hash sharding
   "users starting with 'Sm'"         "users starting with 'Sm'"
   -> 1 shard                         -> ALL shards (scatter-gather)

  Every non-key query becomes a fan-out to every shard, and the client must
  merge and re-sort the results. At 64 shards, one such query is 64 queries,
  and its latency is the SLOWEST shard's, not the average (Chapter 1.2).

 AND THE RESHARDING PROBLEM

  shard = hash(key) % 4      ->  shard = hash(key) % 5
  Changing the modulus remaps ALMOST EVERY KEY. Roughly 80% of the data must
  move to add one shard. This is why naive modulo hashing is a trap, and why
  consistent hashing (two topics ahead) exists.

A common intermediate solution is to fix the number of logical shards high (say 1,024) and map many logical shards onto each physical node. Growing the cluster then means moving whole logical shards between nodes without rehashing anything:

LOGICAL_SHARDS = 1024                     # chosen once, never changed

def logical_shard(key):
    return int.from_bytes(hashlib.md5(str(key).encode()).digest()[:8], "big") % LOGICAL_SHARDS

# A small, explicit map from logical shard -> physical node. Growing the cluster
# updates this map and moves the affected logical shards only.
PLACEMENT = {n: n * 4 // LOGICAL_SHARDS for n in range(LOGICAL_SHARDS)}

def physical_node(key):
    return PLACEMENT[logical_shard(key)]

Property	Hash sharding
Write distribution	Excellent and automatic
Point lookups	Excellent — compute the shard, one hop
Range queries	Poor — scatter-gather across all shards
Metadata	Minimal (or a small logical-shard map)
Adding a shard	Catastrophic with plain modulo; fine with logical shards or consistent hashing
Hotspot risk	Low, unless one key is itself hot

That last caveat matters: hashing distributes keys, not traffic. If one celebrity user's row receives 30% of all reads, hashing puts all of it on one shard just as reliably as range sharding would.

Key Takeaways

• Hashing distributes even sequential keys uniformly, solving write hotspots.
• It destroys ordering, so range queries become scatter-gather across every shard.
• Never use a process-randomized hash function for routing.
• Plain hash % N makes resharding move almost all data — use fixed logical shards or consistent hashing.

🧪 Practice

1. Compute what fraction of keys change shard when going from 4 to 5 shards with plain modulo hashing.
2. Explain why ORDER BY created_at LIMIT 10 is expensive on a hash-sharded table and how a client must implement it.
3. Interview: Hash sharding spread your data evenly but one shard is still at 90% CPU. What is happening? (Hint: even key distribution is not the same as even request distribution.)

Directory-Based Sharding

Both previous schemes compute a shard from the key, which means the mapping is fixed by a formula. Directory-based sharding stores the mapping instead: a lookup service holds an explicit table of key (or key range, or tenant) to shard, and every request consults it.

The gain is total flexibility. You can move one noisy tenant to dedicated hardware, keep European customers on European infrastructure, place a large customer on a bigger machine, and rebalance any individual entity at any time — none of which a hash function can express.

 DIRECTORY LOOKUP

  request(tenant=acme)
        |
        v
  [ directory service ]  tenant -> shard
      acme      -> shard_7   (dedicated, they are 40% of our load)
      globex    -> shard_2
      initech   -> shard_2
      eu_corp   -> shard_eu1 (data residency requirement)
        |
        v
  [ shard_7 ]

  The directory is consulted on EVERY request, so it must be:
    fast     (cached aggressively in each client)
    correct  (a stale entry routes to the wrong shard -> data appears missing)
    available (it is now on the critical path of every query)

class ShardDirectory:
    def __init__(self, store, cache_ttl=60):
        self._store = store                  # durable: Postgres, etcd, ZooKeeper
        self._cache = TTLCache(maxsize=100_000, ttl=cache_ttl)

    def shard_for(self, tenant_id):
        if tenant_id in self._cache:
            return self._cache[tenant_id]
        shard = self._store.get(f"tenant:{tenant_id}")   # durable lookup
        if shard is None:
            raise UnknownTenant(tenant_id)
        self._cache[tenant_id] = shard
        return shard

    def migrate(self, tenant_id, new_shard):
        # Migration is a controlled, per-tenant operation — the whole point of
        # this scheme. The order matters:
        #   1. copy data to the new shard while writes continue on the old
        #   2. briefly freeze writes for this tenant only
        #   3. copy the delta, then flip the directory entry
        #   4. invalidate caches (or accept up to cache_ttl of misrouting)
        #   5. keep the old copy read-only until verified, then delete
        self._store.set(f"tenant:{tenant_id}", new_shard)
        self._cache.pop(tenant_id, None)
        broadcast_invalidation(tenant_id)     # tell every client to drop its entry

Property	Directory sharding
Flexibility	Total — any entity can live anywhere
Rebalancing	Per-entity, incremental, online
Heterogeneous shards	Supported (big tenants on big machines)
Data residency	Supported (route by jurisdiction)
Extra hop	One lookup per request (mitigated by caching)
Failure mode	The directory is a critical dependency and a single point of failure
Cardinality limit	Works per-tenant; impractical per-row at billions of keys

This is the standard architecture for multi-tenant B2B SaaS, where tenants vary enormously in size and some have contractual isolation or residency requirements. It is impractical for per-row mapping at very large scale, since the directory itself becomes the thing that needs sharding.

The operational rule: cache the directory everywhere and make invalidation explicit. A stale entry does not produce an error — it produces a query against the wrong shard that returns no rows, which the application will happily interpret as "the data does not exist".

Key Takeaways

• Directory sharding stores an explicit mapping instead of computing one.
• It enables per-entity placement: dedicated shards, heterogeneous hardware, data residency.
• Migration becomes an online, per-tenant operation rather than a cluster-wide rehash.
• The directory is on every request's critical path — cache it, replicate it, and invalidate deliberately.

🧪 Practice

1. Design the directory schema and cache strategy for 50,000 tenants across 20 shards.
2. Write the step-by-step runbook for migrating one tenant with zero data loss and minimal write downtime.
3. Interview: What happens if a client caches a stale directory entry during a migration? (Hint: the dangerous case is not an error — describe what the application sees instead.)

Consistent Hashing

Plain modulo hashing has one fatal operational property: changing the number of shards remaps nearly every key. Adding a fifth node to a four-node cluster moves roughly 80% of the data, which means growing the cluster is a full migration. Consistent hashing reduces that to roughly 1/n of the keys — only the data that must move, moves.

The trick is to hash both keys and nodes onto the same circular space. A key belongs to the first node encountered walking clockwise from the key's position. Adding a node captures only the arc between it and its predecessor; removing one hands its arc to the next node clockwise. Everything else is untouched.

 THE HASH RING (positions 0 .. 2^32-1, wrapped into a circle)

                 0/2^32
                    |
          nodeC ----+---- nodeA
              \     |     /
               \    |    /          key -> hash -> walk CLOCKWISE -> first node
                \   |   /
                 \  |  /
                  nodeB

  Add nodeD between nodeA and nodeB:
    only keys in the arc (nodeA, nodeD] move — from nodeB to nodeD.
    Keys owned by nodeA and nodeC are completely unaffected.

The naive ring has a flaw: with few nodes, the arcs are wildly uneven, so one node may own three times another's data. The fix is virtual nodes — place each physical node at many positions on the ring, so the law of large numbers evens out the arcs.

import hashlib, bisect

class ConsistentHashRing:
    def __init__(self, nodes, vnodes=150):
        self._vnodes = vnodes            # more vnodes -> more even distribution
        self._ring = {}                  # ring position -> physical node
        self._sorted = []                # sorted positions, for binary search
        for n in nodes:
            self.add(n)

    def _hash(self, s):
        return int.from_bytes(hashlib.md5(s.encode()).digest()[:4], "big")

    def add(self, node):
        for i in range(self._vnodes):
            pos = self._hash(f"{node}#{i}")          # 150 positions per node
            self._ring[pos] = node
            bisect.insort(self._sorted, pos)

    def remove(self, node):
        for i in range(self._vnodes):
            pos = self._hash(f"{node}#{i}")
            del self._ring[pos]
            self._sorted.remove(pos)

    def node_for(self, key):
        pos = self._hash(key)
        idx = bisect.bisect_right(self._sorted, pos)  # first position >= key
        if idx == len(self._sorted):
            idx = 0                                   # wrap around the circle
        return self._ring[self._sorted[idx]]

    def nodes_for(self, key, replicas=3):
        """Walk clockwise past distinct physical nodes — this is how quorum
        replication in 4.4 picks which N replicas hold a key."""
        pos = self._hash(key)
        idx = bisect.bisect_right(self._sorted, pos) % len(self._sorted)
        chosen = []
        for i in range(len(self._sorted)):
            node = self._ring[self._sorted[(idx + i) % len(self._sorted)]]
            if node not in chosen:
                chosen.append(node)
                if len(chosen) == replicas:
                    break
        return chosen

 KEYS MOVED WHEN GROWING A CLUSTER (fraction of total data)

   nodes: 4 -> 5      plain modulo:  ~80%      consistent hashing:  ~20% (1/5)
   nodes: 10 -> 11    plain modulo:  ~91%      consistent hashing:  ~9%  (1/11)
   nodes: 100 -> 101  plain modulo:  ~99%      consistent hashing:  ~1%  (1/101)

 Consistent hashing moves exactly the data the new node will own — the
 theoretical minimum.

Consistent hashing is what underpins Cassandra, DynamoDB, Riak, memcached client libraries, and CDN request routing. Its main limitation is inherited from hashing in general: ordering is still destroyed, so range queries remain scatter-gather, and a single hot key is still concentrated on one node.

Key Takeaways

• Consistent hashing maps keys and nodes onto a ring; a key belongs to the next node clockwise.
• Adding or removing a node moves only about 1/n of keys instead of nearly all of them.
• Virtual nodes are required for even distribution and for smooth load transfer on failure.
• Walking the ring past distinct nodes also selects the replica set for quorum systems.

🧪 Practice

1. Implement the ring above and empirically measure the distribution across 5 nodes with 10,000 keys, at 1 vnode and at 150.
2. Explain why removing a node without virtual nodes dumps all its load on a single neighbour.
3. Interview: Consistent hashing minimizes data movement — what problem does it not solve? (Hint: think about what happens when one particular key is requested a million times per second.)

Hotspots and Skew Mitigation

Every sharding scheme assumes load spreads across keys. Reality rarely obliges. Real data follows power laws: a few products account for most sales, a few users generate most traffic, and a celebrity account has millions of followers while the median has fifty. Skew is uneven data distribution; a hotspot is uneven traffic distribution. They are different problems and need different fixes.

The distinction matters because the diagnosis differs. If one shard holds 10x the rows, that is data skew. If every shard holds the same rows but one shard's CPU is pinned, that is a traffic hotspot — and rebalancing data will not help at all.

 THREE FAILURE SHAPES

 1. DATA SKEW           shard 0: 40 GB   shard 1: 40 GB   shard 2: 400 GB
                        cause: a shard key with uneven cardinality
                        (sharding by country when 80% of users are in one)

 2. TRAFFIC HOTSPOT     shard 0: 2k QPS  shard 1: 2k QPS  shard 2: 90k QPS
                        cause: one popular key (a celebrity, a viral post)

 3. TEMPORAL HOTSPOT    all writes hit the newest shard
                        cause: a monotonically increasing shard key

# FIX 1: SALTING — split one hot key into N sub-keys so it spans shards.
HOT_KEYS = {"user:celebrity_1", "product:flash_sale"}
SALT_BUCKETS = 32

def write_key(key):
    if key in HOT_KEYS:
        return f"{key}#{random.randrange(SALT_BUCKETS)}"   # spread the writes
    return key

def read_all(key, store):
    if key in HOT_KEYS:
        # The cost of salting: reads must gather every bucket and merge.
        parts = [store.get(f"{key}#{i}") for i in range(SALT_BUCKETS)]
        return merge(parts)
    return store.get(key)

# FIX 2: CACHE THE HOT KEY. Often the best answer: a hotspot on a READ path is
# a caching problem, not a sharding problem. One popular row served from Redis
# (or from an in-process cache) never reaches the shard at all.

# FIX 3: SPLIT COUNTERS — the write-hotspot version of salting.
# Instead of one row everyone contends on:
#   UPDATE posts SET likes = likes + 1 WHERE id = 42        <- serialized
# use per-bucket rows and sum on read:
#   UPDATE like_shards SET n = n + 1 WHERE post_id = 42 AND bucket = ?
#   SELECT sum(n) FROM like_shards WHERE post_id = 42       <- 32 cheap rows

# FIX 4: A BETTER SHARD KEY — usually the real fix, and the most expensive.
# Composite keys with high-cardinality leading components distribute naturally.

Symptom	Diagnosis	Effective fix
One shard much larger	Data skew	Better shard key; split the range
One shard's CPU pinned, sizes equal	Read hotspot	Cache the hot keys; add replicas
One row contended by writers	Write hotspot	Salting or split counters
Newest shard takes all writes	Temporal hotspot	Prefix the key with high-cardinality data
Latency fine at p50, awful at p99	Occasional hotspot	Find the hot key; it is usually one entity

Two habits prevent most of this. Measure per-shard load, not cluster averages — an average hides exactly the imbalance you are looking for. And choose the shard key from the access pattern, testing it against realistic, skewed data rather than uniformly generated test data, which will make any key look perfect.

Key Takeaways

• Data skew (uneven storage) and traffic hotspots (uneven load) are different problems with different fixes.
• Real distributions follow power laws — assume skew rather than hoping for uniformity.
• Salting and split counters spread a hot key at the cost of gather-on-read.
• A read hotspot is usually a caching problem; a write hotspot usually needs a key change.

🧪 Practice

1. A social platform shards by user_id. One account has 200 million followers. Identify every hotspot this creates and propose fixes.
2. Implement a split counter for post likes, including the read path and its cost.
3. Interview: Your shards hold equal data but one is at 95% CPU. Walk through your investigation. (Hint: equal storage rules out one whole category — what does that leave, and how do you find the specific key?)

Resharding and Rebalancing

The number of shards you chose at the start will eventually be wrong. Data grows, traffic grows, and hardware changes — so at some point you must add shards and redistribute existing data. Doing that while the system serves live traffic, without data loss or downtime, is one of the more demanding operations in infrastructure engineering.

The reason it is hard is that during the move, a given key exists in two places and writes keep arriving. Any scheme must answer: where do writes go during the copy, how do you catch up the delta, how do you cut over atomically, and how do you roll back if verification fails.

 ONLINE RESHARD OF ONE LOGICAL SHARD

  1. DUAL WRITE      writes -> old shard AND new shard (new is not read yet)
  2. BACKFILL        copy historical rows old -> new, in batches, throttled
  3. VERIFY          compare row counts and checksums; sample-compare rows
  4. SHADOW READ     read from both, compare results, alert on mismatch
                     (this step is what catches subtle bugs before users do)
  5. CUT OVER        flip the routing entry; reads now come from new
  6. SOAK            keep dual-writing for a while — this is your rollback
  7. CLEAN UP        stop dual writes, delete the old copy

  Every step is individually reversible until step 7. That is the design goal.

# Dual-write with a migration state machine. The state lives in the routing
# metadata, so every client behaves consistently.
class MigrationState(Enum):
    STABLE      = "stable"        # only the old shard
    DUAL_WRITE  = "dual_write"    # write both, read old
    VERIFYING   = "verifying"     # write both, read old, compare new
    CUT_OVER    = "cut_over"      # write both, read NEW
    COMPLETE    = "complete"      # only the new shard

def write(key, value, router):
    state, old, new = router.lookup(key)
    if state in (MigrationState.STABLE,):
        return old.put(key, value)
    if state is MigrationState.COMPLETE:
        return new.put(key, value)
    # All intermediate states write BOTH. The new shard must never miss a write
    # that lands after the backfill has already passed that row.
    old.put(key, value)
    new.put(key, value)          # if this fails, the migration must halt, not
                                 # silently continue with a divergent copy

def read(key, router):
    state, old, new = router.lookup(key)
    if state in (MigrationState.CUT_OVER, MigrationState.COMPLETE):
        return new.get(key)
    return old.get(key)

The far better strategy is to avoid resharding entirely by over-provisioning logical shards up front. Create 1,024 logical shards on day one and place them on 4 physical nodes; growing to 8 nodes moves whole logical shards without touching the hash function or any key's identity.

 LOGICAL SHARDS DECOUPLE KEY MAPPING FROM PHYSICAL PLACEMENT

  day 1:  1024 logical shards -> 4 nodes    (256 each)
  later:  1024 logical shards -> 8 nodes    (128 each)
          hash(key) -> logical shard  NEVER CHANGES
          only the logical -> physical placement map changes

  Moving a logical shard is a bounded, resumable, verifiable unit of work.

Approach	Downtime	Complexity	Risk
Stop the world and copy	Hours	Low	Unacceptable for most services
Dual write and backfill	None	High	Divergence if a dual write fails
Logical shard movement	None	Medium	Bounded per shard; resumable
Consistent hashing	None	Low	Only ~1/n moves, handled by the system

Key Takeaways

• Resharding must handle live writes arriving during the copy — that is the hard part.
• The standard sequence is dual write, backfill, verify, shadow read, cut over, soak, clean up.
• Keep every step reversible until the old copy is deleted.
• Over-provision logical shards from day one so growth becomes placement changes, not rehashing.

🧪 Practice

1. Write the full runbook to go from 4 to 8 shards online, including verification and rollback criteria.
2. Explain what breaks if a dual write succeeds on the old shard and fails on the new one, and how to detect it.
3. Interview: How do you verify a migration copied data correctly without comparing every row? (Hint: think about what you can compute cheaply per range and compare across two systems.)

Cross-Shard Queries and Joins

Sharding is what makes a database scale, and cross-shard queries are what it costs. Once rows for one logical question live on different machines, the database can no longer answer it with a local join — someone has to query every shard and combine the results, and that someone is usually your application.

Two operations become fundamentally harder: queries that span shards, and transactions that must update rows on more than one shard atomically.

 SCATTER-GATHER

   query "top 10 orders by total, all customers"
        |
        +--> shard 0: top 10 locally
        +--> shard 1: top 10 locally
        +--> shard 2: top 10 locally      (each must return 10, not fewer)
        +--> shard 3: top 10 locally
        |
        v
   merge 40 rows -> re-sort -> take the global top 10

 Costs:
   - latency = the SLOWEST shard, not the average (tail amplification, 1.2)
   - one logical query becomes N physical queries; capacity divided by N
   - pagination is genuinely hard: page 2 needs each shard's rows 11-20, and
     OFFSET across shards does not compose. Cursor pagination (3.1) per shard
     plus a merge is the workable approach.

async def scatter_gather_top_n(shards, sql, n, key):
    # Each shard must return n rows — the global top n could all live on one.
    results = await asyncio.gather(*[
        s.query(sql, limit=n) for s in shards
    ], return_exceptions=True)

    rows, failed = [], 0
    for r in results:
        if isinstance(r, Exception):
            failed += 1                      # decide deliberately: fail the query,
            continue                         # or return a PARTIAL result and say so
        rows.extend(r)
    if failed:
        raise PartialResult(f"{failed}/{len(shards)} shards unavailable")
    return sorted(rows, key=key, reverse=True)[:n]

The design work is mostly about avoiding these queries rather than optimizing them:

Technique	How it avoids cross-shard work
Co-location by shard key	Put related entities on the same shard (all of a tenant's data keyed by tenant_id)
Denormalization	Copy the joined fields into the row (4.2)
Reference-table replication	Copy small, rarely changing tables to every shard
A derived aggregate store	Send changes to a search or analytics store that holds everything (4.3)
Cursor pagination per shard	Merge sorted streams instead of using global OFFSET

Cross-shard transactions are worse still, because atomicity across machines requires either two-phase commit — which blocks if the coordinator fails — or a saga with compensating actions, which gives up isolation entirely. Both are covered in Chapter 8.5. The practical guidance is to design the shard key so that every transaction is single-shard, treating a cross-shard transaction as a design failure rather than a feature to implement.

 CHOOSING A SHARD KEY THAT AVOIDS THE PROBLEM

  Multi-tenant SaaS  -> shard by tenant_id
     every tenant's users, orders, and settings land together, so joins and
     transactions stay local and cross-shard queries are admin-only reporting.

  Social network     -> shard by user_id
     a user's own data is local; the follower graph is inherently cross-shard
     and is handled by a separate, purpose-built store.

Key Takeaways

• Cross-shard queries become scatter-gather: N physical queries, latency set by the slowest shard.
• Global ORDER BY, LIMIT, aggregation, and pagination must be recomputed by the client after merging.
• Avoid the problem by co-locating related data under one shard key and replicating small reference tables.
• Treat cross-shard transactions as a design failure; choose a key that keeps every transaction single-shard.

🧪 Practice

1. Implement scatter-gather pagination for page 3 of a globally sorted list and explain why OFFSET 20 per shard is wrong.
2. Choose a shard key for a multi-tenant CRM and list which queries remain cross-shard.
3. Interview: A cross-shard transaction is required by a new feature. What are your options? (Hint: one changes the data layout, one changes the consistency guarantee, and one changes the feature.)

<a id="46-data-lifecycle-management"></a>

4.6 Data Lifecycle Management

Data has a life beyond being written and read: it must be recoverable, migratable, eventually cheaper to store, sometimes legally erasable, and often streamed elsewhere.

Backup and Restore Strategies

Replication protects against hardware failure. It does not protect against a DELETE with a missing WHERE clause, a bad migration, ransomware, or a bug that corrupts data over three weeks — because replicas faithfully reproduce all of those within milliseconds. Backups protect against logical damage, which is the category that actually destroys companies.

The distinction is worth stating plainly: replication gives you availability; backups give you the ability to go back in time. They are not substitutes.

 WHAT PROTECTS AGAINST WHAT

  threat                          replication   backup
  -----------------------------------------------------
  disk failure                        yes         yes
  node/AZ failure                     yes         yes
  region failure                   multi-region   yes
  accidental DELETE/DROP               NO         yes
  bad migration                        NO         yes
  slow data corruption bug             NO         yes*   (*if retention is long
  ransomware                           NO         yes*     enough to predate it)
  deleting the backups themselves      NO      immutable/offline copies only

The industry shorthand is the 3-2-1 rule: at least 3 copies of the data, on 2 different media or storage systems, with 1 copy off-site. In cloud terms that usually means the live database, automated snapshots in the same region, and replicated backups in a different region and a different account.

# PHYSICAL backup: a byte-level copy of the data directory plus WAL. Fast to
# take and fast to restore; tied to the exact database version and platform.
pg_basebackup -h db.internal -D /backups/base -Fp -Xs -P

# LOGICAL backup: a portable dump of schema and data. Slower both ways, but
# version-portable, selectively restorable, and human-inspectable.
pg_dump --format=custom --compress=9 orders > orders_2026-03-11.dump
pg_restore --dbname=orders --jobs=8 orders_2026-03-11.dump   # selective restore

# The part teams skip: proving the backup works. An untested backup is a
# hypothesis, not a recovery plan.
def weekly_restore_drill(backup, staging):
    started = time.time()
    staging.restore(backup.latest())                       # full restore
    elapsed = time.time() - started

    checks = {
        "row_counts_match":  staging.count("orders") >= backup.metadata.row_count,
        "constraints_valid": staging.validate_constraints(),
        "spot_check_rows":   staging.sample_matches(backup.metadata.checksums),
        "meets_rto":         elapsed < RTO_SECONDS,        # RTO is a MEASURED number
    }
    if not all(checks.values()):
        page_oncall("restore drill failed", checks)
    metrics.gauge("backup.restore_seconds", elapsed)       # track it over time:
    return checks                                          # restores get slower as
                                                           # data grows

Dimension	Choice	Consequence
Type	Physical vs. logical	Speed and size vs. portability and selectivity
Scope	Full vs. incremental	Restore simplicity vs. backup window
Frequency	Sets your RPO	More frequent = less data lost
Retention	Days vs. years	Protects against slow corruption; costs money
Location	Same region vs. cross-region vs. cross-account	Blast-radius isolation
Immutability	Object lock / WORM	Survives an attacker with your credentials

That immutability row has become essential. If your backups can be deleted with the same credentials that run your infrastructure, a compromised account destroys both the data and its recovery path at once. Object-lock retention and a separate account with separate credentials are what make backups survive an attacker.

Key Takeaways

• Replication protects against hardware failure; only backups protect against logical damage.
• Follow 3-2-1: three copies, two storage systems, one off-site.
• Physical backups restore fast; logical backups are portable and selectively restorable.
• Test restores on a schedule and measure the time — an untested backup is not a backup. Make backups immutable and separately credentialed.

🧪 Practice

1. Design a backup strategy for a 2 TB database with RPO 15 minutes and RTO 1 hour, and state what each component contributes.
2. Explain why a read replica is not a backup, using two concrete failure scenarios.
3. Interview: A corruption bug has been silently damaging rows for three weeks. What does your backup strategy need for you to recover? (Hint: think about retention depth and about how you would even identify the last good moment.)

Point-in-Time Recovery

A nightly backup restores you to midnight. If the damaging UPDATE ran at 14:32, restoring the backup discards fourteen hours of legitimate work along with the mistake. Point-in-time recovery (PITR) removes that trade-off: restore the last full backup, then replay the write-ahead log up to any chosen instant — say 14:31:59.

This is possible precisely because of the WAL from section 4.1. The log already contains every change in order, so a base backup plus a continuous archive of log segments is a complete, replayable history.

 PITR = BASE BACKUP + CONTINUOUS WAL ARCHIVE

  00:00 [ full base backup ]
        |
  ------+====WAL====WAL====WAL====WAL====WAL====WAL====> time
                                    ^
                              14:31:59 target
                              (just before the bad UPDATE at 14:32)

  Restore = copy the base backup, then replay archived WAL segments and stop
  at the target. Everything committed before 14:31:59 is present; nothing
  after it is applied.

  Your RPO is set by how often WAL segments are ARCHIVED, not by how often the
  base backup runs.

# 1. Enable continuous archiving (this is what makes PITR possible at all)
#    postgresql.conf:
#      wal_level = replica
#      archive_mode = on
#      archive_command = 'aws s3 cp %p s3://db-wal-archive/%f'
#      archive_timeout = 60      # force a segment at least every 60s -> RPO ~60s

# 2. Recover to an instant:
#    restore the base backup, then in recovery.signal / postgresql.conf:
#      restore_command = 'aws s3 cp s3://db-wal-archive/%f %p'
#      recovery_target_time = '2026-03-11 14:31:59+00'
#      recovery_target_action = 'pause'   # PAUSE, do not auto-promote: verify
#                                         # the data before making it live

# The hardest part of PITR is not the mechanics — it is choosing the target.
def find_recovery_target(audit_log, table):
    """Locate the last known-good instant before damage began."""
    # Sources that tell you WHEN, in rough order of reliability:
    #   1. the application audit log / CDC stream (what changed, and when)
    #   2. the deploy timeline (a bad release is often the cause)
    #   3. row-level updated_at timestamps on the damaged rows
    #   4. user reports (least precise, usually latest)
    suspicious = audit_log.first_anomaly(table)
    return suspicious.timestamp - timedelta(seconds=1)

# And the decision that surrounds it: a full PITR rolls back EVERYTHING,
# including good writes made after the damage began. Often the better plan is:
#   - PITR into a SEPARATE instance at the target time
#   - extract only the damaged rows from it
#   - merge them back into the live database
# This is slower but does not discard fourteen hours of unrelated work.

Aspect	Restore from backup	Point-in-time recovery
Granularity	Backup boundaries	Any instant covered by the archive
RPO	Backup interval (hours)	WAL archive interval (seconds/minutes)
Storage cost	Backups only	Backups plus the full WAL archive
Restore time	Base copy	Base copy plus replay (longer for a distant target)
Data after the target	Lost	Lost — unless you restore side-by-side and merge

Key Takeaways

• PITR restores a base backup and replays the WAL to any chosen instant.
• RPO is determined by WAL archive frequency, not by backup frequency.
• Pause before promoting a recovered instance and verify the data first.
• Prefer restoring to a side instance and merging the damaged rows over rolling the whole database back.

🧪 Practice

1. Calculate the WAL storage needed for 7 days of PITR at 50 GB of WAL per day, and state your RPO under a 5-minute archive timeout.
2. Describe how you would identify the exact recovery target after a bad migration.
3. Interview: PITR would lose 6 hours of good writes made after the corruption started. What do you do? (Hint: nothing says the recovered copy has to become the live database.)

Schema Migrations at Scale

On a small table, ALTER TABLE is instant. On a 500-million-row table under production load, the same statement can hold an exclusive lock for an hour, queueing every query behind it and taking the application down. Schema migration at scale is the discipline of changing structure without ever holding a long lock and without a moment where the running code and the schema disagree.

Two constraints drive everything. Deploys are not atomic: during a rolling release, old and new application versions run simultaneously against one schema. And locks are contagious: a blocked ALTER blocks every subsequent query on the table, including reads, so a "brief" lock becomes a full outage.

 THE EXPAND-CONTRACT PATTERN (the safe way to change anything)

  EXPAND    add the new structure; leave the old one working
            -> schema now supports BOTH old and new code
  MIGRATE   backfill data into the new structure, in throttled batches
  SWITCH    deploy code that reads and writes the new structure
  CONTRACT  once no code uses the old structure, remove it

  Renaming `name` -> `full_name`, safely, across four deploys:

   deploy 1  ADD COLUMN full_name  (nullable, no default rewrite)
             app writes BOTH columns, reads `name`
   deploy 2  backfill full_name in batches; app reads full_name, writes both
   deploy 3  app reads and writes full_name only
   deploy 4  DROP COLUMN name

  A single `ALTER TABLE ... RENAME` would break every running instance of the
  old code the instant it landed.

-- SAFE: metadata-only in modern PostgreSQL and MySQL 8
ALTER TABLE orders ADD COLUMN currency TEXT;               -- nullable, no default
ALTER TABLE orders ADD COLUMN currency TEXT DEFAULT 'EUR'; -- PG 11+: no rewrite

-- DANGEROUS: rewrites the whole table while holding an exclusive lock
ALTER TABLE orders ALTER COLUMN total TYPE NUMERIC(14,2);
ALTER TABLE orders ADD COLUMN x TEXT NOT NULL DEFAULT '';  -- older versions rewrite

-- SAFE index creation: does not block writes (takes longer, can fail and
-- leave an INVALID index that must be dropped and retried)
CREATE INDEX CONCURRENTLY idx_orders_currency ON orders (currency);

-- SAFE constraint addition: two steps, neither taking a long lock
ALTER TABLE orders ADD CONSTRAINT chk_total CHECK (total >= 0) NOT VALID;
ALTER TABLE orders VALIDATE CONSTRAINT chk_total;   -- scans without blocking writes

-- ALWAYS bound the wait, so a migration fails fast instead of queueing traffic
SET lock_timeout = '3s';
SET statement_timeout = '30s';

# Backfills must be batched, throttled, resumable, and idempotent.
def backfill_currency(db, batch=5_000):
    last_id = load_checkpoint() or 0            # resumable after a restart
    while True:
        rows = db.execute(
            "UPDATE orders SET currency = 'EUR' "
            "WHERE id > %s AND currency IS NULL "
            "ORDER BY id LIMIT %s RETURNING id", (last_id, batch))
        if not rows:
            break
        last_id = rows[-1].id
        save_checkpoint(last_id)
        # Throttle on replication lag, not on a fixed sleep: the backfill must
        # not push followers behind and break read-your-writes (4.4).
        while db.replication_lag_seconds() > 5:
            time.sleep(1)

Operation	Lock impact	Safe approach
Add nullable column	Metadata only	Direct
Add column with default	Version-dependent	Check your version; else add then backfill
Drop column	Metadata only	Direct — but only after no code reads it
Rename column	Metadata only, breaks code	Expand-contract across deploys
Change column type	Full table rewrite	New column, backfill, switch, drop
Add index	Blocks writes	CREATE INDEX CONCURRENTLY
Add NOT NULL	Full scan	CHECK ... NOT VALID, then validate
Add foreign key	Locks both tables	NOT VALID, then validate

For very large tables on MySQL, tools like gh-ost and pt-online-schema-change implement a shadow-table approach: build a copy with the new schema, replay changes from the binlog, then swap atomically. The principle is the same as the online resharding in 4.5 — build alongside, verify, cut over, keep a rollback.

Key Takeaways

• Rolling deploys mean old and new code share one schema — every migration must be compatible with both.
• Use expand-contract: add, backfill, switch, then remove, across separate deploys.
• Know which operations rewrite the table; set lock_timeout so migrations fail fast instead of queueing traffic.
• Backfills must be batched, checkpointed, and throttled on replication lag.

🧪 Practice

1. Write the four-deploy plan to split full_name into first_name and last_name on a 200-million-row table.
2. Explain why SET lock_timeout protects availability more than a carefully-chosen maintenance window does.
3. Interview: A migration must add a NOT NULL column with a computed default to a huge live table. Walk through it. (Hint: separate "the column exists" from "every row has a value" from "the constraint is enforced".)

Data Archival and Tiering

Data does not stay equally valuable. Yesterday's orders are queried constantly; orders from 2019 are queried by an auditor twice a year — yet in a naive design both sit on the same expensive storage, in the same tables, slowing the same queries and inflating the same backups. Tiering matches storage cost to access frequency; archival moves cold data out of the operational path entirely.

The benefit is not only money. A smaller hot table means more of it fits in the buffer pool, indexes are smaller and shallower, VACUUM and backups finish faster, and restores meet their RTO. Archival is a performance strategy that happens to save cost.

 A TEMPERATURE HIERARCHY

  HOT      last 90 days      primary DB, SSD, fully indexed
                             query latency: milliseconds       cost: 1x
  WARM     90 days - 2 years partitioned tables or a columnar store
                             query latency: seconds            cost: ~0.2x
  COLD     2 - 7 years       object storage (Parquet), queried on demand
                             query latency: seconds-minutes    cost: ~0.02x
  FROZEN   7+ years          archival object tier (Glacier-class)
                             retrieval: minutes-hours          cost: ~0.004x

  The decision is driven by ACCESS FREQUENCY and by the LATENCY the rare
  reader will accept — not by the age of the data as such.

-- Partitioning makes archival a metadata operation instead of a mass DELETE.
CREATE TABLE orders (...) PARTITION BY RANGE (created_at);

-- Detach an old partition: instant, no row-by-row work, no bloat.
ALTER TABLE orders DETACH PARTITION orders_2019_q1;

-- Export it to columnar files in object storage, then drop it.
COPY (SELECT * FROM orders_2019_q1) TO PROGRAM
  'aws s3 cp - s3://archive/orders/2019/q1.parquet' (FORMAT parquet);
DROP TABLE orders_2019_q1;

# Archived data must remain QUERYABLE, or you have not archived it — you have
# hidden it. Provide an explicit path, and make its cost visible to callers.
def query_orders(start, end, db, lake):
    if start >= HOT_BOUNDARY:
        return db.query_orders(start, end)                    # milliseconds

    if end < HOT_BOUNDARY:
        return lake.query_parquet("orders", start, end)       # seconds; scanned
                                                              # from object storage
    # Spanning the boundary: query both and merge. Expose the latency
    # difference rather than hiding it behind a uniform API that is sometimes
    # 3 ms and sometimes 40 seconds.
    return merge(db.query_orders(HOT_BOUNDARY, end),
                 lake.query_parquet("orders", start, HOT_BOUNDARY))

Tier	Storage	Query path	Typical relative cost
Hot	Primary DB, SSD	Normal queries	1x
Warm	Compressed partitions, columnar store	Slower SQL	~0.2x
Cold	Parquet in object storage	Query engine on demand	~0.02x
Frozen	Archival object tier	Restore first, then query	~0.004x

Two operational cautions. Retrieval costs money and time — archival tiers charge for restores and can take hours, so "we can always get it back" needs a tested procedure and a known cost. And archived data still has obligations: encryption, access control, and deletion requests apply to it exactly as they do to hot data, which is the subject of the next topic.

Key Takeaways

• Tiering matches storage cost to access frequency; archival removes cold data from the operational path.
• Smaller hot tables mean better cache hit rates, faster maintenance, and faster restores.
• Partitioning turns archival into detach-and-export instead of a mass DELETE.
• Archived data must stay queryable and still carries encryption, access, and deletion obligations.

🧪 Practice

1. Design a tiering policy for an orders table growing 500 GB per year with a 7-year legal retention requirement.
2. Explain why detaching a partition is dramatically cheaper than DELETE FROM orders WHERE created_at < ....
3. Interview: Finance wants seven years of data queryable but the database is too expensive. What do you propose? (Hint: separate "queryable" from "queryable in 5 milliseconds" and price each.)

Data Retention and Deletion

Keeping data forever used to be the default because storage is cheap. It is now a liability: privacy regulation grants individuals a right to erasure, security teams treat old personal data as breach surface, and legal teams know that anything retained is discoverable. Retention has become a design requirement, and "delete this user's data" turns out to be surprisingly hard in a distributed system.

The reason is that data spreads. One user's record lives in the primary database, its replicas, backups, the search index, the cache, the analytics warehouse, log files, the message queue's retained topics, and a third-party CRM. A DELETE against the primary addresses one of those.

 WHERE ONE USER'S DATA ACTUALLY LIVES

   [primary DB] ---> [replicas]        deleted by replication
        |
        +---> [backups]                NOT deleted — they are immutable by design
        +---> [search index]           needs its own deletion path
        +---> [cache]                  needs explicit invalidation
        +---> [analytics warehouse]    needs its own deletion path
        +---> [event stream]           retained for days; may need compaction
        +---> [application logs]       often contain PII; retention-bound
        +---> [third-party services]   needs an API call, and proof it happened

 A deletion request is a WORKFLOW across every one of these, not a SQL statement.

# Crypto-shredding: the technique that resolves the backup problem.
# Encrypt each user's personal data with a per-user key. Deleting the KEY makes
# every copy of the ciphertext — including immutable backups — unreadable.
def store_pii(user_id, data, kms, db):
    key = kms.get_or_create_key(f"user:{user_id}")     # one key per subject
    db.insert("user_pii", {"user_id": user_id, "blob": encrypt(data, key)})

def erase_user(user_id, kms, db, search, warehouse, crm):
    kms.destroy_key(f"user:{user_id}")     # backups now hold undecryptable bytes
    db.execute("DELETE FROM user_pii WHERE user_id = %s", (user_id,))
    search.delete_document(f"user:{user_id}")
    warehouse.delete_subject(user_id)      # or anonymize, if aggregates must stay
    crm.delete_contact(user_id)            # third parties need an explicit call
    audit.record("erasure_completed", user_id)   # you must be able to PROVE it

-- Soft delete keeps a row for recovery and referential integrity, but it is
-- NOT erasure — the personal data is still there.
UPDATE users SET deleted_at = now() WHERE id = 7;

-- Two-stage deletion is the pattern that satisfies both needs:
--   stage 1 (immediate): soft delete; the user disappears from the product
--   stage 2 (after the grace period): hard delete or anonymize
UPDATE users
   SET email = concat('deleted-', id, '@invalid'),   -- keep the row, destroy the
       name  = 'Deleted User',                       -- identifying content
       phone = NULL,
       anonymized_at = now()
 WHERE deleted_at < now() - INTERVAL '30 days';
-- Aggregates and foreign keys survive; the person is no longer identifiable.

Concern	Approach
Accidental deletion recovery	Soft delete with a grace period
Regulatory erasure	Hard delete or anonymization, plus crypto-shredding for backups
Referential integrity	Anonymize in place rather than removing the row
Backups	Crypto-shredding, or documented retention expiry
Analytics value	Aggregate first, then delete the identifiable rows
Legal hold	Suspends deletion — it must override retention automation
Proof	An audit trail of what was erased and when

That legal-hold row is a real conflict: retention policies delete on a schedule, while litigation holds require preservation. Whichever system implements retention must support an override, or an automated job will destroy evidence someone was legally required to keep.

Key Takeaways

• Retention is a design requirement: old personal data is liability, not an asset.
• One user's data spans replicas, backups, indexes, caches, warehouses, logs, and third parties — erasure is a workflow.
• Crypto-shredding (destroy the per-user key) is how immutable backups are handled.
• Soft delete is recovery, not erasure; anonymization preserves aggregates and referential integrity. Keep an audit trail and honour legal holds.

🧪 Practice

1. Enumerate every system that would hold a user's data in an app you know, and write the deletion step for each.
2. Explain why soft delete alone does not satisfy a right-to-erasure request.
3. Interview: How do you delete a user's data from immutable backups? (Hint: you do not have to make the bytes disappear to make them useless.)

Change Data Capture

Section 4.3 ended with a rule: never dual-write. But derived systems — search indexes, caches, analytics warehouses, other services — still need to know when data changes. Change data capture (CDC) solves this by reading the database's own write-ahead log and turning committed changes into an event stream. The database remains the single source of truth, and every derived system follows from its log.

The elegance is that CDC captures exactly what was committed, in commit order, including changes made by scripts, admin tools, and other services that never touched your application code. Nothing can be missed by forgetting to publish an event, because the publisher is the log itself.

 CDC ARCHITECTURE

   application --writes--> [ PostgreSQL ]
                                 |
                              WAL (already exists, section 4.1)
                                 |
                          [ CDC connector ]  (Debezium, native logical decoding)
                                 |
                          [ Kafka topics ]   one per table, keyed by primary key
                    +------------+------------+-------------+
                    v            v            v             v
              [search index] [cache]   [warehouse]   [other service]

   Each consumer is independent, replayable, and idempotent. If one breaks,
   it catches up from its offset; if it needs rebuilding, it replays from the
   beginning of the topic.

{
  "op": "u",
  "ts_ms": 1773225862000,
  "source": { "table": "orders", "lsn": 24857392, "txId": 88123 },
  "before": { "id": 42, "status": "pending", "total": "58.00" },
  "after":  { "id": 42, "status": "paid",    "total": "58.00" }
}

Having both before and after is what makes CDC more useful than a plain event feed: a consumer can react to the transition ("pending became paid") rather than just the current state, which is exactly what audit trails and workflow triggers need.

# Consuming CDC safely. Two properties are mandatory.
def handle_change(event, search, offsets):
    key = f"{event['source']['table']}:{event['after']['id']}"

    # 1. IDEMPOTENCY: delivery is at-least-once, so the same event will arrive
    #    twice. Use the LSN as a version so replays cannot go backwards.
    if search.version_of(key) >= event["source"]["lsn"]:
        return "already_applied"

    if event["op"] in ("c", "u"):                     # create / update
        search.index(key, event["after"], version=event["source"]["lsn"])
    elif event["op"] == "d":                          # delete
        search.delete(key, version=event["source"]["lsn"])

    # 2. OFFSET COMMIT AFTER the side effect, never before — committing first
    #    means a crash silently skips the change.
    offsets.commit(event.offset)

CDC approach	How it works	Trade-off
Log-based	Reads the WAL/binlog	Complete, low overhead, ordered — the standard
Trigger-based	Database triggers write to an audit table	Portable; adds write latency and load
Query-based polling	Poll WHERE updated_at > last_seen	Simple; misses deletes and intermediate states
Application outbox	Write events in the same transaction	Explicit and portable; requires app discipline (Chapter 8.5)

Operationally, three things matter. Replication slots retain WAL: a stopped CDC consumer causes the database's disk to fill, which is a genuine outage risk — monitor slot lag as a first-class alert. Schema changes flow through, so consumers must tolerate new and removed columns exactly as in Chapter 3.3. And snapshot plus stream is the standard bootstrap: an initial consistent snapshot, then the log from that snapshot's position onward.

Key Takeaways

• CDC turns the database's own commit log into an ordered event stream, so derived systems follow the source of truth.
• It captures every committed change regardless of which client made it — nothing can forget to publish.
• before/after images let consumers react to transitions, not just current state.
• Consumers must be idempotent and commit offsets after side effects; monitor replication slot lag or the database's disk will fill.

🧪 Practice

1. Design a CDC pipeline that keeps an Elasticsearch index in sync, including the initial snapshot and a full rebuild path.
2. Explain what happens to the primary database when a CDC consumer is down for a week, and how you would alert on it.
3. Interview: How does CDC differ from the outbox pattern, and when would you choose each? (Hint: one captures rows, the other captures intent — think about which is easier for a consumer to interpret.)

Chapter Summary

Concept	The one-line version
Storage abstractions	Block for databases, file for sharing, object for blobs at unlimited scale
Storage engines	The engine, not the query language, sets performance characteristics
B-tree vs. LSM	In-place updates and predictable reads, versus append-only fast writes
Write-ahead logging	Log before applying; one sequential fsync per commit powers recovery, replication, PITR, and CDC
Indexes	Turn scans into lookups; composite order matters, and functions disable them
Serialization	Text for humans, binary schema-driven for volume, columnar for analytics
Normalization	Every fact in one place; constraints are the only concurrency-safe checks
Denormalization	A cache inside the database — needs sync and reconciliation
ACID	Atomicity/durability from the WAL, isolation from MVCC, consistency from constraints
Isolation levels	Defined by permitted anomalies; write skew survives snapshot isolation
Locking vs. MVCC	Wait to prevent conflicts, or allow concurrency and detect them
Query planning	Read EXPLAIN ANALYZE; estimate-versus-actual gaps reveal stale statistics
Connection pooling	Size from Little's Law; the database's total limit is the real ceiling
Key-value	O(1) lookups, trivial partitioning, every access path is a key you maintain
Document	One entity, one read; embed bounded owned data, reference everything else
Wide-column	Partition key picks the node; model one table per query
Graph	Index-free adjacency makes multi-hop traversal cheap
Time-series	Time partitioning, delta compression, downsampling; guard tag cardinality
Search	Inverted index plus analysis pipeline; a derived, rebuildable index
Vector	ANN over embeddings; same model at index and query time; hybrid beats pure
Polyglot persistence	One owner per fact, everything else derived from a change stream
Leader-follower	One writer removes conflicts; scales reads, never writes
Multi-leader	Local writes at the price of conflict resolution; partition ownership instead
Quorums	W + R > N guarantees overlap; sloppy quorums quietly remove it
Sync vs. async replication	RPO 0 and a round trip, versus speed and possible loss on failover
Replication lag	Causes read-your-writes and monotonic read violations; route on lag
Failover	Detect, elect, redirect; fence the old leader or risk split-brain
Vertical vs. horizontal	Columns/tables versus rows; only sharding scales without limit
Range sharding	Keeps order and range queries; hotspots on sequential keys
Hash sharding	Even distribution, no ordering; plain modulo makes resharding catastrophic
Directory sharding	Explicit mapping enables per-tenant placement and residency
Consistent hashing	Adding a node moves ~1/n of keys; virtual nodes even out the ring
Hotspots	Data skew and traffic hotspots differ; cache reads, salt writes
Resharding	Dual write, backfill, verify, cut over, soak — reversible until cleanup
Cross-shard queries	Scatter-gather at the slowest shard's latency; co-locate to avoid them
Backups	Only backups survive logical damage; test restores and make them immutable
PITR	Base backup plus WAL replay to any instant; RPO set by archive frequency
Migrations	Expand-contract across deploys; batch and throttle backfills
Tiering	Match storage cost to access frequency; keep archives queryable
Retention	Erasure is a workflow across every copy; crypto-shredding handles backups
Change data capture	The commit log becomes an event stream; consumers idempotent, slots monitored

---

<a id="5-caching-and-content-delivery"></a>