PostgreSQL source

DeltaForge captures row-level changes from PostgreSQL using logical replication with the pgoutput plugin.

Prerequisites

PostgreSQL Server Configuration

Enable logical replication in postgresql.conf:

# Required settings
wal_level = logical
max_replication_slots = 10    # At least 1 per DeltaForge pipeline
max_wal_senders = 10          # At least 1 per DeltaForge pipeline

Restart PostgreSQL after changing these settings.

User Privileges

Create a replication user with the required privileges:

-- Create user with replication capability
CREATE ROLE deltaforge WITH LOGIN REPLICATION PASSWORD 'your_password';

-- Grant connect access
GRANT CONNECT ON DATABASE your_database TO deltaforge;

-- Grant schema usage and table access for schema introspection
GRANT USAGE ON SCHEMA public TO deltaforge;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO deltaforge;

-- The replication SLOT is auto-created on first run using the REPLICATION
-- attribute granted above. The PUBLICATION is NOT auto-created — create it
-- yourself (see "Replication Slot and Publication" below).

The same role reads the catalog for capture-time annotations and checks. Each such read first proves, in its own snapshot, that it reaches the node and the live walsender serving the stream: it reads pg_replication_slots, pg_stat_activity (the walsender’s row; visible to the same role, otherwise grant pg_read_all_stats) and pg_control_system(). A capture takes the table’s ACCESS SHARE lock (granted by SELECT on the table) for a few milliseconds: inserts, updates and deletes are not blocked, DDL on that table waits until the capture ends.

Unsupported: a tracked table with a stored generated column (GENERATED ALWAYS AS ... STORED) stops the source before its next row (pg_table_unsupported): pgoutput does not publish generated columns. Exclude the table or drop the generation.

pg_hba.conf

Ensure your pg_hba.conf allows replication connections:

# TYPE  DATABASE        USER            ADDRESS                 METHOD
host    replication     deltaforge      0.0.0.0/0               scram-sha-256
host    your_database   deltaforge      0.0.0.0/0               scram-sha-256

Replication Slot and Publication

DeltaForge creates the replication slot on first run and records durable ownership of it (bound to the server’s system_identifier, the database, the slot name, and the plugin). The publication is never created by DeltaForge: create it, then register it.

CREATE PUBLICATION my_pub FOR TABLE public.orders, public.order_items;

The publication contract

A publication DeltaForge streams from is immutable while it is registered. Registration is a superuser step, done before the source’s first start:

deltaforge pg-publication --dsn-file superuser.dsn register --publication my_pub
# review the database-wide impact printed, then repeat with --confirm

Registering a publication:

• requires an explicit FOR TABLE list of ordinary tables, with all four publish operations, no row filter, no column list, no publish_via_partition_root and no publish_generated_columns. FOR ALL TABLES, TABLES IN SCHEMA and partitioned tables are refused: their membership changes without publication DDL;
• transfers the publication to the NOLOGIN role deltaforge_publication_owner and records its canonical digest and the WAL position of the registration;
• installs enforcement for the whole database. While any publication of the database is registered, every ALTER PUBLICATION and DROP PUBLICATION is refused for every role, superusers included, and any drop that would remove a registered publication’s table (DROP TABLE, DROP SCHEMA ... CASCADE, DROP OWNED, …) is refused inside its transaction. The database enforces this while DeltaForge is stopped.

At startup and on every reconnect, before it snapshots, decodes, emits or advances a checkpoint, the source verifies that enforcement is intact, the publication is registered and its live digest equals the registration it accepted, and that it does not stream from WAL older than the registration. Otherwise it stops with pg_publication_changed or pg_publication_enforcement.

Changing a publication (membership, options, a table drop) is a database-wide maintenance procedure:

1. stop every DeltaForge PostgreSQL source using the database;
2. for each, record how it continues: deltaforge recover apply <pipeline> pg-publication-maintenance --arg mode=resnapshot (then the resnapshot operation) or --arg mode=abandon (the backlog before the new registration is skipped and never delivered; the applied plan is the audit record);
3. deltaforge pg-publication ... unregister --publication <each>;
4. deltaforge pg-publication ... uninstall (refused while any registration remains, naming them);
5. apply the change;
6. register the publications again (new digests);
7. restart the sources.

Outside the guarantee: a superuser can disable or drop the enforcement triggers. The source detects it at its next startup or reconnect and stops, but changes made between the tampering and that check are not guaranteed to have been captured completely.

Replica Identity

For complete before-images on UPDATE and DELETE operations, set tables to REPLICA IDENTITY FULL:

ALTER TABLE public.orders REPLICA IDENTITY FULL;
ALTER TABLE public.order_items REPLICA IDENTITY FULL;

Without this setting:

• FULL: Complete row data in before-images
• DEFAULT (primary key): Only primary key columns in before-images
• NOTHING: No before-images at all

DeltaForge warns at startup if tables don’t have REPLICA IDENTITY FULL.

Unchanged TOAST values

PostgreSQL stores large values out of line (TOAST). When an UPDATE leaves such a value unchanged, PostgreSQL does not send it in the new row.

• Under REPLICA IDENTITY FULL PostgreSQL sends the complete old row, including those values (verified on PostgreSQL 14 through 18). DeltaForge fills each unchanged value from the old row, exactly. The event’s after is complete.
• Otherwise the value is not available. Sinks that accept partial rows (native JSON: Kafka, Redis, NATS, HTTP, S3 JSON Lines) leave it out of after and list it in unchanged_columns.

Sinks that need complete rows (Avro encoding on any sink, S3 Parquet, ClickHouse, Elasticsearch) cannot represent a missing value. A pipeline with at least one of them requires REPLICA IDENTITY FULL on every captured table that has a TOAST-capable column (any variable-length type stored with a strategy other than PLAIN, domains and arrays included, read from the catalog).

• At start: a bulk check over the publication’s tables, before anything is snapshotted or streamed, refuses the pipeline with pg_table_unsupported (class toast_without_full_identity). The source log names every such table with its TOAST-capable columns and the statement that fixes it, for example ALTER TABLE "public"."docs" REPLICA IDENTITY FULL;. The refusal contains table and column names only, never row values, and the checkpoint does not move.
• While running: the check is repeated whenever PostgreSQL announces a table’s definition (stream start, after DDL, after reconnect or failover), so changing a table’s identity away from FULL stops the source before that table’s next row.

A pipeline whose sinks all accept partial rows does not require FULL; one sink that needs complete rows is enough to require it. deltaforge_pg_unchanged_toast_total{pipeline,source,outcome} counts values filled from the old row (reconstructed) and values left unavailable (unavailable).

WAL cost of FULL. Under FULL, every UPDATE and DELETE writes the complete old row to the WAL, and the old row includes its out-of-line values even when the UPDATE leaves them unchanged. The cost grows with row width:

More WAL also means more WAL decoded and sent to the source, and more retained for the slot while the source is behind. For wide, frequently updated tables, weigh this against delivering those tables only to sinks that accept partial rows.

Configuration

Set spec.source.type to postgres and provide a config object:

Message size limit

PostgreSQL sends each row change as one replication message, so a single wide row (or a large pg_logical_emit_message() payload) must fit under max_message_bytes. A larger message is refused before any of it is buffered, and the source stops fail-closed with the incident pg_message_too_large:

• the session that received it is abandoned and the source does not reconnect: a reconnect with the same limit would meet the same message again;
• the checkpoint stays before the message’s transaction, so nothing of it is acknowledged or delivered partially;
• the evidence is the message’s size (message_bytes), the limit (message_limit) and the source, never the payload.

Raise max_message_bytes above the reported size and restart the pipeline: it resumes from the held checkpoint and delivers the whole transaction. The limit bounds one message, not a transaction or total memory.

DSN Formats

DeltaForge accepts both URL-style and key=value DSN formats:

# URL style
dsn: "postgres://user:pass@localhost:5432/mydb"

# Key=value style
dsn: "host=localhost port=5432 user=deltaforge password=pass dbname=mydb"

Table Patterns

The tables field supports flexible pattern matching:

tables:
  - public.orders          # exact match: schema "public", table "orders"
  - public.order_%         # prefix: tables starting with "order_" (same as public.order_*)
  - myschema.*             # wildcard: all tables in "myschema"
  - %.audit_log            # cross-schema: "audit_log" table in any schema
  - orders                 # table "orders" in every schema

A wildcard (* or %) only counts as the last character of a part (a prefix match); _ and any other character are literal. A pattern without a qualifier matches that table name in every database/schema. The initial snapshot copies exactly the tables these patterns capture during CDC.

System schemas (pg_catalog, information_schema, pg_toast) are always excluded.

Start Position

⚠️ Not yet implemented. start_position is accepted by the config parser but currently ignored — a new slot always starts from its consistent point (the current WAL position when it is created). The forms below are aspirational.

Intended to control where replication begins when no checkpoint exists:

# Start from the earliest available position (slot's restart_lsn)
start_position: earliest

# Start from current WAL position (skip existing data)
start_position: latest

# Start from a specific LSN
start_position:
  lsn: "0/16B6C50"

Example

source:
  type: postgres
  config:
    id: orders-postgres
    dsn: ${POSTGRES_DSN}
    slot: deltaforge_orders
    publication: orders_pub
    tables:
      - public.orders
      - public.order_items

Resume Behavior

DeltaForge checkpoints progress using PostgreSQL’s LSN (Log Sequence Number):

1. With checkpoint: Resumes from the stored LSN
2. Without checkpoint: Uses the slot’s confirmed_flush_lsn or restart_lsn
3. New slot: Starts from the slot’s consistent point (the LSN its creation returns): changes committed after it are captured.

The pipeline reports running only once the slot exists, the start position is persisted for every sink and the stream is open, with that position as capture_start. A retry or restart before then starts from the slot again, never from a newer position.

Checkpoints are stored using the id field as the key.

Snapshot (Initial Load)

DeltaForge performs a consistent initial snapshot using PostgreSQL’s exported snapshot mechanism before starting logical replication.

How it works

DeltaForge anchors CDC at the replication slot’s consistent point C - the LSN pg_create_logical_replication_slot returns when the slot is created - rather than a separately sampled pg_current_wal_lsn(). A coordinator connection then exports an MVCC snapshot for worker mutual consistency; worker connections import it into their own REPEATABLE READ transactions, so all workers see one consistent DB state with no locks held on the source. Tables with a single integer primary key use PK-range chunking; others fall back to ctid page-range chunking.

Anchoring at C (which precedes the snapshot read) closes the snapshot-to-CDC seam: no committed row is lost. The trade-off is a bounded at-least-once overlap - a change committed between C and the snapshot export is present in both the snapshot and the CDC stream from C, so it is delivered more than once. This is at-least-once, not exactly-once (see Snapshot events).

Correctness note. Earlier versions sampled pg_current_wal_lsn() decoupled from the exported snapshot, which admitted a (small) window where a row could be absent from both the snapshot and the CDC stream. The slot-consistent-point anchor removes that window, trading the loss risk for the bounded overlap above.

Configuration

source:
  type: postgres
  config:
    id: orders-postgres
    dsn: ${POSTGRES_DSN}
    slot: deltaforge_orders
    publication: orders_pub
    tables:
      - public.orders
    snapshot:
      mode: initial           # initial | always | never (default: never)
      max_parallel_tables: 8  # tables snapshotted concurrently
      chunk_size: 10000       # rows per chunk for integer-PK tables

Snapshot events

Snapshot rows are emitted as Op::Read events (Debezium op: "r"), distinguishable from live CDC Op::Create events. The slot’s consistent point C is the CDC resume position, so no rows are missed. Rows committed in (C, snapshot-export] are delivered by both the snapshot (as Op::Read) and the CDC stream (as their change op) - the bounded at-least-once overlap. The two copies carry distinct event identities:

• Current-state / idempotent sinks (Elasticsearch, ClickHouse upsert, with version_source: source_position) converge to the correct current state - the duplicate is absorbed by last-writer-wins.
• Append-only sinks (Kafka, Redis, HTTP, ClickHouse changelog, legacy S3) receive the overlapping rows twice, by design.

This is at-least-once delivery; DeltaForge does not claim exactly-once for the snapshot-to-CDC boundary.

Legacy-anchor warning and metric

A pipeline whose initial snapshot was completed under the older (pre-hardening) anchor is flagged so you can decide to re-snapshot:

• a structured WARN at startup, and
• the gauge deltaforge_snapshot_unsafe_anchor{pipeline,source} held at 1.

Re-snapshotting under this version records the safe anchor and resets the gauge to 0. See Upgrade guidance.

Resume after interruption

An interrupted snapshot is copied again in full on the next restart, as a new snapshot generation with a new anchor; it is never resumed table by table. When DeltaForge can prove it owns the now-inactive slot, it re-anchors (drops and recreates its slot for a fresh C).

A snapshot counts as complete only once its terminal barrier is committed by the sinks the commit policy requires; a restart after that completes it without copying. Completion, lagging sinks, bounds and blocking are described in Initial Snapshots.

WAL slot retention safety

DeltaForge validates replication slot health before starting a snapshot and monitors it throughout. This prevents the slot from being invalidated during a long snapshot, which would make the captured LSN unreachable for CDC resume.

Preflight checks (before any rows are read):

• Fails hard if the slot does not exist or is already invalidated
• Fails hard if wal_status=lost
• Warns if wal_status=unreserved (WAL retention no longer guaranteed)
• Estimates WAL generated during snapshot (~2× data size) against max_slot_wal_keep_size; warns at ≥50%, HIGH RISK at ≥80%

During snapshot:

• Background task polls pg_replication_slots every 30s
• Cancels immediately on slot invalidation or disappearance
• Warns but continues on wal_status=unreserved

After all tables complete:

• Synchronous final check before the generation records its rows as produced: a snapshot whose slot is lost, invalidated or missing never completes, and blocks with snapshot_anchor_unavailable (see Bounds and blocking)

If you see WAL retention risk warnings:

ALTER SYSTEM SET max_slot_wal_keep_size = '10GB';
SELECT pg_reload_conf();

Upgrade guidance

The slot-consistent-point anchor and durable slot ownership change how existing slots are handled when a snapshot runs. Steady-state CDC resume from a checkpoint is unaffected - the notes below apply only when a snapshot runs.

• Existing slots created before this version have no ownership record. A re-snapshot on such a slot (mode: always, or mode: initial after clearing the checkpoint) cannot prove ownership and fails closed with remediation: drop the slot manually (SELECT pg_drop_replication_slot('<slot>')) so DeltaForge recreates it with ownership, then restart. New pipelines and first-run snapshots are unaffected.
• Ownership ambiguity / foreign slots. If the slot exists but DeltaForge cannot prove exclusive ownership (missing, partial, or mismatched record - e.g. a different server system_identifier after a restore/failover) or the slot is active, it fails closed and never drops the slot. Remediation: confirm no other consumer uses it, drop it, and restart; or set snapshot.mode = never to stream from the current position without an initial load.
• Interrupted snapshots. An owned, inactive slot left by a snapshot that was interrupted before its first checkpoint is re-anchored (dropped and recreated for a fresh consistent point) and fully re-snapshotted automatically.
• Safe re-snapshotting. To take a fresh, correctly-anchored snapshot, stop the pipeline and apply the resnapshot recovery operation (or set mode: always), then resume. Checkpoints are kept; a slot this source owns is re-anchored, and a lost one is recreated only under the operation’s explicit authorization. A completed, safely-anchored snapshot resets deltaforge_snapshot_unsafe_anchor to 0.

Type Handling

DeltaForge preserves PostgreSQL’s native type semantics:

Event Format

Each captured row change produces an event with:

• op: insert, update, delete, or truncate
• before: Previous row state (updates and deletes, requires appropriate replica identity)
• after: New row state (inserts and updates)
• table: Fully qualified table name (schema.table)
• tx_id: PostgreSQL transaction ID (xid)
• checkpoint: LSN position for resume
• schema_version: Schema fingerprint
• schema_sequence: Monotonic sequence for schema correlation

WAL Management

Logical replication slots prevent WAL segments from being recycled until the consumer confirms receipt. To avoid disk space issues:

1. Monitor slot lag: Check pg_replication_slots.restart_lsn vs pg_current_wal_lsn()
2. Set retention limits: Configure max_slot_wal_keep_size (PostgreSQL 13+)
3. Handle stale slots: Drop unused slots with pg_drop_replication_slot('slot_name')

-- Check slot status and lag
SELECT slot_name, 
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as lag
FROM pg_replication_slots;

Troubleshooting

Connection Issues

If you see authentication errors:

-- Verify user has replication privilege
SELECT rolname, rolreplication FROM pg_roles WHERE rolname = 'deltaforge';

-- Check pg_hba.conf allows replication connections
-- Ensure the line type includes "replication" database

Missing Before-Images

If UPDATE/DELETE events have incomplete before data:

-- Check current replica identity
SELECT relname, relreplident 
FROM pg_class 
WHERE relname = 'your_table';
-- d = default, n = nothing, f = full, i = index

-- Set to FULL for complete before-images
ALTER TABLE your_table REPLICA IDENTITY FULL;

Slot/Publication Not Found

-- List existing publications
SELECT * FROM pg_publication;

-- List existing slots
SELECT * FROM pg_replication_slots;

-- Create if missing
CREATE PUBLICATION my_pub FOR TABLE public.orders;
SELECT pg_create_logical_replication_slot('my_slot', 'pgoutput');

WAL Disk Usage Growing

-- Check slot lag
SELECT slot_name, 
       active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as lag
FROM pg_replication_slots;

-- If slot is inactive and not needed, drop it
SELECT pg_drop_replication_slot('unused_slot');

Logical Replication Not Enabled

-- Check wal_level
SHOW wal_level;  -- Should be 'logical'

-- If not, update postgresql.conf and restart PostgreSQL
-- wal_level = logical