cdc

CDC（変更データキャプチャ）— 内部機能マップ

概要

CDC は、データベーステーブルの INSERT／UPDATE／DELETE の変更を追跡し、専用の CDC テーブル（デフォルトでは turso_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 の動作は、次の 2 つの型によって制御されます。

#[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 の場合に true を返し、COMMIT レコードの出力を制御します
• Display／FromStr — "v1" ↔ V1、"v2" ↔ V2 を相互変換します

CaptureDataChangesInfo の主なメソッド：

• parse(value: &str, version: Option<CdcVersion>) — PRAGMA 引数 "<mode>[,<table_name>]" を解析し、"off" の場合は None を返します
• cdc_version() — CdcVersion を返します（version が None の場合はパニックします）。旧来の is_v1()／is_v2()／version() メソッドに代わる単一のアクセサーです。
• has_before()／has_after()／has_updates() — モードの機能を確認します
• mode_name() — モードを文字列として返します

Option<CaptureDataChangesInfo> に対する便宜的なトレイト 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 行をトランザクションごとにグループ化します。接続ごとのトランザクション ID を取得または設定する conn_txn_id(candidate) オペコードによって割り当てられます。
• change_type=2（COMMIT）レコード — トランザクション境界を示します。自動コミットモードではステートメントごとに 1 回、明示的な 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 オペコードを 1 つ出力します。CDC のセットアップ（テーブル作成、バージョン追跡、状態変更）はすべて実行時に行われます

取得（現在のモードの読み取り）： core/translate/pragma.rs

• mode、table、version の 3 列を返します
• オフの場合：("off", NULL, NULL) を返します
• 有効な場合：(mode_name, table, version) を返します

Pragma の登録： core/pragma.rs — ["mode", "table", "version"] 列を持つ CaptureDataChangesConn（および非推奨のエイリアス UnstableCaptureDataChangesConn）

2. 接続状態

フィールド： core/connection.rs — capture_data_changes: RwLock<Option<CaptureDataChangesInfo>> ゲッター： get_capture_data_changes_info() — 読み取りガードを返します セッター： 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 — ビルダーの作成時に接続から読み取ります

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 状態はオペコード内では適用されません。代わりに、トランザクションが正常にコミットされた後に限り、halt() で pending_cdc_info を適用します。これによりアトミック性を確保します。いずれかの手順が失敗してトランザクションがロールバックされた場合、接続の CDC 状態は変更されません。

スキーマにまだ存在しないテーブルに対する DML を PRAGMA の計画に含められないため、テーブルの作成にはバイトコードの出力ではなく、ネストした conn.prepare()／run_ignore_rows() 呼び出しを使用します。

バイトコードの出力（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()	変更された各行につき 1 つの CDC 行を書き込みます（INSERT／UPDATE／DELETE）。DML ループ内で行ごとに呼び出されます。
emit_cdc_commit_insns()	CDC テーブルに COMMIT レコード（change_type=2）を書き込みます（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) のような複数行ステートメントでは、行ごとに 1 つずつではなく、最後に 1 つの COMMIT が出力されます。

統合ポイント — CDC レコードの出力箇所

INSERT — core/translate/insert.rs

• 行ごと： 挿入後に emit_cdc_insns() を呼び出し、REPLACE／競合の場合は削除の前にも呼び出します
• ステートメント末尾： 挿入ループ後の emit_epilogue() で emit_cdc_autocommit_commit() を呼び出します

UPDATE — core/translate/emitter.rs

• 行ごと： 変更前のイメージを取得し、パッチレコードで変更後のイメージを取得して、emit_cdc_insns() を出力します
• ステートメント末尾： 更新ループ後に emit_cdc_autocommit_commit() を呼び出します

DELETE — core/translate/emitter.rs

• 行ごと： 更新前イメージを取得し、emit_cdc_insns() を出力します
• ステートメントの終了時： delete ループの後に emit_cdc_autocommit_commit() を出力します

UPSERT（ON CONFLICT DO UPDATE）— core/translate/upsert.rs

• 行ごと： 次の3つすべてのケースで emit_cdc_insns() を出力します：純粋な挿入、競合後の更新、置換
• ステートメントの終了時に COMMIT はありません — upsert は INSERT のエピローグを共有します

スキーマ変更（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() が読み取ります。設定されていない場合はパニックします。
• イテレーター： 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. バイトコードレベルでの実装。 変換時に、実際の DML バイトコードとともに CDC 命令が生成されます。実行時フックやトリガーはありません。
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 を介して保留され、Halt 時にのみ適用されます。PRAGMA のディスク書き込みに失敗してトランザクションがロールバックされた場合、接続状態は変更されません。
9. ステートメント単位の COMMIT（v2）。 emit_cdc_autocommit_commit() が実行時に is_autocommit() を確認し、COMMIT レコードを行単位ではなくステートメント単位で一度だけ生成します。明示的なトランザクションでは、最後の 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 の設定時とジャーナルモードの切り替え時に確認されます。