Postgres CDC to ClickHouse

CDC from Postgres order_events into one PicoMQ stream per region, then one ClickHouse table.

Source: examples/connectors/postgres-cdc-clickhouse.

Run

cd examples/connectors/postgres-cdc-clickhouse
docker compose up -d
./seed.sh
./verify.sh

Service	Host	Auth
Postgres (pgweb)	http://localhost:8082	pico / pico
PicoMQ	http://localhost:9090
Connectors runtime	http://localhost:8081
ClickHouse	http://localhost:8123/play	pico / pico

Connectors

Source, postgres-source.toml:

[[topics]]
topic = { strategy = "field", path = "data.region", template = "orders.{value}" }
create_topics = true

[plugin_config]
mode = "cdc"
tables = ["public.order_events"]
replication_slot = "picomq_orders"
capture_operations = ["INSERT", "UPDATE"]

Sink, clickhouse-sink.toml:

[[topics]]
pattern = "orders\\..*"

[transforms.unwrap_envelope]
enabled = true
field = "data"

[plugin_config]
table = "order_events"
insert_format = "json_each_row"

The ClickHouse table is not created by the sink. sql/clickhouse.sql creates it at container start.

table	Result
order_events	One table, region as a column
order_events_{topic_segment[-1]}	order_events_eu, order_events_na, order_events_apac

Verify

	Postgres	ClickHouse	PicoMQ
Counts	SELECT region, count(*) FROM order_events GROUP BY region	SELECT region, uniqExact(id) FROM order_events GROUP BY region	pico ls --prefix /orders.
EU rows	SELECT * FROM order_events WHERE region = 'eu'	SELECT * FROM order_events WHERE region = 'eu'	pico read /orders.eu

Resume

docker compose kill connectors
docker compose exec postgres psql -U pico -d example -c \
  "INSERT INTO order_events (order_id, region, merchant_id, \"type\", payload)
   VALUES ('ord-eu-resume', 'eu', 'lemongrass', 'placed', '{}'::jsonb);"
docker compose start connectors
./verify.sh

The row inserted while the runtime was down arrives after restart.