Chunked backfill

Deep dive

How it works

In: a table or time series ordered by key, a deterministic transform, a chunk size · Out: every old record carries the new field or lives in the new store · State: one durable checkpoint (last key done) plus the high-water mark end · Cost: O(N) rows overall, O(log N + n) per chunk of n rows, at most one chunk redone after a crash

1. Switch the live write path first (dual write, trigger or CDC), so every new or changed row carries the new field. Record end = highest key present, or a cut-over timestamp.
2. Read the checkpoint. Select the next n keys in (checkpoint, end] by index range scan.
3. In one short transaction, for each row where the field is unset, compute and write it. A conditional write (WHERE new_col IS NULL, an upsert, or a version check) means a live write or an earlier run always wins.
4. Commit, then save the checkpoint = last key of the chunk. A crash between 3 and 4 re-runs one chunk, which writes nothing.
5. Throttle: sleep in proportion to chunk time, or pause while replica lag, p99 latency or lock waits exceed a limit.
6. Stop when a chunk is empty; keys above end belong to the live path.
7. Verify per key range: row counts and a checksum of the derived values against a recompute. Re-running the whole backfill must write 0 rows.
8. Switch reads to the new field, then finish expand-contract.

For a historical load the key is time. Fix a cut-over timestamp T: the live feed writes everything at or after T, the backfill everything before T, newest first if recent data matters most, oldest first if consumers replay in order. With records keyed by (series, timestamp) an overlap is harmless: upsert on both sides.

Code

use std::collections::BTreeMap;
use std::ops::Bound::{Excluded, Included, Unbounded};
use std::time::Instant;

#[derive(Clone, Debug, PartialEq)]
struct Row { price: i64, qty: i64, notional: Option<i64> } // `notional` is the new column
type Table = BTreeMap<u64, Row>;

fn derive(r: &Row) -> i64 { r.price * r.qty } // deterministic: same input, same output

/// One chunk = one transaction: the next `n` keys in (after, end], fill only unset fields.
/// Returns (last key seen, rows actually written). `None` = range exhausted.
fn backfill_chunk(t: &mut Table, after: Option<u64>, end: u64, n: usize) -> (Option<u64>, usize) {
    let lo = after.map_or(Unbounded, Excluded); // None = nothing done yet (key 0 is a valid key)
    let keys: Vec<u64> = t.range((lo, Included(end))).map(|(k, _)| *k).take(n).collect();
    let mut written = 0;
    for k in &keys {
        let r = t.get_mut(k).unwrap();
        if r.notional.is_none() { // idempotent: live writes and earlier runs win
            r.notional = Some(derive(r));
            written += 1;
        }
    }
    (keys.last().copied(), written)
}

/// Resumable driver. `checkpoint` stands for a durable row (last key done); `crash_after`
/// simulates a crash after chunk N committed but before its checkpoint was saved.
fn run(t: &mut Table, checkpoint: &mut Option<u64>, end: u64, n: usize, crash_after: Option<usize>,
       mut live: impl FnMut(&mut Table, usize)) -> Result<usize, &'static str> {
    let mut total = 0;
    for chunk_no in 0.. {
        live(t, chunk_no); // live traffic interleaves between chunks
        let (last, w) = backfill_chunk(t, *checkpoint, end, n);
        total += w;
        if Some(chunk_no) == crash_after { return Err("crashed before saving checkpoint"); }
        match last { Some(k) => *checkpoint = Some(k), None => return Ok(total) } // save after commit
        // throttle here (step 5)
    }
    unreachable!()
}

fn main() {
    let mut t: Table = (0..10_000u64).map(|k| (k * 2, Row { price: k as i64 + 1, qty: 3, notional: None })).collect();
    let (mut next_id, end) = (20_000u64, 19_998u64); // end = max key when the backfill started
    // Live path (dual write) maintains `notional` itself, including deletes.
    let mut live = |t: &mut Table, step: usize| {
        let r = Row { price: step as i64 + 5, qty: 2, notional: Some((step as i64 + 5) * 2) };
        t.insert(next_id, r); next_id += 1;
        if let Some((_, row)) = t.iter_mut().nth(step * 7 % 5000) { row.price += 1; row.notional = Some(derive(row)); }
        if let Some(k) = t.keys().nth(step * 11 % 5000).copied() { t.remove(&k); } // delete during the window
    };
    let mut checkpoint = None;
    let crashed = run(&mut t, &mut checkpoint, end, 500, Some(6), &mut live);
    assert!(crashed.is_err());
    let resumed = run(&mut t, &mut checkpoint, end, 500, None, &mut live).unwrap(); // resume from checkpoint
    println!("resumed from key {checkpoint:?}, wrote {resumed} more rows");
    let again = run(&mut t, &mut None, end, 500, None, |_, _| {}).unwrap(); // full re-run
    assert_eq!(again, 0, "re-run must change nothing");
    // Verification before switching reads: every row has the field and it matches the source columns.
    assert!(t.values().all(|r| r.notional == Some(derive(r))));
    println!("verified {} rows", t.len());

    // Measure: keyset chunks vs OFFSET paging over a 200k-row table.
    let base: Table = (0..200_000u64).map(|k| (k, Row { price: 7, qty: 3, notional: None })).collect();
    for n in [10usize, 1_000, 100_000] {
        let (mut t, mut cp) = (base.clone(), None);
        let s = Instant::now();
        run(&mut t, &mut cp, u64::MAX, n, None, |_, _| {}).unwrap();
        println!("keyset chunk={n:>6}: {:.1} M rows/s", 200_000.0 / s.elapsed().as_secs_f64() / 1e6);
    }
    let mut t = base.clone();
    let s = Instant::now();
    for off in (0..200_000).step_by(1_000) { // OFFSET paging: skip `off` rows each time
        let ks: Vec<u64> = t.keys().skip(off).take(1_000).copied().collect();
        for k in ks { let r = t.get_mut(&k).unwrap(); r.notional = Some(derive(r)); }
    }
    println!("offset chunk=1000: {:.1} M rows/s", 200_000.0 / s.elapsed().as_secs_f64() / 1e6);
}

Compiled and run, release build, measured on this Mac (Apple M4 Pro). The asserts (resume after crash, idempotent re-run, verification) pass; the table is in memory, so the timings show the paging pattern, not database cost.

Complexity and constants

• Total work O(N) plus one index seek per chunk, amortised over n rows; tiny chunks are dominated by that overhead.
• Measured on this Mac (Apple M4 Pro), 200,000 in-memory rows, 5 runs: keyset chunks of 10 rows ran 9 to 22 M rows/s, 1,000 rows 21 to 31 M rows/s, 100,000 rows 18 to 31 M rows/s. OFFSET-style paging with 1,000 rows per page ran 0.9 to 3.4 M rows/s, and its gap widens with table size. Run-to-run noise was large.
• In a real database the bound is the chunk's write and replication cost, not the loop.

Efficiency

• Cost: one chunk = one index range read, n conditional row writes, one commit and one checkpoint write, plus the throttle sleep. The bound is write I/O and replication, not CPU. No real database was measured.
• Levers:
  • adaptive chunk size aiming at a time per chunk — keeps each transaction short as load changes (pt-online-schema-change --chunk-time, default 0.5 s) — shared primaries
  • lag-based throttling — pause while a replica is behind (gh-ost --max-lag-millis; pt-osc --max-lag, default 1 s) — replicated databases
  • sleep proportional to chunk time — --nice-ratio=1 sleeps 1 ms per ms spent copying (gh-ost); a token bucket caps rows per second instead — simple load cap
  • find the work offline — Stripe listed the objects to migrate from a database snapshot on Hadoop, not the live database (Stripe) — large tables
  • lazy copy on update — rows migrate when traffic touches them, shrinking what remains (Stripe) — hot working sets
• Naive → standard → state of the art:
  • one UPDATE over the table · hours of locks and undo, all lost on crash
  • keyset chunks, checkpoint, idempotent writes, fixed sleep · bounded impact, resumable
  • adaptive chunk time plus lag/latency feedback, per-range checksums, watermark interleaving with the change stream · self-tuning, verifiable

Implementation notes

• Keyset paging needs a unique, indexed, deterministic order; for a non-unique sort column use a composite (col, id).
• Keep each chunk's transaction short (pt-osc aims at 0.5 s) so MVCC cleanup and replication are not held back by one old transaction.
• Store the checkpoint in the same database, ideally in the chunk's transaction, so commit and checkpoint cannot disagree.
• Keep transforms deterministic: no now(), random values or external calls, or a re-run yields different data.
• Retry a failed chunk with backoff and jitter; abort on repeated failure rather than skip a range.

Pitfalls

• OFFSET paging: cost grows with the offset, and an insert between pages gives duplicates (Winand).
• No end bound while the live path inserts at the top of the key range: the loop never finishes.
• Checkpoint 0 with key > checkpoint skips key 0. Use "none yet".
• A backfill write overwrites a newer live write with a stale computed value: make it conditional.
• Missed deletes: a copy into a new store resurrects rows deleted during the run unless the live path also deletes there.
• Hot rows touched by both traffic and backfill cause lock waits: use smaller chunks.

Variants and alternatives

In production

• Relational migrations: Stripe's online migration pattern (dual write, then backfill); gh-ost and pt-online-schema-change copy tables in chunks.
• Change data capture: Debezium incremental snapshots read primary-key-ordered chunks; Netflix built DBLog for this.
• Frameworks: Shopify's maintenance_tasks runs interruptible, throttled iterations over records.
• Time series: history behind a live tail (time-series chunk storage); point-in-time recovery replays the same way.

Implementations

Sources

• Stripe Engineering (J. Xu), Online migrations at scale, 2017. post
• M. Winand, We need tool support for keyset pagination, Use The Index, Luke. post
• GitHub, gh-ost command-line flags (chunk-size, max-lag-millis, nice-ratio). doc
• Percona, pt-online-schema-change (--chunk-time, --max-lag). doc
• A. Andreakis, I. Papapanagiotou, DBLog: A Watermark Based Change-Data-Capture Framework, 2020. arXiv
• Debezium, Connector for PostgreSQL: incremental snapshots. doc
• Shopify, maintenance_tasks. code