/cdc

name: cdc
description: Change Data Capture - architecture, entrypoints, bytecode emission, sync engine integration, tests

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:

#[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`)

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)

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:

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 `em