Rivet Documentation

Rivet exports data from PostgreSQL, MySQL, SQL Server, and MongoDB to Parquet/CSV files on local disk, S3, GCS, Azure Blob Storage, or stdout.

Install from Rust: cargo install rivet-cli (crates.io name is rivet-cli; the binary is rivet). Other install options live in the repo README.

This folder contains modular guides for running exports, a complete configuration and CLI reference, an architecture overview, and the full set of architecture decision records.

Why teams pick Rivet

Three properties, each measured, not asserted:

• Source-safe under load — the batch export holds no long-running query on the source (0.00 s vs 7.7–94.6 s for the field); CDC reads the log, not your tables.
• Flat memory at any scale — a steady 57 MB peak RSS (2×–63× smaller than the field), flat from 10 k rows to a field-proven 454 M-row table with no OOM.
• CDC cost, per engine — an honest per-engine account of the one operational hazard (PostgreSQL slot retention) and why MySQL / SQL Server / MongoDB can’t fill the source disk.

Supported database versions

PostgreSQL and MySQL run the full end-to-end suite on each release; SQL Server and MongoDB carry their own scope and CI coverage:

See reference/compatibility.md for the version-support policy, the exact test matrix, and notes on engine-specific features.

Start here

Pick one — they’re ordered shortest to deepest. Read top-to-bottom, then come back to this index when you need a reference.

Short terminal walkthroughs in gifs/:

• gifs/basic.gif — scaffold config, preflight, run, inspect state (≈25 s)
• gifs/plan-apply.gif — sealed plan/apply with credential redaction (≈20 s)
• gifs/reconcile-repair.gif — chunked + reconcile + targeted repair (≈35 s)

Export Modes

Destinations

Cloud auth & trust: cloud-auth.md — per-backend credential-flow matrix (AWS / GCS / Azure incl. SAS) · cloud-destinations.md — the cloud write trust contract (manifest, quarantine, support matrix).

Output layout

Reference

Trust contracts

The five surfaces a serious operator inspects before adopting Rivet. Same five rows, same order, are mirrored at the top of the project README.

Best Practices

Practical guides explaining why settings matter and when to use them.

Architecture

Production

Operator recipes

Action-first cookbooks for the most common production scenarios.

Architecture Decision Records

Example Configs

Ready-to-use YAML templates live in the examples/ directory. To scaffold YAML from a live database (rivet init), see reference/init.md and the root docker-compose.yaml.

Source-safe under load

The first question a serious operator asks about an extraction tool is what does it do to my source database while it runs? A careless SELECT * holds a long-running transaction, pins a read snapshot, inflates temp space, and spikes p99 latency for every other query on the box. Rivet is built so that the honest answer is “almost nothing you’ll notice.”

It holds no long-running query

A batch export streams the source in bounded pages — one chunk / one page at a time — and flushes each to a Parquet part before asking for the next. The longest query Rivet ever holds open on the source is a single page, not a full-table scan. In the cross-tool benchmark (PostgreSQL → Parquet, measured under a concurrent OLTP workload), the longest single server-side query each tool held was:

Rivet is the only tool in the field that never parks a long-running read on the source. Everything else holds one server-side query open for the length of a full scan — 8 to 95 seconds here, and proportionally longer on a real table. That is the query your DBA sees in pg_stat_activity blocking autovacuum, or the one a pooler’s statement timeout kills at the worst moment.

It stays gentle on concurrent traffic

Under the same concurrent OLTP workload, the p99 latency multiplier (how much the export slows other queries) is mid-field for Rivet — 5.1× single-stream, 3.3× chunked — comparable to the rest, but achieved at 57 MB of RAM and zero long queries instead of hundreds of megabytes to multiple gigabytes with a scan pinned open. Rivet trades a little client-side CPU for a source that never sees a heavy query.

If the source is fragile, production-shared, or behind a pooler (pgBouncer, ProxySQL, MaxScale), that trade is the whole point. See MSSQL gentle extraction and resource-aware extraction for the per-engine knobs.

CDC is a passive reader too

Change-data-capture reads the transaction log, not your tables. Running a looping CDC drain against a live writer, the writer’s throughput barely moves — except on SQL Server, and that cost is inherent to the engine, not to Rivet:

The full per-engine numbers, harnesses, and contracts live in the performance & harm ledger. The one operational hazard that is per-engine — a CDC reader pinning source log retention — is covered on its own page: CDC cost, per engine.

Flat memory at any scale

Rivet’s memory is bounded by how much you buffer before a flush — not by how big the table is. It streams rows into Arrow batches, and every time a batch fills it is written to a Parquet part and dropped. A 10-thousand-row table and a 500-million-row table run at the same resident set size.

The number that matters

In the cross-tool benchmark, peak resident memory was:

Rivet is 2× to 63× smaller than the field — and crucially, that 57 MB is flat. The tools that buffer the result set (or materialise it in an embedded engine) scale their memory with the data; Rivet does not.

Proven at production scale

The benchmark table is small enough to fit in RAM, which is exactly why the flat-memory property is invisible there for the buffering tools. It stops being invisible on a real table. A field run pulled a 454-million-row table to Parquet in about two hours with flat memory and no OOM — the same table on which Airbyte OOM’d. The mechanism the benchmark measures (bounded Arrow batches → Parquet parts) is the mechanism that survives at 454 M rows.

This is what lets Rivet run on a 512 MB–4 GB host, in a small Kubernetes Job, or alongside other work on a shared box. See low-memory runners for the exact settings and the RSS budget formula.

CDC memory is bounded by rollover, not the backlog

The same property holds for change-data-capture. A CDC drain’s peak RSS is O(rollover) — the part-size at which it flushes and checkpoints — not O(backlog). A soak test grows the drain interval 12× (10 → 120 minutes of accumulated changes) and peak RSS stays flat: the run reads, flushes at rollover, checkpoints, and acks in a loop, so a larger backlog just means more loops, not more memory. The harness self-asserts both the flat RSS and that every churned row was captured; details in the performance ledger.

CDC cost, per engine

Change-data-capture reads a source’s transaction log, and the one operational hazard worth understanding before you turn it on is log retention: does the reader hold log data on the source, and can an abandoned reader fill the source disk? The answer is not the same across engines, and Rivet is explicit about it.

The one engine that can pin the disk: PostgreSQL

PostgreSQL’s logical replication slot is a consume-retention mechanism: the server keeps WAL from the slot’s confirmed position forward until the consumer acknowledges it. That is what makes resume lossless — Rivet’s CDC delivery is at-least-once (a crash between flush and acknowledge re-reads the un-acked span; dedupe downstream by PK + __op if you need exactly-once effects) — and it is also the one place where a reader can hurt the source. While Rivet keeps up, retention is bounded to roughly interval × write-rate (≈ 23.5 MB over a 6 s cycle at ~4 MB/s in the harness). But an abandoned or stuck slot is unbounded: during this project’s own testing, a forgotten slot pinned 29 GB of WAL and crashed the PostgreSQL instance by filling its disk.

Operational rule for PostgreSQL CDC: monitor slot lag / retained WAL, and drop slots you no longer consume. This is inherent to how PostgreSQL logical replication guarantees no-loss — every tool that uses a slot inherits it — but Rivet names it plainly rather than hiding it.

The engines that cannot: MySQL, SQL Server, MongoDB

On the other three engines, log retention is decided by the server on its own schedule, independent of any reader. A Rivet CDC run — or a crashed one — structurally cannot fill the source disk:

On these engines the trade-off flips: because the server can trim the log out from under a slow reader, the risk is not disk-fill but falling behind (missing changes if you resume after the log has rolled past your checkpoint). The mitigation is a short enough scheduler interval, not disk monitoring.

Why this shapes the default

Because retention on PostgreSQL is coupled to acknowledgement, an abandoned continuous stream is the worst case — it holds a slot open forever. That is a large reason the OSS model is the bounded drain: until_current now defaults to true, so a scheduled run reads to the log end, checkpoints, acks (advancing the slot), and exits. The next cycle resumes from the checkpoint. Continuous streaming is an explicit opt-in (until_current: false, or rivet cdc --stream) that logs a warning, so you never start a never-terminating slot-holding stream by accident.

The full per-engine numbers, harnesses, and contracts are in the performance & harm ledger; the recovery playbook per symptom is in CDC failure modes.

Who is Rivet for?

A short, honest fit-check. Rivet is intentionally narrow — the goal is to do one thing well, not to be the only data tool you need. This page exists so you can decide in 60 seconds whether to keep reading, or whether something else is a better fit for your problem.

If you came here from a search for “Postgres / MySQL / SQL Server / MongoDB → Parquet” or “extract a big SQL table without an OOM”, you are probably in the right place.

Yes, Rivet is probably a good fit if…

• You need to dump rows from PostgreSQL, MySQL, or SQL Server (or documents from MongoDB) into Parquet or CSV files — locally, on S3, GCS, or Azure Blob Storage.
• The source database is fragile, production-shared, or behind a pooler (pgBouncer, ProxySQL, MaxScale) and a careless SELECT * is going to hurt someone.
• You want resumable extraction — the job can crash, the network can blip, and the next run continues from a checkpoint instead of starting over.
• You want a manifest + _SUCCESS trust contract so a downstream loader can decide exactly which files belong to a given run.
• You are happy operating Rivet from cron, Airflow, GitHub Actions, Argo, a one-off shell script, or a Kubernetes Job — anything that can invoke a single CLI binary with a YAML config.
• You can write a SQL query. Rivet does not abstract SQL away; it runs the query you give it.

No, Rivet is probably not the right tool if…

If you find yourself trying to bend Rivet into one of the above shapes, stop and pick the tool above instead. We are not trying to become any of those, and shoehorning will be painful.

Edge cases — Rivet can do this, but read first

• Very large single-table dumps (100M+ rows). Yes, but read docs/modes/chunked.md first — chunked mode with the right cursor column is the difference between an export that finishes in 20 minutes and one that holds a single SQL statement open for 2 hours.

• Sources with weak or missing primary keys / cursor columns. Rivet has incremental_cursor_mode: coalesce for nullable primaries (see composite cursor walkthrough) and keyset (chunk_by_key) for chunking without an integer key — but these surface tradeoffs in rivet check (sparse range warnings). Look at the warnings, do not ignore them.

• Read replicas with replication lag. On PostgreSQL, a full-mode export runs inside a single snapshot transaction, so the exported rows are point-in-time consistent as of the replica’s “now”. Chunked/keyset exports issue independent short queries (parallel workers each on their own connection), and other engines (e.g. MySQL) run plain autocommit SELECTs — there consistency is per-query, not per-run. Either way, a lagging replica gives you the replica’s “now”, not the primary’s. Operator’s responsibility to choose the right endpoint.

• Sources behind SSH bastions / jump hosts. No native SSH tunnel support yet (tracked in rivet_roadmap.md Epic 13). Use ssh -L or autossh to forward the port and point Rivet at localhost:<forwarded>.

• Stateless / ephemeral runners (Kubernetes pods, Lambda, ECS tasks). Set RIVET_STATE_URL to a PostgreSQL state backend so cursors and checkpoints survive pod death. See the README § Stateless deployment.

Decision shortcut

Need always-on live streaming (into Kafka)?    → Debezium / Estuary
Need CDC captured to files (resumable)?        → Rivet  (mode: cdc)
Need a connector for a non-DB SaaS source?     → Airbyte / Fivetran
Need a managed extract+load product?           → Fivetran / Airbyte Cloud
Load extracted data into BigQuery / Snowflake? → Rivet  (rivet load)
Need a SQL-based transformation framework?     → dbt
Need an orchestrator?                          → Airflow / Dagster
Need to dump PG/MySQL to Parquet/CSV safely?   → Rivet

If you stayed on the last line, the Getting Started guide is ~3 minutes and ends with one Parquet file you can arrow read or duckdb 'SELECT * FROM read_parquet(…)' against.

Last updated: 2026-07-09.

Getting Started

Rivet exports tables from PostgreSQL, MySQL, and SQL Server (and collections from MongoDB) to Parquet (or CSV) files — locally, to S3, GCS, or Azure Blob Storage. Point it at a database, scaffold a config from your real tables, then run. (MongoDB has its own reference: reference/mongodb.md.)

brew install panchenkoai/rivet/rivet
export DATABASE_URL='postgresql://user:pass@localhost:5432/mydb'
# `orders` is a placeholder — use one of YOUR tables, or omit --table to scan the whole schema
rivet init --source-env DATABASE_URL --table orders -o rivet.yaml
rivet run -c rivet.yaml --validate

That’s the whole flow. The four steps below explain each command, expected output, and where to go from each. Read time: ~3 minutes.

Already running it locally? Jump to §3 Preflight & run. If you’re evaluating it for production, finish this page first, then continue with docs/pilot/.

1 · Install

# macOS / Linux — Homebrew (recommended)
brew install panchenkoai/rivet/rivet
rivet --version

# Docker — try without installing anything
docker run --rm ghcr.io/panchenkoai/rivet:latest --version

Pre-built binaries are published for Linux and macOS (x86-64 + arm64). On Windows, install from source with cargo install rivet-cli (a native binary is not currently published). Other install paths — cargo install rivet-cli, build from source, plus the full Docker recipe with database-on-host pointers (host.docker.internal vs --network host) — live in the project README § Installation. Shell completions: rivet completions bash|zsh|fish.

Try it in 60 seconds — no database of your own

Spin up a throwaway PostgreSQL, seed one table, and export it — nothing external to configure:

docker run -d --name rivet-demo -e POSTGRES_PASSWORD=demo -p 5432:5432 postgres:16
sleep 3
docker exec -i rivet-demo psql -U postgres <<'SQL'
CREATE TABLE orders (id serial PRIMARY KEY, name text, price numeric(10,2),
                     updated_at timestamptz DEFAULT now());
INSERT INTO orders (name, price)
  SELECT 'order-'||g, (random()*500)::numeric(10,2) FROM generate_series(1,500) g;
SQL

export DATABASE_URL='postgresql://postgres:demo@localhost:5432/postgres'
rivet init --source-env DATABASE_URL --table orders -o rivet.yaml
rivet run -c rivet.yaml --validate
# → 500 rows of typed Parquet in ./output/orders/.  Clean up: docker rm -f rivet-demo

That is the whole flow against a real (throwaway) database. Then jump to §4 Inspect & iterate, or read on to point Rivet at your own database.

2 · Connect & scaffold a config

Recommended pattern: put the connection URL in an environment variable and reference it from the config so credentials never enter the file or shell history.

export DATABASE_URL='postgresql://user:pass@localhost:5432/mydb'
# MySQL: same flag, just a mysql:// URL
# export DATABASE_URL='mysql://user:pass@localhost:3306/mydb'

rivet init --source-env DATABASE_URL --table orders -o rivet.yaml
# `orders` is a placeholder — use one of YOUR tables, or omit --table to scan the whole schema

rivet init connects once, reads the column list + a rough row estimate from the live database, and writes a YAML file with url_env: DATABASE_URL and a sensible default mode. For a large table with a single-column primary key it picks keyset (chunk_by_key) — seek paging that stays flat-memory and is immune to sparse/gappy keys; keyset is scaffolded sequential (add parallel: N yourself to fan it into row-percentile ranges). A large table with no single-column PK gets a range chunk_column with a row-scaled parallel: (1 / 2 / 4) out of the box — measured ~1.4× faster on a wide 2 M-row table and ~4× on a narrow one. You can also point it at a whole schema (--schema public) or emit a richer JSON discovery artifact instead (--discover -o discovery.json).

Full flag reference: reference/init.md. For a manually-authored YAML instead of rivet init, see reference/config.md.

State file. Rivet creates .rivet_state.db next to the config (cursors, chunk checkpoints, run history). Add it to .gitignore if the folder is version-controlled — see SECURITY.md § Sensitive local artifacts.

3 · Preflight & run

rivet doctor -c rivet.yaml   # verify source + destination auth
rivet check  -c rivet.yaml   # dry-run analysis per export
rivet run    -c rivet.yaml --validate --reconcile

The full basic workflow (init → doctor → check → run → state) recorded as a single terminal cast:

What each step does:

• rivet doctor — connects to the source and writes a tiny probe object (.rivet_doctor_probe) to every destination prefix — removed afterwards on local destinations, while on S3 / GCS / Azure it stays at the prefix (the destination seam has no delete) and is filtered out of manifest and validate listings; fixes nothing, fails loudly on any auth / network issue.

• rivet check — runs EXPLAIN against your queries, estimates row counts, detects whether your cursor / chunk columns are indexed, and emits a verdict + concrete suggestion. Verdicts are EFFICIENT · ACCEPTABLE · DEGRADED · UNSAFE; on the SQL engines the last two carry a mode-aware Suggestion: line (MongoDB is full-scan-only, so its verdicts omit the mode suggestion).

  

• rivet run --validate --reconcile — extracts. --validate reads each output file back and verifies its row count; --reconcile runs SELECT COUNT(*) on the source query and compares with what was exported.

Example summary card after a successful run:

── orders ──
  run_id:      orders_20260519T120000.123
  status:      success
  tuning:      profile=balanced (default), batch_size=10,000 (batch_size_memory_mb=32MiB → effective FETCH in logs)
  rows:        5,432
  files:       1
  output:      file://./output
  bytes read:    1.2 MB
  bytes written: 847.0 KB
  duration:    1.2s
  peak RSS:    15 MB (sampled during run)
  validated:   pass
  schema:      unchanged
  reconcile:   MATCH (5,432/5,432)

4 · Inspect & iterate

rivet state show   -c rivet.yaml             # cursors (incremental exports)
rivet metrics      -c rivet.yaml --last 10   # per-run history
rivet state files  -c rivet.yaml             # files actually written
rivet journal      -c rivet.yaml --export orders   # per-run events / retries / quality issues

To make the second run only export rows that changed, switch the export to incremental mode with a cursor_column: (must be monotonically increasing — usually updated_at or a sequence id):

exports:
  - name: orders
    query: "SELECT id, name, updated_at FROM orders"
    mode: incremental
    cursor_column: updated_at
    format: parquet
    skip_empty: true            # a run with no new rows reports `skipped`
    destination:
      type: local
      path: ./output

Subsequent rivet run invocations will only fetch rows with updated_at > the stored cursor. For tables larger than ~5 M rows, switch to mode: chunked instead — see modes/chunked.md.

5 · Many tables: plan once, apply by waves

When a config has several exports, rivet plan assigns each one a wave — a priority band derived from its size, chunking strategy, and risk (ADR-0006). By default rivet plan is read-only: it prints the schedule for you to review but does not touch the config. Add --annotate-waves to write the wave: / parallel_safe: fields back into the config, where you can see and hand-edit them:

rivet plan -c rivet.yaml                   # review the schedule (read-only)
rivet plan -c rivet.yaml --annotate-waves  # write `wave: N` onto every export, in place

exports:
  - name: users
    wave: 1        # small / cheap → runs first
    # …
  - name: events
    wave: 3        # large → runs later
    # …

rivet apply then runs the whole config wave by wave, lowest first, with a barrier between waves — every export in wave 1 finishes before wave 2 starts. Exports with no wave: run last:

rivet apply rivet.yaml          # a .yaml path → wave-ordered execution

(A .json path still means the sealed single-artifact replay — see reference/cli.md § rivet apply.) The plan suggests the waves; you stay in control — hand-edit wave: and apply respects your order.

Parallel within a wave — only where it’s safe

Add parallel_export_processes: true (or pass rivet apply --parallel-export-processes) and, within each wave, the cheap exports — the ones rivet plan marked parallel_safe: true (cost class Low, under ~100K rows) — run concurrently as separate processes. A heavier export already chunk-parallelizes its own ranges internally, so it runs alone in its wave: two big tables at once would multiply the load on the source. The wave stays bounded because only the cheap parallel_safe exports run concurrently, and each child honors its own batch/memory caps (the adaptive back-pressure governor is a separate opt-in: tuning.adaptive: true with parallel > 1).

parallel_export_processes: true   # top-level: parallelize the cheap (parallel_safe) exports within each wave

Load into BigQuery or Snowflake (optional)

Rivet stops at typed Parquet by default. To load it into a warehouse, add a top-level load: block to the same config and run rivet load — the target table, column types, and source files are all derived from the export (nothing hand-typed):

# rivet.yaml — the export above, plus a load target
load:
  target: bigquery          # or: snowflake (+ connection / warehouse / database / schema / storage_integration)
  project: my-gcp-project
  dataset: analytics
  cleanup_source: true      # wipe the staged Parquet once the load is row-count-verified

rivet run  -c rivet.yaml    # extract → GCS
rivet load -c rivet.yaml    # load → warehouse (native types; count-gated before any cleanup)

The load follows the export’s mode: — full overwrites the latest snapshot; incremental / cdc append to <table>__changes and expose a current-state dedup view keyed on the source primary key rivet run recorded (set pk: [id] in the load: block for a query: export or to override it). Recipes: snowflake-load.md · cdc-bigquery-load.md.

When something is wrong

Rivet tries to fail early and say exactly what to fix — most mistakes are caught at check / doctor time, before a single row is read.

A query that references a table (or column) that doesn’t exist is caught by rivet check — it exits non-zero with the offending name and SQLSTATE, instead of passing through to a half-finished run:

A typo in a config field is caught at parse time with a Did you mean …? suggestion that names the line:

An unreachable database — down, wrong host/port, or a tunnel that isn’t up — is reported by rivet doctor with a reachability hint before you waste a run:

More failure modes (retries, schema drift, crash/resume) and exactly what rivet does for each: semantics.md.

Next steps

Rivet Cheat Sheet

One page covering setup, extract, load and verification. Every command reads the same YAML config (-c rivet.yaml). Full references: reference/cli.md · reference/config.md · reference/cdc.md.

init → doctor → check → run → validate / reconcile → load

Command builder. On the docs site, fill in the form below and every command and config on this page is rewritten with your values, ready to copy. On GitHub the form is not rendered: replace the {{…}} placeholders by hand.

0. Prerequisites: cloud and warehouse

Choose the destination and load target in the form, and the blocks below switch to the matching setup. Skip any step you have already done.

0.1 Destination bucket and credentials

{{CLOUD_SETUP}}

• rivet does not read AWS_PROFILE, and SSO/login profiles do not work as aws_profile:. Export temporary credentials instead (the sso option above).
• Temporary AWS keys expire, usually after about an hour. Re-run the export-credentials line before the next run.
• BigQuery and Snowflake loads read GCS only (destination: type: gcs). A ClickHouse load reads GCS, S3 or Azure.
• rivet doctor -c rivet.yaml confirms the credentials can write to the prefix. On cloud storage it leaves a .rivet_doctor_probe object behind.

More auth paths (MinIO, Azurite, SAS details): cloud-auth.md · cloud-permissions.md.

0.2 Warehouse for rivet load

{{LOAD_SETUP}}

{{LOAD_SQL}}

• BigQuery: the dataset must already exist, because rivet does not create datasets. Keep it in the same location as the bucket.
• Snowflake: database, schema and storage integration must already exist. rivet creates the file format, stage, tables and views itself, which is why the role needs the CREATE grants above.
• ClickHouse (preview): the database must already exist. rivet creates the tables and views over HTTP. Known limits: recipes/clickhouse-load.md.

1. Setup

1.1 Batch (full / incremental / chunked)

brew install panchenkoai/rivet/rivet          # or: cargo install rivet-cli

# Credentials: the password is read without echo, never typed into the command line
printf 'DB password: '; read -rs DB_PASS; echo
export DATABASE_URL="{{URL}}"

# Scaffold a config from the live database
rivet init --source-env DATABASE_URL --table {{TABLE}} --tls {{TLS}}{{INIT_DEST}} -o rivet.yaml    # one table
rivet init --source-env DATABASE_URL --schema {{SCHEMA}} --tls {{TLS}}{{INIT_DEST}} -o rivet.yaml  # whole schema
rivet init --source-env DATABASE_URL --schema {{SCHEMA}} --include 'order*' --exclude '*_tmp' --tls {{TLS}} -o rivet.yaml
rivet init --source-env DATABASE_URL --tls {{TLS}} --discover -o discovery.json   # JSON: row estimates, cursor/chunk candidates

# Preflight
rivet doctor -c rivet.yaml                     # source + destination auth / connectivity
rivet check  -c rivet.yaml                     # EXPLAIN, index checks, verdict per export
rivet check  -c rivet.yaml --type-report --target {{LOAD_KIND}}   # bigquery | snowflake | duckdb | clickhouse
rivet check  -c rivet.yaml --strict            # non-zero exit on any unsafe type mapping

Minimal config:

source:
  type: {{SOURCE_TYPE}}           # postgres | mysql | mssql | mongo | oracle
  url_env: DATABASE_URL           # or url_file: / host+user+password_env+database
  tls: { mode: {{TLS}} }          # disable | require | verify-ca | verify-full (+ ca_file:)
exports:
  - name: {{NAME}}
    table: {{TABLE}}              # or query: / query_file:
    mode: full                    # full | incremental | chunked | time_window | cdc
    format: parquet               # parquet | csv
    compression_profile: balanced # none | fast | balanced | compact
    destination: {{DEST}}

All destination shapes:

destination: { type: local, path: ./output }
destination: { type: gcs,   bucket: my-bucket, prefix: exports/orders/ }
destination: { type: s3,    bucket: my-bucket, prefix: exports/orders/, region: us-east-1 }
destination: { type: azure, bucket: my-container, account_name: acct, account_key_env: RIVET_AZURE_KEY, prefix: exports/ }

.rivet_state.db (cursors, checkpoints, run history) is created next to the config. Add it to .gitignore.

Oracle (preview): the URL path is the service name (oracle://user:pass@host:1521/ORCLPDB1), not a SID. Unquoted names are upper-case (table: orders reads ORDERS). Types, modes and limits: reference/oracle.md.

1.2 CDC

Scaffold:

rivet init --source-env DATABASE_URL --mode cdc --table {{TABLE}} --tls {{TLS}} -o cdc.yaml
rivet init --source-env DATABASE_URL --mode cdc --tls {{TLS}} -o cdc.yaml   # whole DB: one `tables:` export (PG/MySQL),
                                                                             # one export per table (SQL Server)
rivet doctor -c cdc.yaml    # also probes slot WAL retention / binlog config / CDC Agent + retention

Source prerequisites:

Grants for the selected engine:

{{CDC_GRANTS}}

Rules of thumb:

• MySQL: connect directly, not through ProxySQL/MaxScale. Give rivet a unique server_id.
• MySQL on RDS / Aurora: two settings that are not in my.cnf. Binary logging follows automated backups — with retention at 0 the instance runs log_bin = 0 and every binlog query answers ERROR 1381, whatever the parameter group says. And retention is not binlog_expire_logs_seconds: RDS purges a binlog as soon as the engine no longer needs it, so the next run’s resume dies with ERROR 1236 (measured: a checkpoint taken at 13:42 was already past retention at 13:59). Set it explicitly, well above the run interval: CALL mysql.rds_set_configuration('binlog retention hours', 72); A read replica also needs log_replica_updates = 1.
• PostgreSQL: an abandoned slot pins WAL and fills the disk. Drop it with SELECT pg_drop_replication_slot('{{SLOT}}');. Set max_slot_wal_keep_size to cap it.
• SQL Server: change-table retention defaults to about 3 days. A run that falls behind it fails loudly and needs a re-snapshot.
• Reading from a replica is verified on every engine: MySQL (log_replica_updates=ON; rivet refuses a replica without it), PostgreSQL 16+ standbys in continuous mode only (until_current: false), SQL Server readable secondaries and MongoDB secondaries (readPreference=secondary).

CDC config:

source:
  type: {{SOURCE_TYPE}}
  url_env: DATABASE_URL
  tls: { mode: {{TLS}} }
exports:
  - name: {{NAME}}_cdc
    table: {{TABLE}}               # or tables: [a, b]  (one stream, PG/MySQL only)
    mode: cdc
    format: parquet
    cdc:
      initial: snapshot            # first run: anchor → full snapshot → drain stream
      checkpoint: {{CKPT_DIR}}/{{NAME}}.ckpt   # required for a baseline (initial:/backfill:) on every engine but PostgreSQL; MySQL/MongoDB/Oracle need it for any mode: cdc
      until_current: true          # default: drain to the log end as of open, then exit
      {{CDC_PARAM}}
      # rollover: 100000           # rows per part (≈ drain memory)
    destination: {{CDC_DEST}}

With several mode: cdc exports, each one needs its own slot, server_id and checkpoint. The defaults collide, and config validation rejects them.

1.3 One config for the whole cycle: extract → load → compact

Add the warehouse flags to either scaffold above and rivet init writes one file that drives everything — the exports, the top-level load: block, and the base-and-buffer layout rivet compact needs. Nothing is hand-added afterwards.

rivet init --source-env DATABASE_URL --table {{TABLE}} --mode incremental --tls {{TLS}}{{INIT_LOAD}} -o rivet.yaml
rivet init --source-env DATABASE_URL --mode cdc --tls {{TLS}}{{INIT_LOAD}} -o rivet.yaml
                                               # whole DB: one `tables:` stream with backfill: auto

rivet doctor  -c rivet.yaml    # source + destination auth
rivet run     -c rivet.yaml    # Parquet → gs://{{BUCKET}}/exports/{{TABLE}}/
rivet load    -c rivet.yaml    # → the base on the first pass, the buffer on later ones
rivet compact -c rivet.yaml    # BigQuery: MERGE the buffer into the base, drop the buffer

• ClickHouse: the cycle is run + load only. The change log is a ReplacingMergeTree, the view reads it with FINAL, and rivet compact does nothing. The generated block carries no layout: and no partition:.
• Snowflake: rivet init has no Snowflake flags. Scaffold with --gcs-bucket and add the load: block from §3.

Every value in the generated load: block is a guess from the catalog — review it before the first load. It carries target: bigquery, pk: auto, cluster_by: auto, cleanup_source: true; a per-table partition: on the creation stamp at granularity: day (never a mutation stamp, which would move a row between partitions on every update); and, for a mode that carries deltas (incremental, cdc), layout: base_buffer so compact has a base to merge into. Field-by-field reference: §3 below.

--gcs-bucket is required with the BigQuery flags: a BigQuery load reads GCS only, so a load: block over a local or S3 destination is a config its own next step refuses. The ClickHouse flags take --gcs-bucket or --s3-bucket. On a whole-database CDC scaffold the partition guesses land on the stream’s load.tables.<table> blocks — the place the load reads them — not on the per-table recipes, which the load never reads.

2. Extract

2.1 Batch

rivet run -c rivet.yaml                         # all exports
rivet run -c rivet.yaml -e {{NAME}}             # one export
rivet run -c rivet.yaml --validate --reconcile  # verify files + source COUNT(*) match
rivet run -c rivet.yaml --parallel-exports      # exports concurrently (threads)
rivet run -c rivet.yaml --parallel-export-processes   # one child process per export
rivet run -c rivet.yaml -p day=2026-09-14       # substitutes ${day} in queries
rivet run -c rivet.yaml --json --summary-output run.json
rivet run -c rivet.yaml --resume                # continue a crashed chunked run (chunk_checkpoint: true)

# Many tables: plan waves, then apply
rivet plan  -c rivet.yaml                       # read-only schedule
rivet plan  -c rivet.yaml --annotate-waves      # write wave:/parallel_safe: into the config
rivet apply rivet.yaml                          # wave by wave
rivet apply rivet.yaml --resume                 # skip exports with _SUCCESS, resume the rest.
                                                #   WITHOUT it a re-run appends fresh parts beside the old
                                                #   ones and rewrites manifest.json for this run only — a
                                                #   glob reader then double-counts. Clear the prefix first.
rivet apply rivet.yaml --pool 4 --split         # work-stealing pool; split one dominant table.
                                                #   A GENERATED config carries no `parallel_safe:`, so every
                                                #   export counts as heavy and `--pool` alone overlaps
                                                #   NOTHING — run `--annotate-waves` first. `--split` needs a
                                                #   dominant full/chunked/keyset export with a chunk key
                                                #   (never incremental/cdc). Measured on one 1.26M-row set:
                                                #   66s waves · 74s bare --pool · 57s annotated · 42s +split
rivet plan  -c rivet.yaml -e {{NAME}} --format json -o plan.json && rivet apply plan.json   # sealed replay
                                                # `-o` REQUIRES `--format json` (pretty mode ignores it)

Mode snippets:

# incremental: only rows past the stored cursor
mode: incremental
cursor_column: {{CURSOR}}
skip_empty: true
settle: { after: 1h }            # optional: export a row only once it is ≥1h old (s/m/h/d)
                                 # settle.column defaults to the cursor

# chunked, range key
mode: chunked
chunk_column: {{PK}}
chunk_size: 100000
parallel: 4
chunk_checkpoint: true           # enables --resume

# chunked, keyset (unique NOT NULL key, immune to sparse keys)
mode: chunked
chunk_by_key: {{PK}}
chunk_checkpoint: true           # crash recovery only
# keyset_incremental: true       # append-only tables: a clean re-run pulls only new keys

# time_window: rolling N days
mode: time_window
time_column: created_at
days_window: 30

Inspect state:

rivet state show         -c rivet.yaml               # incremental cursors
rivet state reset        -c rivet.yaml -e {{NAME}}   # re-export from scratch
rivet state chunks       -c rivet.yaml -e {{NAME}}   # chunk checkpoint status
rivet state reset-chunks -c rivet.yaml --stuck-checkpoints
rivet state progression  -c rivet.yaml               # committed / verified boundaries
rivet state runs         -c rivet.yaml --running     # run-status ledger
rivet state finish-run   -c rivet.yaml --run-id <id> # close a known-dead `running` row
rivet metrics            -c rivet.yaml --last 10
rivet journal            -c rivet.yaml -e {{NAME}}   # events, retries, quality issues

2.2 CDC

Config-driven (recommended: cloud destinations, TLS, recorded runs):

rivet run -c cdc.yaml                            # bounded: drain, write parts, checkpoint, exit
rivet run -c cdc.yaml --parallel-export-processes   # SQL Server per-table exports in parallel
rivet metrics -c cdc.yaml                        # CDC runs appear with mode=cdc

Schedule rivet run on an interval. Each run resumes from the checkpoint or slot. cdc.until_current: false streams continuously, but only MySQL and MongoDB stay up that way. PostgreSQL and SQL Server still exit on catch-up, and Oracle refuses it at config load (bounded drain only).

Ad-hoc CLI (loopback hosts only, since it has no TLS):

rivet cdc --source-env DATABASE_URL --table {{TABLE}}{{CDC_FLAG}}          # NDJSON to stdout
rivet cdc --source-env DATABASE_URL --table {{TABLE}}{{CDC_FLAG}} \
          --output ./cdc-out --format parquet --checkpoint ./{{NAME}}.ckpt   # typed Parquet (local dir)
rivet cdc --source-env DATABASE_URL --table {{TABLE}}{{CDC_FLAG}} --max-events 10000   # soft cap, stops at a commit boundary
rivet cdc --source-env DATABASE_URL --table {{TABLE}}{{CDC_FLAG}} --stream             # continuous instead of bounded

Output shape: one row per change.

Delivery is at-least-once, so dedupe downstream on PK + (__pos, __seq). Parts are named cdc-<run_id>-NNNNNN.parquet and accumulate across runs. manifest.json and _SUCCESS describe the latest run.

Recovery:

3. Load (BigQuery / Snowflake / ClickHouse)

Put a top-level load: block in the same config. The load reads column types from the state DB and never connects to the source. An Oracle mode: cdc export cannot feed a load: block yet (refused at config load). An Oracle batch export under load: is accepted, but no live test loads one into a warehouse yet.

load:
  {{LOAD_TARGET}}
  pk: auto                      # auto (recorded source PK) | none | [col, ...]   — incremental/cdc dedup key
  cluster_by: auto              # auto | none | [col, ...]  (≤4 on BigQuery; a full-load table's ORDER BY on ClickHouse)
  partition:                    # none (default) | exactly one of column / range / ingestion. Not on ClickHouse
    column: {{CURSOR}}
    granularity: day            # hour | day | month | year
    # range: { column: n, start: 0, end: 1000000, interval: 1000 }
    # ingestion: day
    expiration_days: 90
    require_filter: false
  cleanup_source: true          # delete staged Parquet after the count gate passes
  gc_orphans: false             # also delete unmanifested crash leftovers
  allow_source_drift: false     # load even if the manifest's source count ≠ extracted
  layout: base_buffer           # log_view (default for incremental / capture-only CDC) | base_buffer
                                #   base_buffer: `<table>` is a PHYSICAL table, `<table>__changes` a
                                #   per-cycle buffer that `rivet compact` MERGEs in and drops. BigQuery only.
                                #   Absent: a CDC stream with `backfill:` gets base_buffer, the rest log_view.
  deleted_flag: true            # whether the base carries `__is_deleted`; absent: true for cdc, false otherwise
exports:
  - name: {{NAME}}
    # ...
    load: { pk: [{{PK}}], partition: none }   # per-export override: pk, cluster_by, partition, cleanup_source, gc_orphans, allow_source_drift, layout, deleted_flag
    # a multiplex `tables:` stream adds `tables: { <table>: { pk: [...], partition: none } }` — one block per captured table

Required target fields: BigQuery takes project and dataset. Snowflake takes connection (a snow CLI connection), warehouse, database, schema and storage_integration. ClickHouse takes url, database, user and password_env, plus an optional named_collection (ClickHouse then reads the bucket itself). ClickHouse refuses partition:, layout: base_buffer and a MongoDB CDC stream.

rivet run  -c rivet.yaml              # extract → bucket
rivet load -c rivet.yaml              # load → warehouse
rivet load -c rivet.yaml --run-id "nightly-$(date +%F)"   # tag jobs (BQ label / Snowflake QUERY_TAG)
rivet load -c rivet.yaml --rebuild-changelog               # allow a billed rebuild when partitioning changed
rivet compact -c rivet.yaml           # base_buffer only: MERGE `<table>__changes` into `<table>`, drop the buffer
rivet state loads -c rivet.yaml -t {{WAREHOUSE_TABLE}}

What the load does for each export mode::

Guarantees:

• Manifest-driven. The load uses only the parts listed in Success manifests, never a prefix glob.
• Count gate. The warehouse COUNT(*) must equal the summed manifest rows before the load completes or cleans up.
• Cost labels. BigQuery jobs are labelled managed_by:rivet, rivet_op:{load,count,create,alter,view} and rivet_table:<t>.

4. Data verification

4.1 Built into the run

rivet run -c rivet.yaml --validate      # every manifest part present at its recorded size + _SUCCESS
rivet run -c rivet.yaml --reconcile     # + source COUNT(*) == exported rows (a mismatch fails the run)

exports:
  - name: {{NAME}}
    verify: content          # size (default) | content: require MD5 match per part (no download)
    on_schema_drift: fail    # warn (default) | continue | fail (exit 4)
    quality:
      row_count_min: 1000
      row_count_max: 10000000
      null_ratio_max: { {{PK}}: 0.0 }   # single runner only
      unique_columns: [{{PK}}]          # single runner only
      unique_max_entries: 1000000       # always cap memory

4.2 After the fact, without extracting

rivet validate -c rivet.yaml                       # full: manifest + parts + value-checksum re-read
rivet validate -c rivet.yaml --depth light         # manifest + _SUCCESS only (fast poll)
rivet validate -c rivet.yaml --depth sample        # + part reconcile + untracked surplus
rivet validate -c rivet.yaml -e {{NAME}} --date 2026-09-13   # a prior day's {date} prefix
rivet validate -c rivet.yaml -e {{NAME}} --prefix exports/{{NAME}}/2026-09-13/
rivet validate -c rivet.yaml --format json -o validate.json

For CDC, rivet validate descends into every table prefix and its snapshot/. A missing _SUCCESS means the run did not finish cleanly.

4.3 Source vs export (chunked, chunk_checkpoint: true)

rivet reconcile -c rivet.yaml -e {{NAME}}                           # per-chunk recount; non-zero exit on mismatch
rivet reconcile -c rivet.yaml -e {{NAME}} --format json -o rec.json
rivet repair    -c rivet.yaml -e {{NAME}} --report rec.json         # print the repair plan
rivet repair    -c rivet.yaml -e {{NAME}} --report rec.json --execute   # re-export mismatched ranges only

4.4 Independent oracle (not rivet’s own bookkeeping)

DuckDB fingerprints the source query and the Parquet separately (rows, distinct key, non-null counts, sums, lengths). The script supports PostgreSQL and MySQL sources.

pip install duckdb
python dev/correctness/verify_export.py \
  --source-type {{VERIFY_TYPE}} \
  --dsn "{{DSN}}" \
  --query "SELECT * FROM {{TABLE}}" \
  --parquet "/path/to/{{NAME}}/*.parquet" \
  --key {{PK}}                 # exit 0 = PASS, 1 = FAIL

A prefix with orphaned pre-crash parts reads high. Verify only the parts named in manifest.json.

Run this BEFORE rivet load, or set cleanup_source: false: the generated load: block sets cleanup_source: true, so a successful load deletes the staged Parquet and leaves this oracle nothing to read.

CDC replay check in DuckDB (latest image per key; the LSN parsing is PostgreSQL’s):

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
  FROM read_parquet('cdc-out/cdc-*.parquet')
)
SELECT * FROM (
  SELECT *, row_number() OVER (PARTITION BY {{PK}} ORDER BY lsn_key DESC, __seq DESC) rn FROM ev
) WHERE rn = 1 AND __op <> 'delete';
-- compare with: SELECT * FROM {{TABLE}};  on the source

4.5 Warehouse side

-- BigQuery: what each rivet step cost
SELECT (SELECT value FROM UNNEST(labels) WHERE key='rivet_op')    AS op,
       (SELECT value FROM UNNEST(labels) WHERE key='rivet_table') AS tbl,
       COUNT(*) jobs, SUM(total_bytes_billed) bytes_billed
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE EXISTS (SELECT 1 FROM UNNEST(labels) WHERE key='managed_by' AND value='rivet')
GROUP BY op, tbl ORDER BY bytes_billed DESC;

-- current state of a CDC load vs the source (cdc only: `__is_deleted` exists
-- when `deleted_flag` is on, which is the default for cdc and OFF otherwise —
-- on an `incremental` base this query fails with "Unrecognized name")
SELECT COUNT(*) FROM {{WAREHOUSE_SQL}} WHERE NOT __is_deleted;

-- an incremental / full base has no delete flag: count it plainly
SELECT COUNT(*) FROM {{WAREHOUSE_SQL}};

4.6 Inspection

rivet state files -c rivet.yaml -e {{NAME}} --json   # files actually written
rivet metrics     -c rivet.yaml -e {{NAME}} --json   # rows / files / bytes / status per run
                                                     #   NOTE: neither reaches `--split` sub-units — their
                                                     #   run ids are `<export>#0…#N` and `-e` only accepts a
                                                     #   config export name. Use `rivet state runs`, which
                                                     #   does list them.
rivet journal     -c rivet.yaml -e {{NAME}} --run-id <id>

Last updated: 2026-07-09.

Concepts

A one-page glossary of the terms Rivet uses everywhere — run_id, cursor, chunk, manifest, journal, progression. Read it once, then come back when a doc / CLI output references a term you’ve forgotten.

If you want the binding execution semantics (what’s at-least-once, what survives a crash, what doesn’t), see semantics.md. This page is the orientation; that one is the contract.

Two top-level objects

The same export can be extracted by many runs over time. Each run gets its own run_id; the export’s cumulative state (cursor, chunks done, files written) accretes across runs.

How the data is sliced

What’s recorded after each run

The decisive write order across these (which write happens before which) is the State Update Invariants in ADR-0001 — the I1-I7 sequence. The user-facing summary of “what happens if the process dies between I3 and I4” is in semantics.md § Crash semantics.

Two flags that are easy to confuse

Both are off by default; both add one line to the run summary. For chunked exports the dedicated per-partition reconcile + targeted repair workflow lives in rivet reconcile / rivet repair — see reference/cli.md § rivet reconcile and ADR-0009. The CLI flag --reconcile is the cheaper aggregate-only version.

Plan / apply (auditable execution)

You can stick with rivet run for everything; plan/apply is only there when you want a pre-execution review object that survives in a PR or a CI artifact.

Source-aware connections

Where each concept is exercised in the code

If you want to track a concept from this glossary back to the implementation:

• Source trait + ExportRequest — src/source/mod.rs
• Cursor / incremental — src/source/query.rs (predicate builder) + src/state/cursor.rs
• Chunked / checkpoint — src/pipeline/chunked/{mod, sequential_checkpoint, parallel_checkpoint}.rs + src/state/checkpoint.rs
• File log + Metrics — src/state/{file_log, metrics}.rs
• Journal — src/journal.rs (top-level) + src/state/journal_store.rs
• Progression — src/state/progression.rs (ADR-0008)
• Validate / reconcile (CLI) — src/pipeline/{validate, reconcile_cmd}.rs
• Plan / apply — src/pipeline/{plan_cmd, apply_cmd}.rs + src/plan/

Where to go next

• getting-started.md — the 5-minute install + first export, if you skipped it
• semantics.md — the binding execution contract (what’s at-least-once, what survives crashes)
• modes/ — when to pick full vs incremental vs chunked vs time_window vs cdc
• adr/ — every contract in this glossary has a numbered ADR with the rationale

Execution Semantics

How Rivet behaves under normal execution, retries, crashes, resume, repair, and reconcile. This is the contract a downstream pipeline can build on.

This page is a user-facing summary. The binding contracts live in the ADRs, which this document links into.

Scope

This document describes guarantees and known non-guarantees for:

• single-table exports (full, incremental, chunked, time-window, cdc),
• destination writes (local, S3, GCS, Azure Blob Storage, stdout),
• state and journal updates,
• automatic retries and --resume after crashes,
• rivet reconcile and rivet repair,
• in-process and subprocess parallelism.

It does not cover database / network / disk failures whose mode is external to Rivet (e.g. a corrupted Parquet file caused by a bad disk).

Core concepts

Normal execution

The pipeline is one straight line per export (see architecture.md § Data flow):

begin_query → FETCH batch → write batch to temp file
                          → next FETCH
                          → ...
            → finalize writer
            → destination.write(temp_file)
            → record manifest entry
            → advance cursor (incremental) / record chunk completion (chunked)
            → record metric (at end of run)

The exact state-write ordering is defined by ADR-0001 — State Update Invariants (I1–I7). The pipeline source code references these IDs at the call sites.

Retry semantics

Retries are classified by error type in src/pipeline/retry.rs:

The classifier (RetryClass) has two outcomes — Transient (retry) and Permanent (propagate):

A retried batch starts from the same cursor position as the failed attempt — see ADR-0001 I3 (Write Before Cursor). At-least-once delivery to the destination is therefore possible: on retry after a destination write succeeded but the cursor failed to advance, the same rows are written again, producing a duplicate file.

Retry-safety per destination is declared by capabilities().retry_safe — see ADR-0004 — Destination Write Contracts:

When retries occur against a non-retry-safe destination, the pipeline logs a WARN so operators see the mismatch.

Crash semantics

If the process is killed (SIGKILL, OOM, host reboot), the next state depends on where the crash landed. The full failure-point map is in ADR-0001 § Failure Point Map. Summarised:

Two strong invariants hold across every crash point:

• No row is silently skipped. The cursor only advances after the corresponding write succeeds (I3).
• No file at the destination is incomplete. Writers are finalized before destination upload (I1).

The trade-off is at-least-once at the destination: a crash between write and cursor advancement produces a duplicate file. Downstream consumers must tolerate this — see Known non-guarantees below.

Resume semantics

rivet run --resume consults the state DB to decide what work is outstanding. A plain rivet run (no --resume) never skips the chunks of a run that FINISHED — it does a fresh full pass. A checkpointed run holds its export’s run lease (an flock beside a SQLite state DB, a session advisory lock on a Postgres one) for as long as the process lives, and the OS releases it when the process dies, kill -9 included. So when a plain run finds a chunk-checkpoint run still in progress:

• its process is gone (the lease is free): the run resumes it, exactly as --resume would. A chunk_dense plan cannot be resumed, so it starts over.
• its process is alive (the lease is held): the run is refused, and so is --resume — wait for the live run to finish.

What --resume does:

• Incremental exports resume from export_state.last_cursor_value.
• Chunked exports consult the chunk_task table: tasks in completed are skipped; tasks in pending or running (the latter reset to pending on resume) are re-issued; tasks in failed are retried while attempts < max_chunk_attempts.
• Full and time-window modes do not resume — they restart from the beginning. The previous run’s output files remain at the destination unless cleaned manually.

Chunk task transitions are strictly forward (pending → running → {completed | failed}). A completed chunk is never re-claimed, even after a crash — see ADR-0001 I5 (Chunk Task Acyclicity).

Repair semantics

rivet repair re-exports specific chunks identified as mismatch or unknown by a prior rivet reconcile. The full contract is ADR-0009 — Reconcile and Targeted Repair.

Key properties:

• Repair derives only from a reconcile report (RR1). There is no operator-typed chunk range.
• Dry-run by default (RR2). --execute is required to perform writes.
• Repair writes new files alongside originals (RR5). Rivet does not delete or overwrite prior destination files. Downstream dedup is the operator’s responsibility.
• Repair does not advance the committed boundary (RR4). Run rivet reconcile again afterwards to advance last_verified_*.

Reconcile semantics

rivet reconcile compares per-chunk row counts between source and destination for the latest chunked run:

Scope and limits:

• Chunked mode only in v1. time_window and incremental exports surface a clear “not supported” error.
• COUNT(*) based. No hash-based partition verification yet.
• Verified boundary advances only on a fully clean report — zero mismatches and zero unknowns (ADR-0008 § PG5).
• Exit code gates on mismatch. rivet reconcile exits non-zero when any partition is a mismatch, so rivet reconcile && <next step> does not proceed on disagreeing data (mirrors rivet validate). unknown partitions (an incomplete chunk, or a non-integer keyset key with no source re-count) are surfaced as a warning but do not fail the command — “could not verify” is not “verified wrong”, and a keyset export is structurally all-unknown. The mismatch detail is always in the printed report regardless of exit code.

Reconcile reads from the same source and never writes files itself.

Parallel execution semantics

Rivet runs two distinct parallel engines — see ADR-0010 — Two Parallel Engines:

Both engines honour the same state invariants. Chunk checkpoints serialise the parallel threads’ state writes; subprocess children share one state DB — the parent migrates it once before spawning, and SQLite WAL (or the PostgreSQL state backend) handles the children’s concurrent writes.

Quality gates

When quality: is configured, the pipeline evaluates row-count, null-ratio, and uniqueness checks before advancing committed progression. A failing gate aborts the run with a non-zero exit code; the destination files remain (manual cleanup or replacement is the operator’s call). See docs/best-practices/quality-checks.md.

Destination commit boundaries

Full per-backend table and rationale: ADR-0004 — Destination Write Contracts.

Known non-guarantees

Rivet does not currently guarantee:

• Exactly-once delivery to the destination. Crashes between destination write and cursor advancement can produce duplicate files. Plan downstream dedup or idempotent ingestion — the manifest’s per-part content_fingerprint is the supported dedup key: identical rows produce byte-identical parts (and the same fingerprint) across rivet releases, so a duplicate is safely droppable by fingerprint. See recipes/idempotent-warehouse-load.md.
• Continuous / near-real-time replication. Rivet does capture CDC to files (mode: cdc — inserts/updates/deletes via a Postgres logical replication slot / MySQL binlog / SQL Server CDC change tables / MongoDB change streams, into typed Parquet/CSV — or the JSON-blob document image for MongoDB — resuming from the last committed log position each run), but it is not a continuously-running stream — changes are captured per invocation, not delivered live. For always-on near-real-time replication use Debezium or Estuary.
• Completeness of incremental cursors that can tie. Incremental resume uses a strict WHERE cursor > last_value. If two rows share the high-watermark value and the second becomes visible only after the run that advanced the watermark past it — e.g. a low-resolution updated_at (second granularity) or rows committed at the same timestamp after the read snapshot — the next run skips them and they are never exported. (Keyset pagination is unaffected: its key is planner-enforced unique + NOT NULL.) Use a strictly per-row-distinct, monotonic cursor (a sequence/identity id, or a timestamp with sub-value uniqueness); when the cursor can tie, re-snapshot the affected window with full/chunked mode.
• Automatic cleanup of an interrupted write’s temp file. A crash mid-write on the local destination may leave a dot-prefixed temp file in the target directory — never a partial final file (the commit is an atomic rename, so the final path is the complete file or absent). The stray temp file is harmless and can be removed manually.
• Schema migration handling. If the source schema changes between runs, Rivet does not migrate the destination; it surfaces a schema-drift error (see tests/live/live_schema_drift.rs).
• Correctness of user-authored SQL. Rivet executes query: verbatim. A query that omits a WHERE clause or selects from the wrong table will export the wrong data — there is no semantic validation.
• Protection from poorly indexed source queries. Preflight (rivet doctor, rivet check) warns about missing cursor indexes and unbounded ORDER BY, but it does not refuse to run. The operator decides.
• Stdout state safety. Using stdout with cursor or manifest state is technically allowed but not meaningful; plan validation rejects stdout + chunked and stdout + max_file_size before execution (ADR-0004 Known Gap).
• Atomicity across exports in one run. If a run has three exports and the second fails, the first export’s writes are already committed — the run does not roll back.
• Cross-run ordering when running in parallel from multiple machines against the same state DB. The default state DB is a local SQLite file; concurrent processes on one machine are handled via WAL, but multi-machine deployments should point RIVET_STATE_URL at a shared PostgreSQL state backend — cross-run ordering across machines is still not guaranteed.

Test coverage

The invariants on this page are exercised by:

• tests/invariants.rs — ADR-0001 I1–I7 structural contracts.
• tests/journal_invariants.rs — RunJournal event ordering.
• tests/recovery.rs — chunk checkpoint resume semantics (I5, I6).
• tests/live/live_crash_recovery.rs — crash-and-resume against a live database.
• tests/live/live_chunked_recovery.rs — chunked resume after partial completion.
• tests/live/live_retry_and_faults.rs — retry classification under injected faults.
• tests/live/live_reconcile_repair.rs — reconcile / repair end-to-end.

Invariants and recovery suites run as named semantic gates in PR CI (.github/workflows/ci.yml). Branch protection blocks merges on regression. See reliability-matrix.md for the full coverage matrix.

Rivet type mapping contract

This document describes how Rivet maps source column types to logical RivetType values and Arrow/Parquet/CSV representations. It is aligned with the automated suite in tests/type_roundtrip/ and tests/live_type_golden.rs.

The short version

Wondering “will my decimals / UUIDs / JSON / timestamps silently break on the way out?” — the short answer is no:

• Decimals never become floats — DECIMAL/NUMERIC export as exact Arrow Decimal128/Decimal256 when precision and scale are known.
• UUID and JSON keep native types — Parquet gets native LogicalType::Uuid / LogicalType::Json, not opaque strings.
• Timestamps preserve the instant — the point in time round-trips (Parquet keeps the zone; CSV emits the instant normalised to UTC with a trailing Z, while naive timestamps render bare).
• Rivet fails loud, not silent — a lossy or unmapped type is named by rivet check and aborts the run, never quietly truncated.

The per-engine tables below are the precise contract; this is just the gist.

Guarantees (v0.18.0)

• DECIMAL / NUMERIC are never silently converted to float. They export as Arrow Decimal128 / Decimal256 in Parquet when precision and scale are known (column override, catalog hint, or PostgreSQL wire metadata).
• Binary (BYTEA, BLOB, BINARY/VARBINARY with charset 63) stays Arrow Binary in Parquet.
• Float NaN / ±Infinity are preserved. Parquet stores them natively (IEEE-754); CSV emits the literal NaN / inf / -inf (and -0 keeps its sign) rather than an empty cell — an empty cell would silently conflate a real special value with NULL. Note these are float values: decimal / NUMERIC has no NaN representation and a NaN/Infinity payload there is rejected at extract time, not coerced. Strict CSV loaders that expect Infinity over inf should configure their float parser accordingly; Parquet needs no such care.
• JSON / JSONB is valid JSON text in the file, paired with the Arrow arrow.json canonical extension type so parquet-rs emits native LogicalType::Json in the Parquet footer. The Rivet field metadata (rivet.logical_type=json) stays for Rivet-aware consumers.
• UUID exports as canonical 16-byte FixedSizeBinary(16) paired with the Arrow arrow.uuid canonical extension; parquet-rs emits native LogicalType::Uuid. Downstream Parquet readers (DuckDB, ClickHouse 25.x+, pyarrow, BigQuery autodetect) recover the UUID type without a cast.
• Timestamps: TIMESTAMPTZ and MySQL TIMESTAMP use UTC semantics (timezone: Some("UTC")); naive TIMESTAMP / DATETIME have no timezone.
• Nullability is preserved in Arrow schema and export.

What rivet init auto-detects (and what needs an override)

rivet init reads the source database’s catalog / wire-protocol metadata to build the initial rivet.yaml. Whether a column lands as a native logical type in the resulting Parquet depends on whether the source server advertises the semantic — Rivet never guesses from column names or sample values, by design (a wrong guess is silent corruption; an honest “I don’t know” is a config knob).

Adding overrides looks like:

exports:
  - name: users
    query: "SELECT id, uid, amount FROM users"
    columns:
      uid: uuid              # MySQL VARCHAR(36) → UUID semantic
      amount: decimal(18,2)  # also valid for MySQL DECIMAL without catalog
      event_ts: timestamp_ns # keep SQL Server datetime2(7)'s 100 ns tick (see gap 4)

Supported override types: bool, int2/int4/int8, float4/float8, decimal(p,s), date, timestamp, timestamp_ns, timestamp_tz, timestamp_tz_ns, text, binary, json, uuid. The _ns timestamp variants preserve sub-microsecond precision (range 1677–2262; see gap 4) — the plain timestamp is microsecond with full date range.

The override path is safe: a uid: uuid declaration tries to parse each cell — either 16 raw bytes (BINARY(16) layout) or a 36-char canonical text form. Parse failures emit NULL, never silent garbage. Same convention as decimal overrides whose values do not parse cleanly.

rivet init does not apply heuristics (“column is 36 chars wide and named uid_* → probably UUID”) — guessing risks misclassifying a non-UUID varchar and silently producing wrong-shape Parquet. Operators who want UUID semantics on a MySQL column add the override explicitly.

Test commands

make test-types              # offline mapping contracts (PR-fast)
make test-types-live         # full matrix; requires docker compose
make test-types-validators   # PG/MySQL/SQL Server → Parquet → {DuckDB, ClickHouse} round-trip

test-types-validators re-runs the same canonical PG / MySQL / SQL Server type matrix used by test-types-live, but writes the Parquet into the shared bind-mount under tests/.live-tmp/ and feeds it through three independent readers: DuckDB, ClickHouse (both from docker-compose.yaml), and pyarrow (installed alongside duckdb in the same container). Each reader catches what the others cannot:

Local-reader type fidelity is now covered by the cross-tool harness’ type-loss matrix (dev/bench/smoke.py, rendered in report.html): every source column vs each tool’s Parquet type family. A BigQuery cloud-load type-diff (load each Rivet Parquet via bq load --autodetect, assert decimal sums round-trip) is not currently in the harness — it needs a GCP project + bq auth and is tracked for a future cloud dimension (notably the earlier findings were: LogicalType::Json does not autoload as native BQ JSON — it falls back to BYTES/STRING; values are valid JSON but operators need an explicit --schema='attrs:JSON,...' to query the structure).

In addition, two structural-only tests pin the Parquet layout itself — parquet_schema.rs for physical + logical types per column, and parquet_metadata.rs for the rivet.native_type / rivet.fidelity / rivet.logical_type field metadata.

Coverage extensions live in dedicated files: compression_matrix.rs re-runs the export under zstd / snappy / gzip / none and asserts value parity across all four codecs; csv_load.rs replays the matrix through DuckDB read_csv_auto; pg_edge_cases.rs covers decimal precision boundaries (38, 39 / Decimal128↔Decimal256), tz timestamps pre-epoch and far-future, JSON deep nesting + unicode keys + i64 edges, arrays with NULL elements, large single-cell strings.

v0.7.8 breaking change: UUID export layout

UUID columns (PG native uuid type and MySQL columns explicitly overridden as columns: { col: uuid }) now export as FixedSizeBinary(16) instead of hyphenated Utf8. This is what lets parquet-rs emit native LogicalType::Uuid so downstream readers (DuckDB, ClickHouse 25.x+, pyarrow, BigQuery autodetect) recover the UUID type without a cast.

What this changes:

• Parquet schema: BYTE_ARRAY + LogicalType::String → FIXED_LEN_BYTE_ARRAY + LogicalType::Uuid.
• On-disk bytes: 36-char canonical ASCII (a0eebc99-9c0b-…) → 16 raw bytes (the same UUID, compact encoding).
• CSV: unchanged — the CSV writer still emits hyphenated lowercase text.
• DuckDB autoload: VARCHAR → UUID native type. A view like SELECT uid::VARCHAR FROM read_parquet(...) keeps working because DuckDB’s UUID::VARCHAR cast yields the canonical form.
• ClickHouse 24.8 autoload: String → FixedString(16) (the bytes, not the text). To get back the canonical string, use lower(hex(uid)) and reformat, or upgrade to ClickHouse 25.x which decodes LogicalType::Uuid directly.

Consumers reading the old Utf8-shaped Parquet files keep working — only files produced by Rivet ≥ v0.7.8 carry the new layout.

Findings & fixes from triangulating against external readers

Driving the matrix through DuckDB + ClickHouse + pyarrow exposed three real defects in the PG / MySQL drivers; all have been fixed in v0.7.8:

• PG arrays with NULL elements were silently lost. The driver decoded ARRAY[1, NULL, 3] via try_get::<Vec<i32>>, which errors on a NULL element; the error was swallowed and a whole-row NULL was written. Fixed by deserializing as Vec<Option<T>> and pushing nulls through ListBuilder::append_null (src/source/postgres/arrow_convert.rs).
• MySQL ENUM / SET were misclassified as String. They arrive on the wire as MYSQL_TYPE_STRING / MYSQL_TYPE_VAR_STRING with the ENUM_FLAG / SET_FLAG set, not as MYSQL_TYPE_ENUM. The mapper now checks the flag and emits RivetType::Enum so the rivet.logical_type=enum Parquet metadata is preserved.
• MySQL native_type lost precision. tinyint unsigned, tinyint(1), bit(1), char vs varchar, binary vs varbinary all collapsed to a single label. The mapper now distinguishes them via column_type() + flags() + character_set.

Known limitations (pinned by *currently_fails* tests)

• PG numeric(p, -s) (negative scale) cannot be written to Parquet — the spec requires non-negative DECIMAL scale. Test pg_edge_decimal_negative_scale_currently_fails_at_parquet_write pins the failure with a friendly error message so any future workaround must update the test deliberately.
• DuckDB’s DECIMAL is capped at precision 38 (HUGEINT-backed). For Parquet files with precision > 38 DuckDB silently returns DOUBLE. Our file still carries LogicalType::Decimal(p, s) correctly — pyarrow decodes it as Decimal256. Verified in pg_edge_decimal_boundaries_round_trip.

PostgreSQL

MySQL

SQL Server (MSSQL)

Unmapped SQL Server types resolve to Unsupported and fail loudly at schema build unless a columns: override maps them.

MongoDB (JSON-blob model)

MongoDB has no fixed per-collection schema and no information_schema, so Rivet does not map per-field SQL types the way the three SQL engines do. Every document exports as exactly two columns:

Per-field typing is deferred to the warehouse (PARSE_JSON → VARIANT on Snowflake, native JSON on BigQuery / DuckDB). This is lossless and schema-drift-proof: a new field in a document never breaks a load. Schema inference / auto-discovery into typed columns is deliberately out of OSS scope. CDC (change streams) emits the same two-column shape prefixed with the __op / __pos / __seq meta columns. See reference/mongodb.md for the full contract.

Known gaps (tracked)

1. Nested arrays, ranges, inet, PostGIS, geometry: not in the type matrix — they resolve to Unsupported and fail at schema build unless a columns: override maps them.

2. Nullability: every exported column is OPTIONAL; a source NOT NULL constraint is not propagated into the Parquet schema (ADR-0016, deferred to v0.8 Phase A).

3. CSV complex types: arrays (List), Decimal256 (precision > 38), and non-UUID fixed binary have no CSV cell — the export fails loudly naming the column (see CSV serialization) rather than silently writing empty values.

4. SQL Server datetime2 sub-microsecond precision — default is microsecond; nanosecond is opt-in. rivet maps datetime2 to Timestamp(µs) by default, because Arrow nanosecond timestamps are i64 ns and span only 1677-09-21 .. 2262-04-11, while datetime2 spans 0001–9999 — a blanket ns mapping would silently corrupt any value outside that window (a far worse bug than the precision gap). So the 7th fractional digit of a datetime2(7) (100 ns) is truncated to µs by default; lossless for datetime2(6) and below.

   To preserve the 100 ns tick on a column whose data is inside the ns range, opt in per column with a timestamp_ns override:

   columns:
     event_ts: timestamp_ns       # naive; use timestamp_tz_ns for datetimeoffset
   

   The Parquet file then carries Timestamp(ns) and the full precision survives (verified live 2026-06-07: DuckDB reads it natively as TIMESTAMP_NS, …12:00:00.1234567 intact; the default µs path truncates to .123456). Caveats (all verified live 2026-06-07):

   • A value outside 1677–2262 exports as NULL (Arrow ns range).
   • DuckDB — native TIMESTAMP_NS, fully lossless.
   • Snowflake — autoloads as NUMBER(38,0) (raw nanos); TO_TIMESTAMP_NTZ(col, 9) recovers a lossless TIMESTAMP_NTZ (it holds 9 digits — the 7th survives).
   • BigQuery — autoloads as INT64 (raw nanos, lossless as an integer); a native TIMESTAMP_MICROS(DIV(col,1000)) is lossy (BigQuery TIMESTAMP is microsecond — the 7th digit drops). Keep the default timestamp for BigQuery unless you carry the raw nanos.

   Incremental mode: the default µs cursor on a datetime2(7) lands one tick below the source max, re-exporting the boundary row every run — use timestamp_ns (the cursor literal then carries all 9 digits), a datetime2(6) (or coarser) cursor, or an integer/identity cursor.

(MySQL DECIMAL now resolves its precision/scale from the wire column definition — no override needed; see the MySQL section.)

Fidelity labels

See src/types/fidelity.rs.

CSV serialization

CSV shares the same Arrow RecordBatch as Parquet, so values are identical — only the text rendering differs (src/format/csv.rs):

CSV has no cell representation for list (arrays), Decimal256 (precision > 38), or non-UUID fixed binary. Rather than silently write an empty value, the export fails at writer creation naming the column — use format: parquet or drop the column from the query.

Loading CSV into a warehouse: unlike Parquet (whose loader will not coerce a declared type — see below), BigQuery’s CSV loader honors a declared --schema. Bare --autodetect infers decimal text as FLOAT (precision-lossy) and every semantic type (uuid / json / bytea) as STRING; declare NUMERIC / JSON in the load schema, or recover post-load with PARSE_JSON / FROM_HEX as for Parquet.

Downstream targets (autoload vs native)

Parquet/CSV preserve values and Arrow physical types. Warehouse engines infer types from physical schema on autoload (e.g. JSON columns appear as STRING / VARCHAR). Native target types (JSON, VARIANT, UUID, …) require a materialization step (cast SQL, load schema, or typed view).

See ADR-0014: Target type materialization for the full matrix (DuckDB, BigQuery, Snowflake, ClickHouse) and planned rivet check --type-report --target <engine> extensions.

Verified physical autoload (DuckDB + ClickHouse, v0.18.0)

make test-types-validators re-reads every PG / MySQL matrix column through two independent engines and pins the autoload type. The full matrix:

Round-trip values that the suite pins per row: sum(decimal * 10^scale) matches the in-process Arrow check; json_valid / isValidJSON returns true for every row; TRY_CAST(uid AS UUID) / toUUID(uid) parses every row; BLOB columns compare byte-for-byte (hex(...)); the empty list survives; null bitmaps propagate. The MySQL BIGINT UNSIGNED max value (2^64 − 1) is the load-bearing assertion that exact-width unsigned ints are not silently overflowed to i64.

BigQuery autoload & recovery (verified live)

BigQuery’s Parquet loader is weaker than DuckDB’s: it ignores several Parquet logical types on autoload and — critically — will not coerce a column to a different declared type on load. A bq load into a table that declares JSON/DATETIME is rejected (Field x has changed type from JSON to BYTES), so native types are recovered with a post-load transform, not a load schema.

Rivet writes the Parquet list element as item (arrow-rs default, not the spec’s element), so even --parquet_enable_list_inference yields REPEATED RECORD{item} rather than a clean REPEATED <scalar> — the UNNEST flatten above is required.

rivet check --type-report --target bigquery prints the per-column autoload type, the native type, and a ready-to-run recovery CREATE TABLE … AS SELECT over the autoloaded <table>__staging. Set exports[].target: bigquery to get it without the CLI flag. DuckDB needs none of this — it autoloads every logical type natively.

Snowflake autoload & recovery (verified live)

Snowflake’s INFER_SCHEMA + COPY infers physical types only, so the same semantic types degrade — and the INFER_SCHEMA column names come back lowercase and case-sensitive, so the recovery SELECT must double-quote every source reference ("col").

The load preamble the recovery depends on: CREATE FILE FORMAT … TYPE=PARQUET BINARY_AS_TEXT=FALSE, ALTER SESSION SET TIMEZONE='UTC', CREATE TABLE … USING TEMPLATE (… INFER_SCHEMA …), COPY … MATCH_BY_COLUMN_NAME=CASE_INSENSITIVE.

rivet check --type-report --target snowflake (--target sf) emits the per-column autoload/native types and the post-load CREATE OR REPLACE TABLE … recovery over <table>__staging. Set exports[].target: snowflake to skip the flag.

Value-Based Output Partitioning (partition_by)

When to use

Set partition_by to split one export’s rows into one destination sub-folder per value bucket of a column — the Hive-style col=value/ layout that warehouses and query engines (Snowflake external tables, BigQuery external / Hive-partitioned tables, Spark, DuckDB, Athena) discover automatically. Best for:

• Daily/monthly snapshots that land each day’s rows under its own prefix
• Backfilling history into a partitioned lake layout in one command
• Feeding a warehouse that prunes partitions by a date column

partition_by is orthogonal to mode: each partition runs the export’s own mode, so mode: chunked chunks within a partition.

Required fields

• partition_by — the column whose value buckets the rows (a DATE / TIMESTAMP / TIMESTAMPTZ column).
• a {partition} token in destination.path or destination.prefix — Rivet refuses the run without it, because every partition would otherwise overwrite the same prefix.

Optional fields

• partition_granularity — day (default), month, or year.

Minimal config

source:
  type: postgres
  url: "postgresql://user:pass@host:5432/dbname"

exports:
  - name: events
    table: events
    partition_by: created_at
    partition_granularity: day
    format: parquet
    destination:
      type: s3
      bucket: my-bucket
      prefix: "events/{partition}/"      # → events/created_at=2023-01-01/

Run it

rivet run --config events.yaml --validate

What happens

1. Rivet reads the [min, max] span of partition_by from the source (SELECT min(col), SELECT max(col) over your query).
2. It generates one contiguous bucket per day/month/year across that span.
3. Each bucket becomes its own export: the query is wrapped as SELECT * FROM (<your query>) WHERE col >= '<lo>' AND col < '<hi>' (half-open: lo inclusive, hi exclusive — no row counted twice), and {partition} resolves to col=value.
4. Rows whose partition_by value is NULL land in col=__HIVE_DEFAULT_PARTITION__/ (the Hive default-partition convention) so no row is ever silently dropped.

Each partition is an independent, complete output prefix — its own manifest.json and _SUCCESS — so it can be validated and consumed on its own.

Example output layout

events/
  created_at=2023-01-01/  manifest.json  _SUCCESS  events__2023-01-01_*.parquet
  created_at=2023-01-02/  manifest.json  _SUCCESS  events__2023-01-02_*.parquet
  ...
  created_at=__HIVE_DEFAULT_PARTITION__/  manifest.json  _SUCCESS  ...

DuckDB reads the whole tree as one partitioned dataset and recovers the partition column from the path:

SELECT created_at, count(*)
FROM read_parquet('events/**/*.parquet', hive_partitioning = true)
GROUP BY 1;

Granularity

Notes and limits

• Cloud prefixes: put a / after {partition}. Object stores have no directories — the part filename is appended to the resolved prefix verbatim. prefix: "events/{partition}/" yields events/created_at=2023-01-01/part.parquet; omitting the trailing slash concatenates into …created_at=2023-01-01part.parquet. (Local path: joins as directories, so the slash is optional there.)
• Index the partition column. Expansion runs three probes over the base query (min, max, NULL-count). Without an index on partition_by those are up to three full scans before the first row is exported.
• Time zones. Bucket bounds are emitted as YYYY-MM-DD literals on PostgreSQL and MySQL; SQL Server gets the unseparated YYYYMMDD form, which T-SQL parses as ISO regardless of the session’s DATEFORMAT/language. For a TIMESTAMPTZ column the comparison happens at the session time zone — pin it (e.g. MySQL SET time_zone = '+00:00') when the exact day boundary matters.
• Not compatible with mode: time_window (time_window already filters by a rolling window). Use partition_by with full, chunked, or incremental.
• Not compatible with mode: cdc (CDC reads the log, not a query), a load: block (per-export or top-level: the loader would load a single partition’s manifest), or a MongoDB source (the bucket probes are SQL).
• Not compatible with chunk_by_key (keyset). Each partition reads its bucket as a subquery around the table, and keyset seek pagination over that shape is not supported — Rivet rejects the combination up front. A table: export keeps its table per partition, so a range chunk_column is type-checked and an unset one auto-resolves to the primary key as usual.
• --parallel-export-processes is disabled while partitioning is active (child processes re-load the config and can’t see the synthesised partitions); the run executes in-process.
• plan / check do not expand partitions yet. They report the parent export as one un-partitioned job (its {partition} token stays literal in the shown path, the row estimate is the whole span, strategy is the base mode). Treat their output as the per-partition shape, not the campaign.
• Validating a single partition today: point validate at the concrete prefix — rivet validate --config c.yaml --export events --prefix events/created_at=2023-01-01. Validating every partition of an export by its parent name in one command is not yet wired.

Choosing a chunking strategy per partition (large partitions)

partition_by is orthogonal to mode, so each partition runs the export’s mode inside itself. The right choice depends on how big a single partition is and whether the chunk key is dense within it. Consider a date that holds 100 M rows:

Row counts are always exact regardless of the choice — these trade-offs are about source load and file fan-out, not correctness. There is no keyset/seek option for a partition today (see the chunk_by_key limit above), so for a genuinely huge partition with a sparse key the practical levers are a correlated range key or a finer granularity.

Export modes — which one do I pick?

Every export declares a mode:. Start from the decision shortcut, then open the guide for the mode you land on.

Decision shortcut

• Small-to-medium table, want a fresh snapshot every run → full
• Append-only or has an updated_at, want only new/changed rows → incremental
• Millions+ of rows, want speed and crash-resume → chunked
• Only the last N days matter (event/log table) → time_window
• Continuous low-latency replication from the WAL / binlog / oplog → cdc

When in doubt, rivet init inspects the table and picks a sensible default for you (small → full, large with an integer key → chunked).

At a glance

Notes worth knowing before you run:

• incremental first run = full export. With no cursor yet, every matching row is exported, so the first run behaves like full — size batch_size accordingly (incremental.md).
• chunked clean re-runs are NOT idempotent. A crash + --resume is at-least-once: a re-run chunk can be written twice (byte-identical), so de-duplicate downstream if you re-run (chunked.md).
• Composite cursors are an incremental variant, not a separate mode — see incremental-coalesce.md when one timestamp column isn’t enough.
• Source engine matters. The SQL sources (PostgreSQL, MySQL, SQL Server) support every mode. MongoDB, a document store, supports only full (with keyset / parallel / resume paging on _id) and cdc — see ../reference/mongodb.md.

Full configuration reference: ../reference/.

Full Export Mode

When to use

Use mode: full when you want a complete snapshot of the query result set every time. Each run re-exports all rows from scratch. Best for:

• Small-to-medium tables (up to a few million rows)
• Reference/dimension tables that need a fresh copy each day
• One-time data migrations

MongoDB. full is MongoDB’s primary batch mode — the SQL-runner modes (incremental / chunked / time_window) do not apply to a document store. Within full, MongoDB adds keyset (seek) paging, parallel: N _id-range fan-out, and resume — all keyed on _id (source.mongo.page_size / parallel). See ../reference/mongodb.md.

Minimal config

source:
  type: postgres                                    # postgres, mysql, mssql, or mongo
  url: "postgresql://user:pass@host:5432/dbname"

exports:
  - name: users_daily                               # unique export name
    query: "SELECT id, name, email, created_at FROM users"
    mode: full                                      # re-export everything each run
    format: parquet                                 # parquet or csv
    destination:
      type: local
      path: ./output                                # directory for output files

Output file: ./output/users_daily_20260406_120000_123.parquet (the timestamp ends with a 3-digit millisecond field, e.g. _123, so rapid re-runs never collide)

Run it

# 1. Verify config and connectivity
rivet check --config users.yaml
rivet doctor --config users.yaml

# 2. Run with validation
rivet run --config users.yaml --validate --reconcile

# 3. Check results
rivet metrics --config users.yaml --last 5

What happens

1. Rivet connects to the source database
2. Executes SELECT id, name, email, created_at FROM users
3. Fetches rows in batches controlled by tuning.batch_size (default 10,000 with balanced profile)
4. Writes a timestamped output file
5. Records metrics in the state database

No cursor is stored. Each run produces a new file with all rows.

batch_size directly controls memory usage and source load. For wide tables (many columns, TEXT/JSONB fields), reduce it to 1,000-5,000. See reference/tuning.md.

Common options

exports:
  - name: users_daily
    query: "SELECT * FROM users"
    mode: full
    format: parquet
    compression: zstd           # zstd (default), snappy, gzip, lz4, none
    skip_empty: true            # a 0-row run reports `skipped`, not `success`
    max_file_size: "512MB"      # split into multiple files if output exceeds this
    meta_columns:
      exported_at: true         # add _rivet_exported_at column
    destination:
      type: local
      path: ./output
    tuning:
      profile: safe             # safe/balanced/fast — controls batch size, timeouts

Troubleshooting

Export is slow on a large table – Switch to mode: chunked with parallel: 4 for tables over 1M rows. See chunked.md.

Output file is too large – Add max_file_size: "256MB" to split into parts.

A run that read 0 rows reports success – Add skip_empty: true to record it as skipped. No file is written for 0 rows either way, and a skipped run leaves the prefix describing the last run that delivered — so a later full rivet load keeps the previous data rather than emptying the table.

Incremental Export Mode

When to use

Use mode: incremental when you only want to export rows that are new or updated since the last run. Best for:

• Append-only tables (events, logs, audit trails)
• Tables with a reliable updated_at timestamp
• Tables with a monotonically increasing ID
• Daily/hourly syncs where re-exporting everything is wasteful

SQL sources only (PostgreSQL, MySQL, SQL Server). incremental does not apply to MongoDB, a document store — use full or cdc there (../reference/mongodb.md).

Required fields

• cursor_column – the column used to track progress (must be monotonically increasing)

Minimal config

source:
  type: postgres
  url: "postgresql://user:pass@host:5432/dbname"

exports:
  - name: orders_incremental
    query: "SELECT id, user_id, product, price, status, updated_at FROM orders"
    mode: incremental
    cursor_column: updated_at       # tracks last exported value
    format: parquet
    destination:
      type: local
      path: ./output

Run it

# First run — exports all rows (no cursor yet)
rivet run --config orders.yaml --validate --reconcile

# Second run — only exports rows with updated_at > last cursor
rivet run --config orders.yaml --validate

# Check current cursor position
rivet state show --config orders.yaml

# Reset cursor to re-export everything
rivet state reset --config orders.yaml --export orders_incremental

What happens

1. First run: no cursor exists, so all rows matching the query are exported
2. Rivet records the maximum value of cursor_column as the cursor
3. Subsequent runs: Rivet wraps your query in a subquery, adds WHERE updated_at > <last cursor>, and orders by the cursor column (the cursor is an inlined literal on Postgres/SQL Server — Postgres uses the escaped E'…' form — and a bind parameter on MySQL)
4. Only new/updated rows are exported; cursor advances after successful write

Run 1 (no cursor):  SELECT * FROM (SELECT ... FROM orders) AS _rivet ORDER BY "updated_at"
                     → 5000 rows, cursor saved: 2026-04-05 23:59:59
                     (the sort applies even to the full first run — an index
                      on the cursor column matters)

Run 2 (with cursor): SELECT * FROM (SELECT ... FROM orders) AS _rivet
                       WHERE "updated_at" > E'2026-04-05 23:59:59' ORDER BY "updated_at"
                      → 47 rows (only changes since last run)

Cursor column tips

The cursor column must be:

• Present in the SELECT clause
• Monotonically increasing (new rows always have a larger value)
• Not NULL for rows you want exported

Batch size and tuning

Even in incremental mode, Rivet fetches rows in batches (not all at once). The batch_size from tuning: controls how many rows are fetched per FETCH call:

source:
  type: postgres
  url_env: DATABASE_URL
  tuning:
    batch_size: 5000            # rows per fetch (default: 10,000 for balanced)

exports:
  - name: orders_incremental
    query: "SELECT id, user_id, product, price, updated_at FROM orders"
    mode: incremental
    cursor_column: updated_at
    format: parquet
    destination:
      type: local
      path: ./output
    tuning:
      batch_size: 2000          # per-export override (takes precedence)

On the first incremental run (no cursor yet), all rows are exported. If the table has millions of rows, this first run behaves like a full export — so batch_size directly impacts memory and source load. Use a smaller batch_size (1,000-5,000) for wide tables or production databases.

See reference/tuning.md for all tuning parameters.

Common options

exports:
  - name: orders_incremental
    query: "SELECT id, user_id, product, price, updated_at FROM orders"
    mode: incremental
    cursor_column: updated_at
    format: parquet
    skip_empty: true            # a run with no new rows reports `skipped`
    meta_columns:
      exported_at: true         # add _rivet_exported_at for dedup downstream
    destination:
      type: local
      path: ./output

Switching to incremental

The usual path is a full load first, then mode: incremental on the same export. What the first incremental run does depends on the export’s previous mode (ADR-0033, matrix):

rivet load follows what each run holds. A run that re-read the whole table — a full load, or an incremental export’s first run — lands as a plain <table>. The first delta renames that table to <table>__changes, adds __op / __pos / __seq (NULL on the rows it already held) and makes <table> a view over it; the log keeps the table’s partitioning and clustering, and nothing is copied or dropped. A whole-table load onto a table rivet did not load, one whose partitioning or clustering differs from the config, or a view left by an earlier incremental load fails naming the difference and changes nothing — drop or rename the table, or align the config. The rename refuses the same way when the table’s columns differ from the export’s or a <table>__changes already exists beside it.

Settle window

Some rows keep changing for a while after they are inserted, and no column records when: a page view’s time-on-page arrives with the next hit, a session’s totals grow until it closes. A cursor exports such a row as soon as it appears, before its final values exist, and never sees the later write.

settle holds each row back until it is older than after by the source clock:

exports:
  - name: page_views
    table: page_views
    mode: incremental
    cursor_column: id              # a cheap primary-key range per run
    settle:
      column: server_time          # the row's insert time
      after: 1h                    # longer than the time the row keeps changing

• after takes s, m, h or d. The warehouse copy lags the source by that much.
• Without column, the cursor itself is aged (cursor_column, or the COALESCE in coalesce mode) — the right choice for an updated_at cursor.
• With a separate column, the cursor also stops below the first row that is still settling, so a row committed out of id order is never skipped.
• A row whose settle value is NULL never ages, so it is exported without waiting.
• The column must be a date or timestamp; zone-less values are compared as UTC.
• Deletes stay invisible to a cursor. Use mode: cdc when they must reach the warehouse.

Transactions that commit during a run

Each incremental read must see a transaction either whole or not at all. PostgreSQL and MySQL (InnoDB) read every statement from one snapshot, so they do. SQL Server does only when the database has a row-versioning option on:

Under locking READ COMMITTED a scan can pass a row, wait on a row a writer holds, and read it once the writer commits, so the scan returns that transaction half-applied. The cursor then moves past the rows it had already passed, and no later run reads them. Enable ALLOW_SNAPSHOT_ISOLATION (ALTER DATABASE [<db>] SET ALLOW_SNAPSHOT_ISOLATION ON), or set a settle window longer than your longest write transaction.

A snapshot does not cover everything. A transaction that stamps its rows, commits late, and ends up with a cursor value below a row that committed earlier is skipped on every engine, since each read already moved past it. settle guards that race: set after longer than your longest write transaction.

Troubleshooting

the stored cursor ... was written for ... – The export’s cursor changed (a new cursor_column, or a keyset chunk_by_key export switched to incremental on another column). The old value means nothing for the new column; rivet state reset --config ... --export <name> starts the new cursor with a full pass.

A run with no new rows reports success – Add skip_empty: true to record it as skipped (no file is written for 0 rows either way).

Data appears duplicated across runs – Ensure cursor_column updates when rows are modified. If rows are updated without changing updated_at, they will be missed.

Need to re-export all data – rivet state reset --config ... --export <name> clears the cursor.

rivet apply fails with invalid configuration: password missing – A prior rivet plan silently stripped the plaintext password: from the artifact (ADR-0005 PA9). Migrate to password_env: DB_PASSWORD in the config and re-generate the plan. See the WARN line in the plan output.

Composite cursor (nullable primary)

If your primary column can be NULL for some rows (e.g. updated_at only set on updates), see incremental-coalesce.md — progression switches to COALESCE(primary, fallback) via incremental_cursor_mode: coalesce.

Incremental — Composite Cursor (coalesce mode)

See also: ADR-0007 — Cursor Policy Contracts.

When to use

Use incremental_cursor_mode: coalesce when a single column is not a reliable monotonic key, but combining two columns with COALESCE(primary, fallback) is.

Typical situations:

• updated_at is nullable for some rows — progression must fall back to created_at.
• A partial migration left some rows with only a created_at, others with both.
• You intentionally write updated_at only on changes and need a baseline for new rows.

If your primary column is reliably non-null and monotonic, stay with plain incremental — it is simpler, faster, and has fewer moving parts.

Required fields

Config validation rejects:

• coalesce without cursor_fallback_column
• cursor_fallback_column set for any mode other than coalesce

Minimal config

source:
  type: postgres
  url_env: DATABASE_URL

exports:
  - name: orders_coalesce
    query: "SELECT id, product, quantity, price, updated_at, created_at FROM orders"
    mode: incremental
    cursor_column: updated_at
    cursor_fallback_column: created_at
    incremental_cursor_mode: coalesce
    format: parquet
    destination:
      type: local
      path: ./output

What Rivet runs

A single-level wrapper with an outer ORDER BY on the coalesced expression (so the last Arrow batch carries the maximum progression value; ADR-0007 CC6):

SELECT _rivet.*,
       COALESCE(_rivet."updated_at", _rivet."created_at") AS "_rivet_coalesced_cursor"
FROM (<your query>) AS _rivet
WHERE COALESCE(_rivet."updated_at", _rivet."created_at") > '<last cursor>'
ORDER BY COALESCE(_rivet."updated_at", _rivet."created_at"),
         _rivet."updated_at",
         _rivet."created_at"

On the first run (no stored cursor) the WHERE is omitted.

State and output

• Stored cursor — a single scalar string, the max value of COALESCE(primary, fallback) from the last exported batch (ADR-0007 CC5; same StateStore shape as regular incremental).
• Output files — the synthetic _rivet_coalesced_cursor column is stripped before writing Parquet/CSV. Your output contains only your selected columns.

Apply semantics

Apply uses the cursor snapshot embedded in the plan artifact (ADR-0005 PA4). The comparison is string-wise and mode-agnostic — coalesce does not change the check, only the meaning of the opaque string.

Quoting and escaping

Rivet quotes cursor_column and cursor_fallback_column using the source dialect ("…" for Postgres, `…` for MySQL). On MySQL the cursor value is sent as a ? bind parameter (never inlined into the SQL). On Postgres it is emitted as an E'…' string literal with ' and \ backslash-escaped (O'Brien → E'O\'Brien'). No additional escaping is required on your side.

Caveats

• COALESCE is not indexed — on very large tables the planner may not use the indexes on updated_at or created_at. Preflight (rivet plan) will flag this in diagnostics.
• Monotonicity is best-effort — if new rows can appear with a created_at strictly less than the latest COALESCE(...) already seen, they will be skipped. Prefer setting updated_at on insert when feasible.
• One fallback only — two-level lexicographic cursors (a, b) and more than one fallback are out of scope for v1 (ADR-0007).

Trying it locally

The dev seed (cargo run --bin seed) populates a fixture table orders_coalesce in both Postgres and MySQL with a configurable NULL ratio for updated_at:

# Uses dev/{postgres,mysql}/init.sql + seed defaults (~35% NULL updated_at).
cargo run --bin seed -- --target both --coalesce-rows 2000 --coalesce-null-ratio 0.35

# Then:
rivet plan --config dev/workbench/pg_incremental.yaml --export pg_orders_coalesce
rivet run  --config dev/workbench/pg_incremental.yaml --export pg_orders_coalesce

Troubleshooting

Stored cursor goes backwards — a row was inserted with created_at older than the last seen COALESCE value. Options: set updated_at at insert time, or reset cursor via rivet state reset.

Empty output on second run but new data exists — check that the predicate expression matches the data distribution. rivet plan shows the exact SQL form.

cursor_fallback_column rejected — you need incremental_cursor_mode: coalesce to enable the fallback. Without it, only the primary is used (see incremental.md).

Chunked Export Mode

When to use

Use mode: chunked when the table is too large for a single full export. Rivet splits the data into ranges by a numeric ID column and can process multiple chunks in parallel. Best for:

• Tables with millions or billions of rows
• Tables with a numeric primary key (BIGINT, SERIAL)
• When you need parallel extraction to save time
• Initial loads of large tables

Required fields

• chunk_column – a numeric, date, or timestamp column to partition by (typically the primary key). Auto-resolved from the single-integer primary key if you use the table: schema.name shortcut (works on Postgres, MySQL, and SQL Server) — warn-level log line: export 'orders_chunked': chunk_column not set — auto-resolved to 'id' from the single-integer primary key on public.orders. Set `chunk_column:` explicitly to pin the choice and silence this warning.

Chunking strategies — pick one

Four ways to slice the table. They differ in how chunk boundaries are computed; everything below the strategy line (parallel, checkpoint, retry) is orthogonal and combines with any of them.

Orthogonal options that combine with any strategy:

Picking parallel. Extraction is I/O-bound, so the win comes from overlapping FETCH round-trips, not from CPU. Measured on a 10-core host, 2 M-row tables: a narrow table scales near-linearly (1→4 ≈ 4.2× faster), while a wide table (many large columns) saturates the wire early and plateaus at parallel: 2 (2→4 buys almost nothing for +60 % RSS). Each worker holds its own chunk buffer, so RSS grows roughly linearly with N — raise it for narrow tables, keep it at 2 for wide ones. The init heuristic is a good starting point; tune from there if memory or source connection count is constrained.

Each integer-range chunk runs: SELECT * FROM (<base_query>) AS _rivet WHERE <chunk_column> BETWEEN <lo> AND <hi> — inclusive bounds (hi = lo + chunk_size - 1) inlined as literals, not bind parameters. The subquery wrap applies to query: exports; a table: shortcut renders the unwrapped SELECT * FROM <table> WHERE <chunk_column> BETWEEN <lo> AND <hi>. The date variant (chunk_by_days) uses half-open WHERE col >= '<start>' AND col < '<end>'.

Minimal config

source:
  type: postgres
  url: "postgresql://user:pass@host:5432/dbname"

exports:
  - name: orders_chunked
    query: "SELECT id, user_id, product, price, status, ordered_at FROM orders"
    mode: chunked
    chunk_column: id                # numeric column to split ranges on
    chunk_size: 100000              # rows per chunk (default: 100,000)
    parallel: 4                     # concurrent chunk workers
    format: parquet
    destination:
      type: local
      path: ./output

Output files: one per chunk, named {export}_{YYYYMMDD_HHMMSS}_chunk{N}_{16-hex-nonce}.parquet, e.g. orders_chunked_20260406_120000_chunk0_a1b2c3d4e5f60718.parquet. The random nonce makes retried/re-run parts additive (never overwriting); match on *_chunk{N}_*.parquet, not on an exact stem.

Run it

# Preflight — shows chunk plan (how many chunks, range distribution)
rivet check --config large_table.yaml

# Run with validation and reconciliation
rivet run --config large_table.yaml --validate --reconcile

Progress bar (chunked exports)

In mode: chunked, Rivet shows a terminal progress bar while chunks run: export name, current/total chunks, running row count, elapsed time, and ETA. It appears when stderr is an interactive TTY (a normal terminal window). The bar does not depend on RUST_LOG (that variable only controls env_logger text lines). In CI, or when you pipe or redirect stderr, the bar is usually suppressed — then set RUST_LOG=info (or debug) to follow progress in the log instead.

The GIF above was recorded with RUST_LOG=info on a 50,000-row fixture (10 chunks of 5,000) so the per-chunk log line export 'events': chunk N/10 (...) and the final summary are both visible. On a real interactive terminal you would see the progress bar instead; the log lines appear when stderr is captured.

Use a small chunk_size relative to your table if you want many steps on the bar (each finished chunk advances it once). parallel: 1 still updates the bar after each sequential chunk.

Ready-made example in this repo: dev/scenarios/chunked_postgres_bench.yaml includes bench_content_p4_safe: PostgreSQL content_items with parallel: 4 and tuning.profile: safe (good for trying the bar on a wide table without hammering the source). Other exports in the same file cover serial / highly parallel / fatchunk / balanced profiles.

# From repo root; Postgres up + seeded (e.g. docker compose + cargo run --bin seed ...)
mkdir -p dev/output/bench
rivet check --config dev/scenarios/chunked_postgres_bench.yaml
rivet run --config dev/scenarios/chunked_postgres_bench.yaml --export bench_content_p4_safe
# Optional: RUST_LOG=info for more log detail; RUST_LOG=warn to reduce log noise (bar unchanged in a TTY)

What happens

1. Rivet queries SELECT MIN(id), MAX(id) FROM orders to determine the range
2. Splits into chunks: [min..min+chunk_size), [min+chunk_size..min+2*chunk_size), …
3. Each chunk runs independently: SELECT ... WHERE id BETWEEN <lo> AND <hi> (inclusive, hi = lo + chunk_size - 1)
4. With parallel: 4, up to 4 chunks execute concurrently
5. Each chunk writes a separate output file

Chunk checkpoint (resume after crash)

For very large exports, enable checkpointing so you can resume from where you left off:

exports:
  - name: orders_chunked
    query: "SELECT id, user_id, product, price, ordered_at FROM orders"
    mode: chunked
    chunk_column: id
    chunk_size: 100000
    parallel: 4
    chunk_checkpoint: true          # persist progress per chunk
    chunk_max_attempts: 3           # total attempt budget per chunk (3 attempts = 2 retries)
    format: parquet
    destination:
      type: local
      path: ./output

Resume after a crash:

# Resume only processes incomplete chunks
rivet run --config large_table.yaml --resume

# View checkpoint status
rivet state chunks --config large_table.yaml --export orders_chunked

# Clear checkpoint (to re-export from scratch)
rivet state reset-chunks --config large_table.yaml --export orders_chunked

Clean re-runs are NOT idempotent

Chunked mode is not “extract once, skip on the next clean run”. Two plain rivet run invocations against the same table re-extract every chunk both times — chunk_checkpoint: true only matters after a crashed run, which the next run resumes. Each clean run produces a new file set with a fresh run_id and timestamp suffix.

If you want skip-on-no-change semantics, use mode: incremental with a cursor_column instead — that mode persists the cursor between runs, and skip_empty: true records a run that found nothing new as skipped.

Chunk sizing guidance

Larger chunks = fewer queries but more memory per batch. Smaller chunks = more queries but lower peak RSS.

Date-based chunking

When your table’s natural partition boundary is time rather than a numeric ID, use chunk_by_days instead of relying on integer ranges.

exports:
  - name: orders_by_year
    query: "SELECT id, user_id, product, price, ordered_at FROM orders"
    mode: chunked
    chunk_column: ordered_at        # DATE or TIMESTAMP column
    chunk_by_days: 365              # one chunk per ~year
    format: parquet
    destination:
      type: local
      path: ./output

Rivet fetches MIN / MAX of the column as text, parses the dates, then generates non-overlapping windows:

-- each chunk window (open-end exclusive):
WHERE ordered_at >= '2023-01-01' AND ordered_at < '2024-01-01'
WHERE ordered_at >= '2024-01-01' AND ordered_at < '2025-01-01'
...

The open-end < end_date bound is intentional: it correctly captures all TIMESTAMP values within the day, including 23:59:59.999….

When to use date chunking over numeric chunking:

• The table has no dense numeric PK (UUIDs, composite keys)
• You want even partitions by time, not by row count
• The source DB has better statistics / indexes on the timestamp column
• You want to avoid unix-epoch arithmetic that JDBC tools often get wrong

chunk_by_days can be combined with parallel for concurrent date windows, and supports chunk_checkpoint / --resume like numeric chunked mode.

rivet check will report the strategy as date-chunked(ordered_at, 365d).

Sparse ID ranges

If IDs have large gaps (e.g. UUIDs cast to BIGINT, or deleted rows), many chunks may be empty. On a unique key use keyset (chunk_by_key, below) — it pages by rows, so gaps cost nothing. Otherwise chunk_count: N caps the number of windows.

rivet check will warn you about sparse ranges.

chunk_dense was removed. It paged by ROW_NUMBER() OVER (ORDER BY chunk_column), recomputed per chunk, so concurrent inserts or deletes skipped or duplicated rows even on a unique key. A config that still sets chunk_dense: true is refused at load.

Keyset (seek) pagination — the safe shape without an integer PK

Range chunking needs a single integer PK to slice MIN..MAX. A MySQL table whose PK is a UUID, string, or composite key has no such column, and — unlike PostgreSQL — MySQL has no server-side cursor to bound a mode: full snapshot. That left a real hole: such tables could only be exported as one long-held SELECT * (the exact “don’t hold a long query on prod” risk Rivet exists to avoid).

MongoDB has the same shape, but under mode: full (a document store has no chunked mode): source.mongo.page_size enables keyset (seek) paging on _id, parallel: N fans out over disjoint _id ranges, and both resume on _id. See ../reference/mongodb.md.

Keyset pagination closes it. Rivet pages the table by a unique, NOT NULL, index-backed key:

-- first page
SELECT * FROM (<base>) AS _rivet ORDER BY `uid` LIMIT 1000
-- subsequent pages (cursor = last page's max key)
SELECT * FROM (<base>) AS _rivet WHERE `uid` > ? ORDER BY `uid` LIMIT 1000

Each page is a bounded, index range scan (verified EXPLAIN: type: range on the PK, no filesort, no full scan) and becomes one part file. This bounds both peak RSS (≤ chunk_size rows in flight) and longest-query time.

exports:
  - name: events
    table: app.events            # `table:` shortcut required (index check)
    mode: chunked
    chunk_by_key: event_uuid     # single-column UNIQUE / PRIMARY, NOT NULL
    chunk_size: 1000             # rows per page
    format: parquet
    destination:
      type: local
      path: ./output

Output files: one per page, named {export}_{run_id}_keyset_{tag}.parquet where run_id is {export}_YYYYMMDDTHHMMSS.mmm (filename sanitization maps the . to _) and the tag is start for the first page, then a 16-hex hash of that page’s seek cursor — e.g. events_events_20260529T120000_123_keyset_start.parquet. The run_id/seek-based name makes a crash-resume overwrite its own page idempotently.

Auto-resolution (MySQL). With the table: shortcut and no chunk_by_key, if the table has no single-integer PK but does have a usable single-column unique key, Rivet auto-selects keyset on it and logs a warn naming the key (set chunk_by_key: to pin the choice and silence the warning). On PostgreSQL, auto-resolution stays off — its DECLARE CURSOR snapshot is already bounded, so mode: full is the safe answer there; chunk_by_key: still works if you want per-page files.

The key must be index-backed. This is the load-bearing safety property: an ORDER BY on a non-indexed column degrades to a full-scan + filesort — worse than the snapshot it replaces. Rivet refuses a chunk_by_key that is not a single-column, NOT NULL, UNIQUE/PRIMARY key rather than emit such a query:

chunk_by_key 'payload' is not a usable keyset key on app.events — it must be a
single-column, NOT NULL, UNIQUE or PRIMARY key WHOSE TYPE the keyset cursor can
read (integer / float / string / timestamp / date / uuid). A `decimal`/`numeric`
key is excluded: the cursor cannot advance past it (it would fail mid-run after
a partial write). Without a usable key, `ORDER BY payload LIMIT n` would also
full-scan + filesort. Add a unique index of a supported type, pick another key,
use a range `chunk_column:` (integer), or `mode: full`.

Required privileges: read-only is sufficient — the introspection probe reads information_schema index metadata, no elevated grants needed.

Resumability (chunk_checkpoint). rivet init defaults chunk_checkpoint: true on keyset exports. It is crash-recovery: a run that dies mid-stream resumes from its last committed key on the next run (its in-progress run_id is still open); a run that finished CLEANLY clears that marker, so a plain re-run does a full pass and never silently skips already-exported rows.

Append-only incremental (keyset_incremental). Off by default. When set, a CLEAN re-run continues from the last exported key — pulling ONLY rows past the high-water mark. Correct only for append-only tables: on a mutable table a row whose key already passed is silently never re-read. For a mutable table use mode: incremental on a timestamp cursor instead.

    chunk_by_key: event_uuid
    chunk_checkpoint: true       # crash-recovery (default on for keyset)
    keyset_incremental: true     # append-only ONLY: clean re-run pulls just new keys

Parallel keyset (parallel: N)

By default keyset pages sequentially — each page seeks past the previous page’s last key. Set parallel: N and Rivet instead splits the key into N ROW-based percentile ranges and seeks each range concurrently on its own connection:

    chunk_by_key: event_uuid
    chunk_size: 1000
    parallel: 4                  # 4 workers, each seeks a disjoint key range

The ranges are half-open ([lo, hi)) and adjacent, so together they partition the key — every row is read exactly once (structural parity, proven on all engines). Each worker still pages by seek within its range, so peak RSS stays bounded (≤ N × chunk_size rows in flight) and no worker holds a long query. Per-range crash-recovery is tracked in the state DB (chunk_checkpoint): a run that dies resumes only the unfinished ranges.

Sweet spot. Extraction is I/O-bound, so the speedup plateaus early — ~3.1× at parallel: 4 on an indexed table, little beyond 4. It is tuned for indexed tables up to ~10 M rows. Past that, the range-boundary sampler (an index OFFSET skip to find each percentile cut) grows costly at setup — for very large tables prefer a range chunk_column (integer-PK), which slices MIN..MAX arithmetically with no sampling pass.

Canary first. Parallel keyset is new in 0.23.0. Run it on a canary table and diff row counts against a sequential (parallel: 1) pass before rolling it out across a fleet — the sequential path is unchanged and remains the conservative default.

rivet init scaffolds a row-scaled parallel (≤500 K → 1, <5 M → 2, ≥5 M → 4) on range chunk_column tables (no single-column PK). A keyset (chunk_by_key) table is scaffolded sequential — add parallel: N yourself to opt into parallel keyset. A preflight warns past ~5 M rows that peak RSS scales with N.

Limitations (current):

• Single-column keys only — composite unique keys are not yet supported.
• Decimal (numeric) keys and partition_by are rejected for ALL keyset exports, sequential included: a decimal key is refused at plan time because the keyset cursor cannot read/advance past it, and partition_by is incompatible with chunk_by_key at config validation. (Parallel keyset adds no extra key-type restriction beyond these.)

Troubleshooting

Many empty chunks – Your ID column has gaps. Use chunk_by_key on a unique key, or chunk_count: N.

High memory usage with parallel > 1 – Reduce chunk_size or add tuning.profile: safe.

Export fails midway through 1000 chunks – Enable chunk_checkpoint: true; the next run resumes from the completed chunks, whether the last run crashed or failed.

Time-Window Export Mode

When to use

Use mode: time_window to export only rows within a rolling N-day window from the current timestamp. Best for:

• Event tables where you only need the last 7/30/90 days
• Periodic refresh of a “recent activity” dataset
• When incremental is not suitable because you need overlapping windows

Required fields

• time_column – the timestamp column to filter on
• days_window – how many days back from now to include

Minimal config

source:
  type: postgres
  url: "postgresql://user:pass@host:5432/dbname"

exports:
  - name: recent_events
    query: "SELECT id, user_id, event_type, payload, created_at FROM events"
    mode: time_window
    time_column: created_at         # timestamp column to filter
    days_window: 30                 # include rows from the last 30 days
    format: parquet
    destination:
      type: local
      path: ./output

Run it

rivet check --config events.yaml
rivet run --config events.yaml --validate

What happens

1. Rivet calculates the cutoff: NOW() - 30 days
2. Appends WHERE created_at >= '2026-03-07 00:00:00' to your query
3. Exports all matching rows as a fresh file
4. No cursor is stored – each run re-evaluates the window

Unlike incremental, this mode produces overlapping data across runs (the last 30 days always overlap with yesterday’s last 30 days).

Re-runs always emit a new file (intentional)

time_window does not persist “we already exported this window” anywhere. Each rivet run re-evaluates the rolling window relative to NOW() and writes a fresh file. Two back-to-back runs inside the same minute will produce two near-identical files — same rows, different run_id / timestamp suffix. That is the contract: this mode is built for rolling recent activity (alerting, daily syncs), not for exactly-once delivery. If you need the latter, switch to mode: incremental with a cursor_column and skip_empty: true.

Downstream consumers that need deduplication should key off the id / business key inside the rows themselves, not on file name or run_id.

Time column types

By default, Rivet assumes a TIMESTAMP/DATETIME column. For Unix epoch integers, set time_column_type:

exports:
  - name: recent_events
    query: "SELECT id, user_id, event_type, created_at_epoch FROM events"
    mode: time_window
    time_column: created_at_epoch
    time_column_type: unix          # column stores Unix epoch (seconds)
    days_window: 7
    format: csv
    destination:
      type: local
      path: ./output

Common options

exports:
  - name: recent_events
    query: "SELECT id, user_id, event_type, created_at FROM events"
    mode: time_window
    time_column: created_at
    days_window: 30
    format: parquet
    compression: zstd
    skip_empty: true
    destination:
      type: local
      path: ./output

Troubleshooting

0 rows exported but the table has data – Check that days_window is large enough. Events older than the window are excluded. Also check timezones.

Duplicates across runs – This is by design. Each run exports the full window. Downstream consumers should deduplicate by primary key.

Need non-overlapping exports – Use mode: incremental with cursor_column instead.

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"
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_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.

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 — fall back to a full re-snapshot.

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), delete the checkpoint so CDC anchors on the new host first, then re-snapshot the table (mode: full).

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 by deleting the checkpoint so CDC re-anchors first, then re-snapshot the table with mode: full — in the other order, the changes between the snapshot and the new anchor land in neither. A checkpoint written before rivet recorded the identity resumes with a warning.

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.

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 same information as NDJSON (one JSON object per change) to stdout.

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 is not in that run’s schema, so its values for that run’s window are dropped — re-snapshot the table after an ADD COLUMN if those values matter. MySQL’s binlog carries names only when the server runs with binlog_row_metadata=FULL (8.0.1+ — strongly recommended; the compose test stack sets it):

# my.cnf — makes mid-stream DDL safe for rivet CDC
binlog_row_metadata = FULL

Under the default MINIMAL the binlog is nameless and positional — expect runs to FAIL with an explicit error (“an event … carries N column(s) but the resolved schema has M”) whenever a DDL lands inside a capture window. That is deliberate: mapping by position would put values into the wrong columns silently, and a loud stop is the only safe behavior. Recover by re-snapshotting the table (or resetting the checkpoint past the DDL), and set binlog_row_metadata=FULL to retire this error class. DDL between runs is always fine — each run resolves the schema fresh. A mid-window RENAME is safe in both modes (same arity ⇒ positional fallback keeps the value). Same-arity TYPE changes remain undetectable without schema history (roadmap) — run type migrations and their backfills through a re-snapshot.

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-snapshot hint; delete the checkpoint file only when you explicitly accept a fresh anchor.

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-snapshot. 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.

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 — delete the checkpoint so CDC re-anchors first, then re-snapshot (mode: full). 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), rivet fails loudly — “the resume position is older than the change-table retention … re-snapshot” — rather than resume from the new min and silently skip the gap. Delete the checkpoint so CDC re-anchors first, then re-snapshot. 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. Delete the checkpoint so the next run anchors first, then re-snapshot the table.

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.

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), restart CDC from a new checkpoint first, then re-snapshot the table with mode: full — the only safe recovery once the source log no longer covers the gap. The order matters: the new anchor must exist before the snapshot reads, so the stream overlaps the snapshot (duplicates, which the load deduplicates) instead of leaving the changes in between in neither.

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-snapshot the table 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 is total: every Rivet-mapped type — including PostgreSQL arrays (real List columns, inner NULLs preserved) and NUMERIC/DECIMAL above precision 38 (Decimal256) — is byte-identical to the batch export, enforced per engine by the live *_full_type_matrix_matches_batch tests (ArrayData equality).

CDC failure modes & recovery

What rivet does when a CDC run hits an operational failure, what you do to recover, and how to prevent it. Two guarantees frame every row:

• Loud stop, never a silent gap. When the source can no longer supply the changes since the checkpoint (a dropped/invalidated slot, purged binlog, aged-out change table), rivet fails the run with a specific error and a recovery hint — it never silently re-anchors at “now” and skips the gap. The cost is a re-snapshot; the benefit is you always know the numbers are right.
• At-least-once, so a crash or outage is a delay, not a loss. rivet reads, durably writes, then acks (peek → flush → ack). A crash between the write and the checkpoint re-reads the un-acked changes on the next run. Duplicates are the downstream MERGE’s job; loss does not happen.

rivet doctor -c rivet.yaml is the preventive layer. For mode: cdc exports it probes the engine and turns most of the rows below from an incident into a warning before the run — run it in your scheduler’s pre-flight step.

The table

The shape of every recovery

Two recovery paths cover the table:

1. Re-snapshot — when the source no longer has the changes since the checkpoint (slot invalidated, binlog purged, change table aged out, a same-arity type change). Take a fresh consistent baseline (initial: snapshot or mode: full), then resume CDC from the new anchor. The gap is not recoverable from the log — the honest fix is a new baseline.
2. Re-run — when the changes are still in the source but a write or process failed (destination outage, crash, ENOSPC, revoked grant). The at-least-once contract replays the un-acked tail; no baseline needed.

The rule of thumb: source-side loss ⇒ re-snapshot; sink-side or process failure ⇒ re-run. rivet always fails loud enough to tell you which.

CDC change ordering: __pos is not a total order — add __seq

Shipped in 0.17.0 — this page is the design rationale, not a proposal. __seq is a live CDC output column; the current reference is reference/cdc.md § Output shape. This page records why (__pos, __seq) is the total change order. The per-engine population below covers the three SQL engines; MongoDB (added later) gives every change-stream event a distinct __pos, so its __seq is always 0.

Problem (verified live on all three engines)

The CDC output columns are __op, __pos, and the after-image. __pos is the commit position of the change’s transaction:

Commit position is exactly right for resume/checkpoint (all engines resume at a commit boundary). But it is not a total order over changes: every change in one transaction shares it. Proven — 8000 UPDATEs of one PK in a single transaction produced COUNT(DISTINCT __pos) = 1 on all three engines.

Downstream, the current-state dedup view

ROW_NUMBER() OVER (PARTITION BY <pk> ORDER BY <parsed __pos> DESC) = 1

has 8000 tied rows, so ROW_NUMBER picks an arbitrary one. Live result: the view returned counter = 1 for a row whose committed value was 8000 — silently wrong current state. This bites any transaction that touches the same PK more than once (triggers, ORMs, read-modify-write loops, batch upserts). The row order inside the Parquet part is the change order, but that order is lost the moment the log is loaded into an (unordered) warehouse table.

Design: emit a per-change __seq (intra-transaction ordinal)

Add a __seq column to the CDC output: the change’s ordinal within its commit group. Keep __pos unchanged (still the commit position, still what resume uses). The pair (__pos, __seq) is then a total order that:

• matches log order (commit order across transactions, emission order within),
• is deterministic and log-derived, so a re-emitted change (at-least-once, crash-before-ack) carries the same (__pos, __seq) — the dedup tiebreak is a true tie between identical rows, so either wins and the value is right,
• survives the load (it is a column, not row order).

Dedup view becomes:

ROW_NUMBER() OVER (PARTITION BY <pk> ORDER BY <parsed __pos> DESC, __seq DESC) = 1

Per-engine population

__seq is a 0-based counter over the changes of one commit, in log order:

• SQL Server — use the native __$seqval from the change table (it already orders operations within __$start_lsn). __seq = dense_rank of __$seqval within __$start_lsn (or __$seqval rendered as a comparable fixed-width value). No invention — the engine hands us the order.
• PostgreSQL — logical decoding yields the transaction’s changes in order; assign 0,1,2,…, resetting when __pos (commit LSN) advances.
• MySQL — binlog row events arrive in order within the transaction; assign 0,1,2,…, resetting at each commit __pos.

Because the reset key is __pos (the commit position), the ordinal is reproducible from the log alone on every run — the at-least-once property above holds without any persisted counter.

Why not alternatives

• A single global run counter (0,1,2,… over the whole run) breaks across runs: run 2 resets to 0, so a newer change gets a smaller counter than an older one from run 1. (__pos, __seq) avoids this by resetting per commit, keeping the ordinal log-derived.
• Folding __seq into __pos (making __pos distinct per change) would break resume, which must stop on a commit boundary, not mid-transaction.
• Relying on Parquet row order — lost on load into a warehouse table.

Blast radius

• CDC sink schema gains one column (__seq INT64), all engines.
• One capture-agnostic path populates it — the shared TxnSeq per-commit ordinal, stamped as the stream is consumed on every engine (SQL Server’s __$seqval only orders the change-table read; MySQL/Postgres from a per-commit ordinal).
• validate.rs can additionally assert (__pos, __seq) is strictly increasing in part→row order (today it only checks __pos non-decreasing).
• The rivet-pro dedup view template orders by (__pos, __seq).
• Regression test per engine: N changes to one PK in one transaction → the dedup view returns the last change’s value, not an arbitrary one.

Loading rivet CDC into BigQuery — free ingest, cheap dedup

rivet load on a mode: cdc config does this end to end — it appends the change log for free and builds a current-state dedup view. This note explains the model it implements (verified against BigQuery docs + live behavior): why CDC ingest and dedup to current state can be free, the way the batch loader is free. The one command is at the bottom.

What rivet CDC produces

Per-change typed Parquet: the after-image columns plus __op (insert/update/delete) and __pos (monotonic log position), append-only, at-least-once (a re-run can re-emit a change).

The one hard fact

• Loading raw changes is FREE — it is an ordinary LOAD DATA (native schema, partitioned, clustered), identical to the batch path.
• Deduplication to current state is inherently cross-row (latest row per primary key + drop deletes). Any materialization of that state is a query (MERGE / CREATE TABLE AS SELECT) and is billed. There is no free lunch for collapsing a change log into current state.

So “free CDC + dedup” is really: keep the pipeline free, and defer/limit the dedup cost.

Three options

Recommended: append the log (free) + a dedup view (free)

1. Ingest (free). LOAD DATA INTO <table>__changes (…native schema… , __op STRING, __pos STRING) — the same free, native-schema, daily-batched load the batch path uses (__pos is the JSON log-coordinate string, see the view below). Partition __changes by change date, cluster by the primary key so the view below prunes efficiently.

2. Current state (free to define). A view collapses the log. Note __pos is a JSON string of the log coordinate (verified live), NOT an integer — MySQL renders {"file":"binlog.000047","pos":10840633}, PostgreSQL/SQL Server a {"lsn":…}. So the ordering must parse it; sorting the raw string is wrong ("9" > "10" lexically). The parse is therefore per-engine:

   -- MySQL (binlog file + position):
   CREATE OR REPLACE VIEW `<table>` AS
   SELECT * EXCEPT (__op, __pos, __seq, __rn),
          (__op = 'delete') AS __is_deleted
   FROM (
     SELECT *, ROW_NUMBER() OVER (
       PARTITION BY <pk>
       ORDER BY JSON_VALUE(__pos,'$.file') DESC,
                CAST(JSON_VALUE(__pos,'$.pos') AS INT64) DESC,
                __seq DESC
     ) AS __rn
     FROM `<table>__changes`
   )
   WHERE __rn = 1;
   -- PostgreSQL / SQL Server: ORDER BY JSON_VALUE(__pos,'$.lsn') …
   -- Snowflake: PARSE_JSON(__pos):file … and SELECT * EXCLUDE (…)
   

   One expression does the dedup work: at-least-once dedup (a re-emitted change has the same (__pos,__seq) and loses the tiebreak) and latest-per-PK collapse. Soft delete: the latest change is kept unconditionally, and its __op is projected into a boolean __is_deleted column — a deleted row survives as a tombstone (last-known values + __is_deleted = true) instead of silently vanishing; live state is WHERE NOT __is_deleted. Verified live: three changes (insert/update/delete) loaded twice (10 rows) collapse to 3 distinct-PK rows — the deleted PK present with __is_deleted = true, the other two false (2 live rows).

Ingest + view are both free. Reading <table> scans __changes (billed), but clustering on <pk> keeps it cheap; if current state is read hot, add an optional daily compaction (CREATE OR REPLACE TABLE <table>__snapshot AS SELECT * FROM <table>) — one billed scan per day, not per read. This is the classic log + periodic compaction.

The base-and-buffer layout (backfill: streams) and rivet compact

A stream whose baseline comes from cdc.backfill: does NOT use the view above. Its baseline legs overwrite a physical base table <table> — the source columns plus one service column, __is_deleted BOOL (written as false inside the baseline Parquet, so no NULL ever appears) — and the stream’s runs append into <table>__changes, a per-cycle buffer without a partition. The cycle is

rivet run     -c cfg.yaml   # anchor once, baseline once, then only the changes
rivet load    -c cfg.yaml   # baseline → <table> (batched, staging + CLONE); changes → <table>__changes
rivet compact -c cfg.yaml   # MERGE <table>__changes into <table>; DROP the buffer

compact is one scripted job per table: the latest change per key (the same __pos order the view uses) is upserted; a delete flags the base row (__is_deleted = TRUE, last values kept — the warehouse deletes nothing) and a later insert un-flags it. For a day-partitioned base (init’s default) the script collects the buffer’s distinct days into a variable and every MERGE filters both sides with DATE(col) IN UNNEST(days) — measured: 172 bytes read against 48 KB for a MIN..MAX range on the same buffer, i.e. exactly the touched partitions; more than 4,000 days merge in chunks of 4,000 inside the same script. Then the script drops the buffer and the next load creates it again from its run’s spec. Other partition keys (hour, month, year, integer ranges) keep a constant MIN..MAX range per window in separate jobs — the truncation forms did not prune when measured. An empty buffer is just dropped.

What a cycle bills: BigQuery charges every statement that reads a table at least 10 MB per table, so a compaction with changes bills a 30 MB floor (the probe, and the MERGE over two tables); one without changes bills nothing. The load side is free (CREATE, LOAD DATA). The script’s child statements appear in INFORMATION_SCHEMA.JOBS under parent_job_id with the same labels.

Consumers read <table> directly, WHERE NOT __is_deleted for live rows. The buffer holds no history — __is_deleted in the base is the record that a row was deleted. BigQuery only in this release; Snowflake keeps the view layout.

Every billed step carries its own label

The whole point of the loader’s job labels (managed_by:rivet / rivet_op:<op> / rivet_table:<table> / rivet_run:<load run id>) is that you can price each table’s update, per operation. There are two operations:

• rivet_op:load — everything rivet load runs for a table: the free LOAD DATA jobs (one per batch of at most 4,000 partitions), the COUNT(*) gate, the table DDL, the staging CLONE of a batched whole-table load, the view;
• rivet_op:merge — everything rivet compact runs for a table (the billed MERGE and its partition-range probe).

rivet_table is the base table’s short name for both the table and its __changes, so GROUP BY op, tbl answers “what does keeping this table current cost” in one row per table per operation:

SELECT
  (SELECT value FROM UNNEST(labels) WHERE key='rivet_op')    AS op,      -- load | merge
  (SELECT value FROM UNNEST(labels) WHERE key='rivet_table') AS tbl,
  COUNT(*) AS jobs, SUM(total_bytes_billed) AS bytes_billed
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE EXISTS (SELECT 1 FROM UNNEST(labels) WHERE key='managed_by' AND value='rivet')
GROUP BY op, tbl ORDER BY bytes_billed DESC;

Every rivet-driven job passes through one labelled seam (run_sql(sql, op, table)), so nothing rivet runs is unlabelled — only jobs you run yourself need labels of your own.

The one command: rivet load

rivet load -c cfg.yaml — where the export is mode: cdc and the config carries a top-level load: block with target: bigquery — does both steps automatically. The view’s key is the source primary key rivet run recorded (pk: auto, the default); pk: [..] overrides it and is required only when none was recorded (a query: export, a table without a primary key):

1. free LOAD DATA of the CDC Parquet into <table>__changes (the same native-schema batched loader, with __op/__pos/__seq in the schema);
2. CREATE OR REPLACE VIEW <table> — the exact dedup view above.

exports:
  - name: orders
    table: orders
    mode: cdc
    cdc: { until_current: true, checkpoint: /var/lib/rivet/orders.ckpt }
    destination: { type: gcs, bucket: my-bucket, prefix: cdc/orders/ }
load:
  target: bigquery      # or: snowflake (+ connection/warehouse/database/schema/storage_integration)
  project: my-proj
  dataset: analytics
  # pk: [id]            # the view's PARTITION BY; default: the source primary key
  cleanup_source: true

Both steps are free. The count gate (summed manifest rows == warehouse COUNT(*)) and source cleanup work exactly as in the batch path. There is no --cdc flag — the mode comes from the export’s mode: cdc; one config drives both rivet run (extract) and rivet load.

Live-verified end to end: this flow builds the dedup view shown above, with two refinements over the sketch — on MySQL the binlog file is parsed numerically (CAST(REGEXP_EXTRACT(JSON_VALUE(__pos,'$.file'), r'[0-9]+$') AS INT64) then CAST(JSON_VALUE(__pos,'$.pos') AS INT64)), and the delete flag is COALESCE(__op = 'delete', FALSE) AS __is_deleted so snapshot-backfill rows (NULL __op) stay live — and a deleted PK survives as __is_deleted = true rather than vanishing. See the matrix cells cdc_backfill_snapshot_{mysql,pg,mongo} and the Snowflake parity mongo_cdc_delete_flag_snowflake.

A whole schema: one stream, one warehouse table per source table

rivet init --mode cdc over a whole schema emits ONE multiplex export — every table through one change stream (one PostgreSQL slot / one MySQL binlog connection), rather than one export and one slot per table:

exports:
  - name: cdc
    tables: [orders, customers, line_items]
    mode: cdc
    cdc: { initial: snapshot, until_current: true, checkpoint: /var/lib/rivet/cdc.ckpt }
    destination: { type: gcs, bucket: my-bucket, prefix: cdc/ }
load:
  target: bigquery
  project: my-proj
  dataset: analytics

Loads are batched by partition span. BigQuery writes at most 4,000 partitions per job. Before any job, rivet load reads the partition column’s range from every Parquet footer and packs the files, in order of their lowest value, into jobs whose combined span fits — so a keyset export over an autoincrement key (whose files are date-local because id grows with time) loads eleven years of daily partitions in three or four jobs, never coarsened to month. A whole-table (OVERWRITE) load that needs several jobs fills a <table>__staging table and swaps it in with one zero-copy CLONE. The one shape nothing splits is a single file wider than 4,000 partitions (dates uncorrelated with the read key): that is refused before any job, naming the file and the granularity that fits.

The capture fans each table out under <prefix>/<table>/ (its own manifest.json + _SUCCESS, with initial: snapshot nested a level below as <prefix>/<table>/snapshot/), and rivet load follows that layout: one <table>__changes + one dedup view per SOURCE table, each loaded from its own sub-prefix only. Each table is keyed on its own recorded primary key; the rest of the load: block is shared by every table of the stream unless the export’s load: carries tables: { orders: { partition: { column: created_at, granularity: day } }, customers: { partition: none } } — one block per captured table, layered over the export’s and the top-level load:. With cdc: { backfill: auto, … } in place of initial: snapshot, each table’s baseline is read by the batch export that names it (keyset, chunked, with its columns:) into the same <prefix>/<table>/snapshot/, so the load is unchanged (rivet init --mode cdc scaffolds that shape on MySQL and PostgreSQL); rivet check --target bigquery prints one resolver document per table (Export: cdc/orders), so you see each table’s native schema before loading it. Live-verified against BigQuery over a 3-table PostgreSQL stream (#252).

Bottom line: yes — rivet can ingest CDC into BigQuery and expose a deduplicated current state entirely for free (append + view). The only unavoidable cost is materializing current state, which we defer to read time (a view) or amortize (daily compaction) — never on the ingest path.

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 or a non-public schema get a per-table capture-only stream instead (add initial: snapshot or a recipe + backfill: yourself). Add the load: block and run.
• 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.

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

• 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 baseline OVERWRITEs the base, 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 (MySQL / SQL Server / MongoDB) or let the slot be recreated (PostgreSQL), AND clear the export’s cdc_snapshot row and the table’s snapshot/_SUCCESS, AND truncate <table>__changes before the next load (a re-read baseline has no __pos, so the log cannot be deduplicated across it; the load refuses without the truncate). Deleting the checkpoint alone is refused: prior-run evidence exists and the run would re-anchor over a gap.
• 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.

Running the automated scenario

docker compose --profile cdc up -d
export BIGQUERY_TEST_PROJECT=<gcp-project> RIVET_TEST_GCS_BUCKET=<bucket>   # RIVET_TEST_BQ_DATASET optional
cargo test --test live_suite full_cdc_cycle -- --ignored --test-threads=1

Without the warehouse env the four tests skip, by name.

Database version support

Rivet supports five source engines — PostgreSQL, MySQL, SQL Server, MongoDB and, in preview, Oracle Database — every supported version of which is listed in the table below. PostgreSQL and MySQL run the full end-to-end suite on each release — doctor, check, every export mode (full / incremental / chunked / time_window), every output format (CSV / Parquet) with every compression codec, reconcile, recovery scenarios, state management, date-chunking, and rivet init. The release gate runs that suite against a representative set — PostgreSQL 14, 16 and 18, MySQL 8.0 and 8.4 — chosen as oldest-supported, primary target and newest; the remaining supported versions share the same code paths and are spot-checked rather than gated on every release. No version-specific code paths exist to skip: the same Rust driver builds and the same YAML configs drive every target. SQL Server and MongoDB carry their own scope and CI coverage, detailed below.

MongoDB (the OSS JSON-blob source — batch + CDC) rides its own dedicated CI matrix across 4.4 → 8.0, live-tested through the canonical test rig, rather than the SQL e2e suite above (a document store has no chunked / incremental / time_window modes). See mongodb.md.

Supported versions

“Primary target” means the version that runs the e2e suite by default in the local docker-compose.yaml top-level postgres / mysql / mssql / mongo services. “Legacy” versions are opt-in under the legacy compose profile (see below).

SQL Server (MSSQL) — current scope

Status: GA. The engine is live-validated and feature-complete for the shapes below. The two gaps that formerly held it in Beta are now closed: the transitive rustls-webpki advisory (resolved — see below) and datetimeoffset roundtrip verification, so it is promoted to the same GA bar as PG/MySQL.

Transitive advisory — RESOLVED. tiberius 0.12 formerly linked rustls 0.21 → rustls-webpki 0.101, carrying CA name-constraint advisories (RUSTSEC-2026-0098/0099) and a CRL-parse panic (RUSTSEC-2026-0104). Rather than wait for an upstream tiberius bump, the driver now uses its vendored-openssl TLS backend (OpenSSL, statically linked on every platform), so those advisories are out of the dependency tree entirely — not suppressed. On Linux this also unifies the TLS stack with the PG/MySQL drivers (all three link OpenSSL). On macOS the stacks differ: PG/MySQL use native-tls, which resolves to the system SecureTransport framework, while the MSSQL driver statically links OpenSSL (SecureTransport cannot complete SQL Server’s TDS-wrapped TLS handshake). Strict validation (tls.mode: verify-ca | verify-full) is enforced by OpenSSL and rejects a certificate that does not chain to the trusted CA — verified live on macOS and Linux against a private-CA-configured SQL Server (correct CA connects; wrong CA is refused with certificate verify failed). cargo audit is clean for the MSSQL engine. (native-tls is deliberately not used: on macOS it resolves to SecureTransport, which cannot complete SQL Server’s TDS-wrapped TLS handshake.)

Type fidelity. datetimeoffset, the one type formerly “mapped but not roundtrip-verified”, is now validated through the DuckDB/ClickHouse Parquet oracles (UTC instant + tz-awareness, positive/negative offsets + NULL).

SQL Server is a source engine (source.type: mssql, scheme sqlserver://, default port 1433), driven by the async tiberius client. Supported today:

• Modes: full / snapshot, incremental, chunked (range + dense), keyset (seek) via explicit chunk_by_key — the ideal shape for a non-integer PK (UUID / string) — and CDC (mode: cdc / rivet cdc, reading SQL Server change tables via the Agent capture job; see cdc.md). The page builder emits the T-SQL OFFSET 0 ROWS FETCH NEXT n ROWS ONLY clause (T-SQL has no LIMIT).
• Types (live-validated through the DuckDB + ClickHouse Parquet oracles): int/bigint/smallint/tinyint, bit, decimal/numeric, real/float, money, date, time, datetime2, nvarchar/varchar/ char, varbinary, uniqueidentifier (→ native Parquet LogicalType::Uuid), and datetimeoffset (→ Timestamp(µs, UTC): the offset is applied and the UTC instant re-read correctly by the DuckDB/ClickHouse oracles — positive and negative offsets and NULL all covered in the type matrix).
• TLS: required on the login handshake (SQL Server always encrypts it). Set tls.ca_file: for a private CA, or tls.accept_invalid_certs: true for a self-signed dev cert.

Why these versions

• PostgreSQL 12 — oldest mainstream release still in community support (final minor release; community support ended Nov 2024, but many managed platforms — RDS, Cloud SQL, Azure Database — continue to ship it).
• PostgreSQL 13–15 — actively supported by upstream.
• PostgreSQL 16 — current stable at the time of writing; Rivet’s default.
• MySQL 5.7 — EOL upstream in October 2023 but widely deployed in legacy systems; keeping it in the matrix prevents silent breakage for those users.
• MySQL 8.0 — current stable; Rivet’s default.

Older releases (PostgreSQL ≤11, MySQL ≤5.6) aren’t tested. They may well work — Rivet’s SQL surface is deliberately narrow — but regressions on them aren’t caught by CI.

Running the compatibility matrix locally

Bring up every server the matrix covers:

# Primary versions (PG 16, MySQL 8.0) — plus MinIO and fake-gcs for destinations
docker compose up -d postgres mysql minio fake-gcs

# Legacy versions — opt in via the `legacy` profile
docker compose --profile legacy up -d \
    postgres-12 postgres-13 postgres-14 postgres-15 mysql-57

cargo build --release --bin rivet --bin seed --features dev-seed

Ports assigned (none conflict with the primary services):

Then pick one of:

# Full e2e suite on every version — seeds each DB, then runs
# python3 -m dev.pytools.e2e against it with URLs retargeted via env.
python3 -m dev.pytools.legacy_stand full-matrix

# Just one target
TARGETS="pg-12"     python3 -m dev.pytools.legacy_stand full-matrix
TARGETS="mysql-57"  python3 -m dev.pytools.legacy_stand full-matrix

# Lighter compat smoke (seed + mode sampler + init) — same config, fewer
# assertions; useful when iterating on compat-sensitive code paths.
python3 -m dev.pytools.legacy_stand legacy

How the matrix targets an arbitrary server

python3 -m dev.pytools.e2e no longer hardcodes localhost:5432 / localhost:3306. It reads RIVET_PG_URL and RIVET_MYSQL_URL from the environment (falling back to the primary ports if unset), and every e2e YAML uses url_env: RIVET_PG_URL (or RIVET_MYSQL_URL). So the same script + configs drive any target:

RIVET_PG_URL=postgresql://rivet:rivet@localhost:5412/rivet \
    python3 -m dev.pytools.e2e

python3 -m dev.pytools.legacy_stand full-matrix does nothing more exotic than: seed → export those env vars → invoke python3 -m dev.pytools.e2e.

Engine-specific notes

MySQL 5.7 — window functions

The view orders_sparse_for_export used by the chunked-sparse demo queries uses ROW_NUMBER() OVER (...), which is only available from MySQL 8.0. The dev/mysql/init.sql seeding script creates this view; when 5.7 runs the same script it fails at container bootstrap with

ERROR 1064 (42000) at line 104: You have an error in your SQL syntax …
near '(ORDER BY id) AS chunk_rownum FROM orders_sparse' at line 5

Rivet ships a dedicated dev/mysql/init_57.sql (identical schema minus the view) and docker-compose.yaml mounts it for the mysql-57 service. The seed binary detects the server version via SELECT VERSION() and short-circuits the CREATE OR REPLACE VIEW when the server reports 5.x:

  note: MySQL 5.7.44 has no window functions — skipping `orders_sparse_for_export` view

All other schema (tables, indexes, JSON columns) works unchanged on 5.7.8+. No production YAML depends on orders_sparse_for_export; it’s only used by the chunked-sparse demo.

MySQL 5.7 — arm64 hosts (Apple Silicon)

There is no official arm64 image for mysql:5.7 on Docker Hub. The docker-compose.yaml entry for mysql-57 pins platform: linux/amd64 so Docker Desktop falls back to its amd64 emulator on M-series Macs. Boot is ~2 seconds slower than a native image but otherwise transparent.

MySQL 5.7 — auth plugin and local clients

MySQL 5.7 defaults to mysql_native_password. MySQL 8.0 defaults to caching_sha2_password, and the Homebrew mysql-client@9 package on macOS has dropped the plugin library for native_password entirely:

ERROR 2059 (HY000): Authentication plugin 'mysql_native_password'
    cannot be loaded: dlopen(...) (no such file)

The rust mysql crate (version 28, which Rivet depends on) has native_password built in, so Rivet itself connects fine. Only local CLI tools (mysql, mysqladmin) may refuse to. When scripts need a reachability probe, they use a bash /dev/tcp test rather than mysqladmin:

if (exec 3<>/dev/tcp/127.0.0.1/"$port") 2>/dev/null; then
    echo "reachable"
fi

PostgreSQL — no TLS by default

Rivet’s dev/e2e/*.yaml configs connect without TLS (matches the primary e2e harness). For production, enable transport security explicitly:

source:
  type: postgres
  url_env: DATABASE_URL
  tls:
    mode: verify-full           # disable | require | verify-ca | verify-full
    ca_file: /etc/ssl/certs/rds-ca-2019-root.pem

See config.md for the full TLS block. This works against every supported PostgreSQL version (12+).

PostgreSQL — rivet’s sessions run in UTC

Every PostgreSQL connection rivet opens sets TimeZone = 'UTC', DateStyle = 'ISO, MDY', IntervalStyle = 'postgres' and bytea_output = 'hex', whatever the server, database or role defaults are. Stored values are unaffected: a timestamptz is an instant, and a timestamp has no zone. What changes is calendar arithmetic inside your own query:. For example, current_date and ts::date on a timestamptz column count days in UTC. For the server’s local day, say so explicitly: (ts AT TIME ZONE 'Europe/Kyiv')::date. Behind a transaction-mode pooler (pgBouncer, Odyssey), a session SET would leak to other clients. There rivet sets the zone only inside the export’s own transaction.

What “passes” means per target

Each target in dev/pytools/legacy_stand.py runs 83 assertions against its assigned server. Status reported by the suite:

  pg-12: PASS (83 passed, 0 skipped)
  pg-13: PASS (83 passed, 0 skipped)
  pg-14: PASS (83 passed, 0 skipped)
  pg-15: PASS (83 passed, 0 skipped)
  pg-16: PASS (83 passed, 0 skipped)
  mysql-57: PASS (83 passed, 0 skipped)
  mysql-80: PASS (83 passed, 0 skipped)

581 assertions total, and a broken compat path surfaces which target(s) failed and which specific assertion inside. Per-target logs are written to /tmp/rivet_full_matrix/<target>.log.

Policy

• Adding a new supported version: add a service to docker-compose.yaml under the legacy profile, add the port mapping to dev/pytools/legacy_stand.py, run the matrix, land the PR with the new version listed in this page’s table.
• Dropping a version: remove the service from the compose file, remove the target from dev/pytools/legacy_stand.py, remove its row from this page, and note the change in CHANGELOG.md. Dropping a version is a minor-version bump (no SemVer guarantees apply to unsupported servers).

MongoDB Reference

Rivet reads a MongoDB collection as a JSON-blob table: every document becomes exactly two columns —

Rivet does not flatten fields into typed columns. Documents in one collection rarely share a schema, so rivet keeps each document intact as one JSON value and lets the warehouse type it on the way in (PARSE_JSON → VARIANT on Snowflake, JSON on BigQuery). This is lossless and schema-drift-proof: a new field in a document never breaks a load.

Two run modes:

• Batch — a full snapshot of a collection to Parquet/CSV. Works against a standalone mongod or a replica set.
• CDC — change capture via a change stream. Requires a replica set (a single-node replica set is fine); a standalone mongod cannot open a change stream.

Prerequisites

• A connection URL: mongodb://[user:pass@]host:port/database.
• For a port-mapped single-node replica set (common in local/dev docker), append ?directConnection=true — otherwise the driver tries to re-resolve the replica-set members by their in-container hostnames and fails with ReplicaSetNoPrimary.
• For CDC, the login needs a role that can run changeStream (e.g. read on the database). For delete/update pre-images (MongoDB 6.0+), the collection needs changeStreamPreAndPostImages enabled.

Scenario A — Batch export (full snapshot)

Goal: copy a whole collection to Parquet, verify nothing was lost, and learn how it will land in the warehouse.

1. Scaffold a config

rivet init --source "mongodb://127.0.0.1:27017/shop" -o mongo.yaml

Or write it by hand — the minimal batch config:

# batch.yaml
source:
  type: mongo
  url: "mongodb://127.0.0.1:27017/shop"
  mongo:
    page_size: 5000          # keyset (seek) paging on _id — bounded query time
exports:
  - name: products
    table: products          # the collection name
    mode: full
    format: parquet
    parallel: 4              # _id-range fan-out (optional)
    destination:
      type: local
      path: "./out/products"

2. Preflight — rivet check

$ rivet check -c batch.yaml

Export: products
  Strategy:     full-parallel(4)
  Mode:         full
  Row estimate: ~20K
  Verdict:      ACCEPTABLE

check is advisory — it never blocks the run. (A couple of lines in the report — max_connections, “only chunked mode benefits from parallelism” — are worded for SQL sources and don’t apply to Mongo; ignore them.)

3. Warehouse portability — rivet check --target snowflake

$ rivet check -c batch.yaml --target snowflake

  document  json → VARIANT   warn ~
     autoload: TEXT
     note: JSON autoloads as TEXT; recover native VARIANT with PARSE_JSON after load
     recover: PARSE_JSON("document")

This tells you the exact recovery: load document as TEXT/VARCHAR, then PARSE_JSON it into a VARIANT. Swap --target bigquery for the BigQuery form.

4. Run + validate — rivet run --validate

$ rivet run -c batch.yaml --validate

✓ products      keyset        20,000 rows    6 files   347.5 KB   0.8s
── products ──────────────────────────
  run_id:     products_20260708T120725.974
  rows:       20,000
  files:      6
  validated:  pass

--validate re-reads the output and checks the row counts. Add --reconcile to compare the destination against a fresh countDocuments on the source.

4b. Or: plan → apply (freeze, review, execute)

rivet run decides the strategy and executes in one shot. To separate the decision from the execution — review it, check it into a PR, run it later or on another host — split it into plan + apply:

$ rivet plan -c batch.yaml --format json -o plan.json
Plan written to: plan.json

plan.json is the frozen, self-describing strategy — reviewable and tamper-evident:

{
  "export_name": "products",
  "expires_at": "2026-07-09T12:13:22Z",       // stale plans (>24h) are rejected
  "integrity": "xxh3:0f4e1be244bc0845",       // apply verifies this checksum
  "resolved_plan": {
    "strategy": { "Keyset": { "key_column": "_id", "chunk_size": 5000, "parallel": 4 } },
    "format": "parquet",
    "compression": "zstd"
  }
}

$ rivet apply plan.json

── products ──────────────────────────
  run_id:  products_20260708T121322.293
  rows:    20,000
  files:   6
  verify:  not run — add `--reconcile` or `rivet validate`

apply executes exactly the frozen plan (it re-checks the integrity checksum and the expiry first). Same result as run; the difference is that the strategy was pinned and reviewable in between. (One cosmetic note: the plan’s base_query renders as SELECT * FROM products — a logical placeholder; Mongo does not run SQL.)

5. What landed

$ duckdb -c "SELECT COUNT(*), COUNT(DISTINCT _id) FROM read_parquet('out/products/*.parquet')"
20000, 20000                              # no loss, no duplicates across pages

$ duckdb -c "SELECT document FROM read_parquet('out/products/*.parquet') LIMIT 1"
{"_id":{"$oid":"6a4e…"},"sku":"P000001","name":"Item 1",
 "price":{"$numberDecimal":"1.99"},"qty":1,"tags":[],"meta":{…}}

Every document is verbatim relaxed extended JSON — note $oid and $numberDecimal type tags, which a warehouse PARSE_JSON reconstructs.

Scenario B — CDC (change capture)

Goal: capture inserts/updates/deletes from a collection, resumably, into Parquet. Needs a replica set.

1. Config

# cdc.yaml
source:
  type: mongo
  # directConnection=true is REQUIRED for a port-mapped single-node replica set
  url: "mongodb://127.0.0.1:27018/shop?directConnection=true"
exports:
  - name: orders_cdc
    table: orders
    mode: cdc
    format: parquet
    cdc:
      checkpoint: "./orders.ckpt"   # resume anchor (persisted resume token)
      initial: snapshot             # copy pre-existing docs first, then stream
      until_current: true           # bounded: drain the backlog, then exit
    destination:
      type: local
      path: "./cdc_out/orders"

2. Preflight — rivet doctor

$ rivet doctor -c cdc.yaml

[OK]  CDC replica set — replica set (server 7.0.37)
[OK]  CDC capture tier — full-image-capable (6.0+) — delete/update pre-images
      ride when changeStreamPreAndPostImages is enabled on the collection

doctor proves the source is a replica set and reports the capability tier (see Capability tiers).

3. First run — snapshot + drain

$ rivet run -c cdc.yaml

✓ orders_cdc__snapshot_orders  full   3 rows   1 files       # the snapshot leg
── orders_cdc ──────────────────────────
  rows:   0                                                   # no changes yet

The snapshot leg copies the 3 pre-existing documents (they predate the change stream, so the stream alone would miss them). The CDC leg then drains to “now” — 0 changes — and, because until_current: true, exits. The checkpoint now holds the resume token.

4. Some changes happen

db.orders.insertOne({_id:4, total:75, status:"new"})
db.orders.updateOne({_id:1}, {$set:{status:"shipped"}})
db.orders.deleteOne({_id:3})

5. Resume — captures only what changed

$ rivet run -c cdc.yaml

── orders_cdc ──────────────────────────
  rows:   3                                                   # only the 3 new changes

$ duckdb -c "SELECT __op, _id, document FROM read_parquet('cdc_out/orders/*.parquet')
             WHERE __op IS NOT NULL ORDER BY __pos"
insert  4  {"_id":4,"total":75,"status":"new"}
update  1  {"_id":1,"total":100,"status":"shipped"}           # post-image
delete  3  (null)                                             # _id only, no pre-image

Each change row carries three metadata columns:

Scheduling

Run the same command on an interval (cron, systemd timer, Airflow). Each run resumes from the checkpoint and drains to current. Because part files are named from the millisecond run_id, consecutive runs into the same destination prefix never overwrite each other.

Scenario C — many collections: source impact & parallel tuning

A config can export many collections at once (one - name: block each). rivet plan then writes one plan file per collection (plan.<name>.json), and — a difference from SQL sources — every collection gets the same strategy shape: keyset on _id. Mongo’s _id is always present and always indexed, so there is no per-table strategy diversity to discover (no “table without a primary key”, no chunk-vs-cursor choice). Files scale with size: files = ceil(rows / page_size).

What a full export does to the source

Measured over a 13-collection export (~800K documents, parallel: 1):

Why this is gentle on a production Mongo:

• Index-bound. The seek find({_id:{$gt: last}}).sort({_id:1}).limit(N) uses the _id_ index — examined == returned, no over-scan, on any collection.
• No long-lived cursor. Keyset issues a fresh bounded query per page, not one cursor held open for the whole scan. Nothing is pinned in server memory for minutes, and there is no cursor-timeout risk (this is why no_cursor_timeout is irrelevant to keyset). A naive find() export would hold one cursor for the entire scan; skip/limit paging would be O(n²) (re-scanning each page’s prefix). Keyset is O(n) with a page-lived cursor.

parallel: N — the trade

parallel: N splits a collection into N disjoint _id ranges (quantile boundaries found with $sample, not a full-scan $bucketAuto) and scans them concurrently. Same 8 heavy/medium collections, varying N:

Reading the curve:

• The scan cost is flat. From parallel: 4 up, examined ÷ returned sits at 1.042 and does not move — the only overhead is the one-time $sample boundary probe (~+4%, fixed per collection, independent of N). More workers do not scan the source harder; the range scans stay index-bound and disjoint.
• Connections grow linearly (~3 per unit of N): 8 → 20 → 26 → 32.
• Speed-up has a knee at ~6. 1→4 is 2.9×, 4→6 adds 1.5×, but 6→8 adds only 1.17× (+23% wall improvement for +23% connections — parity). Below the knee, connections buy speed cheaply; above it, they don’t.

So the choice is purely Mongo’s connection budget vs. desired wall-time — the scan footprint barely changes:

The bigger the collection, the more parallel pays off — the fixed ~+4% $sample cost amortises better over 300K rows than over 30K.

Connection pool

There is no external pooler (no pgBouncer/ProxySQL analog) — the MongoDB driver pools connections itself, one pool per Client, and rivet passes the pool knobs straight through from the connection URL (it sets none of its own):

The one thing to know: parallel: N opens N independent clients — N pools. So the connection ceiling is N × maxPoolSize. In practice each worker runs a sequential keyset scan and holds only ~1–2 connections (the measured peak was 20 at parallel: 4, not 4 × 10 = 40 — workers don’t saturate their pools), but N × maxPoolSize is the ceiling to size against the server’s budget:

source:
  type: mongo
  # cap each worker's pool; with parallel: 4 the ceiling is 4 × 5 = 20
  url: "mongodb://host/db?maxPoolSize=5"

directConnection=true (needed for a port-mapped replica set) does not disable the pool — it still holds up to maxPoolSize connections to the single server, just without topology discovery.

Config reference — source.mongo.*

Everything else is the shared surface: parallel (an _id-range fan-out for Mongo), mode: cdc, cdc.{checkpoint, initial, until_current, max_events}, and --target / --format / --validate / --reconcile on the CLI.

Type fidelity

Rivet stores document verbatim — there is no corruption at rest. Fidelity downstream depends on the JSON mode and the reader:

• relaxed (default) renders numbers as bare JSON ("qty": 1), with type tags only where JSON can’t express the type ($oid, $numberDecimal, $date). Compact and directly queryable.
• canonical type-tags every value ({"$numberInt":"1"}, {"$numberLong":"…"}). Verbose but unambiguous.