Sharding & Partitioning

Here's the core idea in one sentence: when a single database can't handle the write load no matter how big you make it, you split the data across multiple databases so each one handles a fraction of the traffic. Read replicas solve read bottlenecks. Sharding solves write bottlenecks. That's the distinction that matters.

InstagramInstagram shards PostgreSQL by user_id. Each shard is a separate PostgreSQL instance holding roughly 10–15 million users. The shard ID is embedded directly into their Snowflake-style IDs, so given any photo or comment ID, they can extract which shard it lives on without a lookup. As of their 2012 engineering blog post, they used about 64 logical shards mapped across fewer physical servers. shards PostgreSQL by user_id — 1 billion users across 64 shards, about 15 million users per shard. DiscordDiscord shards messages in Cassandra by (channel_id, bucket), where a bucket represents 10 days of messages. Old buckets get archived to cold storage. At scale, they manage ~120 billion messages across thousands of Cassandra nodes. In 2022, they migrated from Cassandra to ScyllaDB for better tail latency. shards messages by (channel_id, time_bucket) across thousands of Cassandra nodes. ShopifyShopify uses Vitess (originally built by YouTube) to shard MySQL. Each merchant's data lives on a single shard, so all queries for one store hit one shard. They manage thousands of shards across their fleet. The Vitess proxy (vtgate) routes queries transparently — the application code doesn't know sharding exists. shards MySQL through Vitess, one shard per merchant. These aren't theoretical examples — they're running right now, serving billions of requests per day.

You're the database engineer for a fast-growing e-commerce platform. Your single PostgreSQLA free, open-source relational database used by Instagram, Stripe, Reddit, Notion, and thousands of other companies. You can spin one up right now: docker run -e POSTGRES_PASSWORD=test -p 5432:5432 postgres:16. It stores data in 8 KB pages, supports streaming replication, and can handle tens of thousands of queries per second on modern hardware. instance holds 500 GB of data across three main tables: users (80 million rows), orders (600 million rows), and products (2 million rows). You've already done everything the "Scaling Databases" playbook told you to do:

• Vertical scaling: You're on the biggest RDS instance money can buy — db.r6g.16xlarge (64 vCPUs, 512 GB RAM, $7,200/month).
• Read replicas: You have 3 read replicas absorbing 90% of your SELECT queries.
• Connection pooling: PgBouncer sits in front with 100 pooled connections.
• Query optimization: Every slow query has been indexed, rewritten, or cached in Redis.

And yet your database is still dying. Here's why: the bottleneck isn't reads anymore — it's writes. Your platform processes 50,000 order writes per second during peak hours (flash sales, holiday events). Every single one of those writes — INSERT, UPDATE, DELETE — must go through the single primary database. Read replicas don't help with writes. A bigger machine doesn't exist. You're stuck.

This is the exact moment where every fast-growing company reaches the same conclusion. Instagram hit it when their single PostgreSQL couldn't handle the write volume from 100 million users liking and commenting simultaneously. Discord hit it when message writes across millions of channels overwhelmed a single Cassandra cluster. The answer is always the same: stop trying to push all writes through one machine. Split the data.

Before we reach the breakthrough, let's trace the two obvious ideas that don't work — because understanding why they fail makes the real solution click instantly.

Idea 1: Vertical Scaling (Bigger Machine)

You're already on the biggest machine AWS sells for RDS: the db.r6g.16xlarge with 64 vCPUs and 512 GB RAM. There is no db.r6g.32xlarge. You've hit the ceiling. Even if a bigger machine existed, the gains are sub-linear — doubling cores doesn't double write throughput because of PostgreSQL's internal lock contentionWhen multiple database processes try to modify the same data structure at the same time, they have to wait for each other. PostgreSQL uses lightweight locks (LWLocks) and heavyweight locks to coordinate. More cores means more processes competing for these locks, which means more time spent waiting instead of working. This is why 64 cores don't give you 16x the write throughput of 4 cores — you might get 8-10x at best. on shared memory.

Even at the maximum, you're running at 98% capacity with zero headroom for traffic spikes. And your traffic is growing. In 3 months, you'll need 80K writes/sec. No single machine on Earth can handle that for a PostgreSQL process.

Idea 2: More Read Replicas

This is the most common mistake engineers make when they first see write bottlenecks. "We added read replicas to handle more reads — can't we add write replicas too?" The answer is no, and understanding why is fundamental to understanding sharding.

A read replicaA copy of your primary database that receives a stream of changes (via WAL shipping or logical replication) and applies them locally. It can serve SELECT queries but CANNOT accept INSERT, UPDATE, or DELETE. All writes must go to the primary, which then streams the changes to all replicas. Adding more replicas gives you more read capacity but zero additional write capacity. receives changes FROM the primary. It doesn't generate changes. Think of it like a TV broadcast: the studio (primary) creates the content, and TV sets (replicas) display it. Adding more TVs doesn't help the studio produce content faster. Every write still goes through one machine:

Let's put real numbers to the pain. This isn't "the database is slow" — this is exact arithmetic that proves a single machine physically cannot keep up.

The Write Math

Each write operation (INSERT, UPDATE, DELETE) in PostgreSQL involves several steps: parse the SQL, acquire row locks, write to the WALWrite-Ahead Log. Before PostgreSQL changes any actual data page, it first writes a description of the change to the WAL. This guarantees durability — if the server crashes mid-write, PostgreSQL replays the WAL on startup to recover. The WAL is an append-only file on disk. You can see your WAL location right now: SELECT pg_current_wal_lsn();. Every replica streams this WAL to stay in sync. (write-ahead log), update the data page in shared buffers, and eventually flush to disk. On a well-tuned PostgreSQL, a typical write takes about 0.2 ms (200 microseconds) under low load. But as concurrency increases, lock contention adds up.

At 50,000 writes per second, here's what happens on a single machine:

# Each write: ~0.2ms base + lock contention overhead at high concurrency
# At 50K writes/sec with contention, effective time per write ≈ 0.8ms

50,000 writes/sec × 0.8 ms/write = 40,000 ms of write work per second
                                   = 40 seconds of work per 1 second of wall time

# PostgreSQL can use multiple cores, but WAL writing is largely serialized.
# The WAL insert lock is a SINGLE bottleneck — only one process writes WAL at a time.

# On 64 cores, you can parallelize some work, but:
# - WAL insert: serialized (1 core)
# - Lock manager: shared (contention grows with cores)
# - Buffer pool: shared (contention grows with cores)

# Real measured throughput ceiling on db.r6g.16xlarge:
# ~45K simple INSERTs/sec (pgbench, 64 clients)
# ~30K mixed writes with UPDATEs and foreign keys
# ~20K writes with complex transactions (BEGIN...multiple statements...COMMIT)

# Your 50K writes/sec with real transactions?
# You need roughly 1.5x the capacity of the biggest machine that exists.

The WAL insert lockPostgreSQL 9.4+ uses a WAL insert lock array (8 locks by default) instead of a single lock, but at extremely high write rates, these locks still become a bottleneck. The WAL is fundamentally an append-only sequential log — you can't parallelize sequential writes beyond a certain point. Run SELECT * FROM pg_stat_activity WHERE wait_event = 'WALInsertLock'; to see processes waiting on this lock right now. is the key insight. PostgreSQL writes every change to the WAL before confirming the transaction. At low volume, this is instant. At 50K writes/sec, processes start queuing up to write to the WAL. It's like a single-lane highway — no matter how many lanes (cores) you have leading up to it, everyone has to merge into one lane to cross the bridge.

The insight is almost embarrassingly simple once you see it. If one database can handle 30K writes/sec, and you need 50K writes/sec, then two databases can handle 60K writes/sec — as long as each one only receives half the writes. The trick is making sure each write goes to the right database.

This is shardingThe practice of horizontally splitting a database's rows across multiple independent database instances (shards). Each shard holds a different subset of the data and operates independently with its own CPU, RAM, WAL, and disk. Unlike replicas (which hold copies of ALL data for read scaling), shards hold DIFFERENT data for write scaling. The word "shard" comes from the MMORPG Ultima Online (1997), where the game world was split into parallel "shards" to handle more players.. You pick a column — say user_id — and use it to decide which database a row lives on. User IDs 1–40M go to Shard 0. User IDs 40M–80M go to Shard 1. When a write comes in for user 25,000,000, it goes to Shard 0. When a write comes in for user 65,000,000, it goes to Shard 1. Each shard gets roughly half the traffic, and each shard has its own CPU, its own RAM, its own WAL — its own everything.

The column you pick is called the shard keyThe column (or combination of columns) used to determine which shard a row belongs to. Choosing the right shard key is the single most important sharding decision. A good shard key distributes data evenly AND ensures that related data lives on the same shard. Instagram uses user_id (all of a user's data on one shard). Discord uses (channel_id, time_bucket). A bad shard key creates hotspots — one shard getting far more traffic than others. (sometimes called the partition key). And that choice is the single most important decision in sharding — get it right and everything flows smoothly; get it wrong and you create hotspots that defeat the entire purpose.

But here's the trade-off nobody mentions in the YouTube videos: once you shard, your database isn't "one thing" anymore. It's N separate things. A query that used to scan one table now has to scan N shards. A JOIN between orders and users that used to be instant now requires a scatter-gatherA query pattern where the router sends the same query to ALL shards (scatter), waits for all responses, then merges the results (gather). If orders are on shard 0 and users are on shard 3, the JOIN can't happen inside a single database — the router has to fetch from both shards and combine them in memory. This is O(N) where N is the number of shards, and it's the primary reason sharding makes some queries dramatically slower. across all shards. Transactions that span multiple shards require two-phase commitA distributed transaction protocol (2PC) where a coordinator asks all participants "can you commit?" (Phase 1: Prepare). If ALL say yes, the coordinator sends "commit" (Phase 2: Commit). If ANY says no, the coordinator sends "abort" to everyone. 2PC guarantees atomicity across shards but adds latency (2 round trips instead of 1) and can block if the coordinator crashes between phases., which is slow and fragile.

This is why sharding is called "the weapon of last resort." It gives you unlimited write scaling, but it takes away things you used to get for free: JOINs, transactions, and global queries. Every company that shards wishes they didn't have to. But every company that needs to handle 50K+ writes/sec has no other option.

You've decided to shard. The next question is: how do you decide which rows go to which shard? There are five main strategies, each with different trade-offs. The right choice depends on your data shape and your query patterns — and getting it wrong means hotspots, scattered queries, or painful migrations.

-- Range sharding logic (conceptual)
-- Shard 0: user_id 1 to 20,000,000
-- Shard 1: user_id 20,000,001 to 40,000,000
-- Shard 2: user_id 40,000,001 to 60,000,000
-- Shard 3: user_id 60,000,001 to 80,000,000

-- Query for user 35M → goes to Shard 1
SELECT * FROM orders WHERE user_id = 35000000;

-- Range scan for users 10M-15M → goes to Shard 0 only (efficient!)
SELECT * FROM orders WHERE user_id BETWEEN 10000000 AND 15000000;

-- In MongoDB, this is automatic with range-based sharding:
-- sh.shardCollection("mydb.orders", { "user_id": 1 })
-- MongoDB's balancer automatically creates chunks and moves them

-- Hash sharding: shard = hash(user_id) % 4

-- In MongoDB (hashed shard key):
-- sh.shardCollection("mydb.orders", { "user_id": "hashed" })
-- MongoDB automatically hashes user_id and distributes chunks evenly

-- In Citus (PostgreSQL):
-- SELECT create_distributed_table('orders', 'user_id');
-- Citus uses a hash function internally, creates 32 shards by default

-- In Vitess (MySQL):
-- VSchema defines the sharding function:
-- { "sharded": true,
--   "vindexes": { "hash": { "type": "hash" } },
--   "tables": { "orders": { "column_vindexes": [
--     { "column": "user_id", "name": "hash" }
--   ]}}
-- }
-- vtctl ApplyVSchema -vschema_file=vschema.json commerce

-- Manual hash routing (pseudocode):
-- shard_id = murmur3_hash(user_id) % 4
-- connection = shard_connections[shard_id]
-- connection.execute("INSERT INTO orders ...")

-- The directory: a separate table (or service) mapping keys to shards
CREATE TABLE shard_directory (
    user_id_start  BIGINT,
    user_id_end    BIGINT,
    shard_id       INT,
    shard_host     VARCHAR(255),
    PRIMARY KEY (user_id_start)
);

-- Example entries:
-- | user_id_start | user_id_end | shard_id | shard_host          |
-- |          1    | 20,000,000  |    0     | shard-0.db.internal |
-- | 20,000,001    | 40,000,000  |    1     | shard-1.db.internal |
-- | 40,000,001    | 55,000,000  |    2     | shard-2.db.internal |
-- | 55,000,001    | 80,000,000  |    3     | shard-3.db.internal |
-- Note: ranges can be UNEQUAL — you can split a hot shard without touching others

-- Routing query:
SELECT shard_host FROM shard_directory
WHERE user_id_start <= 35000000 AND user_id_end >= 35000000;
-- Returns: shard-1.db.internal

-- Discord's composite shard key: (channel_id, bucket)
-- bucket = message_timestamp / (10 days in ms)
-- This ensures:
--   1. Recent messages for a channel are on the SAME shard (fast reads)
--   2. Old messages "age out" to different shards (cold storage)
--   3. No single shard grows forever

-- Cassandra (what Discord originally used):
CREATE TABLE messages (
    channel_id   BIGINT,
    bucket       INT,          -- epoch_ms / (10 * 86400 * 1000)
    message_id   BIGINT,       -- Snowflake ID (contains timestamp)
    author_id    BIGINT,
    content      TEXT,
    PRIMARY KEY ((channel_id, bucket), message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

-- The PRIMARY KEY ((channel_id, bucket), ...) means:
-- (channel_id, bucket) is the PARTITION KEY → determines which shard
-- message_id is the CLUSTERING KEY → determines sort order within partition

-- Query: "get latest 50 messages in channel 12345"
-- 1. Compute current bucket: now_ms / (10 * 86400 * 1000)
-- 2. Route to shard holding (12345, current_bucket)
-- 3. Read 50 messages sorted by message_id DESC
-- → Hits exactly ONE shard. Fast.

You've split 80M users across 4 shards. Traffic doubles in 6 months and you need 8 shards. With naive hash(key) % 4 becoming % 8, roughly 50% of all keys change shards — that's 40 million rows physically moving between machines. And what about that JOIN you used to run in 5ms? It now scatter-gathers across every shard, ballooning to 80ms. These are the hard problems that separate "I know what sharding is" from "I can operate a sharded system in production."

-- BEFORE SHARDING: One database, one JOIN, ~5ms
SELECT o.order_id, o.total, u.name, u.email
FROM orders o
JOIN users u ON o.user_id = u.id
WHERE o.created_at > '2025-01-01'
ORDER BY o.total DESC
LIMIT 100;

-- AFTER SHARDING: orders and users might be on DIFFERENT shards
-- The router must:
-- 1. Send the query to ALL N shards (scatter)
-- 2. Wait for ALL N responses
-- 3. Merge and sort ALL results in memory (gather)
-- 4. Return top 100

-- With 64 shards:
-- Best case (single shard, key in WHERE): ~5ms (same as before)
-- Worst case (scatter-gather, no shard key filter): ~5ms × 64 = 320ms
--   + merge overhead + network latency per shard

-- Vitess handles this automatically:
-- vtgate parses the query, sees it needs scatter-gather,
-- fans out to all vttablets, merges results
-- But it's still O(N) — no way around the physics

# Vitess: Split shard "-80" into "-40" and "40-80"
# This is how YouTube/Shopify scale their MySQL clusters

# 1. Create the target shards (empty MySQL instances)
vtctl CreateShard commerce/-40
vtctl CreateShard commerce/40-80

# 2. Start vttablets for new shards
# (launches MySQL + Vitess agent on each)

# 3. Begin the split — Vitess streams data from source to targets
vtctl Reshard -- --source_shards='-80' --target_shards='-40,40-80' Create
vtctl Reshard -- --source_shards='-80' --target_shards='-40,40-80' SwitchTraffic

# 4. During SwitchTraffic:
#    - Reads switch first (safe, idempotent)
#    - Writes switch second (brief pause, usually <1 second)
#    - vtgate automatically routes to new shards

# 5. Clean up the old shard once verified
vtctl Reshard -- --source_shards='-80' --target_shards='-40,40-80' Complete

Interviews love this question: "what's the difference between partitioning and sharding?" Most candidates get it wrong — and the confusion costs real architectural decisions. You can split a 600M-row table into 4 chunks on the same server (partitioning) or across 4 different servers (sharding). Same split, wildly different trade-offs. Let's untangle them.

-- Create a partitioned orders table by date range
CREATE TABLE orders (
    id          BIGSERIAL,
    user_id     BIGINT NOT NULL,
    total       DECIMAL(10,2),
    created_at  TIMESTAMPTZ NOT NULL,
    status      VARCHAR(20)
) PARTITION BY RANGE (created_at);

-- Create partitions for each quarter
CREATE TABLE orders_2025_q1 PARTITION OF orders
    FOR VALUES FROM ('2025-01-01') TO ('2025-04-01');
CREATE TABLE orders_2025_q2 PARTITION OF orders
    FOR VALUES FROM ('2025-04-01') TO ('2025-07-01');
CREATE TABLE orders_2025_q3 PARTITION OF orders
    FOR VALUES FROM ('2025-07-01') TO ('2025-10-01');
CREATE TABLE orders_2025_q4 PARTITION OF orders
    FOR VALUES FROM ('2025-10-01') TO ('2026-01-01');

-- Query automatically uses partition pruning:
EXPLAIN SELECT * FROM orders WHERE created_at > '2025-06-01';
-- Output shows: only orders_2025_q3 and orders_2025_q4 scanned
-- orders_2025_q1 and q2 are SKIPPED entirely

-- Hash partitioning (PostgreSQL 11+):
CREATE TABLE users (
    id      BIGSERIAL,
    name    TEXT,
    email   TEXT
) PARTITION BY HASH (id);

CREATE TABLE users_p0 PARTITION OF users FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE users_p1 PARTITION OF users FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE users_p2 PARTITION OF users FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE users_p3 PARTITION OF users FOR VALUES WITH (MODULUS 4, REMAINDER 3);
-- Each partition holds ~25% of users, evenly distributed

-- Time-based partitioning for application logs
CREATE TABLE app_logs (
    id          BIGSERIAL,
    timestamp   TIMESTAMPTZ NOT NULL,
    level       VARCHAR(10),
    message     TEXT,
    service     VARCHAR(50)
) PARTITION BY RANGE (timestamp);

-- Monthly partitions
CREATE TABLE app_logs_2025_01 PARTITION OF app_logs
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');
CREATE TABLE app_logs_2025_02 PARTITION OF app_logs
    FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');
-- ... one per month

-- Retention: drop data older than 90 days (instant, no DELETE + VACUUM)
DROP TABLE app_logs_2024_12;  -- Gone. Instantly. No IO overhead.

-- TimescaleDB makes this automatic:
-- CREATE EXTENSION timescaledb;
-- SELECT create_hypertable('app_logs', 'timestamp',
--     chunk_time_interval => INTERVAL '1 day');
-- SELECT add_retention_policy('app_logs', INTERVAL '90 days');
-- That's it. Chunks auto-create and auto-delete.

-- Elasticsearch (the most common log store) does the same with Index Lifecycle Management:
-- PUT _ilm/policy/logs_policy
-- { "phases": {
--     "hot": { "actions": { "rollover": { "max_age": "1d" }}},
--     "delete": { "min_age": "90d", "actions": { "delete": {} }}
-- }}

Instagram routes 1 billion users across 64 PostgreSQL shards using a Snowflake ID trick. Discord stores 120 billion messages across thousands of ScyllaDB nodes using a composite key. Shopify hides sharding entirely behind Vitess so developers never see it. MongoDB automates everything with an internal balancer. Four companies, four different architectures — each one shaped by their specific data access patterns.

-- Instagram's routing logic (simplified)
-- Shard = user_id % 64
-- user_id = 482391 → 482391 % 64 = 7 → Shard 7

-- Snowflake ID structure (64 bits):
-- [timestamp 41 bits][shard_id 13 bits][sequence 10 bits]

-- Given photo_id = 1375782301427_07_0042:
--   shard_id = 07 → route to shard-07.instagram.internal

-- pg_partman handles sub-partitioning within each shard:
SELECT partman.create_parent(
  'public.photos',
  'created_at',
  'native',
  'monthly'
);
-- Result: photos_2024_01, photos_2024_02, ... auto-created

-- Migration: split shard_03 into shard_03a + shard_03b
-- Step 1: Create new shard with logical replication
-- Step 2: Start replicating shard_03 → shard_03b
-- Step 3: Once caught up, update routing: user_id % 128
-- Step 4: Cut over, stop old replication

-- Discord's message table (simplified CQL)
CREATE TABLE messages (
  channel_id  bigint,
  bucket      int,        -- 10-day window: floor(epoch_days / 10)
  message_id  bigint,     -- Snowflake ID (contains timestamp)
  author_id   bigint,
  content     text,
  PRIMARY KEY ((channel_id, bucket), message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

-- Partition key = (channel_id, bucket)
-- Each partition = one channel's messages for 10 days
-- "Show latest messages in #general" → hits 1 partition (current bucket)
-- "Search last 90 days" → hits 9 partitions (9 buckets)

-- Hot channels (servers with 1M+ members) get dedicated nodes:
-- Discord's internal tooling detects hotspots and moves
-- heavy partitions to beefier ScyllaDB nodes with more RAM/CPU

-- Total scale: 120+ billion messages across thousands of nodes

# 1. Define how to shard the 'orders' table by merchant_id
vtctl ApplyVSchema -- --vschema='{
  "tables": {
    "orders": {
      "column_vindexes": [{
        "column": "merchant_id",
        "name": "hash"
      }]
    }
  }
}' commerce

# 2. Reshard: split 2 shards into 4 (when load doubles)
vtctl Reshard -- --source_shards='-80,80-' \
  --target_shards='-40,40-80,80-c0,c0-' commerce.reshard1

# 3. vtgate routes transparently — app sees one "database"
mysql -h vtgate.shopify.internal -u app -p commerce \
  -e "SELECT * FROM orders WHERE merchant_id = 12345;"
# vtgate: hash(12345) → keyspace_id → shard '-40' → route there

# 4. Check shard health
vtctl ListAllTablets | grep commerce

# Connect to mongos (the router)
mongosh --host mongos1.prod.internal

# Enable sharding on the database
sh.enableSharding("ecommerce")

# Shard the orders collection by hashed user_id
sh.shardCollection("ecommerce.orders", { "user_id": "hashed" })
# "hashed" = MongoDB hashes user_id for even distribution
# vs { "user_id": 1 } = range-based (good for range queries)

# Check shard distribution
sh.status()
# --- Sharding Status ---
# shards:
#   { "_id": "shard0", "host": "shard0/rs0-0:27017,rs0-1:27017" }
#   { "_id": "shard1", "host": "shard1/rs1-0:27017,rs1-1:27017" }
#   { "_id": "shard2", "host": "shard2/rs2-0:27017,rs2-1:27017" }
# databases:
#   { "_id": "ecommerce", "partitioned": true }
#     ecommerce.orders
#       shard key: { "user_id": "hashed" }
#       chunks:
#         shard0: 42
#         shard1: 41
#         shard2: 43
#       → balanced ✓

# When you query WITH the shard key, mongos routes to 1 shard:
db.orders.find({ user_id: 12345 })  // → targeted query (fast)

# When you query WITHOUT the shard key, mongos asks ALL shards:
db.orders.find({ status: "pending" })  // → scatter-gather (slow)

These are mistakes that look reasonable on a whiteboard but cause real pain in production. Each one comes from a company that learned the hard way.

These aren't theoretical pitfalls. Every one of these has caused production incidents, midnight pages, and painful migrations at real companies.

-- PostgreSQL: check shard sizes (run on each shard)
SELECT pg_size_pretty(pg_database_size('shard_01'));  -- 120 GB
SELECT pg_size_pretty(pg_database_size('shard_03'));  -- 450 GB ← problem!

-- MongoDB: check chunk distribution
sh.status()
-- shard0: 42 chunks
-- shard1: 41 chunks
-- shard2: 180 chunks  ← balancer might be stuck!

-- Vitess: check tablet health
vtctl ListAllTablets | awk '{print $1, $4, $5}'
-- shard-00 SERVING master
-- shard-01 SERVING master
-- shard-02 NOT_SERVING master  ← investigate!

The prompt: "Design the sharding strategy for a social media platform with 500M users, 10B posts, and 50K writes/sec." Here's how a junior, mid-level, and senior engineer would answer — and what the interviewer is looking for at each level.

Try these yourself before opening the solutions. The best way to internalize sharding is to work through the math and the trade-offs on your own.

import hashlib, bisect

class ConsistentHashRing:
    def __init__(self, virtual_nodes=150):
        self.virtual_nodes = virtual_nodes
        self.ring = {}         # hash_value → node_name
        self.sorted_keys = []  # sorted list of hash values

    def _hash(self, key: str) -> int:
        """Hash a string to a position on the ring (0 to 2^32-1)."""
        # TODO: use hashlib.md5 or sha256, return int

    def add_node(self, node: str):
        """Add a node with virtual_nodes copies on the ring."""
        # TODO: for each virtual node, compute hash and insert

    def remove_node(self, node: str):
        """Remove all virtual nodes for this node."""
        # TODO: remove entries from ring and sorted_keys

    def get_node(self, key: str) -> str:
        """Find which node a key maps to."""
        # TODO: hash the key, bisect into sorted_keys, return node

# Test: add 3 nodes, hash 10K keys, add 4th node, count moves

import hashlib, bisect

class ConsistentHashRing:
    def __init__(self, virtual_nodes=150):
        self.virtual_nodes = virtual_nodes
        self.ring = {}
        self.sorted_keys = []

    def _hash(self, key: str) -> int:
        digest = hashlib.md5(key.encode()).hexdigest()
        return int(digest, 16) % (2**32)

    def add_node(self, node: str):
        for i in range(self.virtual_nodes):
            vnode_key = f"{node}#vn{i}"
            h = self._hash(vnode_key)
            self.ring[h] = node
            bisect.insort(self.sorted_keys, h)

    def remove_node(self, node: str):
        for i in range(self.virtual_nodes):
            vnode_key = f"{node}#vn{i}"
            h = self._hash(vnode_key)
            del self.ring[h]
            self.sorted_keys.remove(h)

    def get_node(self, key: str) -> str:
        if not self.ring:
            return None
        h = self._hash(key)
        idx = bisect.bisect_right(self.sorted_keys, h)
        if idx == len(self.sorted_keys):
            idx = 0  # wrap around the ring
        return self.ring[self.sorted_keys[idx]]

# Test
ring = ConsistentHashRing(virtual_nodes=150)
for node in ["shard-0", "shard-1", "shard-2"]:
    ring.add_node(node)

# Hash 10K keys with 3 nodes
before = {f"key-{i}": ring.get_node(f"key-{i}") for i in range(10000)}

# Add 4th node
ring.add_node("shard-3")
after = {f"key-{i}": ring.get_node(f"key-{i}") for i in range(10000)}

# Count moves
moves = sum(1 for k in before if before[k] != after[k])
print(f"Keys moved: {moves}/10000 = {moves/100:.1f}%")
# Expected: ~25% (1/4 of keys move to the new node)