cdc

CDC（更改数据捕获）- 内部功能图

概述

CDC 会跟踪数据库表上的 INSERT/UPDATE/DELETE 更改，并将更改记录写入专用 CDC 表（默认名为 turso_cdc）。CDC 按连接启用，通过 PRAGMA 开启，并在字节码生成（转换）层运行。同步引擎会使用 CDC 记录，将本地更改推送到远程端。

架构图

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                 │
  └─────────────────────────────────────────────────┘

核心数据类型

CaptureDataChangesMode + CaptureDataChangesInfo — core/lib.rs

CDC 的行为由两种类型控制：

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

连接会存储 Option<CaptureDataChangesInfo>：None 表示 CDC 已关闭。

CdcVersion 的关键方法：

• has_commit_record() — self >= V2，用于控制是否发出 COMMIT 记录
• Display/FromStr — 在 "v1" ↔ V1、"v2" ↔ V2 之间进行往返转换

CaptureDataChangesInfo 的关键方法：

• parse(value: &str, version: Option<CdcVersion>) — 解析 PRAGMA 参数 "<mode>[,<table_name>]"；如果为 "off"，则返回 None
• cdc_version() — 返回 CdcVersion（如果版本为 None，则会 panic）。这是替代旧 is_v1()/is_v2()/version() 方法的单一访问器。
• has_before() / has_after() / has_updates() — 检查模式支持的功能
• mode_name() — 以字符串形式返回模式

Option<CaptureDataChangesInfo> 上的便捷 trait CaptureDataChangesExt 提供：

• has_before() / has_after() / has_updates() — 委托给内部值；如果为 None，则返回 false
• table() — 返回 Option<&str>；如果 CDC 已关闭，则返回 None

CDC 表架构 v1

默认表名：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 表架构 v2（当前版本）

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 新增：

• change_txn_id 列 — 按事务对 CDC 行分组。通过 conn_txn_id(candidate) 操作码分配事务 ID，该操作码会获取或设置每连接的事务 ID。
• change_type=2（COMMIT）记录 — 标记事务边界。在自动提交模式下，每条语句发出一次；也会在显式执行 COMMIT 时发出。

CDC 表由 InitCdcVersion 操作码在运行时通过 CREATE TABLE IF NOT EXISTS 创建。

CDC 版本表

首次启用 CDC 时，会创建一个版本跟踪表：

CREATE TABLE turso_cdc_version (
    table_name TEXT PRIMARY KEY,
    version TEXT NOT NULL
);

当前版本：CDC_VERSION_CURRENT = CdcVersion::V2（定义于 core/lib.rs，并从 core/translate/pragma.rs 重新导出）

InitCdcVersion 中的版本检测

InitCdcVersion 操作码会在创建 CDC 表之前检查该表是否已存在，以检测 v1 或 v2：

• 如果 CDC 表已存在，但没有版本行 → v1（版本跟踪功能推出前已存在的表）
• 如果 CDC 表不存在 → 使用当前版本（v2）创建
• 如果版本行已存在 → 按原样使用该版本

DatabaseChange — sync/engine/src/types.rs:229-249

同步引擎中 CDC 行的 Rust 表示。具有用于正向和反向重放的 into_apply() 和 into_revert() 方法。

OperationMode — core/translate/emitter.rs

emit_cdc_insns() 使用它来确定 change_type 的值：

• INSERT → 1
• UPDATE / SELECT → 0
• DELETE → -1
• COMMIT → 2（仅限 v2，由 emit_cdc_commit_insns 发出）

入口点

1. PRAGMA — 启用/禁用 CDC

设置： core/translate/pragma.rs

• 检查是否未启用 MVCC（CDC 和 MVCC 互斥）
• 使用 CDC_VERSION_CURRENT 通过 CaptureDataChangesInfo::parse() 解析模式字符串
• 发出单个 InitCdcVersion 操作码 — 所有 CDC 设置（表创建、版本跟踪、状态更改）都在执行时进行

获取（读取当前模式）： core/translate/pragma.rs

• 返回 3 列：mode、table、version
• 关闭时：返回 ("off", NULL, NULL)
• 启用时：返回 (mode_name, table, version)

Pragma 注册： core/pragma.rs — CaptureDataChangesConn（以及已弃用的别名 UnstableCaptureDataChangesConn），列为 ["mode", "table", "version"]

2. 连接状态

字段： core/connection.rs — capture_data_changes: RwLock<Option<CaptureDataChangesInfo>> Getter： get_capture_data_changes_info() — 返回读锁 Setter： set_capture_data_changes_info(opts: Option<CaptureDataChangesInfo>) 默认值： 初始化为 None（CDC 关闭）

3. ProgramBuilder 集成

字段： core/vdbe/builder.rs — capture_data_changes_info: Option<CaptureDataChangesInfo> 访问器： capture_data_changes_info() — 返回 &Option<CaptureDataChangesInfo> 传入来源： core/translate/mod.rs — 创建 builder 时从连接中读取

4. PrepareContext

字段： core/vdbe/mod.rs — capture_data_changes: Option<CaptureDataChangesInfo> 设置来源： PrepareContext::from_connection() — 从 connection.get_capture_data_changes_info() 克隆

5. InitCdcVersion 操作码 — core/vdbe/execute.rs

始终由 PRAGMA SET 发出。在执行时处理所有 CDC 设置：

1. 对于“off”：在 state.pending_cdc_info 中存储 None，并提前返回
2. 检查 CDC 表是否已存在（用于向后兼容 v1）
3. 创建 CDC 表（CREATE TABLE IF NOT EXISTS <cdc_table_name> ...）— 使用带有 change_txn_id 列的 v2 架构
4. 创建版本表（CREATE TABLE IF NOT EXISTS turso_cdc_version ...）
5. 插入版本行：如果 CDC 表此前已存在 → “v1”，否则 → 当前版本（“v2”）。使用 INSERT OR IGNORE 保留现有版本行。
6. 从表中读回实际版本
7. 将计算出的 CaptureDataChangesInfo 存入 state.pending_cdc_info

连接的 CDC 状态不会在操作码中应用。相反，只有在事务成功提交后，pending_cdc_info 才会在 halt() 中应用。这确保了原子性：如果任何步骤失败并导致事务回滚，连接的 CDC 状态将保持不变。

所有表的创建都通过嵌套的 conn.prepare()/run_ignore_rows() 调用完成，而不是通过发出字节码，因为 PRAGMA 计划无法包含针对架构中尚不存在的表执行的 DML。

字节码生成（core/translate/emitter.rs）

以下是 CDC 代码生成的核心函数：

函数	用途
prepare_cdc_if_necessary()	如果 CDC 已启用且目标不是 CDC 表，则打开 CDC 表游标
emit_cdc_full_record()	从游标读取所有列并放入 MakeRecord（用于变更前／后镜像）
emit_cdc_patch_record()	根据正在使用的寄存器值构建记录（用于 INSERT／UPDATE 的变更后镜像）
emit_cdc_insns()	为每个已变更的行（INSERT／UPDATE／DELETE）写入一条 CDC 行。在 DML 循环中逐行调用。
emit_cdc_commit_insns()	将 COMMIT 记录（change_type=2）写入 CDC 表（仅限 v2）。直接生成，不检查自动提交状态。
emit_cdc_autocommit_commit()	在语句结束时生成 COMMIT。运行时检查 is_autocommit() — 仅在自动提交模式下生成。仅限 v2。

COMMIT 生成策略（v2）

逐行调用点使用 emit_cdc_insns()（不生成 COMMIT）。语句结束时的调用点会调用 emit_cdc_autocommit_commit()，该函数会在运行时检查 is_autocommit()：

• 自动提交模式： 语句完成后生成一条 COMMIT 记录
• 显式事务（BEGIN...COMMIT）： 跳过每条语句后的 COMMIT；显式的 COMMIT 语句通过 emit_cdc_commit_insns() 生成 COMMIT 记录

这样可以确保像 INSERT INTO t VALUES (1),(2),(3) 这样的多行语句只在最后生成一条 COMMIT，而不是每行一条。

集成点 — CDC 记录的生成位置

INSERT — core/translate/insert.rs

• 逐行： 插入后调用 emit_cdc_insns()；对于 REPLACE／冲突，在删除前调用
• 语句结束时： 插入循环后，在 emit_epilogue() 中调用 emit_cdc_autocommit_commit()

UPDATE — core/translate/emitter.rs

• 逐行： 捕获变更前镜像，通过 patch record 捕获变更后镜像，并生成 emit_cdc_insns()
• 语句结束时： 更新循环后调用 emit_cdc_autocommit_commit()

DELETE — core/translate/emitter.rs

• 每行： 捕获前像并发出 emit_cdc_insns()
• 语句结束时： 删除循环后调用 emit_cdc_autocommit_commit()

UPSERT (ON CONFLICT DO UPDATE) — core/translate/upsert.rs

• 每行： 在以下三种情况下均调用 emit_cdc_insns()：纯插入、冲突后更新、替换
• 语句结束时不执行 COMMIT — upsert 与 INSERT 共用其收尾逻辑

Schema 变更（DDL）— core/translate/schema.rs

• CREATE TABLE： 调用 emit_cdc_insns()（插入 sqlite_schema）+ emit_cdc_autocommit_commit()
• DROP TABLE： 在元数据循环中逐行调用 emit_cdc_insns() + 循环后调用 emit_cdc_autocommit_commit()
• CREATE INDEX： 调用 emit_cdc_insns() + emit_cdc_autocommit_commit()（core/translate/schema.rs）
• DROP INDEX： 逐行调用 emit_cdc_insns() + 循环后调用 emit_cdc_autocommit_commit()（core/translate/index.rs）

显式事务中的 DDL（BEGIN; CREATE TABLE t(x); COMMIT）不会为每条语句发出 COMMIT — 自动提交检查会阻止这一操作。

ALTER TABLE — core/translate/update.rs

• 当 CDC 处于 updates 模式时，在更新计划上设置 cdc_update_alter_statement

视图/触发器 — 明确排除

• core/translate/view.rs — 为 CDC 游标传入 None
• core/translate/trigger.rs — 为 CDC 游标传入 None

子查询 — 不启用 CDC

• core/translate/subquery.rs — cdc_cursor_id: None

辅助函数（用于读取 CDC 数据）

table_columns_json_array(table_name) — core/function.rs、core/vdbe/execute.rs

返回表的列名 JSON 数组。用于解析二进制记录。

bin_record_json_object(columns_json, blob) — core/function.rs、core/vdbe/execute.rs

使用列名将二进制记录（来自 before/after/updates 列）解码为 JSON 对象。

同步引擎集成

同步引擎是 CDC 数据的主要使用方。

DatabaseTape — sync/engine/src/database_tape.rs

• CDC 配置： DEFAULT_CDC_TABLE_NAME = "turso_cdc"、DEFAULT_CDC_MODE = "full"
• PRAGMA 名称： CDC_PRAGMA_NAME = "capture_data_changes_conn"
• 初始化： connect() 设置 CDC pragma，并从 turso_cdc_version 表缓存 cdc_version。必须在调用 iterate_changes() 之前调用。
• 版本缓存： cdc_version: RwLock<Option<CdcVersion>> — 由 connect() 设置，由 iterate_changes() 读取。若未设置则触发 panic。
• 迭代器： DatabaseChangesIterator 分批读取 CDC 表，并生成 DatabaseTapeOperation。对于 v2，会从表中生成真实的 COMMIT 记录；对于 v1，会在批次末尾追加一个合成的 Commit。ignore_schema_changes: true（默认值）会过滤掉 sqlite_schema 行变更，但不会过滤 COMMIT 记录。

同步操作 — sync/engine/src/database_sync_operations.rs

• 变更计数： SELECT COUNT(*) FROM turso_cdc WHERE change_id > ?

同步引擎 — sync/engine/src/database_sync_engine.rs

• 初始化： open_db() 会调用 main_tape.connect(coro)，确保在任何 iterate_changes() 调用前设置好 CDC 并缓存版本。
• 在 apply_changes 期间，检查 CDC 表是否存在，并在同步后重新创建该表

重放生成器 — sync/engine/src/database_replay_generator.rs

• 要求填充 updates 列（full 模式）

绑定中的 CDC 接口

所有绑定都将 cdc_operations 作为同步统计信息的一部分公开：

绑定	文件
Python	bindings/python/src/turso_sync.rs
JavaScript	bindings/javascript/sync/src/lib.rs
JS（生成器）	bindings/javascript/sync/src/generator.rs
Go	bindings/go/bindings_sync.go
React Native	bindings/react-native/src/types.ts
SDK Kit（C 头文件）	sync/sdk-kit/turso_sync.h
SDK Kit（Rust）	sync/sdk-kit/src/bindings.rs

测试

• 集成测试： tests/integration/functions/test_cdc.rs — 覆盖所有模式、CRUD、事务、架构变更、版本表和向后兼容性。已在 tests/integration/functions/mod.rs 中注册。
• 同步引擎测试： sync/engine/src/database_tape.rs — CDC 表读取、磁带迭代、架构变更重放。
• JS 绑定测试： bindings/javascript/sync/packages/{wasm,native}/promise.test.ts

运行：cargo test -- test_cdc（集成测试）或 cargo test -p turso_sync_engine -- database_tape（同步引擎）。

面向用户的文档

• CLI 手册页： cli/manuals/cdc.md — 可在 REPL 中通过 .manual cdc 访问
• 数据库手册： docs/manual.md — CDC 章节已链接到目录中

关键设计决策

1. 按连接而非按数据库。 每个连接都有自己的 CDC 模式，并且可以指定不同的表。
2. 在字节码层实现。 翻译期间，CDC 指令与实际的 DML 字节码一起生成，不使用运行时钩子或触发器。
3. 排除自身变更。 对 CDC 表和 turso_cdc_version 表的变更不会被捕获（在 prepare_cdc_if_necessary 中检查）。
4. 跟踪架构变更。 DDL 操作会记录为对 sqlite_schema 表的变更。
5. 二进制记录格式。 before/after/updates 列使用 SQLite 的 MakeRecord 格式（与 B-tree 负载相同）。
6. 感知事务。 CDC 写入与 DML 在同一事务中执行，因此回滚会自然丢弃 CDC 条目。
7. 版本跟踪。 CDC 架构版本记录在 turso_cdc_version 表中，并存储在 CaptureDataChangesInfo.version 中，以支持未来的架构演进。
8. 原子 PRAGMA。 连接 CDC 状态通过 ProgramState 中的 pending_cdc_info 延迟应用，并且仅在 Halt 时生效。如果 PRAGMA 的磁盘写入失败且事务回滚，连接状态将保持不变。
9. 按语句提交（v2）。 每条语句只生成一次 COMMIT 记录（而非每行一次），使用 emit_cdc_autocommit_commit() 在运行时检查 is_autocommit()。在显式事务中，只有最终的 COMMIT 才会生成 COMMIT CDC 记录。
10. 向后兼容的版本检测。 通过在创建表之前检查表是否存在，检测预先存在的 v1 CDC 表（没有 turso_cdc_version）。现有表会在版本表中插入 CdcVersion::V1。
11. 类型化版本枚举。 带有 #[repr(u8)] 和 Ord/PartialOrd 的 CdcVersion 枚举支持通过整数比较进行功能门控（has_commit_record() = self >= V2）。Display/FromStr 可处理数据库往返转换。
12. CDC 与 MVCC 互斥。 在 MVCC 已启用时启用 CDC（或反之）会返回错误。在设置 PRAGMA 和切换日志模式时都会进行检查。