CDC Reference

Rivet’s change-data-capture (CDC) reads a source’s transaction log — not the tables — and emits each INSERT / UPDATE / DELETE as a row change. Because it tails the log that the database already writes for replication and durability, it adds almost no load to the OLTP path: no table scan, no locks, no read snapshot (see Why CDC is gentle on the source).

Status. All three SQL engines support NDJSON streaming and typed Parquet/CSV --output (real Timestamp / Date32 / Decimal128 columns). This page documents all three so the permissions are clear up front — they are the part operators most need to get right.

MongoDB also has CDC — via change streams, with a different setup (a replica set, not per-table grants) and the JSON-blob document image rather than typed columns. It has its own reference: mongodb.md.

The command

# stream changes as NDJSON to stdout (no schema resolution, fewest privileges)
rivet cdc --source 'mysql://rivet_cdc:***@127.0.0.1:3306/app' --table orders

# write typed Parquet files (one row per change, after-image / upsert shape)
rivet cdc --source 'mysql://rivet_cdc:***@127.0.0.1:3306/app' \
          --table orders --output ./cdc-out --format parquet \
          --checkpoint ./orders.ckpt --rollover 100000

Prefer --source-env VAR or --source-file path over an inline URL outside local dev — the URL is otherwise visible in ps / shell history.

The rivet cdc CLI is loopback-only: it carries no TLS configuration, so the TLS gate refuses any remote (non-loopback) host before connecting. For a remote source, use the config-driven rivet run path with a source.tls: block (see From config).

The engine is chosen from the URL scheme (mysql:// / postgresql:// / sqlserver:// / mongodb://) by create_change_stream, the CDC sibling of the batch create_source. With --output, each part goes through the same commit path the batch export uses (ADR-0004) and a manifest.json + _SUCCESS is written at clean end — but the CLI’s --output is a local directory only (it is wired to the local destination; a gs://…/s3://… string would be taken as a literal local path). For a cloud destination, use the config path (mode: cdc with a destination: block) below. Typed columns (real Timestamp / Date32 / Decimal128, not strings) flow through RivetValue structural typing — for all three engines (MySQL binlog values, PostgreSQL test_decoding parse, SQL Server change-table ColumnData).

From config (rivet run)

CDC also runs as an export in a config, so a scheduled rivet run captures changes alongside batch exports and records the run the same way:

source:
  type: mysql
  url_env: DATABASE_URL          # credentials out of the file
  tls: { mode: verify-full }     # required for a remote host (see below)
exports:
  - name: orders_cdc
    table: orders
    mode: cdc
    format: parquet
    cdc:
      checkpoint: /var/lib/rivet/orders.ckpt
      until_current: true        # drain to now and exit — for a scheduler
      # per-engine, all optional:
      server_id: 4271            # MySQL replica id
      slot: rivet_orders         # PostgreSQL logical slot
      capture_instance: dbo_orders  # SQL Server (required for sqlserver://)
    destination: { type: gcs, bucket: my-bucket, prefix: cdc/orders }

rivet run --config cdc.yaml      # captures, writes typed Parquet, records the run
rivet metrics -c cdc.yaml        # the CDC run appears with mode=cdc, like a batch

A mode: cdc export reuses the export’s table, destination, and format; the cdc: block carries only the CDC-specific knobs.

initial: snapshot — the safe switch, enforced by construction. On the first run (no anchor yet) rivet performs, in order: ① create the resume anchor (PostgreSQL slot / MySQL binlog checkpoint / SQL Server max-LSN checkpoint), ② run a full batch snapshot of each table into <destination>[/<table>]/snapshot/ (its own parts + manifest.json + _SUCCESS), ③ drain the change stream. Because the anchor predates the snapshot read, a change landing mid-snapshot appears in both the snapshot and the stream — an overlap the PK + __op dedupe absorbs, never a gap. Subsequent runs skip the snapshot because the state DB records it as done (cdc_snapshot — the authoritative signal; the snapshot/_SUCCESS marker remains a legacy co-signal, so re-snapshotting requires clearing BOTH) and go straight to draining; a run that crashes mid-snapshot re-snapshots on retry (the anchor stays put, so nothing is lost). Once any snapshot completed, a MISSING server-side anchor (a dropped PostgreSQL slot) is a loud error, never a silent re-anchor — see “A vanished slot” below. Load order downstream: the snapshot prefix as the base table, then MERGE the CDC parts. MySQL / SQL Server require cdc.checkpoint: with initial: snapshot (it is the anchor); PostgreSQL anchors in the slot.

  - name: orders_cdc
    table: orders
    mode: cdc
    format: parquet
    cdc: { initial: snapshot, checkpoint: /var/lib/rivet/orders.ckpt, until_current: true }
    destination: { type: gcs, bucket: my-bucket, prefix: cdc/orders }

cdc.backfill: — the baseline by reference, for tables that disagree. initial: snapshot synthesizes one single-connection mode: full scan per captured table. That is right for a small table and wrong for a large one — a 313M-row table read end to end on one statement runs into tuning.statement_timeout_s (300s under the balanced profile) long before it finishes, while the same table as a batch export, keyset-paged with parallel: 4, takes ~22 minutes. And a multiplex stream’s tables do not agree on how they are read: one has a unique id and keysets, another has only a non-unique index and must range-chunk.

So the baseline is declared by REFERENCE — each captured table names the ordinary batch export that already describes how to read it:

exports:
  - name: orders                 # an ordinary export; `rivet init` already writes it
    table: orders
    mode: chunked
    chunk_by_key: id             # keyset
    parallel: 4
    chunk_checkpoint: true       # the baseline is resumable
    format: parquet
    destination: { type: gcs, bucket: my-bucket, prefix: exports/orders/ }
    columns: { price: decimal(10,2) }

  - name: app_cdc
    tables: [orders]
    mode: cdc
    format: parquet
    cdc:
      checkpoint: /var/lib/rivet/app.ckpt
      backfill: auto             # or: [orders, …]
    destination: { type: gcs, bucket: my-bucket, prefix: cdc/ }

auto pairs each entry of tables: with the export whose table: names it; a list names them explicitly. One rivet run then does anchor → baseline → drain, and the ordering is what makes it safe: the anchor is taken before the first row is read, so a row changed mid-baseline also arrives on the stream and the current-state view keeps the higher (__pos, __seq).

What the leg borrows and what stays its own is the whole design. Borrowed: mode, chunk_by_key / chunk_column, page size, parallel, chunk_checkpoint, tuning and the column types. Its own: the name, the <destination>/<table>/snapshot/ prefix, the format and the meta columns — so the referenced export contributes a recipe, never a second load target, and every load invariant that holds for a synthesized leg holds unchanged here. Types MERGE (the recipe’s, with a qualified "table.column" key on the CDC export still winning); a column both sides declare differently is refused, because the two legs write into one <table>__changes and one column cannot have two types.

The run loop skips an export that is named as a backfill recipe, so a full rivet run reads each table once — rivet run -e orders still exports it on its own. An interrupted baseline resumes on the next plain rivet run from its chunk checkpoints (range-chunked and keyset legs alike, both live-proven against a crash after the first page — no --resume, no synthesized name; a leg whose recipe changed after the crash names rivet state reset-chunks -e <leg>) and leaves the anchor alone; once a table’s baseline is recorded (per table, in the state DB), later runs go straight to the drain. cdc.initial: and cdc.backfill: both describe the first run’s baseline, so config load refuses the pair.

Multiple CDC exports: each owns its stream resources. A PostgreSQL slot has ONE consumer (a shared slot is advanced past changes the other export never read), a MySQL server_id has ONE connection (the server kills the older one), and a checkpoint file has ONE writer. Config validation rejects two mode: cdc exports that resolve to the same slot / server_id / checkpoint — including the defaults (rivet_slot, 4271): a multi-table CDC config must set them explicitly per export:

exports:
  - name: orders_cdc
    table: orders
    mode: cdc
    cdc: { slot: rivet_orders, checkpoint: /var/lib/rivet/orders.ckpt }
    ...
  - name: users_cdc
    table: users
    mode: cdc
    cdc: { slot: rivet_users, checkpoint: /var/lib/rivet/users.ckpt }
    ...

Or multiplex: several tables through ONE stream (tables:). N single-table exports cost N slots — and PostgreSQL decodes the WAL once per slot (MySQL: N binlog connections). One export with tables: rides a single slot/connection and a single checkpoint, and routes each table’s changes to its own sub-prefix (<destination>/<table>/, each with its own parts + manifest.json + _SUCCESS — exactly like N exports, minus the N−1 slots):

exports:
  - name: app_cdc
    tables: [orders, users, payments]
    mode: cdc
    format: parquet
    cdc: { slot: rivet_app, checkpoint: /var/lib/rivet/app.ckpt, until_current: true }
    destination: { type: local, path: /data/cdc }   # → /data/cdc/orders/, /data/cdc/users/, …

The resume position is a property of the stream, so the at-least-once sequence generalises: every table’s buffered part is flushed before the one checkpoint/ack advances — a crash mid-roll re-reads for all tables rather than losing any one of them.

columns: overrides on a multi-table export support two key shapes: a bare column name applies to every captured table that has it, and a qualified "table.column" key targets one table and wins over the bare key there — so same-named columns needing different treatments never collide:

    columns:
      amount: "decimal(20,4)"          # every table's `amount`
      "legacy_orders.amount": text     # …except this one

A qualified key naming a table the export does not capture is a config error (a typo must fail at load, never silently miss its target).

Whole-database CDC across engines — and why the config shape differs. tables: multiplexing is PostgreSQL/MySQL only, and that is a property of the engine, not a rivet or driver limit. MySQL exposes one server-wide binlog and PostgreSQL one logical slot — a single stream carrying every table’s changes, so rivet reads it once and routes by table (and two exports sharing the slot/server_id would collide — exactly the scarce resource the one-stream form conserves). SQL Server has no such stream: CDC is enabled per table (sys.sp_cdc_enable_table) and read through a per-capture-instance function (cdc.fn_cdc_get_all_changes_<instance>) — there is no server-wide “all changes” surface to tap, so capture is inherently per-table. Use one export per table there, each with its own capture_instance; sharing a capture_instance between exports is safe (the change-table poll is read-only and resume state lives in the per-export checkpoint), and per-table exports never collide (no slot / server_id — the change tables are populated by one shared capture Agent regardless of how many readers).

The operator flow is identical across the three SQL engines, only the config shape differs (MongoDB’s whole-database change stream uses the same mode: cdc config shape — see mongodb.md):

• rivet init --mode cdc (no --table) scaffolds the whole database on every engine — one tables: export on MySQL (and on PostgreSQL when every table is in the public schema; mixed schemas fall back to per-table exports), one export per table (distinct capture_instance) on SQL Server — so you never hand-list tables. Over two or more tables on MySQL/PostgreSQL the stream gets backfill: auto and one batch recipe per table (the baseline read), and its load: may carry tables: { <table>: { pk, partition, cluster_by, … } } so each captured table has its own warehouse shape.
• rivet run -c <config> drains the whole set; add --parallel-export-processes to run SQL Server’s per-table exports concurrently.
• rivet validate -c <config> descends into every table’s prefix and its initial: snapshot sub-dataset on every engine, so one command certifies the whole stream.

The per-table vs one-stream split is connection/resource topology, not throughput or memory: drain RSS is O(part rollover) per stream on all three engines, independent of table count. Each run produces the standard per-export summary block and an export_metrics row (rows / files / bytes / duration / status), so CDC shows up in rivet metrics and the run aggregate exactly like a batch export. TLS: unlike the CLI (which is loopback-only), the config path passes source.tls to the change stream — so a remote source over TLS requires the tls: block, and a remote host without it is refused before any connection (the same gate the batch path uses).

The four models

Rivet normalises four different source mechanisms behind one ChangeStream:

MySQL and PostgreSQL expose the log to the client; SQL Server does not — there a server-side Agent extracts the log into change tables that rivet polls.

Permissions & prerequisites

MySQL — the binlog grants

Rivet registers as a replica and streams the binlog. Least privilege:

CREATE USER 'rivet_cdc'@'%' IDENTIFIED BY '***';

-- read the binlog stream (register as replica, COM_BINLOG_DUMP).
-- Server-wide: REPLICATION SLAVE cannot be scoped to a database/table.
GRANT REPLICATION SLAVE  ON *.* TO 'rivet_cdc'@'%';

-- read the current binlog coordinate (SHOW MASTER STATUS) when starting
-- without a checkpoint.
GRANT REPLICATION CLIENT ON *.* TO 'rivet_cdc'@'%';

-- ONLY for `--output`: rivet resolves the table's column types with
-- `SELECT * FROM <table> LIMIT 0` (metadata only, no rows). Not needed for NDJSON.
GRANT SELECT ON `app`.`orders` TO 'rivet_cdc'@'%';

FLUSH PRIVILEGES;

Server configuration (my.cnf [mysqld], or SET GLOBAL + restart where allowed):

log_bin           = ON       # binary logging on (often already on for replication/PITR)
binlog_format     = ROW      # rivet needs row images, not statements — MIXED/STATEMENT will not work
binlog_row_image  = FULL     # full before/after image — REQUIRED for the after-image / MERGE shape;
                             # MINIMAL drops unchanged columns and breaks "overwrite all columns"
binlog_row_metadata = FULL   # REQUIRED (8.0.1+): column names in every TABLE_MAP; rivet refuses MINIMAL
server_id         = 1        # any unique id for the source; rivet uses a DIFFERENT --server-id

Notes:

• REPLICATION SLAVE is server-wide by design. You cannot grant binlog access for one database only — the binlog is a single server-wide stream. Scope data exposure with --table (rivet filters client-side) and the SELECT grant.

• A stale server_id collision silently kills the stream. Give rivet a --server-id no other replica uses.

• Connect CDC directly to MySQL — not through ProxySQL / MaxScale. The binlog stream is COM_BINLOG_DUMP, a replication protocol query proxies don’t carry; the batch path can go through a pooler, CDC cannot. Rivet probes the connection and fails fast with this exact reason if it sees a proxy, so point the source at the MySQL host (the replication endpoint), not the proxy port.

• binlog_row_metadata = FULL is required, and it is NOT MySQL’s default. Under MINIMAL the binlog carries no column names, so values could only be mapped by position, and after a column reorder or ALTER they land under the wrong names. rivet refuses instead: the run fails at open with RIVET_SOURCE_CDC_PREREQUISITE, and rivet doctor fails the “CDC binlog server config” check. Fix it with SET PERSIST binlog_row_metadata = FULL; (needs SYSTEM_VARIABLES_ADMIN or SUPER), or binlog_row_metadata = FULL under [mysqld] and a restart. Events already written under MINIMAL stay nameless after the switch: a run that reaches one of a captured table fails with RIVET_SOURCE_CDC_UNDECODABLE, naming the table and the binlog position, and writes no part. If the export already had a checkpoint, re-baseline the stream after the switch. MySQL before 8.0.1 has no such variable and is refused for the same reason.

• binlog_row_image = FULL is MySQL’s default; the risk is a source that has set it to MINIMAL to shrink the binlog — that path needs the column-mask MERGE, not the simple overwrite (see Output shape).

• Amazon RDS / Aurora MySQL: two managed-only settings, and neither is in my.cnf. Both were diagnosed the hard way on a customer replica, a day apart.

  1. Binary logging follows automated backups. With backup retention at 0 the instance runs log_bin = 0 no matter what the parameter group says, and SHOW BINARY LOGS answers ERROR 1381 (HY000): You are not using binary logging. Set backup retention above zero (this restarts the instance), then binlog_format = ROW, binlog_row_image = FULL and binlog_row_metadata = FULL in the parameter group. A read replica also needs log_replica_updates = 1 to re-log what it applies.

  2. binlog_expire_logs_seconds does not govern retention here. RDS purges a binlog as soon as the engine itself no longer needs it — typically within minutes — so a checkpoint written by one run is unreadable by the next and the resume fails with ERROR 1236. Measured: a checkpoint taken at 13:42 was already past retention at 13:59. Set the managed knob instead, sized well above the CDC cadence:

     CALL mysql.rds_set_configuration('binlog retention hours', 72);
     CALL mysql.rds_show_configuration;   -- confirm
     

  The filenames are the tell: mysql-bin-changelog.NNNNNN is RDS’s naming, so an ERROR 1236 naming one of those is this, not PURGE BINARY LOGS.

MySQL — column grants (source.columns_from: grants)

The binlog carries every column of a row. Under source.columns_from: grants rivet reads and delivers only the columns the account may SELECT, in every mode. Batch reads are enforced by the server (the SELECT names only readable columns). The change stream is a client-side projection: the binlog carries every column of a changed row to rivet, which drops the unreadable ones before anything is written. That is data minimisation, not access control. rivet check prints this boundary and a change-stream run logs it at start.

GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'rivet_cdc'@'%';
GRANT SELECT (id, name, email) ON `app`.`users` TO 'rivet_cdc'@'%';  -- not `pass`

source:
  type: mysql
  url_env: DATABASE_URL
  columns_from: grants

Limits:

• MySQL only. On PostgreSQL, SQL Server, Oracle and MongoDB every command that loads the config (run, check, plan, doctor, validate, load, state) and apply of a plan that carries it refuses it with RIVET_CONFIG_SOURCE_MODE_UNSUPPORTED (exit 1, nothing connected or written), because those sources would read and deliver every column (see #711). Remove columns_from: from source: or point source: at a MySQL server.
• The binlog still leaves the server in full: the filter is in rivet, not in MySQL.
• A grant change not written to the binlog (sql_log_bin = 0, a direct mysql.* edit without FLUSH PRIVILEGES) takes effect on the next run only.

PostgreSQL — the logical slot

Rivet’s PostgreSQL reader consumes a logical slot through the normal SQL connection (pg_logical_slot_peek_changes()), not the streaming-replication protocol. That changes what you must grant:

-- REPLICATION attribute: required to create and read a logical slot, even via
-- the SQL functions (pg_create_logical_replication_slot / _get_changes).
ALTER ROLE rivet_cdc WITH LOGIN REPLICATION PASSWORD '***';

-- ONLY for `--output`: schema resolution (SELECT ... LIMIT 0).
GRANT SELECT ON app.orders TO rivet_cdc;

Server configuration (postgresql.conf, needs a restart):

wal_level            = logical   # log enough to decode row changes (default is 'replica')
max_replication_slots = 10       # >= 1 (defaults are usually fine)
max_wal_senders       = 10       # >= 1

Notes:

• No pg_hba.conf replication line is required. That entry is for the streaming walsender protocol; rivet’s poll model uses an ordinary connection, so the normal host app rivet_cdc ... rule suffices. (This is the main way the poll model is operationally lighter than streaming CDC tools.)
• A logical slot pins WAL until it is consumed/advanced. An abandoned slot prevents WAL recycling and fills the disk — the number-one PostgreSQL CDC foot-gun. Drop unused slots with SELECT pg_drop_replication_slot('rivet_slot');.
• wal2json / pgoutput are alternatives to test_decoding; test_decoding is always built in and needs no extension.

SQL Server — CDC change tables

SQL Server has no client-streamable log. A server-side Agent job extracts the log into cdc.* change tables, which rivet polls. Two distinct privilege levels:

-- ONE-TIME ENABLE (requires sysadmin or db_owner):
EXEC sys.sp_cdc_enable_db;                         -- creates the cdc schema + capture job
EXEC sys.sp_cdc_enable_table
     @source_schema = N'dbo', @source_name = N'orders',
     @role_name = N'cdc_reader',                   -- gating role for readers (or NULL = no gate)
     @capture_instance = N'dbo_orders',
     @supports_net_changes = 0;

-- RUNTIME READER (what rivet connects as — least privilege):
CREATE USER rivet_cdc FOR LOGIN rivet_cdc;
GRANT SELECT ON SCHEMA::cdc TO rivet_cdc;          -- read the change tables + functions
ALTER ROLE cdc_reader ADD MEMBER rivet_cdc;        -- if a gating role was set above

Notes:

• SQL Server Agent must be running. The capture job (default ~5 s scan cycle) is what populates the change tables. If the Agent stops, the change tables silently freeze and the transaction log can’t truncate — disk pressure. A production reader should watch for a non-advancing sys.fn_cdc_get_max_lsn(), not read “no rows” as “no changes”.
• Edition gate: CDC is on Enterprise / Standard / Developer — not Express or Web. On Express, use Change Tracking instead (different, lighter, but only tells you which rows changed, not the data).
• Enabling CDC needs sysadmin/db_owner; the runtime reader needs only the SELECT grant above.
• Retention: the cleanup job keeps ~3 days by default. If rivet is offline longer than retention, the saved LSN falls below sys.fn_cdc_get_min_lsn() and the read errors — re-baseline the stream.

Oracle — LogMiner (preview)

Oracle CDC reads the redo logs through LogMiner, which ships with every edition (Free included) and needs no GoldenGate licence. rivet never sets ENABLE_GOLDENGATE_REPLICATION (that one does need the licence).

-- ONCE, as SYSDBA in CDB$ROOT:
SHUTDOWN IMMEDIATE; STARTUP MOUNT; ALTER DATABASE ARCHIVELOG; ALTER DATABASE OPEN;
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;              -- minimal logging
CREATE USER c##rivetcdc IDENTIFIED BY … CONTAINER = ALL;
GRANT CREATE SESSION, SET CONTAINER, LOGMINING TO c##rivetcdc CONTAINER = ALL;
GRANT EXECUTE_CATALOG_ROLE TO c##rivetcdc CONTAINER = ALL;
GRANT SELECT ON v_$database TO c##rivetcdc CONTAINER = ALL;   -- and the same for
--   v_$archived_log, v_$log, v_$logfile, v_$logmnr_contents, v_$logmnr_logs, v_$transaction

-- PER CAPTURED TABLE (its owner or a DBA), in the pluggable database:
ALTER TABLE app.orders ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
GRANT SELECT ON app.orders TO c##rivetcdc;

Notes:

• The URL names the pluggable database (oracle://c%23%23rivetcdc:…@host:1521/ORCLPDB1 — # percent-encoded). rivet switches the session to CDB$ROOT to mine and keeps only that PDB’s changes. A non-CDB works without the switch.
• ALL COLUMNS logging, not PRIMARY KEY. With key logging an UPDATE’s redo carries only the changed columns, so the change cannot represent the row; rivet refuses such a table and prints the statement above.
• Tables the preview refuses by name: a column of type LOB, LONG, XMLTYPE, JSON, INTERVAL, BOOLEAN, VECTOR, ROWID or an object type; a name over 30 bytes; and, before 23ai, an identity column (LogMiner ignores those tables entirely).
• Retention is the DBA’s, as with the binlog: nothing pins archived logs for rivet. If the checkpoint needs a log that was deleted, the run fails with a data-loss error (see Failure modes).
• Loading an Oracle stream with rivet load is not supported in the preview: a mode: cdc Oracle export under a load: block is refused when the config is read.

Reading from a replica (no primary access)

A common real-world constraint: you’re handed a database but only a read replica, never the master. Rivet reads the log of whatever host you point source.url at — it never needs the primary specifically. Whether a replica can serve that log is an engine + replica-config question, not a rivet limitation:

MySQL caveat — the checkpoint is replica-local. Rivet resumes by binlog {file, pos} (not GTID), and a replica’s binlog coordinates are its own, not the primary’s. A checkpoint taken against one replica does not transfer to another host, and rivet refuses one written by a different server (it records server_uuid). If you fail over (to a different replica, or to the primary), re-baseline the stream on the new host.

MySQL — a RESET MASTER does not pass for the same binlog. The checkpoint also records the commit it ends at (where that event starts and its xid or COMMIT text), and a resume reads that event back with SHOW BINLOG EVENTS. A binlog that RESET MASTER renumbered back to the checkpoint’s file name holds other events there, so rivet refuses to resume (RIVET_SOURCE_CDC_FOREIGN_CHECKPOINT, exit 5) instead of reading from an offset in another file, with GTID on or off. Recover with a re-baseline. A checkpoint written before rivet recorded the commit resumes with a warning.

SQL Server — the checkpoint follows a failover, and nothing else. An availability group’s replicas share one log, so a checkpoint written on the primary resumes on a secondary (verified live). The checkpoint records the database’s family_guid and recovery_fork_guid, and rivet refuses to resume against a database whose family_guid differs (another server’s database: its LSNs address a different log) or whose recovery_fork_guid changed (a RESTORE rewound the log). Recover with a re-baseline: the run anchors first and re-reads the table after. A checkpoint written before rivet recorded the identity resumes with a warning.

PostgreSQL — the slot does not survive a failover. The slot is the resume position, and it lives on the server it was created on, so a promoted standby has none. With cdc.checkpoint: set, rivet refuses to continue there (slot … is missing but prior-run evidence exists) and writes nothing. Without a checkpoint it creates a new slot with a warning, and the changes written between the failover and that run are lost — set cdc.checkpoint: on any stream that may fail over. Following a failover (failover slots, PostgreSQL 17+) is not implemented.

MongoDB — the stream follows a member switch. The resume token lives in the replica set’s shared oplog, so pointing the same config at another member continues the stream with every change captured once (verified live; a primary election itself is not yet exercised).

Oracle — not measured. A checkpoint from another database is refused; what a Data Guard failover does to it is untested.

So the answer to “can I read the log from a slave?” is yes on all four engines, each verified live: MySQL (with log_replica_updates = ON), PostgreSQL 16+ in continuous mode, a SQL Server readable secondary, and a MongoDB secondary. Point source.url at the replica; everything else (grants, mode: cdc, output) is identical to running against a primary.

Output shape

--output writes one row per change in the typed after-image (upsert) shape:

__op     __pos                              __seq  id   name    amount
insert   {"file":"binlog.000046","pos":681} 0      1    alice   100
update   {"file":"binlog.000046","pos":682} 0      1    alice   150
delete   {"file":"binlog.000046","pos":683} 0      2    bob     200

• __op — insert / update / delete.
• __pos — the transaction’s commit position (the same value rivet checkpoints). Every change of one transaction shares it — it is not a total order over changes.
• __seq — the change’s ordinal within its transaction (0-based, log order), so (__pos, __seq) is a total order. It is the tiebreak when one transaction touches a key more than once and, being a column, survives the load into an (unordered) warehouse table (Parquet row order does not). See CDC change ordering. (MongoDB gives every event a distinct __pos, so its __seq is always 0.)
• the source columns, typed (resolved from the source schema), carrying the after-image for insert/update and the key (before-image) for delete.

An UPDATE that changes the key is written as two rows: a delete of the old key, then an insert of the new row, at the same __pos with the insert’s __seq after the delete’s (ADR-0030). So the MERGE below retracts the old key with no extra step. The key is the export’s declared load.pk: when there is one, otherwise the table’s primary key, read from the source at the start of the run. pk: none means no key, so nothing is split. A declared key column the table does not have fails the run at the start, naming the column. SQL Server’s change table already records an UPDATE of the primary key as a delete and an insert. This was measured with a clustered key and with a nonclustered key on a heap; rivet reads only the new row of any other UPDATE. MongoDB’s _id cannot change.

An UPDATE counts as a key change only when its old image carries EVERY key column and one of them differs. A column the old image does not carry is absent, not NULL. On PostgreSQL the old image is what the replica identity logs: the whole row under REPLICA IDENTITY FULL, and otherwise only the identity’s columns, and only when one of them changes. So an UPDATE that changes a key column the identity does not log stays one update, and its old key stays live downstream. This happens under REPLICA IDENTITY USING INDEX on an index that is not the key, and with a declared load.pk: that is not the primary key. Use REPLICA IDENTITY FULL for such a table.

A statement that renumbers keys. Oracle checks uniqueness per statement, so UPDATE t SET id = id + 1 is logged as 1 -> 2, then 2 -> 3. The delete of 2 belongs to the row that held 2 before the statement, not to the row that just moved into it, and rivet orders it before that row’s insert. It tells the two rows apart by LogMiner’s ROW_ID, so rows whose other columns are equal (a junction table) keep every row. On a heap table without row movement, a change whose ROW_ID is missing is refused (RIVET_SOURCE_CDC_UNDECODABLE) rather than guessed.

What this does not handle:

• An Oracle table whose ROWID can change inside a transaction. This is an index-organized table (LogMiner gives every row of one the same placeholder ROW_ID) or a table with ENABLE ROW MOVEMENT (an UPDATE that moves a row to another partition reports the row’s old ROW_ID, and the row’s next change its new one). There rivet compares images instead, and a renumber over rows whose non-key columns are all equal loses a row. The same applies to PostgreSQL with a DEFERRABLE primary key under REPLICA IDENTITY FULL, which carries no row identity.
• A key change the old image cannot show stays one update, and its old key stays live. Nothing warns about these cases:
  • PostgreSQL under REPLICA IDENTITY FULL, with a declared load.pk: column that goes from NULL to a value. test_decoding does not print the NULL old cell, so the old image does not carry that key column.
  • MySQL with a session-level binlog_row_image = NOBLOB (the global setting is refused at the start), when one UPDATE changes both the key and a BLOB/TEXT column. The old image then lacks the unchanged BLOBs and cannot be read by name.
  • SQL Server with a declared load.pk: that is not the primary key. An UPDATE of that column is an update in place, and rivet reads only the new row.
• A PostgreSQL table whose changes carry no old image. This covers four cases:
  • a table with REPLICA IDENTITY NOTHING;
  • REPLICA IDENTITY USING INDEX on an index that has since been dropped, which PostgreSQL treats as NOTHING;
  • a table without a primary key under REPLICA IDENTITY DEFAULT;
  • a table whose primary key is DEFERRABLE (PostgreSQL does not use a deferrable key as the replica identity). An UPDATE on such a table carries no old values, and a DELETE carries no columns at all. So a key change stays one update and the old key stays live, and a DELETE retracts nothing downstream. The run warns at the start, naming each such table schema-qualified, with its remedy: REPLICA IDENTITY FULL; for a deferrable key, also REPLICA IDENTITY USING INDEX on a non-deferrable unique index (which then rejects SET id = id + 1); for a dropped index, an existing one; for no key, a primary key.

An old-image cell rivet cannot decode (infinity, a BC date, 24:00) refuses the run only where the old image is written: the delete of a key change, and the before of rivet cdc NDJSON. An UPDATE that keeps its key is delivered as before.

Downstream applies it by primary key:

MERGE target t USING staged s ON t.id = s.id
WHEN MATCHED AND s.__op = 'delete' THEN DELETE
WHEN MATCHED                       THEN UPDATE SET t.* = s.*   -- overwrite all columns
WHEN NOT MATCHED AND s.__op <> 'delete' THEN INSERT (...);

With a full row image, which columns changed is irrelevant — the latest image per key already contains every prior change, so dedup-by-key + overwrite is correct. “Latest” is the highest (__pos, __seq) — the commit position, then the intra-transaction ordinal (a transaction that updates one key twice shares __pos, so __seq breaks the tie). This is why binlog_row_image = FULL matters.

Deduplicating by position, per engine

__pos granularity differs by engine, and the dedup recipe follows from it:

Two distinct problems:

1. At-least-once re-delivery (a crashed run’s part re-read on resume): the re-delivered event is byte-identical — same __op, __pos, and image — so SELECT DISTINCT over the staged rows (or dedup on (pk, __pos, __op)) removes it exactly.
2. Latest-image-per-key (the MERGE): order by (__pos, __seq) — the commit position, then the intra-transaction ordinal. __seq breaks the tie when a transaction touches one key more than once (those changes share __pos); it is a column, so — unlike Parquet row order — it survives the load. See CDC change ordering.

-- DuckDB replay: newest surviving image per key.
WITH ev AS (
  SELECT *,
         upper(lpad(split_part(__pos->>'lsn', '/', 1), 8, '0')) ||
         upper(lpad(split_part(__pos->>'lsn', '/', 2), 8, '0')) AS lsn_key   -- PostgreSQL X/Y → sortable
  FROM read_parquet('…/sessions/cdc-*.parquet')
), latest AS (
  SELECT * FROM (
    SELECT *, row_number() OVER (
      PARTITION BY id
      ORDER BY lsn_key DESC, __seq DESC) AS rn        -- (__pos, __seq) = total order
    FROM ev)
  WHERE rn = 1
)
SELECT * FROM latest WHERE __op <> 'delete';

(MySQL: order by (file, pos) parsed from __pos, then __seq; SQL Server: the fixed-width hex lsn string is already lexically ordered, then __seq.) In a warehouse MERGE, apply the same window to the staged batch first, then merge the winners by PK + __op.

Downstream loading

CDC output is the same typed Parquet the batch export writes (same build_arrow_field pipeline), so the warehouse-loading recipes apply unchanged — the engine-specific MERGE and the JSON-as-BYTES / naive-timestamp autoload recovery are in recipes/idempotent-warehouse-load.md (BigQuery) and recipes/snowflake-load.md, keyed on the PK + __op above.

Verified cross-engine on a CDC part: DuckDB reads json natively, ClickHouse as String (JSONExtract* parses it), BigQuery as BYTES (PARSE_JSON after SAFE_CONVERT_BYTES_TO_STRING); integers keep their width (INT32/INT64) and timestamps their microseconds. The JSON text round-trips losslessly in all three — the type that auto-detects differs, the data does not.

Part naming. Parts are run-stamped — cdc-<run_id>-000000.parquet — so a scheduler re-running into the same prefix appends each cycle’s parts alongside the previous cycle’s (nothing is overwritten). manifest.json / _SUCCESS describe the latest run only; a glob reader over the prefix sees the union of all cycles, which is the intended at-least-once stream — dedupe by PK + __op + __pos downstream, and archive parts you have already loaded if you want the prefix to stay small.

Without --output, rivet emits the changes as NDJSON (one JSON object per change) to stdout, as the engine delivered them: op, schema, table, before, after, pos, seq. NDJSON is NOT split: a key change is one update line. When a drain to stdout ends it acknowledges the last commit it printed, so on PostgreSQL the slot advances and the next drain starts after that commit. A drain that stops before its end acknowledges nothing and the next one prints the same lines again: a consumer of the pipe must tolerate a repeated line. before holds the old image’s cells in the order the engine logged them. On PostgreSQL every UPDATE that carries an old image also gets before_columns, which names those cells. That is the replica identity’s columns, or under REPLICA IDENTITY FULL the whole old row minus its NULL cells, which test_decoding does not print. A column before_columns does not name was either not logged or NULL.

Why CDC is gentle on the source

The log is written anyway (WAL for durability, binlog for replication/PITR), so on MySQL/PostgreSQL CDC mostly reads what already exists — near-zero incremental OLTP cost. The one real CDC cost is disk via log retention if the consumer lags (PG slots pin WAL; MySQL keeps binlog until read). SQL Server is the exception: its Agent writes changes into change tables (extra write volume + storage), so CDC there trades read-contention for an ongoing write/storage overhead.

Failure modes & recovery

Every CDC run is bounded and resumable, and the durable sequence is flush → checkpoint → ack: the resume position only advances after the part is durably written. So on any failure — a dropped connection, a query error, a full source disk — the run fails loudly (non-zero exit, with the per-engine setup hint), the checkpoint/slot is not advanced, and the next run re-reads from the last good position. Rivet never silently loses a change; the trade-off is at-least-once, so a failed run’s already-uploaded parts can reappear — dedupe downstream by primary key + __op (the output is the upsert / after-image shape).

A failed run leaves its durable parts in the destination but no manifest.json / _SUCCESS — that pair marks a clean end, so a missing _SUCCESS is how you (and rivet validate) tell a partial run from a complete one.

PostgreSQL — the slot fills / the source disk fills

A logical slot pins WAL until rivet advances it (confirmed_flush_lsn). The behaviour depends on whether rivet is running:

• Running + advancing — each successful run reads the changes, writes them durably, then advances the slot, so PostgreSQL releases the WAL up to that point. The slot only ever holds the WAL since the last advance — it does not grow unbounded while rivet keeps the slot moving.
• Stopped (abandoned slot) — rivet does nothing (it isn’t running); the slot keeps pinning WAL and the source disk fills. This is the number-one PostgreSQL CDC foot-gun, and it is operator responsibility: SELECT pg_drop_replication_slot('rivet_slot'); when you stop capturing for good.
• Source disk already full — run rivet (it reads WAL to advance the slot, which releases WAL and relieves the pressure) or drop the slot. If PostgreSQL is too degraded to answer, rivet’s query fails → the run fails → re-read next run.

Memory is O(largest transaction). The adapters buffer a whole transaction until its COMMIT (parts never split a transaction — the resume invariant; SQL Server buffers per poll batch). Measured on MySQL: ~1.4 KB of RSS per buffered row (~14× a 100-byte payload): a 100k-row transaction drains at ~170 MB RSS, 300k at ~440 MB — linear. A transaction past the hard caps (5M buffered rows or 2 GiB estimated bytes by default; RIVET_CDC_MAX_TX_ROWS / RIVET_CDC_MAX_TX_BYTES override them) fails the run LOUDLY before it can OOM. Opt-in RIVET_CDC_SPILL_DIR spills the adapter’s copy past the cap to disk (PostgreSQL, MySQL, SQL Server; Oracle always refuses at the cap), but the sink still holds the whole transaction, so it saves only ~11% of peak RSS — see CDC failure modes. Run bulk backfills in batched transactions, or through mode: full/initial: snapshot (the batch path streams).

DDL inside a capture window: safe where the engine names its columns, a LOUD ERROR where it does not. PostgreSQL (wire text) and SQL Server (change tables) always name every image column, so rivet maps values by NAME: a DROP COLUMN or RENAME landing between runs captures correctly, and an equal-arity DROP a + ADD c leaves c NULL for the older images rather than filling it with a neighbour’s value (unless the dropped column sat at c’s position, which looks exactly like a rename and is read as one). A column ADDED while a run is open (a continuous run, until_current: false) shows up in the first change that carries it: the run closes the table’s part under the old schema, reads the table’s columns again, judges the change by on_schema_drift (fail stops the run there, before the change is acknowledged), and continues in a new part with the new column, so none of its values is dropped. One run’s parts then carry two schemas; rivet load takes both into one change log (a part without the column reads it as NULL). A bounded run (until_current: true, the default) resolves its schema after its read bound, so it never meets such a change. SQL Server is the exception: a capture instance carries the columns it was created with, so a column added after it never reaches the change table; a run that sees its table outgrow the instance stops with RIVET_SOURCE_CDC_PREREQUISITE, the refusal a run opening on that instance meets, instead of ending clean — re-enable the table with the full column list, then re-baseline it. MySQL’s binlog carries names only when the server runs with binlog_row_metadata=FULL (8.0.1+), which rivet requires (see MySQL — the binlog grants), so MySQL behaves the same. rivet never maps a binlog image by position: a server at MINIMAL is refused at open, and an event written under MINIMAL is refused when it is read. DDL between runs is always fine, because each run resolves the schema fresh. Same-arity TYPE changes remain undetectable without schema history (roadmap): run type migrations and their backfills through a re-baseline.

The value checksum runs on CDC too. The same always-on two-ended check the batch export performs — an independent fold of the decoded cells vs a fold of the built Arrow column — runs per column before every CDC part is written; a mismatch fails the run naming the column, never writes the corrupted part. Failure behaviour is also parity: a value unrepresentable in the declared column (PostgreSQL 'NaN'::numeric in a Parquet decimal) fails loudly on both paths, never a silent NULL.

For the full operational failure playbook — every symptom, what rivet does, how to recover, how to prevent — see cdc-failure-modes.md.

A vanished slot is a loud error, not a silent restart. When a resume checkpoint exists but the slot is gone (dropped by an operator, or invalidated and removed), rivet refuses to re-create it — a fresh slot would anchor at the current position and silently skip everything since the drop. The run fails with the re-baseline steps; delete the checkpoint file alone only when you explicitly accept a fresh anchor and the gap. Without a checkpoint or a completed baseline, rivet cannot tell a dropped slot from a first run: it creates the slot and warns with the same steps.

Bound the blast radius: set max_slot_wal_keep_size (PG 13+). PostgreSQL then invalidates the slot rather than fill the disk; rivet’s next run fails with a slot-invalidated error and you re-baseline. Monitor pg_replication_slots (active, and restart_lsn vs the current LSN = how much WAL the slot is holding).

rivet doctor automates this monitoring. For a config with mode: cdc exports, doctor probes the engine: PostgreSQL — the export’s slot (retained WAL, fails above 1 GiB) and any other inactive slot pinning WAL (the abandoned-slot foot-gun); MySQL — log_bin/binlog_format=ROW/ binlog_row_image=FULL, and whether the checkpoint’s binlog file is still retained (a purged file is reported before the run fails with ERROR 1236); SQL Server — CDC enabled, the capture instance exists, the checkpoint is within retention, and the Agent service is running. On PostgreSQL, MySQL and Oracle doctor also asks the question the run refuses by, through the same function: a prerequisite the run would refuse (PostgreSQL wal_level, a bounded drain on a standby, a table that does not exist; MySQL binlog_row_metadata, binlog compression, a replica that does not re-log) is a failed CDC run prerequisites check carrying the run’s own message.

MySQL — the binlog was purged

If rivet is offline long enough that the saved binlog position is purged (binlog_expire_logs_seconds / PURGE BINARY LOGS), the resume read fails with MySQL ERROR 1236 (the requested binlog file is gone). The position is unrecoverable — re-baseline the stream. Size binlog retention comfortably above your CDC cadence.

SQL Server — the checkpoint fell below retention

If the saved LSN falls below sys.fn_cdc_get_min_lsn() (the cleanup job — ~3 days by default — removed the changes after it, or the capture instance was disabled and re-enabled), rivet fails loudly (exit 5, RIVET_SOURCE_CDC_LOG_GAP), from the first run after the baseline — “the resume position is older than the SQL Server CDC change-table retention … Re-baseline the stream” — rather than resume from the new min and silently skip the gap. re-baseline the stream. Also watch for a non-advancing sys.fn_cdc_get_max_lsn(): that means the Agent capture job stopped, so the change tables are frozen — read “no rows” as “the job is down”, not “no changes”.

Oracle — the archived logs were deleted

If the checkpoint needs redo older than the oldest archived log still listed (RMAN DELETE INPUT, a retention policy), or a log sequence is missing in between, the run fails with “… LOST to this stream” instead of mining from whatever remains. Restore the archived log, or re-baseline the stream.

A checkpoint written against another database (a different DBID, a RESETLOGS since, or another pluggable database) is refused the same way: an SCN means nothing outside the database that issued it.

Copying a CDC config (staging beside production)

A CDC stream’s position belongs to one reader. A copy of a CDC config that reads the same cluster needs its own stream objects and its own state, or the two configs consume each other’s changes:

• PostgreSQL: its own slot: (rivet init --mode cdc -o <file> names it after the file). The slot is consume-on-read: a second config reading the same slot takes the changes the first one has not read yet. With a checkpoint:, rivet records the last position it confirmed before it advances the slot, and a run whose slot another reader moved past that position is refused, RIVET_SOURCE_CDC_LOG_GAP (exit 5, “another reader consumed the changes between them”): the changes are gone from this stream, so re-baseline it as the text says. Without a checkpoint: nothing is compared.
• MySQL: its own server_id: (two readers with one id evict each other’s binlog connection) and its own checkpoint:; the binlog itself is not consumed by a reader, so a second config loses nothing from it.
• Every engine: its own checkpoint: path and its own state (a state DB of its own, or a config directory of its own for the SQLite file).

Recovery, in one line

Re-run to resume from the last checkpoint (the common case). If the run reports the position is unrecoverable (PostgreSQL slot invalidated, MySQL binlog purged, SQL Server retention exceeded), re-baseline the stream: delete the checkpoint, move every file out of the export’s destination, delete its cdc_snapshot rows, give it cdc.initial: snapshot if it has none, and re-run. That run anchors first and re-reads the table after, so the stream overlaps the snapshot (duplicates, which the load deduplicates) instead of leaving the changes in between in neither. A separate mode: full export is not a re-baseline.

Limitations (current)

Typed output (real Timestamp/Date32/Decimal128), commit-boundary checkpointing, cloud destinations + manifest.json/_SUCCESS, and the config-driven rivet run path with a recorded run are all in place for all three engines. What remains:

• Continuous capture is bounded-poll-and-exit by default on every engine (they drain their backlog and stop). The supported continuous model is a scheduler running the default bounded rivet cdc (or rivet run with cdc.until_current: true, now the default) on an interval, each run resuming from the checkpoint. For an unbounded run, pass --stream (config: cdc.until_current: false; the config-driven rivet run path logs an engine-specific warning), because only MySQL (the binlog dump blocks) and MongoDB (the change stream blocks awaiting events, ending only if the stream is invalidated or closed) genuinely stay up as daemons; PostgreSQL and SQL Server still exit on catch-up (one unbounded pass — run it under a supervisor that restarts it). The bounded run remains the intended model.
• Schema drift: the sink schema is frozen at the first flush — a column added mid-run is not picked up until the next run re-resolves the table, and its values captured in the meantime are dropped (the events are still acked) — re-baseline the stream to recover them.
• No lag metric: the run records rows / files / bytes / duration / status, but not replication lag (“how far behind the source is”) — the next observability step.
• Pre-image completeness depends on the source config: full UPDATE/DELETE before-images need binlog_row_image=FULL (MySQL) / REPLICA IDENTITY FULL (PostgreSQL); otherwise only key columns are carried.
• Type parity with the batch export: every type both modes deliver — including PostgreSQL arrays of text and numbers (real List columns, inner NULLs preserved) and NUMERIC/DECIMAL above precision 38 (Decimal256) — is byte-identical to the batch export. Where the modes differ today (PostgreSQL bare numeric, money, inet, cidr and temporal/uuid/bytea/numeric arrays: batch refuses the column, CDC carries the server’s text) the difference is declared in docs/type-capability-matrix.yaml. Enforced per engine by the live *_batch_and_cdc_deliver_every_ledger_row_alike tests, generated from that ledger.