Chapter 13: Database Partitioning & Sharding

The Scalability Wall

Single-node limits:

• Storage capacity: Dataset exceeds disk capacity (e.g., 10TB data, 2TB disk)
• Write throughput: Single disk I/O maxed out (e.g., 10K writes/sec needed, 2K capable)
• Read throughput: CPU/memory saturated serving queries
• Query performance: Working set doesn't fit in RAM, causing excessive disk I/O

Solution: Split data across multiple nodes (horizontal scaling)

Mechanism: Assign continuous ranges of keys to each partition.

Example: Users A-F → Shard 0, G-M → Shard 1, N-S → Shard 2, T-Z → Shard 3

Mechanism: Hash the partition key, modulo by number of shards.

Example: shard = hash(user_id) % num_shards

Mechanism: Partition by hash of partition key, sort within partition by sort key.

Example: DynamoDB: partition_key = tenant_id (hashed), sort_key = timestamp (sorted)

Mechanism: Maintain a lookup table mapping keys (or key ranges) to shards.

Example: Vitess (MySQL): VSchema maps keyspace ranges to tablets

Mechanism: Partition by geographic region (country, continent, data center).

Use case: Data locality for latency, compliance (GDPR: EU data in EU).

The Celebrity Problem

Scenario: Twitter-like system, hash-partition by user_id. Celebrity with 100M followers posts a tweet.

Problem: All 100M followers' timeline writes go to the same shard (celebrity's shard), overwhelming it.

Mechanism: For hot keys, append a random suffix to distribute across multiple shards.

Original key: celebrity_id = 12345
Salted keys:  celebrity_id_0, celebrity_id_1, ..., celebrity_id_9

Read:  Fan-out to all 10 salted keys, merge results
Write: Randomly pick one salt (or write to all for updates)

Mechanism: Detect hot partitions (by monitoring metrics), split them dynamically.

Example: Range partition [A-M] is hot → split into [A-F], [G-M]

Mechanism: Prepend timestamp keys with random prefix to avoid sequential write hot spot.

Original key: 2024-05-12T10:30:00_event123
With prefix:  7_2024-05-12T10:30:00_event123  (prefix 0-9)

All writes distributed, but range scans now hit all partitions.

Problem: Primary key partitioning (e.g., user_id) is efficient for lookups by user_id. But what if you need to query by email or country?

Solution: Secondary indexes. But in a partitioned DB, where do you store them?

Mechanism: Each shard maintains its own index for documents in that shard.

Query path: Query all shards (scatter), merge results (gather).

Mechanism: Index is independently partitioned (different partition key from main table).

Example: Main table partitioned by user_id, index partitioned by country.

The Rebalancing Problem with Naive Hashing

Naive hash: shard = hash(key) % N

Problem: Adding/removing a shard changes N → rehash ALL keys

Example: 4 shards → 5 shards
  Old: hash(key) % 4
  New: hash(key) % 5
  → 80% of keys move to different shards!

Mechanism:

• Hash space is a ring: 0 to 2^32-1, wraps around
• Nodes placed at random points on ring (hash of node ID)
• Keys placed on ring (hash of key)
• Key assigned to first node clockwise from its position

Problem: With few nodes, hash distribution can be uneven (one node owns 40%, another 10%).

Solution: Each physical node represented by multiple virtual nodes on the ring.

Problem: Hash ring requires O(N) metadata (node positions), O(log N) lookup.

Jump Hash: Stateless algorithm, O(1) space, O(log N) compute.

int32_t JumpConsistentHash(uint64_t key, int32_t num_buckets) {
    int64_t b = -1, j = 0;
    while (j < num_buckets) {
        b = j;
        key = key * 2862933555777941757ULL + 1;
        j = (b + 1) * (double(1LL << 31) / double((key >> 33) + 1));
    }
    return b;
}

Why rebalance?

• Data growth: partitions exceed size limits
• Query throughput: one partition overloaded
• Cluster expansion: added new nodes, need to spread load
• Cluster shrink: removed nodes, need to redistribute data

Mechanism: Create many more partitions than nodes upfront. Assign multiple partitions per node. When adding nodes, move whole partitions.

Example: 1000 partitions, 10 nodes → 100 partitions/node. Add 5 nodes → rebalance to 67 partitions/node.

Mechanism: Start with small number of partitions. Split partitions when they grow too large. Merge when too small.

Example: HBase starts with 1 region per table. Auto-split at 10GB (configurable).

Mechanism: Number of partitions proportional to node count (e.g., 10 partitions per node).

Example: 5 nodes → 50 partitions. Add 5 nodes → split to 100 partitions, rebalance.

1. Incremental Rebalancing

Move one partition at a time, not all at once. Prevents overwhelming the network/cluster.

2. Rate Limiting

Throttle data movement (e.g., max 100MB/s per node). Avoid impacting prod traffic.

3. Prioritization

Move small partitions first (faster, lower risk). Save large partitions for off-peak hours.

4. Snapshot-Based Migration

Take snapshot, copy to new node, replay incremental writes. Minimizes downtime.

5. Monitoring

Track progress, partition counts, data distribution. Rollback mechanism if issues.

Problem: Query needs data from multiple shards (e.g., "Get all orders for product X" when partitioned by customer_id).

Solution: Scatter-gather pattern (query all shards, merge results).

Approach: Duplicate data to support different query patterns without cross-shard access.

Original:
- orders table (partition by customer_id)

Denormalized:
- orders_by_customer (partition by customer_id)
- orders_by_product (partition by product_id)
- orders_by_date (partition by date)

Trade-off: 3x storage, write amplification, eventual consistency

Approach: Pre-compute aggregations across shards, store in separate table.

Example: Daily sales by product
- Background job: scatter-gather query all shards
- Compute SUM(amount) per product per day
- Store in product_daily_sales (single shard or small table)

Query: Fast lookup in materialized view (no cross-shard query)
Freshness: Updated hourly/daily (not real-time)

Approach: Fetch data in multiple round trips, join in application code.

Query: "Get customer details + recent orders"

Step 1: Fetch customer from customers table (partition by customer_id)
Step 2: Fetch orders for that customer from orders table (same shard)
Step 3: Join in application code

Cost: 2 sequential queries (OK if both hit same shard)

Problem: Transaction spans multiple shards (e.g., transfer $100 from account on Shard 1 to account on Shard 2).

Challenge: ACID guarantees are hard across shards (no single coordinator).

Mechanism: Coordinator asks all shards to prepare (Phase 1), then commit (Phase 2) if all agree.

Mechanism: Break transaction into independent local transactions. If one fails, run compensating transactions to undo previous steps.

Trade-off: Eventual consistency (not ACID), but no blocking locks.

Mechanism: Pre-order all transactions deterministically before execution. All shards execute in same order (no coordination needed).

Example: FaunaDB, derived from Calvin research.

Key innovation: TrueTime API (GPS + atomic clocks) provides globally synchronized time with bounded uncertainty.

Result: Serializable distributed transactions without blocking (external consistency).

Most critical decision: Shard key determines data distribution, query patterns, scalability.

Wrong shard key: Hot spots, cross-shard queries, expensive resharding.

Metrics to Monitor

• Data distribution: Standard deviation of data size per shard (goal: <10% variance)
• Request distribution: CV (coefficient of variation) of QPS per shard (goal: <0.2)
• Hot key frequency: % of requests hitting top 1% of keys (goal: <5%)
• Cross-shard query rate: % of queries requiring scatter-gather (goal: <10%)

When to reshard:

• Original shard count insufficient (outgrew initial estimate)
• Bad shard key chosen (hot spots, cross-shard queries)
• Data skew developed over time (uneven growth)
• Major workload change (OLTP → analytics)

Mechanism: Write to both old and new shards during migration. Gradually shift reads to new shards.

Mechanism: Replay production traffic to new shards (reads only), validate correctness before cutover.

1. Mirror 1% of reads to new shards (discard results)
2. Compare results: old vs new (alert on mismatch)
3. Gradually increase to 100% shadow traffic
4. Once validated, flip cutover flag (reads go to new shards)
5. Enable writes to new shards, stop old shards

Vitess: Horizontal sharding layer for MySQL. Handles resharding transparently.

Resharding workflow:
1. SplitClone: Copy data from source shard to destination shards
2. Filtered replication: Stream new writes from source to destinations
3. Cutover: Switch traffic from source to destinations (atomic flag flip)
4. Drop source shard after validation

Example: 1 shard → 2 shards
- Original: shard0 (all key ranges)
- Target: shard0 (keys 0x00-0x7F), shard1 (keys 0x80-0xFF)
- Vitess handles routing during migration (app unaware)

MongoDB: Automatic balancing via chunk migration.

How it works:
1. Config server tracks chunk distribution
2. Balancer runs periodically (every 10 min by default)
3. Detects imbalance: one shard has 8+ more chunks than another
4. Moves one chunk at a time (to minimize impact)
5. Migration steps:
   - Clone chunk to destination shard
   - Apply incremental changes (oplog replay)
   - Atomic cutover (update config server routing)
   - Delete chunk from source shard

Manual resharding: sh.splitAt(), sh.moveChunk() commands

pg_partman: Partition management extension for PostgreSQL.

Features:
- Automatic creation/dropping of time-based/serial-based partitions
- Retention policies (auto-drop old partitions)
- Reindexing, vacuum automation

Resharding approach:
1. Create new partition scheme (e.g., monthly → daily partitions)
2. Backfill data into new partitions (parallel copy)
3. Update partition routing (parent table points to new partitions)
4. Drop old partitions after validation

Not true sharding: Single PostgreSQL instance (vertical scaling limit)

Pre-Migration

• ✓ Validate new shard key (run distribution analysis on sample)
• ✓ Provision new shards (capacity planning: 50% headroom)
• ✓ Setup monitoring (data sync lag, error rate, query latency)
• ✓ Test rollback procedure (dry-run on staging)
• ✓ Communicate to stakeholders (expected timeline, risk)

During Migration

• ✓ Start backfill (low priority, rate-limited)
• ✓ Monitor sync lag (alert if >1 hour)
• ✓ Enable double-write (validate write amplification acceptable)
• ✓ Shadow traffic (validate read results match)
• ✓ Gradual cutover (1% → 10% → 50% → 100%)
• ✓ Monitor error rate (rollback if >0.1% increase)

Post-Migration

• ✓ Validate data completeness (row counts, checksums)
• ✓ Performance comparison (query latency P50/P99)
• ✓ Monitor for 7 days (no alerts → proceed)
• ✓ Decommission old shards (after backup to cold storage)
• ✓ Document lessons learned (post-mortem)

Architecture

• Partition key: Determines partition (consistent hashing)
• Sort key (optional): Defines order within partition
• Partition size limit: 10GB per partition
• Throughput: 3000 RCUs, 1000 WCUs per partition

Auto-Scaling & Adaptive Capacity

Partition split:
- Trigger: Partition exceeds 10GB
- Split: Divide key range in half, create 2 new partitions
- Throughput: Each new partition gets 3000 RCU, 1000 WCU

Adaptive capacity:
- Detect hot partition (exceeds provisioned throughput)
- Isolate hot keys to dedicated capacity
- Automatically increase throughput for hot partition

On-Demand mode:
- No provisioned throughput limits
- Auto-scales to handle any load
- Pay per request (more expensive than provisioned)

Architecture

• Token range: 2^64 hash space (Murmur3 hash)
• Virtual nodes: 256 vnodes per physical node (default)
• Replication: RF=3 (3 replicas), placed on next N nodes clockwise
• Consistency: Tunable quorum (CL=QUORUM: R+W>RF)

Architecture

• Chunk: Contiguous range of shard keys (default 64MB)
• Config servers: 3-node replica set storing chunk mappings
• Mongos: Query router (stateless, determines which shard to query)
• Shards: Replica sets storing actual data

Architecture

• Shard: Lucene index (immutable, fixed count per index)
• Replica: Copy of shard (for HA, read scaling)
• Routing: hash(document_id) % num_primary_shards
• Allocation: Master node assigns shards to data nodes

Architecture

• Tablet: MySQL instance + VTTablet process (query rewriting, pooling)
• Keyspace: Logical database (contains sharded tables)
• Shard: Horizontal partition of keyspace
• VTGate: Query router (stateless, MySQL protocol compatible)