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

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

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

Option<CaptureDataChangesInfo>

• 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

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

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

core/lib.rs

core/translate/pragma.rs

Version Detection in InitCdcVersion

The

InitCdcVersion

• 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()

into_revert()

OperationMode

—

core/translate/emitter.rs

Used by

emit_cdc_insns()

change_type

• 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

UnstableCaptureDataChangesConn

["mode", "table", "version"]

2. Connection State

Field:

core/connection.rs

capture_data_changes: RwLock<Option<CaptureDataChangesInfo>>

get_capture_data_changes_info()

set_capture_data_changes_info(opts: Option<CaptureDataChangesInfo>)

None

3. ProgramBuilder Integration

Field:

core/vdbe/builder.rs

capture_data_changes_info: Option<CaptureDataChangesInfo>

capture_data_changes_info()

&Option<CaptureDataChangesInfo>

core/translate/mod.rs

4. PrepareContext

Field:

core/vdbe/mod.rs

capture_data_changes: Option<CaptureDataChangesInfo>

PrepareContext::from_connection()

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

halt()

All table creation is done via nested

conn.prepare()

run_ignore_rows()

Bytecode Emission (core/translate/emitter.rs)

These are the core CDC code generation functions:

prepare_cdc_if_necessary()

emit_cdc_full_record()

emit_cdc_patch_record()

emit_cdc_insns()

emit_cdc_commit_insns()

emit_cdc_autocommit_commit()

is_autocommit()

COMMIT Emission Strategy (v2)

Per-row call sites use

emit_cdc_insns()

emit_cdc_autocommit_commit()

is_autocommit()

• 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)

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

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

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.

Sync Operations —

sync/engine/src/database_sync_operations.rs

• Change counting:

  SELECT COUNT(*) FROM turso_cdc WHERE change_id > ?

Sync Engine —

sync/engine/src/database_sync_engine.rs

• Initialization:

  open_db()

  calls

  main_tape.connect(coro)

  to ensure CDC is set up and version is cached before any

  iterate_changes()

  calls.
• During

  apply_changes

  , checks if CDC table existed, re-creates it after sync

Replay Generator —

sync/engine/src/database_replay_generator.rs

• Requires

  updates

  column to be populated (full mode)

Bindings CDC Surface

All bindings expose

cdc_operations

bindings/python/src/turso_sync.rs

bindings/javascript/sync/src/lib.rs

bindings/javascript/sync/src/generator.rs

bindings/go/bindings_sync.go

bindings/react-native/src/types.ts

sync/sdk-kit/turso_sync.h

sync/sdk-kit/src/bindings.rs

Tests

• Integration tests:

  tests/integration/functions/test_cdc.rs

  — covers all modes, CRUD, transactions, schema changes, version table, backward compatibility. Registered in

  tests/integration/functions/mod.rs

  .
• Sync engine tests:

  sync/engine/src/database_tape.rs

  — CDC table reads, tape iteration, replay of schema changes.
• JS binding tests:

  bindings/javascript/sync/packages/{wasm,native}/promise.test.ts

Run:

cargo test -- test_cdc

cargo test -p turso_sync_engine -- database_tape

User-facing Documentation

• CLI manual page:

  cli/manuals/cdc.md

  — accessible via

  .manual cdc

  in the REPL
• Database manual:

  docs/manual.md

  — CDC section linked in TOC

Key Design Decisions

1. Per-connection, not per-database. Each connection has its own CDC mode and can target different tables.
2. Bytecode-level implementation. CDC instructions are emitted alongside the actual DML bytecode during translation — no runtime hooks or triggers.
3. Self-exclusion. Changes to the CDC table and

   turso_cdc_version

   table are never captured (checked in

   prepare_cdc_if_necessary

   ).
4. Schema changes tracked. DDL operations are recorded as changes to

   sqlite_schema

   table.
5. Binary record format. Before/after/updates columns use SQLite's MakeRecord format (same as B-tree payload).
6. Transaction-aware. CDC writes happen within the same transaction as the DML, so rollback naturally discards CDC entries.
7. Version tracking. CDC schema version is recorded in

   turso_cdc_version

   table and carried in

   CaptureDataChangesInfo.version

   for future schema evolution.
8. Atomic PRAGMA. Connection CDC state is deferred via

   pending_cdc_info

   in

   ProgramState

   and applied only at Halt. If the PRAGMA's disk writes fail and the transaction rolls back, the connection state stays unchanged.
9. Per-statement COMMIT (v2). COMMIT records are emitted once per statement (not per row), using

   emit_cdc_autocommit_commit()

   which checks

   is_autocommit()

   at runtime. In explicit transactions, only the final

   COMMIT

   emits a COMMIT CDC record.
10. Backward-compatible version detection. Pre-existing v1 CDC tables (without

    turso_cdc_version

    ) are detected by checking table existence before creation. Existing tables get

    CdcVersion::V1

    inserted into the version table.
11. Typed version enum.

    CdcVersion

    enum with

    #[repr(u8)]

    and

    Ord

    /

    PartialOrd

    enables feature gating via integer comparison (

    has_commit_record()

    =

    self >= V2

    ).

    Display

    /

    FromStr

    handles database round-trip.
12. CDC and MVCC mutual exclusion. Enabling CDC when MVCC is active (or vice versa) returns an error. Checked at PRAGMA set time and journal mode switch time.