Metadata-Version: 2.4
Name: pgsync-clickhouse
Version: 0.1.0
Summary: Real-time Postgres -> ClickHouse sink for PGSync (mirror mode; denormalized mode via pgsync-clickhouse-pro)
Author: Tolu Aina
License: MIT
Project-URL: Homepage, https://pgsync.com/clickhouse
Project-URL: Documentation, https://pgsync.com/clickhouse
Project-URL: Source, https://github.com/toluaina/pgsync
Keywords: postgres,postgresql,clickhouse,cdc,change-data-capture,replication,etl,pgsync
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pgsync>=7.3.0
Requires-Dist: SQLAlchemy>=2.0
Requires-Dist: clickhouse-connect>=0.7
Requires-Dist: click>=8.1
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Dynamic: license-file

# pgsync-clickhouse

![CI](https://github.com/toluaina/pgsync-clickhouse/actions/workflows/ci.yml/badge.svg)
![Python](https://img.shields.io/badge/python-3.10%2B-blue)
![License](https://img.shields.io/badge/license-MIT-green)

Real-time **Postgres → ClickHouse** sink for [PGSync](https://github.com/toluaina/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)](#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:

  ```bash
  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**

```bash
# 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`](examples/book.json):

```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](#custom-clickhouse-engine--partitioning).

**4. Bootstrap, then sync**

```bash
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**

```sql
-- 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:

```json
{
  "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`):

```json
"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):

```json
"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`](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.

```sql
-- 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:

```bash
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`):

```bash
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:

```bash
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`](bench/README.md) for the methodology and caveats.

## Develop

```bash
pip install -e ".[dev]"
pytest
```

Or with [uv](https://docs.astral.sh/uv/) (faster; a committed `uv.lock` pins the
exact dependency set, the same one CI uses):

```bash
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`](LICENSE). Denormalized (multi-table) mode is a separate commercial
package, `pgsync-clickhouse-pro`, licensed proprietarily; see
[pgsync.com/clickhouse](https://pgsync.com/clickhouse).
