Change data capture

Source Ingest Engine captures inserts, updates, and deletes from database transaction logs and writes them continuously to Apache Iceberg. A single CDC source can capture many database tables. Ingest Engine creates and maintains a separate Iceberg table for each source table.

Supported databases

Database	Change log
MySQL	Binary log (binlog)
PostgreSQL	Logical replication (WAL)

How it works

Ingest Engine takes an initial snapshot when the selected snapshot mode requires one.
Debezium continues from the corresponding binlog position or PostgreSQL LSN.
Changes are grouped by source table and written to that table’s Iceberg destination.
Source progress and Iceberg data and delete files are coordinated through Ingest Engine checkpoints.
A checkpoint completes only after the Iceberg catalog commit succeeds.

The result is continuous, exactly-once replication into Iceberg. After a restart, Ingest Engine resumes from the last completed checkpoint without omitting or applying committed changes twice. See Shared data services for how shared CDC capture and durable ingestion can be reused across pipelines.

Write modes

Multi-table Iceberg supports two write modes. Choose the mode based on whether the destination should retain change history or expose the latest state of each source row.

Append-only

Append-only mode writes every captured change as a new Iceberg row. Inserts, updates, deletes, and snapshot reads remain available as a change history with operation metadata. Use append-only mode for:

audit and history tables;
downstream change processing;
replayable raw data layers;
the highest write throughput when a current-state table is not required.

appendOnly: true
multiTable:
  enabled: true
  tableNameTemplate: "{{database}}_{{schema}}_{{table}}"

Full CDC

Full CDC mode maintains the current state of every source row in Iceberg. Source tables must have a stable primary key. Ingest Engine writes new row versions and removes the replaced or deleted versions as part of the same checkpoint. Ingest Engine selects the appropriate Iceberg delete representation for the target table:

Equality deletes match rows in previously committed data files by primary key.
Position delete files target recent rows when Ingest Engine knows their data file and row position. This is the position-based representation used by Iceberg v2 tables.
Deletion vectors encode position-based deletes for Iceberg v3 tables when the catalog and query engines support them.

These representations follow the Iceberg row-level delete model. This keeps older changes efficient through primary-key matching while avoiding unnecessary scans for rows written or replaced during the current checkpoint.

appendOnly: false
multiTable:
  enabled: true
  tableNameTemplate: "{{database}}_{{schema}}_{{table}}"

For full CDC, every captured table must expose its primary key. Floating-point columns cannot be used as Iceberg identifier fields.

Create a CDC source

Create CDC sources through the API. The source schema is discovered from the database, so the request does not repeat every source column.

MySQL

curl -X POST http://localhost:8000/api/v1/connectors/cdc/tables \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "mysql_cdc",
    "config": {
      "transport": {
        "mode": "jni",
        "maxHeap": "2g"
      },
      "database": {
        "type": "mysql",
        "host": "mysql.example.com",
        "port": 3306,
        "user": "ingestion_engine_cdc_reader",
        "password": "<password>",
        "database": "ecommerce",
        "tables": ["orders", "customers"],
        "snapshotMode": "initial"
      },
      "debeziumOverrides": {
        "heartbeat.interval.ms": "10000"
      },
      "channelCapacity": 65536
    }
  }'

Unqualified table names are resolved against database. You can also use fully qualified names such as ecommerce.orders.

PostgreSQL

curl -X POST http://localhost:8000/api/v1/connectors/cdc/tables \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "postgres_cdc",
    "config": {
      "transport": {
        "mode": "jni",
        "maxHeap": "2g"
      },
      "database": {
        "type": "postgres",
        "host": "postgres.example.com",
        "port": 5432,
        "user": "ingestion_engine_cdc_reader",
        "password": "<password>",
        "database": "app",
        "schema": "public",
        "tables": ["orders", "customers"],
        "snapshotMode": "initial",
        "slotName": "ingestion_engine_app_slot",
        "publicationName": "ingestion_engine_app_publication"
      },
      "debeziumOverrides": {},
      "channelCapacity": 65536
    }
  }'

Unqualified PostgreSQL table names are resolved against schema. Use distinct replication slot and publication names for independently running CDC sources.

Source configuration

Property	Required	Default	Description
`transport.mode`	Yes	-	Use `jni` to run the CDC reader with the pipeline worker.
`transport.maxHeap`	No	`2g`	Maximum heap available to the embedded CDC reader.
`database.type`	Yes	-	`mysql` or `postgres`.
`database.host`	Yes	-	Database hostname or IP address.
`database.port`	No	`3306` or `5432`	Database port.
`database.user`	Yes	-	User with replication and table-read privileges.
`database.password`	Yes	-	Database password.
`database.database`	Yes	-	Database to capture.
`database.schema`	PostgreSQL only	`public`	Schema used to qualify unqualified PostgreSQL table names.
`database.tables`	No	All eligible tables	Tables to capture. Enumerating tables is recommended.
`database.snapshotMode`	No	`initial`	Controls whether existing rows are read before streaming changes.
`database.slotName`	PostgreSQL only	Generated from the database name	Logical replication slot.
`database.publicationName`	PostgreSQL only	Generated from the database name	Logical replication publication.
`debeziumOverrides`	No	`{}`	Additional Debezium and Ingest Engine CDC tuning properties.
`channelCapacity`	No	`65536`	Number of CDC batches buffered between the reader and pipeline.

Snapshot modes

Snapshot mode values are passed to the bundled Debezium connector. Common modes are:

Mode	Behavior
`initial`	Read existing rows, then continue streaming changes.
`no_data`	Capture the schema and stream new changes without reading existing rows.
`initial_only`	Read existing rows and stop after the snapshot.
`when_needed`	Take a snapshot only when no usable source position exists.

Configure multi-table Iceberg

Enable multiTable on the Iceberg sink. Ingest Engine derives a destination table name from each captured source table and creates the table when it first receives data.

catalog:
  type: rest
  url: https://catalog.example.com
  warehouse: s3://company-lake/warehouse
namespace: replicated
appendOnly: false
multiTable:
  enabled: true
  tableNameTemplate: "{{database}}_{{schema}}_{{table}}"

Supported template variables are:

Variable	Value
`{{database}}`	Source database name.
`{{schema}}`	Source schema name when present.
`{{table}}`	Source table name.
`{{_table}}`	Complete source table identifier, sanitized for Iceberg.

Include the database and schema in the template when the same table name may appear in multiple source namespaces. See Apache Iceberg for catalog authentication, object-store credentials, Parquet settings, and table properties.

Exactly-once commits

Ingest Engine coordinates database progress with Iceberg through checkpoint barriers and two-phase commits:

Writers finish the data files and delete information produced for the checkpoint.
The checkpoint records the corresponding database position.
Ingest Engine commits the files as one atomic Iceberg snapshot for each target table.
The checkpoint is acknowledged only after the catalog accepts the snapshot.

If a worker or controller fails before the catalog commit, the incomplete checkpoint is not published. If the commit succeeds but its acknowledgement is interrupted, Ingest Engine recovers the checkpoint and recognizes the committed work instead of applying it again.

Database prerequisites

MySQL

Enable row-based binary logging:

[mysqld]
server-id=1
log_bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL
binlog_expire_logs_seconds=604800

Create a dedicated replication user:

CREATE USER 'ingestion_engine_cdc_reader'@'%' IDENTIFIED BY '<password>';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT
  ON *.* TO 'ingestion_engine_cdc_reader'@'%';

The binlog retention period must be longer than the maximum expected pipeline downtime.

PostgreSQL

Enable logical replication:

wal_level = logical

Create a user with replication and table-read privileges:

CREATE USER ingestion_engine_cdc_reader WITH REPLICATION PASSWORD '<password>';
GRANT CONNECT ON DATABASE app TO ingestion_engine_cdc_reader;
GRANT USAGE ON SCHEMA public TO ingestion_engine_cdc_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ingestion_engine_cdc_reader;

For complete old row values on updates and deletes, configure the captured table with an appropriate replica identity:

ALTER TABLE public.orders REPLICA IDENTITY FULL;

Monitor replication slots and remove slots that are no longer used. An inactive slot can prevent PostgreSQL from reclaiming old WAL files.

Choosing a mode

Requirement	Recommended mode
Preserve every source change for auditing or replay	Append-only
Maintain queryable current-state replicas	Full CDC
Highest ingestion throughput	Append-only
Apply source updates and deletes in Iceberg	Full CDC

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Supported databases

How it works

Write modes

Append-only

Full CDC

Create a CDC source

MySQL

PostgreSQL

Source configuration

Snapshot modes

Configure multi-table Iceberg

Exactly-once commits

Database prerequisites

MySQL

PostgreSQL

Choosing a mode

​Supported databases

​How it works

​Write modes

​Append-only

​Full CDC

​Create a CDC source

​MySQL

​PostgreSQL

​Source configuration

​Snapshot modes

​Configure multi-table Iceberg

​Exactly-once commits

​Database prerequisites

​MySQL

​PostgreSQL

​Choosing a mode

Supported databases

How it works

Write modes

Append-only

Full CDC

Create a CDC source

MySQL

PostgreSQL

Source configuration

Snapshot modes

Configure multi-table Iceberg

Exactly-once commits

Database prerequisites

MySQL

PostgreSQL

Choosing a mode