Sources

Postgres CDC

Stream a Postgres database's inserts, updates and deletes from a logical replication slot into DuckDB, Snowflake or Parquet files.

A postgres_cdc source captures a Postgres database’s changes. The first sqlload load copies every table under a snapshot. Each later load reads the inserts, updates and deletes committed since the last one from a logical replication slot, applies them, and stops. It loads into DuckDB and Snowflake targets, and into files targets as a changelog.

Use sqlload 0.17.0 or later for postgres_cdc. In 0.16.0, a transaction that changed a table’s columns after writing to it could put values in the wrong columns. Schema changes while streaming load from 0.18.0; in 0.17.0, any change to a table’s columns stops the load until you re-snapshot. From 0.19.0, a table that joins the source is copied on its own, and a first snapshot that stopped part-way resumes; in 0.18.0, both needed --resnapshot. load --follow, which keeps streaming, is in 0.20.0 and later, and files targets in 0.21.0 and later. From 0.22.0, --reload refuses captured tables on every target (see Reloading); before, a warehouse target accepted it but didn’t rebuild the tables.

Requirements

The server runs with wal_level = logical. Changing it needs a restart.
The connection’s user has the REPLICATION attribute and can SELECT the tables.
A publication of exactly the source’s tables; one that also publishes other tables stops the load. When it’s missing, SQLLoad creates sqlload_<source>, which needs a user that owns the tables and has CREATE on the database. Otherwise, create one as the tables’ owner and name it with publication; SQLLoad then only checks it.
Every table has a primary key and a replica identity other than NOTHING.
The target is DuckDB, Snowflake, or a files target with cdc_mode = "changelog" (see Into Parquet files). A Databricks target is refused at plan.

Set up a source

This example streams a shop database into DuckDB, as a user that doesn’t own the tables.

1. Prepare Postgres

Save this as seed.sql. orders has a text[] column, tags:

CREATE TABLE customers (
    customer_id bigint PRIMARY KEY,
    email text NOT NULL,
    country text
);
CREATE TABLE orders (
    order_id bigint PRIMARY KEY,
    customer_id bigint NOT NULL REFERENCES customers,
    status text NOT NULL,
    tags text[],
    total numeric(12,2) NOT NULL
);
INSERT INTO customers
SELECT i, 'customer' || i || '@example.com', (ARRAY['GB', 'FR', 'DE', 'US'])[1 + i % 4]
FROM generate_series(1, 50) AS i;
INSERT INTO orders
SELECT i, 1 + i % 50, (ARRAY['placed', 'paid', 'shipped'])[1 + i % 3], ARRAY['web'],
       round(10 + i * 1.37, 2)
FROM generate_series(1, 1000) AS i;

Start Postgres with logical decoding on, and load the data:

docker run -d --name app-db -e POSTGRES_PASSWORD=secret -p 5432:5432 postgres:17 -c wal_level=logical
docker exec -i app-db psql -q -U postgres < seed.sql

As the tables’ owner, create the loader’s user and a publication of the tables:

CREATE ROLE loader LOGIN REPLICATION PASSWORD 'secret';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO loader;
CREATE PUBLICATION app_changes FOR TABLE customers, orders;

2. Write `sqlload.toml`

default_target = "local"

[connections.app]
kind = "postgres"
host = "localhost"
user = "loader"
password = "${ENV:PGPASSWORD}"
database = "postgres"

[connections.warehouse]
kind = "duckdb"
database = "shop.duckdb"

[targets.local]
connection = "warehouse"
loader_schema = "raw"

[sources.app]
kind = "postgres_cdc"
connection = "app"
description = "The shop's application database, streamed from a replication slot."
include = ["public.*"]
publication = "app_changes"

[sources.app.tables."public.customers"]
description = "Customers, one row per customer."

[sources.app.tables."public.orders"]
description = "Orders, one row per order."
columns = { tags = "text" }

The connection is an ordinary postgres connection. Array and json columns stream as Postgres writes them, so they must be pinned to text, as tags is here. jsonb needs nothing. Without the pin, plan stops:

Errors (1)
└── app__orders
    └── reason  tags: kind = "postgres_cdc" streams text[] values as the source wrote them, which the snapshot normalises as jsonb; pin it with columns = { tags = "text" }, or exclude it

3. Plan, apply and load

export PGPASSWORD=secret

$ sqlload plan
Plan ready  2 selected, 2 objects to create, 2 to load

Objects (2)                                         sqlload apply
├── raw.app__customers                              create table
│   ├── customer_id int    -> BIGINT
│   ├── email       text   -> VARCHAR
│   ├── country     text   -> VARCHAR
│   └── + _sqlload_lsn, _op, _committed_at, _deleted_at, _load_id, _loaded_at
└── raw.app__orders                                 create table
    ├── order_id    int             -> BIGINT
    ├── customer_id int             -> BIGINT
    ├── status      text            -> VARCHAR
    ├── tags        text            -> VARCHAR
    ├── total       decimal(12,2)   -> DECIMAL(12,2)
    └── + _sqlload_lsn, _op, _committed_at, _deleted_at, _load_id, _loaded_at

Loads (2)                                           sqlload load
├── app__customers                                  snapshot copy, then apply changes, soft deletes (key: customer_id)
│   ├── source  postgres_cdc  app  public.customers  slot sqlload_app
│   ├── route   read as Arrow -> DuckDB takes Arrow
│   └── batch   about every 1 GB
└── app__orders                                     snapshot copy, then apply changes, soft deletes (key: order_id)
    ├── source  postgres_cdc  app  public.orders  slot sqlload_app
    ├── route   read as Arrow -> DuckDB takes Arrow
    └── batch   about every 1 GB

✓ Plan complete  2 selected, 2 objects to create, 2 to load

sqlload apply creates the tables. The first sqlload load creates the slot, sqlload_app, and copies every table under its snapshot (the plan is cut here):

$ sqlload load
…
loaded   raw.app__customers  50 rows  0.2s
loaded   raw.app__orders     1,000 rows  0.2s

✓ Completed  2 loaded, 1,050 rows, 0.3s

4. Load the changes

After an insert, an update and a delete in Postgres:

INSERT INTO orders VALUES (1001, 7, 'placed', ARRAY['app'], 25.00);
UPDATE orders SET status = 'shipped' WHERE order_id = 1;
DELETE FROM orders WHERE order_id = 2;

the next load reads just those three changes:

$ sqlload load
…
loaded   raw.app__customers  0 rows  0.2s
loaded   raw.app__orders     3 rows  0.2s

✓ Completed  2 loaded, 3 rows, 0.2s

In raw.app__orders, the deleted order is kept and marked:

`order_id`	`status`	`tags`	`_sqlload_op`	`_sqlload_deleted_at` set
1	shipped	`{web}`	update	no
2	shipped	`{web}`	delete	yes
1001	placed	`{app}`	insert	no

Run sqlload load on a schedule to keep the tables up to date. Each run stops once it has read the changes committed before it started. To load changes as they arrive instead, keep one load running with --follow.

Keep streaming

From sqlload 0.20.0, sqlload load SOURCE --follow keeps streaming after it catches up. It holds one replication session open and writes a batch every batch_interval (default 10s, or 1m into files), at 100,000 changes, or when it stops. A transaction still open on the server stays buffered until it commits. It runs until --max-duration passes or it gets Ctrl-C or SIGTERM:

sqlload load app --follow                    # until Ctrl-C or SIGTERM
sqlload load app --follow --max-duration 10m

It logs each batch it writes, with how far it is behind the server. Here an order is inserted and then updated while it follows, then Ctrl-C stops it:

$ sqlload load app --follow
…
2026-11-07T11:24:53.505015Z  INFO sqlload::cdc::main::follow_cdc: following
2026-11-07T11:25:03.555267Z  INFO sqlload::cdc::_helpers::stream: batch written source="app" changes=2 tables=1 lsn=0/15B3720 behind_bytes=0 behind_secs=6
2026-11-07T11:25:11.518577Z  INFO sqlload::command::_helpers::loader_commands: stopping once the batch or copy under way is written; press Ctrl-C again to stop now
2026-11-07T11:25:11.520357Z  INFO sqlload::cdc::main::follow_cdc: stopped following rows=1
loaded   raw.app__customers  0 rows  18.0s
loaded   raw.app__orders     1 rows  18.0s

✓ Completed  2 loaded, 1 rows, 18.1s

behind_bytes is how far the batch is behind the server’s WAL position, and behind_secs how long ago its newest change committed. The two changes to one order load as one row under merge.

Stopping. --max-duration, or a first Ctrl-C or SIGTERM, writes and confirms what it has read, then exits 0. A snapshot or copy under way finishes first. A second signal stops at once (exit code 130), which is safe: the next load skips what was already written.
While idle it writes nothing and confirms the server’s position, so the slot keeps no WAL for writes to tables it doesn’t publish. It sends Snowflake no queries, so the warehouse can suspend; each batch written resumes it, and a longer batch_interval writes fewer, larger batches.

A column added while it follows is planned again and added to the destination, and the stream carries on from where it was confirmed:

2026-11-07T15:03:20.136978Z  INFO sqlload::cdc::_helpers::stream: re-planning gained=public.customers gained column vip
2026-11-07T15:03:20.160941Z  INFO sqlload::cdc::_helpers::stream: applied line=altered  raw.app__customers  add column  vip bool -> BOOLEAN
2026-11-07T15:03:20.161020Z  INFO sqlload::cdc::main::follow_cdc: following

The same column gained again, with nothing written in between, stops it.

Every other error, including a re-plan that blocks a table, exits 1 without confirming anything unwritten. Run it under a supervisor that restarts it; the restarted load carries on from the last confirmed position.
A table that joins include while it follows is copied at its next re-plan or restart (see Adding tables).
One source per process. It follows one postgres_cdc source and nothing else, so select that source. A selection with other tables or several postgres_cdc sources, plan --follow, apply --follow, --reload, the cursor options, and --max-duration without --follow are refused. --resnapshot applies once, at the start.

Settings

Key	Default	Meaning
`kind`		`postgres_cdc`.
`connection`		A `postgres` connection.
`include`, `exclude`		`schema.table` patterns, as for a Postgres source.
`cdc_mode`	`merge`	`merge` keeps one row per key; `changelog` appends every change.
`deletes`	`soft`	`merge` only: `soft` marks a deleted row with `_sqlload_deleted_at`; `hard` removes it.
`slot`	`sqlload_<source>`	The replication slot.
`publication`	`sqlload_<source>`	The publication. SQLLoad creates `sqlload_<source>` when it’s missing.
`batch_interval`	`10s`, `1m` into files	With `load --follow`: how often it writes what it has read. At least `1s`.

Column settings (columns, exclude_columns, mask and the rest) work as for other sources. A captured table takes no write_strategy, cursor_column, batch_size or state. merge matches rows on the primary key, and a unique_key must lie within it.

Merge and changelog

Every row carries _sqlload_lsn (its transaction’s commit LSN), _sqlload_op (snapshot, insert, update, delete or truncate) and _sqlload_committed_at.

merge keeps one row per key. A soft delete sets _sqlload_deleted_at, and a truncate soft-deletes every older row. A TOASTed value the change left out keeps its current value.
changelog appends every change. _sqlload_seq is its position in the transaction, and _sqlload_unchanged lists the values the change left out.

With cdc_mode = "changelog", two updates to order 5 after the snapshot leave three rows:

`order_id`	`status`	`_sqlload_op`	`_sqlload_seq`
5	shipped	snapshot	0
5	paid	update	1
5	shipped	update	1

Into Parquet files

From sqlload 0.21.0, a postgres_cdc source loads into a files target, local or S3, as a changelog. Set cdc_mode = "changelog"; merge is refused at plan, since Parquet files can’t be updated in place. With the setup above and a local folder:

default_target = "files"

[connections.lake]
kind = "filesystem"
root = "lake"

[targets.files]
connection = "lake"

[sources.app]
kind = "postgres_cdc"
connection = "app"
cdc_mode = "changelog"
# ...the rest as above

$ sqlload plan
…
└── app__orders                                     snapshot copy, then append changes (changelog)
    ├── source  postgres_cdc  app  public.orders  slot sqlload_app
    ├── files   lake/app/orders/  parquet (zstd)
    ├── schema  first files, 12 columns
    ├── batch   about every 1 GB
    └── note    each batch of changes adds new files; a snapshot copy (the first load, a table joining the source, or --resnapshot) writes its files, then deletes every older file in the folder

Every load adds Parquet files to the folder of each table it changed. A snapshot copy (the first load, a table joining the source, or --resnapshot) writes its files, then deletes the folder’s older ones. After the first load and the insert, update and delete from step 4, read the folder with DuckDB:

SELECT order_id, status, _sqlload_op
FROM read_parquet('lake/app/orders/*.parquet', union_by_name = true)
WHERE order_id IN (1, 2, 1001)
ORDER BY order_id, _sqlload_lsn;

`order_id`	`status`	`_sqlload_op`
1	paid	snapshot
1	shipped	update
2	shipped	snapshot
2	NULL	delete
1001	placed	insert

Read with schema merging. Files written after a column was added hold it, and older ones don’t, so read them with union_by_name = true as above. Without it, DuckDB takes the columns of the first file and the new column is missing.
Each change is in exactly one row, across crashes. A replay skips the changes already in a table’s committed files. The files of a load that stopped part-way never count, and are deleted at the table’s next write.
Snapshot records are files under the fixed folders of the source’s destination_path, or else the target’s path, in _sqlload/cdc/<source>/<slot>/. For the default path, {source}/{table}/, that’s lake/app/_sqlload/cdc/app/sqlload_app/. A table’s own destination_path doesn’t move them.
max_footer_reads. A snapshot copy written in more files than the target’s max_footer_reads (default 100, at about a million rows a file) stops the next plan until you raise it.
--resnapshot writes each table’s new snapshot, then deletes its older files, so the history goes, and a reader in between sees both. A table that rejoins the source while its folder still holds files is refused: delete the folder to copy it alone, or run --resnapshot.
Under --follow, batch_interval defaults to 1m, since each batch adds a file to every table it changed. Nothing compacts the files yet.

These are refused at plan:

└── public.customers, public.orders
    └── reason  a files target takes kind = "postgres_cdc" only with cdc_mode = "changelog": Parquet files are append-only, so changes cannot be merged into rows

└── app__customers, app__orders
    └── reason  a postgres_cdc table writes one changelog, not partition folders; remove the cursor placeholder from its path

--reload is refused for a captured table too; --resnapshot copies it again (see Reloading).

What a load guarantees

A batch always holds whole transactions.
The slot confirms a batch only after every table it touched has committed. A change whose _sqlload_lsn a table already holds is skipped. A load that stops part-way is safe to run again: nothing is lost or applied twice.
Each table converges to the source as of the last confirmed position. Tables aren’t promised to be consistent with each other at every moment.
A transaction or batch holding more than about 256 MiB of changes moves them to the system’s temporary folder (TMPDIR; point it at a disk, not a memory-backed tmpfs). Each table is still committed once.

The slot and WAL

Postgres keeps WAL for a slot from the slot’s last confirmed position, so WAL builds up between loads. Every load confirms what it applied, including a load that found no changes in the source’s tables, so Postgres can then free that WAL.

To see how much each slot holds, run this in Postgres:

SELECT slot_name, wal_status,
       pg_size_pretty(pg_current_wal_lsn() - confirmed_flush_lsn) AS not_loaded,
       pg_size_pretty(pg_current_wal_lsn() - restart_lsn) AS wal_kept
FROM pg_replication_slots
WHERE slot_name LIKE 'sqlload_%';

After 100,000 inserts and before the next load:

  slot_name  | wal_status | not_loaded | wal_kept
-------------+------------+------------+----------
 sqlload_app | reserved   | 15 MB      | 15 MB

After the load, not_loaded is 0 bytes. wal_kept shrinks at the first load after Postgres’s next checkpoint.

If loads stop running, the slot keeps WAL until the disk fills. To cap it, set Postgres’s max_slot_wal_keep_size. A slot past the cap is marked lost, and the next load stops:

failed   raw.app__orders     slot sqlload_app lost WAL it had not confirmed, so changes are missing; run `sqlload load --resnapshot` to drop and recreate it and copy every table again

SQLLoad never drops a slot on its own, except under --resnapshot. When you remove a source, drop its slot and publication yourself, or they keep holding WAL:

SELECT pg_drop_replication_slot('sqlload_app');
DROP PUBLICATION app_changes;

Schema changes

From sqlload 0.18.0, changes to a source table’s columns carry through when the stream carries their values. plan and apply add new columns to the destination, as for other sources (see on_schema_change).

Change in Postgres	What loads
A column added without a default	The new column. Rows changed after it was added carry its values.
A column dropped	It stays in the destination with its values; new rows get NULL. A value written just before the drop, in the same load, isn’t loaded for it.
A type `on_schema_change` widens, such as `numeric(12,2)` to `numeric(14,2)`	The widened column, altered in place.

A change from before a column was added leaves that column out. merge keeps the row’s current value; changelog writes NULL and lists the column in _sqlload_unchanged. Here a changelog table loads three transactions, run one after another:

UPDATE customers SET country = 'DE' WHERE customer_id = 3;
ALTER TABLE customers ADD COLUMN vip boolean;
UPDATE customers SET vip = true WHERE customer_id = 3;

`customer_id`	`country`	`vip`	`_sqlload_op`	`_sqlload_unchanged`
3	DE	NULL	update	`,vip,`
3	DE	true	update	NULL

These stop the load, with --resnapshot as the fix, because no change carries the values the table now holds:

a column added with a default, or as an identity or generated column;
a renamed column, or one loaded again after it stopped loading;
a primary key change under merge (a changelog table loads it);
a column dropped and another added while changes to the dropped one are still unloaded, which the stream can’t tell from a rename. Run a load between the two.

failed   raw.app__customers  column tier of app__customers was added with a default (or as an identity or generated column), which fills existing rows without changes to stream; run `sqlload load --resnapshot` to copy every table again, or exclude the column

A column added while a load runs stops that load before anything is written; run it again. Under --follow, it plans again and carries on.

A generated column, or one left out of the publication’s column list, never streams, so a table that loads one is refused. Add it to exclude_columns.

Adding tables

From sqlload 0.19.0, a table that joins a source after its first load, because include now matches it or was widened, is copied on its own at the start of the next load (under --follow, at its next re-plan or restart). The other tables keep streaming, with their history and soft deletes; they aren’t copied again. The copy reads under a temporary slot, so one slot under max_replication_slots must be free when it starts. The table’s changes from before its copy are skipped, and from then on it streams with the others.

Here a refunds table is created, as the tables’ owner:

CREATE TABLE refunds (
    refund_id bigint PRIMARY KEY,
    order_id bigint NOT NULL REFERENCES orders,
    amount numeric(12,2) NOT NULL
);
INSERT INTO refunds SELECT i, i * 10, 5.00 FROM generate_series(1, 20) AS i;
GRANT SELECT ON refunds TO loader;

include = ["public.*"] matches it, and plan shows it as a copy before it streams:

$ sqlload plan
…
└── app__refunds                                    snapshot copy, then apply changes, soft deletes (key: refund_id)
    ├── source  postgres_cdc  app  public.refunds  slot sqlload_app
    ├── route   read as Arrow -> DuckDB takes Arrow
    └── batch   about every 1 GB

SQLLoad adds a new table to its own publication, sqlload_<source>, which needs a user that owns the table. A named publication is yours to change, so until it publishes the table, the load stops:

$ sqlload load
…
failed   raw.app__refunds    publication app_changes does not publish "public"."refunds"; add them with ALTER PUBLICATION "app_changes" ADD TABLE "public"."refunds"

After ALTER PUBLICATION app_changes ADD TABLE refunds; as the owner, and sqlload apply to create the destination table, the next load copies refunds and applies the other tables’ changes, here one update to an order:

$ sqlload load
…
loaded   raw.app__customers  0 rows  0.4s
loaded   raw.app__orders     1 rows  0.4s
loaded   raw.app__refunds    20 rows  0.4s

✓ Completed  3 loaded, 21 rows, 0.4s

A table leaving include still needs --resnapshot. If it joins again later while its destination table still holds rows captured before that, the load stops rather than replace them without asking:

failed   raw.app__refunds    rows captured before slot sqlload_app was recreated are still in app__refunds; copying those tables alone would replace them without asking, so run `sqlload load --resnapshot` to copy every table again (it asks first), or drop the tables from the destination to copy them alone

Drop the destination table and run sqlload apply, and the next load copies it alone. An empty destination table is copied without asking. On a files target, delete the table’s folder instead.

From 0.21.0, a table whose destination is gone, because its table was dropped or its folder deleted, is copied again alone the same way. After DROP TABLE raw.app__orders in DuckDB, plan creates it again and copies it:

$ sqlload plan
…
└── app__orders                                     snapshot copy, then apply changes, soft deletes (key: order_id)

sqlload load creates the table, copies its 1,000 rows, and applies the other tables’ changes.

Re-snapshot

A load stops and names --resnapshot as the fix when the slot is lost or gone, after a schema change the stream can’t carry, or when a table leaves include (see above).

A first snapshot that stopped part-way resumes: the next load copies only the tables it hadn’t finished, and keeps the slot. A --resnapshot that stopped part-way needs --resnapshot again, since its unfinished tables still hold the rows it was replacing.

sqlload load --resnapshot drops and recreates the slot and copies every table again, replacing what it holds. A changelog table loses its history, and a merge table its soft-deleted rows. SQLLoad’s own publication, sqlload_<source>, is set to follow include again. plan --resnapshot shows what it would do:

$ sqlload plan --resnapshot
…
Warnings (2)
├── app__customers
│   └── --resnapshot drops and recreates slot sqlload_app, then copies every table of app again
└── app__orders
    └── --resnapshot drops and recreates slot sqlload_app, then copies every table of app again

✓ Plan complete  2 selected, 2 to load

On a terminal, load --resnapshot asks first. Elsewhere it needs --yes:

$ sqlload load --resnapshot
error: --resnapshot rebuilds app__customers, app__orders from the whole source, replacing what they hold; confirm it in a terminal, or add --yes
$ sqlload load --resnapshot --yes
…
loaded   raw.app__customers  50 rows  0.2s
loaded   raw.app__orders     1,000 rows  0.2s

✓ Completed  2 loaded, 1,050 rows, 0.3s

Reloading

--resnapshot, not --reload, copies a captured table again. From sqlload 0.22.0, --reload refuses captured tables on every target, with or without --resnapshot. In a project that also has other sources, a bare load --reload still rebuilds their tables, but it exits 1. Here shop is an ordinary postgres source next to app:

$ sqlload load --reload --yes
…
Errors (1)
└── app__customers, app__orders
    └── reason  a postgres_cdc table is copied again with --resnapshot, not --reload

✗ Plan failed with errors  3 selected, 1 object to create, 1 to load
loaded   raw.shop__products  30 rows  0.0s

✓ Completed  1 loaded, 30 rows, 0.1s

To reload only the other tables, leave the source out:

$ sqlload load --reload --yes --exclude app
…
loaded   raw.shop__products  30 rows  0.0s

✓ Completed  1 loaded, 30 rows, 0.1s

Refused at plan

Every table of a source shares its slot, so a source loads whole. Selecting some of its tables is refused:

$ sqlload load app__orders
…
Errors (1)
└── app__orders
    └── reason  this selection leaves out public.customers of source app, whose changes share its one slot and would be skipped; select the whole source, or move those tables to another postgres_cdc source

A table without a primary key:

Errors (1)
└── public.events
    └── reason  public.events has no primary key; kind = "postgres_cdc" captures tables with one in this release (publishing a table without a replica identity makes Postgres refuse its updates and deletes); add a primary key, or exclude it

A Databricks target:

Errors (1)
└── public.customers, public.orders
    └── reason  kind = "postgres_cdc" loads into DuckDB and Snowflake targets, and into files targets as a changelog, in this release; not Databricks

A files target with merge, or a {cursor} path: see Into Parquet files.

Tested

The test suite runs postgres_cdc against real Postgres 15, 16 and 17 in containers. Before release, live rounds loaded into DuckDB and into Snowflake (through an S3 external stage), under both merge and changelog, with every table checked against Postgres afterwards. They covered the snapshot, inserts, updates that left a TOASTed value out, deletes, a truncate and reinsert in one transaction, a slot advancing past changes to unpublished tables, a 30,000-change transaction spilled to disk (with the spill limit lowered), and, from 0.18.0, the schema changes above.

Part of SQLStack