Sharding

Replication makes copies of the same data. Sharding (horizontal partitioning) splits different data across machines, so each shard holds a slice of the whole. You reach for it when one node can no longer hold the data or absorb the writes — when even the biggest single primary has run out of room or throughput.

Each shard is its own database (often itself replicated). The whole game is one decision: how do you decide which row lives on which shard? Get that wrong and you’ll spend years paying for it.

The three strategies

Range sharding — partition by ranges of the key. Users A–H on shard 1, I–P on shard 2, Q–Z on shard 3. Range scans are efficient (adjacent keys sit together), but ranges go lopsided: if everyone signs up S–T, shard 3 roasts while shard 1 idles.

Hash sharding — run the key through a hash and assign by the result. shard = hash(key) % N. Distribution is beautifully even, but you lose range queries (adjacent keys scatter), and that naive modulo formula is a resharding landmine (next lesson fixes it).

Directory-based — a lookup service maps key → shard explicitly. Maximum flexibility (move any key anywhere, rebalance freely) at the cost of an extra hop and a new component that must itself be highly available.

Choosing the shard key

This is the highest-leverage decision in the whole design, and it has two requirements that pull against each other:

1. High cardinality and even distribution — so data and load spread evenly.
2. It’s in your hot queries — so a typical request hits one shard, not all of them.

A great shard key for a chat app is often the conversation id: messages for a conversation live together (one-shard reads) and conversations are numerous and roughly even. A terrible shard key is something low-cardinality like country or status — a handful of values means a handful of giant, lopsided shards.

Hot shards and the celebrity problem

Even a well-chosen key has pathological cases. If you shard a social network by user, a celebrity with 100 million followers turns their shard into a furnace while everyone else’s idles — the celebrity (hot-shard) problem. No hashing fixes it, because the heat is concentrated on one key. Real mitigations:

• Split the hot key — fan a celebrity’s writes across sub-partitions.
• Cache the hot entity hard, in front of the shard.
• Replicate read-heavy hot keys to extra read replicas.

The general lesson: uneven access is harder than uneven data, and your shard key controls both.

Resharding pain and cross-shard queries

Two costs you sign up for the moment you shard:

Resharding. When shards fill up and you add machines, data must move. With hash(key) % N, changing N remaps almost every key — a catastrophic reshuffle. (This exact pain is what consistent hashing, next lesson, was invented to solve.)

Cross-shard queries and joins. A query that spans shards must scatter to every shard and gather the results — slow, and as available as your least available shard. Joins across shards are worse: there’s no efficient distributed join, so you either denormalize to keep related data co-located, or join in application code. This is why sharded systems lean so heavily on denormalization.

The JavaScript angle

In Node, sharding usually lives as a thin routing layer: hash the shard key, pick a connection pool. The danger is the innocent-looking query that forgets the shard key and silently fans out to every node:

import { Pool } from 'pg';
import { createHash } from 'node:crypto';

const shards = [new Pool({ host: 'shard-0' }), new Pool({ host: 'shard-1' }), new Pool({ host: 'shard-2' })];

function shardFor(userId) {
  const h = createHash('md5').update(userId).digest().readUInt32BE(0);
  return shards[h % shards.length];           // ⚠️ modulo — reshard hazard
}

// ✅ Single-shard read: has the shard key, hits one node. Fast.
async function getUser(userId) {
  return shardFor(userId).query('SELECT * FROM users WHERE id = $1', [userId]);
}

// ❌ Cross-shard: no shard key in the filter → must scatter to ALL shards
//    and gather. Slow, and fails if any one shard is down.
async function countActiveEverywhere() {
  const results = await Promise.all(
    shards.map((s) => s.query("SELECT count(*) FROM users WHERE active = true")),
  );
  return results.reduce((sum, r) => sum + Number(r.rows[0].count), 0);
}

That % N is the resharding time bomb. Next lesson defuses it with consistent hashing — the technique that lets you add and remove nodes while reshuffling only a small fraction of the keys.