system-design · intermediate

Database Sharding — Split Data Across Many Machines

The Central Question

Consider an online SaaS platform running on the DataLab platform (datalab.com) processing 100,000,000 queries per day across 10,000,000 active users and 500,000,000 orders, growing by 5 TB of new transactional data every month:

• The engineering team has already exhausted all standard single-node database scaling options:

When a single Primary database node reaches its physical hardware ceiling for writes and storage, the database must be partitioned horizontally.

Database Sharding (Horizontal Partitioning) is the architectural pattern of splitting a large database table into smaller, independent chunks called Shards, with each shard residing on an independent database server instance holding its own dedicated CPU, RAM, and storage.

This lesson answers one central question: How does horizontal database sharding partition massive datasets across independent database nodes using shard keys, and how do engineers design hashing/range routing algorithms while managing cross-shard query penalties and resharding re-balance operations?

Topology: Vertical Scaling Ceiling vs. Horizontal Sharding

Understanding when to transition from vertical scale-up to horizontal sharding defines database infrastructure growth:

flowchart TB
  subgraph Single Primary Infrastructure (Scale-Up Ceiling)
    App1[Application Tier] -->|100% of Write Traffic| SingleDB[(Single Primary Node<br/>128 vCPUs / 1 TB RAM)]
    SingleDB --> R1[(Read Replica 1)]
    SingleDB --> R2[(Read Replica 2)]
    Note1["Write Bottleneck! Single Primary cannot handle write QPS."]
  end
  subgraph Sharded Infrastructure (Scale-Out Architecture)
    App2[Application Tier] --> Router[Database Proxy / Router]
    Router -->|user_id: 1-1,000,000| S1[(Shard Server 1)]
    Router -->|user_id: 1,000,001-2,000,000| S2[(Shard Server 2)]
    Router -->|user_id: 2,000,001-3,000,000| S3[(Shard Server 3)]
    Note2["Each Shard is an independent Primary handling a subset of writes."]
  end

Figure 1: Structural comparison between a saturated single Primary node and a horizontally sharded database cluster.

The Shard Key: Choosing the Partitioning Attribute

The most critical architectural decision in database sharding is selecting the Shard Key (Partition Key).

The Shard Key is a specific column (such as user_id, tenant_id, or account_id) present in database tables used by the database router to compute which shard server stores a given row:

flowchart TD
  Req[Incoming Write: INSERT INTO orders user_id = 89041] --> Router{Evaluate Shard Key: user_id}
  Router -->|Hash Partitioning: hash 89041 % 3 = Shard 2| S2[(Shard Node 2)]
  S2 --> Write[Execute Local SQL Insert]

Figure 2: Routing decision flowchart mapping a shard key value to a target database node.

Partitioning Strategies: Hash vs. Range Partitioning

Directory-Based Sharding (Lookup Mapping Table)

Virtual Nodes in Hash Sharding Rings

The Cross-Shard Query Penalty: Scatter-Gather Fan-Out

When an application issues a query that includes the Shard Key (WHERE user_id = 89041), the router forwards the query directly to a Single Shard ($O(1)$ routing latency).

However, if a query omits the Shard Key (WHERE status = 'PENDING'), the router must execute a Scatter-Gather Query:

flowchart TD
  Router[Database Shard Router] -->|1. Parallel Scatter| S1[(Shard Server 1)]
  Router -->|1. Parallel Scatter| S2[(Shard Server 2)]
  Router -->|1. Parallel Scatter| S3[(Shard Server 3)]
  
  S1 -->|2. Return Partial Rows| Merge[Router Aggregation & Sort]
  S2 -->|2. Return Partial Rows| Merge
  S3 -->|2. Return Partial Rows| Merge
  
  Merge --> Final[Return Final Merged Result to Client]

Figure 3: Scatter-Gather fan-out pattern querying all shards in parallel.

Scatter-Gather Cost Formula

$$L_{\text{scatter}} = \max(T_{\text{shard1}}, T_{\text{shard2}}, \dots, T_{\text{shardS}}) + T_{\text{merge}}$$

Scatter-gather queries suffer from the Slowest Shard Bottleneck: a single degraded or slow shard node delays the entire user query response.

Global Secondary Indexes (GSI)

• A separate lookup mapping table (stored in Redis or a dedicated index shard) maintains email -> user_id.
• When an application queries WHERE email = 'user@example.com', it first performs an $O(1)$ lookup in the GSI to retrieve user_id = 89041.
• The application then queries the primary sharded database with WHERE user_id = 89041, routing directly to a single target shard node in $O(1)$ time!

Distributed Transactions Across Shards: Two-Phase Commit (2PC)

1. Prepare Phase: The Transaction Coordinator sends a PREPARE request to both Shard 1 and Shard 2. Each shard acquires local locks, writes the WAL log, and responds VOTE_COMMIT.
2. Commit Phase: If both shards vote YES, the Coordinator sends COMMIT to both shards. If any shard votes NO (or times out), the Coordinator sends ABORT to rollback both shards.

Complete Worked Example: Go Shard Router (Hash & Range Partitioning)

Let's inspect a complete Go implementation of a Shard Router supporting Hash and Range partitioning for the DataLab platform (datalab.com).

package main
import (
	"context"
	"fmt"
	"hash/fnv"
	"sync"
)
type ShardNode struct {
	ID        string
	ConnString string
}
type ShardRouter struct {
	mu     sync.RWMutex
	shards map[int]*ShardNode
}
func NewShardRouter(shards map[int]ShardNode) ShardRouter {
	return &ShardRouter{shards: shards}
}
func (r ShardRouter) HashRoute(shardKey string) ShardNode {
	r.mu.RLock()
	defer r.mu.RUnlock()
hasher := fnv.New32a()
	hasher.Write([]byte(shardKey))
	hashVal := hasher.Sum32()
shardIdx := int(hashVal) % len(r.shards)
	fmt.Printf("[HASH ROUTER] Key '%s' (Hash: %d) -> Assigned to Shard Node %d (%s)\n",
		shardKey, hashVal, shardIdx, r.shards[shardIdx].ID)
return r.shards[shardIdx]
}
func (r ShardRouter) RangeRoute(userID int64) ShardNode {
	r.mu.RLock()
	defer r.mu.RUnlock()
var shardIdx int
	switch {
	case userID <= 1000000:
		shardIdx = 0
	case userID <= 2000000:
		shardIdx = 1
	default:
		shardIdx = 2
	}
fmt.Printf("[RANGE ROUTER] User ID %d -> Assigned to Shard Node %d (%s)\n",
		userID, shardIdx, r.shards[shardIdx].ID)
return r.shards[shardIdx]
}

Failure Modes and Engineering Mitigations

What You Should Remember

1. Sharding provides horizontal write scale: Partition rows across independent database servers to scale write throughput and storage infinitely.
2. Select high-cardinality shard keys: Use unique keys like user_id or uuid to distribute reads and writes uniformly across shard nodes.
3. Avoid Scatter-Gather queries: Ensure critical high-frequency queries include the Shard Key to route to a single target node ($O(1)$ routing).
4. Use Consistent Hashing or Hash Slots: Eliminate 100% data reshuffling when adding shards by using fixed virtual slot assignments.
5. Avoid Cross-Shard Transactions (2PC): Design application boundaries to keep related transaction records inside the same shard or use Sagas.

Glossary of Terms

Practice Scenario and Self-Assessment

Architecture Scenario

• 5,000 enterprise tenant accounts (tenant_id).
• Tenant 1 produces 40% of overall platform write volume (Hot Tenant).

1. Evaluate whether to shard by tenant_id or user_id.
2. Design the Hot Tenant isolation architecture to prevent Tenant 1 from overwhelming shared shards.

Interactive Self-Assessment

It forces a Scatter-Gather query across all shards in parallel, binding total latency to the slowest shard node.

It automatically formats persistent NVMe SSD disk drives across all shard nodes.

It revokes client HTTPS TLS encryption certificates on edge load balancers.

It doubles the physical hardware clock speed of primary database CPUs.

Hash Partitioning distributes data and write traffic uniformly across all nodes, preventing hot-shard bottlenecks.

Hash Partitioning converts relational SQL schemas into un-indexed CSV files.

Hash Partitioning replaces public DNS nameservers with local hosts entries.

Hash Partitioning reboots operating system hypervisors across all shard nodes.

What to Learn Next

• Data Replication — Single-Leader, Multi-Leader, and Leaderless: Master replication topologies and consensus.
• Rebalancing Shards Under Skewed Traffic: Learn live data migration and range splitting.
• Hot Partition / Hot Key — Mitigating Skewed Traffic: Explore salted keys and local micro-caching.

Track: Data, Storage and Messaging

Previous: Database Indexes — Find Rows Without Scanning Everything

Next: Hot Partition / Hot Key — When One Shard Takes All the Heat

By Shubham Jain

All articles · Study paths

Shubham Jain · Learning Lab