Skip to main content

PostgreSQL CDC input connector

Experimental feature

PostgreSQL CDC support is an experimental feature of Feldera.

note

This page describes configuration options specific to the PostgreSQL CDC connector. See top-level connector documentation for general information about configuring input and output connectors.

The PostgreSQL CDC input connector reads changes from a PostgreSQL table using logical replication and streams them into a Feldera table. Unlike the PostgreSQL input connector, which runs a query once, this connector first snapshots the source table and then continues to ingest inserts, updates, and deletes from PostgreSQL's write-ahead log.

The connector uses a PostgreSQL publication that must exist before the pipeline starts. The connecting PostgreSQL user must either be a superuser, or a role prepared according to Running as a non-superuser.

PostgreSQL CDC input connector configuration​

Use transport name postgres_cdc_input.

PropertyTypeDefaultDescription
uri*stringPostgreSQL connection URL, e.g. "postgres://postgres:password@localhost:5432/postgres". The URL must include a username, host, and database name. The user must be a superuser or a role prepared according to Running as a non-superuser.
publication*stringName of an existing PostgreSQL publication. The publication must include source_table; a pipeline whose publication does not carry the table fails to start with an error that lists the tables it does carry.
source_table*stringPostgreSQL table to replicate, schema-qualified, e.g. "public.orders". A name given without a schema refers to a table in public.
run_source_migrationsbooleantrueWhether the connector runs etl's source migrations on startup. The migrations install helper functions and a ddl_command_end event trigger; creating the trigger requires a superuser. Set to false after an administrator has installed the source objects out-of-band (see Running as a non-superuser). This does not skip the state-store migrations, which run on every start.
ssl_ca_pemstringCA certificates in PEM format. Setting this enables TLS and takes precedence over ssl_ca_location.
ssl_ca_locationstringPath to a PEM file containing CA certificates. Used when ssl_ca_pem is not set.
batchobjectBatch processing configuration. See Batch configuration below. When omitted, uses the defaults listed below, which match the etl library's defaults.
memory_backpressureobjectMemory-based backpressure configuration. See Memory backpressure configuration below. When omitted, uses the defaults listed below, which match the etl library's defaults.

[*]: Required fields

The CDC connector does not support client-certificate TLS options (ssl_client_pem, ssl_client_location, ssl_client_key, ssl_client_key_location, or ssl_certificate_chain_location). It also rejects verify_hostname: false: hostname verification is always on, unlike the PostgreSQL source and sink, where it can be disabled. A pipeline that sets any of these fails to start.

Batch configuration​

Controls how replication events are buffered into a batch before being flushed to Feldera.

PropertyTypeDefaultDescription
max_fill_msinteger10000Maximum time, in milliseconds, to wait before flushing a partially filled batch.
memory_budget_ratiofloat0.2Ratio, in (0.0, 1.0], of process memory reserved for incoming replication batch bytes, divided across active streams.
max_bytesinteger33554432Maximum preferred byte size for one batch per active stream (32 MiB). A ceiling, not a target: the smaller of this and the memory-ratio budget applies.

Memory backpressure configuration​

Controls when the connector pauses reading further replication events to avoid exceeding available memory.

PropertyTypeDefaultDescription
activate_thresholdfloat0.85Memory usage ratio, in (0.0, 1.0], above which backpressure is activated.
resume_thresholdfloat0.75Memory usage ratio, in [0.0, 1.0), below which backpressure is released. Must be lower than activate_threshold.

PostgreSQL setup​

The PostgreSQL server must have logical replication enabled:

SHOW wal_level;

The value must be logical. If it is not, configure PostgreSQL with wal_level = logical and restart the server.

The connecting user must be a superuser, or a role prepared according to Running as a non-superuser. A superuser may use logical replication and read every table without further attributes or grants, so the role needs nothing else. The credential is not scoped to the published tables: a superuser can read and write every table in the cluster, run programs on the database server through COPY ... PROGRAM, and load extensions. Give the source tables a database, or a server, of their own when that reach is unwelcome. For example:

CREATE ROLE feldera WITH LOGIN PASSWORD 'password' SUPERUSER;

Create the source table and publication before starting the Feldera pipeline:

CREATE TABLE public.orders (
    id BIGINT PRIMARY KEY,
    customer TEXT NOT NULL,
    amount DECIMAL(10, 2),
    status TEXT NOT NULL
);

ALTER TABLE public.orders REPLICA IDENTITY FULL;

CREATE publication feldera_orders FOR TABLE public.orders;

REPLICA IDENTITY FULL is recommended so update and delete events include the old row values needed to retract records from the Feldera input table.

Objects the connector installs​

The connector embeds the etl replication library, which keeps its state in the source database. Every time the pipeline starts, the connector connects as the configured user and applies the etl migrations that the database does not yet hold. etl divides its migrations into two sets: the state-store migrations create the etl schema and the state tables, and the source migrations create the functions and the event trigger. Of the objects the migrations create, these matter to an operator:

ObjectNamePurpose
SchemaetlHolds the state tables and functions listed in this table.
State tablesetl.replication_state, etl.replication_progress, etl.table_schemas, etl.table_columns, etl.destination_tables_metadataRecord each table's replication phase and column schema, and how far the connector has acknowledged the replication stream.
Migration logetl._sqlx_migrationsRecords which migrations have run, so a later start applies only new ones.
Functionsetl.describe_table_schema, etl.describe_table_identity, etl.emit_schema_change_messagesDescribe a published table's columns and replica identity from the catalog and emit the description as a logical replication message.
Event triggersupabase_etl_ddl_message_triggerFires on ddl_command_end for ALTER TABLE and ALTER PUBLICATION and calls etl.emit_schema_change_messages.

The migrations also create enum types and indexes that only these tables use. The etl repository holds the complete DDL under crates/etl/migrations/source and crates/etl/migrations/postgres_store.

The migrations run forward only. A newer Feldera version migrates the source database on its first start, and an older Feldera version then fails to start a pipeline against that database, because its etl reads the migrated columns with their former types. To run an older version against such a database, drop the etl schema first: that discards the replication state, and every table's initial read runs again.

The event trigger lets the connector learn about schema changes to a published table in order with the row changes around them. It fires for every ALTER TABLE and ALTER PUBLICATION in the database but emits a message only for tables that some publication contains.

PostgreSQL allows only a superuser to create an event trigger, so installing or upgrading the source objects requires a superuser. The connector checks for pending source migrations on every start by default. Already applied migrations are not repeated, but a new migration can require superuser privileges. A start that cannot apply a migration fails. Setting run_source_migrations to false skips the source migrations, so a de-elevated role can connect once an administrator has installed the source objects (see Running as a non-superuser).

The state-store migrations are not gated by run_source_migrations: the connector applies them on every start regardless of the setting. A read-only standby cannot serve as the source: etl skips the source migrations when pg_is_in_recovery() reports a standby, but it runs the state-store migrations on every start, and the standby rejects their writes, so the start fails. The objects stay in the database after the pipeline stops or the user deletes it, and pipelines that read from the same database share them.

Running as a non-superuser​

To run the connector under a role that is not a superuser, install the source objects once as an administrator, then connect as a role with the narrower privileges the connector needs.

First, start the pipeline once as a superuser with the default configuration (run_source_migrations unset, i.e. true), using the same Feldera version that will run the pipeline later so the installed objects match the embedded etl.

Alternatively, install both migration sets out-of-band. Check out the etl repository at the rev pinned for the etl dependency in your Feldera release's root Cargo.toml. From that checkout, with the SQLx CLI installed and an administrator connection URL in ADMIN_DATABASE_URL, run:

psql "$ADMIN_DATABASE_URL" -v ON_ERROR_STOP=1 -c 'CREATE SCHEMA IF NOT EXISTS etl'
PGOPTIONS='-c search_path=etl' sqlx migrate run \
    --database-url "$ADMIN_DATABASE_URL" --ignore-missing \
    --source crates/etl/migrations/postgres_store
PGOPTIONS='-c search_path=etl' sqlx migrate run \
    --database-url "$ADMIN_DATABASE_URL" --ignore-missing \
    --source crates/etl/migrations/source

Apply the state-store migrations first, followed by the source migrations. Both sets share etl._sqlx_migrations: the search_path option places the log in the correct schema, and --ignore-missing lets each set coexist with the other's history. SQLx applies pending up migrations in version order and records their checksums. Do not apply the SQL files alone without recording migration history; subsequent starts would attempt to apply them again.

Then stop the bootstrap pipeline and grant the runtime role only what it needs:

-- The role is not a superuser, but may use logical replication.
CREATE ROLE feldera_runtime WITH LOGIN PASSWORD 'password' REPLICATION NOSUPERUSER;

-- Read the published tables.
GRANT CONNECT ON DATABASE postgres TO feldera_runtime;
GRANT USAGE ON SCHEMA public TO feldera_runtime;
GRANT SELECT ON public.orders TO feldera_runtime;

-- etl runs `CREATE SCHEMA IF NOT EXISTS etl` on every start, so the runtime
-- role needs CREATE on the database even after the schema exists.
GRANT CREATE ON DATABASE postgres TO feldera_runtime;

-- SQLx runs CREATE TABLE IF NOT EXISTS for its migration log on every start.
-- PostgreSQL checks CREATE on the schema even when the table already exists.
GRANT USAGE, CREATE ON SCHEMA etl TO feldera_runtime;
GRANT SELECT ON etl._sqlx_migrations TO feldera_runtime;

-- Read and update etl's state tables (created by the bootstrap start).
GRANT SELECT, INSERT, UPDATE, DELETE ON
    etl.replication_state, etl.replication_progress,
    etl.table_schemas, etl.table_columns, etl.destination_tables_metadata
TO feldera_runtime;

-- etl's source migrations revoke public access to the helper functions it
-- installs; grant the runtime role explicit execute.
GRANT EXECUTE ON FUNCTION etl.describe_table_schema(pg_catalog.oid) TO feldera_runtime;
GRANT EXECUTE ON FUNCTION etl.describe_table_identity(pg_catalog.oid) TO feldera_runtime;

Replace postgres, public, and public.orders with your database, schema, and published tables. etl reads every table in the publication, so grant USAGE and SELECT for each of them even when this connector selects just one source_table. The identity columns in the current etl state tables do not require separate sequence grants. Keep the etl objects, especially the event trigger and its security-definer function, owned by the administrator. The runtime role needs no ownership or write privileges on the published tables.

The recipe requires database and etl-schema creation privileges as described above; it is not a read-only database account. Its etl grants cover state shared by pipelines using that database. PostgreSQL's REPLICATION attribute also permits creating replication slots and is not an access boundary limited to the configured publication.

Finally, run the pipeline with run_source_migrations set to false:

{
    "transport": {
        "name": "postgres_cdc_input",
        "config": {
            "uri": "postgres://feldera_runtime:password@localhost:5432/postgres",
            "publication": "feldera_orders",
            "source_table": "public.orders",
            "run_source_migrations": false
        }
    }
}

Disabling the source migrations does not remove the connector's other requirements: the helper functions, the event trigger, the writable etl state tables, and the state-store migration checks must all still be present and reachable. The connector refuses to start when the event trigger supabase_etl_ddl_message_trigger is missing or disabled, because without it schema changes to the published tables go unnoticed; this check runs with either setting. Before starting an upgraded connector with run_source_migrations set to false, an administrator must apply any new source migrations the upgrade introduces; pending state-store migrations need the same attention when the runtime role lacks the DDL or ownership privileges to apply them. Recheck the grants after any migration that recreates functions or adds tables.

Schema requirements​

Feldera matches columns by name.

  • Every non-nullable Feldera column must exist in the PostgreSQL source table.
  • Nullable Feldera columns may be absent from the PostgreSQL source table.
  • Extra PostgreSQL columns that do not exist in Feldera are ignored.
  • If a required Feldera column is removed from PostgreSQL while the connector is running, the connector reports a fatal error.

Example​

First, create a PostgreSQL table and publication:

CREATE TABLE public.orders (
    id BIGINT PRIMARY KEY,
    customer TEXT NOT NULL,
    amount DECIMAL(10, 2),
    status TEXT NOT NULL
);

ALTER TABLE public.orders replica identity FULL;

CREATE publication feldera_orders FOR TABLE public.orders;

INSERT INTO public.orders VALUES
    (1, 'Alice', 25.00, 'new'),
    (2, 'Bob', 40.50, 'new');

Then create a Feldera table that reads from the PostgreSQL publication:

CREATE TABLE orders (
    id BIGINT NOT NULL,
    customer TEXT NOT NULL,
    amount DECIMAL(10, 2),
    status TEXT NOT NULL
) WITH (
    'materialized' = 'true',
    'connectors' = '[{
        "transport": {
            "name": "postgres_cdc_input",
            "config": {
                "uri": "postgres://feldera:password@localhost:5432/postgres",
                "publication": "feldera_orders",
                "source_table": "public.orders"
            }
        }
    }]'
);

When the pipeline starts, Feldera ingests the existing rows in public.orders. Subsequent PostgreSQL changes are streamed into the Feldera table:

INSERT INTO public.orders VALUES (3, 'Carol', 19.99, 'new');
UPDATE public.orders SET status = 'shipped' WHERE id = 1;
DELETE FROM public.orders WHERE id = 2;

TLS example​

To connect over TLS, provide the trusted root certificate:

CREATE TABLE orders (
    id BIGINT NOT NULL,
    customer TEXT NOT NULL,
    amount DECIMAL(10, 2),
    status TEXT NOT NULL
) WITH (
    'materialized' = 'true',
    'connectors' = '[{
        "transport": {
            "name": "postgres_cdc_input",
            "config": {
                "uri": "postgres://feldera:password@db.example.com:5432/postgres",
                "publication": "feldera_orders",
                "source_table": "public.orders",
                "ssl_ca_pem": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
            }
        }
    }]'
);

Memory vs. throughput tuning example​

For a high-throughput table, you may want larger batches (fewer, bigger flushes) and a memory backpressure window tuned for the worker's available memory:

CREATE TABLE orders (
    id BIGINT NOT NULL,
    customer TEXT NOT NULL,
    amount DECIMAL(10, 2),
    status TEXT NOT NULL
) WITH (
    'materialized' = 'true',
    'connectors' = '[{
        "transport": {
            "name": "postgres_cdc_input",
            "config": {
                "uri": "postgres://feldera:password@localhost:5432/postgres",
                "publication": "feldera_orders",
                "source_table": "public.orders",
                "batch": {
                    "max_fill_ms": 2000,
                    "memory_budget_ratio": 0.3,
                    "max_bytes": 16777216
                },
                "memory_backpressure": {
                    "activate_threshold": 0.9,
                    "resume_threshold": 0.7
                }
            }
        }
    }]'
);

Resume behavior​

The connector stores replication state in PostgreSQL and uses logical replication slots managed by the connector. Restarting a pipeline with the same database host, port, database, publication, and source table resumes from the existing replication state. Changing any of those values creates a different replication identity and can cause a new snapshot.

Rotating the PostgreSQL username or password does not change the replication identity.

With fault tolerance enabled, the connector delivers every change at least once: the replication slot advances only past changes that a Feldera checkpoint contains. The connector reads changes in batches, and a batch big enough to reach the pipeline in more than one piece counts as contained only once its last change is in the checkpoint, so a checkpoint taken while such a batch is still arriving leaves the slot where it is.

What the slot follows depends on the fault tolerance setting. With fault tolerance enabled, a batch counts as contained once a checkpoint covers the step that holds it, and a checkpoint waits for the circuit and for the input connectors alone, so an output connector that runs behind does not hold the slot back. Without fault tolerance, the connector waits instead for the step that holds the batch to complete, and a step is complete only when every output connector has finished it, so an output that runs behind does hold the slot back with it.

PostgreSQL retains the write-ahead log covering everything the slot has not passed, and the connector reads the next batch of changes into Feldera only after the previous one is acknowledged, so a slot that stays put shows up as both disk held on the source and a slower read from it. Sizing a source's log retention means allowing for the checkpoint interval with fault tolerance enabled, and for the slowest sink without it.

The initial read of a table is all or nothing. A checkpoint holding only part of it would be unusable, because PostgreSQL streams changes from the replication slot once the read is complete and cannot supply the missing rows later. While a read is in progress, and for a moment after it until the connector has seen its replication state record the read, the connector therefore blocks checkpoints and suspend requests, and the pipeline reports the connector as the reason. Reading a large table can hold checkpoints back for as long as the read takes, and pausing the pipeline during the initial read extends that block until the pipeline runs again. A suspend, or a stop that takes a checkpoint first, waits the same way, so stopping a pipeline during its first read of a large table may need a forced stop; the next start reads the table again, so nothing is lost. When the publication holds several tables, the changes the connector has already read stay unacknowledged for the same period, and PostgreSQL retains the write-ahead log that covers them.

A pipeline that stops or crashes before its initial read of a table finishes reads that table again on the next start, with or without fault tolerance. The connector's replication state records the read as complete only after the last row of the read has reached the circuit, so an interrupted read leaves the table marked incomplete and the next start reads it again from a fresh snapshot. The connector then delivers the rows it read before the stop a second time; a primary key on the Feldera table makes the repeated rows replace the originals instead of appearing twice.

Once the initial read has finished, what a restart does with the rows of that read depends on how the pipeline starts:

StartReads the table againRows from the initial read
Resume from a checkpoint or from a suspend, with or without fault toleranceNoKept: the connector blocks checkpoints and suspend requests until the read is complete, so the checkpoint holds every row.
Start with no checkpoint to resume from, fault tolerance enabledYesRead again from a fresh snapshot.
Start with no checkpoint to resume from, fault tolerance disabledNoAbsent from Feldera; only changes streamed from the slot arrive.

The first row has an exception: the pipeline discards the connector's checkpointed state when a program change requires a backfill of the Feldera table the connector feeds, or when the connector's configuration changed. The connector then starts as in the two rows for a start with no checkpoint to resume from. The first row has a second exception: when the connector's replication state shows that it finished a read of the table after the checkpoint was taken, for example a read it repeated after an error, the connector reads the table again, because neither the checkpoint nor the changes replayed from the replication slot hold that read's rows.

With fault tolerance enabled, a start that has no checkpoint to resume from, for example after a forced stop or a crash before the pipeline has taken a checkpoint, begins with an empty circuit. The connector detects that no checkpoint holds the initial read, logs a warning to that effect, and reads the table again before resuming from the replication slot, so the Feldera table is complete. The pipeline delivers rows that reached output connectors before the stop a second time.

Without fault tolerance, a start that has no checkpoint to resume from resumes from the replication slot alone: the connector does not read the table again, so the rows the table held when the connector first read it are absent from Feldera, although the changes PostgreSQL streams from the slot still arrive. Whatever the fault tolerance setting, a pipeline with storage configured resumes from the latest checkpoint in its storage, so a stop that takes a checkpoint and a suspend keep those rows. A crash or a forced stop keeps them only with fault tolerance. Without it the connector acknowledges changes to PostgreSQL as the pipeline processes them, so a resume after a crash replays neither the changes nor a repeated read of the table that happened after the checkpoint. A pipeline without storage starts fresh every time, and only Feldera Enterprise Edition takes checkpoints (see fault tolerance). Without fault tolerance the pipeline takes a checkpoint only when the user requests one or suspends the pipeline. With fault tolerance the pipeline also takes checkpoints on its own, by default every 60 seconds; the checkpoint_interval_secs setting described on the fault tolerance page changes the interval, and null disables automatic checkpoints.

How often the pipeline checkpoints is therefore how often the replication slot advances with fault tolerance enabled. With automatic checkpoints off, the slot moves only when someone requests a checkpoint or suspends the pipeline, and PostgreSQL keeps every write-ahead log segment since the last one. A pipeline left running that way holds the source's disk, so either keep checkpoints automatic or make requesting one part of how the pipeline is operated.