Database Sharding Cheat Sheet

import hashlibdef get_shard(user_id: str, num_shards: int) -> int:    # Consistent hash of the key mod number of shards    digest = hashlib.md5(user_id.encode()).hexdigest()    return int(digest, 16) % num_shards# Route a query to the correct shard connectionshard_id = get_shard("user_42", num_shards=8)conn = shard_connections[shard_id]conn.execute("SELECT * FROM orders WHERE user_id = %s", ("user_42",))

import bisectimport hashlibclass HashRing:    def __init__(self, nodes, vnodes=100):        self.ring = {}        self.sorted_keys = []        for node in nodes:            for i in range(vnodes):                key = self._hash(f"{node}:{i}")                self.ring[key] = node                bisect.insort(self.sorted_keys, key)    def _hash(self, key):        return int(hashlib.md5(key.encode()).hexdigest(), 16)    def get_node(self, key):        h = self._hash(key)        idx = bisect.bisect(self.sorted_keys, h) % len(self.sorted_keys)        return self.ring[self.sorted_keys[idx]]

def write_during_migration(key, value):    old_shard = get_shard(key, num_shards=OLD_N)    new_shard = get_shard(key, num_shards=NEW_N)    write(old_shard, key, value)    write(new_shard, key, value)  # dual-write so both topologies stay currentdef backfill(old_shards, new_shards):    # Copy historical rows written before dual-write was enabled    for row in scan_all(old_shards):        target = get_shard(row.key, num_shards=NEW_N)        upsert(target, row)  # idempotent write, safe to re-run# Cutover sequence:# 1. Enable dual-write.# 2. Backfill historical data into new shards.# 3. Verify row counts / checksums match between old and new.# 4. Flip reads to the new shard map.# 5. Stop writing to old shards, decommission.

from concurrent.futures import ThreadPoolExecutordef scatter_gather_query(sql, params, shard_connections):    def query_one(conn):        with conn.cursor() as cur:            cur.execute(sql, params)            return cur.fetchall()    with ThreadPoolExecutor(max_workers=len(shard_connections)) as pool:        results = pool.map(query_one, shard_connections.values())    merged = [row for shard_rows in results for row in shard_rows]    # Sorting/aggregation (ORDER BY, LIMIT, GROUP BY) must be redone here    # since each shard only sorted/limited its own local rows.    return sorted(merged, key=lambda r: r["created_at"], reverse=True)

NUM_VIRTUAL_SHARDS = 4096  # fixed forever, far more than any physical node countdef virtual_shard(key: str) -> int:    return hash(key) % NUM_VIRTUAL_SHARDS# Small, static mapping updated as physical capacity changesvshard_to_node = {    0: "node-a", 1: "node-a", 2: "node-b", 3: "node-b",  # ... 4096 entries}def get_node(key: str) -> str:    return vshard_to_node[virtual_shard(key)]# Adding node-c: only reassign a subset of virtual shard entries (e.g.# 0..1365 -> node-c) instead of rehashing every key in the cluster.