Version: 2.0.0

Publish data from MySQL CDC

Follow the binlog instead of re-reading tables

A CDC source uses @stream_publisher rather than @publisher. A long-lived worker follows the MySQL binlog, 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 stream CDC changes to the orders and customers tables out of mysql

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

That is a MySQL 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 mysql://db.internal:3306/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 MySQLCdcStreamSrcConn to db.internal:3306. Connection check passed, and the user has the replication grants the binlog reader needs.

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 binlog position and never backfills, so the first version holds only changes committed after registration:

@stream_publisher(
  source=MySQLCdcStreamSrc(
      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 binlog, 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 binlog instead of re-reading tables​

Follow the binlog instead of re-reading tables