Partitioning and Sharding

The orders table has 400 million rows on one 2 TB node, and writes are saturating its disk. The obvious fix is to split it: pick a shard key, hash it, and spread rows over 64 shards. Writes parallelize immediately. Two weeks later support reports that the "recent orders for customer 88213" page — which used to return in 40 ms — now times out, and a WHERE customer_id = 88213 ORDER BY created_at DESC scan that once hit one index now fans out to 64 nodes and merges 64 result sets.

That is the trade of hash partitioning in one sentence: you trade spatial locality for balance. The function is simple:

shard = H(key) mod N          # naive version — avoid this for growing clusters
shard = lookup(H(key))        # consistent hash ring: H(key) → ring position → owning node

example (64 shards):
  key user_88213 → sha256 → 0x4f2a... → ring position 0x4f2a → shard 31
  every query for that user goes to shard 31, and only shard 31

Hashing makes distribution a probability question instead of a design question: if the hash is uniform and the key has high cardinality, load evens out by the law of large numbers — no shard holds 10% of rows just because it was created first. That is why Cassandra, DynamoDB, Cosmos DB, and Vitess all hash by default.

What breaks:

The scatter-gather rows are the ones to internalize: the shard key must appear in almost every query, because any predicate without it becomes a full-fleet scan. This is the classic sharding interview question — "you shard by user_id, how do you produce a global leaderboard?" The honest answers are: keep the leaderboard in a separate store designed for it, maintain a per-shard partial aggregate and merge top-K, or accept an eventually-updated precomputed table.

The second failure is cross-shard transactions. UPDATE accounts SET balance = balance - 100 WHERE id = A on shard 3 and ... + 100 WHERE id = B on shard 19 cannot share a local ACID transaction; you need a two-phase commit, a saga (The Saga Pattern), or a redesign that colocates both rows on one shard. Sharding an existing schema also runs into constraints that assumed a global table: unique indexes become per-shard unless the key includes the shard key, foreign keys are unenforceable across shards, and AUTO_INCREMENT collides — which is why sharded systems push you toward UUID/ULID keys or per-shard ID ranges.

Finally, mod N deserves its own warning. With 64 shards, H(id) mod 64 sends nearly every key to a different shard when you go to 65 — a full data reshuffle billed as "adding one node". Consistent hashing fixes this: keys are placed by ring position, and adding a node takes over only the arc of the ring it now owns — roughly 1/N of the keys move, not everything. Virtual nodes (each physical node owns many ring positions) smooth out the imbalance that a sparse ring produces.

Pick hash partitioning when access is point-lookup shaped and you need even load. If your queries are ordered, range partitioning preserves exactly what hashing throws away — the next chapter covers when that locality is worth the hotspot risk it invites.

A multi-tenant analytics product stores events with tenant_id as the primary dimension. Hashing tenant_id would balance rows nicely, but every query starts with WHERE tenant_id = ? AND ts BETWEEN ... ORDER BY ts — a shape hash partitioning punishes. So the team ranges on ts: partition 2026-01, 2026-02, ... each a contiguous, index-ordered block. Queries touching a time window hit only the partitions that overlap it — partition pruning — and ordered scans are served by local indexes. For a while this is the correct answer.

Then one customer backfills two years of history in a weekend, and the 2024-03 partition grows to ten times its siblings. Every write for that period lands on one shard, its WAL fsync queue backs up, and p99 write latency for all tenants sharing the node degrades. This is the hotspot: range partitioning gives locality by accepting that locality is unevenly distributed.

The two axes you can range on lead to two different failure modes:

An increasing-ID range is the sneakiest: the last partition is a write hotspot by construction, because every insert goes to the right edge. Remedies, in increasing order of invasiveness:

1. Salt the range — prepend a random shard prefix (bucket = random(0,7), key bucket:ts) so writes scatter across 8 pre-splits per window, then merge on read. You have bought balance by giving up single-partition ordered reads — the hash trade again, scoped to the query.
2. Pre-split — create empty partitions ahead of time so growth lands in several places rather than one auto-splitting node.
3. Isolate the heavy tenant — dedicate a node per large tenant (shard-by-tenant, range-within-shard: composite key (tenant_id, ts) gives per-tenant balance and per-tenant ordered scans).
4. Backpressure and batching — throttle bulk loads to off-peak or route them to a staging keyspace merged later.

Composite keys deserve emphasis because they combine both worlds: PARTITION BY (tenant_id) RANGE (ts) — common in Cassandra with CLUSTERING ORDER BY ts DESC and in Postgres declarative partitioning — balances across tenants while preserving order inside each. The rule of thumb interviewers listen for: the first column of the key decides distribution; later columns decide locality within a shard.

There is also a maintenance cost specific to ranges: partitions must exist for incoming data. If new ranges are not pre-created, inserts fail with errors like no partition of relation 'events' found for row (Postgres) rather than silently creating one. Scheduling partition creation, archiving old partitions to object storage, and dropping expired ones is operational work you own.

Range partitioning is right when queries carry a bounded, ordered predicate and data has a natural temporal edge. The remaining problem — partitions drifting apart as usage changes — is the rebalance problem, covered in the final chapter.

Your cluster has 16 shards and the traffic forecast says 24 by Q3. Adding nodes is the easy part — the hard part is that the old mapping H(id) mod 16 now disagrees with mod 24, which means every key must move. The team schedules a maintenance window, pauses producers, copies, remaps, restarts: 40 minutes of write downtime and a rollback plan nobody wants to use. The alternative is a rebalance that never stops serving, and it comes down to one invariant: routing must change exactly once, and every write during the copy must be captured either way it could be read.

First, why naive approaches fail:

• mod N remapping — with N changing, virtually every key changes shard. Moving 97% of the data is not a rebalance, it is a migration. Consistent hashing (or Vitess's shard-routing cells, or DynamoDB's partition metadata) limits movement to 1/N of the data per node addition.

• Copy-then-switch without a log — while you copy shard 31's rows to the new owner, writes keep landing on shard 31. At cut-over, those rows are stale on the new owner unless you capture them.

The two production-proven patterns:

Dual-write / routing epoch:
 1. epoch 1: router sends ALL traffic to old owners; background copy runs
 2. capture every change since the copy started (CDC tail or write log)
 3. epoch 2: router flips to new owners in ONE atomic metadata update
    → any write that raced the flip lands on the new owner too (dual-written)
 4. verify checksums, drain old copy, epoch 3: stop dual-write

Chunk move (MongoDB, Vitess, Kafka):
 1. split range into chunks; mark chunk migrating (owner → donor)
 2. copy chunk; while copying, incoming writes queue on the donor
 3. hand over ownership metadata, replay queued writes, chunk now owned
 4. repeat per chunk — each chunk is individually small, so the flip is fast

Both satisfy the same rule: the metadata/routing change is atomic and instantaneous; the data copy is slow and invisible. You never have two masters for a key after the flip, because writes during the copy were replayed or dual-written. Kafka's kafka-reassign-partitions with --throttle and MongoDB's chunk balancer both follow it, and both throttle replication bandwidth so the rebalance does not starve foreground reads.

Practical details that decide success:

One interview follow-up to be ready for: "what if a node dies mid-rebalance?" The answer is that ownership metadata, not the copy process, defines truth — an in-flight chunk move aborts and re-allocates, and replicas already satisfy the replication factor independently of the copy. This is why sharded stores separate the control plane (who owns what, changed by consensus) from the data plane (reads and writes) — the same split Raft-backed systems use (Consensus Overview with Raft).

With data now spread across nodes, the next question is what promise each read makes. Continue to Consistency Models.