ClickHouse Internals

How ClickHouse actually stores, merges, and executes queries against columnar data at analytical scale — the MergeTree part lifecycle, why columnar storage and vectorized execution make full scans cheap, and the replication and distributed-aggregation mechanics that show up once you shard and replicate for real traffic.

MergeTree Storage Engine

ClickHouse's primary table engine — designed for analytical queries over billions of rows. Every INSERT writes its own part; a background thread continually merges parts into fewer, larger ones so reads never have to stitch together thousands of tiny fragments.

graph TD
    classDef write fill:#3498db,stroke:#2471a3,color:#fff
    classDef partclass fill:#7f8c8d,stroke:#616a6b,color:#fff
    classDef merge fill:#e67e22,stroke:#ba6018,color:#fff
    classDef result fill:#27ae60,stroke:#1e8449,color:#fff

    INSERT1["INSERT INTO events VALUES (...)"]:::write --> PART1["Part: 20240115_1_1_0/"]:::partclass
    INSERT2["INSERT INTO events VALUES (...)"]:::write --> PART2["Part: 20240115_2_2_0/"]:::partclass
    INSERT3["INSERT INTO events VALUES (...)"]:::write --> PART3["Part: 20240115_3_3_0/"]:::partclass

    subgraph STRUCT["Every part has this structure on disk"]
        COL_TS["event_date.bin (compressed)<br/>event_date.mrk2 (mark file)"]
        COL_USR["user_id.bin (compressed)"]
        COL_ACT["action.bin (compressed)"]
        PRIMARY["primary.idx<br/>sparse index on ORDER BY columns"]
        MINMAX["minmax_event_date.idx<br/>min/max per granule, used for partition pruning"]
    end

    PART1 -.-> STRUCT
    PART2 -.-> STRUCT
    PART3 -.-> STRUCT

    PART1 --> MERGE["Background merge thread<br/>picks parts from the same partition<br/>(like LSM compaction)"]:::merge
    PART2 --> MERGE
    PART3 --> MERGE
    MERGE --> BIGPART["Larger merged part: 20240115_1_3_1/<br/>rebuilt primary.idx, higher compression ratio"]:::result

The diagram above shows one merge frozen in time. This is the same lifecycle running live: insert rows, watch them buffer in the memtable, flush into a part once 4 rows have accumulated (or force it sooner), and watch the background merge fire automatically once more than 3 parts pile up in the same partition — exactly the "many small parts → fewer, larger ones" cycle described above, just small enough numbers to actually watch happen.

The two index files do different jobs even though both sound like "an index on this column." minmax_event_date.idx works at the partition level — it lets the planner throw out whole partitions (whole months, in this schema) before opening them at all. primary.idx works one level down, inside whatever partitions survive that cut — it's the sparse index over the ORDER BY key that skips individual granules within a part. Partition pruning is the coarse first cut; the primary index skip is the fine-grained second one.

Data-Skipping Indexes — Beyond the Automatic Ones

primary.idx and minmax_event_date.idx are both automatic — every MergeTree table has them, whether you ask for them or not, and they only ever help a query that filters on the partition key or (a prefix of) the ORDER BY key. A query filtering on some other column gets none of that benefit: with no index to consult, ClickHouse has to open every granule in every part and check the condition row-range by row-range. That's the gap data-skipping indexes (also called secondary indexes) close — small, optional, per-granule summaries you explicitly declare on a non-key column so a query can rule out granules that provably can't match, without ClickHouse building anything as heavyweight as a B-tree over that column.

ALTER TABLE events ADD INDEX idx_action action TYPE set(100) GRANULARITY 4;
ALTER TABLE events ADD INDEX idx_value value TYPE minmax GRANULARITY 4;
ALTER TABLE events ADD INDEX idx_ua user_agent TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4;

-- Existing parts aren't indexed retroactively — MATERIALIZE builds the index
-- over data already on disk; new inserts get it automatically going forward.
ALTER TABLE events MATERIALIZE INDEX idx_action;

The mental model that matters more than memorizing the list: a data-skipping index can only tell you which granules to SKIP, never where a matching row actually is. A B-tree secondary index points you straight at a row. These don't — minmax says "this granule's range doesn't contain your value, skip it" (or "it might, go look"); bloom_filter says "this value is definitely not in this granule" (or "it might be, go look"). Either way, a granule that isn't ruled out still gets fully scanned and filtered the normal way. The entire value proposition is I/O reduction — fewer granules decompressed and checked — not point-lookup precision. A data-skipping index with weak selectivity for a given workload doesn't produce wrong results; it just fails to skip anything, and the query degrades to the same full scan it would've done with no index at all.

GRANULARITY here means something more specific than index_granularity: it's how many consecutive index blocks (each already covering index_granularity rows) one skip-index entry summarizes. GRANULARITY 4 on a minmax index means each stored min/max pair spans 4 granules' worth of rows, not 1. A smaller value gives finer-grained skipping (a query can rule out data in tighter chunks) at the cost of more index entries to store and check; a larger value shrinks the index but forces the query to pull in more rows around each match it can't rule out.

Columnar Storage — Why Queries Are Fast

graph LR
    classDef query fill:#8e44ad,stroke:#6c3483,color:#fff
    classDef mustread fill:#e74c3c,stroke:#c0392b,color:#fff
    classDef skip fill:#95a5a6,stroke:#707b7c,color:#fff
    classDef actualread fill:#27ae60,stroke:#1e8449,color:#fff

    Q["Query: COUNT(*) WHERE action='click'"]:::query

    subgraph ROW["Row storage (PostgreSQL, MySQL)"]
        R1["row1: {id, user_id, action, value, ts}"]:::mustread
        R2["row2: {id, user_id, action, value, ts}"]:::mustread
        R3["row3: {id, user_id, action, value, ts}"]:::mustread
    end

    subgraph COL["Column storage (ClickHouse)"]
        C_ID["id column: [1,2,3,4,5...]"]:::skip
        C_USR["user_id column: [101,102,101...]"]:::skip
        C_ACT["action column: ['click','view'...]"]:::actualread
    end

    Q -->|"must read every column<br/>of every row to reach action"| ROW
    Q -->|"reads only the action column<br/>10-100x less I/O"| COL

Compression per column: Each column has uniform data type → high compression ratio. user_id column: sorted integers → delta encoding → LZ4/ZSTD. Typical: 5-10x compression vs raw CSV.

MergeTree Family

MergeTree                 — base engine, append + merge
ReplacingMergeTree        — deduplicate by ORDER BY key on merge
SummingMergeTree          — aggregate numeric columns on merge
AggregatingMergeTree      — store partial aggregates, merge = combine aggregates
CollapsingMergeTree       — delete rows by sign column (CRDT-like)
ReplicatedMergeTree       — MergeTree with ZooKeeper/Keeper replication

The four non-base, non-replicated variants below all differ only in what the merge step does with rows that share the same ORDER BY key — the rest of MergeTree's mechanics (parts, granules, background merges) are identical across all of them. ReplicatedMergeTree is orthogonal to this choice: it's a replication layer that can wrap any of these engines, covered in ReplicatedMergeTree Internals below.

CREATE TABLE events (
    event_date  Date,
    user_id     UInt64,
    action      LowCardinality(String),
    value       Float64
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
PARTITION BY toYYYYMM(event_date)     -- partition by month
ORDER BY (event_date, user_id)         -- sort key = primary key
SETTINGS index_granularity = 8192;    -- rows per granule (sparse index unit)

Query Execution

graph LR
    classDef inputs fill:#8e44ad,stroke:#6c3483,color:#fff
    classDef plan fill:#f39c12,stroke:#ba6018,color:#fff
    classDef io fill:#2980b9,stroke:#1f618d,color:#fff
    classDef exec fill:#27ae60,stroke:#1e8449,color:#fff
    classDef out fill:#7f8c8d,stroke:#616a6b,color:#fff

    QUERY2["SELECT user_id, count() FROM events<br/>WHERE event_date = '2024-01-15'<br/>GROUP BY user_id"]:::inputs --> PLAN

    subgraph PLANNING["Query planner"]
        PLAN["Partition pruning<br/>skip other months via minmax idx"]:::plan --> SKIP["Primary index skip<br/>skip non-matching granules"]:::plan
    end

    SKIP --> READ["Read only event_date + user_id columns<br/>decompress needed granules only"]:::io
    READ --> VEC["Vectorized execution<br/>process 8192 rows/granule at once (SIMD)"]:::exec
    VEC --> AGG["AggregatingTransform<br/>per-thread hash aggregation"]:::exec
    AGG --> MERGEAGG["MergingAggregatedTransform<br/>combine per-thread partials"]:::exec
    MERGEAGG --> RESULT["Result"]:::out

Granule: ClickHouse divides each column file into granules of index_granularity rows (default 8192). The sparse primary index stores the first value of each granule. Queries skip entire granules that can't match the WHERE clause.

Materialized Views

Pre-aggregate data at insert time for fast dashboard queries:

sequenceDiagram
    participant APP as Application
    participant SRC as events (source MergeTree)
    participant MV as events_by_user_daily (AggregatingMergeTree)

    APP->>SRC: INSERT INTO events VALUES (block of rows)
    SRC->>SRC: write new data part
    Note over SRC,MV: the MV's SELECT runs only over<br/>the just-inserted block, not the whole table
    SRC->>MV: feed inserted block through the MV's SELECT
    MV->>MV: countState() / sumState() over the block<br/>write a new partial-aggregate part
    Note over MV: existing historical parts in the MV<br/>are untouched by this insert

    APP->>MV: SELECT ... countMerge(cnt), sumMerge(total) ...
    MV->>MV: merge partial-aggregate states<br/>across every matching part
    MV-->>APP: final aggregated result

-- Source table
CREATE TABLE events (...) ENGINE = MergeTree() ...;

-- Materialized view: count events per user per day
CREATE MATERIALIZED VIEW events_by_user_daily
ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, user_id)
AS SELECT
    event_date,
    user_id,
    countState() AS cnt,        -- partial aggregate state
    sumState(value) AS total
FROM events
GROUP BY event_date, user_id;

-- Query the materialized view
SELECT
    event_date,
    user_id,
    countMerge(cnt) AS count,   -- merge partial states
    sumMerge(total) AS total
FROM events_by_user_daily
WHERE event_date = today()
GROUP BY event_date, user_id;

A materialized view here isn't a live view recomputed on read — it's a trigger wired to INSERT. It only ever sees the rows in the block currently being inserted into the source table; it has no visibility into rows that already existed in events before the view was created. That's why standing up a materialized view against a table that already has data needs a manual one-time backfill (an INSERT INTO events_by_user_daily SELECT ... FROM events covering the existing rows) done alongside CREATE MATERIALIZED VIEW — the view itself only ever populates incrementally, going forward.

Key Configuration

<!-- config.xml -->
<max_memory_usage>10000000000</max_memory_usage>  <!-- 10GB per query -->
<max_threads>8</max_threads>
<max_concurrent_queries>100</max_concurrent_queries>

<!-- Compression -->
<compression>
    <case><min_part_size>10000000000</min_part_size>
        <method>zstd</method><level>3</level>
    </case>
</compression>

Key Metrics

ClickHouseMetrics_Query                    # active queries
ClickHouseAsyncMetrics_MemoryResident      # RSS memory
ClickHouseMetrics_BackgroundMergesAndMutations  # merge queue
ClickHouseProfileEvents_MergedRows         # rows merged per second
ClickHouseMetrics_ReplicasMaxQueueSize      # replication queue depth (pending fetches/merges)

EXPLAIN — Query Execution Analysis

-- See query execution pipeline
EXPLAIN PIPELINE SELECT user_id, count() FROM events
WHERE event_date = today() GROUP BY user_id;

-- Output shows stages:
-- (ExpressionTransform) → (AggregatingTransform) → (MergingAggregatedTransform)
-- → shows parallelism: how many threads per stage

-- Detailed analysis with timing
EXPLAIN ANALYZE SELECT user_id, count() FROM events
WHERE event_date = today() GROUP BY user_id;

-- Show which granules are read (index analysis)
EXPLAIN indexes=1 SELECT count() FROM events WHERE user_id = 123;
-- Marks (1) → only 1 granule read out of thousands (primary index worked)

TTL — Automatic Data Expiry

-- Auto-delete rows older than 30 days
CREATE TABLE events (
    event_date  Date,
    user_id     UInt64,
    action      String
) ENGINE = MergeTree()
ORDER BY (event_date, user_id)
TTL event_date + INTERVAL 30 DAY DELETE;
-- Rows deleted during background merges when TTL expires

-- Move old data to cheaper storage tier (tiered storage)
TTL event_date + INTERVAL 7 DAY TO DISK 'ssd',
    event_date + INTERVAL 30 DAY TO DISK 'hdd',
    event_date + INTERVAL 90 DAY TO VOLUME 's3';

-- Check TTL status
SELECT name, data_compressed_bytes/1e9 AS compressed_gb,
       min_date, max_date
FROM system.parts
WHERE table = 'events' AND active;

Distributed Aggregation

When ClickHouse shards data, aggregations run in two phases:

graph LR
    classDef initiator fill:#8e44ad,stroke:#6c3483,color:#fff
    classDef shard fill:#2980b9,stroke:#1f618d,color:#fff
    classDef result fill:#27ae60,stroke:#1e8449,color:#fff

    APP["Client query on<br/>distributed_events"] --> INIT["Initiator node<br/>receives query, fans out to shards"]:::initiator

    subgraph SHARDS["Phase 1 — local partial aggregation, per shard"]
        S1["Shard 1<br/>partial GROUP BY, local data only"]:::shard
        S2["Shard 2<br/>partial GROUP BY, local data only"]:::shard
        S3["Shard 3<br/>partial GROUP BY, local data only"]:::shard
    end

    INIT --> S1
    INIT --> S2
    INIT --> S3
    S1 & S2 & S3 -->|"partial aggregates,<br/>not final rows"| MERGE["Phase 2 — initiator merges<br/>partials into final GROUP BY result"]:::result
    MERGE --> APP

-- Distributed table fans out to shards, merges results
SELECT user_id, count() FROM distributed_events
WHERE event_date = today() GROUP BY user_id ORDER BY count() DESC LIMIT 10;

-- Under the hood: each shard runs:
-- SELECT user_id, count() FROM local_events WHERE event_date = today() GROUP BY user_id
-- Initiator merges partial counts from all shards

ReplicatedMergeTree Internals

sequenceDiagram
    participant C as Client
    participant R0 as Replica 0 (leader)
    participant KEEPER as ClickHouse Keeper / ZooKeeper
    participant R1 as Replica 1

    C->>R0: INSERT INTO events VALUES (...)
    R0->>R0: write local data part: 20240115_1_1_0/
    R0->>KEEPER: register part metadata<br/>(part name, checksum, block_id)

    alt default — insert_quorum not set (asynchronous)
        R0-->>C: OK — INSERT returns immediately
        KEEPER->>R1: notification: new part available
        R1->>R0: fetch part (HTTP, port 9009)
        R1->>R1: verify checksum, write locally
        R1->>KEEPER: mark part replicated
    else insert_quorum = 2 (synchronous-like)
        KEEPER->>R1: notification: new part available
        R1->>R0: fetch part (HTTP, port 9009)
        R1->>R1: verify checksum, write locally
        R1->>KEEPER: mark part replicated
        KEEPER-->>R0: quorum of 2 replicas confirmed
        R0-->>C: OK — INSERT only returns now
    end

Replication is asynchronous — INSERT returns after writing to one replica. Use insert_quorum for synchronous-like behavior:

SET insert_quorum = 2;  -- wait for 2 replicas to confirm before returning
SET insert_quorum_timeout = 60000;  -- 60 second timeout

INSERT INTO events VALUES (...);  -- blocks until 2 replicas have the data

Query Profiling and Optimization

-- System tables for query analysis
SELECT query, read_rows, read_bytes/1e9 AS read_gb,
       memory_usage/1e9 AS memory_gb,
       query_duration_ms
FROM system.query_log
WHERE type = 'QueryFinish'
  AND event_time > now() - INTERVAL 1 HOUR
ORDER BY query_duration_ms DESC LIMIT 10;

-- Find queries doing too many reads (need better indexes/partitioning)
SELECT query, read_rows, read_rows / result_rows AS selectivity
FROM system.query_log
WHERE type = 'QueryFinish' AND read_rows > 1e8
ORDER BY read_rows DESC LIMIT 10;
-- If selectivity > 10000: reading 10K rows per result row → bad filtering

-- Parts and merges in progress
SELECT table, elapsed, progress, rows_read, rows_written
FROM system.merges;

-- Background merge queue size (high = inserts faster than merges)
SELECT table, count() AS parts_count
FROM system.parts
WHERE active AND database = currentDatabase()
GROUP BY table
HAVING parts_count > 100  -- warning: too many parts → slow queries
ORDER BY parts_count DESC;

Ingestion Best Practices

-- Batch inserts: ClickHouse is optimized for large batches, NOT one-row inserts
-- Bad: INSERT INTO events VALUES (row1); INSERT INTO events VALUES (row2);
-- Good: INSERT INTO events VALUES (row1),(row2),...,(row10000);

-- Use async_insert for high-frequency small inserts
SET async_insert = 1;          -- buffer inserts server-side
SET wait_for_async_insert = 0; -- don't wait for flush confirmation

-- Buffer table: accumulate writes, flush to MergeTree periodically
CREATE TABLE events_buffer AS events
ENGINE = Buffer(currentDatabase(), events, 16, 10, 100, 10000, 1000000, 10000000, 100000000);
-- Flushes when: time > 10-100s OR rows > 10K-1M OR bytes > 10MB-100MB
INSERT INTO events_buffer VALUES (...);  -- fast, goes to RAM buffer