Sharding & Partitioning Strategies

Learning goal: Choose the right partitioning strategy for a given workload, avoid the hot-partition trap, and plan a live migration with explicit safety checks and a bounded cutover pause.

Why this session exists

Imagine a service where 5% of users generate 60% of traffic. Splitting rows by user ID may balance row counts while leaving one shard overloaded. Changing the placement rule then touches routing, backfills, concurrent writes and query plans—not just a modulo expression. This session gives you the vocabulary and the mental model to make the right choice — or at least to recognise when you're about to make the wrong one.

You will be able to

• Name the three canonical partitioning strategies (hash, range, directory) and pick the right one for a given workload.

• Explain consistent hashing and why it lets you add a shard without moving all the data.

• Diagnose a 'hot partition' from monitoring graphs and propose three concrete fixes.

• Plan snapshot/change capture, backfill, validation and fenced cutover; explain why uncoordinated dual writes are unsafe.

• Predict targeted versus scatter/gather queries and product-specific transaction/index restrictions.

Prerequisites

• S071 · Replication — sharding is orthogonal to replication; you almost always do both.
• S072 · Consistency models — cross-shard operations force you to make consistency trade-offs.
• S041 · Indexes — routing selects a shard; an index finds records inside it. These are distinct jobs.

Preparation and pacing

Follow intuition → placement diagrams → offline ring and migration checks → query/failure reasoning → recall. The original 90-minute session, 30-minute preparation and 5/15/25/15/10-minute phase labels are suggested allocations, not measured reading or execution times. The 130-session learning route needs periodic source review rather than an “evergreen” guarantee.

Optional original discovery pointers

These preserve exact original URLs and their intended topics. Videos were not watched; historical titles and creator labels are finding aids with unverified content, duration and availability. Prior praise and numeric production-scale claims are not adopted.

• “Database Sharding Explained” — formerly labelled ByteByteGo: optional sharding introduction. The original also reused this same video URL as a “Discord Engineering blog” reading card for hot-partition diagnosis; it is not a Discord blog URL, and that attribution is not retained as fact.
• “Consistent Hashing” — formerly labelled ByteByteGo: optional ring/movement visualization, not a measured resharding demonstration.
• Database-sharding search for Hussein Nasser: the original labelled this Gaurav Sen, but the URL is a search for Hussein Nasser. It identifies neither a particular speaker nor recording; use it only to discover trade-off explanations.
• MongoDB sharding overview: optional shard-key, chunk and balancer orientation. The /manual/ route is mutable; the 8.0 pages below govern the versioned claims accepted here.
• LinkedIn Kafka ecosystem article: historical discovery pointer for partitioning, skew and rebalancing, not a generic sharding tutorial. Its original “4+ trillion messages/day” annotation was not verified and is not a benchmark used here.

(a) Intuition · 5 min

Splitting a library across three buildings

Your city library has grown too big for one building. You need to split its 3 million books across three buildings. Three natural strategies:

By author's last name: A–H → Building 1, I–P → Building 2, Q–Z → Building 3. Easy to know where a book is, but Building 2 gets crushed if half this hypothetical collection has author names starting with M.

By a hash of the ISBN: Each book is randomly assigned. Book counts tend to balance with a suitable hash and enough keys, but "give me every book by Rowling" requires visiting all three.

By an explicit lookup table: "The DBA decided these books go here, those books go there." Flexible but somebody has to maintain the table.

These are the three canonical sharding strategies. Range (by-author), hash (by-ISBN), directory (by-lookup-table). The choice depends on access patterns, skew and migration requirements.

The bad news: you can't easily switch. Once you've sharded by ISBN, your "all books by Rowling" queries require fan-out unless you introduce an appropriate index, projection, routing scheme or migration.

The three sharding strategies

• Hash: compute a placement from the shard key. Modulo-N is simple; a ring or fixed virtual-partition map can reduce movement when membership changes. A good hash spreads distinct keys, not necessarily bytes or traffic.
• Range: assign contiguous key intervals. A range query can prune irrelevant shards, but a large range may cross many of them. Monotonically increasing keys can concentrate new writes.
• Directory: maintain an explicit tenant/key/range-to-owner map. Cache it with an epoch, and have servers reject or redirect stale ownership rather than trusting TTL alone.
• Composite: combine dimensions such as tenant and bucket, with timestamp sorting inside each bucket. This spreads one tenant's writes only if the finer-grained keys actually distribute its traffic.

Changing a shard key is possible, but plan data movement, query changes and compatibility carefully. It need not move every row; changing hash modulus from 4 to 8, for example, moves half of uniformly distributed hashes if old shard IDs are retained. Moving one large tenant by directory may touch only that tenant.

A history of "we shard now"

The 2007 Dynamo paper describes consistent hashing and virtual nodes. Bigtable's official schema guidance demonstrates why row-key order still matters despite automatic tablet management. MongoDB 8.0 documents cardinality, frequency and monotonicity when choosing a shard key. Vitess's versioned migration workflow makes copy, replication, validation and traffic switching explicit. These are different designs, not evidence that every modern database uses a hash ring.

The three strategies side by side

Route the same user ID through three sharding strategies. Holding the key constant makes the design difference visible: compute a destination, find its range, or consult an explicit mapping.

The destination differs because each strategy stores routing knowledge in a different place. The next question is what happens to that knowledge—and to existing keys—when the cluster gains a node.

Consistent hashing — the trick that makes adding a shard cheap

Hold the existing node positions fixed and insert a new position between B and C. Follow which interval changes ownership instead of recomputing every key modulo a new shard count.

Node E takes only the interval that now reaches E before C; the other ownership intervals remain in place. This limits movement, but says nothing by itself about whether one key receives far more traffic than another.

The lifecycle of a shard

Begin by identifying the limiting resource: storage, IOPS, CPU, write contention or read fan-out. Tune indexes and queries before assuming the remedy is partitioning. Vertical scaling has provider- and workload-specific costs, not a universal 64-core ceiling. Read replicas can offload suitable reads but must still apply the write stream. Sharding distributes ownership and can distribute write work if keys and transactions permit it.

Before adding capacity, choose logical partitions separately from machines and make routing versioned. A managed service can automate splits and movement but does not remove hot-key limits, quota planning, query locality or transaction semantics. A celebrity's single serialized mutation cannot become parallel merely by adding machines.

Hot partitions — the failure mode you'll hit at scale

Where 'skew' comes from

Application layer

One tenant is 100x the size of others. One celebrity user has 10M followers. Trending event drives all traffic to one row.

Key choice

Sharding by timestamp: all writes today go to the same shard. Sharding by country: US shard has 40% of traffic. Sharding by low-cardinality key: only N possible values.

Hash function

A weak or biased hash can pile keys onto some shards. Use MurmurHash / xxHash / CityHash, not % alone on non-uniform inputs.

Range boundaries

Fixed ranges get lopsided as data grows. BigTable + CockroachDB auto-split ranges when they exceed a size threshold — but splitting a range does not split a single indivisible hot key.

Coordinator / router

If routing is centralised and one shard is slow, the coordinator gets backed up. Circuit-breakers and per-shard timeouts help.

When to use which strategy

Choose hash when point lookups dominate and there are enough similarly sized, similarly active keys. Choose range when prefix/range scans dominate and you can spread newest writes or tolerate their concentration. Choose directory when tenant placement, residency or isolation requires exceptions. A directory adds operational state and lookup cost; caching can avoid a network hop on many requests, but stale entries require server-side ownership validation.

A routed query is not automatically fast: it can still scan a huge partition or lack an index. A scatter/gather query is not automatically unusable: projections, pruning, parallelism and workload isolation can make it practical. Cross-shard uniqueness and transactions are engine-specific; a saga is not a drop-in atomic transaction.

Cross-shard transactions and secondary-index routing

Transfer 10 units from account A on shard 1 to B on shard 2. Two independent commits can debit A and then fail before crediting B. A two-phase-commit design asks participants to prepare and preserve the state needed to honor a durable commit/abort decision, then disseminates that decision. A prepared participant cannot safely guess “abort” merely because its coordinator timed out: it may need recovery of the decision to proceed. Replicating participants and coordinator state with consensus improves crash recovery under that protocol; it does not make every partition available. Isolation still needs locking, timestamps or another concurrency-control rule. A saga that later refunds A is a different, externally observable intermediate-state contract.

An index also has an owner. Hashing by user_id routes a profile lookup; a local index on country accelerates each shard's search without telling the router which shards contain India users. A global country projection can supply candidate user IDs/locations, but asynchronously maintained entries can be stale or missing: validate returned rows, and recognize that validation alone cannot recover omitted matches. If a query needs a complete snapshot, coordinate projection freshness/snapshot scope or use an appropriate transactional index contract.

For MongoDB 8.0 ranged sharding, the documented unique-index choices are the shard-key index, a compound unique index with the shard key as prefix, and the default _id index with a crucial limitation: if _id is not the shard key, uniqueness is only enforced per shard. A unique (tenant_id, email) key enforces tenant-scoped pairs, not globally unique email. MongoDB does not support a unique constraint on a hashed index. These restrictions are product-specific, not a theorem that all distributed databases lack global uniqueness.

MongoDB 8.0 also documents that outside reads using snapshot, linearizable, or causal afterClusterTime conditions wait for the relevant multi-shard transaction visibility during commit, while other read concerns can read pre-transaction versions. This does not make linearizable a multi-document transaction isolation level or make two separate outside queries one atomic snapshot. Specify the reader's concern and operation scope as carefully as the writer's transaction. The same version documents transaction/chunk-migration lock interactions and possible transaction aborts; routing movement and transaction execution are not independent.

The mental model to hold

Count work, not just rows. With four shards each holding 25% of keys, a single key generating 60,000 writes/s still sends those writes to one shard. If that shard can apply 20,000/s, the queue grows by 40,000/s even while the other three sit idle. More virtual nodes do not split that key. Diagnose per-shard request rate, queue age, CPU, latency and bytes, then find heavy keys using sampled/top-k telemetry rather than unbounded per-user metric labels.

Three responses address different causes: (1) cache or coalesce repeated reads while defining freshness; (2) split independent writes into meaningful subkeys/buckets and pay fan-in/aggregation cost; (3) move a hot tenant to dedicated or larger capacity through directory routing. Rate limits/backpressure protect the system if semantic splitting is impossible. A larger shard buys finite headroom, not a proof of scale.

Why a ring reduces movement. For uniformly distributed integer hashes and retained shard IDs, moving from modulo N to modulo N+1 leaves about 1/(N+1) on the same ID; N/(N+1) move. Going 4→5 therefore moves about 80%, not literally every key. A consistent ring with balanced equal-capacity ownership instead adds the new node's share, expected near 1/(N+1)=20%. On a finite deterministic ring the actual share depends on its arcs, hash distribution and vnode count. Removing one of N balanced nodes moves about 1/N. Virtual nodes reduce random ownership variance, not traffic skew or every possible imbalance.

Fixed logical partitions provide another approach: route into, say, 1,024 buckets and map buckets onto machines. Moving a bucket changes ownership metadata and still requires copying its data. Do not confuse avoiding rehash with avoiding data transfer.

Fan-out tail calculation. If each of 20 independent shard requests has a 1% chance of exceeding a deadline, the chance at least one exceeds it is 1 - 0.99**20, about 18.2%. This is a synthetic independence model, not a measured p99; correlated failures can be worse. Query pruning, extra per-shard capacity and carefully budgeted hedges can change the tail. Adding shards is not universally unable to improve latency, but more fan-out dependencies deserve scrutiny.

Tenant trade-off. Co-location preserves tenant-local transactions and queries, but does not automatically isolate CPU or make backup/restore trivial. Fine-grained bucketing can spread a giant tenant while turning tenant-wide reads into merges. A hybrid directory can give large tenants dedicated capacity and small tenants shared shards, at the cost of reliable migration and ownership metadata. None is a universal default; list dominant queries, per-tenant peak rates, residency and growth before choosing.

Live resharding with a recoverable cutover

Do not independently write old and new destinations and hope both succeed. Use the old owner as authority while copying and capturing its committed changes, or a supported transactional migration protocol. A concrete sequence is:

1. Prepare: provision targets, version the routing map (epoch 7), record partition ownership and reserve source/target I/O headroom. Define acceptable pause and rollback limits.
2. Snapshot and stream: establish a coordinated snapshot/change position. Backfill historical rows while replaying later inserts, updates and deletes. A copied old value must not overwrite a newer streamed value; use positions or a tool that coordinates copy and apply.
3. Validate: compare keys/counts/checksums at a common position, test constraints and shadow reads, measure lag and capacity. Equal counts alone cannot detect a swapped or stale row.
4. Fence and switch: stop or briefly buffer old-owner writes, drain changes to cutover position 9,500, and verify targets through that position. Atomically publish epoch 8. Reject/redirect requests using epoch 7; TTL expiry alone cannot fence an old router.
5. Retain rollback capability: after targets accept new writes, the old copies become stale. Reverse replication or another documented recovery path is required before switching back. Keep old data until validation and rollback criteria are satisfied; then decommission deliberately.

Vitess 22's MoveTables workflow exposes copy/catch-up, VDiff validation, SwitchTraffic and ReverseTraffic phases. That supports the mechanism, not a promise that every migration has zero failed requests. Test each system's cutover, retry and transaction behavior.

Failure exercise: backfill copies balance=10 from position 100, then a stream update sets 11 at 101, then a delayed backfill write sets 10 again. What went wrong? Answer: copy and change ordering were not coordinated. Stage the snapshot before applying later positions or guard each apply against older versions. A delete at 102 also needs a tombstone/version marker until backfill and replay can no longer resurrect the row.

Tiny version and ownership model

A source sequence is a log position, not a wall-clock guess. This single-key model rejects stale routing epochs, retains tombstone versions and treats repeated identical positions idempotently. Each position must name one immutable mutation. Epoch validation represents an already-authoritative fence at the owner; it does not implement the distributed publication of that fence.

from dataclasses import dataclass
 
@dataclass
class MigratingKey:
    epoch: int = 8
    position: int = -1
    value: object = None  # None is a tombstone in this model
 
    def apply(self, epoch, position, value):
        if epoch != self.epoch:
            return 'stale-owner'
        if position < self.position:
            return 'older'
        if position == self.position:
            if value != self.value:
                raise ValueError('one log position cannot name two mutations')
            return 'duplicate'
        self.position, self.value = position, value
        return 'applied'
 
k = MigratingKey()
assert k.apply(8, 101, 11) == 'applied'
assert k.apply(8, 100, 10) == 'older'
assert k.apply(8, 102, None) == 'applied'
assert k.apply(8, 100, 10) == 'older'  # no resurrection by delayed backfill
assert k.apply(8, 102, None) == 'duplicate'
assert k.apply(7, 103, 12) == 'stale-owner'
assert (k.position, k.value) == (102, None)

After the target accepts position 103, the old source at 102 is no longer a safe rollback destination until reverse catch-up/validation. If the new owner loses its replica quorum, do not silently revive the stale old owner. Keep tombstones until no old copy/replay can resurrect their keys. This fixture models one key and an installed epoch only: it proves neither atomic multi-key cutover nor CDC completeness, routing-cache invalidation, replica failover or tombstone garbage collection.

(c) Hands-on · 25 min

Measure movement on 100,000 synthetic keys with a modulo router and a consistent ring. This is a deterministic example, not a proof of production traffic balance.

#!/usr/bin/env python3
"""consistent_hash.py — a working consistent hash ring in one file.
 
Run: python consistent_hash.py
 
Compares naive `hash % N` to consistent hashing with virtual nodes.
Shows: 'when you add a shard, how many keys move?'
"""
from __future__ import annotations
 
import bisect
import hashlib
from collections import Counter
 
 
def h(x: str) -> int:
    """Stable 128-bit hash truncated to 32 bits for readability."""
    return int(hashlib.md5(x.encode()).hexdigest(), 16) & 0xFFFF_FFFF
 
 
class NaiveHashRouter:
    """The `hash(key) % N` approach. Simple, terrible at resharding."""
    def __init__(self, nodes: list[str]):
        if not nodes or len(set(nodes)) != len(nodes):
            raise ValueError("nonempty unique nodes required")
        self.nodes = list(nodes)
 
    def route(self, key: str) -> str:
        return self.nodes[h(key) % len(self.nodes)]
 
    def add_node(self, node: str) -> None:
        if node in self.nodes:
            raise ValueError("duplicate node")
        self.nodes.append(node)
 
 
class ConsistentHashRouter:
    """Consistent hashing with virtual nodes for balance."""
    def __init__(self, nodes: list[str], virtual_per_node: int = 150):
        if not nodes or virtual_per_node < 1 or len(set(nodes)) != len(nodes):
            raise ValueError("nonempty unique nodes and positive vnode count required")
        self.vpn = virtual_per_node
        self.nodes = set()
        self.ring: list[tuple[int, str]] = []  # (hash, node)
        for n in nodes:
            self._add(n)
 
    def _add(self, node: str) -> None:
        if not node or node in self.nodes:
            raise ValueError("nonempty unique node required")
        self.nodes.add(node)
        for v in range(self.vpn):
            self.ring.append((h(f"{node}#{v}"), node))
        self.ring.sort()  # deterministic tie-break if 32-bit vnode hashes collide
 
    def route(self, key: str) -> str:
        if not self.ring:
            raise RuntimeError("empty ring")
        hv = h(key)
        # Find the first virtual node whose hash >= key's hash; wrap around.
        idx = bisect.bisect_left(self.ring, (hv, ""))
        if idx == len(self.ring):
            idx = 0
        return self.ring[idx][1]
 
    def add_node(self, node: str) -> None:
        self._add(node)
 
    def remove_node(self, node: str) -> None:
        self.nodes.discard(node)
        self.ring = [(hv, n) for hv, n in self.ring if n != node]
 
 
def demo(router_class, name: str, N_INITIAL: int, N_KEYS: int) -> None:
    print(f"\n=== {name} ===")
    nodes = [f"node-{i}" for i in range(N_INITIAL)]
    router = router_class(nodes) if router_class is NaiveHashRouter \
        else router_class(nodes, virtual_per_node=150)
    keys = [f"user-{i}" for i in range(N_KEYS)]
 
    # Distribution before
    before = Counter(router.route(k) for k in keys)
    print(f"Before ({N_INITIAL} nodes):")
    for n, c in sorted(before.items()):
        pct = 100 * c / N_KEYS
        bar = "#" * int(pct)
        print(f"  {n:<10} {c:>6}  {bar}")
 
    # Snapshot mapping
    old_map = {k: router.route(k) for k in keys}
 
    # Add a node
    router.add_node(f"node-{N_INITIAL}")
    new_map = {k: router.route(k) for k in keys}
    moved = sum(1 for k in keys if old_map[k] != new_map[k])
    pct_moved = 100 * moved / N_KEYS
 
    print(f"After adding one node ({N_INITIAL + 1} nodes total):")
    after = Counter(new_map.values())
    for n, c in sorted(after.items()):
        pct = 100 * c / N_KEYS
        bar = "#" * int(pct)
        print(f"  {n:<10} {c:>6}  {bar}")
    print(f"Keys moved: {moved}/{N_KEYS} = {pct_moved:.1f}%")
    expected = (N_INITIAL if router_class is NaiveHashRouter else 1) / (N_INITIAL + 1)
    print(f"Uniform-hash expected movement: {100 * expected:.1f}%")
 
 
def hot_partition_demo() -> None:
    """Simulate the 'celebrity user' effect."""
    print("\n=== Hot-partition simulation ===")
    router = ConsistentHashRouter([f"node-{i}" for i in range(4)])
    # 10 000 normal users, 1 celebrity
    events: list[str] = []
    for i in range(10_000):
        events.append(f"user-{i}")
    # Celebrity generates 100 000 events (all with same key)
    events.extend(["user-42"] * 100_000)
 
    load = Counter(router.route(k) for k in events)
    total = sum(load.values())
    print("Traffic per node (celebrity concentrates load):")
    for n, c in sorted(load.items()):
        pct = 100 * c / total
        bar = "#" * int(pct / 2)
        print(f"  {n:<10} {c:>7}  {pct:5.1f}%  {bar}")
    print("Fix: shard by (user_id, event_id % 16) so one hot user")
    print("     spreads across 16 sub-shards.")
 
 
if __name__ == "__main__":
    demo(NaiveHashRouter, "Naive hash % N", 4, 100_000)
    demo(ConsistentHashRouter, "Consistent hashing (150 vnodes/node)",
         4, 100_000)
    hot_partition_demo()

What each block does

h uses the low 32 bits of an MD5 digest as a stable demonstration hash, not a cryptographic security primitive or collision-free namespace. The ring sorts (position,node) tuples, making collisions deterministic. Production designs must specify hash width, collisions, membership and weighted capacity; the toy stores no data and performs no migration.

bisect_left finds the successor vnode with logarithmic lookup in the number of vnodes, wrapping to index 0 after the final position. No sub-microsecond claim follows from asymptotic complexity. demo reports actual movement; its ideal 20% is an expectation for balanced ownership, not an acceptance threshold. Try 1 and 150 vnodes and compare the observed distributions—there is no universal ±3% bound or mandatory vnode count.

In the executed 100,000-key fixture, modulo 4→5 moved 80,100 keys (80.1%) and the 150-vnode ring moved 19,202 (19.202%). These are local model observations, not distributed throughput or migration measurements.

hot_partition_demo assigns 100,000 extra events to one key, so that owner receives at least 100,000 of 110,000 total events (90.9%); this fixture placed 102,298 events (about 93.0%) on that node. To split independent events, replace the hot-key extension—not append a second copy—with:

# Context fragment inside hot_partition_demo; replace its hot-key extension.
events.extend(f"user-42#{i % 16}" for i in range(100_000))

Each bucket gets 6,250 extra events. Multiple buckets may land on one physical node, so four-node traffic will not be perfectly flat. Reads of the complete user history need up to 16 logical-bucket lookups plus an ordered/deduplicated merge. This is suitable for independently keyed events, not permission to split one atomic bank balance into unrelated writers.

(d) Production reality · 15 min

Query exercise: propose keys for profiles, nearby drivers, trips and ride history. Answer: hash(user_id) supports a point profile lookup. Nearby-driver search needs a geospatial index/cell projection, not merely sorting by driver ID inside a city. A large city may span many physical partitions; moving drivers require projection updates. Hash(trip_id) supports trip lookup but not all trips for a user without an index. (user_id, month) buckets bound logical history scans to twelve monthly buckets for one year, not necessarily twelve physical shards or a guaranteed query latency. Budget late events, hot users and retention.

Migration arithmetic: copying 2 TB at 100 MB/s takes 20,000 seconds (about 5.56 hours) using decimal units and ignoring overhead. If apply capacity is 30 MB/s while changes arrive at 20 MB/s, a 60 GB change backlog needs 6,000 seconds to drain at the net 10 MB/s. If apply is only 15 MB/s, it never catches up under the continuing 20 MB/s workload. Throttle copy without starving catch-up; measure service latency during both phases.

Historical incident pointers

The original Discord message-storage article is an optional hot-partition investigation pointer, and Instagram sharding/IDs article is an optional identity/placement pointer. Their exact article availability, former celebrity anecdote and quantitative traffic claims were not freshly verified here. The mechanism survives as explicit synthetic hot-key and backlog calculations above, not an invented account of either company's incident.

Where this shows up in the rest of the plan

Replication copies partitions; queues uses key-based ordering and bounded consumption; CDC teaches coordinated snapshots and replay. Consensus can replicate each shard, but one thousand logical shards need not mean exactly one thousand Raft groups in every product. Caching revisits placement and hot keys; do not assume memcached and Redis Cluster use the same client-side ring. The original Kafka-partitioner, Elasticsearch document-routing and Twitter-style celebrity-design connections remain optional comparison topics, not release-specific claims tested by this ring program. For each, ask which key selects ownership and whether changing placement changes query fan-out. These reuse placement and ordering ideas but do not share one universal routing algorithm.

(e) Recall + stretch · 10 min

1. Hash, range, directory? Compute a placement, select a key interval, or look up explicit ownership. Cassandra/Dynamo-style rings illustrate hash placement, Bigtable illustrates ordered row ranges, and a tenant routing map illustrates directory placement. Some products combine them.
2. Why is a country query expensive after hashing user ID? The routing key does not identify matching users. Fan out or maintain a country projection/global index with its own update/consistency cost.
3. What does adding a fifth balanced node move? A ring expects around one fifth of keys; modulo 4→5 expects around four fifths. Neither is exact for every finite sample. Modulo 4→8 moves about half, showing why “all keys always move” is false.
4. Can range auto-splitting solve every hotspot? No. It can divide a large interval, not an indivisible heavily written key. Row-key/schema choices remain important in managed stores.
5. Three hot-key responses? Cache/coalesce reads, split independent writes into subkeys, or move/isolate capacity. Apply rate limits if none can preserve the invariant at the requested rate.
6. What makes the reshard cutover safe? A defined source authority, coordinated snapshot and stream, validated cutover position, fencing/epoch enforcement, and a rollback plan that includes writes accepted after cutover. Naive dual-write plus a config flip omits those requirements.
7. Does a TTL-cached directory eliminate a single point of failure? No. Replicate the directory appropriately and reject stale ownership at servers; define behavior when the authoritative map is unreachable.

Additional original recall answers

• Timestamp-only write hotspot: monotonically increasing time ranges can direct newest writes into one range. Add a meaningful bucket dimension if events can be partitioned independently; range queries then merge buckets. Late timestamps and uneven bucket traffic still need handling.
• Why virtual nodes? Many ring positions can smooth random ownership arcs and support placement flexibility. Neither 150 vnodes nor a ±3% balance guarantee is universal; the deterministic ring fixture measures one distribution, not a traffic bound.
• Why not universally shard by tenant? Tenant locality can keep queries/transactions local, but a large tenant may require buckets or dedicated capacity. CPU isolation, per-tenant restores and SLAs require additional mechanisms, not merely a tenant key or an invented 99% workload assumption.

Explain-out-loud test

Explain the 80% versus 20% movement expectations, the 90.9% hot-key trace, the 18.2% fan-out calculation, and the version-100 backfill overwriting version 101. Then draw the epoch-7→8 cutover including a stale router and rollback after a new write.

Continue with message queues. Learning plan.

Sources and review boundary

Current complete lesson reviewed 2026-10-10. Primary snapshots from 2026-10-07 were reused after raw/canonical/quotation verification; MongoDB 8.0 transaction and unique-index pages were retrieved 2026-10-10. This is not a fresh fetch of every mutable reference; mutable documentation is bound to saved source hashes in the accompanying claim evidence. Numerical examples are explicit teaching assumptions. The bounded Python examples were run locally; no cloud/distributed deployment, PostgreSQL cluster, broker, object store, load benchmark or regional drill was executed. Compilation and diagram parsing do not establish deployed-system correctness.

• Dynamo Paper — Dynamo uses multiple ring positions per physical node.

• MongoDB 8.0 shard keys — Shard key review needs cardinality, frequency, monotonicity and query distribution.

• Bigtable — Bigtable row-key order affects locality.; Hotspot prevention requires write-pattern analysis.

• Vitess — Vitess migration offers explicit data validation.; Vitess reverse replication supports rollback after new-target writes in its workflow.

• MongoDB 8.0 transaction production considerations — chunk-migration interaction and outside-read visibility during commit.

• MongoDB 8.0 unique indexes — shard-key prefix, hashed-index and per-shard _id constraints.

• Spanner, OSDI 2012 — a concrete two-phase-commit-over-Paxos design; not a claim that every sharded store implements it.