How ClickHouse Works Internally: The MergeTree Storage Engine, Columnar Compression, and Distributed Query Execution

Most databases optimize for transactional workloads: point reads, small writes, row-level locks. ClickHouse optimizes for exactly one thing instead: scanning billions of rows as fast as the hardware allows. It trades away random write performance, full ACID transactions, and row-level mutations to achieve analytical query throughput that row-oriented databases simply cannot reach. Understanding why requires going past the marketing claims and into how the storage engine, compression, and query executor actually work.

The Core Design Decision: Columns, Not Rows

A row-oriented database stores all columns for a given row adjacent in memory. A columnar database stores all values for a given column adjacent. For a table with 100 columns and a query that touches 3 of them, the columnar layout reads roughly 3% of the data a row store would read. On modern SSDs, I/O bandwidth is the real bottleneck for analytical queries. Columnar storage turns a bandwidth problem into a much smaller one.

ClickHouse takes this further by operating on blocks of column values (typically 8,192 rows per block) rather than individual values. Every operation in the query pipeline processes a batch. Aggregation, filtering, projection: all operate on vectors of column data. This is vectorized processing, and it maps well onto modern CPU SIMD instructions.

The MergeTree Storage Engine

MergeTree is the foundational storage engine in ClickHouse. Nearly every production table uses MergeTree or one of its variants (ReplicatedMergeTree, AggregatingMergeTree, CollapsingMergeTree, ReplacingMergeTree). The name reflects the core mechanic: data is written as immutable parts, and parts are continuously merged in the background.

Parts and the Write Path

When ClickHouse receives an INSERT, it does not write individual rows into an existing structure. It creates a new immutable directory on disk called a data part. Each part contains:

• One binary file per column (e.g., timestamp.bin, user_id.bin, event_type.bin)
• One marks file per column (timestamp.mrk2) storing byte offsets for granule boundaries
• A primary index file (primary.idx) holding the primary key values at each granule boundary
• Partition metadata and a checksums file

A single insert statement produces one part. If you insert 1,000 rows, you get one new part. If you insert 100 times in rapid succession, you get 100 parts. ClickHouse merges small parts into larger ones in the background, continuously. Too many parts (the “too many parts” error is infamous) indicates inserts are arriving faster than merges can keep up.

This design has a direct consequence: inserts are fast (sequential writes to a new directory) but small-batch inserts are expensive to query until merged. The recommended pattern is batching inserts server-side or client-side to 100,000 to 1,000,000 rows per insert.

Granules and Sparse Primary Indexing

ClickHouse does not build a dense index that maps every row to a disk offset. It builds a sparse primary index. Column data is divided into granules of 8,192 rows (configurable via index_granularity). The primary index stores the primary key value of the first row in each granule.

A query with a WHERE filter on the primary key does not do a binary search over individual rows. It does a binary search over the primary index to find the range of granules that could contain matching rows, then reads only those granules. For a table with 10 billion rows and granule size 8,192, the primary index has roughly 1.2 million entries. It fits comfortably in memory. Reading it is fast.

The granule-skip mechanic means primary key selectivity is granule-level, not row-level. If your filter matches 1 row out of 8,192 in a granule, you still read the whole granule. This is why choosing the right primary key ordering matters enormously. High-cardinality columns that appear in most query filters should come first in the ORDER BY clause (which defines the primary key in MergeTree).

Skip Indexes

Beyond the primary index, ClickHouse supports secondary skip indexes that operate on the same granule unit. A minmax skip index on a column stores the min and max value per granule. If a query filter cannot possibly match the granule (e.g., WHERE temperature > 100 and the granule’s max is 80), ClickHouse skips it entirely without reading the column data.

Other skip index types include set (stores a set of distinct values per granule, useful for low-cardinality columns), bloom_filter (approximate membership for string/array columns), and ngrambf_v1 (n-gram bloom filter for substring searches).

Columnar Compression and Codec Chains

Each column file is compressed independently. The default codec is LZ4, which offers fast decompression at moderate compression ratios. ZSTD provides better ratios at higher CPU cost. ClickHouse lets you stack codecs and apply pre-processing transformations before compression.

A codec chain like CODEC(Delta, LZ4) works as follows: the Delta codec subtracts each value from the previous one, turning a monotonically increasing timestamp column into a sequence of small deltas (often all the same value for regular time-series). LZ4 then compresses that delta sequence extremely efficiently. A column of Unix millisecond timestamps might compress at 50:1 or better with this chain.

Other useful pre-processing codecs:

• DoubleDelta: second-order delta, useful for slowly varying numeric sequences
• Gorilla: XOR-based float compression borrowed from Facebook’s Gorilla TSDB, effective for floating-point metrics
• T64: transposes the bit representation of integers, grouping bits of the same significance across values

Codecs are declared per column in the table DDL. You choose them based on the data’s statistical properties.

The Query Execution Pipeline

ClickHouse query execution follows a push-based pipeline model. The planner builds a DAG of operators. The executor runs operators in threads, with each operator pulling column blocks from its input and pushing results to its output.

Key steps in a typical analytical query:

1. Partition pruning: The query planner inspects the WHERE clause and eliminates entire partitions (subdirectories keyed on the partition expression) that cannot contain matching data. This happens before any I/O.

2. Part-level pruning: Within surviving partitions, parts outside the query’s time/key range are skipped.

3. Granule selection via primary index: Within surviving parts, the primary index binary search narrows the read to a range of granules.

4. Skip index evaluation: Per-granule skip indexes are checked to eliminate granules that cannot match.

5. Column read and decompression: Only the columns referenced in the query are read. Each column block is decompressed using its codec chain.

6. Vectorized filter evaluation: The WHERE predicate runs over column vectors using SIMD operations where possible.

7. Aggregation: Hash table aggregation operates on column blocks. ClickHouse uses a two-level hash table that avoids rehashing for large aggregation cardinalities.

8. Result serialization: The result set is serialized in the requested output format.

TypeScript Client Example

The official @clickhouse/client package exposes a typed interface for queries and inserts.

import { createClient } from "@clickhouse/client";

const client = createClient({
  host: "https://your-clickhouse-host:8443",
  username: "default",
  password: process.env.CLICKHOUSE_PASSWORD,
  database: "analytics",
});

// Typed insert with column-per-array format (efficient for large batches)
interface EventRow {
  timestamp: string;
  user_id: string;
  event_type: string;
  properties: string;
}

async function insertEvents(events: EventRow[]): Promise<void> {
  await client.insert<EventRow>({
    table: "events",
    values: events,
    format: "JSONEachRow",
  });
}

// Parameterized query returning typed results
interface FunnelStep {
  event_type: string;
  step_count: number;
  unique_users: number;
}

async function getFunnelMetrics(
  startDate: string,
  endDate: string
): Promise<FunnelStep[]> {
  const result = await client.query({
    query: `
      SELECT
        event_type,
        count() AS step_count,
        uniq(user_id) AS unique_users
      FROM events
      WHERE toDate(timestamp) BETWEEN {startDate: Date} AND {endDate: Date}
      GROUP BY event_type
      ORDER BY step_count DESC
    `,
    query_params: { startDate, endDate },
    format: "JSONEachRow",
  });

  return result.json<FunnelStep>();
}

// Streaming large result sets to avoid memory pressure
async function streamLargeReport(date: string): Promise<void> {
  const stream = await client.query({
    query: `
      SELECT user_id, sum(revenue) AS total_revenue
      FROM orders
      WHERE toDate(created_at) = {date: Date}
      GROUP BY user_id
    `,
    query_params: { date },
    format: "JSONEachRow",
  });

  for await (const rows of stream.stream<{ user_id: string; total_revenue: number }>()) {
    rows.forEach((row) => {
      // process each row without loading full result into memory
      console.log(row.json());
    });
  }
}

Batching inserts to at least 10,000 rows per call significantly reduces part creation overhead. In high-throughput pipelines, an async buffer that flushes every 500ms or every 100,000 rows is a common pattern.

Distributed Tables and Two-Phase Query Execution

A single ClickHouse node has limits. For very large datasets, ClickHouse distributes data across shards. The Distributed table engine sits in front of a cluster of shards. A query against a Distributed table executes in two phases:

Phase 1 (scatter): The coordinating node sends a rewritten query to every shard. Each shard executes the query locally against its portion of the data, performing local aggregation where possible. For a GROUP BY with sum(), each shard computes partial sums per key.

Phase 2 (gather): The coordinating node receives partial results from all shards and merges them. Partial sums are added together. uniq() estimates are merged using HyperLogLog union. quantile() estimates are merged using their respective sketch structures.

This two-phase model means some aggregation functions work differently in distributed mode. count(distinct x) requires exact deduplication across shards, which forces a full data shuffle. uniq(x) uses HyperLogLog with merge-friendly sketches and avoids the shuffle. Choosing between them is a genuine tradeoff between accuracy and query performance at scale.

Shard keys matter for query locality. If your most common queries filter on user_id, sharding by intHash64(user_id) % num_shards ensures most queries hit only one shard. A poor shard key forces every query to scatter across all shards and merge.

Materialized Views

Materialized views in ClickHouse are insert-triggered transformations, not precomputed snapshots. When data is inserted into the source table, ClickHouse simultaneously runs the view’s SELECT query against the inserted block and inserts the result into the target table.

This means materialized views are a low-latency aggregation layer. A view that computes per-minute event counts is updated atomically with each source insert, with no batch refresh lag. The target table is a normal MergeTree table, so it accumulates rows over time and queries against it are just queries against a smaller, pre-aggregated table.

The AggregatingMergeTree engine is the standard target for materialized views that maintain running aggregates. It stores partial aggregate states (e.g., HyperLogLog state for uniqState(), partial sum state for sumState()) and merges them during background part merges using finalizeAggregation().

Replication via ClickHouse Keeper

ClickHouse replication historically depended on Apache ZooKeeper for distributed coordination: tracking which parts exist on which replicas, coordinating merges, storing replication queues. ClickHouse Keeper is a ZooKeeper-compatible replacement written in C++ and embedded directly in the ClickHouse process. It implements the Raft consensus algorithm to provide strong consistency for its coordination metadata.

ReplicatedMergeTree tables register their parts and merge operations in Keeper. When replica A performs a merge producing part 20240601_1_100_5, it records this in Keeper. Replica B reads the Keeper log, sees the merge instruction, and either fetches the merged part from replica A directly or re-executes the merge locally from its own copy of the constituent parts, whichever is faster.

Replication is asynchronous. Replicas may lag. SELECT queries with SETTINGS select_sequential_consistency = 1 wait for all replicas to catch up before returning, at the cost of added latency. Most read workloads tolerate eventual consistency across replicas and skip this setting.

Production Considerations

Partition strategy: The PARTITION BY expression determines how data is physically split into subdirectories. Monthly partitions (toYYYYMM(timestamp)) are a common default. Partitions that fall entirely outside a query’s time range are skipped without reading any data. Very fine-grained partitions (hourly or by individual user ID) create too many parts and directories, degrading merge performance and metadata overhead.

Primary key ordering: Place columns with the highest query filter selectivity first in ORDER BY. Date or timestamp columns that appear in almost every query should typically come first. If your queries always filter by (tenant_id, date), the ORDER BY should be (tenant_id, toDate(timestamp), ...). Columns with very high cardinality (UUID, random hash) placed first in the primary key produce an index that is rarely selective enough to skip granules.

Projections: Projections are alternative sort orders stored inside the same data part. A table with ORDER BY (date, user_id) can have a projection with ORDER BY (user_id, date) for queries that filter by user_id first. Projections add storage overhead (the data is stored twice in different column orderings) but eliminate the need for secondary tables or manual denormalization.

Mutations: ALTER TABLE ... UPDATE and ALTER TABLE ... DELETE in ClickHouse are not in-place row updates. They rewrite entire parts asynchronously in the background. Frequent mutations destroy the read performance benefits of immutable columnar storage. Design schemas to avoid updates. Use ReplacingMergeTree for deduplication or CollapsingMergeTree for event-based corrections instead.

Insert deduplication: ReplicatedMergeTree supports insert deduplication using a checksum-based token stored in Keeper. If an insert fails and is retried with the same data, ClickHouse recognizes the duplicate and silently ignores it. This requires inserts to supply a dedup_token or rely on the built-in block checksum mechanism, and it only deduplicates within a configurable window (default 100 recent inserts).

Memory and query limits: Analytical queries can consume significant memory for hash table aggregation. The max_memory_usage setting caps memory per query. Queries exceeding the limit spill to disk or return an error, depending on configuration. For production dashboards with concurrent users, setting per-query memory limits and using read replicas to distribute load are standard practices.

Tradeoffs: ClickHouse vs Alternatives

Dimension	ClickHouse	PostgreSQL	BigQuery	Apache Druid	TimescaleDB
Storage model	Columnar, MergeTree parts	Row-oriented, heap	Columnar, serverless	Columnar, segments	Row + columnar chunks
Write model	Batched inserts, immutable parts	Row-level MVCC	Streaming or load jobs	Real-time ingestion via Kafka	Row-level MVCC
Query model	Vectorized, single-node or distributed	Row-at-a-time, MVCC	Distributed serverless MPP	Scatter-gather, approximate	Row-at-a-time
Aggregation at scale	Excellent (billion-row in seconds)	Poor above ~100M rows	Excellent (scales automatically)	Good (approximate by default)	Moderate
Exact vs approximate	Exact by default, HLL opt-in	Exact	Exact	Approximate by default (Theta sketches)	Exact
OLTP capability	None (no row updates, limited joins)	Full ACID	None	None	Full ACID
Join support	Limited (broadcast joins work, shuffle joins are slow)	Full	Full	Limited	Full
Operational complexity	Medium (cluster setup, part management)	Low	None (managed)	High (many moving parts)	Low
Cost model	Self-hosted or cloud	Self-hosted or cloud	Per-byte scanned	Self-hosted	Self-hosted or cloud
Best fit	Event analytics, logs, metrics at scale	General-purpose OLTP/OLAP	Ad-hoc analytics without ops burden	Sub-second dashboards on pre-aggregated data	Time-series with PostgreSQL compatibility

When to Choose ClickHouse

ClickHouse is the right choice when you have event-scale data (hundreds of millions to trillions of rows), queries are primarily aggregations over time ranges, and write patterns are append-only or nearly so. Product analytics, log analysis, metrics storage, billing aggregation: these workloads match ClickHouse’s architecture directly.

It is the wrong choice when your queries require complex multi-table joins across large datasets, when you need row-level updates at high frequency, or when you need full transactional semantics. For those patterns, PostgreSQL or a hybrid approach combining a transactional store with ClickHouse for the analytical layer is more appropriate.

The architecture is not accidental complexity. Every constraint, from immutable parts to sparse indexing to granule-oriented skip logic, exists to serve the core goal: scan as many column values per second as the hardware allows. That singular focus is what makes ClickHouse perform the way it does, and it is worth understanding before designing a schema or picking a shard key.