R10 — Storing data: choosing stores, replication, partitioning and sharding

Every mechanism so far has assumed a database underneath it that keeps its promises. This section is about that database: which kind to pick, how copies of it stay in step, and how data is split once one machine is no longer enough. R11 picks up what to do when the store you chose becomes the bottleneck.

The story: the ledger and the read model want different things

A customer sends a payment. Two pieces of data change.

1. The ledger posts a debit and a credit. Two entries that must both exist or neither, against a balance that must never go below what the rules allow.
2. The customer's activity feed gets a new line. "You paid Asha ₹1,200." A denormalised row in a read model, fed by the event from R4.

Now imagine the network between two data centres drops for thirty seconds.

For the ledger, the right behaviour is to stop. A ledger that keeps accepting debits on both sides of a split can approve two payments against the same balance, and nothing afterwards can make that money exist. Refusing a request is an inconvenience. Posting a wrong entry is an incident.

For the feed, the right behaviour is to keep serving. Showing a feed that is thirty seconds behind costs nothing. Showing an error page costs a support call.

Same payment, same moment, opposite answers. That is the whole of this section in one example: the store is chosen per kind of data, not per company, and the question that decides it is what that data must do when something breaks.

Choosing a store

SQL and NoSQL: what each one is actually good at

The labels are blunt, but the trade-off underneath them is real.

A relational database stores structured rows in tables, enforces a schema, gives you joins and ad-hoc queries, and wraps work in ACID transactions (R0). It scales up easily and out with effort — sharding a relational store is a project, not a setting.

A NoSQL store gives up some of that in exchange for horizontal scale, throughput and availability. Schemas are flexible or absent, access patterns are simple and must be known in advance — each store is built to answer a particular kind of question fast — and the default consistency is eventual rather than transactional — often described as BASE, the deliberate opposite of ACID, which the next subsection unpacks.

They are used together far more often than instead of each other. A social product keeps users in a relational store and the social graph and posts in NoSQL. A payments platform keeps the ledger relational and puts sessions, idempotency caches and feeds wherever they are cheapest.

The trap to name: "NoSQL scales, SQL doesn't." Relational databases scale a very long way vertically, and R13 shows how far a single well-tuned primary goes before anything else is needed. What NoSQL buys is not scale in general — it is scale without the transaction boundary. If your data needs that boundary, you are buying the wrong thing.

BASE: what a NoSQL store promises instead of ACID

ACID is a precise promise about one transaction: it happens completely or not at all, it leaves the rules intact, it is not disturbed by other transactions, and once committed it stays committed. BASE is a looser promise about a replicated store: it keeps answering even when its copies disagree, and it lets them agree later. The acronym was chosen to be the chemical opposite of ACID, and the fit is a little forced. Learn the three ideas behind it rather than the letters.

To make them concrete, take one example and keep it for the rest of this subsection. A shopping cart lives in a store that keeps three copies of every key, on three machines called A, B and C. Each copy is a replica. To keep writes fast, the store tells the client "saved" as soon as one replica has the write, and copies it to the other two in the background. (The replication chapter below explains these choices properly. One idea is enough for now: the copies are not updated at the same instant.)

"Soft state" is the letter people get wrong. It does not mean the data is kept only in memory, and it does not mean data is lost. It means the state is not settled yet: background processes are still moving it towards agreement, so two reads a second apart can disagree with no write in between. ACID's state is "hard" in comparison. Once a commit returns, every later read sees it until someone changes it.

Step through the cart below. Watch three things: the write accepted while a replica is down, the read that sees an empty cart, and the background repairs that bring all three copies to the same answer. The last tab shows what happens when two people change the cart at the same moment on opposite sides of a network failure.

ACID and BASE, question by question

How "eventually" actually happens

Convergence is not magic. Three background mechanisms do the work, and a BASE store runs all of them:

• Hinted handoff. When a replica is down, a live replica keeps the write on its behalf as a "hint", and delivers it when the replica returns.
• Read repair. A read that asks several replicas compares their answers. If one is stale, the newest value goes back to the client and is also written to the stale replica.
• Anti-entropy repair. A scheduled job compares whole replicas, using Merkle trees so it does not have to compare every row, and fixes every difference. This is what catches keys that nobody reads.

R6 covers read repair and anti-entropy in detail, because they are reconciliation inside the database.

Repair handles copies that are merely behind. Concurrent writes are harder: two clients change the same key on different replicas before either change has spread. Something has to decide the result:

"Eventually" has no deadline.

On a healthy cluster, the copies usually agree within milliseconds. But the promise has no time limit. Cassandra keeps hints for a replica for three hours by default. A replica that is down longer than that misses those writes until anti-entropy repair runs. A deleted row is kept as a marker (a tombstone) for ten days by default, and a replica that misses the delete and is not repaired within that window can bring the deleted data back. Running repair on a schedule is part of operating a BASE store, not an optional extra.

BASE is a dial, not a label on the box

Most BASE stores let you buy stronger consistency for a single request when you need it. In Cassandra you choose a consistency level per query, and QUORUM reads and writes overlap so that a read sees the latest acknowledged write. In DynamoDB, a read on the table can ask for ConsistentRead, and multi-item transactions are available at a higher cost. The quorum arithmetic below shows exactly when that overlap holds.

It works the other way too. A relational database read through an asynchronous replica behaves like a BASE store for that read: the replica can be behind, and R11's stale-read problem is exactly that. So BASE describes how a particular read or write behaves, not which product you bought.

What BASE hands to your application

• Make every write idempotent, so retries and redeliveries are harmless (R3).
• Prefer changes that merge — add to a set, increment a counter, append an event — over overwriting a whole value.
• Put a version or timestamp on every record, so a conflict can at least be detected.
• Design the screen for staleness: route the user who just wrote to a fresh read, and let everyone else see "updated a few seconds ago" (R11).
• Reconcile anything that matters against a source of truth (R6).
• Keep rules that must never break out of BASE stores. If nothing in the database enforces "no negative balance", nothing does.

BASE gives up the promise of a correct answer now in exchange for an answer now. Choose it where a slightly old answer harms nobody.

In payments, that line falls where you would expect: a feed, a notification history or a product catalogue can be BASE; a balance, a hold or an idempotency record cannot. The CAP section below draws the line precisely.

The four NoSQL families

"NoSQL" is not one thing. Name the family, because each is built for a different kind of access pattern — and, as the next subsection shows, most of them expect you to know those patterns before you design anything.

Columnar stores — Redshift, BigQuery, ClickHouse — are a different thing despite the similar name: they store each column separately for analytics scans, and the wide-column deep dive below sets the two side by side.

Two details worth having ready. Graph databases can be ACID-capable, so "NoSQL" does not automatically mean "no transactions." And a document store's format is JSON or something like it, which is what makes it attractive when the schema genuinely is not fixed — and dangerous when it is fixed and nobody wrote it down.

Access patterns come first: designing a NoSQL store backwards

An access pattern is one specific question the application asks of its data, with its inputs and the order it wants the answer in. "Fetch payment P123." "List customer C42's payments, newest first, 20 at a time." "Total a merchant's payments for one day." Writing these down is the first step in choosing and designing any store. For most NoSQL stores, it is the step the whole design depends on.

A relational database lets you model the data first and ask questions later. You describe the entities — customers, payments, merchants — normalise them into tables, and write whatever query you need, today or next year. The query planner combines joins and indexes to answer questions nobody planned for. It may be slow until you add an index, but it can answer.

A key-value or wide-column store works the other way round. You list every question first, then shape the data so that each question is answered by one lookup on a key, or one range read inside a partition. The reason is how these stores find data. Rows are spread across machines by their key. There are no joins. A query that does not name the key has no machine to go to. The store must then read every partition, refuse, or make you opt in explicitly — Cassandra's ALLOW FILTERING exists to make you feel that cost.

Take the three payment questions above in Cassandra. Each one gets its own table, laid out for exactly that question:

The same payment is written three times. That duplication — denormalisation — is the price of making every read a single fetch. In DynamoDB the same idea appears as a partition key and sort key chosen for the main question, plus secondary indexes for the others.

Now add a question nobody listed: "every failed payment over ₹1 lakh last week, across all customers." No table is keyed that way. The fix is a new table or index, plus a backfill of the old data, or sending the events to a search engine or warehouse that handles ad-hoc questions. In Postgres it would have been a WHERE clause and perhaps an index.

In a relational store you model the data and ask anything; in a key-value or wide-column store you model the questions, and the data follows.

Each family suits a different kind of question

How strictly the questions must be known in advance depends on how much the store relies on the key to find data:

So the rule is a gradient, not a switch. Key-value and wide-column stores want the questions fixed before the first table exists. Document stores are more forgiving. Graph stores are built for questions you discover later, as long as they are questions about relationships.

The trap: choosing a NoSQL store before listing the access patterns.

"Schemaless" does not mean "no design". It moves the design from the data to the queries, and it makes changing your mind expensive. A team that picks a wide-column store for flexibility, then discovers a reporting requirement, ends up running a second store to answer it.

How to handle access patterns in a design round

• List the access patterns out loud before naming any store: each question, its inputs, its sort order, and how often it runs.
• Mark which are hot (thousands per second) and which are rare (a nightly report).
• Choose the store that serves the hot questions with one key lookup, and say how each one maps to a key.
• Send the rare, ad-hoc questions somewhere built for them — a search engine or a warehouse fed by events — instead of bending the main store.
• Say what happens when a new question arrives: a new table or index, and a backfill.

Wide-column stores in depth, with Cassandra as the reference

Wide-column is not columnar

The names are close and the ideas are opposite, so settle this first.

"Wide" refers to the partition, which can hold a very large number of rows, each with its own set of columns. Cassandra is the wide-column store this subsection uses.

How Cassandra stores and finds data

Every table has a primary key with two parts, and each part has one job.

• The partition key decides which machines hold the row. Cassandra hashes it to a number (a token) and places it on a ring of nodes. The node owning that range, plus its replicas, holds every row with that partition key. The consistent hashing chapter below explains the ring.
• The clustering columns decide the order of rows inside the partition. Rows are stored sorted by them on disk, so "the newest 50" or "everything between two times" is one sequential read.

Together they give the one query shape Cassandra is built for: go to one partition, then read a contiguous slice of it in sorted order.

Writes are cheap by design. A write is appended to a commit log for durability and added to an in-memory table (the memtable). Full memtables are flushed to disk as immutable files called SSTables, and a background process called compaction merges them. There is no read before a write: an INSERT of an existing key simply overwrites it (an upsert), and a delete writes a marker called a tombstone. This structure is a log-structured merge tree. It is why Cassandra absorbs very high write rates, and why every SSTable carries a Bloom filter (end of this section) to keep reads fast.

Reads go to the partition's replicas, merge the memtable and the relevant SSTables, and keep the newest value of each cell by timestamp. You choose per query how many replicas must answer, using the consistency levels from the quorum arithmetic below.

What CQL lets you ask follows directly from that layout:

A worked example: message history in a chat app

Chat history is the classic Cassandra workload. Discord, for example, has described storing its messages in Cassandra with a design very close to the one below. Start from the access patterns, as the previous subsection insists:

Why Cassandra fits: the writes are relentless appends, every read is "one channel, in time order", the data grows forever, and a message appearing a moment late on one device harms nobody. This is AP data (the CAP section below).

Things the app will also want, but not from this table: searching message text, and counting a channel's messages. The first goes to a search engine fed by events; the second to a counter kept separately.

First attempt: one partition per channel.

CREATE TABLE messages (
  channel_id  bigint,
  message_id  bigint,    -- Snowflake-style id: unique and ordered by time (R0)
  author_id   bigint,
  content     text,
  PRIMARY KEY ((channel_id), message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

The double parentheses mark the partition key; message_id is the clustering column, stored newest first. Q1 is now WHERE channel_id = ? LIMIT 50, and Q2 adds AND message_id < ?. Both are a single slice of a single partition.

The flaw: a busy channel's partition grows without limit.

Every message ever posted to #general lands in one partition, on the same few replicas. After a few years it holds millions of rows and many gigabytes. Those replicas become hot spots, and compaction, repair and node replacement all slow down around it. A common rule of thumb is to keep a partition under about 100 MB.

Second attempt: bucket each channel's history by time.

Add a time bucket to the partition key, so each channel is split into a series of bounded partitions:

CREATE TABLE messages (
  channel_id  bigint,
  bucket      int,       -- which 10-day window, computed from the time inside message_id
  message_id  bigint,
  author_id   bigint,
  content     text,
  PRIMARY KEY ((channel_id, bucket), message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

Size the bucket so that the busiest channel's messages for one window stay under the partition limit. Ten days is the figure Discord described. Every query now names both parts of the partition key:

-- Q1: latest 50 — start in the current bucket
SELECT * FROM messages WHERE channel_id = ? AND bucket = ? LIMIT 50;

-- Q2: the 50 before message X — X's own bucket comes from X's id
SELECT * FROM messages WHERE channel_id = ? AND bucket = ? AND message_id < ? LIMIT 50;

-- Q3: post — an upsert, with no read first
INSERT INTO messages (channel_id, bucket, message_id, author_id, content) VALUES (?, ?, ?, ?, ?);

-- Q4: edit — the bucket is again derived from the message id
UPDATE messages SET content = ? WHERE channel_id = ? AND bucket = ? AND message_id = ?;

If a read returns fewer than 50 messages, the application reads the previous bucket and continues until it has 50. This is why the time-ordered id matters: the bucket of any message can be computed from its id, so no lookup is needed. A quiet channel may have empty buckets to step over. The usual fix is to remember each channel's most recent non-empty bucket, or to stop after a few empty ones.

A new question means a new table. Suppose moderators need "a user's recent messages across all channels". Nothing above is keyed by author, so add messages_by_author, with PRIMARY KEY ((author_id, bucket), message_id), and write every message to both tables. A logged batch makes Cassandra eventually apply both writes or neither. It is not a transaction: a reader can briefly see one table updated and not the other.

Where this design bites

• Deletes are writes. Every deleted message leaves a tombstone until compaction removes it, after the grace period described under BASE. A channel where thousands of messages were deleted makes Q1 read past thousands of tombstones to find 50 live rows. Watch tombstone counts per read, and do not model queues or frequently cleared lists in Cassandra.
• Consistency is your choice per query. Writing and reading at LOCAL_QUORUM gives read-your-own-writes inside one data centre. ONE is faster, and can show a message missing for a moment.
• No ad-hoc questions. Text search, analytics and "messages containing a link last week" go to stores built for them, fed by the same events.
• The same shape fits payments read models. An account's activity feed — latest transactions, newest first, scroll back by month — is the same design with account_id in place of channel_id. The ledger behind it stays relational and CP.

Cassandra rewards one discipline: pick a partition key that spreads load and stays bounded, and a clustering order that matches how you read.

The questions that decide it

Pick the store by answering these, in this order. Each one forks the design; a question that forks nothing is not worth asking.

In payments, the first question settles the ledger before you reach the second. Double-entry postings, balances and holds need a transaction boundary, so the ledger lives on a relational ACID store. Everything around it — feeds, notification history, search, analytics — is fed from the ledger's events and can live wherever its access pattern points.

Choose the store by what the data must survive, not by what the team already runs.

CAP, applied to store choice

R0 defines CAP: when the network partitions, a replicated system chooses between consistency (every read sees the latest write) and availability (every request gets an answer). Here the useful move is to apply it per kind of data, the way the opening story did.

The CA row is there to be named and dismissed. A single node has no partition because it has no network between copies; the moment you add a replica, you are choosing between C and A again.

Two precise points interviewers listen for. First, CAP's C is not ACID's C — CAP means one up-to-date copy, ACID means the rules and constraints still hold. A store can be both, either or neither. Second, many modern stores are not one letter: DynamoDB and Cassandra let you pick consistency per request, which moves them along the dial rather than across it. PACELC (R0) is the more honest frame, because on a healthy day the trade you actually feel is latency against consistency.

Why money is CP, and what can be AP

Money data is CP because the two failure modes are not symmetric. An unavailable ledger means a customer retries in a minute. An inconsistent ledger means two debits approved against one balance, and the second one cannot be taken back by any amount of convergence. R0's rule — prefer the failure you can clean up — points straight at CP.

But most of a payments platform is not the ledger:

The pattern is: the write model is CP, the read models are AP, and the events from R4 carry changes from one to the other. That is CQRS in one sentence, and R11 covers how to run it.

The design that makes CP affordable is a single writer per account. If every write for an account goes through one home — one shard, one region — there is never a second copy to disagree with, and the partition question becomes "is the home reachable?" rather than "which copy is right?". R13 covers how writes are homed across regions.

Replication: keeping copies in step

A store is replicated for three reasons: availability (a replica takes over when the primary dies), read scale (replicas serve reads), and disaster recovery (a copy in another place). Each copy costs you one of two things — latency or freshness — and which one is decided by when the write is acknowledged.

Synchronous and asynchronous

Synchronous replication acknowledges the write only once the replicas have it. Asynchronous replication acknowledges as soon as the primary has it and ships the change afterwards.

The practical middle ground is semi-synchronous: wait for one replica (or a quorum), not all of them. You keep durability against a single failure without letting the slowest replica set your latency.

The read that bites: reading your own write from an async replica.

A customer pays, the write commits on the primary, the confirmation page reads from a replica that has not caught up, and the payment is not there. Nothing is broken; the replica is simply behind. R11 covers the fixes — read-your-writes routing, reading from the primary after a write, and version checks.

The three replication patterns

Primary–secondary is the pattern to reach for first, because it has exactly one place where a write is decided. That is the single-writer property again, and it is why the ledger sits on it.

Primary–primary trades that away for local writes everywhere. The cost is that two primaries can accept contradictory writes, and something has to decide which wins. For a profile picture, last-write-wins is fine. For a balance, there is no merge rule that makes two approved debits into one, which is why money is homed rather than multi-mastered (R13).

Quorum arithmetic: N, R and W

Leaderless stores make consistency a dial with three settings.

• N — how many replicas hold each key.
• W — how many must acknowledge a write before it counts.
• R — how many are consulted on a read.

If R + W > N, every read set overlaps every write set in at least one replica, so a read sees the latest acknowledged write. Picture three seats: a write sits in two of them, a read looks at two of them, and two groups of two out of three must share a seat.

Raising W makes writes slower and less available; raising R does the same for reads. Every row is a different way of satisfying the rule, or deliberately breaking it.

The trap to name: quorum overlap is necessary but not sufficient for linearizability. Concurrent writes resolved by last-write-wins can lose to clock skew. Sloppy quorums let writes land outside the home replicas, so a read of the home set misses them. A write that reached one replica and "failed" can reappear later through repair. And a timed-out write is not a failed write — R0's "unknown is not failed" applies here too. For true compare-and-set you need a consensus round per key, which is R8's territory. R17 walks through a full leaderless key-value store built on these settings.

When copies disagree

Replication with any asynchrony produces copies that diverge. Two jobs follow: detect the disagreement — version vectors or checksums — and resolve it — last-write-wins (which trusts clocks that NTP never perfectly syncs), merge-friendly data types (CRDTs), or a repair process. R6 covers read repair and Merkle-tree anti-entropy; the point here is only that choosing an AP store means choosing to run one of them.

Before moving on, see both replication modes side by side. Write to the primary in synchronous mode and watch the acknowledgement wait for the replica; switch to asynchronous, write again, and read from the replica before the change arrives. That stale read is the one R11 spends a section fixing.

Now cut the network. The demo below has two replicas and a partition switch. In CP mode, watch the minority side refuse the write; in AP mode, watch both sides answer and one of them answer wrongly. Ask yourself which of the two you would want behind a balance check.

Replication gives every node the whole dataset. When the dataset, or the write load on it, outgrows one machine, the next step is to split it.

Partitioning and sharding: not the same thing

These get used interchangeably and are different decisions with different costs. Getting the distinction right in the first sentence of an answer buys you a lot.

Partitioning splits one table inside one database. The rows live in separate physical chunks, but it is still one server, one connection, one transaction manager. Your application does not know it happened. Postgres calls this declarative partitioning; the planner prunes the chunks a query cannot match.

Sharding splits data across independent databases. Each shard is a separate server with its own storage, connections and transaction boundary. Nothing spans them for free — no joins, no foreign keys, no transactions — and something has to route each query to the right one.

The rule: partition early, shard late. Partitioning is a schema change with real benefits and modest risk. Sharding changes what your application is allowed to assume, permanently — and the day you shard is the day cross-account transfers stop being a transaction and start being a saga.

Choosing a partitioning scheme

Start with the problem partitioning solves. Picture ledger_entry after three years: two billion rows in one table. A query for "this month's entries for account 42" has to walk an index built over all three years. Deleting entries older than the retention period means a DELETE that runs for hours and bloats the table. Routine maintenance — vacuum, index rebuilds, backups — takes longer every month.

Partitioning fixes this by splitting the one big table into many smaller tables, called partitions, while the application keeps reading and writing a single table name. Two decisions define it:

• the partition key — the column that decides where a row goes, such as posted_at or account_id;
• the scheme — the rule that maps a key value to a partition.

Postgres, like most relational databases, offers three schemes:

1. Range — each partition owns a continuous range of values, almost always a span of time.
2. List — each partition owns an explicit list of values, such as a rail or a country.
3. Hash — each partition owns a share of the hash of a key, so rows spread evenly.

Choose the scheme from how you query the data and how you delete it. Everything below follows from that one sentence.

1. Range partitioning: split by time

How it works. Each partition holds the rows whose key falls in its range. The October 2026 partition of ledger_entry holds every row with posted_at from 1 October up to, but not including, 1 November. A row is routed to the partition whose range contains it.

Where it fits. Data that only grows and is read and retired by date: ledger entries by month, outbox rows by day, reconciliation staging by value date.

What you gain.

• Queries read less. A query for one week touches one monthly partition instead of three years of data. The planner skips the other partitions entirely; this is called partition pruning.
• Retention becomes free. Removing a month is detaching and dropping one partition — a metadata change that takes milliseconds — instead of deleting a billion rows one by one.
• Maintenance shrinks. Vacuum and index rebuilds run per partition, and old partitions that no longer change need almost none.

What it costs.

• The newest partition takes every write, so it is the hot one. That is usually fine, because it is also the one already in memory.
• Future partitions must exist before data arrives, so a scheduled job creates them ahead of time.

2. List partitioning: split by category

How it works. Each partition is given an explicit set of values: 'UPI' goes to payment_upi, 'NEFT', 'RTGS' and 'IMPS' go to payment_bank. A default partition can catch any value nobody listed.

Where it fits. A small, known set of categories that need different treatment — rail, region, tenant tier.

What you gain.

• Different rules per group. Each partition can have its own retention, its own storage, its own maintenance window.
• Data residency becomes structural. Put the Indian partition on storage in India, and the residency rule is enforced by the schema rather than by every query remembering it.

What it costs.

• Skew. If UPI carries 80% of the volume, its partition is nearly as large as the original table. List partitioning organises data; it does not balance it.
• A new category needs a new partition, or it piles up in the default.

3. Hash partitioning: split evenly

How it works. The database hashes the key and takes the remainder by the number of partitions: with four partitions, hash(account_id) % 4 picks one. Rows spread roughly evenly, whatever the key values look like.

Where it fits. Data with no natural range, looked up by an exact key, where the goal is even load — account_balance by account_id.

What you gain.

• Even spread. No partition is hot just because keys are sequential.
• Exact lookups stay fast. A query for one account goes straight to one partition.

What it costs.

• No range pruning. Hashing scatters neighbouring values, so a query by date or by a range of accounts must visit every partition.
• The partition count is hard to change. Going from 4 to 8 partitions means moving data, so choose the count with growth in mind.

The three side by side

The schemes also combine. A very large ledger can be range by month, then hash by account inside each month: monthly retention and pruning, with no single month becoming one enormous table.

Before you commit to a partition key

Four questions, in order. A "no" to any of the first three means the key is probably wrong.

1. Does your most common query filter on it? Pruning only works when the query names the partition key. If the usual query cannot, every query visits every partition, and partitioning has made things slower.
2. Does your retention policy line up with it? Data deleted by age wants partitions by time, so that deleting is dropping.
3. Where will the writes land? Range by time concentrates writes on the newest partition; hash spreads them. Decide whether that concentration is acceptable.
4. Can you live with the uniqueness rule? Postgres checks uniqueness inside each partition, not across them.

In Postgres, every unique constraint on a partitioned table must include the partition key. A payment_id that must be unique across a table partitioned by rail therefore needs either the rail in the constraint or a separate registry table keyed by payment_id. This is the detail that catches teams halfway through a migration.

Partitioning well

• Partition by time for append-only data that is queried and retired by date.
• Choose a key your most frequent query always filters on, and keep that filter on the bare column — posted_at >= …, not date_trunc('month', posted_at) = … — so the planner can prune.
• Never partition by a value that changes, such as status; every change moves the row to another partition.
• Create future partitions ahead of time, and retire old ones by detaching and dropping, never by mass DELETE.
• Keep the total number of partitions in the hundreds; planning time grows with the count.

The walkthrough below turns each scheme into SQL. Pick a scenario at the top, read the explanation on the left, and follow the statements on the right — highlighted lines are the ones that step is about. It covers range by time, list by rail and region, hash by account, sub-partitioning, the uniqueness rule and its fix, how partitioning changes query plans, and what changes when the same idea crosses servers as sharding.

How partitioning changes your queries

Partitioning is transparent to the SQL: the application still reads and writes the parent table by name, and Postgres routes each row to its partition. It is not transparent to performance, and it adds a few rules.

One question decides how a query behaves: does it filter on the bare partition key?

Queries that get faster

• Filters on the partition key — posted_at >= '2026-10-01' AND posted_at < '2026-10-08' — let the planner skip every partition that cannot match.
• Values known only at run time, such as bind parameters or now() - interval '7 days', are pruned when the query executes (Postgres 11+), reported as Subplans Removed in EXPLAIN ANALYZE.
• Retention stops being a query at all: you drop a partition instead of running a mass DELETE.

Queries that get slower or behave differently

Rules that change what you can write

• Unique constraints and ON CONFLICT targets must include the partition key. An upsert on payment_id alone is rejected on a table partitioned by rail.
• There are no global indexes. Each partition has its own, so a lookup on a non-key column probes all of them.
• Foreign keys that reference a partitioned table (Postgres 12+) must point at a unique key — which includes the partition key — so the referencing table must carry that column too.
• Updating the partition key moves the row, as a delete plus an insert: slower, it fires both sets of row triggers, and a concurrent update of the same row can fail because it has already moved to another partition.

Two maintenance surprises. CREATE INDEX CONCURRENTLY is not allowed on the partitioned parent — create the index ON ONLY the parent, build it concurrently on each partition and attach each one. And autovacuum analyzes partitions but never the parent, so the statistics used for queries spanning many partitions go stale unless you schedule ANALYZE on the parent.

Writing queries for a partitioned table

• Put the partition key in the WHERE clause of every frequent query, as a range or equality on the bare column.
• Check with EXPLAIN that the plan touches only the partitions you expect; look for Subplans Removed when parameters are involved.
• Give lookups by a non-key identifier their own small lookup table instead of fanning out across partitions.
• Include the partition key in upserts, unique constraints and foreign keys from the start, not mid-migration.
• Schedule ANALYZE on the parent table, and build new indexes partition by partition.

The Query impact tab of the walkthrough above shows each of these with EXPLAIN output and the SQL to fix it.

Choosing a shard key

This is the consequential one, because it is effectively permanent. A good shard key has six properties:

1. High cardinality — millions of distinct values, so shards can be balanced.
2. Even distribution — no value carrying a disproportionate share.
3. Present in almost every query — otherwise you scatter-gather across all shards and lose the benefit.
4. Stable — it must never change for a row, because changing it means moving the row between databases.
5. Aligned with your transaction boundary — the things that must commit together should live on one shard.
6. Aligned with your ordering requirement — whatever must stay in sequence should share a shard, for the same reason a Kafka key groups a partition (R9).

In payments the answer is almost always the account or customer, and the reasoning is properties 5 and 6. An account's ledger entries, its materialised balance, its holds and its payment history all belong together: the balance invariant is per account, the ordering requirement is per account, and the hot row is per account.

Payment ID is the tempting wrong answer. It has perfect cardinality and distribution, and it scatters one account's history across every shard in the estate — so "show me this customer's statement" becomes a fan-out, and "debit this account" can no longer be a single transaction with its own balance row.

Four anti-patterns to name:

• Monotonically increasing keys — timestamps, auto-increment IDs. Every write goes to the newest shard, which is one hot shard and N idle ones.
• Low cardinality — country, currency, rail. Four values cannot balance forty shards, and one of them is always 80% of the traffic.
• Mutable keys — anything a customer can change. A shard key that changes means a cross-shard move, which means a distributed transaction you deliberately avoided having.
• A key nobody queries by — technically perfect distribution, practically a scatter-gather on every read.

Build the key, do not just pick a column

The move that makes sharding survivable is decoupling the logical key from the physical shard with virtual buckets:

bucket   = hash(account_id) % 4096      -- fixed forever
shard    = lookup(bucket)               -- a small table, changeable

The application only ever computes a bucket. A directory maps buckets to shards, and rebalancing means moving one bucket at a time rather than rehashing the world. 4,096 buckets is small enough to cache in memory in every service and fine-grained enough to balance.

Two alternatives and why the directory usually wins:

• Pure hash routing (shard = hash(key) % N) needs no lookup, but changing N remaps almost everything. Consistent hashing softens that and still moves data on every topology change.
• A per-entity directory (account → shard, one row each) is maximally flexible — per-account overrides, residency pinning, moving one whale — but the table is enormous and read on every request.

Buckets are the middle ground, and they buy you the thing you actually need at 3am: the ability to move a slice of traffic without a migration. R28's global design uses exactly this, with a fence token on the bucket so a stale writer cannot keep writing after a move (R8).

Compose the key when one column is not enough. Multi-tenant platforms shard on hash(tenant_id, account_id) so one tenant cannot monopolise a shard. Time-series systems shard on the entity and partition within the shard by time — sharding for throughput, partitioning for retention, both at once.

What breaks once you shard, and what to do about it

Deciding you need it, and getting there

The escalation order matters, and saying it shows judgement rather than enthusiasm:

1. Vertical — a bigger instance. Fast, boring, often enough.
2. Read replicas for read load (with the replication-lag caveat).
3. Partitioning for table size, maintenance and retention.
4. Caching and CQRS read models to take load off the write path.
5. Sharding — only when writes, connections or storage genuinely exceed one machine.

The signals that you have arrived at 5: write throughput at the instance ceiling, connection counts you cannot proxy away, storage or index size beyond what fits in memory, maintenance windows that no longer fit the night, and p99 degrading with load rather than with any one query.

The migration is R16's expand-contract applied to data: dual-write to old and new topology, backfill history, run a continuous comparison, switch reads, then retire the old path. With virtual buckets you do it one bucket at a time, which means the blast radius of a mistake is a slice of customers rather than all of them — and the rollback is a directory update.

One thing to do before any of it: run the distribution against real production data. Count rows and write volume per candidate key, and look at the top 100 values. Skew that looks theoretical in a design review is usually obvious in the data, and it is much cheaper to discover it there.

How data is placed: modulo, consistent hashing and rendezvous

Once data is split across nodes — shards, cache servers, replicas in a leaderless ring — every request has to answer one question: which node owns this key? The answer has to be cheap to compute, the same on every client, and stable when the set of nodes changes. That last property is the one the obvious answer fails.

Modulo hashing, and why it breaks

The first thing everyone writes:

node = hash(key) % N

It is O(1), stateless and perfectly even. It works until N changes.

Take key 12. With 10 nodes it lives on node 12 % 10 = 2. Add one node and it moves to 12 % 11 = 1. Lose one and it moves to 12 % 9 = 3. That is not one unlucky key: changing N remaps almost every key, roughly N/(N+1) of them when you add one node.

For a cache, that is a cold start across the whole fleet at the moment you were adding capacity because of load. For a database, it is moving nearly every row between machines. At scale, a full rehash is not a migration — it is an outage.

Consistent hashing

The fix is to stop tying the answer to N.

1. Hash each node — its ID or address — onto a ring of positions from 0 to the hash maximum.
2. Hash each key onto the same ring.
3. Walk clockwise from the key's position. The first node you meet owns the key.

Now look at what a topology change costs:

• A node joins: it takes over only the keys between itself and its predecessor on the ring. Nothing else moves.
• A node leaves: only its keys move, to the next node clockwise.

That is roughly 1/N of the data per change instead of nearly all of it. The ring is held as a sorted array or balanced tree of node positions, so a lookup is a binary search: O(log N). The hash function matters less than people think — Dynamo used MD5, current systems typically use MurmurHash — as long as it spreads evenly.

The basic ring is uneven.

With one position per node, the arcs between nodes are random lengths, so one node can own far more of the ring than another. And a new node relieves only its one clockwise neighbour, not the cluster.

Virtual nodes

The fix is to give every physical node many positions on the ring — N1_1, N1_2, N1_3 and so on.

• More points smooth the distribution: arcs average out, and every physical node ends up with close to its fair share.
• A new node takes small slices from many peers, so it relieves the whole cluster and the data streams in parallel.
• Heterogeneous hardware gets weight: a bigger server simply gets more virtual nodes.

DynamoDB, Cassandra and Riak all use them. Cassandra's default moved from 256 tokens per node to 16 in version 4.0, alongside a smarter token allocator, because more tokens also means more peers involved in every repair. The trade-off is a larger token map that every node must hold.

Virtual nodes also give you replication for free: store each key on the first r distinct physical nodes clockwise from it. A newly joined node starts empty, and the next replicas along the ring cover for it while it fills.

Hot spots: the problem hashing does not solve

Consistent hashing balances keys. It does nothing about one key that receives a disproportionate share of traffic — the celebrity account, the merchant running a flash sale. One key lives on one node, and that node melts however even the ring is.

The fixes live above the placement layer: an application-level cache in front of the hot key (R12), extra replicas serving its reads, a CDN for public content, or splitting the key itself — the whale treatment in the sharding table above. In payments, a hot account is also a hot row, and R11 covers sharded counters for that.

Rendezvous hashing (highest random weight)

A different answer to the same question, with no ring at all.

For each key, compute a weight for every server — weight = hash(key, server) — and pick the server with the highest weight. When a server leaves, only the keys it was winning move, each to its own second-highest choice. When a server joins, it wins only the keys whose new highest weight is its own. Unequal capacity is handled by scaling the weights.

It is simple to implement, needs nothing in memory beyond the server list, and moves the minimum on change. The price is O(N) per lookup, because every server is scored. That makes it a good fit for small and medium clusters — load balancers, Memcached clients, distributed hash tables — and a poor one for thousands of nodes.

The comparison

How this connects to the bucket directory above. Virtual buckets are the same idea taken one step further: hash to a fixed number of buckets, then map buckets to shards with a table you control. Consistent hashing decides placement by arithmetic; a bucket directory decides it by a lookup you can edit at 3am. For a payments estate, where moving one account deliberately matters more than moving none automatically, the directory usually wins — as the sharding section argues.

Walk through it step by step. The walkthrough below starts with the modulo problem and builds the ring one piece at a time — hashing nodes, hashing keys, the clockwise walk, a node joining and leaving, then virtual nodes and the binary-search lookup. At the join and leave steps, count how many keys actually move.

Then route some traffic. The shard router below sends keys through range and hash sharding. Feed it sequential IDs under range sharding and watch every write pile onto the last shard — the monotonic-key anti-pattern from the shard key list, happening live. Switch to hash and watch the load spread, then try a range query and see it touch every shard.

Bloom filters: asking "is it here?" without looking

A different storage problem, and one that shows up inside the stores you choose. You need to know whether a key is in a very large set without reading the set. A read in an LSM store is the classic case: the key could be in any of dozens of immutable files on disk, and opening each one to find out is most of the cost of the read.

A Bloom filter answers the question in a few bits per key, with one deliberate weakness.

How it works

• Start with a bit array of size m, all zeros, and k hash functions.
• Add a key: compute h_i(key) % m for each of the k functions and set those bits to 1.
• Check a key: compute the same k positions. If any bit is 0, the key is definitely not present. If all are 1, it is probably present.

Each operation is O(k), effectively constant. The filter stores bits, not keys, so it is tiny. And it cannot delete: clearing a bit might erase evidence of another key. Where deletes matter, use a counting Bloom filter or a cuckoo filter.

Sizing it

Two formulas, for n expected keys and a target false-positive rate p:

m = -(n × ln p) / (ln 2)²      bits in the array
k = (m / n) × ln 2             hash functions

Worked example: one million keys at a 2% false-positive rate.

m = -(1,000,000 × ln 0.02) / (ln 2)²
  = (1,000,000 × 3.912) / 0.4805
  ≈ 8.1 million bits  ≈ 1 MB

k = 8.1 × 0.693 ≈ 5.6  →  6 hash functions

One megabyte to rule out a million keys. The rule of thumb behind it: about 10 bits per key gives roughly a 1% false-positive rate, and every halving of p costs more memory. The filter only helps if it sits in RAM, so it is sized per node — R17's key-value store works through that budget for a petabyte cluster.

Where it earns its place

In LSM stores — Cassandra, Bigtable, HBase — every SSTable carries a Bloom filter. A read checks each file's filter first and opens only the files that say "maybe." A key that is nowhere costs a few memory lookups instead of a disk read per file, and a key that exists costs one or two reads instead of dozens. This is what keeps the read amplification of an LSM tree bounded, and it is why write-heavy stores can still serve point reads quickly.

Elsewhere, the same shape: a cache that checks a filter before going to the database for a key that does not exist (the cache-penetration fix in R12), spam and IP blocklists, intrusion detection.

The part people miss: the false positive is safe only because the "maybe" is followed by a real check. A Bloom filter is a fast way to skip work, never the authority. Used as the only duplicate check for payments, a false positive would silently reject a legitimate payment as a duplicate — the R3 idempotency record stays the source of truth, and a filter can only sit in front of it.

Try it. Add a handful of keys and check a few that you did not add. Keep adding until a key you never inserted comes back "maybe present" — that is the false positive, and you can watch the array fill up to cause it.

What to say in the interview

"I list the access patterns before I name a store, then choose the store per kind of data, by the questions it must answer and what it must do when the network breaks. The ledger, balances and idempotency records go on a relational ACID store, single writer per account, because a refused request is recoverable and a double debit is not; feeds, notifications and read models are AP and rebuilt from the ledger's events. I partition early by time for retention and pruning, and I shard late, on the account, through a fixed set of hashed buckets mapped to shards by a directory — so rebalancing moves one bucket at a time instead of rehashing the world."