How ClickHouse Works Internally: MergeTree Storage, Vectorized Execution, and Distributed Query Processing

You run a query that scans 10 billion rows and gets results back in under two seconds. If you have used ClickHouse for real-time analytics, that experience is probably familiar. If you have not, that number sounds like marketing. It is not.

Understanding why ClickHouse is fast requires understanding its architecture at a level deeper than “it’s columnar.” A lot of databases are columnar. ClickHouse is fast for a specific set of reasons that compound: storage layout designed around merge semantics, an execution engine that processes data in vectors aligned to CPU cache lines, a distributed model that pushes computation close to storage, and a suite of precomputation primitives that make common aggregations nearly free.

This article walks through each layer.

The MergeTree Storage Engine

Everything in ClickHouse is built around the MergeTree family of table engines. The name is literal: data is written in parts, and parts are merged in the background.

Parts and the Write Path

When you insert a batch of rows, ClickHouse writes them as an immutable directory on disk called a part. A part contains one file per column (plus index and checksum metadata). Each column file is compressed independently. This is not the WAL-driven buffer pool model you know from PostgreSQL. There is no heap file. There is no transaction log that gets replayed into a shared buffer pool at read time.

The consequence is that inserts are fast (you are writing sequential column files) and reads are fast (you read only the columns you need, already laid out contiguously), but parts accumulate. ClickHouse merges parts in the background into larger parts, keeping the total part count bounded. The MergeTree engine’s core invariant is: rows within a part are sorted by the primary key, and after a merge, the resulting part is still sorted.

If you insert too frequently in small batches (say, one row at a time), you create too many parts and ClickHouse will start rejecting inserts with “too many parts” errors. The intended write pattern is batches of at least 1000 rows, and ideally 100k or more.

Primary Key: Sparse Index, Not Unique Constraint

ClickHouse’s primary key is not a unique constraint. It is a sparse index. For every 8192 rows (one granule, the default index_granularity), ClickHouse writes one index entry pointing to the starting row values of that granule. The index itself fits in memory for most tables.

When a query arrives with a WHERE clause on primary key columns, ClickHouse uses the sparse index to identify which granules might contain matching rows, then reads only those granules from disk. Everything else is skipped entirely at the storage layer.

// Simulating the query planning decision in TypeScript terms
interface Granule {
  startKey: [string, number]; // e.g., (user_id, timestamp)
  fileOffset: number;
  rowCount: number; // always <= 8192
}

function findCandidateGranules(
  index: Granule[],
  keyRange: { from: [string, number]; to: [string, number] }
): Granule[] {
  // Binary search for the first granule where startKey >= keyRange.from
  // and last granule where startKey <= keyRange.to
  return index.filter(
    (g) =>
      g.startKey >= keyRange.from && g.startKey <= keyRange.to
  );
}

The sparse index is why primary key column order matters so much. Cardinality rules apply: put low-cardinality columns first (tenant, date) so the index eliminates large swaths of data before the high-cardinality columns refine further.

Data Skipping Indexes

Beyond the primary sparse index, ClickHouse supports secondary data skipping indexes stored per granule block. Common types:

• minmax: stores min and max value of a column per block. Eliminates blocks where the queried value is outside range.
• set(N): stores up to N distinct values per block. Eliminates blocks that definitely do not contain a value.
• bloom_filter: probabilistic membership test. Useful for high-cardinality string lookups.
• ngrambf_v1 and tokenbf_v1: Bloom filters over n-gram or token decompositions of strings.

These indexes are skipping indexes, not lookup indexes. They eliminate blocks that cannot contain results. They never guarantee a block contains results. False positives are expected; false negatives are not.

Columnar Compression Codecs

Each column file is compressed with a codec chain. The defaults are LZ4 for general data. ClickHouse ships specialized codecs that are applied before the general compressor:

• Delta: stores differences between adjacent values instead of the values themselves. Highly effective for timestamps and monotonically increasing integer sequences.
• DoubleDelta: second-order difference, for sequences with near-constant rate of change.
• Gorilla: XOR-based float compression, same concept used in Prometheus.
• T64: transposes bits across a block of 64 integers to expose patterns that improve subsequent compression.

A column declared as DateTime CODEC(Delta, LZ4) first delta-encodes the timestamps, then LZ4-compresses the delta stream. Timestamp columns routinely compress to 1-2 bytes per value this way, versus 8 bytes raw.

Vectorized Query Execution

ClickHouse processes data in column chunks rather than row-by-row. The execution unit is a Block: a set of columns, each containing a fixed number of rows (typically 65536, the default max_block_size). Every operator in the query pipeline consumes Blocks and produces Blocks.

The practical consequence is that tight inner loops operate over arrays of the same type. This is exactly the access pattern SIMD instructions require. A filter on an integer column becomes a loop over a contiguous int32 array, which the compiler (or explicit intrinsics) can vectorize to process 8 or 16 values per CPU cycle instead of one.

// Conceptual illustration of vectorized filter vs row-oriented filter
// Row-oriented: one comparison per loop iteration
function rowOrientedFilter(rows: Array<{ age: number; name: string }>, minAge: number) {
  return rows.filter((row) => row.age >= minAge);
}

// Column-oriented: operate over a contiguous typed array
function columnOrientedFilter(ages: Int32Array, minAge: number): Uint8Array {
  const mask = new Uint8Array(ages.length);
  for (let i = 0; i < ages.length; i++) {
    mask[i] = ages[i] >= minAge ? 1 : 0;
    // In C++ this loop becomes a SIMD AVX2 comparison, 8 int32s at once
  }
  return mask;
}

The pipeline model means operators are fused where possible. A filter followed by a projection does not materialize an intermediate Block; the selection mask from the filter is applied to the column arrays in-place as the projection reads them.

ClickHouse also exploits late materialization: string columns and other expensive-to-decompress columns are not read from disk until after cheaper filters have reduced the row set. This means a query with a selective filter on a numeric column followed by a SELECT on a string column only decompresses the string values for rows that passed the filter.

Distributed Query Processing

A single ClickHouse node can handle impressive workloads, but production deployments are typically distributed across multiple shards, each with one or more replicas.

Sharding and Distributed Tables

The Distributed table engine is a virtual table that routes queries across shards. It does not store data itself. Each shard stores a local MergeTree (or ReplicatedMergeTree) table, and the Distributed table points at all of them.

-- Local table on each shard node
CREATE TABLE events_local ON CLUSTER my_cluster (
    tenant_id   UInt32,
    event_ts    DateTime,
    user_id     UInt64,
    event_type  LowCardinality(String)
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
PARTITION BY toYYYYMM(event_ts)
ORDER BY (tenant_id, event_ts, user_id);

-- Distributed table: query this, writes route to shards
CREATE TABLE events ON CLUSTER my_cluster AS events_local
ENGINE = Distributed(my_cluster, default, events_local, rand());

The sharding key (rand() above, but typically a hash of a column) determines which shard receives each inserted row. Choose the sharding key to co-locate data that is frequently joined or grouped together.

Two-Phase Aggregation

When you run SELECT count(), sum(amount) FROM events GROUP BY tenant_id, the Distributed engine sends the query to all shards. Each shard runs a partial aggregation over its local data and returns partial aggregate states. The initiating node merges the partial states into final results.

This two-phase model works well for most aggregations (count, sum, avg, min, max). It breaks down for exact distinct counts and quantiles, which require either approximation functions (uniq, quantile) or a distributed_group_by_no_merge = 0 setting that forces full data movement to the initiating node.

For high-cardinality GROUP BY on distributed data, the distributed_group_by_no_merge setting matters. Setting it to 1 skips the merge step and returns partial results per shard. This is useful for debugging and for queries where you control the sharding key to guarantee that all rows for a group reside on one shard.

Replication via ClickHouse Keeper

ReplicatedMergeTree uses ClickHouse Keeper (or ZooKeeper) as a coordination service. Each replica registers itself in Keeper and watches for operations from other replicas. When a replica finishes writing a new part, it creates a Keeper node describing the part. Other replicas fetch the part from the first replica and apply it locally.

Replication is asynchronous by default. An insert that completes on one replica may not be visible on another for a moment. The insert_quorum setting makes inserts synchronous: the insert only succeeds after N replicas have confirmed they have the data. This increases insert latency but eliminates the inconsistency window.

Materialized Views and Projections

Materialized Views

A materialized view in ClickHouse is a trigger that fires on insert. When rows arrive in the source table, the view’s SELECT query runs over the new rows and inserts the result into a target table. This happens synchronously within the same insert transaction.

-- Source table: raw events
-- Materialized view: precomputed hourly counts per tenant
CREATE MATERIALIZED VIEW events_by_hour_mv
TO events_by_hour
AS SELECT
    tenant_id,
    toStartOfHour(event_ts) AS hour,
    event_type,
    count()                 AS event_count
FROM events_local
GROUP BY tenant_id, hour, event_type;

Queries on events_by_hour scan the precomputed table instead of aggregating billions of raw rows. The read path becomes trivially fast. The cost is insert overhead (one additional aggregation per insert batch) and storage for the precomputed table.

Materialized views are a core performance primitive in ClickHouse, not an optimization of last resort. Designing the right set of materialized views is usually the highest-leverage tuning activity for a new deployment.

Projections

Projections are an alternative approach introduced later in ClickHouse’s development. A projection defines an alternative sort order and optional aggregation stored within the same physical part files as the base table. The query planner automatically selects a projection when it reduces the scan cost versus the primary index.

Unlike materialized views (which require explicit query routing to a separate table), projections are transparent to the query layer. The tradeoff is that projections cannot express aggregations as flexible as materialized views, and they add write amplification proportional to the number of projections defined.

Production Considerations

Partition Granularity

The PARTITION BY key controls how parts are grouped for retention, detach/attach operations, and TTL. A common mistake is partitioning by day when the data volume and query patterns call for monthly or weekly partitions. Too many partitions means too many parts, higher memory pressure on the part metadata, and slower merges.

Rule of thumb: you want 1000-3000 active parts per table total. If you see part counts in the tens of thousands, the partition key is too granular.

Merge Tuning

Background merges are controlled by background_pool_size (number of merge threads) and merge_max_block_size. Under heavy insert load, merges may fall behind. Signs of falling behind include growing part counts and increasing query latency on scans that touch many small parts.

Increasing background_pool_size helps on nodes with spare CPU. The max_bytes_to_merge_at_max_space_in_pool setting bounds how large a single merge can grow, which affects tail latency of merges but also the final part sizes.

Memory Limits

ClickHouse is aggressive about using available memory for aggregation hash tables and caches. The max_memory_usage setting (per query) and max_server_memory_usage (global) are essential guardrails in multi-tenant deployments. Without them, a single heavy analytical query can OOM the server.

For GROUP BY queries that exceed max_bytes_before_external_group_by, ClickHouse spills hash table data to disk. This is significantly slower but avoids OOM. Set it to roughly half of max_memory_usage.

// Example: constructing ClickHouse query settings for controlled execution
const querySettings = {
  max_memory_usage: 8 * 1024 * 1024 * 1024, // 8 GB per query
  max_bytes_before_external_group_by: 4 * 1024 * 1024 * 1024, // 4 GB before spill
  max_execution_time: 30, // seconds
  distributed_group_by_no_merge: 0, // full merge for correctness
  optimize_move_to_prewhere: 1, // push filters before decompression
};

async function runAnalyticsQuery(
  client: ClickHouseClient,
  sql: string
): Promise<unknown[]> {
  const result = await client.query({
    query: sql,
    query_params: {},
    clickhouse_settings: querySettings,
  });
  return result.json();
}

The optimize_move_to_prewhere Setting

ClickHouse’s PREWHERE clause is the manual version of what this setting automates. PREWHERE filters are applied before decompressing all other columns: ClickHouse reads only the PREWHERE columns, evaluates the filter, then reads remaining columns only for surviving rows. For selective filters on cheap columns, this is a major win. With optimize_move_to_prewhere = 1, ClickHouse automatically promotes suitable WHERE conditions to PREWHERE without you changing SQL.

Tradeoffs: ClickHouse vs Alternatives

Dimension	ClickHouse	PostgreSQL	DuckDB	Apache Druid	TimescaleDB
Processing model	Columnar, vectorized, distributed	Row-oriented, single-node primary	Columnar, vectorized, in-process	Columnar, pre-segment aggregation	Row+columnar hybrid, TimescaleDB extension
Write model	Batched parts, async merge	WAL-driven heap, MVCC	Append or in-process mutation	Real-time ingestion + deep storage	WAL-driven, hypertable chunks
Read latency	Sub-second on billions of rows	Fast for OLTP, slow on large analytics scans	Fast for local files and moderate data sizes	Sub-second on pre-aggregated dimensions	Fast for time-range scans, slower for ad-hoc
Distributed	Yes, native sharding + replication	Read replicas only (no distributed query)	No (single process)	Yes, native Historicals + Brokers	Yes, via Citus or read replicas
Exact distinct counts at scale	Expensive (full shuffle) or approximate	Expensive	Exact, in-process	Approximate (HyperLogLog default)	Exact, but slower at scale
Operational complexity	Medium (Keeper, schema design)	Low (single node)	Very low (no server)	High (many moving parts)	Low to medium
SQL compatibility	Mostly ANSI, some gaps	Full ANSI + extensions	Full ANSI	Limited (no joins in real-time)	Full PostgreSQL compatibility
Sweet spot	Real-time analytics on event streams, multi-tenant reporting	Transactional OLTP, moderate analytics	Local analytics, data engineering, embedded	Pre-aggregated dashboards, gaming/ad-tech	Time-series with transactional access patterns

Closing

ClickHouse’s performance is not magic. It is the result of consistently choosing the right data structure for analytical access patterns at every layer: column files instead of row pages, sparse indexes instead of B-trees, merge semantics instead of in-place updates, vectorized loops instead of interpreted tuple-at-a-time evaluation. The distributed model extends the same ideas across nodes with a coordination layer that is narrow in scope and well-understood in behavior.

The system rewards engineers who understand its internals. Schema design (primary key order, partition granularity, codec selection), materialized view architecture, and insert batch sizing all compound. Getting them right on the first deployment avoids a class of performance problems that are painful to fix retroactively on a live cluster.