[emoss08]

What does this PR do?

Mirrors each source table into a ClickHouse ReplacingMergeTree, keyed by the source primary key and versioned by the event's LSN. Target tables are created from the source catalog on first sight, so there is no target DDL to write:

CREATE TABLE gtc.orders (
    `id` Int32,
    `total` Nullable(Decimal(10, 2)),
    `notes` Nullable(String),
    `_version` UInt64,              -- the event's LSN
    `_deleted` UInt8 DEFAULT 0,     -- tombstone marker
    `_synced_at` DateTime64(3) DEFAULT now64(3)
) ENGINE = ReplacingMergeTree(`_version`, `_deleted`)
ORDER BY (`id`)

Current state is SELECT * FROM gtc.orders FINAL WHERE _deleted = 0.

Each design decision follows from a guarantee GTC already makes:

• Versioned by LSN — a redelivered event carries the same version, so replays are idempotent, and a live change always outranks a backfilled one for the same row. Backfill and streaming interleave safely.
• Deletes become tombstones, so a late-arriving delete cannot resurrect a row.
• Unchanged TOAST columns are carried forward. ReplacingMergeTree replaces whole rows, so an update that left a large column untouched would otherwise blank it; the stored value is read back and rewritten.
• Batched inserts without weakening delivery. A change stream is exactly the many-small-inserts pattern ClickHouse dislikes. Rather than batch in-process — which would let the WAL position advance past undelivered rows — inserts use ClickHouse's server-side async batching and wait for the flush.
• DDL follows automatically. The sink implements ports.SchemaObserver, so a column added upstream is added to the mirror. Dropped and retyped columns are logged but not applied: the mirror holds history the source no longer has, and discarding it silently would be worse than a warning.

Column types come from introspecting the source catalog, since CDC events carry values but not types (the sink keeps its own non-replication connection for this). Mapping is exact where ClickHouse has an equivalent — int4→Int32, numeric(10,2)→Decimal(10, 2), timestamptz→DateTime64(6, 'UTC'), uuid→UUID — and falls back to String, preserving PostgreSQL's exact text, for unconstrained numeric, arrays, JSON and user-defined types.

Type of change

• New feature / sink

Checklist

• go test ./... passes (including -race)
• gofmt / go vet are clean
• Tests added or updated for changed behavior — unit tests for type mapping, DDL generation, value coercion, tombstone zero values and target naming; the integration test asserts the mirror's contents (live rows, tombstone, DDL-added column) and reads GTC_TEST_CLICKHOUSE_URL, which CI now provides with a readiness wait
• README.md / CLAUDE.md updated (new "ClickHouse mirror" section, all env vars, config/sinks.example.yaml)
• If this touches delivery semantics: at-least-once holds by default — inserts wait for the server-side flush before success, so the WAL position never advances past unpersisted rows. CLICKHOUSE_WAIT_FOR_INSERT=false deliberately trades that for throughput and is warned about both at startup and by gtc doctor.

Notes for the reviewer

Verified end to end against a real ClickHouse 26.8 and PostgreSQL 16, not a mock: auto-created schema, backfill, a TOAST-preserving update (a 5000-character column survived an update that touched only a sibling column), delete tombstones, and ALTER TABLE ... ADD COLUMN reaching the mirror. The integration test ran green four times in a row.

Tables without a primary key are skipped, with an error naming them, while the rest keep mirroring — without a key there is no ORDER BY, so row versions cannot be collapsed and deletes cannot address a row. Failing the whole pipeline over one unmirrorable table would be worse.

TRUNCATE is not mirrored. Emptying the target would race with rows still in flight, so it is logged as a warning and the target keeps the truncated rows. Worth a follow-up if you want it handled.

🤖 Generated with Claude Code

https://claude.ai/code/session_019dEj24Ta2o9QxzW9ZrSPtb

---

Generated by Claude Code