Postgres

Logical replication over pgoutput, via pgx. One replication connection per pipeline, on a logical decoding slot.

slotName is required — the slot is the consistency anchor.

Connection

Two ways to configure the connection — mutually exclusive:

URI (default)

URI connection
source:
  kind: postgres
  uri: postgres://user:password@host:port/database?sslmode=disable
  slotName: my_slot

Structured fields

Structured connection with TLS/SSH
source:
  kind: postgres
  slotName: my_slot
  postgres:
    host: db.example.com
    port: 5432
    database: mydb
    username: reader
    password: secret
    params:
      application_name: urutau
    maxThreads: 10
    retryCount: 3
    cdc:
      plugin: pgoutput
      initialWaitTime: 300
    ssl:
      mode: verify-full
      ca: /etc/ssl/ca.pem
      cert: /etc/ssl/client-cert.pem
      key: /etc/ssl/client-key.pem
    ssh:
      host: bastion.example.com
      port: 22
      username: ubuntu
      privateKey: /home/user/.ssh/id_ed25519

When source.postgres is set, source.uri is ignored for connection building. slotName and snapshotUri remain flat regardless.

TLS (`ssl`)

Mode	Behavior
`disable` (default)	No TLS
`require`	TLS, skip certificate verification
`verify-ca`	TLS, verify chain against CA but not hostname
`verify-full`	TLS, verify chain and hostname

cert + key enable mutual TLS (client certificate). Both must be set together.

SSH tunnel (`ssh`)

Tunnels the connection through an SSH bastion. Supports password and private key authentication. The tunnel is established once per connection and reused.

Connection tuning

Field	Default	Description
`maxThreads`	`runtime.NumCPU()`	Max concurrent connections for snapshot chunk SELECTs (1..32), and the size of the concurrent row-normalization pool
`retryCount`	3	Transient-error retries with exponential backoff: snapshot queries are retried, and a lost replication stream reconnects and resumes from the committed position. 0 means "use the default"
`cdc.plugin`	`pgoutput`	Logical decoding plugin: `pgoutput` (binary) or `wal2json` (JSON). wal2json must be installed on the server; the slot is created with the selected plugin
`cdc.initialWaitTime`	300	Seconds the CDC reader waits for the first WAL message before failing with a non-retryable error (minimum 30). Detects a misconfigured slot or publication that would otherwise hang forever; the timer is satisfied by the first WAL data message
`schemas`	all accessible	Limits `discover` to these schemas
`discover`	`false`	Replicates every table the user may `SELECT` (base tables and partitioned parents) instead of an explicit `tables` list. Mutually exclusive with `tables`

Discovery

See Table discovery below — discover: true replaces the explicit tables list. schemas limits which schemas are scanned.

Distributed mode

In distributed mode the worker opens the snapshot chunk SELECT from this block. ssl.ca, ssl.cert and ssl.key are sent as paths, so every worker must mount those files at the same paths as the coordinator. An ssh block is shipped to the worker in its Assignment — a DSN cannot carry a tunnel — so the worker also needs the private key at the configured path (see the Kubernetes guide).

Table discovery

Instead of listing every table, discover: true replicates all tables the connected user may SELECT. The spec then omits tables entirely — discovery and an explicit tables list are mutually exclusive (declaring both is a validation error).

Table discovery
pipeline: shop
source:
  kind: postgres
  slotName: shop_slot
  postgres:
    host: db.example.com
    database: shop
    username: repl
    password: secret
    discover: true
    schemas: [public, analytics]   # optional; omit to scan every accessible schema
sink:
  uri: http://polaris:8181/api/catalog
  namespace: raw
  warehouse: quickstart_catalog

There is no tables: block. Every discovered table is written to <sink.namespace>.<table name>: public.orders → raw.orders, analytics.events → raw.events.

What is discovered — and what is not

Included	Excluded	Why
Base tables (`relkind = 'r'`)	Leaf partitions (`relispartition`)	The partitioned parent already holds every leaf's rows; replicating both would write each row twice.
Partitioned parents (`relkind = 'p'`)	Materialized views (`'m'`)	They cannot take `REPLICA IDENTITY FULL` or join a publication.
	Foreign tables (`'f'`)	Same, and their rows live on another server.
	`pg_*`, `information_schema`	System catalogs.

A table is included only when the connected user has SELECT on the table and USAGE on its schema.

Target names

The source schema is dropped, not prefixed — the target is <sink.namespace>.<table name>. So public.orders and analytics.orders both derive raw.orders: a collision, which is a boot error, not a silent overwrite. To replicate two schemas that share a table name, use an explicit tables list with distinct targets (discovery off).

Primary keys

Discovery declares no keys. For each discovered table the source reads the table's actual primary key from the catalog and replicates it in the default upsert mode. A table without a primary key has no equality key: the pipeline fails at boot with a clear error rather than writing an unkeyed upsert. Replicate such a table with writeMode: append and an explicit tables list (which means discovery off for that pipeline). A partitioned parent needs a primary key on the parent, as usual.

Re-discovery on every start

Discovery runs at each start, before the snapshot and the publication sync:

A table added since the last run is discovered, added to the publication, and snapshotted (it has no committed position yet).
A table removed is dropped from the publication.
A table whose schema changed is re-introspected.

No restart flag is needed — the discovered set is recomputed on every boot.

Partitioned tables and CDC

A discovered partitioned parent is replicated as one logical table. Its live changes are published as the parent (the publication is created with publish_via_partition_root = true), which requires PostgreSQL 13+. On an older server a partitioned table is rejected at setup, rather than silently dropping its live stream.

Distributed (Kubernetes) mode

A discovery pipeline has no tables at operator time, so the operator cannot render a per-table worker Pod template. It renders one generic template instead, and the coordinator clones it for every discovered target. Nothing extra to configure — see the Kubernetes guide.

Incremental mode

A table can sync by cursor column instead of the replication log: it reads SELECT ... WHERE <cursor> > <last> ORDER BY <cursor> on every boot and stores the last cursor value. No replication slot, no publication — useful for slowly changing tables or servers where logical replication is not available.

Incremental mode
tables:
  - source: public.accounts
    target: raw.accounts
    primaryKey: [id]
    mode: incremental
    cursor: updated_at

The cursor column must be NOT NULL; the pass reads past the stored value.
Rows are upserted (op: insert) with __phase: incremental. Deletes are not detected — an incremental read only sees rows that still exist.
The cursor is the table's committed cdc.position, written in the same commit as the rows, so a restart resumes exactly where it left off. The resume predicate is >=, so a non-unique cursor (e.g. updated_at) never drops a row that shares the last value.
Incremental and CDC tables can share one pipeline: the slot covers only the CDC tables. Incremental mode is currently supported in the collapsed runner (not distributed mode).

Requirements

wal_level=logical.
Enough max_replication_slots and max_wal_senders for the pipeline's slots.
A user with REPLICATION (and rights to create the publication/slot).

On start the runner makes the server side ready, idempotently: it sets REPLICA IDENTITY FULL on every replicated table (so updates and deletes carry the full old row, matching MySQL's row_image=FULL), creates the logical publication <slotName>_pub listing exactly the pipeline's tables, and creates the pgoutput slot — before the snapshot begins, so no transaction between slot creation and the stream start is lost.

Behavior

Position — an LSN, tracked through the slot. The reader reports the pipeline's minimum committed position back to the server as confirmed_flush, so the slot never discards WAL for events still in flight to the sink. A restart resumes from the slot.
Snapshot — the worker runs the chunk SELECT with snapshotUri when set, so it can use a SELECT-only user (the replication credential stays coordinator-side).
Before image — deletes and updates carry the old row (PK-only unless REPLICA IDENTITY FULL, which the runner sets).
Snapshot consistency — each chunk runs in a REPEATABLE READ READ ONLY transaction, so the chunk sees one consistent snapshot even under concurrent writes.
Snapshot chunking — the default is physical CTID block ranges: no primary key required, uniform chunks regardless of key skew, sized from sink.defaults.targetFileSize (default 512Mi) divided by the server block size. Range-partitioned tables are split proportionally across their leaf partitions. Set chunkColumn to chunk by a key column instead (value-range for integer/float, cursor stepping otherwise).
Worker partitioning — workers: {number: N > 1} splits a single-column primary key into N contiguous ranges that govern both the snapshot and the live stream, so a key never changes owner. CTID is not routable, so a partitioned table is always chunked by its key.
Column projection — columnFilter narrows the snapshot SELECT list and the CDC projection; the excluded columns are absent from the target. It must include the primary key.
Row filter — filter is pushed into the snapshot WHERE and evaluated on each CDC row before the Arrow hot-path. A row that leaves the filter produces a delete, so an upsert target drops the stale row.

Example

URI-based

URI-based pipeline
pipeline: shop
source:
  kind: postgres
  uri: postgres://repl:replpass@postgres:5432/shop?sslmode=disable
  slotName: urutau_shop
sink:
  uri: http://polaris:8181/api/catalog
  namespace: raw
  warehouse: quickstart_catalog
tables:
  - source: public.orders
    target: raw.orders
    primaryKey: [id]
    createIfNotExists: true

Structured config with TLS and SSH

Structured config with TLS and SSH
pipeline: shop
source:
  kind: postgres
  slotName: urutau_shop
  postgres:
    host: postgres.internal
    port: 5432
    database: shop
    username: repl
    password: replpass
    ssl:
      mode: verify-full
      ca: /etc/ssl/ca.pem
    ssh:
      host: bastion.example.com
      username: ubuntu
      privateKey: /home/user/.ssh/id_ed25519
    maxThreads: 10
    retryCount: 3
    cdc:
      plugin: pgoutput
      initialWaitTime: 300
sink:
  uri: http://polaris:8181/api/catalog
  namespace: raw
  warehouse: quickstart_catalog
tables:
  - source: public.orders
    target: raw.orders
    primaryKey: [id]
    createIfNotExists: true

Connection​

URI (default)​

Structured fields​

TLS (ssl)​

SSH tunnel (ssh)​

Connection tuning​

Discovery​

Distributed mode​

Table discovery​

What is discovered — and what is not​

Target names​

Primary keys​

Re-discovery on every start​

Partitioned tables and CDC​

Distributed (Kubernetes) mode​

Incremental mode​

Requirements​

Behavior​

Example​

URI-based​

Structured config with TLS and SSH​