Change data capture & incremental loads

Why the log, not the table

Every transactional database writes changes to a write-ahead log (Postgres WAL, MySQL binlog, SQL Server transaction log) before applying them, and uses that log for crash recovery and replication. Log-based CDC attaches a reader to that log and decodes it into change events: for each committed row change, the table, the operation (create, update, delete), the row before and after, the transaction id and the commit position. Because it reads what the database already writes, it sees every change including deletes and columns without timestamps, in exact commit order, without querying the tables at all.

A Debezium pipeline from Postgres

The most common open-source setup is Debezium running in Kafka Connect: a connector per source database reads the log and publishes one Kafka topic per table, keyed by the row's primary key. Postgres needs wal_level = logical, a replication slot (the server keeps WAL until the slot's consumer has read it) and a publication listing the tables. The connector records its position (LSN) in Kafka Connect's offset topic, so a restart resumes exactly where it stopped.

# postgresql.conf (restart required)
#   wal_level = logical
#   max_replication_slots = 10
#   max_wal_senders = 10

psql -U postgres -d shop <<'SQL'
CREATE ROLE debezium WITH LOGIN REPLICATION PASSWORD 'REPLACE_ME';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
CREATE PUBLICATION shop_pub FOR TABLE public.orders, public.customers;
ALTER TABLE public.orders REPLICA IDENTITY FULL;   -- include full 'before' image on updates/deletes
SQL

{
  "name": "shop-postgres",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "db.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${secrets:shop/debezium:password}",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_shop",
    "publication.name": "shop_pub",
    "table.include.list": "public.orders,public.customers",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "true",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "heartbeat.interval.ms": "10000"
  }
}

Each change event carries an envelope: op (c create, u update, d delete, r read during snapshot), before and after row images, and source metadata (LSN, transaction id, commit timestamp). Deletes are followed by a tombstone (a null value for the key) so that log-compacted topics drop the key. The event is the unit of truth for everything downstream.

{
  "op": "u",
  "ts_ms": 1772352000123,
  "before": {"id": 9001, "customer_id": 1042, "status": "placed", "amount_cents": 4250, "updated_at": 1772351940000},
  "after":  {"id": 9001, "customer_id": 1042, "status": "paid",   "amount_cents": 4250, "updated_at": 1772352000000},
  "source": {"db": "shop", "schema": "public", "table": "orders", "lsn": 48219044, "txId": 771203}
}

Applying changes downstream

The consumer side turns a stream of events into a table that mirrors the source. Two patterns. Append-only changelog: land every event as a row in a bronze table (cheap, complete history, the audit trail), then derive the current state with a windowed deduplication — latest event per key, excluding deletes. Merge into a mirror: apply events to a table-format table with MERGE, in micro-batches, ordered by commit position, so the silver table is always the source's current state. Most platforms do both: the changelog for history and reprocessing, the merged mirror for queries.

MERGE INTO lake.silver.orders t
USING (
  SELECT *
  FROM (
    SELECT
      e.after.id            AS id,
      e.after.customer_id   AS customer_id,
      e.after.status        AS status,
      e.after.amount_cents  AS amount_cents,
      e.op,
      e.source.lsn          AS lsn,
      ROW_NUMBER() OVER (PARTITION BY COALESCE(e.after.id, e.before.id) ORDER BY e.source.lsn DESC) AS rn,
      COALESCE(e.after.id, e.before.id) AS key_id
    FROM lake.bronze.orders_changes e
    WHERE e.ingested_at >= TIMESTAMP '2026-03-04 10:00:00'
  ) WHERE rn = 1
) s
ON t.id = s.key_id
WHEN MATCHED AND s.op = 'd' THEN DELETE
WHEN MATCHED AND s.lsn > t.last_lsn THEN UPDATE SET status = s.status, amount_cents = s.amount_cents, last_lsn = s.lsn
WHEN NOT MATCHED AND s.op != 'd' THEN INSERT (id, customer_id, status, amount_cents, last_lsn)
  VALUES (s.key_id, s.customer_id, s.status, s.amount_cents, s.lsn);

Two details make this robust. Keeping last_lsn on the mirror and only applying newer events makes replays idempotent: a batch applied twice changes nothing. Handling deletes explicitly (op = 'd') is the reason CDC was chosen; a mirror that ignores them drifts. For slowly changing dimensions the same events feed a Type 2 history table, with each change becoming a new versioned row and the previous row closed at the event's commit time.

Snapshots, schema changes and restarts

A new connector must capture the existing rows before it can stream changes: the initial snapshot reads each table (as op = 'r' events) at a consistent point and then continues from the log position recorded at the snapshot's start, so nothing between snapshot and streaming is missed. Snapshots of large tables take time and load; incremental snapshot features chunk them and interleave with streaming. Re-snapshotting a single table after a problem is a normal operation, and downstream merges handle it because r events apply like upserts.

• Schema changes at the source (a new column, a type change) appear in the event schema; with a schema registry in backward mode, additive changes flow through, and downstream tables using on_schema_change or table-format evolution pick them up. Breaking changes need coordination, exactly as the data contract described.
• Slot growth: if the connector stops, Postgres retains WAL for the slot indefinitely and the disk fills. Monitor slot lag (pg_replication_slots), alert on it, and drop slots for decommissioned connectors.
• Heartbeats: on quiet databases the slot position does not advance; heartbeat events keep it moving so WAL can be released.
• Restarts resume from the committed offset with at-least-once semantics; downstream idempotency (the last_lsn guard) absorbs the replayed events.
• Ordering is per key within a partition; keying topics by primary key preserves per-row order, which is all a mirror needs.

SELECT slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS retained_wal
FROM pg_replication_slots;

-- alert if retained_wal exceeds a few GB, or if active is false for more than a few minutes

When CDC is worth it

CDC is the right tool when deletes and every intermediate change matter (finance, inventory, auditing), when latency must be seconds to minutes, when the source cannot tolerate polling, or when many consumers need the same changes (a mirror, a search index, a cache, a fraud model). It is the wrong tool when a nightly snapshot of a small table is enough, when the team cannot operate Kafka and Connect, or when the source database team will not enable logical replication. Managed CDC services and warehouse-native connectors reduce the operational burden and are often the pragmatic middle.