Online resharding

Deep dive

How it works

Parts:

• source shard — serves the range, appends every change to a positioned log (WAL, binlog, oplog)
• snapshot reader — consistent key-ordered read of the range as of log position p
• target shard — takes the bulk copy, then applies log entries in order
• routing table — shard map with a version in a metadata store; clients cache it
• orchestrator — drives phases, watches lag, decides cutover and cleanup

Steps:

1. Choose the split point (size or load midpoint) or the target node; mark the target "not serving".
2. Snapshot at p and bulk copy in key order; writes continue on the source and append after p.
3. Tail the log from p into the target; repeat while lag > limit.
4. Verify with row counts or chunked checksums (Vitess VDiff).
5. Cutover: fence the source (fencing-tokens), drain entries past the last applied position, publish routing version v+1 atomically.
6. Stale clients get "moved(v+1, owner)" and retry; reads follow the same rule.
7. After a grace period (rollback window, open cursors drained) delete the source copy.

 phase    snapshot copy      catch-up (lag -> 0)     fence+drain   serving
 source   [serve ============================================]   fenced   -
 target   [<== copy @p ==][<== replay log p..n ==]  [drain]      [serve ====]
 routing  v ...................................................| v+1 (atomic)

Code

Simulation: snapshot at p, copy while writes arrive, replay, fenced cutover, stale-client rejection. Compiled and run, std only.

use std::collections::BTreeMap;

type Op = (u64, Option<u64>); // (key, Some = put | None = delete)
#[derive(Clone, Copy, PartialEq, Debug)]
enum Owner { Source, Target }
#[derive(PartialEq, Debug)]
enum Reject { Moved { version: u64, owner: Owner }, Fenced }

#[derive(Default)]
struct Shard { data: BTreeMap<u64, u64>, log: Vec<Op>, fenced: bool }
impl Shard {
    fn apply(&mut self, (k, v): Op) { // idempotent: replaying an entry twice is harmless
        match v { Some(v) => { self.data.insert(k, v); } None => { self.data.remove(&k); } }
    }
}

struct Cluster { src: Shard, dst: Shard, version: u64, owner: Owner } // version = routing table

impl Cluster {
    fn write(&mut self, client_version: u64, op: Op) -> Result<(), Reject> {
        if client_version != self.version {
            return Err(Reject::Moved { version: self.version, owner: self.owner });
        }
        match self.owner {
            Owner::Source if self.src.fenced => Err(Reject::Fenced),
            Owner::Source => { self.src.log.push(op); self.src.apply(op); Ok(()) }
            Owner::Target => { self.dst.apply(op); Ok(()) }
        }
    }
    /// Copy a snapshot taken at log position p; `tick` = writes arriving meanwhile.
    fn copy(&mut self, chunk: usize, mut tick: impl FnMut(&mut Cluster)) -> usize {
        let p = self.src.log.len();
        let snap: Vec<(u64, u64)> = self.src.data.iter().map(|(k, v)| (*k, *v)).collect();
        for part in snap.chunks(chunk) {
            for &(k, v) in part { self.dst.data.insert(k, v); }
            tick(self);
        }
        p
    }
    /// Replay the log from `pos`, `rate` entries per round, until lag <= limit.
    fn catch_up(&mut self, mut pos: usize, rate: usize, limit: usize,
                mut tick: impl FnMut(&mut Cluster)) -> (usize, Vec<usize>) {
        let mut lags = vec![];
        loop {
            let lag = self.src.log.len() - pos;
            lags.push(lag);
            if lag <= limit { return (pos, lags); }
            for i in pos..pos + rate.min(lag) { let op = self.src.log[i]; self.dst.apply(op); }
            pos += rate.min(lag);
            tick(self);
        }
    }
    /// Fence the source, drain the last entries, flip the routing version.
    fn cutover(&mut self, pos: usize) {
        self.src.fenced = true;
        for i in pos..self.src.log.len() { let op = self.src.log[i]; self.dst.apply(op); }
        self.version += 1;
        self.owner = Owner::Target;
    }
}

fn main() {
    let mut s = 42u64;
    let mut writer = move |c: &mut Cluster| for _ in 0..100 {
        s ^= s << 13; s ^= s >> 7; s ^= s << 17; // xorshift64
        c.write(c.version, (s % 12_000, if s % 8 == 0 { None } else { Some(s) })).unwrap();
    };
    let mut c = Cluster { src: Shard::default(), dst: Shard::default(), version: 0, owner: Owner::Source };
    for k in 0..10_000 { c.write(0, (k, Some(k))).unwrap(); }
    c.src.log.clear();
    let p = c.copy(500, &mut writer); // copy 500 > write 100 per tick
    let (pos, lags) = c.catch_up(p, 500, 100, &mut writer);
    assert!(lags.windows(2).all(|w| w[1] < w[0]), "lag must shrink: {lags:?}");

    c.src.fenced = true;
    assert_eq!(c.write(0, (1, Some(1))), Err(Reject::Fenced));
    c.src.fenced = false;
    c.cutover(pos);
    assert_eq!(c.src.data, c.dst.data); // no lost or duplicated writes

    let seen = c.src.log.len();
    let stale = c.write(0, (7, Some(777))); // client still holds version 0
    assert_eq!(stale, Err(Reject::Moved { version: 1, owner: Owner::Target }));
    c.write(1, (7, Some(777))).unwrap(); // retry with the new map
    assert_eq!(c.dst.data[&7], 777);
    assert_eq!(c.src.log.len(), seen); // source saw nothing after the flip
    println!("lag per round {lags:?}: ok");
}

Complexity and constants

• Copy: one sequential scan, O(bytes); the target can bulk-build its index from key-ordered input.
• Catch-up: write rate w, replay rate r > w, copy duration T: lag after the copy is w·T, catch-up takes about w·T / (r − w). If r ≤ w it never ends.
• Freeze: limit/r for the final drain, plus one metadata write and routing propagation. The router check is one integer compare per request.

Efficiency

• Cost: measured on an Apple M4 Pro, one thread, in memory (BTreeMap<u64,u64>, 1M keys, 100 writes per 1,000 copied): copy plus interleaved writes 134-166 ms (3 runs), catch-up of the 100k-entry lag 51-65 ms, fenced drain of at most 100 entries 17-144 µs, full compare 3-4 ms. Real moves are bound by disk and network, not CPU.
• Levers:
  • bulk copy in key order — sequential reads, bulk index build on the target — always
  • parallel range copy — scales with disks and links until source foreground latency suffers — large ranges
  • throttle copier and replayer — protects foreground p99 (Vitess has a throttler, see its docs) — hot shards
  • metadata-only split — no data moves (CockroachDB) — range-partitioned stores
  • chunked checksums — re-copy only bad chunks — verify step
• Naive → standard → state of the art:
  • stop writes, copy, restart · downtime proportional to data size
  • snapshot + log tail + fenced cutover · downtime of the last drain, milliseconds
  • metadata-only split, throttled replica moves, reversible switch with reverse replication (Vitess) · no copy for splits, rollback after the flip

Implementation notes

• Take the snapshot and record p atomically (an MVCC snapshot with its log position); a gap loses or doubles writes.
• Make apply idempotent and store the applied position with the target data, so a crash replays safely.
• Fence with an epoch or lease, not "I told the source to stop": a partitioned old source must not accept writes after the flip.
• Send the routing version in every request; the "moved" reply carries the new map, costing one extra round trip.

Pitfalls

• Lost writes in the cutover window: a client with an old map writes to the source after the drain. The version check or fence closes it.
• Read-your-writes across the move: serve from the target only after the drain, never from a partial replay.
• Lag that never shrinks: copy and foreground writes compete for the same disk; throttle one.
• Hot keys do not split: one key stays on one shard.

Variants and alternatives

In production

Vitess MoveTables and Reshard workflows (VReplication copy, VDiff, SwitchTraffic, reverse replication); MongoDB balancer range migrations with a metadata update on the config servers; Redis Cluster slot migration; CockroachDB range splits and replica rebalancing; TiKV with Placement Driver scheduling; Kinesis shard split and merge; Bigtable tablet splits.

Implementations

Sources

• Vitess docs, MoveTables, Reshard, VDiff. MoveTables · Reshard · VDiff
• MongoDB manual, Sharded Cluster Balancer (range migration procedure). docs
• Redis docs, Redis Cluster specification (hash slot migration, ASK/MOVED). docs
• CockroachDB docs, Distribution Layer (range splits). docs
• PingCAP docs, Scheduling (Placement Driver). docs
• AWS docs, Resharding a stream (shard split and merge). docs
• Fay Chang et al., Bigtable: A Distributed Storage System for Structured Data, OSDI 2006. paper