16.1 Sharding, Replica Groups, and Consensus
This lesson follows a common auto-sharding design: divide the key space into units, replicate each unit, and coordinate access through routing and consensus. A distributed database can also replicate without sharding; this design is one implementation family.
First, determine why you need distributed systems
Common goals fall into four categories, each requiring different approaches:
- Capacity: The single machine's storage, memory, or I/O is no longer sufficient;
- Throughput: You can scale horizontally by processing disjoint keys in parallel;
- Availability: The system remains operational even after node or failure domain failures;
- Geographic latency: Data is placed closer to primary users or compute workloads.
For high availability alone, a single master with synchronous or asynchronous replicas may be simpler than an auto-sharding database. For pure capacity scaling, an object store paired with a compute cluster might be more appropriate. Distributed SQL becomes useful when cross-shard transactions, relational semantics, and automatic rebalancing are needed together, and the team can operate the resulting system.
Avoid citing fixed thresholds like "a single machine can only handle a few TB." Capacity limits depend on hardware, data access patterns, index amplification, maintenance windows, recovery time, and workload characteristics. The actual trigger for sharding should come from observed capacity trends and service level objectives (SLOs), not arbitrary benchmarks.
Shard Functions Determine Locality and Hotspots
Range Sharding
The key space is divided into contiguous intervals:
[a, g) -> shard 1
[g, n) -> shard 2
[n, z) -> shard 3Range queries only access adjacent shards, making data compression and sequential scans more efficient. However, monotonically increasing time or ID values can concentrate recent writes toward the last range. The system can automatically split and migrate ranges, but it cannot eliminate hotspots at individual logical keys.
Hash Sharding
After hashing the partition key, the result is mapped to a bucket, typically distributing random key writes:
bucket = hash(partition_key) mod bucket_countThis approach sacrifices locality among adjacent keys. Directly taking the modulus of the physical node count causes a large number of keys to be remapped during scaling up or down. Production systems usually introduce fixed logical buckets, range/slot mappings, or a consistent hashing layer to map logical shards onto nodes.
Composite Keys
In practice, combinations of tenant ID, time bucket, and entity ID are commonly used:
tenant_id / month / event_idA tenant field groups related data; access control must still enforce authorization. Month limits the time span, not the data volume, and event IDs need uniqueness within their chosen scope. Still, large tenants require further dispersion, while small tenants suffer from excessive fan-out. The shard key is a long-term data contract, not a temporary choice of a high-cardinality field picked at deployment time.
Sharding and Replication Are Two Distinct Axes
Sharding answers the question "which portion of data is managed by whom," while replication answers "how many copies exist and how are they kept in sync." A common architecture looks like this:
range A -> replicas A1, A2, A3 -> one Raft group
range B -> replicas B1, B2, B3 -> another Raft groupEach range typically has its own leader/leaseholder and consensus log. When a client connects to any gateway, it does not mean that any replica can unconditionally return the latest read; the gateway may forward the request to the current leader. A follower may serve historical reads locally when conditions such as a safe closed timestamp and sufficient application progress are met.
Raft Solves Log Replication Ordering, Not All Database Problems
Raft appends client commands to the replicated log. A leader can advance the commit index once an entry from its current term is persisted by a majority of voting members, thereby committing preceding entries too. Merely counting copies of an old-term entry is insufficient. State machines apply committed commands in order. Three voters need two votes; five need three. Non-voting replicas do not add votes. Terms, election restrictions, and log matching protect the committed prefix.
It primarily addresses consensus within a single Raft group and does not automatically provide:
- Atomic transactions across multiple Raft groups;
- SQL isolation levels and predicate conflict detection;
- Shard key selection, hot-spot splitting, and global secondary indexing;
- Backup, accidental deletion recovery, or cross-version upgrades;
- The low-level hardware persistence details that guarantee "no data loss when written to a majority."
Databases must build upon Raft with additional mechanisms such as transaction records, locks/intents, MVCC timestamps, and recovery protocols.
Leader, Lease, and Read Paths
Writes typically pass through the current leader and wait for a majority of replicas to acknowledge. A linearizable read needs current read authority and a state machine applied through the relevant committed position. ReadIndex or a lease protocol with valid clock assumptions can supply those checks. Remembering past leadership is insufficient: if a read must include log position 105 but local application has reached only 102, returning the state at 102 is unsafe.
Follower reads reduce cross-region read latency, but they must be explicitly time-bound:
- Linearizable latest reads may require cross-region lookup to find the leader;
- Historical reads need a safe timestamp and sufficient local application progress; bounded staleness additionally requires satisfying the requested freshness lower bound;
- Ordinary asynchronous replica reads without bounds can only be described as "potentially stale."
"Read-write separation" is not a free scalability toggle. Business logic must clearly define which data pages can be stale, for how long, and where to read from after a write.
The Hidden Costs of Rebalancing
After a node joins the system, it must replicate range snapshots, catch up on logs, transfer leader or lease responsibilities, and update routing metadata. Rebalancing consumes network bandwidth, disk I/O, and CPU resources. If multiple replicas migrate simultaneously, it can even reduce fault tolerance.
Operations staff should at least monitor:
- Range size and QPS skew;
- Distribution of leaders and leases;
- Under-replicated ranges and log lag;
- Traffic from splits, merges, snapshots, and rebalances;
- Single-key hotspots and cross-shard fan-out;
- Which ranges become unavailable when a majority of replicas are lost.
References
- Raft: In Search of an Understandable Consensus Algorithm
- CockroachDB: Architecture Overview
- CockroachDB: Distribution Layer
The next section addresses a more challenging problem than replication: how to ensure both atomicity and serializability when a transaction spans multiple shards.