cdc

# CDC (Change Data Capture) - Internal Feature Map ## Overview CDC tracks INSERT/UPDATE/DELETE changes on database tables by writing change records into a dedicated CDC table (`turso_cdc` by default). It is per-connection, enabled via PRAGMA, and operates at the bytecode generation (translate) layer. The sync engine consumes CDC records to push local changes to the remote. ## Architecture Diagram ``` User SQL (INSERT/UPDATE/DELETE/DDL) | v ┌─────────────────────────────────────────────────┐ │ Translate layer (core/translate/) │ │ ┌───────────────────────────────────────────┐ │ │ │ prepare_cdc_if_necessary() │ │ │ │ - checks CaptureDataChangesInfo │ │ │ │ - opens CDC table cursor (OpenWrite) │ │ │ │ - skips if target == CDC table itself │ │ │ └───────────────────────────────────────────┘ │ │ ┌───────────────────────────────────────────┐ │ │ │ emit_cdc_insns() │ │ │ │ - writes (change_id, change_time, │ │ │ │ change_type, table_name, id, │ │ │ │ before, after, updates) into CDC tbl │ │ │ └───────────────────────────────────────────┘ │ │ + emit_cdc_full_record() / emit_cdc_patch_record() │ └─────────────────────────────────────────────────┘ | v CDC table (turso_cdc or custom name) | v ┌─────────────────────────────────────────────────┐ │ Sync engine (sync/engine/) │ │ DatabaseTape reads CDC table → DatabaseChange │ │ → apply/revert → push to remote │ └─────────────────────────────────────────────────┘ ``` ## Core Data Types ### `CaptureDataChangesMode` + `CaptureDataChangesInfo` — `core/lib.rs` CDC behavior is controlled by two types: ```rust #[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)] #[repr(u8)] enum CdcVersion { V1 = 1, V2 = 2, } const CDC_VERSION_CURRENT: CdcVersion = CdcVersion::V2; enum CaptureDataChangesMode { Id, // capture only rowid Before, // capture before-image After, // capture after-image Full, // before + after + updates } struct CaptureDataChangesInfo { mode: CaptureDataChangesMode, table: String, // CDC table name version: Option<CdcVersion>, // schema version (V1 or V2) } ``` The connection stores `Option<CaptureDataChangesInfo>` — `None` means CDC is off. Key methods on `CdcVersion`: - `has_commit_record()` — `self >= V2`, gates COMMIT record emission - `Display`/`FromStr` — round-trips `"v1"` ↔ `V1`, `"v2"` ↔ `V2` Key methods on `CaptureDataChangesInfo`: - `parse(value: &str, version: Option<CdcVersion>)` — parses PRAGMA argument `"<mode>[,<table_name>]"`, returns `None` for "off" - `cdc_version()` — returns `CdcVersion` (panics if version is None). Single accessor replacing old `is_v1()`/`is_v2()`/`version()` methods. - `has_before()` / `has_after()` / `has_updates()` — mode capability checks - `mode_name()` — returns mode as string Convenience trait `CaptureDataChangesExt` on `Option<CaptureDataChangesInfo>` provides: - `has_before()` / `has_after()` / `has_updates()` — delegates to inner, returns false for None - `table()` — returns `Option<&str>`, None when CDC is off ### CDC Table Schema v1 Default table name: `turso_cdc` (constant `TURSO_CDC_DEFAULT_TABLE_NAME`) ```sql CREATE TABLE turso_cdc ( change_id INTEGER PRIMARY KEY AUTOINCREMENT, change_time INTEGER, -- unixepoch() change_type INTEGER, -- 1=INSERT, 0=UPDATE, -1=DELETE table_name TEXT, id <untyped>, -- rowid of changed row before BLOB, -- binary record (before-image) after BLOB, -- binary record (after-image) updates BLOB -- binary record of per-column changes ); ``` ### CDC Table Schema v2 (current) ```sql CREATE TABLE turso_cdc ( change_id INTEGER PRIMARY KEY AUTOINCREMENT, change_time INTEGER, -- unixepoch() change_type INTEGER, -- 1=INSERT, 0=UPDATE, -1=DELETE, 2=COMMIT table_name TEXT, id <untyped>, -- rowid of changed row before BLOB, -- binary record (before-image) after BLOB, -- binary record (after-image) updates BLOB, -- binary record of per-column changes change_txn_id INTEGER -- transaction ID (groups rows into transactions) ); ``` v2 adds: - `change_txn_id` column — groups CDC rows by transaction. Assigned via `conn_txn_id(candidate)` opcode which get-or-sets a per-connection transaction ID. - `change_type=2` (COMMIT) records — mark transaction boundaries. Emitted once per statement in autocommit mode, or on explicit `COMMIT`. The CDC table is created at runtime by the `InitCdcVersion` opcode via `CREATE TABLE IF NOT EXISTS`. ### CDC Version Table When CDC is first enabled, a version tracking table is created: ```sql CREATE TABLE turso_cdc_version ( table_name TEXT PRIMARY KEY, version TEXT NOT NULL ); ``` Current version: `CDC_VERSION_CURRENT = CdcVersion::V2` (defined in `core/lib.rs`, re-exported from `core/translate/pragma.rs`) ### Version Detection in InitCdcVersion The `InitCdcVersion` opcode detects v1 vs v2 by checking whether the CDC table already exists before creating it: - If CDC table already exists but has no version row → v1 (pre-existing table from before version tracking) - If CDC table doesn't exist → create with current version (v2) - If version row already exists → use that version as-is ### `DatabaseChange` — `sync/engine/src/types.rs:229-249` Sync engine's Rust representation of a CDC row. Has `into_apply()` and `into_revert()` methods for forward/backward replay. ### `OperationMode` — `core/translate/emitter.rs` Used by `emit_cdc_insns()` to determine `change_type` value: - `INSERT` → 1 - `UPDATE` / `SELECT` → 0 - `DELETE` → -1 - `COMMIT` → 2 (v2 only, emitted by `emit_cdc_commit_insns`) ## Entry Points ### 1. PRAGMA — Enable/Disable CDC **Set:** `core/translate/pragma.rs` - Checks MVCC is not enabled (CDC and MVCC are mutually exclusive) - Parses mode string via `CaptureDataChangesInfo::parse()` with `CDC_VERSION_CURRENT` - Emits a single `InitCdcVersion` opcode — all CDC setup (table creation, version tracking, state change) happens at execution time **Get (read current mode):** `core/translate/pragma.rs` - Returns 3 columns: `mode`, `table`, `version` - When off: returns `("off", NULL, NULL)` - When active: returns `(mode_name, table, version)` **Pragma registration:** `core/pragma.rs` — `CaptureDataChangesConn` (and deprecated alias `UnstableCaptureDataChangesConn`) with columns `["mode", "table", "version"]` ### 2. Connection State **Field:** `core/connection.rs` — `capture_data_changes: RwLock<Option<CaptureDataChangesInfo>>` **Getter:** `get_capture_data_changes_info()` — returns read guard **Setter:** `set_capture_data_changes_info(opts: Option<CaptureDataChangesInfo>)` **Default:** initialized as `None` (CDC off) ### 3. ProgramBuilder Integration **Field:** `core/vdbe/builder.rs` — `capture_data_changes_info: Option<CaptureDataChangesInfo>` **Accessor:** `capture_data_changes_info()` — returns `&Option<CaptureDataChangesInfo>` **Passed from:** `core/translate/mod.rs` — read from connection when creating builder ### 4. PrepareContext **Field:** `core/vdbe/mod.rs` — `capture_data_changes: Option<CaptureDataChangesInfo>` **Set from:** `PrepareContext::from_connection()` — clones from `connection.get_capture_data_changes_info()` ### 5. InitCdcVersion Opcode — `core/vdbe/execute.rs` Always emitted by PRAGMA SET. Handles all CDC setup at execution time: 1. For "off": stores `None` in `state.pending_cdc_info`, returns early 2. Checks if CDC table already exists (for v1 backward compatibility) 3. Creates CDC table (`CREATE TABLE IF NOT EXISTS <cdc_table_name> ...`) — v2 schema with `change_txn_id` column 4. Creates version table (`CREATE TABLE IF NOT EXISTS turso_cdc_version ...`) 5. Inserts version row: if CDC table pre-existed → "v1", otherwise → current version ("v2"). Uses `INSERT OR IGNORE` to preserve existing version rows. 6. Reads back actual version from the table 7. Stores computed `CaptureDataChangesInfo` in `state.pending_cdc_info` The connection's CDC state is **not applied in the opcode**. Instead, `pending_cdc_info` is applied in `halt()` only after the transaction commits successfully. This ensures atomicity: if any step fails and the transaction rolls back, the connection's CDC state remains unchanged. All table creation is done via nested `conn.prepare()`/`run_ignore_rows()` calls rather than bytecode emission, because the PRAGMA plan can't contain DML against tables that don't exist yet in the schema. ## Bytecode Emission (core/translate/emitter.rs) These are the core CDC code generation functions: | Function | Purpose | |----------|---------| | `prepare_cdc_if_necessary()` | Opens CDC table cursor if CDC is active and target != CDC table | | `emit_cdc_full_record()` | Reads all columns from cursor into a MakeRecord (for before/after image) | | `emit_cdc_patch_record()` | Builds record from in-flight register values (for after-image of INSERT/UPDATE) | | `emit_cdc_insns()` | Writes a single CDC row per changed row (INSERT/UPDATE/DELETE). Called per-row inside DML loops. | | `emit_cdc_commit_insns()` | Writes a COMMIT record (change_type=2) into CDC table (v2 only). Raw emission, no autocommit check. | | `emit_cdc_autocommit_commit()` | End-of-statement COMMIT emission. Checks `is_autocommit()` at runtime — only emits COMMIT if in autocommit mode. v2 only. | ### COMMIT Emission Strategy (v2) Per-row call sites use `emit_cdc_insns()` (no COMMIT). End-of-statement sites call `emit_cdc_autocommit_commit()` which checks `is_autocommit()` at runtime: - **Autocommit mode:** emits a COMMIT record after the statement completes - **Explicit transaction (`BEGIN...COMMIT`):** skips per-statement COMMIT; the explicit `COMMIT` statement emits the COMMIT record via `emit_cdc_commit_insns()` This ensures multi-row statements like `INSERT INTO t VALUES (1),(2),(3)` produce one COMMIT at the end, not one per row. ## Integration Points — Where CDC Records Are Emitted ### INSERT — `core/translate/insert.rs` - **Per-row:** `emit_cdc_insns()` after insert, and before delete for REPLACE/conflict - **End-of-statement:** `emit_cdc_autocommit_commit()` in `emit_epilogue()` after the insert loop ### UPDATE — `core/translate/emitter.rs` - **Per-row:** captures before-image, after-image via patch record, emits `emit_cdc_insns()` - **End-of-statement:** `emit_cdc_autocommit_commit()` after the update loop ### DELETE — `core/translate/emitter.rs` - **Per-row:** captures before-image and emits `emit_cdc_insns()` - **End-of-statement:** `emit_cdc_autocommit_commit()` after the delete loop ### UPSERT (ON CONFLICT DO UPDATE) — `core/translate/upsert.rs` - **Per-row:** `emit_cdc_insns()` for all three cases: pure insert, update after conflict, replace - No end-of-statement COMMIT — upsert shares INSERT's epilogue ### Schema Changes (DDL) — `core/translate/schema.rs` - **CREATE TABLE:** `emit_cdc_insns()` (insert into `sqlite_schema`) + `emit_cdc_autocommit_commit()` - **DROP TABLE:** `emit_cdc_insns()` per-row in metadata loop + `emit_cdc_autocommit_commit()` after loop - **CREATE INDEX:** `emit_cdc_insns()` + `emit_cdc_autocommit_commit()` (`core/translate/schema.rs`) - **DROP INDEX:** `emit_cdc_insns()` per-row + `emit_cdc_autocommit_commit()` after loop (`core/translate/index.rs`) DDL in explicit transactions (`BEGIN; CREATE TABLE t(x); COMMIT`) does NOT emit per-statement COMMIT — the autocommit check prevents it. ### ALTER TABLE — `core/translate/update.rs` - Sets `cdc_update_alter_statement` on the update plan when CDC has updates mode ### Views/Triggers — Explicitly excluded - `core/translate/view.rs` — passes `None` for CDC cursor - `core/translate/trigger.rs` — passes `None` for CDC cursor ### Subqueries — No CDC - `core/translate/subquery.rs` — `cdc_cursor_id: None` ## Helper Functions (for reading CDC data) ### `table_columns_json_array(table_name)` — `core/function.rs`, `core/vdbe/execute.rs` Returns JSON array of column names for a table. Used to interpret binary records. ### `bin_record_json_object(columns_json, blob)` — `core/function.rs`, `core/vdbe/execute.rs` Decodes a binary record (from `before`/`after`/`updates` columns) into a JSON object using column names. ## Sync Engine Integration The sync engine is the primary consumer of CDC data. ### DatabaseTape — `sync/engine/src/database_tape.rs` - **CDC config:** `DEFAULT_CDC_TABLE_NAME = "turso_cdc"`, `DEFAULT_CDC_MODE = "full"` - **PRAGMA name:** `CDC_PRAGMA_NAME = "capture_data_changes_conn"` - **Initialization:** `connect()` sets CDC pragma and caches `cdc_version` from `turso_cdc_version` table. Must be called before `iterate_changes()`. - **Version caching:** `cdc_version: RwLock<Option<CdcVersion>>` — set by `connect()`, read by `iterate_changes()`. Panics if not set. - **Iterator:** `DatabaseChangesIterator` reads CDC table in batches, emits `DatabaseTapeOperation`. For v2, real COMMIT records from the table are emitted. For v1, a synthetic Commit is appended at end of batch. `ignore_schema_changes: true` (default) filters out `sqlite_schema` row changes but not COMMIT records.