Hey everyone, welcome to the forty-second issue of The Main Thread.
For a long time, a single DB is the right answer. We add indexes, we add replicas, we put a cache in front, we buy a bigger box. Each of these buys us time, and we should use all of them before we do what this issue is about. Vertical scaling is underrated; the largest cloud instances have hundreds of cores and terabytes of RAM, and a well-tuned single Postgres can carry a startup to a valuation most startups never reach.
But there is a wall. Eventually the working set doesn’t fit in RAM on the biggest box money can buy, or write throughput exceeds what one node’s disk can sustain, or the dataset is simply too large to live on a single machine. Read replicas also don’t help. They scale reads, not writes, and every replica still holds the entire dataset. At that point we can only do one thing: split the data across multiple machines so that each one holds a subset. This is called Sharding.
Sharding is a one-way door. Once the data lives on N machines, a long list of conveniences we took for granted go away: cross-table joins, foreign keys, SELECT across the whole dataset, multi-row transactions, ORDER BY over everything, AUTO INCREMENT ids. The DB is no longer a single coherent thing we query but becomes a fleet we orchestrate. The decision we make at the start, especially the shard key, is extraordinarily expensive to change later, sometimes requiring a full re-copy of the data under production load.
This is the pragmatic guide to doing it well: the three ways to split data, the one decision that matters more than all others, why rebalancing is important, and how to query data that no longer lives in one place.
When We Actually Need To Shard
Before the strategies, a warning, because premature sharding has killed more roadmaps than later sharding ever did.
Sharding multiplies the operational complexity and amputates the query model. We should reach for it only when we have exhausted the cheaper option and have evidence (not a hunch) that we are hitting a single-node limit
There are a few legitimate triggers:
Write Throughput
If we are saturating the write capacity of one primary, and we can’t fix it with batching, async writes, or a write-optimized engine. Read scales with replicas; writes don’t. This is the most common honest reason to shard.
Dataset Size
The data no longer fits on one machine’s disk, or the working set no longer fits in RAM and we are constantly hitting disk. Index maintenance on a multi-TB table starts to dominate.
Blast Radius / Isolation
We want a noisy or compromised tenant to be unable to degrade everyone else, so we isolate tenants onto separate shards. This is a valid business reason, not a capacity reason.
If the actual problem is “reads are slow”, the answer is replicas and caching, not sharding. If the problem is “one table is huge but the rest is fine”, we should consider partitioning that single table within one DB before going distributed. Sharding is the tool for write and size limits, and it’s a poor tool for anything else. When we commit, we must commit deliberately and pick the strategy based on how the data is accessed, because that’s what every strategy below is really about.
Hash-Based Sharding: Even Distribution By Default
The first strategy takes the shard key, hashes it, and uses the hash to assign the row to a shard. Because a good hash function scatters inputs uniformly, the rows spread evenly across shards regardless of what the keys actually are.
import hashlib
def shard_for_key(key: str, num_shards: int) -> int:
# Use a stable hash that's randomized per-process
# and will route the same key differently on every
# restart.
digest = hashlib.sha256(key.encode()).hexdigest()
return int(digest, 16) % num_shardsThe appeal is uniformity. Sequential user ids, monotonic timestamps, alphabetically clustered names, all of them spread evenly once hashed, because the hash destroys whatever structure the key had. This is the right default when the access pattern is point lookups by key: “fetch user 12345“, “get cart for session abc“. Each lookup hashes to exactly one shard, and load spreads naturally. Cassandra and DynamoDB use hash partitioning as their foundation precisely because it makes hotspots-free distribution rather than something we have to engineer.
The cost is that hashing destroys ordering, and we often need ordering. Adjacent keys land on different shards. A query like “all events between Tuesday and Thursday” or “users whose name starts with K” can no longer be served by one shard reading a contiguous range; it becomes a scatter across all shards, because the rows we want are smeared uniformly across all of them. We traded a range-scan efficiency for distribution.
There’s a partial escape, and it’s worth knowing because the good systems use it. Cassandra’s compound primary key splits the key into a partition key (hashed, decides the shard) and clustering columns (kept sorted within the shard). So PRIMARY KEY ((user_id), event_time) hashes on user_id to spread user across shards, but stores each user’s events sorted by time. We get uniform distribution across users and efficient range scans of one user’s timeline, as long as the range query is scoped to a single partition key. The general lesson is that hashing for distribution and sorting for locality are not mutually exclusive if we structure the key so the hashed part is the access boundary.
One trap that catches everyone exactly once: the naive hash(key) % num_shards. It works perfectly until we add a shard, at which point % N becomes % (N + 1) and almost every key remaps to a different shard. This is not just a rebalance but a full data migration disguised as a config change. The fix is explained in the rebalancing section below but flag it for now, because it’s the single most common way teams paint themselves into a corner on day 1.
Range-Based Sharding: Locality When You Need It
This strategy keeps keys in sorted order and assigns each shard a contiguous range of keys. Shard 1 owns a-f, shard 2 owns g-m, and so on. BigTable, HBase, and MongoDB’s ranged sharding all work this way.
# Range map: each shard owns a half-open interval [start, end)
SHARD_RANGES = [
("", "f000", 0), # shard 0: keys < "f000"
("f000", "m000", 1), # shard 1: "f000" <= key < "m000"
("m000", "t000", 2), # shard 2: "m000" <= key < "t000"
("t000", None, 3), # shard 3: key >= "t000"
]
def shard_for_key(key: str) -> int:
for start, end, shard in SHARD_RANGES:
if (start is None or key >= start) and (end is None or key < end):
return shard
raise ValueError("no shard owns this key") The payoff is locality: adjacent keys live on the same shard, so range scans are efficient. “All orders from January”, “all log lines from this hour“, “every document with a key prefix” each hits one shard (or a small contiguous set) instead of fanning out. If the dominant query pattern is range scans, this is the strategy that makes them cheap. It is also why time-series analytics stores favor range partitioning.
The danger is the mirror image of hashing’s strength: range sharding creates hot spots when the access pattern is not uniform. The textbook disaster is a timestamp shard key. If we partition by time and our key is a created_at, then every new write goes to the shard that owns “now“, while every other shard sits idle holding cold history. We have sharded our storage but not the write load, which was probably the whole reason we sharded.
There are a few mitigations:
Prefix the key to spread the hot range
Lead the key with something high cardinality: {region}#{timestamp} or {tenant_id}#{timestamp}. This way “now” is split across as many shards as there are prefixes. We keep range-scan locality within a prefix while spreading write load across prefixes. This is the standard HBase row-key design.
Pre-split the key space
When we create the table, we define the boundaries upfront based on expected distribution instead of letting one giant range exist and split under load.
Just use hashing
If we don’t actually need range scans. The most common reason teams suffer range-sharding hot spots is that they chose range sharding without needing its one benefit.
The decision between hash and range is, at bottom, a single question: do you need efficient range scans on the shard key, badly enough to manage hot spots? If yes, range. If no, hash, and never think about hot spots again.
Directory-Based Sharding: Flexibility At A Price
The third strategy refuses to compute a shard from the key at all. Instead it keeps an explicit lookup table that maps each key (or each range or each tenant) to a shard. To find a row, we first ask the directory “where does this live?“ then go to that shard.
class ShardDirectory:
"""Authoritative mapping from logical key -> physical shard.
Backed by a strongly-consistent store (etcd, ZooKeeper, a small
replicated SQL table) and cached aggressively on every client."""
def __init__(self, store, cache):
self.store = store
self.cache = cache # short-TTL local cache to avoid a hop per query
def locate(self, tenant_id: str) -> str:
cached = self.cache.get(tenant_id)
if cached:
return cached
shard = self.store.get(f"shardmap:{tenant_id}")
self.cache.set(tenant_id, shard, ttl=30)
return shard
def reassign(self, tenant_id: str, new_shard: str):
# The migration code moves the data, THEN flips the pointer.
self.store.put(f"shardmap:{tenant_id}", new_shard)
self.cache.delete(tenant_id)The reason to do this is flexibility, and it’s real. Because placement is an explicit decision rather than a function of the key, we can:
Put a specific large tenant on its own dedicated shard, and tiny tenants packed many-to-a-shard.
Move a single hot tenant to a beefier machine without touching anyone else’s data.
Rebalance one key at a time instead of reshuffling the whole key space.
Co-locate keys that are queries together, even if they have nothing in common hash- or range-wise.
This is why directory based sharding is the backbone of large multi-tenant systems. Vitess’s vindexes, and the sharding layers behind products like Slack, Notion, and Figma, are directory schemes at heart: a tenant or workspace id maps, via a lookup, to a physical home that operators can move around freely.
The price is a layer of indirection that becomes a layer of responsibility.
An extra hop on every query
We must consult the directory before we can route. In practice, we cache the map aggressively on every client (it changes rarely), but now we have a cache-invalidation problem on top of everything else.
Single point of failure
A directory is a single point of failure and a consistency hazard. If the map is wrong or stale, we route to the wrong shard and read stale or missing data. The directory has to be strongly consistent and highly available which usually means putting it on Zookeeper, etcd, or a small replicated SQL table, and treating it as the most critical component in the system.
It can become a bottleneck
It happens when every query hits it synchronously. Caching solves throughput but introduces staleness. There is no free lunch; we are trading the rigidity of a computed mapping for the operational burden of an authoritative one.
All three approaches can be summarized as:
Strategy | Distribution | Range scans | Rebalance granularity | Main cost |
|---|---|---|---|---|
Hash | Uniform by default | Lost (scatter) | Whole key space (unless fixed partitions) | No locality |
Range | Skew-prone, needs care | Efficient | Split/merge ranges | Hot spots |
Directory | Arbitrary, operator-chosen | Depends on map | Per-key / per-tenant | Lookup hop + critical directory |
Choosing a Shard Key
Everything above is a downstream consequence of one choice. The shard key determines distribution, which queries are cheap, which queries are catastrophic, where your hot spots form, and how hard rebalancing will be. And unlike most engineering decisions, it is nearly impossible to change in production; switching shard keys means re-deriving the location of every row and re-copying the entire dataset. Get this right.
Four properties make a good shard key:
High Cardinality
The key must have enough distinct values to spread across all the shards, now and in the future. Sharding on country caps us at ~195 shards and guarantees that India shard is a monster. Sharding on status (3 values) is a non-starter. We want a key with far more distinct values that we will ever have shards.
Uniform Request Distribution
High cardinality isn’t enough if access is skewed. user_id has high cardinality but 1% of users generate 90% of traffic, hashing on user_id still produces hot shards - the celebrity problem. The key’s traffic must spread evenly, not just its values. This is the property most people often get wrong, because they check cardinality and stop.
Query Alignment
The most frequent queries should be answerable by hitting one shard. This means the shard key should be present the hot-path queries. If we shard orders by order_id but the dominant query is “all orders for customer X“, it will be scattered across all shards. If we shard by customer_id, then only one shard will be hit. The shard key should match the highest-volume access pattern, even at the expense of others. We are optimizing the common case and accepting that the rare case fans out.
Avoid Monotonicity (for hash) / manage it (for range)
A monotonically increasing key such as auto increment id, timestamp, is the classic hot-spot generator under range sharding. Under hashing it is fine for distribution but useless for range scans. Either way, a raw timestamp as the primary shard key is almost always a mistake.
When no single column satisfies all four, we should use a composite shard key. I have found a pattern that works repeatedly: combine a high-cardinality distribution dimension with a locality dimension.
-- Cassandra: hash on (tenant_id, bucket) for spread,
-- sort by event_time within each partition for range scans.
-- The bucket splits a hot tenant across multiple partitions.
CREATE TABLE events (
tenant_id uuid,
bucket int, -- e.g. hash(entity_id) % 16, splits hot tenants
event_time timestamp,
entity_id uuid,
payload text,
PRIMARY KEY ((tenant_id, bucket), event_time)
) WITH CLUSTERING ORDER BY (event_time DESC);This gives us tenant isolation (a tenant’s data is co-located), protection against a single hot tenant (the bucket spreads one tenant across 16 partitions), and time-range scans within a partition. The cost is that a query for a tenant’s events now reads up to 16 partitions and merges: a bounded scatter, which is acceptable, versus the unbounded scatter we’d get from a bad key.
There are a few anti-patterns too:
Low cardinality keys (
status,country,booleanflags) - we can’t have more shards than key values, and distribution is lopsided.Monotonic keys as the sole key (
auto_increment, rawtimestamp) - all new writes hit one shard.Keys absent from hot queries - every important query scatters.
Keys with a power-law distribution (
celebrity_user_id,viral_video_id) - a few values carry most of the load and overwhelm their shard.
We must spend time here. A whiteboard session listing the top five queries by volume, checking each against the candidate key, is worth more than any amount of cleverness downstream.
Rebalancing: The Ops Nightmare
It is a surprise to me that most tutorials on the web don’t cover the rebalancing problem in as much depth as they should.
The data grows, we add machines, and we have to move data onto them; while serving production traffic, without losing writes, without corrupting reads. This is rebalancing, and it’s where sharding goes from “interesting design problem” to “I have been paged in the middle of night for a week”.
Let’s start with what NOT to do, because it is the most natural thing to reach for:
Never hash(key) % N
As noted earlier, when N changes, nearly every key’s % N result changes, so nearly every row has to move. Going from 4 shards to 5 shards remaps ~80% of the keys. This is more than just rebalancing; it is full migration triggered by adding one node, and it makes scaling so painful that teams start to avoid scaling. The strategies below all exist to avoid this single failure mode.
Fixed number of partitions
Create many more partitions than nodes up front. Say 1024 partitions for an eventual 10-50 nodes, and assign a bunch of whole partitions to each node. When we add a node, it steals a few entire partitions from the existing nodes. Keys never move between partitions; only whole partitions move between nodes, and only a small fraction of them. The number of partitions is fixed for the life of the cluster, so we must pick it high enough to never run out, but not so high that per-partition overhead hurts. This is how Elasticsearch, Citus, Riak, and Couchbase handle it, and it’s the most operationally pleasant strategy if we can commit to a partition count early.
# Fixed partitions: key -> partition is stable forever.
# Only the partition -> node assignment changes on rebalance.
NUM_PARTITIONS = 1024 # chosen once, never changes
def partition_for_key(key: str) -> int:
return int(hashlib.sha256(key.encode()).hexdigest(), 16) % NUM_PARTITIONS
# partition_to_node is the thing that moves during a rebalance;
# it lives in the directory/coordinator, not in code.
def node_for_key(key: str, partition_to_node: dict) -> str:
return partition_to_node[partition_for_key(key)]Dynamic partitioning
Start with few (or one) partition and split a partition in two when it grows past a size threshold; merge adjacent partitions when they shrink. The number of partitions tracks the data volume, which is elegant for range-partitioned stores when we can’t predict the key distribution. HBase and MongoDB work this way. The catch is that an empty database starts with one partition and therefore one node doing all the work until the first split - which is why these systems support pre-splitting an empty table into ranges we specify upfront.
Partitioning proportional to nodes
Fix the number of partitions per node rather than globally. When we add a node, it splits a few existing partitions and takes half of each. Cassandra (with vnodes) does this. The partition count grows with the cluster, keeping partition size roughly stable.
Three more things experience teaches:
Keep a human in the loop
Fully automatic rebalancing combined with automatic failure detection is a known way to cause cascading disasters: a node gets slow, the system declares it dead and starts rebalancing its data onto an already-loaded peers, which makes them slow, which declares them dead. Many production systems require an operator to click “go” on a rebalance for exactly this reason. Automation that moves TBs should not also decide when to move them.
Online resharding is dual-write + backfill + cutover
The standard zero-downtime flow: (1) start dual-writing to both old and new placement, (2) backfill historical data from old to new in the background, (3) verify the copy, (4) flip reads to the new placement, (5) stop writing to the old. Vitess's reshard workflow is this pattern productized. Every step is reversible until the cutover, which is the point of doing it in this order.
Routing has to stay correct mid-movement
When data is moving, requests must still find it. This is the request routing problem, and there are 3 standard answers (from DDIA): clients contact any node and get forwarded (gossip-based, like Cassandra); a dedicated routing tier holds the map (Vitess's vtgate, Mongo's mongos); or clients are partition-aware and consult the map directly. In all three, the authoritative map usually lives in a coordination service (ZooKeeper, etcd) so that every router converges on the same view as partitions move.
If this section reads as more daunting than the others, that’s the honest signal: rebalancing is a recurring operational cost of sharding, not a one-time set up. We must choose a strategy whose rebalancing story we are willing to live for years, because we will live with it for years.
Cross-Shard Queries: Patterns and Anti-Patterns
The last thing sharding takes from us is the easy JOIN. When data lives on one machine, the query planner does the work. When it’s spread across shards, we do the work, in application code, and the patterns we choose determine whether our system stays fast or fails at scale.
The good case: single shard queries. If the shard key is well-chosen, the hot-path queries include it, hit one shard, and behave like normal single-database queries. This should be the overwhelming majority of the traffic. Everything below is about the queries that can’t be answered by one shard, and the goal is to keep those rare and bounded.
Scatter/gather (fan-out). A query that doesn’t include the shard key must query every shard and merge the result.
async def scatter_gather(query, shards, limit):
results = await asyncio.gather(*[
shard.execute(query) for shard in shards
])
merged = heapq.merge(*results, key=lambda r: r.sort_value)
return list(itertools.islice(merged, limit))Scatter/gather is fine when it is occasional and bounded (a handful of shards, an admin report, a background job). It’s an anti-pattern in the hot path, for two reasons. First, latency is governed by the slowest shard. Second, it doesn’t scale: as we add shards to handle more data, every scatter query gets expensive, so the operation we most want to scale degrades as we grow. If a scatter/gather is in our critical path, we chose the wrong shard key, or we need a secondary index.
Secondary indexes: local vs global. When we want to query by a non-shard key attribute, we have two structural choices, and they trade write cost against read cost.
Document partitioned (local) index. Each shard indexes only its own rows. Writes are cheap - a write updates only the local index on the shard the row already lives on. Reads by the secondary key are a scatter/gather, because matching rows could be on any shard. Elasticsearch, Cassandra, and MongoDB use local secondary indexes. Good default when writes dominate and secondary-key reads are rare.
Term partitioned (global) index. The secondary index is itself sharded, by the indexed term. A read by secondary key goes to the one shard that owns that term - efficient. But a single document write may have to update index entries on several shards (one per indexed attribute), making writes more expensive and usually async (so the index is eventually consistent). Good when secondary-key reads dominate and we can tolerate index lag.
The choice is direct write-vs-read trade: local indexes make writes cheap and secondary reads expensive; global indexes make secondary reads cheap and writes expensive. We should pick based on what our workload does more of.
Cross-shard joins: denormalize instead. A join across shards means pulling data from multiple machines and joining in our application. It is slow, memory hungry, and fragile. The standard solutions, in order of preference:
Co-locate joined data under the same shard key. If we always join orders to their line-items, shard both by
customer_idso a customer’s orders and line-items live on the same shard, and the join becomes local. This is the single most effective technique: design joins out of existence by placement.Denormalize. We should store the data we’d join, pre-joined, at write time. Trade storage and write complexity for read simplicity. The cart row carries the product name and price it needs, so reading a cart never has to join the product shard.
Application-side join as a last resort. Fetch from each shard and combine in code, accepting the cost, only for cold paths.
Cross-shard aggregations and transactions. A COUNT / SUM over all data is a scatter/gather with a merge step. This is fine for analytics, poison in the hot path; for live counters, maintain a denormalized running total instead of aggregating on read. And a write that must atomically touch rows on multiple shards has left the land of single-node transactions entirely - we are not doing a distributed transaction, which is exactly the problem the saga pattern exists to solve. The cleanest fix is usually to not need it: choose a shard key that keeps transactionally-related rows on the same shard, so the transaction stays local. When we genuinely can't, reach for sagas or two-phase commit with eyes open about the cost.
The unifying principle is: sharding rewards queries that respect the shard boundary and punishes queries that cross it. Good sharding design is mostly the work of arranging the data so that the common queries never have to cross. (This is the same locality-and-placement reasoning behind consistent hashing and cache co-location from the load-balancer issue: the shapes recur across distributed systems because the underlying constraint does).
Takeaway
Sharding is the tool we use when one machine genuinely isn't enough, and not one moment sooner. It scales writes and storage past a single node, and in exchange it takes away the unified query model that made your database pleasant to use. That trade is worth making at the right time and ruinous at the wrong one, so the first discipline is honesty about whether we have actually hit a single-node wall.
Once we commit, three decisions carry almost all the weight. The strategy: hash for uniform distribution, range for locality, directory for flexibility, follows directly from how the data is accessed. The shard key is the decision we can't take back, and it should be chosen to make the highest-volume queries single-shard while spreading traffic, not just values, evenly. And rebalancing is the recurring cost we will pay for the life of the cluster, so pick a model fixed partitions, dynamic splitting, proportional, whose operational story we can stomach, and never, ever route with hash % N.
Everything else is consequence. Cross-shard queries are expensive because they cross a boundary you created; the fix is almost always to arrange the data so the common queries don't have to cross. Secondary indexes force a write-vs-read trade. Distributed transactions push us toward sagas. None of these are surprises once we internalize the core fact: sharding rewards locality and punishes everything that breaks it. Design for locality and sharding feels like a superpower. Ignore it and sharding feels like the worst decision we ever made, usually right around the time we try to add the fifth shard.
A single database is a gift. Use it as long as you honestly can. And when you finally outgrow it, split your data along the grain of how you actually use it; that grain is the shard key, and getting it right is most of the job.
What's the most painful shard migration you've lived through? I am collecting war stories, the timestamp shard key that hot-spotted, the % N that triggered a full re-copy, the celebrity tenant that took down a shard. The failure modes are remarkably consistent across companies, which means they're learnable.
Hit reply. I read everything.
Namaste!
If this clicked, forward it to someone staring down their first sharding project, it's the kind of decision that's much cheaper to get right the first time. And if you want more deep dives like this, subscribe to The Main Thread, practical distributed systems engineering, one essay per week.
