The full CDC cycle, step by step — every engine

The operator’s sequence for a mode: cdc export from nothing to a warehouse table that follows the source: preflight → anchor + baseline → load → changes → load → an interruption on the CDC leg → load → an idle cycle. Each step names what to run and what must be true afterwards, checked by two readers that share nothing with rivet: the source itself (COUNT(*)) and the warehouse (bq).

The same sequence runs unattended as full_cdc_cycle_{mysql,postgres,mssql,mongo} in tests/live/live_cdc_full_cycle.rs — one body, four engines, through the Rig.

0. Prerequisites per engine

rivet doctor --config cfg.yaml checks the log-side state it can read (PostgreSQL: the slot; MySQL: binlog mode and whether the checkpoint is still retained; SQL Server: the Agent and retention; MongoDB: the replica set) and prints the fix per line. Grants, wal_level, max_replication_slots and RDS retention are not probed — check them by hand.

1. The config: one CDC export, one recipe, one load:

source: { type: mysql, url_env: SOURCE_URL }
exports:
  - name: orders                       # the RECIPE: how to read the table
    table: orders
    mode: chunked
    chunk_by_key: id
    chunk_size: 250000
    chunk_checkpoint: true
    parallel: 4
    destination: { type: gcs, bucket: my-bucket, prefix: "exports/orders/" }

  - name: stream                       # the CDC export
    tables: [orders]
    mode: cdc
    cdc:
      checkpoint: ./cdc/stream.ckpt
      backfill: auto                   # baseline through the recipe, after the anchor
      until_current: true
    destination: { type: gcs, bucket: my-bucket, prefix: "exports/stream/" }

load: { target: bigquery, project: my-proj, dataset: my_ds, pk: auto }

• rivet init --source-env SOURCE_URL --mode cdc over two or more tables (MySQL, PostgreSQL public) scaffolds exactly this shape: one recipe per table — keyset where the table has a single-column keysettable key, range or full otherwise — and one tables: stream with backfill: auto. A single table, SQL Server, MongoDB, Oracle or a non-public schema get one stream per table with initial: snapshot instead. Pass --bigquery-project/--bigquery-dataset and init writes the load: block too (not for Oracle: its CDC does not load yet, ADR-0037).
• A stream over several tables rarely shares one partition column or one key. Put the per-table layer on the stream’s own load: block: load: { partition: { column: created_at, granularity: day }, tables: { customers: { partition: none }, line_items: { pk: [id, line_no] } } } — each table’s block overrides the export’s, which overrides the top-level load:.
• orders is a read recipe: a whole-config rivet run (and rivet apply) skips it — at warn — and the CDC export runs it after the anchor into its own exports/stream/orders/snapshot/. rivet run -e orders still exports it alone. It is never a load target.
• Only a full or chunked recipe is admitted; an incremental one reads a slice and is refused at config load, before any anchor exists.
• A column typed on the recipe (columns:) reaches the baseline, the stream and the recorded load spec alike — one type per column in one __changes.

2. Preflight

rivet check  --config cfg.yaml     # grades the CDC export as a log reader, not a scan
rivet doctor --config cfg.yaml     # binlog/slot/Agent/replica-set readiness, per line
rivet plan   --config cfg.yaml     # plans the batch exports; skips the stream and the recipe, saying so

Expect: no DEGRADED/UNSAFE on the CDC export; every doctor line green. plan skips the stream and the recipe, saying so — on the §1 config that leaves nothing to plan and it exits non-zero with “nothing to plan” (expected — not a fault in the config); with a plain batch export alongside it plans that one and exits 0.

3. Run 1 — anchor, then baseline — then load 1

rivet run  --config cfg.yaml
rivet load --config cfg.yaml

Order inside the run: anchor first (checkpoint pinned / slot created), then the baseline read through the recipe, then the drain of whatever changed during the baseline. A change landing mid-baseline is therefore in both — a duplicate, which the dedup view absorbs — never in neither.

Check:

-- source
SELECT COUNT(*) FROM orders;
-- warehouse (bq query --use_legacy_sql=false)
SELECT COUNT(*) FROM `my-proj.my_ds.orders`;

Both equal. The warehouse object is a plain table after run 1.

If the baseline is interrupted (a kill, a statement timeout on one chunk), just run again: a chunk_checkpoint: true recipe resumes its own leg on the next plain rivet run — no --export <leg> --resume, no synthesized names.

4. Changes → run 2 → load 2

Insert, update and delete a few rows at the source, then:

rivet run  --config cfg.yaml
rivet load --config cfg.yaml

Check:

• With a backfill: recipe the layout is base + buffer (load.layout: base_buffer, the default for this shape): orders is a physical table holding the baseline, and this cycle’s changes landed in the buffer orders__changes — exactly the number of changed rows (an update is one row, a delete is one row with __op = 'delete'). The base does not move until you compact:

  rivet compact --config cfg.yaml
  

  COMPACT OK merges the latest change per key into orders, flags a deleted key with __is_deleted = TRUE (its last values kept — the disappearance is data too) and drops the buffer. Live state is WHERE NOT __is_deleted:

  SELECT COUNT(*), COUNT(DISTINCT id) FROM `my-proj.my_ds.orders` WHERE NOT __is_deleted;
  

  Both equal the source’s COUNT(*); orders__changes is gone until the next load.

• Under load.layout: log_view (a capture-only stream, or written explicitly) there is no compaction: orders is a view over orders__changes, which keeps every change, and the same WHERE NOT __is_deleted reads live state.

• On Snowflake, base + buffer is the default (no layout: to write; load.layout: log_view keeps the log and view), and a CDC stream needs cdc.backfill: or cdc.initial: snapshot to build the base, or the config is refused at load. rivet compact runs one script on one warehouse resume: it renames orders__changes to orders__changes__merging, then in one transaction deletes the winning keys from orders, inserts their live rows and flags deleted keys, and drops the renamed table after the commit. A script that died before the drop is finished by the next compact. The buffer is a transient table with no time travel.

• A second rivet load with no new run appends nothing (the load ledger); a second rivet compact finds no buffer and says so.

• A table that holds no row at the baseline still gets its base: the load creates orders empty, with the source’s columns, and its first insert, update or delete is buffered and compacted like any other table’s.

• rivet compact works table by table. A table it fails or refuses does not stop the others: each prints its own COMPACT OK / COMPACT FAILED line and gets its own ledger row, and the command exits 1 when at least one table failed, naming it.

• On ClickHouse (preview) there is no compaction either: the change log is a ReplacingMergeTree that collapses versions per key by itself, and the view reads it with FINAL. The cycle is rivet run + rivet load; see the ClickHouse recipe.

A load never merges: the first baseline creates the base, later baselines and changes LOAD DATA INTO the buffer (or the changelog); only rivet compact rewrites base rows.

5. An interruption ON THE CDC LEG → run 3 → load 3

Make more changes, start rivet run, and kill it while it is draining (kill -9; the automated scenario injects RIVET_TEST_PANIC_AT=cdc_after_flush_before_ack and cdc_after_ack). Then simply:

rivet run     --config cfg.yaml
rivet load    --config cfg.yaml
rivet compact --config cfg.yaml   # base + buffer; a `log_view` stream skips this

Check: live state (WHERE NOT __is_deleted) equals the source, one row per key. orders__changes may hold a change twice if the kill landed after the part was flushed but before the checkpoint advanced — that is at-least-once, and compact (or the view, under log_view) collapses it. What must never happen is a change in neither.

6. Idle cycle

rivet run  --config cfg.yaml   # nothing changed
rivet load --config cfg.yaml   # "up to date"

Check: no orders__changes buffer was created (under log_view, the changelog did not grow); live state still equals the source.

Recovery orders that matter

• Log gone (slot invalidated, binlog purged — ERROR 1236, MSSQL below retention): re-baseline in ONE run, in the product’s own order — the run pins the anchor first, then re-reads the baseline. To make it do that: delete the checkpoint file if there is one, AND move every file out of the export’s destination (old parts still hold removed rows; every snapshot/_SUCCESS goes with them), AND delete the export’s cdc_snapshot rows, AND give the export cdc.initial: snapshot if it has none, AND truncate <table>__changes before the next load (a key deleted during the gap has no baseline row to supersede its older changes; the load refuses without the truncate); then re-run. The full list: re-baseline. Deleting the checkpoint alone is refused once a baseline completed (prior-run evidence exists), and on an export without a baseline once the state DB records that the export wrote the checkpoint on an earlier run (RIVET_SOURCE_CDC_CHECKPOINT_INVALID, exit 5): restore the file, or accept the gap with rivet state reset -c <config> -e <export>. Only an export the state DB has never seen anchors at the current position, with a warning.
• MySQL checkpoint used against another server: refused on purpose; same order on the new host.
• rivet validate --config cfg.yaml certifies both legs — the baseline under snapshot/ and the change parts — and never reports the baseline as stray.

Putting a table back into tables:

A table removed from tables: is no longer captured, so what changes in it afterwards is not in the destination. Adding it again is refused (RIVET_STATE_CDC_TABLE_REJOINED, exit 5, nothing read or written) while its old baseline is still recorded, because that baseline predates those changes. To put it back: move every file out of the table’s directory in the destination, delete its cdc_snapshot row from the state DB, truncate its __changes table if a warehouse load consumes the stream, and run again. That run reads the table in full and captures it from there. The same refusal answers a table: pointed at another table over a destination that already holds a completed baseline.

Running the automated scenario

docker compose --profile cdc up -d
cargo test --test live_suite full_cdc_cycle -- --ignored --test-threads=1

The project and bucket are dev/stand/registry.yaml’s (BIGQUERY_TEST_PROJECT / RIVET_TEST_GCS_BUCKET override them). Without a Google credential (gcloud auth login) the four tests skip, by name.