Version: 2.0.0

PostgresCdcStreamSrc

class

class PostgresCdcStreamSrc(
    tables: list[PostgresTableNameSpec],
    replication_slot: str,
    start_from: PostgresCdcStartFromSpec,
    replication_slot_behavior: ReplicationSlotBehaviorSpec = 'reuse',
    publication_name: str | None = None,
    columns_format: CdcColumnsFormatSpec = 'columns',
    flatten_columns: bool = True,
    src_cfg: CdcSrcCfgSpec | None = None,
)

Bases: StreamSrc

Categories: source

Postgres CDC stream source for @stream_publisher.

A streaming change-data-capture source: it emits one tableframe-batch per WAL flush across the tracked tables. Postgres logical replication requires a replication slot -- a persistent cursor on the WAL the database will not garbage-collect until the consumer confirms. replication_slot names it and replication_slot_behavior controls create vs reuse. The decorated @stream_publisher function receives the flushed batch as a list of frames and must NOT declare a ctx parameter.

Parameters

parameter

tableslist[PostgresTableNameSpec] (list[str])

Qualified schema.table names to track.

parameter

replication_slotstr

Logical replication slot to attach to.

parameter

start_fromPostgresCdcStartFromSpec (Literal['head', 'tail'] | LsnPosition)

Where to begin -- "head" (replay from the oldest retained WAL), "tail" (current WAL LSN) or an LsnPosition.

parameter

replication_slot_behaviorReplicationSlotBehaviorSpec (Literal['create', 'reuse'])

"reuse" (default, slot must exist) or "create" (create it, no-op if it exists).

parameter

publication_namestr | None

Optional publication scoping the slot's output (wal2json's publication_names option).

parameter

columns_formatCdcColumnsFormatSpec (Literal['columns', 'struct', 'map'])

Layout for the new (and old) columns payload on each change row -- "columns" (default), "struct" or "map".

parameter

flatten_columnsbool

When True (default), the new columns are inlined as top-level columns.

parameter

src_cfgCdcSrcCfgSpec | None (Mapping[Literal['postgres.cdc.blocking_timeout_seconds', 'postgres.cdc.logging_level', 'postgres.cdc.wal2json_format_version', 'tabsdata.cdc.buffer.max_rows', 'tabsdata.cdc.buffer.max_bytes', 'tabsdata.cdc.buffer.max_secs', 'tabsdata.cdc.trigger.max_rows', 'tabsdata.cdc.trigger.max_bytes', 'tabsdata.cdc.trigger.max_secs', 'tabsdata.cdc.poll_interval_seconds'], Any] | None)

Optional config mapping over the allowed CDC keys only -- buffer / trigger thresholds, polling cadence, wal2json format version.

Parameters​

Parameters