ClickHouse for Analytics Workloads

ClickHouse is the fastest analytical database I have used in production. That sentence requires qualification: fastest for the workloads it was designed for — wide tables, aggregations over hundreds of millions of rows, time-series scans, high-cardinality GROUP BY operations — and genuinely terrible for everything else. The speed is not magic. It is the result of a coherent set of architectural decisions that trace back to one foundational choice: column-oriented storage. Understanding why that choice matters, and what trade-offs it forces, is the prerequisite for using ClickHouse well.

This guide is written from production experience. It covers the internal storage model, the MergeTree engine family in depth, the subtle but critical distinction between sorting key and primary key, materialized views for pre-aggregation, query profiling tools, replication with ClickHouse Keeper, distributed sharding, and an honest assessment of when you should reach for ClickHouse versus when you should stay with PostgreSQL.

As of this writing, the current ClickHouse LTS release is 25.8, with the latest stable release at 26.3. The version numbering follows a year.month scheme — 25.8 means the eighth release of 2025. ClickHouse Keeper is the recommended coordination layer for new deployments; ZooKeeper remains supported but is being phased out in favor of the native C++ implementation.

Why Columnar Storage Wins for Analytics

To understand columnar storage, start with how a row-oriented database like PostgreSQL stores data on disk. Each row is written as a contiguous sequence of bytes: all columns of row 1, then all columns of row 2, and so on. This layout is ideal for OLTP workloads. Fetching a single order by its primary key means reading one contiguous block from disk. Updating a row means overwriting one location. Inserting a row means appending bytes at the end of a heap file.

The problem appears when you run analytics queries. Consider this table:

orders(order_id, customer_id, product_id, quantity, unit_price, discount,
       shipping_cost, tax_amount, revenue, status, created_at, updated_at)

Now run: SELECT sum(revenue) FROM orders WHERE created_at >= '2025-01-01'

In a row-oriented layout, to compute that sum, the database must read every column of every qualifying row — order_id, customer_id, product_id, quantity, unit_price, discount, shipping_cost, tax_amount — even though only two columns matter (revenue and created_at). At 100 bytes per row and 500 million rows, you are reading 50 GB of data to extract maybe 3 GB of actual signal.

Row-Oriented Storage (PostgreSQL)
==================================
Block 1:
  [order_id=1][customer_id=42][product_id=7][quantity=2][unit_price=49.99]
  [discount=0.0][shipping_cost=5.99][tax_amount=4.00][revenue=99.98]
  [status="shipped"][created_at=2025-01-15][updated_at=2025-01-16]

  [order_id=2][customer_id=108][product_id=3][quantity=1][unit_price=19.99]
  [discount=2.00][shipping_cost=0.00][tax_amount=1.44][revenue=17.99]
  [status="pending"][created_at=2025-01-15][updated_at=2025-01-15]

  ... (more rows interleaved)

To compute sum(revenue): must read ENTIRE block including all unused columns
Bytes read for 2 rows: ~200 bytes
Bytes actually needed: ~16 bytes (revenue + created_at)

Column-Oriented Storage (ClickHouse)
======================================
revenue.bin:
  [99.98][17.99][34.50][129.00][8.99][55.49]...
  (all revenue values, contiguous, nothing else)

created_at.bin:
  [2025-01-15][2025-01-15][2025-01-16][2025-01-14][2025-01-16][2025-01-17]...
  (all timestamps, contiguous, nothing else)

order_id.bin:   (not touched for this query)
customer_id.bin: (not touched for this query)
product_id.bin:  (not touched for this query)
...

To compute sum(revenue): read revenue.bin and created_at.bin ONLY
I/O reduction: 10-column table = ~10x less data read

This I/O reduction compounds with compression. Columns contain values of the same type with often similar magnitudes — revenue values are all floating-point numbers in a similar range, timestamps cluster together, status is one of a handful of strings. This homogeneity compresses dramatically. ClickHouse achieves 5-15x compression ratios versus the 2-3x typical of row stores. LZ4 is the default codec — extremely fast decompression, good compression ratio. ZSTD trades some CPU for better compression (typically 30-40% better than LZ4 at similar decompression speeds), useful for archival data or when I/O is the bottleneck.

Beyond I/O, ClickHouse’s execution engine is designed around the columnar layout. Processing happens on vectors — contiguous arrays of column data — rather than row-by-row. Modern CPUs support SIMD (Single Instruction, Multiple Data) instructions that operate on 16, 32, or even 64 values simultaneously. When your revenue column arrives as a packed array of Float64 values, the CPU can sum 4 or 8 of them in a single instruction. This vectorized execution is why ClickHouse benchmarks show it processing billions of rows per second on commodity hardware.

None of this applies to OLTP workloads. If your application needs to fetch a single user’s profile, update an order status, or enforce foreign key constraints in a transaction, columnar storage is the wrong choice. Reading one row means touching every column file on disk. There are no real transactions. UPDATE and DELETE are implemented as background mutations, not in-place modifications. ClickHouse is an append-only system by design, and you should treat it that way.

ClickHouse Architecture

A single ClickHouse node is a single binary that runs everything: the query parser, planner, execution engine, storage engine, and embedded HTTP and native TCP servers. There is no separate process for storage, no buffer pool manager in the traditional sense. This simplicity makes single-node ClickHouse easy to operate — install the package, write a config file, start the service.

The fundamental storage unit is a part. When you execute INSERT INTO orders VALUES (...), ClickHouse does not write rows directly into an existing file. Instead, it creates a new part — an immutable directory on disk containing one binary file per column, plus an index file and a checksums file. Parts are small after an insert; they accumulate until the background merge process combines them into larger parts. This is conceptually similar to an LSM tree (Log-Structured Merge-tree), though the details differ significantly.

INSERT path:
  Client INSERT (batch of rows)
        |
        v
  New Part created (immutable directory)
  /var/lib/clickhouse/data/mydb/orders/
    20250115_1_1_0/          <- new part
      order_id.bin
      revenue.bin
      created_at.bin
      primary.idx
      checksums.txt

  Background merge picks up multiple small parts:
    20250115_1_1_0/
    20250115_2_2_0/
    20250115_3_3_0/
        |
        v (merge process)
    20250115_1_3_1/          <- merged part (spans rows 1-3, merge level 1)
      (all column files merged and re-sorted)

The merge runs continuously in the background. ClickHouse processes queries against the current set of parts (including unmerged ones) at all times — there is no maintenance window, no VACUUM, no manual optimization required for correctness. You can inspect the current parts with:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

SELECT
    name,
    rows,
    bytes_on_disk,
    bytes_on_disk / rows AS bytes_per_row,
    data_compressed_bytes,
    data_uncompressed_bytes,
    round(data_uncompressed_bytes / data_compressed_bytes, 2) AS compression_ratio
FROM system.parts
WHERE table = 'orders' AND active = 1
ORDER BY rows DESC;

The active = 1 filter shows only current (non-superseded) parts. After a merge completes, the source parts are marked inactive and eventually deleted. You can force a full merge with OPTIMIZE TABLE orders FINAL, which is occasionally useful in testing or after bulk loads, but should be avoided on production tables with tens of billions of rows — it will saturate I/O and potentially run for hours.

The MergeTree Engine Family

MergeTree is not one engine — it is a family of storage engines, each with different merge-time behavior. The base MergeTree does nothing special during merges except sort and compact data. The variants intercept the merge process to perform deduplication, pre-aggregation, or partial state accumulation.

MergeTree (Base Engine)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

CREATE TABLE orders (
    order_id     UInt64,
    customer_id  UInt64,
    product_id   UInt32,
    quantity     UInt16,
    revenue      Float64,
    status       LowCardinality(String),
    created_at   DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (customer_id, created_at)
SETTINGS index_granularity = 8192;

ORDER BY defines the sorting key — within each part, rows are physically sorted by these columns. PARTITION BY splits data into separate directories per month. The combination enables two tiers of pruning: partition pruning skips entire monthly directories if the query predicate excludes that month, and then the sparse primary index skips granules within the relevant partition.

The index is sparse, not a B-tree. ClickHouse stores one index entry per 8,192 rows (the index_granularity). When a query arrives with WHERE customer_id = 42, ClickHouse binary-searches the sparse index to find the granule range where customer_id=42 could appear, reads only those 8,192-row blocks, and discards rows that do not match. This sparse index is typically a few megabytes even for tables with billions of rows, fits entirely in memory, and is far more cache-friendly than a B-tree page structure.

ReplacingMergeTree

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

CREATE TABLE orders_latest (
    order_id    UInt64,
    customer_id UInt64,
    revenue     Float64,
    status      LowCardinality(String),
    updated_at  DateTime,
    version     UInt64
)
ENGINE = ReplacingMergeTree(version)
ORDER BY order_id;

During merges, ReplacingMergeTree deduplicates rows that share the same ORDER BY key, keeping only the row with the highest value of the optional version column (or the last-inserted row if no version column is specified). This makes it suitable for CDC (change data capture) patterns: write a new row for every update to an order, and let the engine converge on the latest state eventually.

The critical caveat: deduplication happens at merge time, not at insert time. Before a merge, a query may return multiple versions of the same order_id. You have two options for query-time correctness:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- Option 1: FINAL modifier -- forces merge before returning results
-- Correct, but significantly slower (disables parallel reads on each part)
SELECT order_id, revenue, status
FROM orders_latest FINAL
WHERE customer_id = 42;

-- Option 2: GROUP BY deduplication -- use when you control the query
SELECT
    order_id,
    argMax(revenue, updated_at)   AS revenue,
    argMax(status,  updated_at)   AS status
FROM orders_latest
WHERE customer_id = 42
GROUP BY order_id;

The argMax(value, ordering_column) idiom returns the value corresponding to the maximum of the ordering column, effectively picking the latest state. This pattern is faster than FINAL on large tables because it remains parallelizable across parts.

SummingMergeTree

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

CREATE TABLE pageview_counts (
    page_path   String,
    date        Date,
    hour        UInt8,
    views       UInt64,
    unique_ips  UInt64
)
ENGINE = SummingMergeTree((views, unique_ips))
PARTITION BY toYYYYMM(date)
ORDER BY (page_path, date, hour);

During merges, SummingMergeTree collapses rows with the same ORDER BY key by summing the specified numeric columns (or all numeric columns if none are specified). This is the simplest pre-aggregation pattern: instead of storing 1 billion raw pageview events, you store per-page per-hour counts. Queries that ask “how many views did /blog/post-1 get this month?” now scan a few thousand rows instead of millions.

As with ReplacingMergeTree, pre-aggregation is not instantaneous — it happens at merge time. Queries should always include a final SUM to handle the case where multiple parts have not yet been merged:

1
2
3
4
5
6
7
8

SELECT
    page_path,
    sum(views)      AS total_views,
    sum(unique_ips) AS total_unique_ips
FROM pageview_counts
WHERE date >= '2025-01-01' AND page_path LIKE '/blog/%'
GROUP BY page_path
ORDER BY total_views DESC;

AggregatingMergeTree

AggregatingMergeTree is the most powerful and most complex variant. Instead of summing integers, it stores partial aggregate states — intermediate objects that can be finalized by a merge function. This allows any aggregate function (count, avg, uniq, quantile, histogram) to be pre-computed incrementally.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

CREATE TABLE pageview_agg (
    page_path   String,
    date        Date,
    hour        UInt8,
    views_state AggregateFunction(count, UInt64),
    revenue_state AggregateFunction(sum, Float64),
    uniq_users_state AggregateFunction(uniqExact, UInt64)
)
ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (page_path, date, hour);

You almost never insert into an AggregatingMergeTree table directly. Instead, you pair it with a materialized view that uses *State aggregate functions to generate the partial states:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

CREATE MATERIALIZED VIEW pageview_agg_mv
TO pageview_agg
AS
SELECT
    page_path,
    toDate(created_at)   AS date,
    toHour(created_at)   AS hour,
    countState()         AS views_state,
    sumState(revenue)    AS revenue_state,
    uniqExactState(user_id) AS uniq_users_state
FROM pageviews
GROUP BY page_path, date, hour;

Queries against the AggregatingMergeTree use *Merge functions to finalize:

1
2
3
4
5
6
7
8
9

SELECT
    page_path,
    countMerge(views_state)     AS total_views,
    sumMerge(revenue_state)     AS total_revenue,
    uniqExactMerge(uniq_users_state) AS unique_users
FROM pageview_agg
WHERE date >= '2025-01-01'
GROUP BY page_path
ORDER BY total_views DESC;

This pattern scales to genuinely large datasets. The partial state for a uniqExact aggregate is a HyperLogLog sketch — it can represent millions of distinct values in a few kilobytes, and multiple sketches can be merged in constant time. You can aggregate a year of hourly data into a single result in milliseconds.

MergeTree Engine Family Reference

Primary Key vs. Sorting Key: The Critical Distinction

In PostgreSQL, the primary key is an immutable unique identifier — it drives a B-tree index, enforces uniqueness, and is used by foreign keys. In ClickHouse, the concepts split into two separate things that happen to be the same by default.

The sorting key (ORDER BY) defines the physical sort order of rows within each part. When ClickHouse writes a part to disk, it sorts all rows by the sorting key before writing the column files. This determines which range queries are fast: a query with WHERE customer_id = 42 AND created_at >= '2025-01-01' is a sorted range scan, blazing fast. A query with WHERE product_id = 99 on a table sorted by (customer_id, created_at) is a full scan — ClickHouse must read every granule.

The primary key (PRIMARY KEY) defines which columns are included in the sparse index. By default, the primary key equals the sorting key. But you can set them independently:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

CREATE TABLE orders (
    customer_id  UInt64,
    created_at   DateTime,
    order_id     UInt64,
    revenue      Float64,
    region       LowCardinality(String)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (customer_id, created_at, order_id)
PRIMARY KEY (customer_id, created_at);

Here, data is sorted by (customer_id, created_at, order_id) — the sort order includes order_id for uniqueness within a customer-day. But the sparse index only covers (customer_id, created_at) — the index is smaller, and order_id being in the sort key still guarantees efficient range scans on the first two columns. Adding order_id to the sort key (but not the primary key index) ensures rows within the same customer-day granule are consistently ordered, which can improve compression and scan predictability.

Practically speaking, the sorting key should be chosen based on your most common query patterns:

• WHERE clauses that use equality or range predicates on specific columns
• The leftmost columns are most important — ClickHouse uses a prefix of the sorting key for index lookups
• High-cardinality columns that appear often in WHERE clauses belong early in the key
• Time columns (created_at, event_date) often appear second, after a customer or entity ID, because most queries are “for customer X in time range Y”
• Do not put a UUID as the first column — UUIDs are essentially random, which destroys sort order and prevents any useful range scans

Sorting key cardinality principle:
  customer_id  ->  ~10M distinct values  (good first column: good range scan)
  created_at   ->  continuous range      (good second column: time ranges)
  order_id     ->  unique per row        (good for dedup; no range benefit)

  Queries that benefit from ORDER BY (customer_id, created_at):
    WHERE customer_id = X                        <- prefix match
    WHERE customer_id = X AND created_at > Y     <- prefix + range
    WHERE customer_id BETWEEN X AND Z            <- prefix range scan

  Queries that DO NOT benefit:
    WHERE created_at > Y (without customer_id)   <- not a prefix match
    WHERE revenue > 100                          <- revenue not in sort key

Data Types and Compression Codecs

Choosing the right data types is not a micro-optimization in ClickHouse — it directly affects storage size, compression ratio, and execution speed. The wrong types can double your storage footprint.

Integer types: ClickHouse has a full range — UInt8, UInt16, UInt32, UInt64, Int8 through Int64. Always use the smallest type that fits. A boolean stored as UInt8 (values 0 and 1) compresses to near nothing. An event_type stored as UInt32 when there are only 50 event types wastes three bytes per row.

LowCardinality(String): For any string column with fewer than roughly 10,000 distinct values — status, country_code, event_type, currency, device_type — use LowCardinality(String) instead of plain String. ClickHouse stores these as dictionary-encoded integers internally, reducing storage dramatically and enabling faster comparisons and GROUP BY operations. The dictionary is per-block, so cardinality should genuinely be low (the name is accurate).

Nullable(T): Avoid this when possible. Nullable types store a separate null bitmap alongside every row, add overhead to every aggregation, and disable certain optimizations. If you can represent “no value” with a sentinel (0 for integers, empty string, a Unix epoch of 0 for timestamps), prefer that.

DateTime vs DateTime64: DateTime stores Unix timestamps with second precision in 4 bytes. DateTime64(3) stores milliseconds in 8 bytes. Use DateTime where second precision is sufficient — most created_at columns, date partitions, hourly aggregations. Use DateTime64 for event logs where millisecond timestamps matter for ordering and analysis.

Per-column codecs allow you to specify the compression algorithm and any pre-processing transformation per column:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

CREATE TABLE metrics (
    host        LowCardinality(String),
    metric_name LowCardinality(String),
    ts          DateTime CODEC(Delta, LZ4),
    value       Float64  CODEC(Gorilla, LZ4),
    tags        String   CODEC(ZSTD(3))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(ts)
ORDER BY (host, metric_name, ts);

Delta pre-processes monotonic integer sequences (timestamps, sequence IDs) by storing differences rather than absolute values — timestamps that increment by 1-60 seconds become tiny differences that compress to near nothing. Gorilla is the Facebook Gorilla float compression algorithm, designed for time-series floating-point data with small delta-of-delta changes — typical metrics workloads. ZSTD(level) trades more CPU for better compression; level 3 is a reasonable balance. These codecs stack: CODEC(Delta, LZ4) applies Delta transformation first, then LZ4 compression.

A production-quality events table combining these techniques:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

CREATE TABLE events (
    event_id      UInt64         CODEC(Delta, LZ4),
    user_id       UInt64         CODEC(Delta, LZ4),
    session_id    UInt64,
    event_type    LowCardinality(String),
    page_path     String         CODEC(ZSTD(3)),
    referrer      String         CODEC(ZSTD(3)),
    country_code  LowCardinality(String),
    device_type   LowCardinality(String),
    revenue       Decimal(18, 4) CODEC(Gorilla, LZ4),
    occurred_at   DateTime64(3)  CODEC(Delta, LZ4),
    properties    String         CODEC(ZSTD(9))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(occurred_at)
ORDER BY (user_id, occurred_at, event_id)
PRIMARY KEY (user_id, occurred_at)
SETTINGS index_granularity = 8192;

Materialized Views

ClickHouse materialized views are fundamentally different from PostgreSQL materialized views. In PostgreSQL, a materialized view is a cached query result that you refresh manually with REFRESH MATERIALIZED VIEW. In ClickHouse, a materialized view is an insert trigger: it fires on every INSERT into its source table, runs the view’s query against the newly inserted rows, and writes the results to a target table. Historical data is not processed — the view only captures data inserted after the view is created.

This distinction matters operationally. When you add a new materialized view to a live table, you must manually backfill the target table from existing historical data:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

-- 1. Create the target table first
CREATE TABLE pageviews_hourly (
    page_path  String,
    hour       DateTime,
    views      UInt64,
    revenue    Float64
)
ENGINE = SummingMergeTree((views, revenue))
PARTITION BY toYYYYMM(hour)
ORDER BY (page_path, hour);

-- 2. Create the materialized view (triggers on future inserts only)
CREATE MATERIALIZED VIEW pageviews_hourly_mv
TO pageviews_hourly
AS
SELECT
    page_path,
    toStartOfHour(occurred_at) AS hour,
    count()                    AS views,
    sum(revenue)               AS revenue
FROM events
WHERE event_type = 'pageview'
GROUP BY page_path, hour;

-- 3. Backfill historical data manually
INSERT INTO pageviews_hourly
SELECT
    page_path,
    toStartOfHour(occurred_at) AS hour,
    count()                    AS views,
    sum(revenue)               AS revenue
FROM events
WHERE event_type = 'pageview'
GROUP BY page_path, hour;

The TO table syntax (used above) is strongly preferred over the implicit .inner. table syntax. With TO, the target table exists independently — you can query it directly, truncate it for backfills, or attach a different view to it later. The .inner. table is owned by the view and is dropped when the view is dropped, which has caused more than one production incident.

The AggregatingMergeTree + Materialized View Pattern

For complex aggregations — especially distinct counts, quantiles, and histograms — the combination of AggregatingMergeTree and materialized views with *State functions is the canonical ClickHouse pre-aggregation pattern:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

-- Target table stores partial states
CREATE TABLE user_stats_hourly (
    hour              DateTime,
    country_code      LowCardinality(String),
    event_count_state AggregateFunction(count, UInt64),
    revenue_state     AggregateFunction(sum, Float64),
    uniq_users_state  AggregateFunction(uniqHLL12, UInt64)
)
ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, country_code);

-- Materialized view populates partial states on every insert
CREATE MATERIALIZED VIEW user_stats_hourly_mv
TO user_stats_hourly
AS
SELECT
    toStartOfHour(occurred_at)  AS hour,
    country_code,
    countState()                AS event_count_state,
    sumState(revenue)           AS revenue_state,
    uniqHLL12State(user_id)     AS uniq_users_state
FROM events
GROUP BY hour, country_code;

-- Query merges partial states into final results
SELECT
    hour,
    country_code,
    countMerge(event_count_state)      AS events,
    sumMerge(revenue_state)            AS revenue,
    uniqHLL12Merge(uniq_users_state)   AS approx_unique_users
FROM user_stats_hourly
WHERE hour >= '2025-01-01'
GROUP BY hour, country_code
ORDER BY hour, country_code;

uniqHLL12 uses a HyperLogLog sketch with 12-bit precision — approximately 2% error for distinct counts, but the partial states are tiny (about 2KB each) and merge in constant time. For exact distinct counts on large datasets, uniqExact is available but stores a full hash set as its state, which is far larger. Choose based on whether approximate is acceptable.

Query Profiling and EXPLAIN

ClickHouse ships with a rich set of profiling tools. The most important are not in external monitoring systems — they are built into the database itself.

EXPLAIN Variants

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- Logical query plan
EXPLAIN SELECT page_path, sum(views)
FROM pageviews_hourly
WHERE hour >= '2025-01-01'
GROUP BY page_path;

-- Execution pipeline with parallel processor count
EXPLAIN PIPELINE SELECT page_path, sum(views)
FROM pageviews_hourly
WHERE hour >= '2025-01-01'
GROUP BY page_path;

-- Estimated rows and marks to be read -- most useful for index analysis
EXPLAIN ESTIMATE SELECT page_path, sum(views)
FROM pageviews_hourly
WHERE hour >= '2025-01-01'
GROUP BY page_path;

The EXPLAIN ESTIMATE output is the most actionable for performance troubleshooting. It shows:

• rows — estimated rows to be scanned
• marks — number of index granules to be read

If a query is reading 100,000 marks on a table with index_granularity = 8192, that means 819 million rows are being scanned. If the query returns 1,000 rows, something is wrong with your partition pruning or sorting key choice. Either the WHERE clause does not align with the partition key, the sort key columns are not being used effectively, or the query is hitting data across many partitions.

system.query_log

system.query_log is the primary production profiling tool. Every completed query is logged here with full timing and resource usage:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- Find the 20 most expensive queries in the last hour
SELECT
    query,
    query_duration_ms,
    read_rows,
    formatReadableSize(read_bytes)     AS read_size,
    formatReadableSize(memory_usage)   AS memory,
    result_rows,
    exception
FROM system.query_log
WHERE
    type = 'QueryFinish'
    AND event_time >= now() - INTERVAL 1 HOUR
    AND is_initial_query = 1
ORDER BY query_duration_ms DESC
LIMIT 20;

The is_initial_query = 1 filter excludes sub-queries issued by distributed tables, which would otherwise show the same query multiple times. read_rows divided by query_duration_ms gives you rows-per-millisecond throughput — a well-optimized ClickHouse query should be reading tens to hundreds of millions of rows per second on modern hardware.

For investigating a specific slow query, look at the marks read:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

SELECT
    query,
    query_duration_ms,
    read_rows,
    ProfileEvents['SelectedMarks']      AS marks_read,
    ProfileEvents['SelectedParts']      AS parts_read,
    ProfileEvents['SelectedRanges']     AS ranges_read
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY event_time DESC
LIMIT 10;

A high marks_read relative to result_rows means the query is reading far more data than it returns. Investigate whether:

1. The WHERE clause predicate aligns with the partition key (partition pruning)
2. The WHERE clause columns appear in the ORDER BY key (granule pruning via sparse index)
3. There are secondary indexes (bloom filter, minmax) that could help

Secondary indexes in ClickHouse (INDEX name expr TYPE bloom_filter GRANULARITY 4) can help for low-cardinality filters on columns not in the sort key, but they are not a substitute for good sort key design.

clickhouse-benchmark

For load testing, ClickHouse ships a built-in benchmarking tool:

1
2

echo "SELECT sum(revenue) FROM orders WHERE toYYYYMM(created_at) = 202501" \
  | clickhouse-benchmark --iterations=100 --concurrency=10 --host=localhost

This runs the query 100 times with 10 concurrent connections and reports percentile latencies (p50, p95, p99), QPS, and rows-per-second. Useful for measuring the impact of schema changes, index additions, or configuration tuning before deploying to production.

Replication with ClickHouse Keeper

ClickHouse replication is implemented at the table level, not the server level. A non-replicated MergeTree table exists only on the node where it was created. To make a table replicated, you use the ReplicatedMergeTree engine family — any MergeTree variant prefixed with “Replicated”:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

-- On replica 1
CREATE TABLE orders (
    order_id    UInt64,
    customer_id UInt64,
    revenue     Float64,
    created_at  DateTime
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/orders',   -- ZooKeeper/Keeper path
    '{replica}'                            -- replica name (from macros config)
)
PARTITION BY toYYYYMM(created_at)
ORDER BY (customer_id, created_at);

The path /clickhouse/tables/{shard}/orders is a node in the Keeper coordination tree. All replicas of the same table share the same path prefix — this is how they find each other. The {shard} and {replica} placeholders are filled from the server’s macros configuration:

1
2
3
4

<macros>
    <shard>01</shard>
    <replica>clickhouse-01</replica>
</macros>

ClickHouse Keeper

ZooKeeper was the original coordination service for ClickHouse replication. Managing a separate Java-based ZooKeeper ensemble alongside ClickHouse adds operational overhead — separate deployment, separate monitoring, separate JVM tuning. ClickHouse Keeper is a native C++ ZooKeeper-compatible replacement that ships as part of the ClickHouse binary and implements the Raft consensus algorithm directly.

As of ClickHouse 25.x/26.x, ClickHouse Keeper is fully production-ready and is the recommended choice for all new deployments. ZooKeeper remains supported for existing installations, but new features (such as S3Queue) may require Keeper specifically. Migration from ZooKeeper to Keeper requires a cluster stoppage — it is not a live migration.

To configure Keeper, add a keeper_server section to ClickHouse’s configuration:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

<keeper_server>
    <tcp_port>9181</tcp_port>
    <server_id>1</server_id>
    <log_storage_path>/var/lib/clickhouse/coordination/log</log_storage_path>
    <snapshot_storage_path>/var/lib/clickhouse/coordination/snapshots</snapshot_storage_path>
    <coordination_settings>
        <operation_timeout_ms>10000</operation_timeout_ms>
        <session_timeout_ms>30000</session_timeout_ms>
        <raft_logs_level>warning</raft_logs_level>
    </coordination_settings>
    <raft_configuration>
        <server>
            <id>1</id>
            <hostname>clickhouse-01</hostname>
            <port>9234</port>
        </server>
        <server>
            <id>2</id>
            <hostname>clickhouse-02</hostname>
            <port>9234</port>
        </server>
        <server>
            <id>3</id>
            <hostname>clickhouse-03</hostname>
            <port>9234</port>
        </server>
    </raft_configuration>
</keeper_server>

A Keeper ensemble should have an odd number of nodes (3 or 5) to maintain quorum. For small clusters, it is common to co-locate Keeper on the ClickHouse nodes themselves — one Keeper process per node, three nodes total.

Replication Topology

3-Shard x 2-Replica Cluster with ClickHouse Keeper
====================================================

  Keeper Ensemble (3 nodes, Raft quorum):
  +------------------+   +------------------+   +------------------+
  |  clickhouse-01   |   |  clickhouse-03   |   |  clickhouse-05   |
  |  Keeper (id=1)   |<->|  Keeper (id=2)   |<->|  Keeper (id=3)   |
  +------------------+   +------------------+   +------------------+
         |                       |                       |
         | coordinates replication metadata              |
         v                       v                       v
  +------------------+   +------------------+   +------------------+
  |  clickhouse-01   |   |  clickhouse-03   |   |  clickhouse-05   |
  |  shard=01        |   |  shard=02        |   |  shard=03        |
  |  replica=ch-01   |   |  replica=ch-03   |   |  replica=ch-05   |
  +------------------+   +------------------+   +------------------+
         |                       |                       |
    replicates to          replicates to          replicates to
         |                       |                       |
  +------------------+   +------------------+   +------------------+
  |  clickhouse-02   |   |  clickhouse-04   |   |  clickhouse-06   |
  |  shard=01        |   |  shard=02        |   |  shard=03        |
  |  replica=ch-02   |   |  replica=ch-04   |   |  replica=ch-06   |
  +------------------+   +------------------+   +------------------+

  Each shard pair replicates all data for that shard.
  Distributed table fans out queries to all 3 shards.
  Total capacity = 3 shards; fault tolerance = 1 replica per shard.

Monitoring replication health:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

-- Check replication lag and queue depth across all replicated tables
SELECT
    database,
    table,
    replica_name,
    is_leader,
    total_replicas,
    active_replicas,
    queue_size,
    inserts_in_queue,
    merges_in_queue,
    replica_is_active
FROM system.replicas
ORDER BY queue_size DESC;

-- Inspect pending replication tasks
SELECT
    table,
    type,
    create_time,
    required_quorum,
    source_replica,
    num_tries,
    last_exception
FROM system.replication_queue
WHERE last_exception != ''
ORDER BY create_time;

A non-zero last_exception in the replication queue is worth investigating immediately — it usually indicates a network partition, a missing part, or a Keeper connectivity issue.

Distributed Tables and Sharding

A single ClickHouse node can handle tens of billions of rows and hundreds of terabytes of data if you have the hardware for it. Sharding becomes necessary when the working set exceeds a single node’s I/O, memory, or compute capacity — or when you need write throughput higher than one node can absorb.

The Distributed table engine is a virtual table that sits in front of a set of local tables across shards. Queries to the Distributed table are fanned out to all shards in parallel, and results are merged before being returned to the client.

Cluster Configuration

The cluster topology is defined in ClickHouse’s XML configuration:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

<remote_servers>
    <analytics_cluster>
        <shard>
            <replica>
                <host>clickhouse-01</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>clickhouse-02</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <replica>
                <host>clickhouse-03</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>clickhouse-04</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <replica>
                <host>clickhouse-05</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>clickhouse-06</host>
                <port>9000</port>
            </replica>
        </shard>
    </analytics_cluster>
</remote_servers>

Creating the Distributed Table

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

-- Local table exists on every shard
CREATE TABLE orders_local ON CLUSTER analytics_cluster (
    order_id    UInt64,
    customer_id UInt64,
    revenue     Float64,
    created_at  DateTime
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/orders',
    '{replica}'
)
PARTITION BY toYYYYMM(created_at)
ORDER BY (customer_id, created_at);

-- Distributed table is a routing layer
CREATE TABLE orders ON CLUSTER analytics_cluster AS orders_local
ENGINE = Distributed(
    analytics_cluster,   -- cluster name from config
    currentDatabase(),   -- database
    orders_local,        -- local table name on each shard
    cityHash64(customer_id)  -- sharding key
);

The sharding key determines which shard receives each row. Two common choices:

• rand() — uniform random distribution. Even shard sizes, but related rows (same customer) are split across shards, making customer-level JOINs and GROUP BY operations require cross-shard data movement.
• cityHash64(customer_id) — deterministic per customer. All rows for a given customer land on the same shard. Cross-shard JOINs on customer_id become local. The tradeoff is that large customers can create data skew — a customer with 100M events overwhelms one shard while others sit idle.

For most analytics workloads, cityHash64(customer_id) (or whatever your primary entity key is) produces better query performance than rand(). Monitor shard sizes with:

1
2
3
4
5
6
7
8

SELECT
    hostName()                           AS host,
    sum(rows)                            AS total_rows,
    formatReadableSize(sum(bytes_on_disk)) AS disk_size
FROM clusterAllReplicas('analytics_cluster', system.parts)
WHERE table = 'orders_local' AND active = 1
GROUP BY host
ORDER BY total_rows DESC;

Write Path Considerations

Writing to the Distributed table is convenient but introduces overhead — the receiving node writes the data to a local spool directory, then asynchronously routes it to the target shard. This adds latency and creates a replay queue that can back up under high write load. For high-throughput ingestion pipelines, write directly to the local table on each shard and bypass the Distributed routing layer:

1
2
3

# Kafka consumer: write to shard-local table directly
# Determine shard by: cityHash64(customer_id) % num_shards
clickhouse-client --host=clickhouse-01 --query="INSERT INTO orders_local ..."

When ClickHouse Beats PostgreSQL (and When It Doesn’t)

The honest answer is that ClickHouse and PostgreSQL solve different problems, and using the wrong one for your workload is costly in either direction.

Where ClickHouse Wins

Aggregations over large tables. A query like SELECT country, sum(revenue), count() FROM orders WHERE created_at >= '2025-01-01' GROUP BY country against 10 billion rows will finish in under a second on a reasonably sized ClickHouse cluster. The same query in PostgreSQL would run for minutes or require pre-aggregation through a separate pipeline.

Time-series analytics. Event logs, metrics, access logs, click streams — ClickHouse was built for exactly this shape of data. Partitioning by date and sorting by (entity_id, timestamp) aligns perfectly with the “show me all events for user X between time A and B” access pattern.

High-cardinality GROUP BY. ClickHouse’s hash table implementation for GROUP BY is extremely fast. Grouping 1 billion rows by 10 million distinct user IDs is a workload where ClickHouse outperforms virtually every other database.

Dashboard queries that must be fast. Materialized views and pre-aggregated tables make sub-second responses over petabyte-scale datasets achievable. Business intelligence tools like Grafana, Metabase, and Superset work well against ClickHouse.

Where PostgreSQL Wins

Transactions and row-level updates. If your application needs ACID transactions, foreign key enforcement, or true UPDATE/DELETE semantics, PostgreSQL is the right choice. ClickHouse mutations (ALTER TABLE orders UPDATE status = 'cancelled' WHERE order_id = 42) are background operations that rewrite entire parts — they complete eventually, not immediately, and cannot be rolled back.

Complex JOIN workloads. ClickHouse JOINs require the right-side table to fit in memory per query (or per shard in distributed mode). Multi-table JOINs with multiple large tables are difficult. PostgreSQL’s hash join and merge join implementations are far more flexible. ClickHouse is best used with denormalized tables or with the right-side table small enough to broadcast.

Small datasets. For tables under 100M rows, PostgreSQL with appropriate indexes is competitive with ClickHouse and far simpler to operate. The columnar storage advantage shrinks on small tables where the data fits in PostgreSQL’s buffer cache.

Full-text search, geospatial. PostgreSQL with pg_trgm, tsvector, and PostGIS handles text search and geospatial queries natively. ClickHouse has some text search functions but is not competitive with dedicated search systems or PostgreSQL’s mature extensions.

Operational simplicity. A single PostgreSQL instance needs nothing except itself. A production ClickHouse cluster needs ClickHouse nodes, a Keeper ensemble, shared storage or block device replication, and careful configuration management. The operational burden is real.

The Sweet Spot: ClickHouse as Analytics Sidecar

The most common production pattern is not ClickHouse replacing PostgreSQL — it is ClickHouse running alongside PostgreSQL. Operational data lives in PostgreSQL: the orders table, the user table, the product catalog. For reporting and analytics, data is replicated to ClickHouse using CDC tools (Debezium, Airbyte, ClickHouse’s built-in MaterializeMySQL/PostgreSQL table engines) or a batch ETL process.

This gives you OLTP where you need it (PostgreSQL) and OLAP where you need it (ClickHouse), without trying to force one system to do both well.

Comparison Table

DuckDB 1.5 (released March 2026) deserves a specific mention. For single-machine analytics — data science notebooks, local data exploration, ETL scripts that aggregate Parquet files — DuckDB offers ClickHouse-class columnar query performance with zero operational overhead. It runs in-process as a library. It is not a server, cannot be shared across multiple clients, and tops out at the memory and storage of one machine. But for the workloads it targets, it is exceptional. The choice between DuckDB and ClickHouse is usually the choice between single-analyst local work and production multi-user systems.

Operational Considerations

A few things learned the hard way:

INSERT batch size matters. ClickHouse creates one part per INSERT call. Inserting one row at a time creates thousands of tiny parts, and the merge process cannot keep up. The minimum recommended batch size is 1,000 rows; 10,000-100,000 is more typical for production pipelines. Use a buffer layer (Kafka, a Kafka consumer with local batching, or ClickHouse’s own Buffer table engine) to aggregate small writes.

Too many partitions is a problem. If you partition by day and your table has three years of data, that is over 1,000 partitions. ClickHouse loads partition metadata at startup and query planning time — too many partitions increases planning overhead. Monthly partitioning is more common than daily for large tables. Partition on whatever time granularity you actually use for pruning in queries, not finer.

Backup strategy. ClickHouse has a built-in BACKUP TABLE command (since 22.x) and supports backup to S3, GCS, and local disk. For production systems, combine clickhouse-backup (the open-source tool) with S3 for full and incremental backups. Test restores — a backup you have never restored is not a backup.

Memory limits. ClickHouse is aggressive about memory use by default. Set max_memory_usage at the user or query level to prevent runaway queries from OOM-killing the server. A reasonable starting point is 80% of available RAM for queries, with the remaining 20% reserved for merges and OS cache.

1
2
3
4
5

-- Set per-query memory limit in session
SET max_memory_usage = 10000000000;  -- 10 GB

-- Set per-user limits in users.xml or via SQL
ALTER USER analyst SETTINGS max_memory_usage = 5000000000;

ClickHouse rewards careful schema design and query pattern analysis far more than most databases. The performance ceiling is extremely high, but reaching it requires understanding the storage model, choosing sorting keys deliberately, and building your materialized view pipeline around your actual query patterns. When you get it right, the query times feel almost implausible — seconds on tables where Postgres takes hours. When you get it wrong, you are scanning entire datasets for every query and wondering why it is not faster than Postgres.

The investment in understanding the system pays off quickly in production.