PostgreSQL CDC to Iceberg

What you’ll build

PostgreSQL (orders, customers) ──CDC──▶ Ingest Engine ──▶ Iceberg catalog
                                                           ├── cdc_db.orders
                                                           └── cdc_db.customers

Initial snapshot of all existing rows, then continuous streaming from PostgreSQL’s logical replication
One Iceberg table per source table, auto-created on first write
Every change (insert, update, delete) is appended as a new row with an _op column indicating the operation type, plus _ts_ms and _timestamp for event timing
Exactly-once file commits via checkpointed two-phase commit

Prerequisites

A running Ingest Engine deployment with the controller, operator, and storage configured
PostgreSQL with wal_level = logical and a user with REPLICATION privilege — see CDC connector prerequisites
An Iceberg catalog (REST, AWS Glue, or Hive Metastore) with a target database/namespace
Object storage for Iceberg data files (S3, GCS, or ADLS)

PostgreSQL configuration

Ensure these settings on the PostgreSQL instance:

-- Check WAL level (must be 'logical')
SHOW wal_level;

-- Grant replication to the CDC user
ALTER ROLE cdc_user REPLICATION;

-- Create a publication for the tables to capture
CREATE PUBLICATION laminar_cdc_publication FOR TABLE orders, customers;

Pipeline Deployment

An Ingest Engine CDC pipeline consists of three resources: a source (where to read changes from), a sink (where to write them), and a pipeline (the SQL query connecting them). These can be deployed in two ways:

Kubernetes Manifest — declare all three resources as YAML and apply with kubectl. The Ingest Engine operator reconciles them and starts the pipeline. This is the recommended approach for production deployments and GitOps workflows.
REST API — create each resource individually via HTTP calls to the Ingest Engine controller API. Useful for scripting, automation, and environments where kubectl access is not available.

Both methods use identical configuration — only the delivery mechanism differs.

Via Kubernetes Manifest

Create a YAML manifest with the following three resources. All three can be placed in a single file separated by --- for convenience.

Step 1: CDC Source

Define a LaminarTable resource for the CDC source. This tells Ingest Engine which PostgreSQL database to connect to, which tables to capture, and how to handle the initial snapshot.

apiVersion: laminar.stream/v1alpha1
kind: LaminarTable
metadata:
  name: pg-cdc-source
  namespace: e6data
spec:
  clusterRef: e6data
  connector: cdc
  config:
    name: pg_cdc_src
    config:
      transport:
        mode: jni
        maxHeap: "512m"
      database:
        type: postgres
        host: <postgres_host>
        port: 5432
        user: cdc_user
        password: <password>
        database: mydb
        schema: public
        tables: [orders, customers]
        snapshotMode: initial
        publicationName: laminar_cdc_publication
      debeziumOverrides: {}
      channelCapacity: 65536
    schema:
      inferred: true
      fields:
        - field_name: _table
          field_type:
            type:
              primitive: String
          nullable: false
        - field_name: _schema_epoch
          field_type:
            type:
              primitive: Int64
          nullable: false
        - field_name: _num_rows
          field_type:
            type:
              primitive: Int64
          nullable: false
        - field_name: _offsets
          field_type:
            type:
              primitive: Bytes
          nullable: true
        - field_name: _payload
          field_type:
            type:
              primitive: Bytes
          nullable: false

Key configuration:

Field	Description
`transport.mode`	`jni` runs the CDC engine in-process. Use `grpc` for out-of-process isolation.
`transport.maxHeap`	JVM heap for the CDC engine. 512m is sufficient for most tables. Increase for wide schemas (200+ columns).
`snapshotMode`	`initial` — full snapshot then streaming. Other modes: `no_data` (stream-only), `when_needed`, `incremental`.
`channelCapacity`	Internal buffer between CDC engine and pipeline. Default 65536 is suitable for most workloads.
`debeziumOverrides`	Pass additional Debezium properties (e.g., `snapshot.max.threads`, `poll.interval.ms`).

No column schema needed for the source tables — column names and types are read from PostgreSQL automatically and carried with each batch of rows.

Step 2: Iceberg Sink

Define a LaminarTable resource for the Iceberg sink. This configures the catalog connection, target namespace, multi-table routing, storage credentials, and output format.

apiVersion: laminar.stream/v1alpha1
kind: LaminarTable
metadata:
  name: pg-cdc-iceberg-sink
  namespace: e6data
spec:
  clusterRef: e6data
  connector: iceberg
  config:
    name: pg_cdc_iceberg_sink
    config:
      catalog:
        provider: direct
        format: iceberg # table format
        type: rest # catalog type: rest, glue, hive, or biglake
        url: <catalog_rest_url>
        warehouse: <warehouse>
      namespace: cdc_db
      tableName: _unused
      appendOnly: true
      multiTable:
        enabled: true
        tableNameTemplate: "{{table}}"
      storageOptions:
        s3.region: us-east-1
        s3.access-key-id: <access_key>
        s3.secret-access-key: <secret_key>
      parquet:
        compression: snappy
    schema:
      format:
        parquet: {}
      fields:
        - field_name: _table
          field_type:
            type:
              primitive: String
          nullable: false
        - field_name: _schema_epoch
          field_type:
            type:
              primitive: Int64
          nullable: false
        - field_name: _num_rows
          field_type:
            type:
              primitive: Int64
          nullable: false
        - field_name: _payload
          field_type:
            type:
              primitive: Bytes
          nullable: false

Polaris catalog — replace the catalog and storageOptions sections with:

catalog:
  provider: direct
  format: iceberg
  type: rest
  url: https://<polaris_host>/api/catalog
  warehouse: <warehouse_name>
  auth: # OAuth2 credentials for Polaris catalog API
    credential: "<client_id>:<client_secret>"
    oauth2_server_uri: https://<polaris_host>/api/catalog/v1/oauth/tokens
    scope: PRINCIPAL_ROLE:ALL
storageOptions:
  client.assume-role.arn: "arn:aws:iam::<account_id>:role/<role_name>"
  client.assume-role.session-name: laminar
  s3.region: <region>

Polaris uses OAuth2 for authentication. The credential is the client ID and secret separated by a colon. If your data is in a different AWS account, use client.assume-role.arn in storageOptions for cross-account S3 access. For production deployments, avoid static credentials in the manifest. Instead, use your cloud platform’s pod identity mechanism to grant storage access:

AWS (EKS): IAM Roles for Service Accounts (IRSA) — the pod assumes an IAM role with S3 permissions
GCP (GKE): Workload Identity — the pod uses a GCP service account with GCS permissions
Azure (AKS): Workload Identity — the pod uses a managed identity with ADLS permissions

When pod identity is configured, the storageOptions credentials can be omitted entirely — the pod inherits storage permissions from its service account. The auth block for catalog authentication is still required. Key configuration:

Field	Description
`namespace`	Iceberg namespace where tables are created
`multiTable.enabled`	Routes each source table to its own Iceberg table
`multiTable.tableNameTemplate`	Template for output table names. `{{table}}` uses the source table name. Use `cdc_{{table}}` to add a prefix.
`tableName`	Required field, but ignored when `multiTable` is enabled. Use any placeholder value.
`appendOnly`	`true` for append-only CDC (inserts only, no delete files). `false` for full CDC with equality deletes.
`storageOptions`	S3/GCS/ADLS credentials. Omit if using IAM roles.

Step 3: Pipeline

Define a LaminarPipeline resource that connects the CDC source to the Iceberg sink using a SQL query. The query forwards the CDC envelope from source to sink, and the sink automatically unwraps and routes each table.

apiVersion: laminar.stream/v1alpha1
kind: LaminarPipeline
metadata:
  name: pg-cdc-to-iceberg
  namespace: e6data
spec:
  clusterRef: e6data
  replicas: 1
  config:
    name: pg_cdc_to_iceberg
    query: |
      INSERT INTO pg_cdc_iceberg_sink
      SELECT _table, _schema_epoch, _num_rows, _payload
      FROM pg_cdc_src
    parallelism: 1
    checkpointIntervalMicros: 10000000

Configuration:

Field	Description
`parallelism`	Number of parallel tasks. Use `1` for CDC pipelines (required for offset consistency).
`checkpointIntervalMicros`	How often to checkpoint. `10000000` = 10 seconds. Lower values reduce latency but increase commit overhead.

Step 4: Apply Manifest

Apply the manifest to your Kubernetes cluster. The Ingest Engine operator will reconcile the resources, create the connection tables via the controller API, and start the pipeline automatically.

kubectl apply -f postgres-cdc-to-iceberg.yaml

Output:

namespace/e6data unchanged
laminartable.laminar.stream/pg-cdc-source created
laminartable.laminar.stream/pg-cdc-iceberg-sink created
laminarpipeline.laminar.stream/pg-cdc-to-iceberg created

Check reconciliation:

kubectl get laminartables,laminarpipelines -n e6data

Output:

NAME                                          CONNECTOR   STATUS
laminartable.laminar.stream/pg-cdc-source     cdc         Ready
laminartable.laminar.stream/pg-cdc-iceberg-sink   iceberg     Ready

NAME                                              STATUS    PARALLELISM
laminarpipeline.laminar.stream/pg-cdc-to-iceberg  Running   1

The config block inside each resource is identical to the API format — only the Kubernetes wrapper (apiVersion, kind, metadata, spec.clusterRef) differs.

Via REST API

Set the API values used by the requests:

export INGESTION_ENGINE_API_URL="https://<ingest-engine-api-host>"
export INGESTION_ENGINE_API_TOKEN="<api-token>"

Step 1: Create the CDC source

Register the PostgreSQL CDC source by posting the connection config to the Ingest Engine API. This creates a connection table that Ingest Engine uses to connect to PostgreSQL and capture changes.

curl -X POST "$INGESTION_ENGINE_API_URL/api/v1/connectors/cdc/tables" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H 'Content-Type: application/json' -d '{
  "name": "pg_cdc_src",
  "config": {
    "transport": {
      "mode": "jni",
      "maxHeap": "512m"
    },
    "database": {
      "type": "postgres",
      "host": "<postgres_host>",
      "port": 5432,
      "user": "cdc_user",
      "password": "<password>",
      "database": "mydb",
      "schema": "public",
      "tables": ["orders", "customers"],
      "snapshotMode": "initial",
      "publicationName": "laminar_cdc_publication"
    },
    "debeziumOverrides": {},
    "channelCapacity": 65536
  },
  "schema": {
    "inferred": true,
    "fields": [
      {"field_name": "_table", "field_type": {"type": {"primitive": "String"}}, "nullable": false},
      {"field_name": "_schema_epoch", "field_type": {"type": {"primitive": "Int64"}}, "nullable": false},
      {"field_name": "_num_rows", "field_type": {"type": {"primitive": "Int64"}}, "nullable": false},
      {"field_name": "_offsets", "field_type": {"type": {"primitive": "Bytes"}}, "nullable": true},
      {"field_name": "_payload", "field_type": {"type": {"primitive": "Bytes"}}, "nullable": false}
    ]
  }
}'

Response:

{
  "id": "ct_cdc_xxx",
  "name": "pg_cdc_src",
  "connector": "cdc",
  "tableType": "source"
}

Step 2: Create the Iceberg sink

Register the Iceberg sink by posting the catalog connection, namespace, multi-table routing, and storage credentials. Ingest Engine will auto-create Iceberg tables in this namespace as CDC data arrives.

curl -X POST "$INGESTION_ENGINE_API_URL/api/v1/connectors/iceberg/tables" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H 'Content-Type: application/json' -d '{
  "name": "pg_cdc_iceberg_sink",
  "config": {
    "catalog": {
      "provider": "direct",
      "format": "iceberg",
      "type": "rest",
      "url": "<catalog_rest_url>",
      "warehouse": "<warehouse_name>"
    },
    "namespace": "cdc_db",
    "tableName": "_unused",
    "appendOnly": true,
    "multiTable": {
      "enabled": true,
      "tableNameTemplate": "{{table}}"
    },
    "storageOptions": {
      "s3.region": "us-east-1",
      "s3.access-key-id": "<access_key>",
      "s3.secret-access-key": "<secret_key>"
    },
    "parquet": {
      "compression": "snappy"
    }
  },
  "schema": {
    "format": {"parquet": {}},
    "fields": [
      {"field_name": "_table", "field_type": {"type": {"primitive": "String"}}, "nullable": false},
      {"field_name": "_schema_epoch", "field_type": {"type": {"primitive": "Int64"}}, "nullable": false},
      {"field_name": "_num_rows", "field_type": {"type": {"primitive": "Int64"}}, "nullable": false},
      {"field_name": "_payload", "field_type": {"type": {"primitive": "Bytes"}}, "nullable": false}
    ]
  }
}'

Response:

{
  "id": "ct_iceberg_xxx",
  "name": "pg_cdc_iceberg_sink",
  "connector": "iceberg",
  "tableType": "sink"
}

For Polaris catalog, replace the catalog and storageOptions sections with the Polaris-specific config shown in the Kubernetes Manifest section above.

Step 3: Create the pipeline

Create the pipeline with a SQL query that connects the CDC source to the Iceberg sink. The pipeline starts automatically once created and begins with a full snapshot of existing data.

curl -X POST "$INGESTION_ENGINE_API_URL/api/v1/pipelines" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H 'Content-Type: application/json' -d '{
  "name": "pg-cdc-to-iceberg",
  "query": "INSERT INTO pg_cdc_iceberg_sink SELECT _table, _schema_epoch, _num_rows, _payload FROM pg_cdc_src",
  "parallelism": 1,
  "checkpointIntervalMicros": 10000000
}'

Response:

{
  "id": "pl_xxx",
  "name": "pg-cdc-to-iceberg",
  "parallelism": 1,
  "actionText": "Running"
}

Watch the snapshot

The pipeline begins with a full snapshot of existing data. In worker logs:

Snapshot - Final stage
Exported 11569 of 11569 records for table 'public.orders' after 00:00:03
Exported 34162 of 34162 records for table 'public.customers' after 00:00:06
Snapshot completed
Starting streaming

A single-threaded snapshot streams at roughly 200–250k rows/s. Once complete, the pipeline transitions to streaming mode and captures live WAL changes.

Verify the output

Check Iceberg tables

Tables are auto-created in the configured namespace. Verify via your catalog:

# REST catalog
curl <catalog_rest_url>/v1/namespaces/cdc_db/tables

Response:

{
  "identifiers": [
    { "namespace": ["cdc_db"], "name": "orders" },
    { "namespace": ["cdc_db"], "name": "customers" }
  ]
}

# AWS Glue
aws glue get-tables --database-name cdc_db --query "TableList[].Name"

Response:

["orders", "customers"]

Query with pyiceberg

from pyiceberg.catalog import load_catalog

catalog = load_catalog("rest", **{
    "type": "rest",
    "uri": "<catalog_rest_url>",
    "warehouse": "<warehouse>",
    "s3.endpoint": "<s3_endpoint>",
    "s3.access-key-id": "<access_key>",
    "s3.secret-access-key": "<secret_key>",
})

tbl = catalog.load_table("cdc_db.orders")
df = tbl.scan().to_pandas()
print(df[["id", "customer_id", "total", "_op", "_ts_ms"]].head())
print(f"{len(df)} rows")

Output:

     id  customer_id  total _op          _ts_ms
   1            3  29.99   r  1720000000000
   2            7  49.99   r  1720000000000
   3            1  12.50   r  1720000000000
   4            5  99.00   r  1720000000000
   5            2  35.00   r  1720000000000
rows

Query with DuckDB

SELECT _op, count(*) FROM iceberg_scan('s3://<bucket>/<prefix>/cdc_db/orders') GROUP BY _op;

Output:

┌──────┬──────────────┐
│ _op  │ count_star() │
├──────┼──────────────┤
│ r    │        11569 │
└──────┴──────────────┘

Test streaming

Make changes in PostgreSQL and verify they land in Iceberg:

-- Insert
INSERT INTO orders (id, customer_id, total) VALUES (1001, 7, 49.99);

-- Update
UPDATE orders SET total = 59.99 WHERE id = 1001;

-- Delete
DELETE FROM orders WHERE id = 1001;

Each change lands in Iceberg within the checkpoint interval (default 10 seconds). Rows carry:

Column	Description
`_op`	`r` (snapshot read), `c` (insert), `u` (update), `d` (delete)
`_ts_ms`	Source event timestamp (when the change occurred at the database)
`_timestamp`	Ingest time (when Ingest Engine processed the event)

Snapshot modes

Mode	Behavior	Use case
`initial` (default)	Full snapshot of existing rows, then stream	First-time setup
`no_data`	Skip snapshot, stream from current WAL position	When historical data is not needed
`when_needed`	Snapshot only if no prior offset exists	Recovery after redeployment
`incremental`	Chunked reads without table locks	Large tables where locking is not acceptable

Set via "snapshotMode": "incremental" in the CDC source config.

Multi-table setup

A single pipeline can capture any number of tables. List the tables to capture in the source config:

"tables": ["orders", "customers", "products", "invoices", "shipments"]

To capture all tables in the database, leave the list empty:

"tables": []

Each table lands in its own Iceberg table in the configured namespace. Table names are derived from the tableNameTemplate:

Template	Source table `orders` becomes
`{{table}}`	`orders`
`cdc_{{table}}`	`cdc_orders`
`{{schema}}_{{table}}`	`public_orders`

Cleanup

Via Kubernetes Manifest

kubectl delete laminarpipeline pg-cdc-to-iceberg -n e6data
kubectl delete laminartable pg-cdc-iceberg-sink -n e6data
kubectl delete laminartable pg-cdc-source -n e6data

Via REST API

# Stop and delete pipeline
curl -X PATCH "$INGESTION_ENGINE_API_URL/api/v1/pipelines/{pipeline_id}" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"stop": "graceful"}'

curl -X DELETE "$INGESTION_ENGINE_API_URL/api/v1/pipelines/{pipeline_id}" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN"

# Delete tables
curl -X DELETE "$INGESTION_ENGINE_API_URL/api/v1/connection_tables/{source_table_id}" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN"
curl -X DELETE "$INGESTION_ENGINE_API_URL/api/v1/connection_tables/{sink_table_id}" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN"

Production considerations

Checkpoint interval: 10 seconds is a good balance. Lower values (1–3s) reduce end-to-end latency but increase Iceberg commit frequency. Higher values (30–60s) reduce commit overhead for high-throughput pipelines.
JVM heap: 512m is sufficient for most workloads. Increase to 1–2g for databases with very wide schemas (200+ columns) or large initial snapshots.

Notes

Source configuration, snapshot modes, and database prerequisites: CDC connector reference.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

PostgreSQL CDC to Iceberg

Prerequisites

PostgreSQL configuration

Pipeline Deployment

Via Kubernetes Manifest

Step 1: CDC Source

Step 2: Iceberg Sink

Step 3: Pipeline

Step 4: Apply Manifest

Via REST API

Step 1: Create the CDC source

Step 2: Create the Iceberg sink

Step 3: Create the pipeline

Watch the snapshot

Verify the output

Check Iceberg tables

Query with pyiceberg

Query with DuckDB

Test streaming

Snapshot modes

Multi-table setup

Cleanup

Via Kubernetes Manifest

Via REST API

Production considerations

Notes

​Prerequisites

​PostgreSQL configuration

​Pipeline Deployment

​Via Kubernetes Manifest

​Step 1: CDC Source

​Step 2: Iceberg Sink

​Step 3: Pipeline

​Step 4: Apply Manifest

​Via REST API

​Step 1: Create the CDC source

​Step 2: Create the Iceberg sink

​Step 3: Create the pipeline

​Watch the snapshot

​Verify the output

​Check Iceberg tables

​Query with pyiceberg

​Query with DuckDB

​Test streaming

​Snapshot modes

​Multi-table setup

​Cleanup

​Via Kubernetes Manifest

​Via REST API

​Production considerations

​Notes

Prerequisites

PostgreSQL configuration

Pipeline Deployment

Via Kubernetes Manifest

Step 1: CDC Source

Step 2: Iceberg Sink

Step 3: Pipeline

Step 4: Apply Manifest

Via REST API

Step 1: Create the CDC source

Step 2: Create the Iceberg sink

Step 3: Create the pipeline

Watch the snapshot

Verify the output

Check Iceberg tables

Query with pyiceberg

Query with DuckDB

Test streaming

Snapshot modes

Multi-table setup

Cleanup

Via Kubernetes Manifest

Via REST API

Production considerations

Notes