Flusher¶

The flusher connects to PostgreSQL via a logical replication slot (using the pgoutput plugin) and consumes WAL messages as they are produced.

WAL capture¶

For each transaction, the flusher:

1. Buffers all row change events (INSERT, UPDATE, DELETE) until the transaction COMMIT

2. Serializes each table's buffered events into a Parquet file with a fixed staging schema:

   Column	Type	Description
   _op	string	Operation: I, U, or D
   _lsn	int64	Source WAL LSN
   _ts	timestamptz	PostgreSQL commit timestamp
   _xid	int64	PostgreSQL transaction ID
   _unchanged_cols	string	Comma-separated unresolved TOAST columns
   _data	binary	The row's columns, binary-encoded
   _moved_from	binary	An update's old key, when it changed the key with TOAST columns unchanged

   The staging schema is fixed regardless of the source table structure. Schema evolution only affects the materialized Iceberg table, not the staged files. A row's columns are numbered into the column names its file lists once, in its key-value metadata, beside the format's version: a version of pg2iceberg reads only files of its own format, so materialize what's staged before upgrading across a change of it.

3. Uploads the Parquet files to S3 and atomically registers them in the leaderless log

The replication slot's confirmed flush LSN is advanced immediately after staging — once events are durably registered in S3 + log_index, PostgreSQL can recycle WAL up to that point. On a crash between staging and materialization, pg2iceberg recovers by re-reading the staged Parquet files from the log; the WAL is no longer needed.

Staged log¶

The staged-log primitive provides offset-ordered, durable delivery of WAL events from the flusher to one or more materializer workers without a dedicated leader process.

The coordinator role is filled by the source PostgreSQL database itself. This is a deliberate choice: the coordinator is technically a single point of failure, but if the source database goes down, replication cannot happen regardless — so there is no additional availability cost to colocating coordination there. It also means pg2iceberg needs no extra infrastructure beyond the database you are already replicating from. (Operators who want isolation can point state.postgres_url at a separate Postgres.)

Coordination tables (in _pg2iceberg):

Append (flusher → log):

1. Upload staged Parquet files to S3 in parallel (up to 8 at a time)
2. In a single PostgreSQL transaction: increment log_seq by the batch size, insert per-claim rows into log_index with the assigned offset range, S3 path, and flushable_lsn, and persist any blue-green markers

Step 2 is atomic — a flush is either fully registered or not registered at all. Orphaned S3 files from failed step 2s are collected by table maintenance.

While streaming, a flush — encoding its changes, both steps — runs on a task of its own, and the flusher goes on receiving the next changes meanwhile. One flush runs at a time, so batches register in commit order, and the slot is acked past a flush only once its step 2 commits. A flush that fails stops pg2iceberg, which resumes from the last registered batch on restart — or, if the replication stream dropped with it, the reconnected stream starts there — so none of its changes is skipped.

Read (materializer ← log):

The materializer reads log_index entries with offsets greater than its cursor, downloads the referenced files — up to eight ahead of the one it's applying — decodes them via the staging codec, and processes events in offset order. There is no in-memory cache optimization for the single-process case; profiling showed the S3-fetch overhead is amortized well below the materializer cycle interval.

Restart record¶

Replication state is split across several narrow tables — no single "checkpoint" blob. Each concern that needs durability gets its own row, so updates from different code paths don't contend on a shared OCC token:

• Confirmed flush LSN — recorded in _pg2iceberg.flushed_lsn before every standby ack to the slot, which never goes past it: if recording fails, the slot is acked only as far as the last record. The record only moves forward. Since the slot can't lead it, a slot that does was advanced by something else (pg_replication_slot_advance, drop+recreate, stray pg_recvlogical), and the WAL in between was never staged: startup refuses to run, and the runtime watcher stops the pipeline. The slot lagging what's staged is normal — it's acked on the standby tick, after the flush — and shows as the pg2iceberg_slot_ack_lag_bytes gauge.
• Per-table snapshot status — _pg2iceberg.tables rows store snapshot_complete, pg_oid, and snapshot_lsn per table. The pg_oid field drives the TableIdentityChanged invariant (DROP+recreate detection).
• Per-table mid-snapshot resume — _pg2iceberg.snapshot_progress stores the canonical-PK cursor of the last successfully staged row. Cleared on completion.
• Cluster fingerprint — singleton row in _pg2iceberg.pipeline_meta stores IDENTIFY_SYSTEM's system_identifier. Stamped at first run; subsequent runs against a different cluster fail with SystemIdMismatch.

On startup, pg2iceberg cross-references all four against PG and the slot, and refuses to start on any mismatch. See validate_startup for the full invariant list.