Choosing a database

The problem it solves

A chat app stores 40 billion messages and takes 700,000 new ones a second at peak. Almost every read is the same question: the latest 50 messages of one conversation. A second team wants a search box: every message containing “invoice” from last month.

Put both questions to two databases and the answers swap.

Question	One PostgreSQL primary	A Cassandra cluster
Latest 50 of conversation 7	One index range scan on (chat_id, seq), about 1 ms	One partition, rows already sorted, about 1 ms
700,000 inserts a second	About 70 times what one primary sustains	2.1 million replica writes over about 200 nodes, and more nodes scale it about linearly
Messages containing “invoice”	A full-text index works, slowly at 40 billion rows	Impossible without scanning every partition on every node
Move a message between chats	One transaction	Two writes with no transaction between them

Neither store is “better”. Each makes some questions cheap and others impossible. Choosing a database means listing the questions the system asks, how often and how fresh, and picking the store that makes the frequent ones cheap. A store picked because “NoSQL scales” makes some question that matters expensive, and moving billions of rows out later takes months.

What makes this hard

• The access patterns decide, not the shape of the data. The same messages fit a relational table, a document and a wide-column partition. The questions you ask of them fit only some.
• Consistency is a requirement like any other. “Two rows change together or not at all” rules out most stores that scale writes across machines.
• Every extra store is another copy that can disagree with the first. Keeping copies in step is the real cost of using several databases.

How it works

Step 1: Write down the access patterns

Before naming a database, list every read and write the system does, with its rate and its consistency need. For a ticketing system it looks like this:

Access pattern	Rate (peak)	Shape	Must be fresh?
Hold seats for a buyer	2,000/s	Conditional update on several rows at once	Yes: two buyers, one seat
Show the seat map of an event	50,000/s	All seats of one event	A few seconds stale is fine
Find events by name, city and date	5,000/s	Full-text plus filters	Minutes stale is fine
Sales per event per hour for organizers	10/s	Aggregate over millions of rows	Minutes stale is fine

Four patterns, four different needs. The first needs a transaction across rows. The second is a hot read that a cache absorbs. The third is a search engine’s job. The fourth is an analytics query that should not run on the database taking bookings. The table is the input to every later step on this page.

Step 2: One relational database

Start with what a relational database, such as PostgreSQL or MySQL, gives you on one machine. It answers more of that table than any other single store.

• Tables, and a B-tree per index. The primary key and every secondary index are sorted trees. So both a point lookup and a range (“seats of event 42”, “orders since Monday”) take a few page reads. A new question needs a new index, built online, not a new copy of the data.
• Joins and a query planner. You store each fact once and combine facts at read time. The planner picks the indexes, so a query you did not plan for still runs, if not always fast.
• Transactions. Several rows change together or not at all, and constraints (unique keys, foreign keys, checks) are enforced by the engine. “Hold seats 14 and 15 only if both are free” is one conditional update and a row count check inside one transaction.
• Isolation levels decide what concurrent transactions see of each other. Postgres defaults to READ COMMITTED, which is enough for a conditional UPDATE ... WHERE status = 'FREE' because the row lock serializes the two writers. Invariants that span rows nobody locks, such as “at least one doctor on call”, need SERIALIZABLE or an explicit lock. The consistency models page covers the anomalies.

Here’s a planning figure. One Postgres primary with NVMe disks sustains about 10,000 indexed write transactions a second. It serves tens of thousands to a few hundred thousand cached point reads a second, depending on cores. Treat these as orders of magnitude, not benchmarks. Small rows, few indexes and batched commits can reach several times that. Wide rows with many indexes reach less. Read replicas multiply reads, not writes. Ten billion rows and a few terabytes on one node are normal.

Go deeperIsolation levels in one table1 min

Level	Prevents	Still allows	Default in
READ COMMITTED	Dirty reads	Lost updates from read-then-write, write skew	Postgres, Oracle, SQL Server
REPEATABLE READ (snapshot)	Non-repeatable reads, lost updates (Postgres aborts the second writer)	Write skew	MySQL InnoDB (with gap locks)
SERIALIZABLE	Every anomaly: the result equals some serial order	Nothing, but transactions abort and must retry	CockroachDB, Spanner

MySQL’s REPEATABLE READ is not quite snapshot isolation. A plain SELECT reads the snapshot, but an UPDATE reads the latest row. So read-then-write code can still lose updates. Postgres’s SERIALIZABLE (SSI, since 9.1) runs at close to snapshot cost. It aborts a transaction when its reads and writes could form a cycle with others. The check is conservative, so some aborts are false positives. Callers must retry on serialization errors.

Step 3: Postgres until it hurts

The default for a new system is one relational primary, with a replica or two. It stays the default until a specific signal says otherwise. The signals are measurable:

1. Write rate. Sustained writes above about half of what the primary can take, with WAL volume and replica lag growing at peak.
2. Operating size. A backup restore, a replica rebuild, or a major version upgrade takes many hours. That usually happens somewhere past 5 to 10 TB on one node.
3. Working set. The hot data no longer fits in RAM, and the buffer cache hit ratio falls below about 99%. So reads start waiting on disk.
4. Geography. Users on several continents need low-latency writes, or the business needs zero data loss when a whole region fails.
5. A different kind of question. Relevance-ranked text search, deep graph traversals, or scans over billions of rows for dashboards.

None of these is “we have a lot of rows” or “joins are slow”. A billion rows is routine, and a slow join is almost always a missing index. When a signal appears, climb one rung at a time:

The ladder before leaving one primary

Scroll or pinch to zoom, drag to pan

• Color key
• Database
• Service
• Cache

The ladder is ordered by cost. An index costs minutes. A replica costs a machine and a little stale reading. Sharding costs a permanent tax on every query that does not carry the shard key, and every schema change runs N times. Each rung is worth it only when the cheaper ones are used up.

Step 4: Key-value stores

Drop everything but “get and put by key” and a store can be very fast and very simple to scale. The key is hashed to a partition, so any number of machines can share the data, and no request touches more than one partition.

• Redis keeps everything in RAM, and executes commands on a single thread per shard. That’s about 100,000 to 200,000 simple commands a second per core without pipelining, and well over a million with it, at sub-millisecond latency. Its costs are RAM prices and durability. With the append-only file synced every second, a crash loses up to a second of writes. And asynchronous replication means a failover can drop writes the client saw succeed.
• DynamoDB is the durable, managed version. A partition key spreads items, a sort key orders items inside one partition, so “orders of user 42, newest first” is one Query. Reads and writes take single-digit milliseconds at any size. Each physical partition accepts about 1,000 writes and 3,000 reads a second, which is why a single hot key needs its own sharding scheme. Transactions cover up to 100 items at twice the cost.

The price is everything else. There are no joins. A query on any attribute other than the key needs a secondary index designed in advance, or a full scan. (In DynamoDB, global secondary indexes are always eventually consistent.)

Step 5: Document stores

A document store such as MongoDB stores a whole aggregate, an order with its line items and address, as one JSON-like document. One read returns everything a screen needs, and changes to one document are atomic. Secondary indexes on any field, including fields inside arrays, make it far more queryable than a key-value store.

The modeling rule is: embed what you read together, and reference what grows without bound or is shared. An order embeds its line items. A user does not embed every order they ever placed. MongoDB caps a document at 16 MB, and rewriting a large document on every change is slow. Joins ($lookup) exist, but cost more than in SQL. Multi-document transactions work, also across shards, at a higher price than single-document writes.

Postgres’s jsonb columns with GIN indexes cover a lot of this ground, inside a relational database. So “the data is JSON” alone is not a reason to add a document store. The reasons are operational: built-in sharding, and a team that already runs it well.

Step 6: Wide-column stores

Cassandra, ScyllaDB, HBase and Bigtable store rows inside partitions, and the primary key has two parts that do different jobs:

• The partition key is hashed to pick the nodes that hold the partition. All rows with the same partition key live together on the same replicas.
• The clustering key sorts rows inside the partition on disk. A range over it is one sequential read.

PRIMARY KEY ((chat_id), seq): the partition key picks the node, the clustering key sorts the rows

Scroll or pinch to zoom, drag to pan

• Color key
• Service
• Database

Writes go to a commit log and an in-memory table. That table is flushed to sorted files later (an LSM tree, covered on the indexes page). So a write never reads first, and never updates in place. That’s why a node takes roughly 10,000 to 30,000 writes a second, counting every replica write. And a cluster scales about linearly. Netflix measured 1.1 million client writes a second on 288 nodes in 2011. That’s about 11,000 replica writes per node, at replication factor 3, on the cloud instances of the time. Every datacenter can accept writes. Conflicts resolve per column by timestamp: last write wins.

The model is query-first. A table serves one query, so “messages by sender” is a second table that the application also writes to. Keep each partition under about 100 MB, or reads and compaction slow down. There are no transactions across partitions, and compare-and-set is a lightweight transaction, a Paxos round of about four round trips.

JWideColumnTable.javaPywide_column_table.pyC#WideColumnTable.csGowide_column_table.goTSwide-column-table.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

import java.nio.charset.StandardCharsets;
import java.util.*;

// A wide-column table, messages_by_chat, reduced to its two ideas:
//   PRIMARY KEY ((chat_id), seq)  WITH CLUSTERING ORDER BY (seq DESC)
// The partition key picks the node. The clustering key sorts rows inside the partition.
record Message(String chatId, long seq, String sender, String text) {}

class WideColumnTable {
    private final int nodes;
    // One sorted map per partition: this is what makes "latest 50" a single slice read.
    private final Map<String, TreeMap<Long, Message>> partitions = new HashMap<>();
    int nodesTouched;

    WideColumnTable(int nodes) { this.nodes = nodes; }

    // Real stores use Murmur3 tokens on a ring. FNV-1a keeps the example deterministic
    // and the same in every language; the point is only that the key alone picks the node.
    int nodeFor(String partitionKey) {
        int h = 0x811c9dc5;
        for (byte b : partitionKey.getBytes(StandardCharsets.UTF_8)) {
            h ^= (b & 0xff);
            h *= 0x01000193;
        }
        return Integer.remainderUnsigned(h, nodes);
    }

    // A write is an append into one partition on one node: no read first, no lock.
    void insert(Message m) {
        partitions.computeIfAbsent(m.chatId(), k -> new TreeMap<>()).put(m.seq(), m);
    }

    // The query the table was designed for: names the partition key, reads a sorted slice.
    List<Message> latest(String chatId, int limit) {
        nodesTouched = 1;
        TreeMap<Long, Message> p = partitions.getOrDefault(chatId, new TreeMap<>());
        return p.descendingMap().values().stream().limit(limit).toList();
    }

    // A query that does not name the partition key: every node must look at every partition.
    // This is why wide-column modeling starts from the queries, and adds a second table
    // (messages_by_sender) instead of a filter.
    List<Message> bySender(String sender) {
        nodesTouched = nodes;
        List<Message> out = new ArrayList<>();
        for (TreeMap<Long, Message> p : partitions.values())
            for (Message m : p.values())
                if (m.sender().equals(sender)) out.add(m);
        return out;
    }
}

class Main {
    public static void main(String[] args) {
        WideColumnTable table = new WideColumnTable(6);
        for (long seq = 1; seq <= 5; seq++) {
            table.insert(new Message("chat-7", seq, seq % 2 == 0 ? "ana" : "bo", "hello " + seq));
            table.insert(new Message("chat-9", seq, "ana", "hi " + seq));
        }
        System.out.println("chat-7 lives on node " + table.nodeFor("chat-7"));
        List<Message> last = table.latest("chat-7", 3);
        System.out.println("latest 3: " + last.stream().map(m -> m.seq()).toList()
                + ", nodes touched: " + table.nodesTouched);
        List<Message> ana = table.bySender("ana");
        System.out.println("by sender ana: " + ana.size() + " rows, nodes touched: " + table.nodesTouched);
    }
}

Step 7: Specialized stores beside the primary

Three kinds of question need data structures a general store does not keep. Each specialized store usually runs beside the main database, fed from it, not instead of it.

• Search engines (Elasticsearch, OpenSearch) keep an inverted index, from each term to the documents that contain it. They have analyzers for stemming and typos, and BM25 relevance scoring. A write becomes searchable after the next refresh, 1 second by default. They are not a system of record. They have no transactions, and the index is rebuilt from the source when the mapping changes. The search page goes deeper.
• Time-series and columnar stores (Prometheus, ClickHouse, Apache Pinot, TimescaleDB) store each column separately, sorted by series and time. They compress hard. Facebook’s Gorilla encoding averages about 1.4 bytes per sample. A query that sums one column over a billion rows reads only that column. And a server scans billions of values a second. Point lookups and updates are slow, and inserts want batches.
• Graph databases (Neo4j, Amazon Neptune) store each node with direct pointers to its edges. So a hop follows a pointer, instead of an index lookup. A friends-of-friends-of-friends query costs the edges it visits, not the size of the graph. Most deployments keep the whole graph on every server, and one leader takes every write.

Many large “graphs” are not served by graph databases. Facebook’s social graph lives in sharded MySQL, behind its TAO cache. That’s because almost every query is one hop from a known node: “friends of user 42”, “likes on post 9”. A graph database earns its place when queries walk three or more hops.

Step 8: Distributed SQL (NewSQL)

Google Spanner, CockroachDB and YugabyteDB keep the relational model and its transactions, while spreading data over many machines. Rows are sorted by primary key, and split into ranges (512 MiB by default in CockroachDB). Each range is a consensus group of three or five replicas: Paxos in Spanner, Raft in CockroachDB. Transactions that touch several ranges commit with two-phase commit across those groups.

Spanner adds TrueTime, a clock API with bounded uncertainty. It waits out that uncertainty (a few milliseconds) before acknowledging a commit. That gives external consistency: if transaction A commits before B starts, B sees A.

What it costs:

• Latency. Every write is a consensus round. In one region, that is a few milliseconds. With replicas in several regions, every commit waits on a quorum of regions. That’s tens of milliseconds when they sit on one continent, and 100 ms or more when the quorum spans continents. Add the trip to the range’s leader, if the writer is far from it. Reads stay local only if they accept slight staleness (follower or stale reads), or the data is pinned near its readers.
• Money. More machines for the same load, or a managed price that starts above a single primary’s.
• Hot ranges. A monotonically increasing key (a timestamp, a sequence) sends every insert to the last range. Hash or reverse the key, as the unique ID generator discusses.
• Cross-node joins cost network hops that a single machine never paid.

It wins when a system needs transactions that span shards, or must survive the loss of a region with zero data loss. And when the team does not want to build sharding itself.

Step 9: Several stores, one truth

Real systems end up with several stores: Postgres for orders, Redis for sessions, Elasticsearch for search, a columnar store for analytics. This is polyglot persistence, and its cost is consistency. Each extra store holds a copy, and copies drift unless one store is the source of truth and the others are derived from it.

The tempting design is a dual write: the service writes Postgres, then writes Elasticsearch. Now say the process dies between the two. Or the second write times out. Or two updates to the same order reach the index in the opposite order. The stores disagree, and nothing notices. A distributed transaction across the two is not on offer, because Elasticsearch has no prepare phase.

The fix is to write once. There are two ways:

• The service writes the event into an outbox table, in the same transaction as the change. A relay publishes it.
• A change data capture (CDC) tool, such as Debezium, reads the database’s own log (the Postgres WAL, the MySQL binlog). It publishes every committed change.

Either way, the derived store receives every change at least once: late, not never. It applies each change only if it is newer than what it holds.

One write, every store follows: outbox or CDC into a stream, consumers apply by version

Scroll or pinch to zoom, drag to pan

• Color key
• Database
• Queue or stream
• Service

JDatabase.javaPydatabase.pyC#Database.csGodatabase.goTSdatabase.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

import java.util.*;

// The transactional outbox: how a second store (a search index here) follows the source of
// truth without a dual write. The order row and the event that describes it commit together.
record OutboxEvent(long id, String orderId, long version, String status) {}

record Order(String id, long version, String status) {}

class Database {
    private final Map<String, Order> orders = new HashMap<>();
    private final List<OutboxEvent> outbox = new ArrayList<>();
    private final Set<Long> published = new HashSet<>();
    private long nextEventId = 1;

    // One transaction: the row change and its event are both kept or both lost. Writing to
    // the index here instead would be a dual write: a crash between the two leaves them apart.
    synchronized void setStatus(String orderId, String status) {
        Order old = orders.get(orderId);
        long version = old == null ? 1 : old.version() + 1;
        orders.put(orderId, new Order(orderId, version, status));
        outbox.add(new OutboxEvent(nextEventId++, orderId, version, status));
    }

    // Oldest first, so one order's events leave in the order they were written.
    synchronized List<OutboxEvent> unpublished() {
        return outbox.stream().filter(e -> !published.contains(e.id())).toList();
    }

    synchronized void markPublished(long eventId) { published.add(eventId); }
}

class SearchIndex {
    final Map<String, Order> docs = new HashMap<>();
    int applied, ignored;

    // Idempotent and order-safe: a document only moves to a higher version, so a redelivered
    // or late event changes nothing. The relay may repeat itself; the index does not care.
    void apply(OutboxEvent e) {
        Order cur = docs.get(e.orderId());
        if (cur != null && cur.version() >= e.version()) { ignored++; return; }
        docs.put(e.orderId(), new Order(e.orderId(), e.version(), e.status()));
        applied++;
    }
}

class Relay {
    // At least once: publish, then mark. A crash between the two means the event goes out
    // again after restart. Late or repeated, never lost.
    void runOnce(Database db, SearchIndex index, boolean crashBeforeMark) {
        for (OutboxEvent e : db.unpublished()) {
            index.apply(e);
            if (crashBeforeMark) return; // the process dies here
            db.markPublished(e.id());
        }
    }
}

class Main {
    public static void main(String[] args) {
        Database db = new Database();
        SearchIndex index = new SearchIndex();
        Relay relay = new Relay();
        db.setStatus("o-1", "PLACED");
        db.setStatus("o-1", "PAID");
        db.setStatus("o-2", "PLACED");

        relay.runOnce(db, index, true);  // publishes event 1, crashes before marking it
        relay.runOnce(db, index, false); // restart: event 1 again, then 2 and 3

        System.out.println("index: " + index.docs.get("o-1") + ", " + index.docs.get("o-2"));
        System.out.println("applied " + index.applied + ", ignored " + index.ignored
                + ", left in outbox " + db.unpublished().size());
    }
}

The derived stores lag the source by the relay’s delay, typically under a second and minutes during an incident. Any read that must see the user’s own write goes to the source of truth.

Options and tradeoffs

Pick a workload, or set the requirements yourself. Then open any row for its reasons. The summary picks the store with no blockers and the fewest costs, counting the cost of running it at all. So Postgres wins whenever nothing rules it out.

Load a workload

Payments ledgerChat historyCatalog with searchMetricsMulti-region bankSocial graphAlways-writable cart

Access patterns (pick any)

Get by keyRange within a keyAd hoc filters and joinsFull-text searchMulti-hop relationshipsAggregates over time

Consistency

Eventual is fineRead your writes per keyTransactions across rows

Peak writes

Under 5k/s5k to 50k/sOver 50k/s

Data size

Under 1 TB1 to 10 TBOver 10 TB

Regions

One regionReads in manyWrites in many

✓ fits! fits, at a cost✕ does not fitPick a row to see every reason.

The data model

Option	Shape	Wins when	Costs
Relational (Postgres, MySQL)	Normalized tables, joins, secondary indexes	Invariants across rows, questions that change, under one primary’s load	One primary for writes; sharding is yours to build
Key-value (DynamoDB, Redis)	Value by key, sometimes a sorted range	Every access names the key: sessions, carts, feature flags, counters	Each new query is a new index or table; Redis costs RAM and durability
Document (MongoDB)	One aggregate per document	Aggregates read and written whole, with filters on many fields	Cross-document invariants are expensive; schema drifts without validation
Wide-column (Cassandra, Bigtable)	Partitions of rows sorted by clustering key	Huge write rates, time-ordered data per key, writes in every region	One table per query, no transactions, last write wins

TradeoffRelational by default, another model for a named access pattern

We gain

One store answers questions nobody planned for, enforces invariants itself, and is the system every engineer already knows how to run, back up and debug.

We give up

Writes stop at one primary. A pattern such as 700,000 appends a second, or writes in three regions, eventually forces a second model. The migration is cheaper if the access patterns were written down from the start.

Specialized stores

Each answers a different question, so the decision is whether one earns its pipeline.

Go deeperSearch, columnar and graph stores side by side1 min

Option	Data structure	Wins when	Costs
Search engine (Elasticsearch, OpenSearch)	Inverted index, per-field doc values	Relevance-ranked text, facets, log search	Not a system of record; refresh lag; reindex on mapping changes
Columnar or time-series (ClickHouse, Pinot, Prometheus)	Compressed columns sorted by key and time	Aggregates over billions of rows, metrics, dashboards	Slow point lookups and updates; wants batched inserts
Graph (Neo4j, Neptune)	Nodes with direct edge pointers	Queries of three or more hops: fraud rings, recommendations	One writer, hard to shard, a separate skill set

TradeoffA secondary store fed by CDC, not a second source of truth

We gain

Each question runs on the structure built for it, and the source of truth keeps its transactions. A broken index is rebuilt from the source, not repaired by hand.

We give up

Another system to run, a pipeline that can stall, and reads that lag by seconds. Every consumer must be idempotent and must handle events for the same key arriving twice.

Scaling a relational database past one primary

Option	How it scales writes	Wins when	Costs
Bigger machine, replicas, cache	It does not: it buys headroom	Under about 10,000 writes a second, which is most systems	One region for writes, failover takes seconds to a minute
Sharded Postgres or MySQL (Citus, Vitess, app-level)	Each shard has its own primary	One key owns almost every query: user, tenant, account	Cross-shard queries fan out, cross-shard transactions need 2PC, resharding is a project
Distributed SQL (Spanner, CockroachDB)	Ranges with their own consensus groups	Cross-shard transactions are common, or RPO zero across regions	Consensus latency on every write, higher price, hot ranges on monotonic keys
Move the hot pattern to NoSQL	Partitions spread across nodes	The hot pattern is a key-value or time-ordered append	That data loses joins and transactions; two stores to keep in step

TradeoffShard by the main key before reaching for distributed SQL

We gain

Instagram, Facebook, YouTube and Uber ran their largest datasets on sharded MySQL or Postgres. Each shard is a familiar database, and queries that carry the key never notice the sharding.

We give up

Anything that crosses shards, a global unique constraint, a transfer between two tenants, a report, becomes application code. When that is most of the workload, distributed SQL is the cheaper choice despite its latency.

Keeping several stores in step

Option	How	Wins when	Costs
Dual write	The service writes each store in turn	Never, for data that must agree	A crash or timeout between writes leaves them apart, silently
Transactional outbox	An outbox row in the same transaction, published by a relay	You control the service and want explicit events	An extra insert per change; the relay is another process to run
Change data capture (Debezium)	A connector reads the WAL or binlog	Many consumers, or services you cannot change	Events mirror table rows, not business meaning; a stalled slot fills the primary’s disk

TradeoffOutbox for business events, CDC for mirrors

We gain

Both make the derived store late, never wrong: every committed change is published at least once, in commit order per key.

We give up

At least once means duplicates, so every consumer checks a version or event id. The lag is real, so a user who just saved and then searches may not find their own change.

How it fails

Failure	What breaks	How you detect it	Blast radius	Fix
Hot partition	One key (a celebrity, a big tenant, today’s date) takes most writes; its partition throttles	Per-partition throttling, one node’s CPU far above the rest	Every key on that partition	Spread the key: add a suffix bucket, split the tenant, never partition by date alone
Scatter-gather query	A query without the partition key asks every node; one slow node sets the latency	p99 tracks the slowest node; query fan-out metrics	Every node, on every such query	A second table or index keyed for that query, or move it to a search or columnar store
Unbounded wide partition	One partition grows to gigabytes; reads time out, compaction stalls, repair fails	Partition size histograms, large-partition warnings in logs	The replicas holding it	Add a time bucket to the partition key, such as (chat_id, month)
Replica lag	A user saves, then reads from a replica and sees the old value	Replication lag in seconds, alerts at a threshold	Reads routed to lagging replicas	Read your own writes from the primary for a short window, or route by a version the client holds
Vacuum falls behind (Postgres)	Dead row versions bloat tables; in the worst case transaction IDs near wraparound and writes stop	Dead tuple counts, age of the oldest transaction ID	The whole database	Tune autovacuum per hot table, kill long-running transactions, monitor XID age well before 2 billion
Derived store drifts	A dual write failed halfway, or a consumer applied events out of order	Periodic count and checksum comparison against the source	Searches and dashboards show wrong data	Outbox or CDC with version checks; rebuild the index from the source

Go deeperFour more failures, store by store1 min

Failure	What breaks	How you detect it	Blast radius	Fix
Tombstone pile-up	Deletes in Cassandra write tombstones; a read scanning 100,000 of them fails	Tombstone warnings, failed reads on queue-like tables	Reads of that partition	Do not use a wide-column table as a queue; use TTLs with time-window compaction
Connection exhaustion	Each Postgres connection is a process using several MB; thousands of app threads overwhelm it	Connection count near max_connections, rising latency	Every client of the database	A pooler such as PgBouncer in transaction mode; a few hundred server connections at most
Stalled CDC slot	A dead consumer keeps a Postgres replication slot; the primary keeps all WAL for it	Retained WAL size per slot	The primary’s disk, then all writes	Alert on retained WAL, cap it with max_slot_wal_keep_size, drop abandoned slots
Failover loses writes	An async replica is promoted before it received the last writes (Redis, async Postgres)	Gaps in sequence numbers after failover, client errors	Writes in the last second before the crash	Synchronous replication for data that must not be lost, or accept and reconcile

The two failures that cause the most damage are the quiet ones. One is a derived store that slowly diverges. The other is a Postgres instance whose vacuum has been losing ground for weeks. Both are visible long before they hurt, on dashboards nobody looks at until the outage.

At scale

Numbers per node. These are order-of-magnitude planning figures for current hardware, not benchmarks. They move by several times with row size, indexes, consistency level and configuration. So use them to decide between one node and fifty. Measure your own workload before sizing anything finer.

Store	Writes per node	Reads per node	Latency	Practical size per node
Postgres or MySQL primary	About 10,000 indexed write transactions/s	Tens of thousands to a few hundred thousand cached point reads/s	Under 1 ms server side, cached	A few TB comfortably, 10+ TB with pain
Cassandra or ScyllaDB node	About 10,000 to 30,000/s (ScyllaDB claims several times more)	Similar to writes for single-partition reads	1 to 5 ms	1 to 2 TB per node is common
Redis shard	100,000 to 200,000 simple commands/s, 1M+ pipelined	Same	Under 1 ms	Bound by RAM, tens of GB
DynamoDB partition	1,000 write units/s	3,000 read units/s	Single-digit ms	10 GB per partition, table unbounded
Elasticsearch data node	Several thousand to tens of thousands of docs/s, bulk	Hundreds to thousands of searches/s	10 to 100 ms	A few TB of index
ClickHouse server	Millions of rows/s, in batches	Billions of values scanned/s	10 ms to seconds	Tens of TB compressed

At 10 times the load, the relational default usually hits its first real limit. Writes reach tens of thousands a second, so the busiest table gets partitioned or the workload gets sharded by its main key. Search and analytics leave the primary for stores fed by CDC, and a cache absorbs the hottest reads. This is where the Instagram and payment designs on this site shard.

At 100 times, the questions change. Resharding has to happen online. So plan for many logical shards (thousands), mapped onto fewer physical machines. Then growth moves logical shards, instead of rehashing rows. Multi-region becomes a requirement. With it comes a choice of three:

• A home region per user (sharded relational).
• Last-write-wins in every region (wide-column).
• Consensus across regions (distributed SQL).

The partitioning and replication pages cover those mechanics.

What to monitor, in order of how often it pages someone:

1. Replication lag and retained WAL per replication slot.
2. p99 latency per query shape, not just the average, and the top queries by total time.
3. Connections in use against the pool limit.
4. For Postgres: dead tuples, autovacuum progress and the oldest transaction ID’s age. For Cassandra: pending compactions, tombstones per read and the largest partitions.
5. Throttled requests per partition (DynamoDB) or per node, which is how hot keys show up.
6. Disk headroom, with weeks of warning, not hours.
7. Lag between the source and every derived store, checked end to end with a canary write.

In the interview

Interviewer asksWhich database would you use for this system, and why?

Answer from the access pattern table, not the product name. “Three of our four patterns are point lookups and conditional updates on bookings, at about 2,000 writes a second at peak. That fits one Postgres primary, which sustains about 10,000 indexed write transactions a second. And it gives us the transaction we need, to hold two seats at once. The search pattern goes to Elasticsearch, fed by CDC from Postgres, so the booking path never depends on it. Two things would change the answer. Writes past roughly 10,000 a second sustained mean sharding by event. And a requirement to accept bookings in several regions, with no data loss, means distributed SQL.” Then name what you give up with the choice.

Interviewer asksYour Postgres primary is at 85% CPU at peak. What do you do?

Find out what the CPU is doing, before changing architecture. Look at the top queries by total time. A missing index, or a sequential scan on a hot path, is the most common cause. It costs an hour to fix. If the load is reads, add replicas and a cache. If it is writes, batch them, and buy a bigger machine, which costs one failover. Only when sustained writes approach what one primary can do, shard by the key almost every query carries. Use many logical shards, so the next step is moving shards, not rehashing.

Interviewer asksHow do you keep the search index in sync with the database?

Never by writing both from the service. The order change and an outbox row commit in one transaction, or Debezium reads the WAL. Events go to Kafka, keyed by order id, so one order’s changes stay in order. The indexer applies an event only if its version is newer than the document’s. So duplicates and late events do nothing. Monitor end-to-end lag with a canary write. Compare counts and checksums nightly. Keep a full rebuild from the source as the recovery for anything else. Reads that must see the user’s own write go to Postgres.

Interviewer asksWhen would you choose Spanner or CockroachDB over sharded Postgres?

When transactions across shards are the normal case, not the exception. For example, transfers between any two accounts. Or when the business needs writes to survive losing a whole region, with zero data loss. Distributed SQL gives serializable transactions across shards and regions, without application code. The price is a consensus round on every write. That’s a few milliseconds in one region, tens of milliseconds across nearby regions, and 100 ms or more across continents. Add a higher bill, and hot ranges if keys increase monotonically. If one key, such as tenant or user, owns almost every transaction, sharded Postgres is cheaper and faster.

Common mistake

"SQL does not scale, so use NoSQL"

Relational databases scale writes by sharding. Some of the largest systems in the world run on sharded MySQL or Postgres. What a single relational primary does not do is spread writes over machines by itself. Wide-column and key-value stores do that, by giving up joins, transactions and ad hoc queries. The real choice is which of those you can live without, given the access patterns. And at what write rate the single primary stops being enough. Below about 10,000 writes a second, it rarely is.

Common mistake

Choosing a store by the shape of the data

“The data is JSON, so MongoDB” and “it is a social network, so a graph database” pick a store by how the data looks. The queries decide. JSON fits a jsonb column, next to relational data. A social network whose queries are one hop from a known user is a key lookup. That’s why Facebook serves its graph from MySQL, behind a cache. Write the access patterns first. The store follows.

Mid-level covers

Picks a reasonable store and gives a reason: Postgres for transactions, Cassandra for heavy writes, Redis for a cache. Knows what a partition key is.

Senior adds

Starts from the access patterns, with rates and freshness needs. Sizes them against per-node numbers. Defaults to one relational primary, until a named signal says otherwise. Says what would change the choice. Treats every extra store as a copy to keep in step, and explains how: outbox or CDC, version checks, a rebuild path, and a lag metric.

Seen in

• Payment system: Postgres, because the payment state, the ledger lines and the outbox row must commit in one transaction. That’s 3,600 transactions a second at peak, against one primary’s 10,000. It’s sharded by payment at ten times the load.
• Hotel reservation: Postgres books a stay with conditional updates on every night, in one transaction, sharded by hotel later. DynamoDB transactions are the managed alternative.
• Ticketmaster: Postgres holds seats with conditional updates that are safe under READ COMMITTED, and is sharded by event later.
• Job scheduler: Postgres with partial indexes and SKIP LOCKED, partitioned from day one because 100,000 run writes a second at peak is ten primaries’ worth.
• Notification system: Postgres for user preferences behind a cache, Cassandra for the delivery log that takes a write per status change.
• Chat: a wide-column store partitioned by conversation and clustered by sequence number, exactly the “latest 50” pattern from the top of this page.
• Ad click aggregator: Postgres for raw clicks until the rate passes one primary, then Apache Pinot, a columnar store, for the rollups.
• File sync: sharded MySQL or Postgres by namespace rather than a wide-column store, because moving a folder must be atomic.
• News feed: Cassandra for posts, sharded MySQL for the follow graph, as Facebook does.
• Instagram: Postgres sharded by user, 8,192 logical shards on 64 physical primaries.
• Metrics and monitoring: a purpose-built time-series database with blocks on object storage, after ruling out a wide-column store.
• Proximity service: Postgres as the source of truth, with the search index fed by CDC through Kafka.
• Video streaming: sharded relational metadata, and a wide-column store for counters and watch positions.
• Key-value store builds the inside of a Dynamo-style store: an LSM tree, consistent hashing and leaderless quorums.

Recap

The method. List the access patterns, with rate, shape and freshness. Size them against per-node numbers. Start with one relational primary. Leave it only for a named signal: write rate, operating size, working set, geography, or a different kind of question.

Numbers. Postgres primary: about 10,000 indexed write transactions a second, a few TB comfortably. Cassandra: 10,000 to 30,000 writes a second per node, partitions under 100 MB. Redis: 100,000 to 200,000 commands a second per core. DynamoDB: 1,000 writes a second per partition. Distributed SQL: a few ms per commit in one region, tens of ms across one continent, 100 ms or more across continents. All of these are orders of magnitude, not benchmarks.

The models. Relational for invariants and changing questions. Key-value when every access names the key. Document for aggregates read whole. Wide-column for huge, time-ordered writes per key. Search, columnar and graph stores beside the primary, fed by CDC. Distributed SQL when transactions cross shards or regions.

The traps. Dual writes, queries without the partition key, monotonic keys on range-sharded stores, unbounded partitions, vacuum and replication slots nobody watches.

Further reading

• On this site: indexes for B-trees and LSM trees, replication, partitioning and sharding, consistency models, CAP and PACELC, distributed transactions and search.
• Designing Data-Intensive Applications, Martin Kleppmann. Chapters 2, 3 and 7 cover data models, storage engines and transactions, in more depth than any other single source.
• Transaction isolation, PostgreSQL documentation: exactly what each level prevents in Postgres.
• Pattern: Transactional outbox, microservices.io: the outbox and the ways to relay it.