Database Sharding and Partitioning: How It Works and When You Need It

In article 07, a messaging app turned out to need about 3.5 PB of storage for two years of messages, and article 06 put its peak at about 92,000 messages written per second. No single database server stores 3.5 PB, and no single primary absorbs 92,000 writes per second with room to spare. Replicas do not help: each one would need all 3.5 PB and all 92,000 writes too.

The only way forward is to split the data across many machines, each owning a part. That is sharding. It is one of the most powerful and most expensive things you can do to a system, so this article covers both how to do it and how to avoid doing it too early.

• Vertical versus horizontal partitioning, and partitioning versus sharding.
• When sharding is necessary, and what to try first.
• How to choose a partition key.
• Range, hash, consistent-hashing and directory-based sharding, with a runnable consistent-hashing example.
• Rebalancing and resharding without downtime, using logical shards.
• Hot partitions and hot keys.
• Cross-shard queries, secondary indexes, joins and transactions.
• How requests are routed to the right shard, and a sizing example.

• Vertical partitioning: split by columns or tables. Move large blobs to object storage, move a busy table to its own database, or split one service's data from another's. Each piece holds different kinds of data.
• Horizontal partitioning: split by rows. Users 1 to 1,000,000 here, the rest there. Each piece has the same schema, different rows.
• Sharding: horizontal partitioning across separate machines (each a shard, usually with its own replicas).
• Partitioning within one database: many databases can split one big table into partitions on the same server. PostgreSQL's declarative partitioning, for example, supports range, list and hash partitioning of a table (PostgreSQL: Table Partitioning). This does not add capacity, but it makes huge tables manageable: queries for one month read one partition, and old data is removed by dropping a partition instead of deleting billions of rows.

Sharding is expensive in complexity: routing, rebalancing, cross-shard queries, operational overhead. Try these first, roughly in order:

1. Fix queries and add indexes (article 13).
2. Cache hot reads (article 09).
3. Add read replicas for read volume (article 15).
4. Move blobs out of the database (article 10) and split unrelated data into separate databases (vertical partitioning).
5. Archive or tier old data.
6. Use a bigger machine. Vertical scaling is unglamorous and often very effective.

Shard when, after that, one of these is still true:

• The data does not fit on one machine (with room to grow).
• Write throughput exceeds what one primary can sustain.
• You need isolation (for example, one large customer must not affect others) or data residency (some users' data must stay in a given country).

Even if you do not shard on day one, choose keys that would shard well, such as including a tenant or user id in the primary key of tenant data. Retrofitting a partition key into a large system is far harder than starting with one.

The partition key (or shard key) decides which shard owns each row. It is the single most important sharding decision, and it is hard to change later. A good key:

1. Spreads data and load evenly. Many distinct values, none of which is much busier than the rest.
2. Appears in the hot queries. If most queries say WHERE user_id = ?, a user_id key sends each query to exactly one shard.
3. Keeps data that is read or changed together on the same shard. A conversation's messages together; an order with its items.
4. Does not concentrate new writes. A key that increases over time (a timestamp, an auto-increment id) sends every new write to the "newest" shard.

AWS's guidance on DynamoDB partition key design makes the same points: choose keys with many distinct values that are requested fairly uniformly, so that traffic spreads across partitions (Best practices for designing partition keys).

Assign contiguous key ranges to shards: user ids A to F on shard 1, G to M on shard 2, or January on shard 1, February on shard 2.

• Pros: range queries ("this user's messages from last week", "orders in September") hit one or few shards; easy to understand.
• Cons: uneven ranges become hot spots; time-based ranges send all current writes to one shard. Needs splitting of ranges as they grow.

Compute hash(key) and map it to a shard, for example hash(key) mod N.

• Pros: even spread, even for sequential keys.
• Cons: range queries on the key scatter across all shards. And with mod N, changing N moves almost every key: going from 4 to 5 shards changes the shard of about 80% of keys, which means moving most of your data.

Place shards (nodes) at positions on a circular hash space, the ring. A key belongs to the first node found by walking clockwise from the key's hash. When a node is added, it takes over only the keys between itself and its predecessor, about 1/N of the data; other keys stay where they are.

With only one position per node, the gaps between nodes on the ring are uneven, so some nodes own far more keys than others. The fix is virtual nodes: place each physical node at many positions (100 or more). Each physical node then owns many small arcs, which average out. Amazon's Dynamo paper describes this combination of consistent hashing and virtual nodes for partitioning data across a changing set of storage nodes (Dynamo: Amazon's Highly Available Key-value Store).

           0 / 2^31        A2     B1     C3           A1        Ring positions: A1, A2 are node A's virtual nodes, and so on.   B3     key k ->  C1      Key k walks clockwise to the next position (C1),     A3           B2        so node C owns it.        C2     ...

This Java program measures it. It maps 100,000 keys to 4 nodes and then to 5, using both hash mod N and a consistent-hashing ring with 200 virtual nodes per physical node, and counts how many keys change owner. It also compares how evenly keys spread with 1 versus 200 ring positions per node.

// File: ConsistentHashDemo.javaimport java.nio.charset.StandardCharsets;import java.security.MessageDigest;import java.security.NoSuchAlgorithmException;import java.util.Locale;import java.util.Map;import java.util.TreeMap;public class ConsistentHashDemo {    // First 4 bytes of MD5: deterministic and well mixed (used to spread keys, not for security).    static int hash(String s) {        try {            byte[] d = MessageDigest.getInstance("MD5").digest(s.getBytes(StandardCharsets.UTF_8));            int h = ((d[0] & 0xFF) << 24) | ((d[1] & 0xFF) << 16) | ((d[2] & 0xFF) << 8) | (d[3] & 0xFF);            return h & 0x7FFFFFFF;       // keep it non-negative        } catch (NoSuchAlgorithmException e) {            throw new IllegalStateException(e);        }    }    // Each physical node is placed on the ring several times ("virtual nodes").    static TreeMap<Integer, Integer> ring(int nodes, int virtualNodes) {        TreeMap<Integer, Integer> ring = new TreeMap<>();        for (int n = 0; n < nodes; n++) {            for (int v = 0; v < virtualNodes; v++) {                ring.put(hash("node-" + n + "#" + v), n);            }        }        return ring;    }    // Walk clockwise from the key's position to the next node on the ring.    static int owner(TreeMap<Integer, Integer> ring, String key) {        Map.Entry<Integer, Integer> e = ring.ceilingEntry(hash(key));        return (e != null ? e : ring.firstEntry()).getValue();    }    static String spread(int nodes, int virtualNodes, int keys) {        TreeMap<Integer, Integer> r = ring(nodes, virtualNodes);        int[] load = new int[nodes];        for (int i = 0; i < keys; i++) {            load[owner(r, "user:" + i)]++;        }        int min = Integer.MAX_VALUE;        int max = 0;        for (int l : load) {            min = Math.min(min, l);            max = Math.max(max, l);        }        return String.format(Locale.US, "smallest node %.1f%%, largest node %.1f%%", 100.0 * min / keys, 100.0 * max / keys);    }    public static void main(String[] args) {        int keys = 100_000;        TreeMap<Integer, Integer> before = ring(4, 200);        TreeMap<Integer, Integer> after = ring(5, 200);        int movedMod = 0;        int movedRing = 0;        for (int i = 0; i < keys; i++) {            String key = "user:" + i;            int h = hash(key);            if (h % 4 != h % 5) {                movedMod++;            }            if (owner(before, key) != owner(after, key)) {                movedRing++;            }        }        System.out.printf(Locale.US, "hash mod N, 4 -> 5 nodes:         %.1f%% of keys move%n", 100.0 * movedMod / keys);        System.out.printf(Locale.US, "consistent hashing, 4 -> 5 nodes: %.1f%% of keys move%n", 100.0 * movedRing / keys);        System.out.println("5 nodes, 1 ring position each:    " + spread(5, 1, keys));        System.out.println("5 nodes, 200 ring positions each: " + spread(5, 200, keys));    }}

hash mod N, 4 -> 5 nodes:         79.8% of keys moveconsistent hashing, 4 -> 5 nodes: 19.1% of keys move5 nodes, 1 ring position each:    smallest node 12.0%, largest node 37.2%5 nodes, 200 ring positions each: smallest node 19.1%, largest node 21.3%

Adding a fifth node with mod N would move four-fifths of the data. With consistent hashing, about a fifth moves, which is exactly the share the new node should end up with. And with one ring position per node, one node gets three times the keys of another; with 200 positions, every node is within about a percentage point of the ideal 20%.

Keep an explicit mapping from key (or key range, or tenant) to shard in a small, highly available lookup service.

• Pros: total flexibility: move one big tenant to its own shard, place users in the region where they live.
• Cons: an extra lookup on every request (usually cached), and the directory must be extremely reliable, because if it is unavailable nothing can be routed.

A common practical design avoids rehashing altogether: create many more logical partitions than machines from day one (for example 1,024), map each key to a logical partition with a fixed hash, and assign logical partitions to physical servers in a small table. Redis Cluster follows this idea with 16,384 hash slots distributed among its nodes (Redis cluster specification).

key --hash--> logical partition 0..1023 --assignment table--> physical shard

To add a machine, move some whole logical partitions to it and update the table. Keys never change partition, only partitions change machines.

Moving a partition while it is live typically follows these steps:

1. Copy the partition's existing data to the new shard (a bulk copy, or a new replica that catches up).
2. Keep the copy current with ongoing changes (replication or dual writes).
3. Briefly pause writes for that partition, or use a short cutover protocol, and switch the assignment table.
4. Redirect traffic; verify; delete the old copy.

Microsoft's Sharding pattern discusses these strategies and rebalancing concerns in more detail (Sharding pattern).

Even with a good key, load is rarely uniform. A celebrity's account, a viral post, a flash-sale product or one huge tenant can make one shard far busier than the rest.

Mitigations:

• Cache hot reads in front of the shard (article 09).
• Salt hot write keys: write post:42#0 to post:42#9 (ten sub-keys on different shards) and combine on read. Good for counters and append-heavy data.
• Split a hot range or move a hot tenant to its own shard (easy with a directory or logical shards).
• Different handling for outliers: a celebrity's posts may be fetched at read time instead of pushed to millions of feeds (article 35).
• Rate-limit abusive or runaway clients.

A query that includes the partition key goes to one shard. A query that does not must ask every shard and merge the results: scatter-gather.

"Photos with tag #sunset, newest 20" (photos partitioned by owner_id)  Router --> shard 1: top 20 matching --\         --> shard 2: top 20 matching ---+--> merge, take newest 20         --> ...                          |         --> shard 64: top 20 matching --/

Costs: every shard does work for every such query, and the query is as slow as the slowest shard (article 02's tail-latency amplification). At 64 shards, a 1%-slow shard makes about half of these queries slow.

Ways to avoid frequent scatter-gather:

• Design access patterns around the partition key in the first place.
• Global secondary index: a separate table or index partitioned by the other key (tag, email, product id), updated when the main data changes, often asynchronously. Queries by that key go to one index shard, then fetch the items. The cost is that the index can lag the data.
• Local secondary indexes (each shard indexes its own rows) are cheap to maintain but still need scatter-gather.
• A separate search or analytics system fed by events (articles 27 and 44).

• Joins across shards are slow and complex. Co-locate data that is joined (shard orders and order_items by the same key) or denormalise.
• Transactions across shards need distributed protocols (two-phase commit) or sagas (article 21). Design so that the things that must change atomically live in one shard: one user's wallet and their transaction log, one order and its items.

Something must know which shard holds a key:

• In the application, through a client library holding the shard map (cached and refreshed).
• In a proxy or router in front of the shards, so applications see one database. Sharding middleware such as Vitess for MySQL or Citus for PostgreSQL takes this approach.
• Inside the database, for systems designed to be distributed (many NoSQL stores and distributed SQL databases), where clients may be redirected to the right node, as Redis Cluster does with its redirection responses.

Each shard is usually itself replicated (a primary and replicas), so you have both partitioning for scale and replication for availability.

Using the numbers from articles 06 and 07 (assumptions, not measurements):

• Data: about 1.2 PB of messages for two years, before replication.
• Writes: about 92,000 messages per second at peak.
• Partition key: conversation_id, with messages ordered by a per-conversation sequence number inside the partition.

Assume each shard primary can comfortably hold 4 TB of message data and sustain several thousand writes per second on the chosen hardware (a figure you would confirm with load tests).

Shards by storage = 1.2 PB / 4 TB = 300 shardsWrites per shard at peak = 92,000 / 300 = about 310 per secondWith 2 replicas each = 900 database servers

Storage, not writes, decides the shard count: each shard is lightly loaded for writes, which leaves room for skew (some conversations are far busier) and growth. Since old messages are rarely read, a sensible refinement is to keep recent months on fast shards and move older data to cheaper storage, which could cut the hot fleet considerably. Use logical partitions (for example 4,096) so shards can be added without rehashing.

A hot-key check: a group chat with 100,000 members sending thousands of messages per second would overload its shard. Large groups might get special handling (sub-partitioning by time, or a dedicated shard), and very large broadcast channels are often designed differently from chats.

• Sharding before trying indexes, caching, replicas and a bigger machine.
• Choosing a key with few values or one that grows over time.
• Partitioning so that the most common query must hit every shard.
• hash mod N with plans to grow N.
• Needing cross-shard transactions for everyday operations.
• Forgetting that each shard also needs replication and backups.

• Show with numbers why one machine is not enough (storage or writes).
• Name the partition key and justify it with the main access pattern.
• Name the strategy (hash with logical partitions, or consistent hashing) and how you add capacity.
• Mention one hot-key case and how you handle it.
• Say which query would require scatter-gather and how you avoid or accept it.

Logical data: 2 TB/day x 90 days = 180 TBShards for storage: 180 TB / 10 TB per shard = 18 shardsShards for writes: 300,000 / 20,000 per shard = 15 shardsMachines: 18 shards x 2 copies (primary + replica) = 36 machines, 360 TB in total

• Next: SQL vs NoSQL: Data Models and How to Choose the Right Database, including stores that shard for you.
• Later: consistent hashing returns in the distributed cache design (article 32); quorums over replicas in article 19; sagas for cross-shard workflows in article 21.