Change Data Capture (CDC)

ParticleDB supports Change Data Capture (CDC) through changefeeds that stream INSERT, UPDATE, and DELETE events from tables to external systems in real time. Each change event carries a monotonically increasing sequence number for ordering and resumption.

Creating a Changefeed

Kafka Sink

Stream changes to an Apache Kafka topic:

CREATE CHANGEFEED orders_feed
  FOR TABLE orders
  INTO KAFKA 'broker:9092/orders-topic';

The broker address and topic are specified in a single string separated by /.

Webhook Sink

Send changes as HTTP POST requests to a webhook endpoint:

CREATE CHANGEFEED audit_feed
  FOR TABLE users
  INTO WEBHOOK 'https://api.example.com/hook';

Options

Changefeeds accept optional configuration via a WITH clause:

CREATE CHANGEFEED detailed_feed
  FOR TABLE orders
  INTO KAFKA 'broker:9092/orders-topic'
  WITH (
    include_old_values = true,
    batch_size = 500,
    flush_interval_ms = 2000,
    filter = 'status = ''shipped'''
  );

Option	Default	Description
include_old_values	false	Include the old row image on UPDATE and DELETE events
batch_size	100	Number of events to batch before sending
flush_interval_ms	1000	Maximum time (ms) to wait before flushing an incomplete batch
filter	none	SQL WHERE-style expression to filter events

Managing Changefeeds

Pause and Resume

Temporarily stop a changefeed without dropping it. The changefeed retains its position and resumes from where it left off:

PAUSE CHANGEFEED orders_feed;
RESUME CHANGEFEED orders_feed;

Show Active Changefeeds

List all changefeeds with their current status, table, and delivery statistics:

SHOW CHANGEFEEDS;

The output includes:

Column	Description
name	Changefeed name
table	Watched table
state	Running, Paused, or Error
last_sequence	Last delivered sequence number
events_delivered	Total events delivered since creation

Drop a Changefeed

DROP CHANGEFEED orders_feed;

Change Event Format

Each change event is serialized as JSON with the following structure:

{
  "sequence": 42,
  "timestamp": "2025-03-15T10:30:00Z",
  "table": "orders",
  "operation": "INSERT",
  "key": {"id": 1001},
  "new_row": {"id": 1001, "customer": "acme", "amount": 250.00},
  "old_row": null
}

Field	Presence	Description
sequence	Always	Monotonic sequence number for ordering and resumption
timestamp	Always	Wall-clock time when the change was captured
table	Always	Name of the modified table
operation	Always	INSERT, UPDATE, or DELETE
key	Always	Primary key value(s) of the affected row
new_row	INSERT, UPDATE	The new row image
old_row	UPDATE, DELETE	The old row image (only when include_old_values is enabled)

For Kafka sinks, the message key is formatted as table:pk_json to enable partition routing by primary key.

Architecture

Each changefeed runs as a background task that:

1. Tails the WAL via a WalTailer to capture DML events in order.
2. Batches events up to batch_size or until flush_interval elapses.
3. Delivers to the sink (Kafka producer or HTTP POST).
4. Tracks position via the last_sequence counter for resumption after pause or restart.

The changefeed manager orchestrates the lifecycle of all active feeds, handling creation, pause/resume, error recovery, and teardown.