본문 바로가기
WIKI 기술 지식 베이스

Change Data Capture

원문 보기 위키 갱신

데이터베이스 테이블에 일어난 모든 데이터 변경(삽입, 갱신, 삭제)을 추적하는 Change Data Capture(CDC) 기능이에요. 변경 기록을 일반 테이블처럼 쿼리할 수 있어서 반응형 애플리케이션 구축, 시스템 간 데이터 동기화, 감사(auditing) 등에 유용해요.

출처: 문서

본문

Change Data Capture(CDC)는 데이터베이스 테이블에 가해진 모든 데이터 변경(INSERT, UPDATE, DELETE)을 추적해요. 변경은 다른 테이블처럼 쿼리할 수 있는 로컬 테이블에 기록되고, 이를 활용해 반응형 애플리케이션을 만들거나 시스템 간 데이터를 동기화하거나 감사 로그를 남길 수 있어요.

Enable CDC

PRAGMA로 연결 단위에서 CDC를 켜요.

PRAGMA capture_data_changes_conn('full');

끄는 방법은 이래요.

PRAGMA capture_data_changes_conn('off');
CDC는 [MVCC](/tursodb/concurrent-writes)와 함께 사용할 수 없어요. 같은 연결에서 둘은 상호 배타적이에요.

Capture Modes

Mode Description
id 변경된 행의 기본 키/rowid만 기록해요
before 변경 전의 행 상태를 기록해요 (UPDATE/DELETE)
after 변경 후의 행 상태를 기록해요 (INSERT/UPDATE)
full 변경 전과 후 상태, 그리고 컬럼별 갱신 상세까지 기록해요

CDC Table

변경은 기본적으로 turso_cdc 테이블에 저장돼요. 커스텀 테이블 이름을 지정할 수도 있어요.

PRAGMA capture_data_changes_conn('full,my_changes_table');

CDC 테이블의 스키마는 다음과 같아요.

Column Type Description
change_id INTEGER 자동 증가하는 고유 식별자
change_time INTEGER 타임스탬프 (Unix epoch)
change_txn_id INTEGER 트랜잭션 ID (행들을 트랜잭션 단위로 묶어요)
change_type INTEGER 1 = INSERT, 0 = UPDATE, -1 = DELETE, 2 = COMMIT
table_name TEXT 변경된 테이블의 이름
id varies 변경된 행의 기본 키/rowid
before BLOB 변경 전 행 데이터 (모드: before, full)
after BLOB 변경 후 행 데이터 (모드: after, full)
updates BLOB 컬럼별 변경 상세 (모드: full)

Querying Changes

-- 모든 변경
SELECT * FROM turso_cdc;

-- INSERT만
SELECT * FROM turso_cdc WHERE change_type = 1;

-- UPDATE만
SELECT * FROM turso_cdc WHERE change_type = 0;

-- DELETE만
SELECT * FROM turso_cdc WHERE change_type = -1;

-- 특정 테이블의 변경
SELECT * FROM turso_cdc WHERE table_name = 'users';

-- 지난 1시간 동안의 변경
SELECT * FROM turso_cdc WHERE change_time > unixepoch() - 3600;

Decoding Binary Records

before, after, updates 컬럼은 바이너리 형식으로 데이터를 저장해요. 내장 헬퍼 함수로 디코딩할 수 있어요.

SELECT
  change_type,
  table_name,
  id,
  bin_record_json_object(table_columns_json_array('users'), after) AS after_state,
  bin_record_json_object(table_columns_json_array('users'), before) AS before_state
FROM turso_cdc
WHERE table_name = 'users' AND change_type != 2;
Function Description
table_columns_json_array(table_name) 테이블의 컬럼 이름들을 JSON 배열로 돌려줘요 (예: ["id","name","email"])
bin_record_json_object(columns_json, blob) 주어진 컬럼 이름을 기준으로 바이너리 레코드를 JSON 객체로 디코딩해요

Transactions

CDC는 트랜잭션 경계를 존중해요. 변경은 트랜잭션이 커밋될 때만 기록되고, 트랜잭션이 롤백되면 CDC 항목이 만들어지지 않아요.

같은 트랜잭션의 모든 행은 동일한 change_txn_id를 공유하며, 마지막에 COMMIT 레코드(change_type = 2)가 하나 붙어요.

BEGIN;
INSERT INTO users VALUES (1, 'Alice', '[email protected]');
INSERT INTO users VALUES (2, 'Bob', '[email protected]');
UPDATE users SET name = 'Charles' WHERE id = 1;
COMMIT;
-- 변경 3개 + COMMIT 레코드 1개가 같은 change_txn_id를 공유해요

Schema Changes

CDC는 DDL 연산(CREATE TABLE, DROP TABLE, CREATE INDEX 등)도 sqlite_schema 테이블에 대한 변경으로 기록해요.

PRAGMA capture_data_changes_conn('full');

CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT);
-- sqlite_schema에 대한 변경으로 turso_cdc에 기록돼요

Multiple Connections

연결마다 독립적인 CDC 설정을 가질 수 있어요. 서로 다른 연결이 서로 다른 테이블에 기록할 수도 있어요.

-- 연결 1: 'audit_log'에 기록
PRAGMA capture_data_changes_conn('full,audit_log');

-- 연결 2: 'sync_queue'에 기록
PRAGMA capture_data_changes_conn('id,sync_queue');

Example

-- 테이블을 만들어요
CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    name TEXT,
    email TEXT
);

-- full CDC를 켜요
PRAGMA capture_data_changes_conn('full');

-- 몇 가지 변경을 만들어요
INSERT INTO users VALUES (1, 'Alice', '[email protected]');
INSERT INTO users VALUES (2, 'Bob', '[email protected]');
UPDATE users SET email = '[email protected]' WHERE id = 1;
DELETE FROM users WHERE id = 2;

-- 캡처된 변경을 디코딩해서 봐요
SELECT
  change_id,
  CASE change_type
    WHEN 1 THEN 'INSERT'
    WHEN 0 THEN 'UPDATE'
    WHEN -1 THEN 'DELETE'
    WHEN 2 THEN 'COMMIT'
  END AS operation,
  table_name,
  id,
  bin_record_json_object(table_columns_json_array('users'), after) AS after_state
FROM turso_cdc
WHERE change_type != 2
ORDER BY change_id;

더 알아보기 (Learn more)