Databases · Replication and Sharding

Online Resharding: Moving a Range Without Stopping Writes

How a range is migrated while it is being written to, why the cutover is a metadata transaction, and what the dual-write window costs in correctness.

A range of keys has to move from one machine to another because the machine is full, or because a new machine joined and the cluster is unbalanced. The range is under continuous write load. Stopping the writes for the duration of the copy is the simple answer, and on a range of tens of gigabytes over a slow link that duration runs into hours, which is not an outage an application can absorb. The alternative is to copy while serving, and the entire difficulty is in the moment when the copy is complete and the ownership changes.

The copy is the easy part

Start by snapshotting the range on the source at a known log position. The snapshot is a consistent view of the data as of that position, taken without blocking writes because the engine’s snapshot mechanism is already how consistent reads work. Transfer it to the destination in chunks, in whatever order is convenient, at whatever rate the network allows without starving foreground traffic.

Writes continue on the source during the transfer, appending to the log. The destination’s copy is therefore stale by exactly the writes that occurred after the snapshot position, and the log is the record of them. Once the bulk copy finishes, the destination requests the log records from the snapshot position forward and applies them. It is now at some position that is close to the source’s current position, and the gap is the log tail produced during the catch-up.

The catch-up converges only if the destination can apply records faster than the source produces them. If the write rate on the range exceeds the destination’s apply rate, the gap grows and the migration never completes. This is the first hard constraint, and it is not a matter of tuning: a range that is write-hot relative to what a single destination can absorb cannot be migrated by log catch-up at all. The options are to throttle the source’s writes for the range, which is a partial outage, or to split the range first and migrate the halves.

Throttling is a real technique and its cost should be stated plainly. Every throttled write is a write the application waited for, so the migration shows up as added latency on the affected keys for its duration. The alternative, copying a range that is receiving writes faster than the copy can track, produces a migration that runs forever while consuming network and CPU, which is worse.

The cutover is a metadata decision, not a data decision

When the gap is small enough, the migration stops the world for that range only. The sequence is:

  1. The source stops accepting writes for the range, or forwards them to the destination.
  2. The remaining log is drained, so the destination is at the source’s exact position.
  3. A metadata transaction changes the owner from source to destination, incrementing the range’s epoch.
  4. Clients that were routed to the source are redirected, and requests carrying the old epoch are rejected.

Steps 1 and 3 must be ordered such that no write is lost. If the metadata changes first, clients send writes to the destination while the source still has data the destination lacks, and those writes land on an incomplete copy. If the source stops first and the metadata never changes, the range is unavailable until the migration is retried. The safe order is stop-then-transfer-then-commit-metadata, with the stop being a lease rather than an indefinite halt.

The epoch increment in step 3 is what makes the transition detectable rather than merely intended, for the reasons set out in shard routing and ownership metadata. A client holding the old epoch gets an explicit rejection; a client that has never seen the range gets the new owner directly.

The window in step 1 is the one that has to be bounded. Draining the log tail is fast if the tail is short, and the tail is short precisely because the catch-up was keeping up. So the design works when the destination has spare apply capacity, and fails when it does not — the same condition as above, appearing at a different point in the process.

Dual writes and the ordering they need

Some systems avoid stopping the range by having the source forward writes to the destination as they arrive, from the moment the copy begins. The destination applies both the bulk copy and the forwarded writes, and the source remains the owner until the destination is caught up, at which point the metadata flips and the source stops forwarding.

The forwarding path is a second replication stream, with all the ordering requirements of the replication log. The forwarded writes must be applied in their original order relative to each other, and they must not be applied before the bulk copy has covered the keys they touch — otherwise a forwarded update overwrites a row the copy has not yet written, and the copy then overwrites the update. The usual resolution is to apply forwarded writes to a staging area keyed by position and merge after the bulk copy, or to tag each copied chunk with its position and skip chunks that the forwarded stream has already superseded.

This is the part of online resharding that is genuinely subtle, and it is why many systems choose the simpler stop-the-range approach with a short stop. The stop is bounded by the log tail, which is bounded by the catch-up rate, so the operation is safe as long as the catch-up was healthy. Dual writes remove the stop and add a merge problem whose failure mode is silent data loss.

source: snapshot at pos 1000, bulk copy to dest
source: writes 1001..1200 during copy
dest:   applies bulk copy (as of 1000)
dest:   requests log 1001.., applies -> at 1200
source: freeze range, write final records 1201..1203, drain
metadata: owner = dest, epoch += 1
clients: old epoch rejected, refresh, retry

The freeze in that sequence is the only point at which the range is unavailable, and its duration is the drain of the final few records. Everything before it is concurrent with normal service.

Splitting and merging change the boundary set

Resharding a range in half is a different operation from moving one. The split produces two ranges with new boundaries, and the metadata must record both. The data does not move: both halves stay on the same machine, and the split is a metadata operation plus a change in how the engine addresses the data — two logical ranges over one physical store, or a physical split of the underlying structure.

Merging is the inverse and is harder, because it requires both ranges to be on the same machine before the merge, so it is usually a move followed by a merge. The reason to merge at all is metadata pressure: a cluster that has split aggressively accumulates small ranges whose per-range overhead dominates, and the overhead is not proportional to the data.

The boundary set is the object that every router must agree on, so splits and merges are metadata transactions with the same epoch discipline as moves. A split that is applied on the owner but not yet in the metadata produces a machine that believes it owns two ranges while routers believe it owns one; requests for the second range arrive at a machine that is prepared to serve them, which is benign, and requests routed by the old single boundary still work because the same machine owns both halves. The reverse — metadata split applied, owner not yet split — produces a machine that receives requests for a boundary it does not recognise, which is an error rather than a silent misroute. That asymmetry is why the safe order is owner-first, metadata-second.

What the migration does to the neighbours

Moving a range shifts load, and the shift is not confined to the two machines involved.

The destination now holds the range’s data plus its write stream, and its own existing ranges compete for the same CPU, memory, and disk bandwidth. A migration that fills the destination’s cache with cold data evicts the hot data of the ranges it already served, and the resulting latency increase on those ranges is a real cost that a capacity model based only on bytes will miss.

The source benefits, but not immediately: it holds the range’s data until the migration is confirmed, so its disk usage does not fall until the metadata change and the subsequent cleanup. A cluster that is migrating to relieve disk pressure needs the source’s space to be freed before it can accept another range, and the confirmation delay is part of the capacity planning.

The metadata store sees a burst of writes for each migration — the epoch bump, the new owner, and for a split the new boundaries. A cluster running many migrations in parallel generates a metadata write rate that can exceed the steady-state rate by orders of magnitude, and the metadata store is the component whose failure stops every routing decision. Rate-limiting migrations is therefore a correctness-preserving measure, not merely a courtesy to the network.

When not to migrate

The strongest argument against online resharding is that a well-chosen partition count makes it unnecessary for a long time. If the key space is divided into far more partitions than there are machines, growth is a matter of moving whole partitions, and the mapping from key to partition never changes. That indirection is described in hash and range partitioning, and its purpose is exactly to make the expensive operation rare.

The second argument is that a range that is hot because of a single key cannot be fixed by moving it. A key receiving a disproportionate share of requests is a hotspot wherever it lives, and migrating the range moves the hotspot rather than spreading it. The remedies for that are key redesign or a caching layer in front of the store, neither of which is a resharding operation.

What remains is genuine growth and genuine skew in key volume, and for those the migration is unavoidable. The design question is then how much machinery to build: a stop-the-range cutover is simple enough to reason about and safe under a bounded log tail, and a dual-write path removes the stop at the price of a merge protocol. Most systems start with the former and add the latter only when the stop is measured to be too long, which is the right order — the simple mechanism’s failure is visible, and the complex one’s is not.

References

  1. Bigtable: A Distributed Storage System for Structured Data — Fay Chang et al., OSDI 2006
  2. Spanner: Google's Globally-Distributed Database — James C. Corbett et al., OSDI 2012
  3. Designing Data-Intensive Applications, Chapter 6: Partitioning — Martin Kleppmann, O'Reilly Media, 2017