Version: 2.0.0

Publish data from Postgres CDC

Follow the WAL instead of re-reading tables

A CDC source uses @stream_publisher rather than @publisher. A long-lived worker follows the write-ahead log through a logical replication slot, and each time a rollover threshold trips it writes the completed buffer out and the publisher commits it. One output table per tracked table, so tables and output_tables line up position by position.

Postgres is the one CDC source that needs a replication slot named up front, because the slot is what holds the WAL position between runs.

warning

There are several ways to pass secrets into Tabsdata. See Secrets Management for more info if you are planning to have an LLM register secrets into Tabsdata's credential store.

Can you stream CDC changes to the orders and customers tables out of postgres

Listed 1 directory, called tabsdata 3 times (ctrl+o to expand)

That is a Postgres CDC stream publisher. Its guide and connection template are loaded. Six values left, then I generate and register. Reply with them one per line, or point me at a file that has them:

collection name
connection URI
user and password
tables to track, fully qualified
replication slot name
where to start reading, and rollover interval

everything you need is in

Read 1 file, called tabsdata 2 times (ctrl+o to expand)

Password found. Reading back what is not secret: uri postgres://db.internal:5432/sales, tracking public.orders and public.customers, slot tabsdata_sales, starting from tail, collection sales_cdc.

Called tabsdata 4 times (ctrl+o to expand)

Collection sales_cdc created in group sources, with a PostgresCdcStreamSrcConn to db.internal:5432/sales. Connection check passed, wal_level is logical, and the slot tabsdata_sales already exists so the default "reuse" behavior applies.

Called tabsdata 2 times (ctrl+o to expand)

The stream publisher, registered in sales_cdc. flatten_columns defaults to true, so each source column arrives as its own column rather than nested inside a struct:

@stream_publisher(
  source=PostgresCdcStreamSrc(
      tables=["public.orders", "public.customers"],
      replication_slot="tabsdata_sales",
      replication_slot_behavior="reuse",
      start_from="tail",
  ),
  flush_interval_mins=2,
  output_tables=["orders_changes", "customers_changes"],
)
def read_sales_changes(
  orders: TableFramesSpec,
  customers: TableFramesSpec,
) -> tuple[TableFrameSpec, TableFrameSpec]:
  return (concat(orders), concat(customers))

Called tabsdata 6 times (ctrl+o to expand)

Stream worker started. First rollover committed after the 2 minute interval, 0 failures.

Two tables are live in sales_cdc, and both gain a version per rollover:

┌───────────────────┬──────────────────┬───────────────────┐
│       table       │     tracking     │ rows this version │
├───────────────────┼──────────────────┼───────────────────┤
│ orders_changes    │ public.orders    │ 1,842             │
├───────────────────┼──────────────────┼───────────────────┤
│ customers_changes │ public.customers │ 96                │
└───────────────────┴──────────────────┴───────────────────┘

Set replication_slot_behavior to "create" to have the connector make the slot instead of expecting it. start_from also accepts "head" for the oldest retained WAL, or an explicit LSN. A slot that is never consumed holds WAL on the server indefinitely, so delete the collection rather than leaving a stopped stream behind.

Follow the WAL instead of re-reading tables​

Follow the WAL instead of re-reading tables