Connect

PostgreSQL CDC Patterns

The postgres_cdc input captures row-level changes from PostgreSQL tables using logical replication and the Write-Ahead Log (WAL). Use these patterns to filter, transform, and route PostgreSQL CDC events to Redpanda, S3, and other destinations.

Use this cookbook to:

Apply reusable patterns for capturing PostgreSQL CDC events
Adapt integration patterns to route CDC data to Redpanda and S3
Identify patterns for filtering and transforming change events
Backfill and re-snapshot tables with incremental snapshots

Prerequisites

Before using these patterns, configure the following.

Redpanda CLI

Install the Redpanda CLI (rpk) to run Redpanda Connect. See rpk installation for installation instructions.

PostgreSQL logical replication

The source PostgreSQL database must have logical replication enabled. Verify the setting by running:

SHOW wal_level;

The wal_level value must be logical. For instructions on enabling logical replication, see the postgres_cdc prerequisites. For cloud-managed PostgreSQL, see:

Replication slot

The postgres_cdc input uses a PostgreSQL replication slot to track its position in the WAL. The slot is created automatically when the pipeline starts, using the name specified in the slot_name field. Each pipeline must use a unique slot name. Replication slots persist in the database, so the pipeline resumes from where it left off after a restart.

Environment variables

The examples in this cookbook use environment variables for configuration:

export PG_DSN="postgres://user:password@localhost:5432/mydb" (1)
export PG_SCHEMA="public" (2)
export PG_SLOT_NAME="my_connect_slot" (3)
export REDPANDA_BROKERS=localhost:9092 (4)
export S3_BUCKET=cdc-archive (5)

1	The PostgreSQL DSN in `postgres://user:password@host:port/dbname` format.
2	The schema containing the tables to capture.
3	A unique name for the replication slot.
4	The Redpanda broker addresses (for Redpanda output examples).
5	The S3 bucket name (for S3 output examples).

Capture CDC events

The simplest pattern captures all change events from PostgreSQL tables and outputs them with metadata:

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root.operation = meta("operation")
        root.table = meta("table")
        root.lsn = meta("lsn")
        root.data = this
        root.timestamp = now()

output:
  stdout:
    codec: lines

For details on the CDC event message structure and available metadata fields, see the metadata section in the connector reference.

Filter CDC events

Filter events to process only specific change types using the operation metadata field:

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: false

pipeline:
  processors:
    - mapping: |
        root = if meta("operation") != "insert" && meta("operation") != "update" {
          deleted()
        }
    - mapping: |
        root.operation = meta("operation")
        root.table = meta("table")
        root.data = this
        root.timestamp = now()

output:
  stdout:
    codec: lines

This pattern:

Filters to only insert and update operations, dropping delete and read events
Transforms the event to a simplified format with a timestamp

Route to Redpanda

Stream PostgreSQL changes to Redpanda for real-time processing:

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
      - customers
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        meta topic = meta("table")

output:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topic: ${! meta("topic") }
    key: ${! json("id") }
    batching:
      count: 100
      period: 1s

This pattern:

Uses the table name as the Redpanda topic
Batches messages for efficient delivery
Sets the message key to the row’s primary key field

Route to S3

Archive CDC events to S3 for long-term storage and analytics:

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root.operation = meta("operation")
        root.table = meta("table")
        root.data = this
        root.timestamp = now()

output:
  aws_s3:
    bucket: ${S3_BUCKET}
    path: >-
      cdc/${! meta("table") }/${! timestamp_unix().format_timestamp("2006/01/02/15") }/${! uuid_v4() }.ndjson
    batching:
      count: 1000
      period: 5m
      processors:
        - archive:
            format: lines

This pattern:

Organizes files by table and time-based partitions (year/month/day/hour)
Batches events and archives them as newline-delimited JSON
Uses UUID file names to prevent collisions

Route by event type

Route different event types to different destinations:

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

output:
  switch:
    cases:
      - check: meta("operation") == "insert"
        output:
          redpanda:
            seed_brokers:
              - ${REDPANDA_BROKERS}
            topic: postgres.orders.inserts
      - output:
          redpanda:
            seed_brokers:
              - ${REDPANDA_BROKERS}
            topic: postgres.orders.changes

This pattern:

Routes insert events to one Redpanda topic and all other change events to another
Supports specialized downstream consumers per operation type

Configure replication mode

The postgres_cdc input supports two replication modes controlled by the stream_snapshot field:

stream_snapshot: true: Captures a full snapshot of existing table data before streaming live changes. Use this when you need a complete initial load.
stream_snapshot: false: Skips the snapshot and streams only changes from the current WAL position. Use this when you only need new changes going forward.

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true
    snapshot_batch_size: 1000 (1)

1	Number of rows to read per batch during snapshot processing.

To load or reload existing table data without delaying live changes, use an incremental snapshot instead. See Backfill tables with incremental snapshots.

Backfill tables with incremental snapshots

An incremental snapshot reads a table’s existing rows in chunks while the postgres_cdc input keeps streaming live changes. You request a snapshot by inserting a row into a signal table, so you can backfill or re-snapshot a table at any time without restarting the pipeline, dropping the replication slot, or changing the configuration.

Incremental snapshots require Redpanda Connect 4.112.0 or later.

Requirements

Before you enable incremental snapshots, check that your configuration and tables meet these requirements. The input fails to start if any of the configuration requirements are not met.

Set stream_snapshot to false. Incremental snapshots and stream_snapshot read the same rows, so enabling both would deliver every row twice.
Set signal_table_name to the name of a signal table. You request every incremental snapshot through this table.
Set incremental_snapshot.checkpoint_cache to a cache resource. The input stores snapshot progress in this cache, so a restart resumes the snapshot instead of starting over. Use a cache that persists across restarts, such as redis. With a memory cache, a restart discards an in-progress snapshot: the input does not resume or restart it, so you must insert a new signal.
Keep heartbeat_interval set to a non-zero value. The snapshot advances only when the input receives a transaction from the replication stream, and on a quiet table, heartbeats are the only transactions.

Each table that you snapshot must also meet these requirements. The input rejects and logs a signal for a table that does not meet them, and continues streaming live changes.

The table must appear in tables, or tables must be empty.
The table must have a primary key. The input supports composite keys. The following primary key types are not supported: bytea, interval, bit, bit varying, and any range or multirange type.
If the table is partitioned, its publication must set publish_via_partition_root = true. The input creates its publication as pglog_stream_<slot_name> without this option, so set it after the input has started for the first time. For example, if slot_name is my_connect_slot, run ALTER PUBLICATION pglog_stream_my_connect_slot SET (publish_via_partition_root = true);.

Create the signal table

Create the signal table in the schema that you set in the schema field. The SQL statements on this page use public. If your schema is different, replace public in each statement:

CREATE TABLE public.rpcn_signal_table (
    id   SERIAL PRIMARY KEY,
    type VARCHAR(32),
    data TEXT
);

The table must have the id, type, and data columns. At startup, the input checks that these columns exist, but not their types, so a wrong type (for example, data JSONB instead of TEXT) causes an error only when the input reads the first signal row.

Do not add the signal table to tables. The input adds the signal table to its publication automatically and fails to start if the table is also listed in tables.

Configure the input

This configuration enables incremental snapshots for the orders and customers tables:

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
      - customers
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: false (1)
    signal_table_name: rpcn_signal_table (2)
    incremental_snapshot:
      enabled: true
      chunk_size: 1024 (3)
      checkpoint_cache: snapshot_progress (4)

pipeline:
  processors: (5)
    - mapping: |
        root = if @table == "rpcn_signal_table" { deleted() } else { this }

output:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topic: ${! meta("table") }
    key: ${! json("id") }

cache_resources:
  - label: snapshot_progress
    redis:
      url: ${REDIS_URL}

1	Required. Incremental snapshots and `stream_snapshot` are mutually exclusive.
2	The signal table, without the schema name.
3	The number of rows to read in each chunk. The default is `1024`.
4	The label of the cache resource that stores snapshot progress.
5	The input publishes signal rows downstream as regular `insert` messages. This mapping drops them by filtering on the `table` metadata field.

Set REDIS_URL to the address of your Redis server.

For all incremental snapshot settings, see the incremental_snapshot field reference.

Trigger a table snapshot

To snapshot one or more tables, insert a snapshot-execute signal into the signal table. The data column is a JSON object with a tables key that lists the table names, without the schema name:

INSERT INTO public.rpcn_signal_table (type, data)
VALUES ('snapshot-execute', '{"tables": ["orders", "customers"]}');

The input reads signals from the replication stream, so insert the signal after the input has created its replication slot. Check that the slot exists:

SELECT slot_name, active FROM pg_replication_slots WHERE slot_name = 'my_connect_slot';

Replace my_connect_slot with your slot_name value. A signal inserted before the slot exists is not in the replication stream, and the input never reads it.

Each table joins the end of the snapshot queue. The input emits the snapshot rows as messages with the operation metadata field set to read, alongside the live insert, update, and delete messages.

Re-snapshot a table

The input records each table that it has snapshotted in the checkpoint cache. If you signal a table that is already recorded, the input skips the table and logs a message, so a repeated signal does not read the table again.

To snapshot a table again, set incremental_snapshot.checkpoint_cache_key to a key that has not been used before, restart the pipeline, and insert a new snapshot-execute signal. Changing the key discards the record of which tables have been snapshotted.

If several pipelines share the same cache, give each one a different checkpoint_cache_key.

Handle duplicate rows

An incremental snapshot can deliver the same row twice, once from the replication stream and once from the snapshot. For example, this happens when a row is inserted after replication starts but before the snapshot reaches its table. Write rows to your destination as idempotent upserts keyed by primary key, so a duplicate overwrites the earlier copy instead of creating a second record.

Snapshot tables with large column values

If a table has large (TOASTed) column values, set REPLICA IDENTITY FULL on the table before you snapshot it:

ALTER TABLE public.orders REPLICA IDENTITY FULL;

Under the default replica identity, PostgreSQL omits an unchanged TOAST value from an UPDATE message, and the input emits the unchanged_toast_value placeholder for that column. If a row is updated while its snapshot chunk is buffered, the input drops the snapshot copy as a duplicate, so the placeholder can be the only value your destination receives for that column. With REPLICA IDENTITY FULL, PostgreSQL sends the previous row, and the input reads the real value from it. You can revert the replica identity after the snapshot completes.

Also set unchanged_toast_value to a distinctive value instead of the null default, so downstream consumers can tell a placeholder apart from a column that is null.

Troubleshoot common issues

Use these steps to diagnose and fix the most common problems with the postgres_cdc input.

No events received

If no events arrive:

Verify logical replication is enabled:
```
SHOW wal_level;
```
Grant the required PostgreSQL privileges:
```
ALTER ROLE your_user WITH REPLICATION;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO your_user;
```
ALTER ROLE … WITH REPLICATION requires superuser privileges. Run this statement as a superuser, or have a database administrator run it on your behalf.
Check that slot_name is unique and not already in use:
```
SELECT slot_name, active FROM pg_replication_slots;
```

Snapshot fails mid-run

If the snapshot fails, the replication slot has an incomplete record of your database. To maintain data integrity, drop the replication slot and restart the pipeline:

SELECT pg_drop_replication_slot('my_connect_slot');

Replace my_connect_slot with your slot_name value.

Duplicate events

The postgres_cdc input provides at-least-once delivery. If the pipeline fails between WAL acknowledgments, events may be re-read on restart. To handle duplicates:

Use idempotent processing in downstream systems.
Deduplicate using the lsn metadata field.
Lower checkpoint_limit to reduce the window of possible duplicates.

Next steps

Suggested labs

Stream Jira Issues to Redpanda for Real-Time Metrics

Search all labs

Was this helpful?

group Ask in the community

mail Share your feedback

What do you think of this page?

Let us know more:

Let us contact you about your feedback:

PostgreSQL CDC Patterns

Prerequisites

Redpanda CLI

PostgreSQL logical replication

Replication slot

Environment variables

Capture CDC events

Filter CDC events

Route to Redpanda

Route to S3

Route by event type

Configure replication mode

Backfill tables with incremental snapshots

Requirements

Create the signal table

Configure the input

Trigger a table snapshot

Re-snapshot a table

Handle duplicate rows

Snapshot tables with large column values

Troubleshoot common issues

No events received

Snapshot fails mid-run

Duplicate events

Next steps

Suggested labs