Sharding

Sharding divides a dataset across storage nodes so that each shard owns a different portion. It can spread storage capacity, working sets and write traffic beyond one machine. The application or database router must then determine where each operation belongs. A transaction that was local may become a conversation between machines.

Sharding and Replication solve different problems: shards hold different portions; replicas hold copies. A production shard can itself have replicas. Adding shards does not by itself protect a shard's data from machine failure.

The downloadable lab needs Python 3.11+ with standard-library SQLite. Extract it and run from its directory:

python3 lab.py
python3 -m unittest discover -s . -p 'test_*.py' -v
# Seven tests, including child-process crashes and cross-tenant rejection.

The lab models logical shards inside one persistent SQLite database. It tests routing, checkpoint recovery and ownership gates, using SQLite's local transaction boundary. It does not measure network behavior, independent-machine availability or production database migration performance.

Start with the queries

Consider tenant activity events. Most requests append an event or retrieve one tenant's recent activity. A smaller administrative workload lists recent activity across tenants. Choosing tenant_id as the partition key keeps the common operations on one shard. Within that shard, an index on (tenant_id, event_id) supports ordered retrieval. The partition key selects the destination; the index selects rows there.

Evaluate candidate keys against request frequency, bytes stored and query shape. Counting distinct keys is insufficient. A key with many values still performs poorly if one value receives most traffic. A mutable key also turns an ordinary field update into data movement.

The fixture creates 10,000 requests: 9,000 belong to one hot tenant and 1,000 belong to different cold tenants. Compare routing whole tenants with routing sixteen event buckets per tenant:

from collections import Counter
from lab import rendezvous

nodes = ("a", "b", "c", "d")
requests = [("hot", n) for n in range(9000)]
requests += [(f"cold-{n}", n) for n in range(1000)]
tenant_load = Counter(rendezvous(t, nodes) for t, event in requests)
bucket_load = Counter(rendezvous(f"{t}:{event % 16}", nodes)
                      for t, event in requests)
assert max(tenant_load.values()) == 9258
assert max(bucket_load.values()) == 3082

These are request counts for fixed inputs, not throughput measurements. Hashing a tenant ID cannot split that tenant's traffic. A directory can isolate the tenant on a larger dedicated shard, but its capacity remains finite. Bucketing splits its writes while making a complete tenant read contact multiple buckets, here spread across four shards. Tenant-wide transactions become harder too. The workload determines whether that exchange is worthwhile.

Routing changes when membership changes

Range routing assigns intervals such as tenant IDs [0, 1000) to a shard. Related ranges can be scanned together, but ever-increasing timestamps concentrate new writes at the current range's edge. Splitting old ranges may free storage without distributing current traffic.

Hash routing transforms the key before placement. It removes natural ordering, so adjacent business values need not share a destination. A time-range query without the hash key may therefore contact every shard.

Directory routing looks up a tenant or virtual partition in a maintained map. It permits explicit placement and movement, at the cost of maintaining available, consistent routing metadata and refreshing stale clients. A common combination hashes keys into many stable virtual partitions, then maps those partitions onto physical shards.

The simplest hash rule, hash(key) % shard_count, changes many assignments when the count changes. Rendezvous hashing instead scores each key/node pair and chooses the highest score:

from lab import digest, modulo

def rendezvous(key, nodes):
    return max(nodes, key=lambda node:
               (digest(f"{len(key)}:{key}{node}"), node))

old, new = ("a", "b", "c"), ("a", "b", "c", "d")
keys = [f"tenant-{n}" for n in range(10_000)]
assert sum(modulo(k, old) != modulo(k, new) for k in keys) == 7541
assert sum(rendezvous(k, old) != rendezvous(k, new) for k in keys) == 2559
assert all(rendezvous(k, old) == rendezvous(k, new)
           or rendezvous(k, new) == "d" for k in keys)

digest uses SHA-256, avoiding Python's process-dependent built-in string hash. With existing scores unchanged, adding d can only move keys that d wins. Removing a node similarly affects its winners. This implementation scores every node per lookup; larger systems need to account for that routing work. Membership lists must agree, and changing the hash encoding changes placement.

Neither algorithm copies records or resolves in-flight writes. Fewer moved keys also does not imply proportionally fewer moved bytes or requests: one moved tenant may dominate both.

Moving data means moving write ownership

A migration needs a source of truth throughout the move. Copying rows while accepting unrestricted source writes can miss updates or resurrect deleted records. A more available protocol can snapshot, capture and replay changes, then briefly fence writes for final catch-up. That requires a defined change-log position and handling for updates, deletes and retries.

The lab chooses a simpler contract: pause writes for the migrating tenant while reads continue from its source. Other tenants remain active between batch transactions. Inside a transaction, begin_move freezes the tenant and records its largest event ID as highwater. IDs are nonnegative; checkpoint -1 means nothing copied. Each batch uses ordered keyset pagination:

SELECT id, payload FROM events
WHERE shard = :source AND tenant = :tenant
  AND id > :checkpoint AND id <= :highwater
ORDER BY id LIMIT :batch_size;

Destination upserts and checkpoint advancement commit together. Restart repeats the last uncommitted batch safely. The test actually terminates a child process at two boundaries:

# Inside the fixture's test case: eight source events already exist.
self.store.begin_move("acme", "b")
self.child_crash("before-checkpoint", 23)
assert self.store.info("acme")["checkpoint"] == -1  # Copy rolled back too.
self.child_crash("after-commit", 24)
assert self.store.info("acme")["checkpoint"] == 2
assert self.store.batch("acme") == 3
assert self.store.batch("acme") == 2
assert self.store.batch("acme") == 0
self.store.cutover("acme")
assert self.store.route("acme") == ("b", 2)

Before cutover, the fixture compares IDs and payloads in both directions; matching counts alone would accept corrupted contents. It rejects premature cutover. The directory then changes owner and increments the ownership epoch atomically. Every service write checks the current owner, epoch and migration state inside its mutation transaction. A paused write raises Frozen; an old ("a", 1) route after cutover raises StaleRoute. Refreshing routing is separate from retrying an operation safely, as discussed in Queues and Retries.

The batch limit bounds rows per transaction, not bytes or total pause duration. One large tenant may take too long to freeze. SQLite serializes these local mutations; it supplies no distributed ownership authority. Independent shards need a durable control protocol and enforcement at every writer, including background jobs. The lab exposes inserts only; updates and deletes would need the same gate.

Source rows remain after cutover for inspection. Reads select the directory's current owner so retained copies do not count twice. Production cleanup needs its own safe checkpoint. Once destination writes begin, simply flipping the map backward would lose those writes.

Local guarantees stop at the shard

Placement never establishes permission. Two tenants can share a shard, and a caller can guess another tenant's valid routing key. The service must derive grants from authenticated identity, authorize the requested tenant, and constrain its query by tenant as well as object ID. The lab assumes verified identity grants as input:

# After creating beta on the same shard as acme:
beta_grant = frozenset({"beta"})
store.write(beta_grant, "beta", ("a", 1), 0, "beta private")
assert store.read(beta_grant, "beta") == [(0, "beta private")]
store.read(frozenset({"acme"}), "beta")  # Raises PermissionError.
# A cross-tenant write is denied too, despite using the correct shard and epoch.

The administrative global query has a separate access check. For a global top k, it asks each owning shard for its local top k, merges candidates and keeps k. With S shards, this means S shard queries and at most S × k candidate rows. The fixture orders ties by tenant ID consistently. Its test injects one unavailable shard and requires an incomplete-result error; silently returning the remaining rows would claim a global answer without global evidence. Independently timed shard reads also need an explicit consistency contract if a common snapshot matters.

A local uniqueness constraint has the same boundary. The fixture accepts alias shared-name once on shard a and once on shard b, while rejecting a second copy on b. Each shard obeys its constraint; global uniqueness fails. Tenant-scoped names avoid that requirement. Globally unique names instead need an authoritative reservation path or suitable cross-shard coordination. A cross-shard transfer similarly needs an atomic protocol or an explicit pending/compensation workflow; two successful local transactions are insufficient.

For reporting, an asynchronously maintained analytical view can avoid repeatedly querying every serving shard, while introducing measurable freshness lag. Choose that contract deliberately. Sharding moves a capacity boundary; the useful design keeps frequent queries and invariants local and makes the remaining coordination visible.