Partitioning (sharding)

Why this matters

Replication makes one machine of data fault-tolerant. Partitioning lets you have more data than fits on one machine. They're complementary: in real systems you partition first (split data across machines), then replicate each partition (for HA).

Partitioning is also where you'll feel the most operational pain at scale. Bad partition choices are extraordinarily expensive to undo because they encode themselves into the API, the schema, the business logic.

From first principles: what problem are we solving?

One server can hold maybe 100 TB of disk and handle maybe 100K writes/sec. If your dataset is larger or your traffic is higher, you need to split it across multiple servers. Partitioning is how.

The hard questions:

• How do you decide which server holds which data?
• How do clients know where to look?
• What happens when a query needs data from multiple servers?
• How do you add or remove servers without downtime?
• What about secondary indexes — they're on different attributes than the partition key?

The four partitioning strategies

Hash partitioning

flowchart LR
    subgraph Ring["Consistent hash ring (positions 0–360°)"]
        N1["Node A — pos 45"]
        N2["Node B — pos 130"]
        N3["Node C — pos 220"]
        N4["Node D — pos 310"]
        N1 -->|next CW| N2
        N2 -->|next CW| N3
        N3 -->|next CW| N4
        N4 -->|wraps| N1
    end
    K1["key foo
hash=80"] -.assigned to.-> N2
    K2["key bar
hash=15"] -.assigned to.-> N1
    K3["key baz
hash=200"] -.assigned to.-> N3
    K4["key qux
hash=350"] -.assigned to.-> N1

Rule: hash the key onto the ring; assign it to the next node clockwise. Adding or removing a node only re-maps ~1/N of keys.

Compute a hash of the partition key and take it modulo the number of partitions:

partition = hash(key) % N

Hashes are uniform: keys spread evenly across partitions, no hot spots (assuming the key itself isn't skewed).

The naive problem: when you add a partition (N changes from 8 to 9), almost every key's modulo changes. ~89% of keys would move. Catastrophic during a resharding.

Fix: consistent hashing. Imagine a ring from 0 to 2^32. Each node owns a range on the ring. To find a key's partition, hash the key onto the ring, then walk clockwise to the next node.

Ring (simplified):
    0 ────────── 1
    │           │
   Node A      Node B
    │           │
   4 ───────── 2
    │
   Node C
    │
    3

Key with hash 0.5 → walks to Node B (first node clockwise).
Key with hash 1.5 → walks to Node C.

Adding Node D between A and B: only the keys between A's position and D's position move (from Node B to Node D). Just ~1/N of keys move per node addition.

Virtual nodes: instead of one ring position per physical node, give each physical node 100-200 virtual positions. Distributes load more evenly; rebalancing is finer-grained. Used by Cassandra (256 vnodes per node default), DynamoDB, Riak.

Used by: Cassandra (default), DynamoDB, Memcached with consistent hashing, Riak.

Range partitioning

Each partition owns a contiguous key range.

Partition 1: keys A-F
Partition 2: keys G-M
Partition 3: keys N-T
Partition 4: keys U-Z

Pros: efficient range scans. "Give me all keys from K to M" hits one or two partitions.

Cons: hot spots if keys cluster. The classic disaster: range-partition by created_at. All new writes go to the latest partition. One partition is white-hot; others are idle.

Fix for the hotspot: prepend a salt or hash to the key before range-partitioning, or use composite partition keys (more on this below).

Used by: HBase, Bigtable, Spanner, CockroachDB (which also uses range partitioning to provide ordered scans for SQL).

Directory-based (lookup) partitioning

An explicit table maps keys to partitions:

key   | partition
------|----------
foo   | shard-3
bar   | shard-1
baz   | shard-3
qux   | shard-7

Maximum flexibility: you can rebalance per-key without re-hashing. But the directory itself is a bottleneck and a SPOF.

Used by: Slack (channel→shard mapping in a "topology table"), Vitess (vschema lookup tables), many internal multi-tenant systems where one tenant might be 1000× bigger than another.

Composite partitioning

Most production systems combine strategies. Common pattern:

• Partition key: determines which shard.
• Sort key: determines order within a shard.

Example for a chat app: (channel_id, timestamp). channel_id is hashed to pick a shard; timestamp orders messages within a channel. Queries for "recent messages in channel X" hit one shard and read in sorted order.

Used by: Cassandra (composite primary keys), DynamoDB (partition key + sort key), most modern wide-column stores.

The secondary index problem

Your primary key shards your data. Great. Now what about queries by a different attribute? "Find all users with email = X" — but email isn't the partition key.

Two approaches, each with painful tradeoffs.

Local secondary indexes

Each shard maintains its own secondary index on its own data.

Shard 1: index on email → rows in shard 1 only
Shard 2: index on email → rows in shard 2 only
...

Writes: cheap (one shard write, one local index update). Reads by email: scatter-gather — query every shard, aggregate results.

Used by: Cassandra secondary indexes, DynamoDB local secondary indexes (LSIs).

Global secondary indexes

The index is itself sharded by the indexed attribute.

Index shard 1: emails a-m → original-row pointers
Index shard 2: emails n-z → original-row pointers

Reads by email: cheap (one index shard lookup, one data shard lookup). Writes: distributed transaction across data shard and index shard.

Used by: DynamoDB global secondary indexes (GSIs), Vitess vindex tables, Spanner.

Hot keys and the celebrity problem

The single most common production failure of a sharded system. Symptoms: one shard's CPU/IOPS at 100% while siblings are idle.

Where hot keys come from

• Celebrity user (Twitter's Justin Bieber problem): one user followed by millions, so their tweets fan out heavily.
• Popular content: one viral post in a Discord channel.
• Time-based partitioning: all writes go to "today's" partition.
• Login flow on a tiny tenants table.
• Counter on an admin's row.

Mitigations

Append-the-key. Instead of partitioning on user_id, partition on user_id_bucket = (user_id, random 0..N-1). The same user's data is now spread across N shards. Reads must scatter-gather across the N buckets — accept the cost.

L1 cache. Put a local cache in front. 99% of hot-key reads never hit the data tier. The cache is your shock absorber.

Read replicas. Add more replicas for the hot shard, spread reads across them. Doesn't help writes.

Dedicated shard for the hot tenant. The 1% biggest tenant gets its own physical shard. Operational complexity; sometimes worth it.

Application-level rate limiting. Cap the request rate for the hot key. User-visible.

Online resharding — adding and removing nodes without downtime

This is the operation that decides whether a sharded system is operable in production.

The slot-based approach (Redis Cluster, Vitess)

Pre-tokenise the keyspace into many small, fixed-size slots. Redis Cluster has 16,384 slots. Each slot is independently movable.

To add a node:

1. Decide which slots to move to the new node (typically 1/N of total).
2. For each slot:
   • Mark it MIGRATING on source, IMPORTING on destination.
   • Stream keys from source to destination, in batches, throttled.
   • Client requests for those slots: if source has the key, serve it; if not, return an ASK redirect to the destination.
3. When migration of a slot finishes, flip ownership atomically in cluster topology.
4. Broadcast new topology; clients refresh their slot map on next MOVED error.

Throttling matters: migration competes with serving traffic for I/O. Cap migration rate (e.g., 1000 keys/sec) to keep production p99 stable. A 1 TB shard at 10 MB/sec moves in 28 hours. Plan resharding in days, not hours.

Real-world online resharding examples

Notion: sharded Postgres from 1 → 32 → 96 logical shards over multiple years. Each transition involved logical replication + dual-write + cutover. Their writeup: notion.so/blog/sharding-postgres-at-notion. Key insight: shard early, even if only on one physical box. Going from 1 logical shard to 32 is the painful migration; going from 32 on one box to 32 on 32 boxes is straightforward.

Figma: online Postgres sharding by table groups. Each group is independently shardable. They wrote about it: figma.com/blog/how-figmas-databases-team-lived-to-tell-the-scale. Used a custom routing layer in front of pgbouncer.

Vitess (YouTube → GitHub → Slack): a whole framework for sharded MySQL with online resharding as a first-class operation. vitess.io/docs.

Cross-partition transactions — the hardest problem

Updating two rows on different shards atomically is hard.

Most production systems pick avoidance. Design the schema so the partition key co-locates the rows that need atomic updates (e.g., all of a user's data on the same shard). Cross-shard transactions become rare exceptions handled by saga or eventual consistency.

Interview drills

Red flags

• "We can shard later." Almost always wrong — by the time you need to, you have business logic assuming single-shard transactions.
• Picks timestamps as partition key without recognising the hot spot.
• Doesn't address secondary indexes.
• Proposes 2PC for cross-shard ops without acknowledging cost and blocking.
• Doesn't quantify scatter-gather impact.
• Confuses partitioning with replication.