Phase 2 · Chapter 3

Sharding: When, How & What Breaks

How to split a database across multiple machines — and what you permanently give up when you do

Replication gives you multiple copies of the same data. Every node holds everything. Sharding is the opposite move: split the data itself so each node holds only a fraction of it. This is the only solution to the one problem replication cannot solve — the write bottleneck on a single primary. After this chapter you will understand exactly when sharding becomes necessary, the three strategies for deciding which node owns which data, and — most importantly — the specific database operations that become impossible or permanently expensive once your data is split. Sharding is not an upgrade. It is a tradeoff you make once and live with. This chapter will give you the judgment to know when to make it and what you are accepting when you do.

Part 1: The Problem Replication Cannot Solve

In Chapter 2 you added a replica and spread reads across two nodes. Your primary handles writes; your replica handles reads. This buys you significant headroom.

But consider what happens as your write volume grows. Every INSERT, UPDATE, and DELETE still goes to one machine — the primary. Replicas are irrelevant to write throughput. You can add ten replicas and your write capacity does not increase by a single operation per second.

Eventually the primary saturates. The signs: write latency climbs, WAL generation rate exceeds disk I/O capacity, connection slots fill up with waiting write transactions, CPU pegs at 100% processing commits. No hardware upgrade fully solves this — a bigger machine delays the problem, it does not eliminate it.

The only solution is to stop having one machine responsible for all writes. You need to split the data so that writes to different parts of it go to different machines simultaneously.

This is sharding — horizontal partitioning of a dataset across multiple independent database nodes, each called a shard, where each shard is the authoritative store for its portion of the data and accepts both reads and writes for that portion.

Each shard is typically its own primary-replica pair. You get both the write scaling of sharding and the read scaling and availability of replication within each shard.

Part 2: Before You Shard — The True Cost

Sharding is irreversible in practice. Once your data is split across shards, merging it back is a major engineering project. Before reaching for sharding, exhaust these options in order — each one is cheaper by an order of magnitude:

1. Vertical scaling. Bigger machine. More RAM means more of your working set fits in shared_buffers and page cache. More CPU handles more concurrent queries. This is embarrassingly cheap compared to sharding and buys years of headroom for most SaaS products.

2. Connection pooling. Postgres has a hard limit on concurrent connections (each connection forks a process — Phase 1 Chapter 1). PgBouncer in transaction mode multiplexes thousands of application connections onto tens of database connections. Many "the database is overwhelmed" situations are actually connection exhaustion, not capacity exhaustion.

3. Caching. Most read load is for data that does not change often. Redis in front of Postgres for hot data eliminates the read pressure entirely. A Postgres node that handles 100k reads/second might handle 10k if the hot 10% of data is cached.

4. Read replicas. Covered in Chapter 2. Spreads read load horizontally with no data model changes required.

5. Postgres native partitioning. Postgres supports table partitioning — splitting one logical table into multiple physical tables (called partitions) on the same node, managed transparently by the query planner. This is not sharding across machines, but it dramatically improves query performance and vacuum efficiency on very large tables (hundreds of millions of rows) by limiting the pages that need to be scanned for time-ranged or key-ranged queries.

Only when these options are genuinely exhausted does sharding become necessary. For context: Instagram ran on a single Postgres primary until tens of millions of users. Shopify ran a single database per merchant for years. The bar for needing sharding is higher than most engineers think.

Part 3: The Shard Key — The Decision That Determines Everything

When you shard, you must choose a shard key: the field whose value determines which shard a given row belongs to. Every query, insert, update, and delete will be routed based on this key.

This is the most consequential architectural decision in a sharded system. A wrong shard key produces hot spots, cross-shard queries, or an unreshard-able mess. Getting it right requires understanding your access patterns before you design your sharding strategy.

The shard key must satisfy three properties:

High cardinality. The key must have enough distinct values to distribute data across shards evenly. A boolean field (active/inactive) cannot be a shard key — all active users would pile onto one shard. A user_id with millions of distinct values is suitable.

Even distribution. The values must spread evenly across the key space. If 80% of your users were created in the last month and you shard by created_at, 80% of your data is on the "recent" shard. Uneven distribution creates hot shards that defeat the purpose of sharding.

Query locality. Your most frequent queries should be answerable from a single shard. If your most common query is "get all data for user X," then sharding by user_id means every such query hits exactly one shard. If you shard by something unrelated to your access pattern, every query becomes a cross-shard scatter-gather — you ask all shards and merge the results. This is slower than not sharding at all.

Part 4: Three Sharding Strategies

Given a shard key, you need a rule that maps a key value to a shard. There are three approaches, each with different properties.

Range sharding

Divide the key space into contiguous ranges. Each shard owns a range.

user_id 1          → 33,333,333   : Shard A
user_id 33,333,334 → 66,666,666   : Shard B
user_id 66,666,667 → 100,000,000  : Shard C

The routing rule is simple: compare the key value against the range boundaries, return the matching shard. A lookup table or configuration file holds the boundaries.

Advantage: Range queries are efficient. "Give me all users created between date X and date Y" — if you shard by created_at, this query might hit only one shard. Scanning a range of keys means scanning a range of shards, not all of them.

Disadvantage: Hot spots. New users always get the highest user_id. The shard that owns the highest range receives all inserts. The other shards sit idle. This is called a write hotspot and it is the primary failure mode of range sharding on monotonically increasing keys (IDs, timestamps). You can mitigate this by sharding on a non-monotonic key or by periodically rebalancing ranges — but rebalancing requires moving data between shards, which is expensive.

Hash sharding

Apply a hash function to the shard key. Assign rows to shards based on the hash value modulo the number of shards.

function getShard(userId, numShards) {
  return murmurhash(userId) % numShards;
}

getShard('user_123', 3)  → shard 0
getShard('user_456', 3)  → shard 2
getShard('user_789', 3)  → shard 1

Hash functions distribute uniformly by design. No hot spots on insert. Every shard receives approximately equal write volume regardless of key distribution or whether keys are monotonically increasing.

Advantage: Even distribution. No write hotspots. Works well for any key type.

Disadvantage: Range queries become scatter-gather. "Give me all users with IDs between 1000 and 2000" — the hash of 1001 lands on shard 2, the hash of 1002 lands on shard 0, the hash of 1003 lands on shard 1. There is no locality. You must query all shards and merge. If range queries are common in your access pattern, hash sharding is the wrong choice.

The second disadvantage: resharding. When you add a 4th shard, hash % 4 produces different shard assignments than hash % 3. Approximately 75% of all rows map to a different shard. You must physically move 75% of your data. This is the resharding problem — the same problem as modulo hashing in distributed caches, which is exactly why consistent hashing was invented (Phase 3 Chapter 2 will derive this from first principles). For now: hash sharding with static shard count is operationally painful to scale.

Directory sharding (lookup-based sharding)

Maintain a separate lookup table that maps each shard key value (or range of values) to a shard. Every routing decision queries this table first.

Lookup table:
  tenant_id = 'acme_corp'    → Shard C
  tenant_id = 'globex'       → Shard A
  tenant_id = 'initech'      → Shard A
  tenant_id = 'umbrella'     → Shard B

Advantage: Complete flexibility. You can move any tenant to any shard at any time by updating the lookup entry. You can rebalance hot shards by migrating specific tenants. You can give large enterprise customers dedicated shards. No mathematical constraint on routing.

Disadvantage: The lookup table is a single point of failure and a performance bottleneck. Every database operation requires a lookup table read first. You must cache this table aggressively (it fits in Redis easily) and handle the cache invalidation problem when you move tenants.

Directory sharding is the dominant pattern for multi-tenant SaaS — each tenant (customer organisation) is a shard key, and you route their data to a specific shard. Shopify, Salesforce, and similar platforms use this model. It maps naturally to the business concept of tenant isolation and gives you the flexibility to handle "whale" customers with dedicated resources.

Range versus hash is easier to feel than to read. Predict which shard takes the writes, then switch strategies and watch what a time-range query costs.

Part 5: What Breaks When You Shard

This is the section most sharding tutorials skip. Sharding does not just split your data — it invalidates a class of database operations that you have been using without thinking about them. Every item below is something that works trivially on a single node and becomes a significant engineering problem on a sharded system.

Cross-shard joins

On a single Postgres node, joining two tables is a B-tree lookup and a hash join — the query planner handles it automatically, both tables are on the same machine.

When the tables are on different shards, the join cannot happen inside the database. You must fetch the relevant rows from each shard into your application and join them in memory. This means:

• Multiple round trips to multiple database connections
• The join logic is now in your application code, not in SQL
• The result set might require fetching millions of rows into memory if the join is not selective
• Query planner optimizations (index scans, hash joins, merge joins) do not apply

If your most common queries involve joining tables that will be on different shards, your shard key is wrong. Redesign the sharding strategy before deploying, not after.

Cross-shard transactions

ACID transactions (Chapter 1) are enforced by a single Postgres instance. A transaction that touches rows on two different shards cannot be handled by either shard's transaction manager alone.

The solution is a distributed transaction — a two-phase commit (2PC) protocol where a coordinator tells each involved shard to prepare, waits for confirmation, then tells each to commit. If any shard fails to prepare, the coordinator tells all shards to rollback.

Two-phase commit works but it is:

• Slow (multiple network round trips in the commit path)
• Failure-prone (coordinator crash leaves shards in prepared state indefinitely)
• Not supported natively by standard Postgres sharding setups

In practice, most sharded systems avoid distributed transactions by designing the data model so that operations requiring atomicity always touch a single shard. This is the primary reason shard key selection matters so much — the key must align with your transactional boundaries.

When distributed transactions are unavoidable, the alternative is saga patterns — break the operation into a sequence of local transactions on individual shards, with compensating transactions to undo completed steps if a later step fails. This moves the consistency logic into your application. It is correct but complex.

Global uniqueness

On a single node, a UNIQUE constraint on a column prevents duplicate values across the entire table. On a sharded system, each shard enforces uniqueness only within its own data. Two shards can each contain a row with the same value and neither will detect the conflict.

If you need global uniqueness — for usernames, email addresses, order numbers — you have three options:

Route all uniqueness-constrained operations through a single dedicated node (a global directory). This node becomes a bottleneck for any operation that must check global uniqueness.

Generate IDs in a way that guarantees global uniqueness without coordination: UUIDs (128-bit random values — collision probability is negligible), or Snowflake IDs (64-bit IDs encoding timestamp

• machine ID + sequence number — used by Twitter, Discord, and most large-scale systems). Snowflake IDs are globally unique by construction, require no coordination, and are sortable by time.

Use a distributed ID generation service (Twitter's Snowflake, Instagram's ID service, Postgres's sequence on a dedicated node with caching).

Aggregations and analytics

SELECT COUNT(*) FROM users on a sharded system requires asking every shard for its count and summing the results in the application. The same applies to SUM, AVG, MIN, MAX, GROUP BY, ORDER BY with LIMIT — any query whose result depends on data across multiple shards.

For aggregate analytics, the standard solution is to not run them against the sharded transactional database at all. Instead, replicate all shards to a centralized analytics database (a data warehouse or a dedicated Postgres instance) via logical replication and run analytics there. The analytics database is eventually consistent but allows cross-shard queries and complex aggregations.

Schema migrations

On a single node, ALTER TABLE runs once. On a sharded system with 10 shards, the migration must run on every shard. If the migration takes time (adding an index to a large table), it takes time on each shard. You need tooling to track migration state across shards, handle partial failures, and ensure the application can operate during migrations that have completed on some shards but not others (requiring backward-compatible migration practices: add column before using it, never remove a column in the same deploy that stops writing to it).

Part 6: Postgres Native Partitioning

Before building application-level sharding or deploying Citus, consider Postgres's built-in table partitioning. This is not sharding across machines — everything is on one node — but it solves a large class of performance problems for tables with hundreds of millions of rows.

-- Create a partitioned table (partition by month)
CREATE TABLE test_attempts (
  id          BIGSERIAL,
  user_id     BIGINT,
  created_at  TIMESTAMPTZ NOT NULL,
  score       INT
) PARTITION BY RANGE (created_at);

-- Create partitions
CREATE TABLE test_attempts_2025_01
  PARTITION OF test_attempts
  FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

CREATE TABLE test_attempts_2025_02
  PARTITION OF test_attempts
  FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');

A query with WHERE created_at BETWEEN '2025-01-01' AND '2025-01-31' scans only the test_attempts_2025_01 partition. The query planner calls this partition pruning — it eliminates partitions that cannot contain matching rows before the query executes. For a table with 5 years of data split into monthly partitions (60 partitions), a monthly query reads 1/60th of the data a non-partitioned table would require.

Vacuum also benefits: each partition is vacuumed independently. A high-churn current-month partition gets frequent vacuuming without touching cold historical partitions.

Partitioning works best with time-series data (logs, events, attempts) where queries are naturally scoped to recent time windows. For AttemptIQ's test_attempts table with millions of historical rows, partitioning by month is the correct scaling step before considering cross-machine sharding.

Part 7: Application-Level Sharding vs Citus

When you do need cross-machine sharding, you have two implementation paths.

Application-level sharding

Your application maintains a shard map (in memory, in Redis, or in a config file). Every database query is routed by the application to the correct shard connection based on the shard key.

class ShardRouter {
  constructor(shards) {
    this.shards = shards;  // array of database connections
  }

  getConnection(shardKey) {
    const shardIndex = murmurhash(String(shardKey)) % this.shards.length;
    return this.shards[shardIndex];
  }
}

// Usage
const db = router.getConnection(userId);
const user = await db.query('SELECT * FROM users WHERE id = $1', [userId]);

This is simple, transparent, and gives you full control. The downside is that every query in your codebase must be shard-aware. Cross-shard operations are explicit — you must implement them manually. Schema migrations must be run manually against each shard.

Citus (now part of Azure Postgres)

Citus is a Postgres extension that implements sharding inside the database engine. Your application connects to a Citus coordinator node and sends normal SQL queries. The coordinator routes queries to the correct worker shards transparently, parallelises cross-shard queries automatically, and returns merged results to the application.

-- Citus: declare sharding strategy (run on coordinator)
SELECT create_distributed_table('users', 'id');  -- shard users by id
SELECT create_distributed_table('orders', 'user_id', colocate_with => 'users');
-- orders are on the same shard as the user they belong to
-- JOIN users u ON o.user_id = u.id now executes locally on each shard

The colocate_with parameter is the key concept: put related tables on the same shard so that joins between them remain local. This is the Citus answer to the cross-shard join problem — design co-location deliberately rather than discovering join pain later.

Citus is the correct path when you want sharding with minimal application changes and you are already on Postgres. The cost: Citus is a significant operational dependency, and the coordinator is a single point of failure (though it can be made redundant).

Part 8: Resharding — The Operation Nobody Wants

Your system is sharded across 4 nodes. Traffic grows. You need 8 nodes. You must move half your data from existing shards to new shards without downtime.

This is resharding and it is genuinely hard. The naive approach:

1. Stand up 4 new empty shards
2. For each row in existing shards, compute its new shard assignment
3. Copy rows that belong to new shards to those new shards
4. Flip application routing to new shard map
5. Delete migrated rows from old shards

The problem is steps 3 and 4. During the copy, new writes are arriving on the old shards. By the time you finish copying, some rows have been updated after they were copied — the copies are stale. You must either lock the old shard during copy (downtime) or implement a change-data-capture mechanism that streams writes to new shards during the migration (complex, but zero-downtime).

Consistent hashing (Phase 3 Chapter 2) was specifically invented to minimise the amount of data that moves during resharding — from O(all data) to O(data / new shard count). But it reduces the pain, it does not eliminate it. Any production resharding operation requires careful planning, staging, and a practiced rollback plan.

This is why the advice at the start of this chapter is genuine: shard when you must, not when you think you might need to. The initial sharding and every subsequent resharding are expensive operations that consume engineering time and introduce operational risk.

Vocabulary Map

Practice Problems

Problem 1 — Failure Analysis: AttemptIQ shards its test_attempts table by user_id using hash sharding across 4 shards. The product launches a leaderboard feature: "Top 100 scorers this month across all users."

Describe exactly what happens when this query executes. How many database round trips does it require? What does the application code need to do that a single-node Postgres would handle automatically? What would you change about the architecture to make leaderboard queries efficient without abandoning sharding for the attempts table?

Problem 2 — Design From Scratch: Samiksha is a multi-tenant SaaS. Each "tenant" is a coaching institute with up to 50,000 students. You currently have 200 tenants and expect 2,000 within two years.

Design a sharding strategy. Choose a shard key with justification. Which of the three sharding strategies (range, hash, directory) is most appropriate and why? What operations become cross-shard and how do you handle them? What do you do when a single large tenant (10,000 students, very active) generates 30% of total write traffic on their shard?

Problem 3 — What Breaks: Your application has this query that runs 10,000 times per day:

SELECT u.name, u.email, COUNT(a.id) as attempt_count, AVG(a.score) as avg_score
FROM users u
JOIN test_attempts a ON a.user_id = u.id
WHERE u.subscription_tier = 'pro'
GROUP BY u.id, u.name, u.email
ORDER BY avg_score DESC
LIMIT 50;

You shard by user_id using hash sharding across 8 shards. Walk through every part of this query and describe what breaks, what becomes expensive, and what the application must now do to produce the same result. Is this query reshapeable to be shard-local, or is it fundamentally cross-shard?

Problem 4 — Go Deeper: Hash sharding with static shard count has a resharding problem: adding a shard requires moving approximately (N-1)/N of all data. You are about to study consistent hashing in Phase 3 Chapter 2, but reason from first principles now: what property would an ideal shard assignment function have that minimises data movement when a new shard is added? Why does hash(key) % N not have this property? Sketch what a better function's output would look like — not the algorithm, just the desired behaviour.

Problem 5 — Connect to Previous Chapters: In Chapter 1 you learned about MVCC and long-running transactions preventing VACUUM on a Postgres node. In Chapter 2 you learned that a long-running read on a replica can prevent VACUUM on the primary via the replication slot mechanism.

In a sharded system with 8 shards, each a primary-replica pair: a buggy analytics job opens a transaction on Shard 3's replica and holds it open for 6 hours while processing results. Trace the full chain of consequences: what happens on Shard 3's replica, what happens on Shard 3's primary, and — if this shard holds 12.5% of user data — what is the user-visible symptom when the primary's table bloat causes sequential scans to degrade? Which Postgres system view would show you the root cause?

What's Next

You now understand the three layers of Postgres scaling: transactions on a single node (Chapter 1), replication to multiple copies of all data (Chapter 2), and sharding to split the data itself (Chapter 3). There is one more dimension of database performance that operates entirely within a single node and is responsible for more production slowdowns than any of the above: writing queries that Postgres can execute efficiently, and understanding why it sometimes chooses the wrong plan. Chapter 4 opens Postgres's query planner — the engine that decides how to execute every SQL statement you write — and gives you the tools to diagnose and fix slow queries before they become incidents.

End of Phase 2, Chapter 3.