# CDC（变更数据捕获）- 内部特征图 ## 概述 CDC 通过将更改记录写入专用 CDC 表（默认情况下为“turso_cdc”）来跟踪数据库表上的 INSERT/UPDATE/DELETE 更改。它是针对每个连接的，通过 PRAGMA 启用，并在字节码生成（翻译）层运行。同步引擎使用 CDC 记录将本地更改推送到远程。 ## 架构图用户 SQL（插入/更新/删除/DDL） | v ┌──────────────────────────────────────────────────┐ │ 翻译层（核心/翻译/） │ │ ┌──────────────────────────────────────────┐ │ │ │ 准备_cdc_if_necessary() │ │ │ │ - 检查 CaptureDataChangesInfo │ │ │ │ - 打开 CDC 表游标 (OpenWrite) │ │ │ │ - 如果目标 == CDC 表本身则跳过 │ │ │ └────────────────────────────────────────────┘ │ │ ┌──────────────────────────────────────────┐ │ │ │ emit_cdc_insns() │ │ │ │ - 写入（change_id、change_time、 │ │ │ │ 更改类型，表名，id， │ │ │ │ 进入 CDC 表格之前、之后、更新） │ │ │ └────────────────────────────────────────────┘ │ │ + emit_cdc_full_record() / emit_cdc_patch_record() │ └──────────────────────────────────────────────────┘ | v CDC 表（turso_cdc 或自定义名称） | v ┌──────────────────────────────────────────────────┐ │ 同步引擎（sync/engine/） │ │ DatabaseTape 读取 CDC 表 → DatabaseChange │ │ → 应用/恢复 → 推送到远程 │ └──────────────────────────────────────────────────┘ ## 核心数据类型 ### `CaptureDataChangesMode` \+ `CaptureDataChangesInfo` — `core/lib.rs` CDC 行为由两种类型控制： #[导出（调试、克隆、复制、Eq、PartialEq、Ord、PartialOrd）] #[代表(u8)] 枚举 CdcVersion { V1=1， V2=2， } const CDC_VERSION_CURRENT: CdcVersion = CdcVersion::V2; 枚举 CaptureDataChangesMode { id, // 只捕获rowid Before, // 捕获前图像 After, // 捕捉残像 Full, // 更新前+更新后 } 结构CaptureDataChangesInfo { 模式：CaptureDataChangesMode， table: String, // CDC 表名 version: Option, // 架构版本（V1 或 V2） } 连接存储“Option”——“None”表示 CDC 已关闭。 `CdcVersion` 的关键方法： * `has_commit_record()` — `self >= V2`，gates COMMIT 记录发射 * `Display`/`FromStr` — 往返 `"v1"` ↔ `V1`, `"v2"` ↔ `V2` `CaptureDataChangesInfo` 的关键方法： * `parse(value: &str, version: Option)` — 解析 PRAGMA 参数 `"[,]"`，对于“off”返回 `None` * `cdc_version()` — 返回 `CdcVersion` （如果版本为 None 则出现恐慌）。单个访问器替换旧的 `is_v1()`/`is_v2()`/`version()` 方法。 * `has_before()` / `has_after()` / `has_updates()` — 模式功能检查 * `mode_name()` — 以字符串形式返回模式 `Option` 上的便利特征 `CaptureDataChangesExt` 提供： * `has_before()` / `has_after()` / `has_updates()` — 委托给内部，对于 None 返回 false * `table()` — 返回 `Option<&str>`，当 CDC 关闭时返回 None ### CDC 表架构 v1 默认表名称：“turso_cdc”（常量“TURSO_CDC_DEFAULT_TABLE_NAME”）创建表 turso_cdc ( change_id 整数主键自动增量， change_time INTEGER, -- unixepoch() change_type INTEGER, -- 1=插入，0=更新，-1=删除表名文本， id , -- 更改行的 rowid before BLOB, -- 二进制记录（前映像） after BLOB, -- 二进制记录（残像）更新 BLOB——每列更改的二进制记录）； ### CDC 表架构 v2（当前）创建表 turso_cdc ( change_id 整数主键自动增量， change_time INTEGER, -- unixepoch() change_type INTEGER，-- 1=插入，0=更新，-1=删除，2=提交表名文本， id , -- 更改行的 rowid before BLOB, -- 二进制记录（前映像） after BLOB, -- 二进制记录（残像）更新 BLOB, -- 每列更改的二进制记录 change_txn_id INTEGER -- 事务 ID（将行分组为事务））； v2 增加： * `change_txn_id` 列 — 按事务对 CDC 行进行分组。通过“conn_txn_id(candidate)”操作码分配，该操作码获取或设置每个连接的事务 ID。 * `change_type=2` (COMMIT) 记录 — 标记事务边界。在自动提交模式下，或在显式“COMMIT”上，每个语句发出一次。 CDC 表是在运行时由“InitCdcVersion”操作码通过“CREATE TABLE IF NOT EXISTS”创建的。 ### CDC 版本表首次启用 CDC 时，会创建版本跟踪表：创建表 turso_cdc_version ( 表名文本主键，版本文本不为空）；当前版本：`CDC_VERSION_CURRENT = CdcVersion::V2`（在`core/lib.rs`中定义，从`core/translate/pragma.rs`重新导出） ### InitCdcVersion 中的版本检测 `InitCdcVersion` 操作码通过在创建 CDC 表之前检查 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` 值： * `插入` → 1 * `更新` / `选择` → 0 * `删除` → -1 * `COMMIT` → 2（仅限 v2，由 `emit_cdc_commit_insns` 发出） ## 入口点 ### 1\. PRAGMA — 启用/禁用 CDC **设置：** `core/translate/pragma.rs` * 检查MVCC是否启用（CDC和MVCC是互斥的） * 通过 `CaptureDataChangesInfo::parse()` 和 `CDC_VERSION_CURRENT` 解析模式字符串 * 发出单个“InitCdcVersion”操作码 - 所有 CDC 设置（表创建、版本跟踪、状态更改）都在执行时发生 **获取（读取当前模式）：** `core/translate/pragma.rs` * 返回 3 列：`mode`、`table`、`version` * 关闭时：返回 `("off", NULL, NULL)` * 激活时：返回 `(mode_name, table, version)` **编译指示注册：** `core/pragma.rs` — `CaptureDataChangesConn` （以及已弃用的别名 `UnstableCaptureDataChangesConn`），其中包含列 `["mode", "table", "version"]` ### 2\.连接状态 **字段：** `core/connection.rs` — `capture_data_changes: RwLock>` **Getter：** `get_capture_data_changes_info()` — 返回读取防护 **Setter：** `set_capture_data_changes_info(opts: Option)` **默认值：** 初始化为“None”（CDC 关闭） ### 3\.程序构建器集成 **字段：** `core/vdbe/builder.rs` — `capture_data_changes_info: Option` **访问器：** `capture_data_changes_info()` — 返回 `&Option` **传递自：** `core/translate/mod.rs` — 创建构建器时从连接读取 ### 4\.准备上下文 **字段：** `core/vdbe/mod.rs` — `capture_data_changes: Option` **设置自：** `PrepareContext::from_connection()` — 来自 `connection.get_capture_data_changes_info()` 的克隆 ### 5\. InitCdcVersion 操作码 — `core/vdbe/execute.rs` 始终由 PRAGMA SET 发出。在执行时处理所有 CDC 设置： 1. 对于“off”：将“None”存储在“state.pending_cdc_info”中，提前返回 2.检查CDC表是否已经存在（为了v1向后兼容） 3. 创建 CDC 表 (`CREATE TABLE IF NOT EXISTS ...`) — 具有 `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 状态**未应用于操作码**。相反，只有在事务成功提交后，才会在“halt()”中应用“pending_cdc_info”。这确保了原子性：如果任何步骤失败并且事务回滚，连接的 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()` |从运行中的寄存器值构建记录（用于插入/更新的后像） `emit_cdc_insns()` |每个更改的行写入一个 CDC 行（插入/更新/删除）。在 DML 循环内按行调用。 `emit_cdc_commit_insns()` |将 COMMIT 记录 (change_type=2) 写入 CDC 表（仅限 v2）。原始发射，没有自动提交检查。 `emit_cdc_autocommit_commit()` |语句结束 COMMIT 发射。在运行时检查“is_autocommit()”——仅在自动提交模式下发出 COMMIT。仅限 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 记录的发出位置 ### 插入 — `core/translate/insert.rs` * **每行：** `emit_cdc_insns()` 插入后，删除前替换/冲突 * **语句结束：**插入循环后`emit_epilogue()`中的`emit_cdc_autocommit_commit()` ### 更新 — `core/translate/emitter.rs` * **每行：** 通过补丁记录捕获前像、后像，发出 `emit_cdc_insns()` * **语句结束：** 更新循环后的 `emit_cdc_autocommit_commit()` ### 删除 — `core/translate/emitter.rs` * **每行：** 捕获前图像并发出 `emit_cdc_insns()` * **语句结束：** 删除循环后的 `emit_cdc_autocommit_commit()` ### UPSERT（冲突时更新）- `core/translate/upsert.rs` * **每行：** `emit_cdc_insns()` 适用于所有三种情况：纯插入、冲突后更新、替换 * 没有语句结束 COMMIT — upsert 共享 INSERT 的尾声 ### 架构更改 (DDL) — `core/translate/schema.rs` * **创建表：** `emit_cdc_insns()` （插入到 `sqlite_schema`）+ `emit_cdc_autocommit_commit()` * **删除表：** 元数据循环中的每行 `emit_cdc_insns()` + 循环后的 `emit_cdc_autocommit_commit()` * **创建索引：** `emit_cdc_insns()` \+ `emit_cdc_autocommit_commit()` (`core/translate/schema.rs`) * **删除索引：** 每行 `emit_cdc_insns()` + 循环后的 `emit_cdc_autocommit_commit()` (`core/translate/index.rs`) 显式事务中的 DDL (`BEGIN; CREATE TABLE t(x); COMMIT`) 不会发出每条语句 COMMIT — 自动提交检查会阻止它。 ### 更改表 — `core/translate/update.rs` * 当 CDC 具有更新模式时，在更新计划上设置 `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>` — 由 `connect()` 设置，由 `iterate_changes()` 读取。如果没有设置会出现恐慌。 * **迭代器：** `DatabaseChangesIterator` 批量读取 CDC 表，发出 `DatabaseTapeOperation`。对于 v2，会发出表中的真实 COMMIT 记录。对于 v1，在批处理结束时附加一个合成提交。 `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”列（完整模式） ## 绑定 CDC 表面所有绑定都将“cdc_operations”公开为同步统计信息的一部分：绑定|文件 ---|--- 蟒蛇 | `绑定/python/src/turso_sync.rs` JavaScript | `绑定/javascript/sync/src/lib.rs` JS（发电机）| `绑定/javascript/sync/src/generator.rs` 去 | `绑定/go/bindings_sync.go` 反应本机 | `绑定/react-native/src/types.ts` SDK 套件（C 头文件）| `sync/sdk-kit/turso_sync.h` SDK 套件（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 树负载相同）。 6. **事务感知。** CDC 写入与 DML 发生在同一事务内，因此回滚自然会丢弃 CDC 条目。 7. **版本跟踪。** CDC 模式版本记录在“turso_cdc_version”表中，并在“CaptureDataChangesInfo.version”中携带，以供将来的模式演变。 8. **原子 PRAGMA。** 连接 CDC 状态通过“ProgramState”中的“pending_cdc_info”推迟，并且仅在停止时应用。如果 PRAGMA 的磁盘写入失败且事务回滚，则连接状态保持不变。 9. **每语句 COMMIT (v2)。** 每个语句（而不是每行）使用 `emit_cdc_autocommit_commit()` 发出一次 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设置时间和日志模式切换时间。

CDC

包含技能