1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# End-to-end Postgres CDC -> Postgres upsert mirror, exactly-once.
#
# Streams logical-replication change events from a source database, normalizes
# the CDC envelope with `cdc_unwrap`, and mirrors them into a destination table
# with `write_mode: upsert` — inserts/updates become UPSERTs and deletes become
# row deletes, so the mirror table stays an exact replica of the source table.
#
# Setup on the SOURCE database (logical replication must be enabled):
# psql "$SOURCE_PG_URL" <<SQL
# CREATE TABLE IF NOT EXISTS users (id int4 PRIMARY KEY, name text);
# CREATE PUBLICATION faucet_pub FOR TABLE users;
# SQL
#
# Setup on the DEST database (upsert/delete need a UNIQUE/PRIMARY KEY on `key`):
# psql "$DEST_PG_URL" <<SQL
# CREATE TABLE IF NOT EXISTS users_mirror (id int4 PRIMARY KEY, name text);
# SQL
#
# Then:
# export SOURCE_PG_URL=postgres://faucet:faucet@localhost:5432/appdb
# export DEST_PG_URL=postgres://faucet:faucet@localhost:5432/warehouse
# faucet run cli/examples/postgres_cdc_to_postgres_upsert.yaml
#
# INSERT/UPDATE/DELETE rows in the source `users` table from another psql
# session — every change is mirrored into `users_mirror` each fetch cycle.
version: 1
name: pg_cdc_mirror
# Exactly-once delivery composes a CDC source + an idempotent SQL sink + a state
# store with no DLQ: each committed batch and its monotonic commit token land in
# one transaction, so a crash/resume never re-applies or skips a batch.
delivery: exactly_once
pipeline:
source:
type: postgres-cdc
config:
connection_url: ${env:SOURCE_PG_URL}
slot_name: faucet_mirror
publication_name: faucet_pub
create_slot_if_missing: true
idle_timeout: 30
# Normalize the postgres-cdc envelope ({op, before, after}) into a flat row
# plus a `__op` marker. Inserts/updates emit the post-image; deletes emit the
# key from the pre-image. DDL/truncate events are dropped.
transforms:
- type: cdc_unwrap
sink:
type: postgres
config:
connection_url: ${env:DEST_PG_URL}
table_name: users_mirror
# Upsert/delete require column-mapping mode (not the JSONB blob mode).
column_mapping: auto_map
max_connections: 5
# Insert-or-update by `id` (the mirror table needs a UNIQUE/PRIMARY KEY on
# it); rows whose `__op` is "d" are routed to DELETE instead.
write_mode: upsert
key: [id]
delete_marker: { field: __op, values: [d] }
state:
type: file
config:
path: ./state