pgsync-clickhouse

Real-time Postgres → ClickHouse sink for PGSync.

Reuses PGSync's WAL reader, checkpointing, schema tree and change fan-out; swaps the Elasticsearch/OpenSearch sink for a ClickHouse one via PGSync's Sync.make_search_client() hook (see pgsync.sink.Sink).

Status: working vertical slice (mirror + denormalized). Shipped so far:

• types, PostgreSQL → ClickHouse type mapping.
• ddl, mirror-mode CREATE TABLE … ReplacingMergeTree DDL generator (versioned upserts + tombstone deletes).
• reflect, builds Table/Column specs from SQLAlchemy reflection.
• client, ClickHouseClient (implements pgsync.sink.Sink): translates PGSync actions into versioned inserts / tombstones.
• settings, ClickHouse connection config from env.
• sync, ClickHouseSync(Sync): wires the sink in via make_search_client(), reflects the schema into one table (mirror or denormalized), and supplies _version = source WAL LSN.
• cli, pgsync-clickhouse command (bootstrap / sync / daemon / wal / teardown).

Verified end-to-end by the smoke tests (real Postgres + ClickHouse + Redis) across thirteen scenarios: mirror (single table), schema evolution (additive ALTER), engine control (custom PARTITION BY/SETTINGS), denormalized (parent + children), deep denormalization (grandchildren fold into nested tuples), TRUNCATE propagation (native TRUNCATE TABLE), the type roundtrip matrix (20 column types survive PG→CH), enum / column_types override (a user enum pinned to Enum8), exactly-once on replay (versioned idempotency), the incremental WAL consumer (CDC tail via slot decode), the streaming WAL consumer (live CDC via wal_consumer()), the trigger + Redis default path (pg_notify → Redis queue → consumer), and the stoppable daemon. Insert → update (higher _version wins) → delete (tombstone drops the row under FINAL); one-to-one children flattened into typed columns and one-to-many children folded into native Array(Tuple(...)) columns (queryable like a ClickHouse Nested); a child update fanning out to the wide row; and changes reaching ClickHouse via all three of PGSync's change-capture mechanisms.

Getting started, the book example

PGSync ships a canonical book database (see its examples/book). Here's how to mirror the book table into ClickHouse in real time. This walks through single-table (mirror) mode, the nested book → authors/publisher/… denormalized wide table is denormalized mode, covered in Denormalized (multi-table) below.

1. Have the pieces running

• A PGSync book Postgres database (follow PGSync's examples/book to create and seed it).

• A ClickHouse instance, local Docker is fine:

  docker run -d --name ch -p 8123:8123 \
    -e CLICKHOUSE_USER=book -e CLICKHOUSE_PASSWORD=book \
    -e CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1 \
    clickhouse/clickhouse-server:24.8-alpine
  

• Install this package (pulls in pgsync): pip install pgsync-clickhouse.

2. Point at both databases

# Postgres source (same PG_* vars PGSync already uses)
export PG_USER=postgres PG_PASSWORD=postgres PG_HOST=localhost PG_PORT=5432
# ClickHouse sink
export CLICKHOUSE_HOST=localhost CLICKHOUSE_PORT=8123
export CLICKHOUSE_USER=book CLICKHOUSE_PASSWORD=book CLICKHOUSE_DATABASE=default

3. Use the mirror schema, examples/book.json:

[
  {
    "database": "book",
    "index": "book",
    "clickhouse": {
      "order_by": ["id"],
      "partition_by": "id % 8",
      "settings": "index_granularity = 8192"
    },
    "nodes": {
      "table": "book",
      "columns": ["id", "isbn", "title", "description", "publish_date"]
    }
  }
]

The "clickhouse" block is optional (omit it for the default ReplacingMergeTree ORDER BY (id)); it's shown here to demonstrate per-table engine control, see Custom ClickHouse engine / partitioning.

4. Bootstrap, then sync

pgsync-clickhouse -c examples/book.json --bootstrap   # replication slot + ClickHouse `book` table
pgsync-clickhouse -c examples/book.json               # initial sync (one pass)
pgsync-clickhouse -c examples/book.json --daemon      # keep it live (continuous CDC)

--bootstrap reflects the source book table and issues the CREATE TABLE book (…, _version UInt64, _is_deleted UInt8) ENGINE = ReplacingMergeTree(_version, _is_deleted) PARTITION BY id % 8 ORDER BY (id) SETTINGS index_granularity = 8192 in ClickHouse (the PARTITION BY / ORDER BY / SETTINGS come from the optional "clickhouse" block).

5. Query ClickHouse

-- easiest: the companion _live view already applies FINAL + tombstone filter
SELECT count() FROM book_live;
SELECT title, isbn FROM book_live LIMIT 5;

-- equivalently, against the base table:
-- FINAL collapses versions; filter tombstones for the live view
SELECT count() FROM book FINAL WHERE _is_deleted = 0;
SELECT title, isbn FROM book FINAL WHERE _is_deleted = 0 LIMIT 5;

The connector creates a <index>_live view for every table (… FINAL WHERE _is_deleted = 0 over the data columns) so consumers get the clean, deduplicated, tombstone-free state without knowing the ReplacingMergeTree mechanics. It's on by default, set "live_view": false in the schema's clickhouse block to skip it, or "live_view_suffix" to rename it. FINAL does merge-on-read work; for hot paths on very large tables set "live_view_strategy": "argmax", which builds the view as SELECT pk…, argMax(col, _version) … GROUP BY pk HAVING argMax(_is_deleted, _version) = 0, the same clean state, without FINAL.

Now insert / update / delete a book in Postgres and re-run the query, the row converges in ClickHouse (an update appends a higher _version that wins under FINAL; a delete writes a tombstone that _is_deleted = 0 filters out).

A TRUNCATE of the root table is propagated as a native ClickHouse TRUNCATE TABLE, O(1), rather than a tombstone per row, so the target is physically emptied. (Note: in the default trigger mode, PGSync core currently leaves TRUNCATE notifications without an index and filters them out before the sink; TRUNCATE propagation therefore applies to the WAL paths today.)

Scope note. Bootstrap + initial sync is the fully verified path. --daemon uses PGSync's standard change capture (inherited unchanged). The ClickHouse table contains exactly the columns you declare in "columns" (plus the primary key, always kept for the ORDER BY).

Schema evolution

When you add a column to the source (and to "columns" in schema.json), just re-run --bootstrap. The connector reconciles the existing ClickHouse table additively:

ALTER TABLE book ADD COLUMN IF NOT EXISTS <new_col> <type>

Existing rows keep their default for the new column until they next sync. Reconciliation is additive only, a column you remove from the schema is left in place in ClickHouse (logged, not dropped), and a type change is not auto-migrated; both are destructive and left to you to do deliberately.

Custom ClickHouse engine / partitioning

By default the target is ReplacingMergeTree(_version, _is_deleted) ORDER BY (<pk>). Add an optional top-level "clickhouse" object to a schema doc to control the engine, sort key, partitioning, TTL and settings:

{
  "database": "book",
  "index": "book",
  "clickhouse": {
    "engine": "ReplacingMergeTree",
    "order_by": ["id"],
    "partition_by": "id % 8",
    "ttl": "created_at + INTERVAL 2 YEAR",
    "settings": "index_granularity = 8192"
  },
  "nodes": { "table": "book", "columns": ["id", "title", "created_at"] }
}

For a replicated cluster, name the replicated engine and pass its path/replica as engine_params (the connector appends _version, _is_deleted):

"clickhouse": {
  "engine": "ReplicatedReplacingMergeTree",
  "engine_params": ["'/clickhouse/tables/{shard}/book'", "'{replica}'"]
}

Two hard rules the connector enforces / relies on:

• The engine must be a ReplacingMergeTree family engine, the versioned upsert / tombstone mechanic depends on it (validated; a non-Replacing engine is rejected).
• order_by must include every primary-key column (it is the dedup key; validated), and partition_by must be computable from the primary key. Deletes arrive as PK-only tombstones and ReplacingMergeTree dedups only within a partition, so id % 8 or intDiv(id, 1000000) are safe but toYYYYMM(created_at) would send the tombstone to the wrong partition and the delete would never take effect.

Type mapping

Each Postgres column is reflected to a native ClickHouse type: integers → IntN, boolean → Bool, real/double → Float32/64, numeric(p,s) → Decimal(p,s) (unqualified numeric → Decimal(38, 9)), uuid → UUID, date → Date32, timestamp/timestamptz → DateTime64(6, 'UTC') (UTC-pinned for portability; a naive timestamp is treated as UTC), text/char → String, and Postgres arrays (integer[]) → Array(...). json/jsonb and specialized types with no better ClickHouse scalar (hstore, ranges/multiranges, interval, money, xml, tsvector, geometric, PostGIS, …) map to String, a loss-free textual mirror. Values are coerced to the right native type on insert (ISO strings → date/datetime, arrays stay native lists, numeric → Decimal).

A user-defined enum or domain can't be recognized by type name, so it falls back to String (never dropped) and the connector logs one startup warning naming those columns. To pin such a column to a specific ClickHouse type, add a column_types override to the schema's clickhouse block (emitted verbatim, including any Nullable wrapping):

"clickhouse": {
  "column_types": {
    "mood": "Nullable(Enum8('happy' = 1, 'sad' = 2, 'meh' = 3))",
    "tag":  "LowCardinality(Nullable(String))"
  }
}

An overridden column is no longer flagged as unmapped, and its value (the enum label string / the text) inserts as-is.

Denormalized (multi-table)

Add children to the schema and the sink builds one wide table, the root's typed columns plus, per direct child:

• a one-to-one object child → flattened typed columns <label>_<col> (native, analytics-friendly), or
• a one-to-many object child → a native Array(Tuple(...)) column named <label>, typed from the child's columns (queryable like a ClickHouse Nested, <label>.<col> element access, ARRAY JOIN to explode to rows).

Object grandchildren fold recursively (no depth limit): a one-to-one grandchild flattens into its parent, at the top level as more <label>_<col> columns, or inside a nested array's tuple as prefixed fields; a one-to-many grandchild becomes a nested Array(Tuple(...)) within the parent's tuple. So order_items → products yields items Array(Tuple(qty …, product_sku String, product_name String …)), queried two levels deep as items.product_sku (or via ARRAY JOIN). A child with a scalar variant, or a subtree with nothing foldable, falls back to a JSON String column (which preserves the subtree).

See examples/book_denormalized.json: book with a publisher (one-to-one → flattened to publisher_id / publisher_name) and authors (one-to-many via book_author → Array(Tuple(...))). A change to a child row re-emits the root document, so the wide row stays in sync.

-- flattened one-to-one column, plus native nested-array access / explode
SELECT title, publisher_name, authors.name AS author_names
FROM book_wide FINAL
WHERE _is_deleted = 0;

SELECT title, a.name AS author
FROM book_wide FINAL ARRAY JOIN authors AS a
WHERE _is_deleted = 0;

Smoke tests

Thirteen scenarios drive real Postgres → ClickHouse syncs through ClickHouseSync and assert rows land / update / tombstone correctly: mirror, schema evolution (additive ALTER ADD COLUMN), engine control (custom PARTITION BY/SETTINGS), denormalized, deep denormalization (grandchildren fold into nested tuples), TRUNCATE propagation (a source TRUNCATE empties the target via native TRUNCATE TABLE), the type roundtrip matrix (a 20-column table, integers/bool/floats/numeric/uuid/date/timestamp/ jsonb/native arrays/specialized-as-String, mirrored and read back intact), enum / column_types override (a user enum pinned to a ClickHouse Enum8), exactly-once on replay (re-applying a change at the same _version is idempotent; a stale lower-version replay is ignored), the incremental WAL consumer (commits after the initial load reach ClickHouse by decoding the replication slot), the streaming WAL consumer (commits made while wal_consumer() streams flow through live), trigger + Redis (DML fires pg_notify, a producer queues to Redis, a consumer syncs), all three of PGSync's change-capture mechanisms, and the stoppable daemon (receive()/stop()).

Two runners, same scenarios:

smoke/native.sh   # uses your locally-installed Postgres + ClickHouse + Redis (no Docker)
smoke/smoke.sh    # spins the stack up/down via docker compose

native.sh needs a local Postgres with wal_level = logical, a local ClickHouse, and a local Redis (for the trigger path); put ClickHouse credentials in smoke/.env.local (gitignored, copy smoke/.env.local.example). It creates a throwaway smoke database, resets the target ClickHouse tables and the Redis queue between scenarios, and tears everything down on exit.

CLI

Installs a pgsync-clickhouse command mirroring pgsync's workflow (same schema.json):

pgsync-clickhouse -c schema.json --bootstrap   # create slot + ClickHouse table
pgsync-clickhouse -c schema.json               # initial sync (one pass)
pgsync-clickhouse -c schema.json --daemon      # initial + continuous trigger+Redis CDC
pgsync-clickhouse -c schema.json --wal         # WAL-streaming CDC
pgsync-clickhouse -c schema.json --teardown    # drop slot + ClickHouse table

--daemon runs the trigger+Redis pipeline on background threads and shuts down gracefully on Ctrl-C/SIGTERM (workers drain and join, then the sink closes), needs Redis. --wal streams the replication slot directly (no Redis, no triggers).

Daemon knobs: --num-workers N runs N concurrent consumer threads, each with its own ClickHouse connection (out-of-order writes are safe under _version dedup); --health-interval S logs sink metrics every S seconds. The daemon also watches worker liveness, if a worker thread crashes it shuts down and exits non-zero, so a supervisor (systemd/k8s) restarts a clean process instead of running half-dead.

Postgres source overrides (-h/--host, -p/--port, -u/--user, --password, --sslmode, --sslrootcert) match pgsync.

Configuration

Point at ClickHouse via a DSN or discrete vars (read lazily from the environment):

Env var	Default	Notes
CLICKHOUSE_URL / CLICKHOUSE_DSN		full DSN; takes precedence
CLICKHOUSE_HOST	localhost
CLICKHOUSE_PORT	driver default	8123 (http) / 8443 (tls)
CLICKHOUSE_USER	default
CLICKHOUSE_PASSWORD	(empty)
CLICKHOUSE_DATABASE	default	qualifies the target table
CLICKHOUSE_SECURE	false	TLS
CLICKHOUSE_VERIFY	true	verify the server cert (TLS only)
CLICKHOUSE_CA_CERT		CA bundle path to trust (TLS only)
CLICKHOUSE_CHUNK_SIZE	1000	insert batch size
CLICKHOUSE_CONNECT_RETRIES	3	retries on a transient/connection error
CLICKHOUSE_RETRY_BACKOFF	0.5	initial backoff seconds (doubles each retry)
CLICKHOUSE_RETRY_MAX_BACKOFF	30	backoff cap

Postgres source connection uses PGSync's own PG_* settings, unchanged.

Resilience & exactly-once

Every ClickHouse call (DDL, DESCRIBE, insert) is wrapped with retry + reconnect: a transient network/connection error is retried with exponential backoff, reconnecting a dropped connection between attempts; query/type errors are not retried. Retrying an insert is safe, _version is the source WAL LSN, so a re-applied change re-inserts an identical (key, _version) row that ReplacingMergeTree collapses under FINAL. The same property makes crash replay idempotent (PGSync reprocessing already-applied changes can't create duplicate live rows or resurrect deletes), and a late, lower-version replay is ignored. The sink exposes metrics via ClickHouseClient.metrics(), counters (insert_batches, rows_written, retries, reconnects) plus gauges rows_per_sec (windowed since the previous sample) and applied_version (the highest source WAL LSN written). ClickHouseSync.metrics() adds source_version (the live source LSN) and wal_lag (how far ClickHouse trails the source). --health-interval logs them periodically, and they're logged on daemon shutdown.

Modes

• Mirror (free), single-node schema → one ClickHouse table of the declared columns.
• Denormalized (paid), multi-node schema → one wide table: root columns, plus one-to-one children flattened into typed <label>_<col> columns and one-to-many children folded into native Array(Tuple(...)) columns (queryable like a ClickHouse Nested). A scalar-variant child, or one with its own children, falls back to a JSON String column. The differentiator.

Benchmarks

bench/bench.sh measures the two numbers that matter, backfill throughput (rows/sec for the initial sync) and CDC apply latency (p50/p95/max to decode and apply one WAL change), against a local Postgres + ClickHouse, reading the sink's own metrics gauges:

BENCH_ROWS=1000000 BENCH_MODE=both bench/bench.sh   # mirror + denormalized
BENCH_CDC=0 BENCH_ROWS=5000000 bench/bench.sh       # backfill only, at scale

It benchmarks this connector only, not an apples-to-apples comparison with other pipelines (which would need all of them on shared infrastructure). See bench/README.md for the methodology and caveats.

Develop

pip install -e ".[dev]"
pytest

Or with uv (faster; a committed uv.lock pins the exact dependency set, the same one CI uses):

uv sync --extra dev   # create .venv + install from uv.lock
uv run pytest

After changing dependencies in pyproject.toml, run uv lock and commit the updated uv.lock (CI runs uv sync --locked and fails if it drifted).

The DDL generator and type mapper are dependency-free and DB-free, so the test suite runs with no Postgres or ClickHouse.

Continuous integration

.github/workflows/ci.yml runs three jobs on push / PR: lint (ruff, format check + lint/import sorting), unit (pytest across Python 3.10–3.12), and smoke (the full suite against Postgres + ClickHouse + Redis via docker compose). All three install via uv sync --locked from the committed uv.lock, so every run resolves the identical dependency set, pgsync (>=7.3.0) is an ordinary PyPI dependency.

License

This package (pgsync-clickhouse, mirror mode) is MIT licensed; see LICENSE. Denormalized (multi-table) mode is a separate commercial package, pgsync-clickhouse-pro, licensed proprietarily; see pgsync.com/clickhouse.