You have an application database and you want a copy of one of its tables somewhere else, for reporting. Copying the whole table every night works until the table gets big, or until you need the deletes too.

The usual answer is change data capture: read the inserts, updates and deletes from PostgreSQL's write-ahead log and apply them to the copy. The usual tool for that is Debezium, which typically runs on Kafka Connect. That's a lot of infrastructure for one mirror table.

PostgreSQL has had everything needed built in since version 10: logical replication, a publication that says which tables to follow, and a replication slot that remembers how far a reader got. Hydra ETL 0.20 adds a source that uses exactly that, with nothing to install on the server.

The interesting part isn't reading the changes. It's what happens when writing them fails halfway.

Everything below runs from one repository: the database, its creation and seed scripts, the Hydra ETL job and the SQL for every step. You only need Docker and Python.

Setup 

git clone https://github.com/bejaouibechir/hydra-cdc-mirror-demo.git cd hydra-cdc-mirror-demo pip install "hydra-etl[postgres]>=0.20" ./scripts/start.sh # Windows PowerShell: .\scripts\start.ps1

start launches PostgreSQL 16 in Docker with wal_level=logical (the only server setting CDC needs), on port 5433 so it doesn't clash with a PostgreSQL you may already run. At its first start, the container runs db/initdb.sql and db/seed.sql: two databases, shop (the application) and analytics (the mirror), and 1,000 orders in shop.orders. Prefer a zip to git? The repository has one, next to the code.

The connection settings live in .env, shared by Docker and Hydra ETL. They are named DEMO_PG_* on purpose: a PG_PORT or PG_USER already set on your machine can't interfere with the demo.

The Hydra ETL job is two files. The source reads the changes of public.orders:

# jobs/orders_mirror/sources.yaml sources: order_changes: type: postgresql_cdc connection: host: "{{ env:DEMO_PG_HOST }}" port: "{{ env:DEMO_PG_PORT }}" user: "{{ env:DEMO_PG_USER }}" # needs the REPLICATION attribute password: "{{ env:DEMO_PG_PASSWORD }}" database: shop slot: hydra_orders # created at the first run publication: hydra_orders # created at the first run extract: table: public.orders

The destination applies them to the mirror:

# jobs/orders_mirror/destinations.yaml destinations: mirror: type: postgresql connection: host: "{{ env:DEMO_PG_HOST }}" port: "{{ env:DEMO_PG_PORT }}" user: "{{ env:DEMO_PG_USER }}" password: "{{ env:DEMO_PG_PASSWORD }}" database: analytics load: table: orders mode: upsert # inserts and updates are upserted, deletes are applied key: [order_id] atomic: true # all or nothing

First run: the whole table 

hdrctl run . ./scripts/sql.sh check # PowerShell: .\scripts\sql.ps1 check

The first run creates the publication and the slot, then copies the whole table. The mirror table is created with the source's types (bigint, numeric, jsonb, timestamptz) and order_id as primary key.

  • shop.orders: 1,000 rows, total 453855.00
  • analytics.orders: 1,000 rows, total 453855.00

A day of changes

db/changes.sql adds 20 orders, refunds 3 and deletes 1:

INSERT INTO orders (order_id, customer_id, amount, status, details) SELECT g, 1000 + g % 50, 42.00, 'paid', '{"channel": "web"}' FROM generate_series(1001, 1020) g; UPDATE orders SET status = 'refunded', updated_at = now() WHERE order_id IN (5, 17, 230); DELETE FROM orders WHERE order_id = 999; 
./scripts/sql.sh changes hdrctl run . ./scripts/sql.sh check

This time Hydra ETL reads only the 24 changes, in the order of the transactions:

  • shop.orders: 1,019 rows, 3 refunded, total 454582.31
  • analytics.orders: 1,019 rows, 3 refunded, total 454582.31

Each change arrives as a row with three extra columns: _op (insert, update, delete), _lsn (its position in the log) and _commit_time. In upsert, the PostgreSQL destination reads _op: the last change of each key wins, deletes are applied by key, and the three columns aren't written. In append mode they are written, which gives you an audit log of every change instead of a mirror.

Now the failure

db/break.sql gives the mirror a rule the application doesn't have, no negative amounts, then records 500 legitimate updates and one bad order in the shop:

-- in analytics ALTER TABLE orders ADD CONSTRAINT amount_not_negative CHECK (amount >= 0); -- in shop UPDATE orders SET amount = amount + 1 WHERE order_id <= 500; INSERT INTO orders (order_id, customer_id, amount, status) VALUES (1021, 1001, -15.00, 'paid'); 
./scripts/sql.sh break hdrctl run . 
✗ Step 'orders' FAILED: … CheckViolation: new row for relation "orders" violates check constraint "amount_not_negative" 
./scripts/sql.sh check
  • shop.orders: 1,020 rows, total 455067.31
  • analytics.orders: 1,019 rows, total 454582.31, exactly as before the run

None of the 500 updates was applied. That part is atomic: true: the load is one transaction.

But a CDC pipeline has a second way to lose data. If the source had marked those 501 changes as "read" before the load failed, the next run would start after them, and the 500 updates would be gone for good, without any error.

That's why Hydra ETL reads the changes without consuming them (pg_logical_slot_peek_binary_changes), and only moves the slot forward (pg_replication_slot_advance) after the destination has published. The failed run didn't move the slot.

db/repair.sql fixes the bad order at the source:

UPDATE orders SET amount = 15.00 WHERE order_id = 1021; 
./scripts/sql.sh repair hdrctl run . ./scripts/sql.sh check
  • shop.orders: 1,020 rows, total 455097.31
  • analytics.orders: 1,020 rows, total 455097.31

The 500 updates from the failed run were read again and applied, together with the new ones. The order that was inserted at -15.00 and then corrected arrives in the mirror at 15.00: within a batch, only the last change of each key is written.

Want to start over? ./scripts/reset.sh removes the container and its data and creates them again.

Details that matter in practice

  • Unchanged large values. When an UPDATE doesn't touch a large column (a big JSON document stored out of line, "TOAST"), PostgreSQL doesn't put its value in the log. A naive mirror would write NULL there. Hydra ETL keeps the value the mirror already has.
  • Primary key. The source table needs one, so that updates and deletes can be applied by key.
  • Unused slots. A slot keeps the server's write-ahead log until it's read. If you stop the pipeline for good, drop it: SELECT pg_drop_replication_slot('hydra_orders');.
  • Rollback. A workflow with rollback: true refuses a CDC source: the slot is confirmed at the end of the job, and a later rollback couldn't move it back. atomic: true is the right tool here.
  • Your own server. No Docker? The repository's README shows how to run the same scripts on any PostgreSQL 11 or later with wal_level = logical.

Links

Disclosure: I'm the author of Hydra ETL. It's a batch tool: each run applies the changes since the last one, it isn't a streaming platform. For one or a few mirror tables, it's a lot less to run than a Kafka cluster.