Sharding Strategies and Shard Keys

Scenario

Example company: a B2B collaboration SaaS. One Postgres primary handles 40,000 writes/s at peak, the largest table is 9 TB, and the biggest instance size is nearly reached. Most queries are scoped to one workspace; a few enterprise workspaces are 1,000× bigger than average.

What you see in production

Vertical scaling is running out, replicas don't help write volume, and VACUUM on the largest table takes days. A first sharding prototype keyed on user_id breaks many queries (a workspace's documents spread over every shard), and one enterprise customer still overloads "their" shard.

The naive code

def shard_for(user_id):
    return user_id % NUM_SHARDS                     # documents of one workspace scatter everywhere

docs = query_all_shards("SELECT * FROM documents WHERE workspace_id = %s", workspace_id)   # fan-out

Reproduce it

How data lands on 4 shards with different strategies, including one "whale" tenant:

import zlib
tenants = {f"t{i:03}": 10 for i in range(1, 41)}        # 40 normal tenants, 10 units of load each
tenants["t007"] = 400                                     # one whale customer
SHARDS = 4

def h(key): return zlib.crc32(key.encode())               # stable hash (not Python's randomized hash())

def load(assign):
    per = [0] * SHARDS
    for t, l in tenants.items():
        per[assign(t)] += l
    return per

hash_shard = lambda t: h(t) % SHARDS
range_shard = lambda t: min(int(t[1:]) * SHARDS // 41, SHARDS - 1)     # t001-t010 -> 0, …
directory = {"t007": 3}                                                  # pin the whale to its own shard
dir_shard = lambda t: directory.get(t, h(t) % (SHARDS - 1))

for name, fn in (("hash", hash_shard), ("range", range_shard), ("directory", dir_shard)):
    per = load(fn)
    neighbours = sum(1 for t in tenants if t != "t007" and fn(t) == fn("t007"))
    print(f"{name:9} load per shard {per}  tenants sharing the whale's shard: {neighbours}")

hash      load per shard [100, 100, 100, 490]  tenants sharing the whale's shard: 9
range     load per shard [490, 100, 100, 100]  tenants sharing the whale's shard: 9
directory load per shard [150, 120, 120, 400]  tenants sharing the whale's shard: 0

Root cause

A shard key decides two things: which queries stay on one shard and how evenly load spreads. Sharding by user scattered every workspace query. Any key-based scheme also maps a whale tenant to a single shard, so it can still overload it.

The fix

1. Shard by the entity most queries are scoped to — here, the workspace. Keep everything related to it (documents, comments, permissions) on the same shard so joins and transactions stay local.
2. Use many logical shards mapped to fewer physical databases, so you can rebalance by moving logical shards.
3. Directory overrides for whales: a lookup table pins a huge tenant to dedicated capacity.

LOGICAL_SHARDS = 480
physical_for_logical = load_mapping()               # e.g. 480 logical shards → 32 databases
overrides = {"ws_enterprise_42": "db-dedicated-1"}  # directory for whales

def database_for(workspace_id: str) -> str:
    if workspace_id in overrides:
        return overrides[workspace_id]
    logical = zlib.crc32(workspace_id.encode()) % LOGICAL_SHARDS
    return physical_for_logical[logical]

SELECT create_distributed_table('documents', 'workspace_id');
SELECT create_distributed_table('comments', 'workspace_id', colocate_with => 'documents');   -- co-located joins

# VSchema: documents are sharded by a hash of workspace_id ("vindex"); queries with workspace_id route to one shard

Real-world examples (summarised from their engineering posts):

• Notion sharded its Postgres block data by workspace ID into 480 logical shards spread across 32 physical databases, migrating with double-writes, a backfill, verification and a switch-over (Notion engineering).
• Figma first split groups of tables into vertical partitions, then horizontally sharded Postgres; it separated logical from physical sharding and built a DBProxy service that routes queries by shard key (Figma engineering).

Trade-offs

• Hash sharding spreads load evenly but makes range scans across keys impossible to localise.
• Range sharding keeps ranges together (good for time series) but concentrates new writes on the newest range (a hot shard).
• Directory sharding is flexible (move any tenant anywhere) but the directory is critical infrastructure — cache it and make it highly available.
• Every query must know its shard key; queries without it fan out to all shards. Sharding is a one-way door — exhaust vertical scaling, partitioning, caching and read replicas first.

Detect it in production

• Load and storage per shard; alert on imbalance (max/avg) and on single tenants exceeding a share of a shard.
• Count fan-out (scatter-gather) queries — they should be rare.

Key points

• Pick the shard key that most queries and transactions are scoped to; co-locate related tables.
• Use many logical shards on fewer physical databases to make rebalancing possible.
• Handle whales with directory overrides; shard only when other options are exhausted.