Version: 2.0.0

Publish data from Db2 CDC

Follow the change log instead of re-reading tables

A CDC source uses @stream_publisher rather than @publisher. A long-lived worker follows the Db2 change log, 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.

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 publish changes to the ORDERS and CUSTOMERS tables out of db2

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

That is change data capture rather than a query read, so this is a Db2 CDC stream publisher. Its guide and connection template are loaded. Five 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
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 db2+ibm_db://db.internal:50000/sales, tracking SALES.ORDERS and SALES.CUSTOMERS, starting from tail, collection sales_cdc.

Called tabsdata 4 times (ctrl+o to expand)

Collection sales_cdc created in group sources, with a Db2CdcStreamSrcConn to db.internal:50000/sales. Connection check passed, and both tables have change capture enabled.

Called tabsdata 2 times (ctrl+o to expand)

The stream publisher, registered in sales_cdc. start_from="tail" means the worker begins at the current log position and never backfills, so the first version holds only changes committed after registration:

@stream_publisher(
  source=Db2CdcStreamSrc(
      tables=["SALES.ORDERS", "SALES.CUSTOMERS"],
      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    │ SALES.ORDERS    │ 1,842             │
├───────────────────┼─────────────────┼───────────────────┤
│ customers_changes │ SALES.CUSTOMERS │ 96                │
└───────────────────┴─────────────────┴───────────────────┘

start_from also accepts "head" to begin at the oldest retained log, or an explicit position. Rows carry the change operation, so a downstream transformer can apply inserts, updates and deletes rather than treating every row as new. A stream publisher takes no trigger_by and must not declare a ctx parameter.

Follow the change log instead of re-reading tables​

Follow the change log instead of re-reading tables