Articles in this section

Listen for real-time changes from PostgreSQL

Saikumar Rachamalla
Updated

Overview

Use Change Data Capture (CDC) to stream inserts, updates, and deletes from PostgreSQL tables into your flows in near real-time. CDC reads from PostgreSQL’s write-ahead log (WAL) via Debezium, so changes arrive within seconds rather than waiting for a scheduled run.

Creating a PostgreSQL CDC listener provides the following benefits:

  • Reduces database load by reading the WAL instead of running heavy polling queries on source tables.
  • Streams only changes (inserts, updates, deletes), cutting data transfer and downstream compute.
  • Provides clean operation semantics — after values, before values, and operation type (insert/update/delete) — reducing downstream conflicts and rework.
  • Works with Slowly Changing Dimension (SCD) Type 2 history tables without comparing entire tables, and can capture before-and-after row states for each change.
  • Compatible with existing Celigo transformations, output filters, and preSavePage scripts.

Note: PostgreSQL CDC uses Debezium’s pgoutput plugin and automatically creates a replication slot and publication on your PostgreSQL server if you have not provided one. Ensure the connecting user has REPLICATION privileges, and sufficient WAL retention is configured.

Prerequisites

Before you begin:

  • PostgreSQL CDC listener requires a Professional or Enterprise edition. See Celigo platform editions.
  • You need an existing PostgreSQL connection, or you can create one during setup. See Set up a connection to PostgreSQL.
  • Your PostgreSQL server must have wal_level set to logical. Contact your database administrator to verify or update this setting.

Create a PostgreSQL listener

Follow these steps to configure a listener that captures change data from PostgreSQL and connects it to your flow.

  1. Go to Build > Flows > Create flow.
  2. In Flow builder, select Add source, then configure the details in the new flow step.

Configure step and connection details

  1. In Application, search and select the database connector PostgreSQL from the list.
  2. In Step type, select Listen for real-time data from source application.
  3. In Name your step, enter a unique name. You can reuse this listener across flows, so a specific name makes it easier to identify in a list.
  4. Optional: In Describe your step, enter a clear and concise description of the flow step to help anyone using this listener in their flows understand its purpose. Keep this description updated as you make changes.

After configuring the step, a listener configuration panel opens. Complete the fields below.

Configure listener

Configure the scope and filters that determine which PostgreSQL changes your listener captures.

  1. In Connection, select a PostgreSQL connection from the list. If you don’t have a connection, select Create connection at the end of the list and set up a connection to PostgreSQL.
  2. Optional (only when you provide publication): In Tables, select or enter a comma-separated list of table names to listen to. Select the refresh icon next to the field to load available tables from your database. If you leave this blank, events from all non-system tables are captured. To exclude specific tables instead, use table.exclude.list in Additional properties.

  3. In Additional properties, enter key-value pairs that control which tables, schemas, and columns are captured. Use the dropdown to select a property key and enter its value. Based on your requirements, you can set optional properties such as:

    • table.exclude.list: Comma-separated list of fully-qualified table names (schema.table) to exclude from capture.
    • schema.include.list: Comma-separated list of schemas to include. When set, only tables in these schemas are captured.
    • schema.exclude.list: Comma-separated list of schemas to exclude from capture.
    • column.include.list: Comma-separated list of fully-qualified column names to include in the change event payload.
    • column.exclude.list: Comma-separated list of fully-qualified column names to exclude from the change event payload.
    • snapshot.mode: Controls how the connector behaves on startup. Accepts no_data (default, captures only new changes after the connector starts) or when_needed (re-snapshots if the replication slot offset is no longer available). You can also enter other Debezium-supported values directly.

    Tip: Default: snapshot.mode defaults to no_data. Change this only if you need to backfill existing rows. See Debezium’s PostgreSQL connector documentation for the full list of supported properties.

  4. In Fields to include, select which parts of the CDC change event to include in the listener payload. By default, only after is selected. Choose only the fields you need to reduce payload size and simplify downstream processing.
    Available fields:
    • after: The full row state after the change (insert or update). Default selection.
    • before: The full row state before the change (update or delete). Requires REPLICA IDENTITY FULL on the source table.
    • op: Operation type: c (create/insert), u (update), d (delete), r (read/snapshot).
    • source: Metadata about the source event, including database name, table name, schema, and LSN position.
    • schema: The full Debezium schema descriptor for the event.
    • tableChanges: The updated field values from the change event (equivalent to payload.updateDescription.updatedFields).
    • transaction — Transaction metadata, including the transaction ID and event order within the transaction.
    • ts_ms: Timestamp of the change event in milliseconds.
    • ts_us: Timestamp in microseconds.
    • ts_ns: Timestamp in nanoseconds.
  5. Cursor reset setting: In Cursor management, reset the export's stored CDC cursor (offset). The section is displayed only when a cursor exists and is typically available after several minutes of active CDC streaming. To reset the cursor, expand Cursor management, click Reset, and then save the export. When the export is saved, the connector clears the stored cursor and restarts CDC processing. After restart, behavior depends on the configured snapshot.mode value. For more information, see FAQs – PostgreSQL listener for change data capture (CDC).

Advanced settings

Configure these settings only if your scenario requires them.

  1. In Max wait time, set the maximum time (in seconds) for the listener to wait before sending a batch of events. Typically, events are sent when the page size limit is reached. If the events are fewer, they are sent after the wait time. The default is 300 seconds. The minimum wait time is 30 seconds.
  2. In Page size, specify how many records you want in each page of data. The Celigo platform splits exported data into pages of records. The default is 250 records per page; pages are capped automatically at 5 MB. The application you are importing data into will most often be the bottleneck on page size.
  3. In Data URI template, when your flow runs but has data errors, use this field to ensure that all the errors in your job dashboard have a link to the original data from the export application. This field uses a handlebars template to generate the dynamic links based on the data being exported.
  4. Turn on Do not store retry data if you don’t want the Celigo platform to store retry data for records that fail in your flow. Storing retry data can slow down your flow’s overall performance if you are processing very large numbers of failing records.
  5. In Override trace key template, define a trace key that the Celigo platform will use to identify a unique record. Any value you provide overrides the default trace key for your app. You can specify a single field, such as {{record.field1}}, or use a handlebars expression. For example, {{join "_" record.field1 record.field2}} generates a trace key such as 123_456. If you have applied a transformation to exported data, reference its fields in the trace key template without the path record. — for example, {{field1}}.

After saving your listener, it appears in the flow builder. Connect it to an import step to begin capturing changes from PostgreSQL.