Trigger off CDC in an SQL Database
A CDC stream source reads a database's replication log and emits one Tabsdata table per
tracked table. Every insert, update, and delete arrives, rather than whatever a SELECT
happens to see when it runs.
Three databases have a CDC source, each with its own connection type:
| Connection type | Database | Class |
|---|---|---|
mysql-cdc | MySQL | MySQLCdcStreamSrc |
postgres-cdc | PostgreSQL | PostgresCdcStreamSrc |
db2-cdc | IBM Db2 | Db2CdcStreamSrc |
What runs the publisher is a rollover rather than a clock or an upstream commit. The stream worker consumes the log continuously, and when a threshold trips it hands the batch to the publisher, which de-duplicates and offset-filters the rows before ingesting them.
The database URI and its credentials live on the collection's connection, not in the function, so the collection has to exist with a CDC connection bound to it before a stream publisher can be registered there.
Needs a stream publisher, and a source collection with a mysql-cdc connection to hold it. Four values left, then I generate and register. Reply with them one per line, or point me at a file that has them:
- collection name
- database URI and credentials
- tables to track
- where in the log to start
Read 1 file, called tabsdata 2 times (ctrl+o to expand)
Password found. Reading back what is not secret: mysql://db.internal:3306/shop, tracking shop.products and shop.stock, starting at the tail, collection inventory.
Called tabsdata 3 times (ctrl+o to expand)
Collection inventory created in group sources, with a MySQLCdcStreamSrcConn to db.internal:3306. Connection check passed and binlog access confirmed.
Generating — MySQLCdcStreamSrc tracking both tables. A CDC source emits one output table per tracked table, so this one has two.
Called tabsdata 3 times (ctrl+o to expand)
The stream publisher, registered in inventory:
@stream_publisher(
source=MySQLCdcStreamSrc(
tables=["shop.products", "shop.stock"],
start_from="tail",
),
flush_interval_mins=5,
output_tables=["products", "stock"],
)
def read_changes(
products: TableFramesSpec,
stock: TableFramesSpec,
) -> tuple[TableFramesSpec, TableFramesSpec]:
return (products, stock)
Called tabsdata 2 times (ctrl+o to expand)
Execution committed, 0 failures.
inventory/products and inventory/stock are live — the worker is reading the binlog from the current position:
┌───────────────┬──────────────┬───────┐ │ tracked table │ output table │ start │ ├───────────────┼──────────────┼───────┤ │ shop.products │ products │ tail │ ├───────────────┼──────────────┼───────┤ │ shop.stock │ stock │ tail │ └───────────────┴──────────────┴───────┘
One output table per tracked table, in order, and one input slot per output table. The function declares no transaction context, since it runs inside the runtime's stream loop rather than per-transaction.
Where the stream starts
start_from decides what the worker reads first. For PostgreSQL, "head" starts at the
oldest retained WAL, "tail" starts at the current LSN, and an LsnPosition starts at a
position you name. Starting at the tail skips the history and streams only what arrives
from now on.
PostgreSQL needs a replication slot
PostgreSQL logical replication is slot-based, so its connection carries fields the other two do not.
Called tabsdata 3 times (ctrl+o to expand)
Switched to PostgresCdcStreamSrc, which needs a logical replication slot. The connection now carries replication_slot, and replication_slot_behavior defaults to reusing an existing slot rather than creating one.
┌───────────────────────────┬──────────────┐ │ field │ value │ ├───────────────────────────┼──────────────┤ │ replication_slot │ tabsdata_cdc │ ├───────────────────────────┼──────────────┤ │ replication_slot_behavior │ reuse │ ├───────────────────────────┼──────────────┤ │ columns_format │ columns │ ├───────────────────────────┼──────────────┤ │ flatten_columns │ true │ └───────────────────────────┴──────────────┘
publication_name is optional. columns_format also accepts struct and map if you would rather keep the change payload nested.
Closing a batch sooner
The flush interval is a fallback ceiling, minimum 1 whole minute. The connector's own thresholds cover buffer size and polling cadence, and whichever threshold trips first closes the batch, so the connector's window is usually the one that fires.
Drivers
MySQL and PostgreSQL CDC both need a database driver that Tabsdata cannot redistribute. Declare it during installation, as described in Installing Third-Party Database Drivers. Adding it later means re-running the installation.
What to do next
Trigger off a Kafka Stream covers the fourth stream source, which follows the same rollover model against a topic.
See Stage Triggers for how a batch closes and becomes a run.