Parsing pgoutput format with psycopg2

Decode PostgreSQL's native pgoutput binary replication stream in Python by driving psycopg2's LogicalReplicationConnection and parsing each message…

Decode PostgreSQL’s native pgoutput binary replication stream in Python by driving psycopg2’s LogicalReplicationConnection and parsing each message byte-for-byte — the concrete decoding step inside Python CDC parser development. Getting the byte alignment wrong does not throw a clean error; it silently mis-attributes columns or desyncs the stream, so a restart_lsn that never advances quietly fills pg_wal until the primary refuses writes.

This page assumes the logical decoding subsystem is already enabled and a logical replication slot exists on the pgoutput plugin. It covers the exact wire semantics of each message tag, diagnostic queries to isolate a decoder bug from a slot problem, a zero-downtime deployment sequence, and the feedback loop that keeps WAL bounded. Everything targets PostgreSQL 14 through 17.

Message Type Semantics

pgoutput emits a strict binary protocol over the WAL stream mechanics that decode committed transactions. psycopg2 hands each replication message to your callback as a complete payload bytes object — there is no separate length prefix to consume. Every message begins with a single-byte type tag; the payload layout that follows is fixed per tag. All integers are network byte order (big-endian); strings are null-terminated UTF-8.

Tag Message Key payload fields Decoder action Consumer behavior
B BEGIN final_lsn (Q), timestamp (q), xid (I) Open a transaction buffer keyed by xid Delimits an atomic batch; no rows yet
R RELATION rel_id (I), namespace, name, replica identity, natts (H), per-column metadata Cache column list under rel_id Schema context for every later tuple; re-emitted after DDL
I INSERT rel_id (I), N tuple Decode new-tuple against cached relation Append full new row to the batch
U UPDATE rel_id (I), optional K/O old tuple, N new tuple Decode old (if present) + new tuple Old image present only under REPLICA IDENTITY FULL/USING INDEX
D DELETE rel_id (I), K/O old tuple Decode old tuple (key or full row) Carries only the replica identity columns
C COMMIT flags (B), commit_lsn (Q), end_lsn (Q), timestamp (q) Flush batch to sink, then acknowledge end_lsn The only point at which it is safe to advance the slot
T TRUNCATE nrelids (I), flags, relation ids Emit a truncate event per relation Cascades and RESTART IDENTITY encoded in flags
O/Y/M ORIGIN / TYPE / MESSAGE origin name, type OIDs, logical message Optional; skip if unused Origin filtering, custom types, pg_logical_emit_message

Inside every tuple, each column value is prefixed by a one-byte kind flag that decides how many bytes to read next. This is the highest-risk part of the parser:

Flag Meaning Bytes that follow
n SQL NULL none
u Unchanged TOASTed value (not sent) none — reuse the prior value or mark absent
t Textual value 4-byte length I, then that many UTF-8 bytes
b Binary value (proto v2+, binary option) 4-byte length I, then raw bytes

The u flag is the classic silent-corruption trap: it is not NULL. Writing NULL to the sink on a u flag overwrites an unchanged large value with nothing. Under REPLICA IDENTITY FULL — the safe default for a hand-built parser, see publication and subscription models — UPDATE and DELETE carry a full old-tuple image so downstream upserts and tombstones reconstruct exactly.

The pgoutput message stream within one transaction Within a single decoded transaction the stream begins at a start marker and enters BEGIN, which opens a transaction buffer keyed by xid. From BEGIN the decoder either takes the RELATION path when a table's schema is new or has changed — caching its column list — or goes directly to the DML state when the relation is already cached; RELATION also flows into DML. The DML state self-loops once per row change, decoding each INSERT, UPDATE or DELETE tuple against the cached schema. When the transaction ends the stream transitions to COMMIT, which flushes the batch to the sink; only then does control reach the end state and the flush LSN advance so PostgreSQL can reclaim WAL. BEGIN open xid buffer RELATION cache column list DML decode tuple COMMIT flush batch new / changed schema schema cached INSERT / UPDATE / DELETE flush LSN advances
The pgoutput message stream within one transaction, from BEGIN to a COMMIT that advances the flush LSN. RELATION is emitted only when the schema is new or has changed; every row change loops back through DML before the transaction commits.

A production decoder is stateful: it caches RELATION messages so rel_id resolves to a fully qualified table name and column set without redundant catalog lookups. The core parser below reads a message into an io.BytesIO cursor and dispatches on the tag.

python
import struct
import io
from typing import Dict, Any

class PgOutputParser:
    """Binary pgoutput decoder aligned with the PostgreSQL 14+ protocol."""

    def __init__(self) -> None:
        self.relations: Dict[int, Dict[str, Any]] = {}  # rel_id -> schema cache

    @staticmethod
    def _read_string(buf: io.BytesIO) -> str:
        out = bytearray()
        while (byte := buf.read(1)) and byte != b"\x00":
            out += byte
        return out.decode("utf-8", errors="replace")

    def _read_tuple(self, buf: io.BytesIO, columns) -> Dict[str, Any]:
        natts = struct.unpack("!H", buf.read(2))[0]
        row: Dict[str, Any] = {}
        for i in range(natts):
            name = columns[i]["name"] if i < len(columns) else f"col_{i}"
            flag = buf.read(1)
            if flag in (b"n", b"u"):
                # 'n' = SQL NULL; 'u' = unchanged TOAST — DISTINCT meanings.
                row[name] = None if flag == b"n" else _UNCHANGED
            elif flag in (b"t", b"b"):
                length = struct.unpack("!I", buf.read(4))[0]
                value = buf.read(length)
                row[name] = value.decode("utf-8", errors="replace") if flag == b"t" else value
        return row

    def parse_message(self, payload: bytes) -> Dict[str, Any]:
        buf = io.BytesIO(payload)
        tag = buf.read(1)
        if not tag:
            return {}

        if tag == b"B":  # BEGIN
            final_lsn, ts, xid = struct.unpack("!QqI", buf.read(20))
            return {"type": "BEGIN", "final_lsn": final_lsn, "xid": xid}

        if tag == b"C":  # COMMIT
            flags = buf.read(1)
            commit_lsn, end_lsn, ts = struct.unpack("!QQq", buf.read(24))
            return {"type": "COMMIT", "commit_lsn": commit_lsn, "end_lsn": end_lsn}

        if tag == b"R":  # RELATION (schema)
            rel_id = struct.unpack("!I", buf.read(4))[0]
            schema = self._read_string(buf)
            table = self._read_string(buf)
            buf.read(1)  # replica identity setting
            ncols = struct.unpack("!H", buf.read(2))[0]
            cols = []
            for _ in range(ncols):
                buf.read(1)  # column flags (1 = part of key)
                cname = self._read_string(buf)
                type_oid, atttypmod = struct.unpack("!Ii", buf.read(8))
                cols.append({"name": cname, "type_oid": type_oid})
            self.relations[rel_id] = {"schema": schema, "table": table, "columns": cols}
            return {"type": "RELATION", "rel_id": rel_id, "schema": schema, "table": table}

        if tag in (b"I", b"U", b"D"):  # INSERT / UPDATE / DELETE
            rel_id = struct.unpack("!I", buf.read(4))[0]
            rel = self.relations.get(rel_id)
            if rel is None:
                raise KeyError(f"rel_id {rel_id} seen before its RELATION message")
            cols = rel["columns"]
            old = new = None
            if tag == b"U":
                sub = buf.read(1)  # 'K' key, 'O' full old tuple, or start of 'N'
                if sub in (b"K", b"O"):
                    old = self._read_tuple(buf, cols)
                    buf.read(1)   # 'N'
                new = self._read_tuple(buf, cols)
            elif tag == b"I":
                buf.read(1)       # 'N'
                new = self._read_tuple(buf, cols)
            else:  # DELETE
                buf.read(1)       # 'K' or 'O'
                old = self._read_tuple(buf, cols)
            return {"type": tag.decode(), "table": f'{rel["schema"]}.{rel["table"]}',
                    "old": old, "new": new}

        return {"type": "unknown", "tag": tag.decode(errors="replace")}


_UNCHANGED = object()  # sentinel: leave the sink's existing value untouched

Diagnostic Patterns

Decoder defects and slot defects look identical from the client — the stream just stops. Separate them from the server side before touching Python.

Peek at exactly what the slot will emit, without consuming it. pg_logical_slot_peek_binary_changes reads WAL without advancing the cursor, so you can inspect the raw bytes your parser is about to see:

sql
-- Inspect the next 20 messages WITHOUT acknowledging anything.
SELECT lsn, xid, left(encode(data, 'hex'), 80) AS first_bytes
FROM pg_logical_slot_peek_binary_changes(
       'cdc_pgoutput_slot', NULL, 20,
       'proto_version', '1', 'publication_names', 'cdc_py');

If peek returns well-formed messages (a B/0x42 first byte on the first row of a transaction) but your consumer chokes, the bug is in the Python decoder, not the slot.

Watch retained WAL and liveness. active = false on a slot that should be streaming means the parser died or never reconnected:

sql
SELECT slot_name, active, restart_lsn, confirmed_flush_lsn,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots
WHERE slot_name = 'cdc_pgoutput_slot';
-- Alert when retained_wal exceeds ~25% of max_slot_wal_keep_size, or when
-- active = false persists beyond 60 s on a slot expected to be live.

Confirm the sender is shipping bytes. A non-zero, steadily-growing unflushed value means the walsender is ahead of your acknowledged position — normal briefly, a stall if it never shrinks:

sql
SELECT application_name, state,
       pg_size_pretty(pg_wal_lsn_diff(sent_lsn, flush_lsn)) AS unflushed,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn)) AS total_lag
FROM pg_stat_replication;

These map onto the same panels described in async monitoring integration.

Isolating a stalled pgoutput consumer: decoder bug versus slot problem The stream stops the same way whether the Python decoder or the replication slot is at fault, so the first move is a server-side peek. Call pg_logical_slot_peek_binary_changes: if the first byte of a transaction's first row is 0x42 (the ASCII 'B' that opens a BEGIN), the slot is emitting well-formed messages and the defect is in the Python decoder — remediate by fixing the tuple byte alignment, redeploying the consumer and re-peeking to confirm. If peek returns empty or garbled bytes, inspect pg_replication_slots: active = false or a steadily growing retained_wal indicates a slot or WAL problem — remediate by reconnecting the consumer, and if the slot is lost drop it, recreate it and resnapshot from the last durably persisted LSN. Consumer stream stalls peek_binary_changes: first byte = 0x42 ('B')? read WAL without consuming Bug in the Python decoder slot emits clean messages Fix tuple byte alignment, redeploy, re-peek to confirm Slot / WAL problem active = false · retained_wal grows Reconnect consumer; if lost, drop, recreate, resnapshot yes · well-formed no · empty / garbled
Isolate the fault from the server before touching Python: a clean 0x42 first byte from peek puts the bug in the decoder, while empty or malformed bytes plus a stalled slot point at a WAL/slot problem — each branch has its own remediation.

Safe Deployment Sequence

Roll a pgoutput consumer out without disturbing the primary, and keep a clean revert path.

1. Enforce the server parameter baseline. Deviations cause silent WAL truncation or protocol rejection. wal_level requires a restart, so plan the bounce:

sql
ALTER SYSTEM SET wal_level = 'logical';            -- restart required
ALTER SYSTEM SET max_replication_slots = '10';     -- restart required
ALTER SYSTEM SET max_wal_senders = '10';           -- restart required
ALTER SYSTEM SET wal_sender_timeout = '60s';       -- reload-safe
ALTER SYSTEM SET max_slot_wal_keep_size = '10GB';  -- PG 13+; finite = self-protecting
SELECT pg_reload_conf();

Verify with SHOW wal_level; and confirm the slot object exists. Set max_slot_wal_keep_size to a finite value so a stalled parser invalidates its own slot rather than filling the disk — a resnapshot is almost always the right trade over a downed primary.

2. Create the slot and narrow the stream. The publication decides which tables and operations reach the wire; the slot guarantees WAL retention until you acknowledge an LSN. Full provisioning detail is under initializing replication slots:

sql
CREATE PUBLICATION cdc_py FOR TABLE public.orders, public.line_items;
ALTER TABLE public.orders     REPLICA IDENTITY FULL;   -- full old images
ALTER TABLE public.line_items REPLICA IDENTITY FULL;
SELECT * FROM pg_create_logical_replication_slot('cdc_pgoutput_slot', 'pgoutput');

3. Open the replication connection in binary mode. The replication=database DSN parameter is mandatory; omitting it defaults to standard query mode and raises psycopg2.ProgrammingError. decode=False is required so pgoutput bytes reach your parser intact:

python
import psycopg2
from psycopg2.extras import LogicalReplicationConnection

DSN = ("host=primary-db port=5432 dbname=cdc_source user=cdc_reader "
       "password=use-a-secret-manager replication=database connect_timeout=10")

def init_consumer(slot_name: str, publication: str):
    conn = psycopg2.connect(DSN, connection_factory=LogicalReplicationConnection)
    cur = conn.cursor()
    cur.start_replication(
        slot_name=slot_name,
        decode=False,                                  # binary pgoutput
        options={"proto_version": "1", "publication_names": publication},
        status_interval=10,                            # keepalive cadence, seconds
    )
    return conn, cur

Always confirm the slot is inactive before start_replication — a second consumer on an active slot desyncs the protocol. On PG 14+, set proto_version to 2 to stream in-progress transactions and bound memory on very large commits.

4. Validate with a dry read. Consume a handful of messages and log the decoded tags before wiring the real sink; confirm you see BEGIN → RELATION → INSERT/... → COMMIT in order and that column names resolve from the schema cache.

5. Revert cleanly. To back out, stop the consumer, then drop the slot so no WAL is retained — this is the single most important teardown step:

sql
SELECT pg_drop_replication_slot('cdc_pgoutput_slot');  -- releases retained WAL
DROP PUBLICATION IF EXISTS cdc_py;

Never leave an inactive slot behind on rollback; it pins restart_lsn indefinitely.

Pipeline Integration

The decoder is one stage; correctness depends on when you acknowledge. PostgreSQL retains WAL until the consumer confirms an LSN via a standby status update. send_feedback is called on the cursor, not the connection, and only after the sink has durably persisted the batch.

python
import select

def consume_stream(cur, parser: PgOutputParser, sink) -> None:
    batch, begin_lsn = [], None
    while True:
        msg = cur.read_message()
        if msg is None:
            # Idle: block on the socket but wake to send a keepalive so the
            # walsender does not drop us at wal_sender_timeout.
            if not select.select([cur], [], [], 10)[0]:
                cur.send_feedback()          # keepalive only; no LSN advance
            continue

        event = parser.parse_message(msg.payload)
        etype = event.get("type")
        if etype == "BEGIN":
            batch, begin_lsn = [], msg.data_start
        elif etype in ("I", "U", "D"):
            batch.append(event)
        elif etype == "COMMIT":
            sink.write_batch(batch, event["commit_lsn"])   # raises on failure
            # ONLY now is it safe to let PostgreSQL reclaim WAL up to here.
            cur.send_feedback(flush_lsn=msg.data_start, write_lsn=msg.data_start,
                              apply_lsn=msg.data_start, reply=True)
            batch = []

Advancing flush_lsn before write_batch confirms durability is how you lose data on a crash: PostgreSQL discards WAL you never stored. Persist the sink offset and the LSN in one atomic write so recovery resumes from a single source of truth.

Backpressure. Keep the feedback cadence at or below wal_sender_timeout / 3. When a downstream sink lags, buffer into a bounded queue and pause read_message before the queue saturates — never stall the acknowledgment path to wait on downstream I/O.

Typed contracts and routing. pgoutput yields raw column text; convert it into a registry-governed record via JSON to Avro transformation, then partition and dead-letter events through event routing and Kafka integration. Embed xid and commit_lsn in the payload so the sink can deduplicate on replay.

Failover. Logical slots are not replicated to standbys by default; PG 17 adds opt-in slot synchronization via the failover flag and pg_sync_replication_slots(). Without it, treat a lost slot as a resnapshot: recreate the slot, reconcile against your last durably persisted LSN — mirroring subscription sync procedures — and only then resume. Watch pg_replication_slots.wal_status for lost or extended, which flag WAL gaps needing intervention.

Authoritative references

← Back to Python CDC Parser Development