> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/connectors/databases/cdc.md). # Change data capture **Source** e6 Ingestion 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. e6 Ingestion 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 1. e6 Ingestion Engine takes an initial snapshot when the selected snapshot mode requires one. 2. Debezium continues from the corresponding binlog position or PostgreSQL LSN. 3. Changes are grouped by source table and written to that table's Iceberg destination. 4. Source progress and Iceberg data and delete files are coordinated through e6 Ingestion Engine checkpoints. 5. A checkpoint completes only after the Iceberg catalog commit succeeds. The result is continuous, exactly-once replication into Iceberg. After a restart, e6 Ingestion Engine resumes from the last completed checkpoint without omitting or applying committed changes twice. See [Shared data services](/ingestion-engine/architecture/shared-data-services.md) 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. ```yaml 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. e6 Ingestion Engine writes new row versions and removes the replaced or deleted versions as part of the same checkpoint. e6 Ingestion 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 e6 Ingestion 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](https://iceberg.apache.org/spec/#row-level-deletes). This keeps older changes efficient through primary-key matching while avoiding unnecessary scans for rows written or replaced during the current checkpoint. ```yaml 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 ```bash 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": "", "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 ```bash 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": "", "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 e6 Ingestion 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. e6 Ingestion Engine derives a destination table name from each captured source table and creates the table when it first receives data. ```yaml 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](/ingestion-engine/connectors/lakehouse/iceberg.md) for catalog authentication, object-store credentials, Parquet settings, and table properties. ## Exactly-once commits e6 Ingestion Engine coordinates database progress with Iceberg through checkpoint barriers and two-phase commits: 1. Writers finish the data files and delete information produced for the checkpoint. 2. The checkpoint records the corresponding database position. 3. e6 Ingestion Engine commits the files as one atomic Iceberg snapshot for each target table. 4. 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, e6 Ingestion Engine recovers the checkpoint and recognizes the committed work instead of applying it again. ## Database prerequisites ### MySQL Enable row-based binary logging: ```ini [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: ```sql CREATE USER 'ingestion_engine_cdc_reader'@'%' IDENTIFIED BY ''; 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: ```ini wal_level = logical ``` Create a user with replication and table-read privileges: ```sql CREATE USER ingestion_engine_cdc_reader WITH REPLICATION 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: ```sql 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 | --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.e6data.com/ingestion-engine/connectors/databases/cdc.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.