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:

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:

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 ALTERs 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

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').

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:

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:

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)

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

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.

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.

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

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:

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:

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:

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:

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:

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:

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.