Sources

Postgres CDC

The postgres_cdc source: settings, Databricks and Parquet targets, what a load guarantees, the slot and WAL, schema changes, and what plan refuses.

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, Snowflake and Databricks 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. Databricks targets are in 0.23.0 and later. From 0.24.0, a load into a files target names each table by its folder.

To set one up, and for what the server, user and tables need, follow Capture Postgres changes.

Settings

A postgres_cdc source takes a postgres connection, include and exclude patterns, and cdc_mode, deletes, slot, publication and batch_interval. Each is in sqlload.toml, with its default. Column settings work as for other sources. A captured table takes no write_strategy, cursor_column, batch_size or state, and merge matches rows on the primary key.

Into Databricks

From sqlload 0.23.0, a postgres_cdc source loads into a Databricks target, under merge (soft or hard deletes) or changelog. The snapshot copy, schema changes while streaming, tables joining the source and load --follow work as for DuckDB and Snowflake. With the setup from the guide:

default_target = "lakehouse"

[connections.lakehouse]
kind = "databricks"
host = "${ENV:DATABRICKS_HOST}"
client_id = "${ENV:DATABRICKS_CLIENT_ID}"
client_secret = "${ENV:DATABRICKS_CLIENT_SECRET}"
warehouse_id = "${ENV:DATABRICKS_WAREHOUSE_ID}"
catalog = "main"
volume = "main.raw.stage"

[targets.lakehouse]
connection = "lakehouse"
loader_schema = "raw"

plan shows each table’s snapshot copy, then how its changes apply:

$ 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   -> STRING
│   ├── country     text   -> STRING
│   └── + _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            -> STRING
    ├── tags        text            -> STRING
    ├── 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 -> Parquet -> PUT to /Volumes/main/raw/stage -> COPY INTO
│   └── 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 -> Parquet -> PUT to /Volumes/main/raw/stage -> COPY INTO
    └── batch   about every 1 GB

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

With cdc_mode = "changelog" a table shows snapshot copy, then append changes (changelog), and with deletes = "hard", apply changes, hard deletes.

Databricks’ Statement Execution API commits every statement on its own, so each table’s batch is written by one statement, after its rows are staged as Parquet in the volume and copied in with COPY INTO, as the plan’s route shows:

merge is one MERGE on the key. A truncate in the batch is that MERGE’s WHEN NOT MATCHED BY SOURCE clause, which marks (soft deletes) or removes (hard deletes) the rows from before it. Readers never see the table emptied between the truncate and the rows after it.
changelog is one INSERT of the batch’s changes.

As on every target, the slot confirms a batch only after every table it touched has committed, and a replay skips the changes a table already holds (see What a load guarantees).

One case isn’t covered by tests: if the sqlload process is killed while its MERGE or INSERT is still running on the warehouse, that statement can still finish. The restarted load’s replay then relies on Delta’s conflict detection between the two statements.

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 from the guide 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 in the guide

$ 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, and names each table by its folder. A snapshot copy (the first load, a table joining the source, or --resnapshot) writes its files, then deletes the folder’s older ones. The first load, then the insert, update and delete from step 4:

$ sqlload load
…
loaded   lake/app/customers/  50 rows  0.2s
loaded   lake/app/orders/     1,000 rows  0.2s

✓ Completed  2 loaded, 1,050 rows, 0.2s
$ sqlload load
…
loaded   lake/app/customers/  0 rows  0.3s
loaded   lake/app/orders/     3 rows  0.3s

✓ Completed  2 loaded, 3 rows, 0.3s

Then 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.

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 BigQuery target:

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

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.

Before 0.23.0 was released, live rounds loaded into a Databricks SQL warehouse (2X-Small, serverless) and into DuckDB from the same Postgres, and both matched Postgres exactly. They covered the snapshot, inserts, an update that left a TOASTed value out, deletes, and a truncate and reinsert in one transaction, under merge with soft and with hard deletes and under changelog. They also covered columns added, dropped and widened while streaming, and load --follow with a column added and a truncate mid-follow. While the follow was idle, the warehouse’s query history showed no statements for 40 seconds. A live test then covered a load stopped after its first table committed, a changelog load stopped before the slot confirmed, and a truncate alone and a truncate with a reinsert under soft and hard deletes.

Part of SQLStack