Change data capture

Deep dive

How it works

Parts:

• log — the commit log the database writes anyway (WAL, binlog, oplog)
• decoder — rebuilds each transaction's row changes and emits them at commit; aborted ones never appear
• position — LSN, GTID set or resume token: "everything up to here is done"
• retention — the source keeps log segments until the consumer acknowledges a position
• snapshot — consistent copy of existing rows taken at a known position
• sink — applies events idempotently, then stores or acknowledges the position

Operations:

1. Enable: PostgreSQL wal_level = logical, a publication (tables, and which of INSERT/UPDATE/DELETE/TRUNCATE) and a slot; MySQL binlog_format = ROW, binlog_row_image = FULL (Debezium requires both).
2. Snapshot: a new slot can export a snapshot showing exactly the state after which all changes are in the stream, so copy and stream join without a gap. Debezium streams from the position it read at snapshot start.
3. Decode: PostgreSQL decodes concurrent transactions in commit order, drops rolled-back ones, folds savepoints into their transaction. A plugin formats the output: pgoutput (built in), wal2json, test_decoding.
4. Emit: Debezium event = before, after, op (c, u, d, r snapshot row, t, m), source (LSN, transaction id, time).
5. Route: key = primary key. Kafka writes equal keys to one partition and a partition is read in write order, so each row's changes stay ordered; order across rows is not kept.
6. Apply, then acknowledge: PostgreSQL frees WAL below confirmed_flush_lsn; the slot position is persisted only at checkpoint, so a restart may resend recent changes.
7. Schema change: PostgreSQL logical replication does not carry DDL. MySQL's binlog does; Debezium replays it into a schema history so each event is read with the schema of its time.
8. Re-snapshot: Debezium incremental snapshots (design DDD-3) read a table by primary key in chunks (default 1,024 rows) while streaming continues and resolve a row seen twice in a watermark window; started by a signal row.

 writers -> [ table + commit log ] --slot/offset--> decoder --> events --> broker (key = PK)
                    | snapshot @ position P                                      |
                    +---------- rows (op=r) --------------------------------->  sink
                                stream from P --- apply, then ack LSN <---------+

The trigger row is this page's inference. Alibaba's canal README says its early sync used business triggers before it moved to log parsing.

Code

Toy table with a WAL of interleaved transactions (one aborted), a decoder emitting at commit, a sink with a stored offset and a crash between apply and offset, and a polling baseline.

use std::collections::{BTreeMap, HashMap};

#[derive(Clone)]
enum Rec { Begin(u32), Put(u32, u32, u32), Del(u32, u32), Commit(u32), Abort(u32) }

#[derive(Debug, PartialEq)]
struct Event { commit_lsn: u64, idx: u32, key: u32, after: Option<u32> }

#[derive(Default)]
struct Db {
    wal: Vec<(u64, Rec)>,
    table: BTreeMap<u32, (u32, u64)>,
    pend: HashMap<u32, Vec<Rec>>,
}
impl Db {
    fn log(&mut self, r: Rec) -> u64 { let lsn = self.wal.len() as u64 + 1; self.wal.push((lsn, r)); lsn }
    fn begin(&mut self, t: u32) { self.log(Rec::Begin(t)); self.pend.insert(t, vec![]); }
    fn put(&mut self, t: u32, k: u32, v: u32) { let r = Rec::Put(t, k, v); self.log(r.clone()); self.pend.get_mut(&t).unwrap().push(r); }
    fn del(&mut self, t: u32, k: u32) { let r = Rec::Del(t, k); self.log(r.clone()); self.pend.get_mut(&t).unwrap().push(r); }
    fn abort(&mut self, t: u32) { self.log(Rec::Abort(t)); self.pend.remove(&t); }
    fn commit(&mut self, t: u32) -> u64 {
        let lsn = self.log(Rec::Commit(t));
        for r in self.pend.remove(&t).unwrap() {
            match r { Rec::Put(_, k, v) => { self.table.insert(k, (v, lsn)); } Rec::Del(_, k) => { self.table.remove(&k); } _ => {} }
        }
        lsn
    }
}

/// Reorder buffer: hold each transaction's changes, emit them at its commit record, drop aborted ones.
/// Emits only commits after `after_lsn`. (A real slot keeps restart_lsn; this toy rescans the log.)
fn decode(wal: &[(u64, Rec)], after_lsn: u64) -> Vec<Event> {
    let (mut buf, mut out): (HashMap<u32, Vec<(u32, Option<u32>)>>, Vec<Event>) = (HashMap::new(), vec![]);
    for (lsn, rec) in wal {
        match rec {
            Rec::Begin(t) => { buf.insert(*t, vec![]); }
            Rec::Put(t, k, v) => buf.get_mut(t).unwrap().push((*k, Some(*v))),
            Rec::Del(t, k) => buf.get_mut(t).unwrap().push((*k, None)),
            Rec::Abort(t) => { buf.remove(t); }
            Rec::Commit(t) => {
                let changes = buf.remove(t).unwrap();
                if *lsn > after_lsn {
                    for (i, (key, after)) in changes.into_iter().enumerate() {
                        out.push(Event { commit_lsn: *lsn, idx: i as u32, key, after });
                    }
                }
            }
        }
    }
    out
}

#[derive(Default)]
struct Sink { replica: BTreeMap<u32, u32>, offset: u64, delivered: usize }

/// Apply a transaction, then save the offset. `crash_at`: die between the two.
/// Applying an after-image (set or remove) is idempotent.
fn consume(wal: &[(u64, Rec)], s: &mut Sink, crash_at: Option<u64>) {
    let evs = decode(wal, s.offset);
    let mut i = 0;
    while i < evs.len() {
        let c = evs[i].commit_lsn;
        while i < evs.len() && evs[i].commit_lsn == c {
            match evs[i].after { Some(v) => { s.replica.insert(evs[i].key, v); } None => { s.replica.remove(&evs[i].key); } }
            s.delivered += 1;
            i += 1;
        }
        if crash_at == Some(c) { return; }
        s.offset = c;
    }
}

/// Baseline: SELECT ... WHERE updated_at > last
fn poll(db: &Db, last: &mut u64, replica: &mut BTreeMap<u32, u32>) -> usize {
    let rows: Vec<_> = db.table.iter().filter(|(_, (_, ts))| ts > last).map(|(k, (v, ts))| (*k, *v, *ts)).collect();
    for &(k, v, ts) in &rows { replica.insert(k, v); *last = (*last).max(ts); }
    rows.len()
}

fn main() {
    let mut db = Db::default();
    db.begin(1); db.put(1, 1, 10); db.put(1, 2, 20); db.commit(1);
    db.begin(2); db.begin(3); db.put(2, 3, 30); db.put(3, 1, 11);
    db.commit(3); db.commit(2);                       // commit order differs from begin order
    db.begin(4); db.put(4, 9, 99); db.abort(4);       // must never be emitted
    let (mut last, mut poll_replica) = (0, BTreeMap::new());
    poll(&db, &mut last, &mut poll_replica);         
    db.begin(5); db.put(5, 1, 12); db.commit(5); db.begin(9); db.put(9, 1, 13); db.commit(9);
    db.begin(6); db.del(6, 2); db.commit(6);         
    db.begin(7); db.put(7, 5, 5); db.commit(7); db.begin(8); db.del(8, 5); db.commit(8);
    let seen = poll(&db, &mut last, &mut poll_replica);
    let all = decode(&db.wal, 0);
    assert!(all.windows(2).all(|w| (w[0].commit_lsn, w[0].idx) < (w[1].commit_lsn, w[1].idx))); // commit order
    assert!(all.iter().all(|e| e.key != 9));                                                // aborted txn dropped
    assert_eq!(all.len(), 9);                          // committed changes: 2+1+1+1+1+1+1+1
    assert_eq!(seen, 1);                               // the poll saw 1 row for 5 later changes
    let source: BTreeMap<u32, u32> = db.table.iter().map(|(k, (v, _))| (*k, *v)).collect();
    let (mut s, crash_lsn) = (Sink::default(), db.wal.iter().position(|(_, r)| matches!(r, Rec::Commit(6))).unwrap() as u64 + 1);
    consume(&db.wal, &mut s, Some(crash_lsn));        // crash after applying tx 6, offset not saved
    assert!(s.offset < crash_lsn);
    consume(&db.wal, &mut s, None);                   // restart: tx 6 is delivered again
    assert_eq!(s.delivered, all.len() + 1);           // exactly one redelivered event
    assert_eq!(s.replica, source);                    // idempotent: replica equals source
    assert_ne!(poll_replica, source);                 // polling kept the deleted key 2
    assert!(poll_replica.contains_key(&2));
    println!("source {source:?}\nCDC    {:?} ({} events, {} delivered)\npoll   {poll_replica:?}", s.replica, all.len(), s.delivered);
}

Run on this Mac (Apple M4 Pro), asserts pass. Exact counts: 9 committed changes decoded, the aborted one dropped; 10 delivered after the crash, one repeated; replica equals source ({1: 13, 3: 30}); the poller kept deleted key 2 and saw one row for five later changes.

Complexity and constants

• Decode O(records in the log); memory = open transactions' changes.
• Retained WAL = write rate × time since the last acknowledgement (10 MB/s × 1 h = 36 GB, arithmetic).
• Cost drivers: update rate, row width, old-value logging.

Efficiency

• Cost: one sequential log read and decode per change; no table queries. Bound: sink apply rate and acknowledgement latency. No benchmark claimed.
• Levers:
  • publication filter on tables and operations (PostgreSQL docs) — decodes and ships less — most traffic unneeded
  • REPLICA IDENTITY DEFAULT, not FULL — logs only key columns of old rows (PostgreSQL docs) — sink needs only the key
  • heartbeat events — slot advances although captured tables are idle (Debezium docs) — busy database, quiet tables
  • max_slot_wal_keep_size — caps retained WAL; a slot past it becomes lost (PostgreSQL docs) — stalled-consumer risk
  • incremental snapshot — resumable, streaming continues (Debezium docs) — large tables
• Naive → standard → state of the art:
  • dual write in application code · drifts on a crash between the writes
  • poll updated_at · misses deletes and intermediate updates
  • log + initial snapshot + idempotent sink · every change, resumable
  • plus incremental snapshots and streamed in-progress transactions · no long snapshot, bounded memory

Implementation notes

• If the sink is a database, store the position in the same transaction as the data; else save it after the apply.
• One slot per consumer; drop unused slots.
• Apply after-images (set or remove by key), never deltas, so a repeat is harmless; or skip events at or below the stored LSN (event deduplication).

Pitfalls

• A slot without a live consumer still holds WAL and catalog rows; in extreme cases the database shuts down to prevent transaction-id wraparound (PostgreSQL docs).
• Table without primary key under REPLICA IDENTITY DEFAULT logs no old row; Debezium emits only create events for it.
• Debezium sends a primary-key update as delete (old key) plus create (new key).
• Failover: a slot lives on one server; PostgreSQL's slot synchronization (failover option) is needed on a standby, else re-snapshot.

Variants and alternatives

In production

• Debezium into Kafka; its docs state at-least-once delivery.
• Maxwell (MySQL): README lists ETL, audit log, cache expiry, search indexing, inter-service communication.
• Canal (MySQL): README lists database mirroring, real-time backup, index maintenance, cache refresh.

Implementations

Sources

• S. Das et al., All aboard the Databus!, ACM SoCC, 2012. paper
• PostgreSQL docs: logical decoding, output plugins, REPLICA IDENTITY, pg_replication_slots, replication settings, restrictions, publications
• Debezium: PostgreSQL, MySQL, features, log-based vs polling, DDD-3
• MySQL 8.4 (archived): binary log formats, binlog_row_image, GTIDs
• MongoDB change streams · Kafka concepts · Confluent JDBC source
• READMEs: canal, maxwell, wal2json, pgwire-replication, mysql_cdc