Sharding

When database size and query traffic exceed the physical capacities of a single server, traditional scaling reaches its limit. While read replication helps handle read-heavy workloads, it does not scale write capacity or total storage limit. To achieve unlimited write scalability and manage multi-terabyte datasets, databases must be horizontally partitioned. This architectural technique of breaking up a single logical database into smaller, autonomous database instances across separate physical nodes is called Sharding.

1. Learning Objectives

• Differentiate between Horizontal Partitioning (Sharding) and Vertical Partitioning.
• Master the mechanics of three main sharding strategies: Range-Based, Hash-Based, and Directory-Based sharding.
• Analyze how sharding alters the Database Read and Write paths.
• Evaluate strategies for selecting shard keys and mitigating write hotspots.
• Understand the operational complexities of cross-shard joins, transactions, and rebalancing.
• Implement a fully functioning sharding router with multiple routing policies in Java, Python, and C++.

2. Prerequisites

To get the most out of this lesson, you should review the following topics first:

• Databases and DBMS: Basic knowledge of database servers, tables, and rows.
• SQL vs NoSQL: How relational engines store data versus schema-less document stores.
• Database Replication: Primary-replica setups for scaling reads and achieving high availability.
• Transactions & ACID: Standard isolation levels and single-node atomicity guarantees.

3. Why This Topic Matters

A single, high-spec relational database server can comfortably handle tens of thousands of reads per second when indexed properly. However, physical hardware constraints present hard limits:

• Storage Capacity: A single server can only host as much data as its physical NVMe/SSD drives allow.
• Write Throughput: All writes must eventually commit to the storage engine's Transaction Log/Write-Ahead Log (WAL), creating a strict I/O bottleneck.
• Memory Limits: Working sets (indexes and hot data) must ideally fit into RAM to maintain sub-millisecond query performance.

Adding read replicas allows you to scale read volume infinitely, but *every replica must process all writes* to stay synchronized. Therefore, write scalability does not improve. Sharding solves this by distributing the storage and write processing load across a cluster of independent physical machines. Without sharding, global-scale systems like YouTube, Instagram, or financial ledgers could not exist.

4. Real-world Analogy

Imagine you manage a rapidly expanding city archive containing physical paper folders for millions of citizens.

Vertical Partitioning: If folders get too thick, you split them up. You keep medical records in building A, tax records in building B, and employment history in building C. However, if the city grows 100x, the medical records building itself will eventually run out of physical space.

Horizontal Partitioning (Sharding): Instead of splitting records by category, you keep the same full folder structure but divide the citizens alphabetically. Citizens with last names starting with A-F are housed in Archive Office 1, G-M in Office 2, N-T in Office 3, and U-Z in Office 4. Each office is completely independent, has its own filing staff, and manages its own building space. If the population doubles, you can add a fifth office and re-distribute some alphabet ranges without changing how the records are structured.

5. Core Concepts

• Horizontal Partitioning: Splitting table rows across multiple tables or databases. The schema remains identical across all instances, but each database contains a unique subset of the rows. This is the structural basis of Sharding.
• Vertical Partitioning: Splitting table columns into different tables. For example, moving a large, rarely queried binary column (like user profile images) from a primary users table to a separate user_metadata table to reduce row size and optimize disk memory layout.
• Shard Key: A specific column (or combination of columns) chosen to determine which physical shard a given row belongs to. Common choices include user_id, tenant_id, or country_code.
• Sharding Strategies:
  • Range-Based: Mapping ranges of the shard key value to specific shards (e.g., values 1 to 1,000,000 go to Shard 1).
  • Hash-Based (Algorithmic): Applying a hash function to the shard key and using modulo math to select the shard index: shard_id = hash(shard_key) % total_shards.
  • Directory-Based (Lookup): Using a central lookup database or service (like ZooKeeper/etcd or a mapping table) to explicitly map shard keys to their target shards.
• Query Routing (Proxy vs. Client-Side):
  • Proxy-Based: A middleware component (e.g., Vitess for MySQL) sits between the application and the shards. The application sends standard queries to the proxy, which parses the SQL, resolves the shard key, routes the query, and aggregates results.
  • Client-Side Routing: The application library itself knows the sharding rules and maintains direct connections to every shard, routing traffic natively.

6. Visualizations

Horizontal vs. Vertical Partitioning

Query Routing Architecture (Proxy vs. Client-Side)

7. How It Works Step-by-Step

Let's follow how a client application saves a record and retrieves it in an algorithmic (hash-based) sharded system:

1. Extract Key: The application issues a write command containing data with a chosen shard key field, e.g., user_id = 948271.
2. Apply Hash Function: The routing layer hashes the key to generate a uniform integer value:

Apply Shard Modulo: The routing layer calculates the target physical shard index using the current total shard count (e.g., 4 shards):

Connection Routing: The routing engine checks its internal connection pool, fetches a socket handle for Shard 3, and issues the exact query:

Local Storage: Shard 3 (an independent database instance) executes the write command locally, indexing the row in its own local B-Tree or LSM tree, and returns success to the router.

Result Aggregation: The router bubble-up response returns to the calling client application code.

8. Internal Architecture

A fully sharded database system contains several critical architectural blocks:

• Application Client: The user service generating read and write queries.
• Routing Coordinator (Proxy or SDK Router): The brain of the sharding routing logic. It parses SQL parsed syntax trees, extracts the shard key from WHERE or INSERT clauses, and matches keys to physical connection strings.
• Configuration Metadata Store: An ultra-reliable, highly available distributed key-value store (e.g., etcd, ZooKeeper) that tracks physical coordinates (IP, port, master/replica status) for all shards, and defines the range boundaries or lookup tables.
• Shard Nodes (Data Engine): The actual worker nodes (e.g., MySQL, Postgres, MongoDB) executing queries. Each shard node is configured as a standalone replica set (a primary writer and secondary readers) to guarantee high availability at the node level.

9. Request Lifecycle

Let's walk through how request lifecycles behave differently depending on the query payload.

Lifecycle A: Single-Key Point Write

Lifecycle B: Scatter-Gather Read Query

What happens when the query does not contain the shard key? (e.g. searching users by registration age):

10. Deep Dive

A. The Read & Write Paths on a Single Shard Node

When a write query reaches a specific shard database (e.g., PostgreSQL Shard A), it processes it using standard single-node query execution. It logs the statement to the Write-Ahead Log (WAL), reserves memory space in cache pool blocks, and inserts the record into the local table indexes.

The indexing challenge: Under sharding, indexes are local. A secondary index (e.g., indexing email columns on a table sharded by user_id) is only valid *within that physical shard*. If you query a user by email, the router cannot compute the target shard. It must broadcast the query to all shards to check their local indexes. This is known as a scatter-gather query and it degrades system performance if done frequently.

B. Shard Key Selection Criteria

Choosing the correct shard key is the single most critical decision in database sharding. Once selected, changing it is extremely difficult, requiring a complete backup, re-partition, and restore. A good shard key must satisfy:

• High Cardinality: The key must have a large range of unique values. For example, country_code is a poor shard key (only ~200 values), causing data skew when one country has 100x the users of another. user_id or uuid has millions of unique values.
• Even Load Distribution: The hash distribution of keys must disperse rows evenly to avoid overloading a single physical disk.
• Query Pattern Alignment: If 90% of your queries filter by tenant_id, then tenant_id should be your shard key. This allows the router to direct queries to a single shard, bypassing costly scatter-gather operations.

C. Hotspot Mitigation & Salting

Even with a high cardinality key, real-world workloads can create hotspots. Consider a social media app sharded by user_id. When a celebrity (e.g., with ID 9999) posts, millions of users read/write comments to that ID simultaneously, overloading Shard Node 3.

To mitigate this, you can apply Salting. You append a random prefix or suffix to the shard key for hot records:

This distributes the celebrity's data across 10 different shards. The downside is that reading a celebrity's profile now requires querying all 10 shards and merging results, trading read latency for write durability.

D. Resharding and Rebalancing

When your dataset grows beyond the capacity of your initial shards, you must scale out by adding new physical nodes and rebalancing the data. In simple modulo sharding (hash(key) % N), changing N to N + 1 invalidates the target shard index of almost all existing keys, requiring a migration of 90%+ of your data.

To prevent this, production systems use Consistent Hashing or a directory-based router to limit data movement. The rebalancing workflow must be executed online:

• Dual Writing: Start writing all new data updates to both the old and new shards.
• Backfilling: Copy the historical data from the old shard to the new shard in background batches.
• Validation: Run data integrity checks to ensure the backfilled data matches the source.
• Cutover: Change the routing configuration mapping to point to the new shard, and safely delete the migrated data from the old shard.

11. Production Examples

• Vitess (YouTube): Built to scale YouTube's massive MySQL databases. It acts as an SQL proxy proxying web clients, performing query parsing, and automatically managing sharding schema configurations.
• Instagram: Shards PostgreSQL databases by User ID. To avoid centralized ID generation bottlenecks, Instagram generates unique Snowflake-like primary IDs that embed the Shard ID and a epoch time timestamp inside a 64-bit integer, letting local shards generate keys independently.
• Pinterest: Scaled its MySQL architecture using client-side sharding. Users are sharded to specific database shards, and all pins created by a user are stored on the same shard to ensure profile pages can load in a single database read.

12. Advantages

• Horizontal Write Scalability: Unlike replication, adding shard nodes increases the cluster's aggregate write capacity.
• Massive Storage Capacity: Datasets can scale to petabytes since data is partitioned across multiple server disks.
• Fault Isolation (Blast Radius): If Shard 3 crashes, only 25% of users (those mapped to Shard 3) are affected. The remaining 75% of the platform functions normally.
• Resource Optimization: CPU, RAM, and Disk I/O limits are scaled concurrently, preventing memory pressure on a single database process.

13. Limitations

• Loss of Referential Integrity: Modern relational databases cannot enforce foreign key constraints across different database engines over network boundaries.
• Scatter-Gather Latency: Queries that do not contain the shard key must query all shards in parallel, incurring network tail latency bottlenecks.
• Resharding Complexity: Moving live production data between shards without downtime requires complex pipeline tooling.
• No Cross-Shard Joins: Relational queries joining tables across separate physical instances are impossible or require custom in-memory join stitching at the application tier.

14. Trade-offs

• Range-Based Sharding vs. Hash-Based Sharding: Range sharding makes range queries (WHERE age BETWEEN 20 AND 30) fast since the data resides on the same shard, but it risks write hotspots (e.g., chronological inserts hitting only the highest range). Hash-based sharding guarantees uniform writes but forces range queries to execute as scatter-gather across all shards.
• Proxy-Based Routing vs. Client-Side Routing: A proxy simplifies client logic, maintains unified connection pooling, and allows dynamic re-routing without redeploying applications. However, the proxy introduces a network hop and can become a single point of failure. Client routing is faster and cheaper but requires updating config code on all microservices whenever shards change.

15. Performance Considerations

• Connection Pool Exhaustion: If 50 application servers each maintain a pool of 100 connections to 10 shards, the databases must manage 5,000 active connections, which degrades database performance. Use connection pooling proxies (like PgBouncer or Vitess).
• Tail Latency: If a scatter-gather query hits 10 shards, its final response latency is determined by the slowest shard. If one shard is experiencing a CPU spike, the entire request suffers.
• Data Skew: If a shard key maps too many rows to a single shard, that instance will hit its physical limits while other shards sit idle.

16. Failure Scenarios

• Single Shard Outage: If a physical shard node fails, the portion of users mapped to that shard cannot access the app.
  Mitigation: Configure each shard as a replica group (Active Primary with Active-Standby Replicas) so that a failover occurs automatically.
• Metadata Split-Brain: If the configuration registry (ZooKeeper) has a partition and reports stale mapping coordinates, the proxy will route writes to the wrong physical node, causing data corruption.
  Mitigation: Require strict consensus (e.g. Raft protocol) for all configuration database changes.
• Partial Batch Failure: If you write a batch of 10 records and they map to 3 different shards, the write to Shard 1 and 2 might succeed while the write to Shard 3 fails due to a timeout.
  Mitigation: Implement application-level rollbacks or compensate with Saga queues.

17. Best Practices

• Choose an immutable shard key with high cardinality.
• Design application schemas to be tenant-isolated or user-isolated, guaranteeing that typical transactions can execute entirely within a single shard.
• Pre-shard your database: start with a large number of logical shards (e.g., 256 logical databases mapped to 4 physical servers) so you can easily migrate logical databases to new physical servers as you scale.
• Maintain comprehensive dashboards monitoring disk space, CPU load, and IOPS skew across all shards.

18. Common Mistakes

• Sharding Too Early: Sharding introduces huge operational overhead. Do not shard until you have exhausted optimization avenues: read replicas, query optimization, indexing, and vertical scaling.
• Using Auto-Incrementing IDs as Shard Keys: Inserting sequential IDs (1, 2, 3, 4...) into range-based or modulo sharding will target the same partition repeatedly, creating a severe write hotspot.
• Ignoring Cross-Shard Queries in API Paths: Building a dashboard landing page that performs scatter-gather queries on every API request will eventually crash under user load.

19. Implementation (Sharding Router)

Below is a complete, production-grade Sharding Router Proxy simulation. It demonstrates range-based, hash-based, and directory-based sharding policies, managing writes, single-key reads, and a scatter-gather scan across mock database connections.

20. Interview Questions & Answers

Q1. What is the celebrity or hotkey problem in sharding, and how do you handle it?

Answer: The celebrity problem occurs when a high cardinality shard key has highly unbalanced activity. For example, in a database sharded by username, a request regarding a celebrity with millions of followers receives millions of reads or updates, completely overwhelming the single shard housing that celebrity's record.

To handle this:

• Salting: Append a random tag to the end of the key (e.g. celebrity_user_1, celebrity_user_2) to distribute the write/read load over multiple shards.
• Caching: Cache reads of extremely hot data in an in-memory layer (like Redis or Memcached) to intercept read spikes before they reach the physical databases.

Q2. How does sharding affect database transactions and foreign key constraints?

Answer: Standard databases can only enforce constraints (like FOREIGN KEY) and guarantee transaction atomicity (ACID) within a single engine process. Once you shard data across separate physical nodes:

• Foreign Keys: Cross-shard referential integrity checks are not supported by the database engine. Referential safety must be handled at the application logic layer or through denormalization.
• Transactions: Standard commits fail. You must implement a distributed transaction protocol like Two-Phase Commit (2PC) or use Saga workflows (eventual consistency with compensating rollbacks), which introduce substantial latency and architectural complexity.

Q3. What is the difference between sharding and partitioning in databases like PostgreSQL or MySQL?

Answer:

• Partitioning (Table Partitioning): Dividing a single logical table into sub-tables *on the same physical database engine instance*. The filesystem writes different chunks of the table to different files, but they are still managed by a single database server process.
• Sharding: Distributing the partitioned tables *across separate database instances running on different physical hardware servers*. Sharding implies a multi-machine, distributed shared-nothing system architecture.

21. Practice Exercises

• Exercise 1 (Easy): Calculate which shard index the keys id_10, id_99, and id_1000 will route to under a simple modulo strategy: hash(id) % 3. Assume hash(x) returns its numeric value.
  Answer:
  • 10 % 3 = 1 -> Shard 1
  • 99 % 3 = 0 -> Shard 0
  • 1000 % 3 = 1 -> Shard 1
• Exercise 2 (Medium): Write a design outline showing how to perform a multi-shard order-by query (SELECT * FROM users ORDER BY age DESC LIMIT 10) at the proxy coordinator level without pulling all users from all databases.
  Answer: The proxy should send SELECT * FROM users ORDER BY age DESC LIMIT 10 to all shards in parallel. Each shard returns only its local Top 10 oldest users (at most 10 rows per shard). The proxy collects these results (N shards * 10 rows), merges the lists in-memory, sorts them, and returns the absolute Top 10 to the user. This avoids loading the entire dataset into memory.
• Exercise 3 (Hard): Implement a pseudo-code routine in Python simulating a background "rebalancer thread" that reads keys from a source shard and moves them to a destination shard, handling the case where users are actively writing to the source key during the transfer. (Hint: use locking or shadow dual-writes).

22. Challenge Problem

The Zero-Downtime Migration Problem: You have a database sharded into 4 physical instances using range sharding. Due to a viral campaign, Shard 2 is running out of disk space. You need to split Shard 2's key range in half, migrating the upper half of its records to a brand new physical instance, Shard 5, without taking the system offline or rejecting user writes.

Draft a comprehensive step-by-step operational script outlining:

• How the sharding proxy rules must be updated dynamically.
• How new writes targeting the moving range are routed during backfill.
• The locking mechanism (or sequence number check) used to prevent overwriting new writes with older backfilled rows.
• The validation routine that confirms Shard 5 is identical to Shard 2's upper range before deleting data from Shard 2.

23. Summary

Sharding is a horizontal partitioning technique that splits a single logical database across multiple physical servers. It is the ultimate tool for scaling write-intensive systems and large-scale datasets. However, it trades off system simplicity, making queries without shard keys slow, cross-shard transactions heavy, and database rebalancing a complex operations challenge.

24. Cheat Sheet

Strategy	How it Works	Pros	Cons
Range-Based	Maps key ranges (e.g. A-M) to dedicated shards.	Allows fast local range queries.	High risk of hotspots (e.g. chronological keys).
Hash-Based	Applies modulo math on the hash of the shard key.	Uniform data and query distribution.	Requires scatter-gather for range scans.
Directory-Based	Queries a central lookup index table to locate shard ID.	Highly flexible; easy to rebalance individual keys.	Lookup table becomes a network bottleneck and SPOF.

25. Quiz

26. Further Reading

• Designing Data-Intensive Applications (Chapter 6: Partitioning) — Martin Kleppmann.
• Vitess Sharding Guide: Deep dive into YouTube's horizontal scaling proxy layer.
• Instagram Engineering Blog: Sharding and Snowflake ID generation at scale.

27. Next Lesson Preview

Now that we understand the trade-offs of sharding, we see that resizing a modulo-sharded database causes huge data migration overhead. In the next lesson, we will explore Consistent Hashing—the mathematical solution that lets us add or remove storage nodes while moving a minimum amount of data.