# Postgres To Clickhouse Integration

> PostgreSQL → ClickHouse Data Integration

- Skill: `mikerrr/postgres-to-clickhouse-integration` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add mikerrr/postgres-to-clickhouse-integration`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mikerrr/postgres-to-clickhouse-integration/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: mikerrr (https://skillmd.com/u/mikerrr)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/mikerrr/postgres-to-clickhouse-integration

---


# PostgreSQL → ClickHouse Data Integration

I design and execute the full path from a transactional PostgreSQL database to an analytical ClickHouse deployment: live pass-through querying, mirrored tables and databases, persisted MergeTree tables, and continuous Change Data Capture. My guiding architecture is: **PostgreSQL stays the transactional source of truth for row-based OLTP operations; ClickHouse becomes the analytical "speed layer" delivering millisecond aggregations over column-oriented storage.**

Before starting, I consult `references/reference.md` for the complete type-mapping table, the exhaustive pushdown rules, the expected benchmark figures, the CDC comparison matrix, and the full catalogue of pitfalls.

## Instructions

### Step 0: Verify prerequisites

I confirm all of the following before touching data:

1. **ClickHouse target** — a running self-managed cluster or ClickHouse Cloud service. A Cloud *development* service (up to 1 TB storage, 16 GB total memory, e.g. 8 GB RAM / 2 cores per node) is sufficient for datasets in the tens of millions of rows; a production service works equally well.
2. **PostgreSQL source** — self-managed, a managed service (AWS RDS, Google Cloud SQL, Azure Database), or a hosted platform such as Supabase. On Supabase free tier I note the **500 MB database size limit** and the **global 120 s / 2-minute statement timeout**.
3. **Network reachability** — host, port (default `5432`), database name, user, password. I place both systems **in the same cloud region**; streaming performance is bandwidth/latency-bound, and cross-region setups can dominate total query time.
4. **Postgres privileges** — `SELECT` on the tables to read; for CDC/replication additionally the `REPLICATION` attribute and the right to create publications; for the `PostgreSQL` *database* engine, DDL rights (`ALTER`/`DROP COLUMN`) if I intend to modify Postgres tables through ClickHouse.
5. **Clients** — `psql` for Postgres (a web SQL editor also works) and a ClickHouse client (`clickhouse-client`, the Cloud SQL console, or a programmatic client such as Airflow's `ClickHouseHook`).
6. **Schema knowledge** — the source DDL, column types and **cardinalities** (needed for correct ClickHouse type mapping and primary-key design).
7. **Data loaded** into Postgres (e.g. a Postgres-compatible SQL dump restored with `psql`), plus any Postgres indexes I intend to exploit for filter pushdown.
8. **For logical-replication CDC** — PostgreSQL 9.4+; 10+ strongly preferred so the built-in `pgoutput` logical decoding plugin is available (below 10 I install `wal2json` or `decoderbufs` manually). I need edit access to `postgresql.conf` (`wal_level = logical`) and `pg_hba.conf` (a replication host line), plus permission to restart/reload Postgres.
9. **For `MaterializedPostgreSQL`** — the ClickHouse setting `allow_experimental_database_materialized_postgresql = 1` and a Postgres instance configured for logical replication.
10. **Optional** — an orchestrator (e.g. Airflow with a ClickHouse connection/hook) if the copy must be repeatable, and the target role/user that will receive `SELECT` grants after loading.
11. **Integration pattern decided up front** — pass-through querying, one-off/batched persistence, or continuous CDC.

---

## Part A — Prepare and evaluate the Postgres side

### Step 1: Create the source table and indexes in PostgreSQL

Reference schema I use throughout:

```sql
CREATE TABLE uk_price_paid
(
   id integer primary key generated always as identity,
   price INTEGER,
   date Date,
   postcode1 varchar(8),
   postcode2 varchar(3),
   type varchar(13),
   is_new SMALLINT,
   duration varchar(9),
   addr1 varchar(100),
   addr2 varchar(100),
   street varchar(60),
   locality varchar(35),
   town varchar(35),
   district varchar(40),
   county varchar(35)
);
```

I add the indexes my analytical filters could use:

```bash
psql -c "CREATE INDEX ON uk_price_paid (type)"
psql -c "CREATE INDEX ON uk_price_paid (town)"
psql -c "CREATE INDEX ON uk_price_paid (extract(year from date))"
```

### Step 2: Benchmark the candidate analytical queries in Postgres first

I enable timing (`psql -c "\timing"`), run each query **at least 5 times and report the fastest run** so both systems are compared "hot" with warm file-system caches, and I inspect `EXPLAIN` for every query.

### Step 3: Interpret the Postgres plans to predict pushdown behaviour

- **Low-cardinality filters cause sequential scans.** `type` has only 5 distinct values; `type='flat'` matches ~6.3M of 34M rows (~1/6). With ~24 rows per block almost every block contains a match, so the planner chooses a **parallel sequential scan** over the index. The yearly-average query over `type='flat'` took **~28.5 s**.
- **Selective filters do use indexes.** Filtering `town='BRISTOL'` and grouping by `postcode1` ran in **~543 ms** via the `town` index. Changing the literal (e.g. to `LONDON`) can flip the planner back to a sequential scan depending on matching row counts.
- **Combined selective predicates can use a bitmap scan** over several indexes: the London 2002-vs-2022 median comparison used a bitmap scan on both the `town` and `extract(year from date)` indexes and finished in **~8.9 s**.
- **I never distort the Postgres schema just to force index usage** (heavy denormalization, extra helper columns). It wastes storage, is a poor use of Postgres, and such `ALTER`s can exceed a managed service's 120 s statement timeout.

---

## Part B — Query Postgres data from ClickHouse without persisting it

### Step 4: Choose the access mechanism

| Mechanism | What it does | When I pick it |
|---|---|---|
| `postgresql(...)` **table function** | Opens a connection per query and streams rows into ClickHouse | Ad-hoc analysis; joining small Postgres datasets to local ClickHouse tables |
| `PostgreSQL` **table engine** | Mirrors a single Postgres table as a ClickHouse table; identical implementation and pushdown behaviour to the function, but trivial query syntax | Repeated access to one table |
| `PostgreSQL` **database engine** | Mirrors an entire Postgres database, exposing all its tables; also allows DDL that modifies/drops columns in the underlying Postgres | Whole-schema read-through access |

In ClickHouse Cloud the table function and the table engine are available.

### Step 5: Use the table function

Signature: `postgresql('host:port', 'database', 'table', 'user', 'password')`.

```sql
SELECT toYear(date) AS year, round(avg(price)) AS price
FROM postgresql('db.example.supabase.co', 'postgres', 'uk_price_paid', 'postgres', '<password>')
WHERE type = 'flat'
GROUP BY year ORDER BY year ASC;
```

Row filtering happens **in Postgres where possible**; aggregations, JOINs, sorting and `LIMIT` are **always executed in ClickHouse**.

### Step 6: Write WHERE clauses that can be pushed down

Only simple comparisons are pushed down: `=`, `!=`, `>`, `>=`, `<`, `<=`, `IN`.

Any ClickHouse-specific function in the predicate (e.g. `toYear(date) = 2002`) **blocks pushdown**, so the Postgres `extract(year from date)` index cannot be used and the whole table is streamed. I rewrite such predicates as native comparisons on the raw column (e.g. a `date` range) or knowingly accept the full scan.

### Step 7: Minimize the number of separate queries sent to Postgres

Each call opens a new connection and re-reads data. I replace JOINs of two Postgres subqueries with a **single scan plus conditional aggregate combinators**.

Slow variant — two streams of the same table, **~59.9 s**:

```sql
SELECT med_2002.postcode1, median_2002, median_2022,
       round(((median_2022 - median_2002) / median_2002) * 100) AS percent_change
FROM (SELECT postcode1, median(price) AS median_2002 FROM postgresql(...) WHERE town='LONDON' AND toYear(date)='2002' GROUP BY postcode1) AS med_2002
INNER JOIN (SELECT postcode1, median(price) AS median_2022 FROM postgresql(...) WHERE town='LONDON' AND toYear(date)='2022' GROUP BY postcode1) AS med_2022
  ON med_2002.postcode1 = med_2022.postcode1
ORDER BY percent_change DESC LIMIT 10;
```

Faster, simpler rewrite with `medianIf` — single read, **~36.2 s**:

```sql
SELECT postcode1,
       medianIf(price, toYear(date) = 2002) AS median_2002,
       medianIf(price, toYear(date) = 2022) AS median_2022,
       round(((median_2022 - median_2002) / median_2002) * 100) AS percent_change
FROM postgresql(...)
WHERE town = 'LONDON'
GROUP BY postcode1 ORDER BY percent_change DESC LIMIT 10;
```

I always balance "fewer Postgres queries" against "keep predicates pushdownable so Postgres indexes reduce the streamed volume".

### Step 8: Create a mirror table with the PostgreSQL table engine (inferred types)

```sql
SET external_table_functions_use_nulls = 0;  -- represent NULLs as column default values
CREATE TABLE uk_price_paid_postgresql AS postgresql('db.example.supabase.co', 'postgres', 'uk_price_paid', 'postgres', '<password>');
SHOW CREATE TABLE uk_price_paid_postgresql;  -- inspect the mapped types
```

With the default `external_table_functions_use_nulls = 1`, ClickHouse wraps **every** column in `Nullable(...)` — which degrades performance and changes semantics. Typical inferred mapping: `integer` → `Int32`, `smallint` → `Int16`, `varchar(n)` → `String`, `date` → `Date`.

### Step 9: Or declare the mirror table explicitly with tighter types

```sql
CREATE TABLE default.uk_price_paid_v2
(
  `price` UInt32,
  `date` Date,
  `postcode1` String,
  `postcode2` String,
  `type` Enum8('other'=0,'terraced'=1,'semi-detached'=2,'detached'=3,'flat'=4),
  `is_new` UInt8,
  `duration` Enum8('unknown'=0,'freehold'=1,'leasehold'=2),
  `addr1` String, `addr2` String, `street` String, `locality` String,
  `town` String, `district` String, `county` String
)
ENGINE = PostgreSQL('db.example.supabase.co', 'postgres', 'uk_price_paid', 'postgres', '<password>');
```

---

## Part C — Mirror a whole Postgres database

### Step 10: Read-through mirror with the `PostgreSQL` database engine

Parameters: `host:port`, database, user, password, schema, and the flag enabling use of tables from that schema.

```sql
CREATE DATABASE postgres_analytic_datamarts2
ENGINE = PostgreSQL('10.10.2.74:5432', 'analytic', 'm_rubinov', 'pass', 'datamarts', 1);
```

Every Postgres table then becomes addressable as `postgres_analytic_datamarts2.<table>`.

### Step 11: Continuous replication with the experimental `MaterializedPostgreSQL` database engine

I enable `allow_experimental_database_materialized_postgresql = 1` and treat this engine as **experimental, not production-grade**. Its lifecycle:

- On creation it takes a **snapshot** of the PostgreSQL database and loads the required tables (any subset of tables from any subset of schemas of the specified database).
- Together with the snapshot it acquires an **LSN**; once the initial dump completes it starts pulling updates from the **WAL**.
- Tables created in PostgreSQL **after** the ClickHouse database was created are **not** replicated automatically — I attach each manually: `ATTACH TABLE db.table`.
- Replication uses the PostgreSQL Logical Replication Protocol, which **cannot replicate DDL** but does signal replication-breaking changes (column type change, adding/removing columns). Affected tables stop receiving updates; I recover them with `DETACH PERMANENTLY` followed by `ATTACH` to reload the table completely.
- Non-breaking DDL (e.g. renaming a column) leaves replication working, because inserts are applied **by column position, not by name** — which also means reordering columns can silently mis-map data.

---

## Part D — Persist data into a native ClickHouse MergeTree table

### Step 12: Create the target MergeTree table

I choose analytics-optimized types and an `ORDER BY` (primary key) matching real filter/grouping patterns, ordered **from lowest to highest cardinality**.

```sql
CREATE TABLE default.uk_price_paid
(
  `price` UInt32,
  `date` Date,
  `postcode1` LowCardinality(String),
  `postcode2` LowCardinality(String),
  `type` Enum8('other'=0,'terraced'=1,'semi-detached'=2,'detached'=3,'flat'=4),
  `is_new` UInt8,
  `duration` Enum8('unknown'=0,'freehold'=1,'leasehold'=2),
  `addr1` String, `addr2` String,
  `street` LowCardinality(String), `locality` LowCardinality(String),
  `town` LowCardinality(String), `district` LowCardinality(String), `county` LowCardinality(String)
)
ENGINE = MergeTree
ORDER BY (type, town, postcode1, postcode2);
```

Rules I apply: `LowCardinality(String)` for repeated string dimensions, `Enum8` for small fixed value sets, and drop unneeded Postgres surrogate keys via `SELECT * EXCEPT id`.

### Step 13: Load the data

```sql
INSERT INTO uk_price_paid SELECT * EXCEPT id
FROM postgresql('db.example.supabase.co', 'postgres', 'uk_price_paid', 'postgres', '<password>');
```

### Step 14: Chunk the load if the Postgres service enforces a statement timeout

With Supabase's global 2-minute limit, a single bulk insert aborts (e.g. after ~21.58M rows / ~121 s) with:

```
pqxx::sql_error: Failure during '[END COPY]': ERROR: canceling statement due to statement timeout
```

I split the load on a column of appropriate cardinality whose filter **is pushed down**, so each chunk finishes inside the timeout:

```sql
INSERT INTO uk_price_paid SELECT * EXCEPT id FROM postgresql(...) WHERE type = 'other';
INSERT INTO uk_price_paid SELECT * EXCEPT id FROM postgresql(...) WHERE type = 'detached';
INSERT INTO uk_price_paid SELECT * EXCEPT id FROM postgresql(...) WHERE type = 'flat';
INSERT INTO uk_price_paid SELECT * EXCEPT id FROM postgresql(...) WHERE type = 'terraced';
INSERT INTO uk_price_paid SELECT * EXCEPT id FROM postgresql(...) WHERE type = 'semi-detached';
```

Self-managed or unrestricted instances usually do not need this workaround.

### Step 15: Repeatable copy from a mirrored database into a local MergeTree table

Idempotent, orchestrator-friendly pattern — drop, recreate from the mirror's schema, insert, grant:

```python
clickhouse_connection = 'analytic-clickhouse1'
clickhouse_database = 'datamarts'

clh_hook = ClickHouseHook(clickhouse_conn_id=clickhouse_connection, database=clickhouse_database)
clh_client = clh_hook.get_conn()
clh_client.execute('DROP TABLE datamarts.ewa_for_cso_billing')
clh_client.execute('CREATE TABLE IF NOT EXISTS datamarts.ewa_for_cso_billing AS postgres_analytic_datamarts2.ewa_for_cso_billing Engine=MergeTree() ORDER BY tuple()')
clh_client.execute('INSERT INTO datamarts.ewa_for_cso_billing SELECT * FROM postgres_analytic_datamarts2.ewa_for_cso_billing')
clh_client.execute('GRANT SELECT ON datamarts.ewa_for_cso_billing TO superset_crystalball')
```

I use `ORDER BY tuple()` **only** when no meaningful sorting key exists; otherwise I always choose a real `ORDER BY`.

### Step 16: Re-benchmark against the local MergeTree table

Expected orders of magnitude on the 28M-row dataset:

| Query | Postgres | ClickHouse MergeTree | Rows scanned |
|---|---|---|---|
| Yearly average, `type='flat'` | ~28.5 s | **~0.079 s** | 5.01M |
| Bristol postcode averages | ~543 ms | **~0.077 s** | 27.69M |
| London median 2002 vs 2022 (`medianIf`) | ~8.9 s | **~0.062 s** | 2.62M |

I add sanity filters such as `postcode1 != ''` to exclude empty dimension values from ranked results.

---

## Part E — Change Data Capture from PostgreSQL

### Step 17: Choose the CDC approach

Three options exist; **log-based logical replication is the recommended default**. Full comparison matrix in `references/reference.md`. I never rely on batch full-extract pipelines instead of CDC — they are inefficient, error-prone, and let downstream systems drift out of sync.

### Step 18: Trigger-based CDC

Captures INSERT/UPDATE/DELETE instantly and stores events in a Postgres audit table.

- Install the community generic audit trigger function (PostgreSQL 9.1+), which writes all change events into `audit.logged_actions`.
- Enable per table: `SELECT audit.audit_table('public.users');`
- **Advantages:** instant capture; all three event types; rich metadata by default (the statement that caused the change, transaction ID, session user name).
- **Disadvantages:** triggers lengthen the original statement's execution time and hurt Postgres performance; they require schema changes inside Postgres; a separate pipeline must still poll `audit.logged_actions` to ship events onward; extra operational complexity.

### Step 19: Query-based CDC (polling on a modification timestamp)

- Requires a column such as `updated_at` recording the last modification time.
- Poll periodically, remembering the previous high-water mark:

```sql
SELECT * FROM public.users WHERE updated_at > 'TIMESTAMP_LAST_QUERY';
```

- **Advantage:** no changes to Postgres needed if the timestamp column already exists.
- **Disadvantages:** goes through the query layer and adds load to Postgres; wastes resources when data rarely change; requires the timestamp column; **cannot capture DELETEs** unless the application uses soft deletes.

### Step 20: Log-based CDC via logical replication (PostgreSQL 9.4+, recommended)

a) Enable logical replication in `postgresql.conf` — `wal_level = logical` — and **restart Postgres**. Replication is *not* enabled by default.

b) Allow replication connections in `pg_hba.conf`, for example:

```
host     all     repuser     0.0.0.0/0     md5
```

I tighten the CIDR and auth method for production per the PostgreSQL documentation — I never ship `0.0.0.0/0` verbatim.

c) Ensure a logical decoding plugin is available: PostgreSQL 10+ ships `pgoutput` by default; for versions older than 10 I install `wal2json` or `decoderbufs` manually.

d) Create a publication for the tables of interest:

```sql
CREATE PUBLICATION newpub FOR TABLE public.users;
```

e) Create the subscription/consumer. A subscription begins with an **initial snapshot** and then streams all incremental changes. From another PostgreSQL instance:

```sql
CREATE SUBSCRIPTION newsub CONNECTION 'dbname=foo host=bar user=repuser' PUBLICATION newpub;
```

For non-Postgres consumers I use an established open-source CDC implementation (e.g. Debezium) or a purpose-built Postgres→ClickHouse replication tool (e.g. PeerDB) rather than writing a decoder myself; inside ClickHouse the equivalent is the `MaterializedPostgreSQL` database engine from Step 11.

f) Verify the managed provider supports logical replication — AWS RDS, Google Cloud SQL and Azure Database all do.

**Advantages:** real-time, event-driven capture; detects INSERTs, UPDATEs **and** DELETEs; consumes the WAL from the file system so it does not load the query layer.

### Step 21: Document and hand over the architecture

I state explicitly: Postgres remains the transactional source of truth for row-level operations; ClickHouse serves complex aggregations at millisecond scale; data flows one way via `INSERT INTO ... SELECT` batches or CDC; ad-hoc and small joins may still be served live via the `postgresql()` function.

---

## Success criteria

I consider the work correctly completed when:

1. **Connectivity works** — a `SELECT` through the `postgresql()` function or the PostgreSQL table engine returns the same rows and identical aggregate values as the same query run directly in `psql` (the yearly average price series and the top-10 postcode lists match exactly).
2. **Types are mapped as intended** — `SHOW CREATE TABLE` on the mirrored table shows the expected ClickHouse types, with no unwanted `Nullable(...)` wrappers when `external_table_functions_use_nulls = 0` was set, and passwords displayed as `[HIDDEN]`.
3. **Filter pushdown is confirmed** — queries with simple predicates (`=`, `!=`, `>`, `>=`, `<`, `<=`, `IN`) on indexed Postgres columns stream only a small subset of rows (visible in ClickHouse's "Processed N rows" statistics), while unfiltered or ClickHouse-function-filtered queries stream the whole table.
4. **Full load completed without error** — the `INSERT INTO ... SELECT ... FROM postgresql(...)` (or the chunked variants) finished with no statement-timeout exception, and `SELECT count()` in ClickHouse equals the Postgres row count (all chunks summed, e.g. all 28M rows).
5. **Analytical performance improved by orders of magnitude** — the queries that took ~28.5 s, ~0.5 s and ~8.9 s in Postgres run in well under 0.1 s against MergeTree, scanning only primary-key-selected rows (5.01M / 27.69M / 2.62M rows processed).
6. **The MergeTree target is well designed** — `ORDER BY` reflects real query filters (low → high cardinality), `LowCardinality`/`Enum8` applied to repeated dimensions, no unnecessary surrogate `id` column.
7. **Downstream access granted** — e.g. `GRANT SELECT ON db.table TO <role>`, and BI/reporting tools can query the table.
8. **CDC verified** — after an INSERT, UPDATE and DELETE in the monitored Postgres table, the corresponding change events appear downstream in real time (in `audit.logged_actions` for the trigger approach, or through the publication/subscription for logical replication); DELETEs are present, proving trigger- or log-based CDC rather than query-based.
9. **`MaterializedPostgreSQL` verified** — the initial snapshot is present, subsequent WAL changes appear in ClickHouse, and any newly created Postgres table has been explicitly attached with `ATTACH TABLE db.table`.
10. **Benchmarks measured fairly** — each query executed 5 times with the fastest run reported, and hardware/resource differences between the two systems explicitly noted.

---

## Troubleshooting

**Problem: `pqxx::sql_error: Failure during '[END COPY]': ERROR: canceling statement due to statement timeout` during the bulk load.**
The managed Postgres killed the read mid-flight (Supabase aborts at ~120 s, e.g. after ~21.58M rows). I split the load into chunks on a column of suitable cardinality whose filter is pushed down — e.g. one `INSERT ... WHERE type = '<value>'` per distinct `type` value — so each chunk completes inside the timeout, then verify `SELECT count()` in ClickHouse equals the Postgres row count. Alternatively I raise or disable the statement timeout on a self-managed instance.

**Problem: a query against `postgresql()` is far slower in ClickHouse than the same query in Postgres.**
The predicate is almost certainly not pushed down, so the whole table is streamed over the network (e.g. ~59.9 s in ClickHouse vs ~8.9 s in Postgres). I check for ClickHouse-specific functions inside `WHERE` (`toYear(date) = 2002` blocks pushdown; only `=`, `!=`, `>`, `>=`, `<`, `<=`, `IN` are pushed down), rewrite them as raw-column comparisons or date ranges, collapse JOINs of two `postgresql()` subqueries into a single scan with `-If` combinators such as `medianIf`, and confirm both systems sit in the same cloud region — cross-region latency and bandwidth can dominate the runtime.

**Problem: replication from `MaterializedPostgreSQL` silently stopped for one table, or a new Postgres table never appears.**
Logical replication does not replicate DDL. A replication-breaking change (column type change, adding/removing columns) halts updates for that table — I recover it with `DETACH PERMANENTLY` and then `ATTACH` to reload it completely. Newly created Postgres tables are never auto-discovered; I add each one with `ATTACH TABLE db.table`. If no data flows at all, I verify `wal_level = logical` in `postgresql.conf`, the replication line in `pg_hba.conf`, that Postgres was restarted, that a decoding plugin exists (`pgoutput` on 10+, otherwise `wal2json`/`decoderbufs`), and that `allow_experimental_database_materialized_postgresql = 1` is set.