ClickHouse¶

03:14 AM. Grafana's checkout-service tile times out. EXPLAIN indexes = 1 on the query behind it shows every granule in the partition selected — none skipped — for a filter on service = 'checkout'. The table has an ORDER BY. The predicate is right there in the WHERE clause.

Predict before you read on: (A) ORDER BY doesn't include service at all, (B) service is in ORDER BY but not first, (C) the query wraps service in a function, or (D) the primary index is just too small for the data volume?

Hundreds of millions of events land every day, Grafana wants p95 latency by endpoint in a few hundred milliseconds, and Postgres cannot answer that chart as a lifestyle — ClickHouse can, but only because ORDER BY is the index, and the design decision you cannot postpone is which column goes first.

Start with the situation¶

Observability / SaaS analytics:

{timestamp, customer_id, service, endpoint, status_code, latency_ms, bytes, trace_id}

~400 million events/day. Queries that must be fast:

-- Q1 fleet: one service, recent time
SELECT endpoint, count(), quantile(0.95)(latency_ms)
FROM requests
WHERE service = 'checkout'
  AND timestamp >= now() - INTERVAL 1 HOUR
GROUP BY endpoint;

-- Q2 tenant: one customer, recent time
SELECT status_code, count()
FROM requests
WHERE customer_id = 'bigcorp'
  AND timestamp >= now() - INTERVAL 6 HOUR
GROUP BY status_code;

-- Q3 wide scan: whole fleet, one hour (on-call wall)
SELECT service, count(), avg(latency_ms)
FROM requests
WHERE timestamp >= now() - INTERVAL 1 HOUR
GROUP BY service;

If the product is customer-facing 5k QPS of Q2, read Pinot too. If the product is internal dashboards plus ad-hoc SQL, stay here.

Why the obvious approach breaks¶

The engine must:

• ingest hundreds of thousands of rows/s in batches without creating a million tiny files;
• skip almost all bytes for Q1/Q2;
• still answer Q3 (which matches time, not tenant);
• survive a restart without a 14-hour REPAIR.

Row stores give you B-trees per query. ClickHouse gives you one physical order per table (plus optional projections). That order cannot be equally perfect for Q1, Q2, and Q3. You pick the revenue query and you copy or project the others.

Build the mental picture¶

Writes do not update pages. They create immutable parts. Background merges compact parts into larger parts. Reads walk a sparse primary index built on the sort key, jump to granules (~8,192 rows), and decode only the columns the query named.

flowchart LR
    I1[INSERT batch] --> P1[Part A]
    I2[INSERT batch] --> P2[Part B]
    I3[INSERT batch] --> P3[Part C]
    P1 --> M[merge]
    P2 --> M
    P3 --> M
    M --> P4[Larger part]

ORDER BY is not an SQL nicety. It is the primary index. Rows in a part are stored in that order. The index stores one key every index_granularity rows (a mark). Query planning is: binary-search marks, skip granules that cannot match, read the rest.

Simulate this: ORDER BY explorer.

Under the hood¶

MergeTree parts¶

Each part is a directory:

20240612_0_120_3/
    timestamp.bin     latency_ms.bin     service.bin     ...
    timestamp.mrk2    latency_ms.mrk2    service.mrk2
    primary.idx
    columns.txt       checksums.txt      count.txt

Parts are immutable. A merge writes a new part and deletes the inputs after readers drain.

Granules, marks, sparse primary index¶

Default granule = 8,192 rows. 8 million rows → ~1,000 marks. The primary index for that part fits in memory easily. It is not a B-tree of every row. If 8,000 consecutive rows share the same service and your query filters a different service, one mark comparison skips all 8,000.

If those 8,000 rows are a mix of every service (because you ordered only by time), the mark’s min/max for service is “almost everything” and you skip nothing.

EXPLAIN indexes = 1 shows granules selected vs total. That is the first debug tool, not EXPLAIN in the Postgres sense.

ORDER BY vs PARTITION BY¶

Typical observability:

PARTITION BY toYYYYMMDD(timestamp)   -- drop a day, prune by day
ORDER BY (service, endpoint, timestamp)  -- skip inside the day

Anti-pattern: PARTITION BY timestamp (DateTime) → one partition per second. Anti-pattern: 10,000 partitions “for prune.” Merge and Keeper metadata will explode. Hundreds of partitions is a lot; thousands is an incident.

Worked ORDER BY for observability¶

Same table, three keys. Same three queries. This is the lesson.

A. ORDER BY (timestamp)¶

Rows are a time stream. Granules are “a few seconds of the fleet.”

CREATE TABLE requests_by_time (...)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (timestamp);

Use only if almost every query is a time range over the whole fleet (on-call wall) and never “one tenant” or “one service.” That is rare in SaaS.

B. ORDER BY (service, endpoint, timestamp)¶

Granules are “this endpoint of this service, in time order.”

CREATE TABLE requests_by_svc (...)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (service, endpoint, timestamp);

This is the default for SRE / service dashboards. PARTITION BY day still lets Q3 drop old days.

C. ORDER BY (customer_id, timestamp)¶

Granules are “this tenant’s stream.”

CREATE TABLE requests_by_cust (...)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (customer_id, timestamp);

This is the default for tenant product analytics (and a prerequisite if you later expose charts to customers). BigCorp 40% of traffic is then one dense key range — good for their dashboard, a hot range on one shard if you sharded poorly.

Hybrid that teams actually ship¶

You cannot have B and C in one ORDER BY. Options:

• Two tables (or a materialized view) from the same Kafka sink: SRE table (service, endpoint, timestamp), product table (customer_id, timestamp) or (customer_id, service, timestamp).
• Projection on one table (ClickHouse projections store another sort order).
• If Q2 is rare, keep B and accept tenant scans on a day partition.

ALTER TABLE requests_by_svc ADD PROJECTION p_cust
(
    SELECT *
    ORDER BY (customer_id, timestamp)
);
ALTER TABLE requests_by_svc MATERIALIZE PROJECTION p_cust;

Projections cost disk and merge CPU. They are still cheaper than the wrong primary key.

Mutations vs collapsing / replacing¶

ALTER TABLE ... UPDATE/DELETE is a mutation: rewrite parts in the background. It is not Postgres UPDATE. Concurrent reads may see old values until parts finish. Mutations on a hot table are how on-call spends the weekend.

For “latest row wins” facts:

CREATE TABLE customer_dims
(
    customer_id String,
    plan        LowCardinality(String),
    updated_at  DateTime,
    version     UInt64
)
ENGINE = ReplacingMergeTree(version)
ORDER BY (customer_id);

ReplacingMergeTree does not make rows unique at query time. Dedup happens during merges. Until then you have duplicates.

SELECT customer_id, plan FROM customer_dims FINAL;  -- correct, expensive
SELECT customer_id, argMax(plan, version)           -- usually better
FROM customer_dims
GROUP BY customer_id;

CollapsingMergeTree / AggregatingMergeTree are for signed rows and incremental states. Same caveat: correctness after merge, or via FINAL / explicit -State / -Merge aggregates.

Distributed tables and the sharding key¶

-- on each shard: local MergeTree
CREATE TABLE requests ON CLUSTER '{cluster}' (...)
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (service, endpoint, timestamp);

-- router
CREATE TABLE requests_d ON CLUSTER '{cluster}' AS requests
ENGINE = Distributed('{cluster}', currentDatabase(), requests, cityHash64(customer_id));

The last argument is the sharding key. rand() spreads load and destroys locality for tenant queries (every shard scanned). cityHash64(customer_id) colocates a tenant; BigCorp becomes a hot shard.

Replication ≠ sharding. Replicas are copies of a shard (Keeper coordinates). Distributed is a fan-out query. SELECT on the Distributed table hits every shard unless the optimizer can infer a shard from the key (often it cannot for WHERE service = ...).

Inserts are batches¶

One INSERT = at least one part (more with partitions). Target tens of thousands of rows per insert, not one. Kafka Engine + materialized view batches for you; HTTP insert from a naive app does not.

CREATE TABLE kafka_requests (...) ENGINE = Kafka
SETTINGS kafka_broker_list = 'kafka:9092',
         kafka_topic_list = 'requests',
         kafka_group_name = 'ch-requests',
         kafka_format = 'JSONEachRow',
         kafka_num_consumers = 4;

CREATE MATERIALIZED VIEW requests_mv TO requests AS
SELECT * FROM kafka_requests;

Production alternatives: Vector/Redpanda Connect/NiFi inserting batches; Flink sink with clickhouse-jdbc batching; object storage → INSERT SELECT from S3. All of them batch. The Kafka engine is not mandatory.

Put it to work¶

Full serving table for SRE dashboards:

CREATE TABLE requests
(
    timestamp    DateTime64(3),
    customer_id  LowCardinality(String),
    service      LowCardinality(String),
    endpoint     LowCardinality(String),
    status_code  UInt16,
    latency_ms   UInt32,
    bytes        UInt64,
    trace_id     String CODEC(ZSTD(1))
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(timestamp)          -- monthly if you TTL by month; daily if you drop days
ORDER BY (service, endpoint, timestamp)
TTL timestamp + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;

-- skip index if you still filter customer on this table
ALTER TABLE requests
    ADD INDEX idx_cust customer_id TYPE bloom_filter GRANULARITY 4;

TTL:

-- drop raw after 90d; keep a rollup forever
ALTER TABLE requests
    MODIFY TTL timestamp + INTERVAL 90 DAY;

CREATE TABLE requests_1h
(
    hour DateTime,
    service LowCardinality(String),
    endpoint LowCardinality(String),
    c UInt64,
    lat_sum UInt64
)
ENGINE = SummingMergeTree
PARTITION BY toYYYYMM(hour)
ORDER BY (service, endpoint, hour);

CREATE MATERIALIZED VIEW requests_1h_mv TO requests_1h AS
SELECT
    toStartOfHour(timestamp) AS hour,
    service,
    endpoint,
    count() AS c,
    sum(latency_ms) AS lat_sum
FROM requests
GROUP BY hour, service, endpoint;

Query the rollup for “last 30 days” charts; raw for last few hours.

Where teams get caught¶

How it fails¶

How to investigate¶

Parts and merges:

SELECT partition, count() AS parts, sum(rows) AS rows,
       formatReadableSize(sum(bytes_on_disk)) AS disk
FROM system.parts
WHERE table = 'requests' AND active
GROUP BY partition
ORDER BY partition DESC
LIMIT 14;

SELECT * FROM system.merges WHERE table = 'requests';

Marks / index usage:

EXPLAIN indexes = 1
SELECT endpoint, count()
FROM requests
WHERE service = 'checkout'
  AND timestamp >= now() - 3600
GROUP BY endpoint;

You want primary key granules far below total granules. If selected ≈ total, the predicate is not a prefix of ORDER BY (or you wrapped the column in a function).

Query log:

SELECT
    type, query_duration_ms,
    read_rows, formatReadableSize(read_bytes) AS read,
    result_rows, memory_usage, query
FROM system.query_log
WHERE event_time > now() - 600 AND type = 'QueryFinish'
ORDER BY query_duration_ms DESC
LIMIT 20;

read_rows ≈ all rows in the time partition → sort key miss. memory_usage huge → aggregation cardinality.

Replication:

SELECT database, table, is_leader, absolute_delay, queue_size, log_max_index, log_pointer
FROM system.replicas;

Scale 10× / 100× / 1000׶

Start: 40 million events/day, one node, ORDER BY (service, endpoint, timestamp), daily partitions, batched Kafka.

Hardware scale without the right ORDER BY is paying 1000× IO forever.

Trade-offs¶

Alternatives¶

Apply¶

Before CREATE TABLE:

1. Write the three queries with real predicates.
2. Put equality columns first in ORDER BY, time usually last.
3. PARTITION BY a coarse time bucket you will drop.
4. Batch inserts; never one row from the app.
5. Decide uniqueness: none (facts), argMax (dims), not FINAL.
6. TTL raw; MV rollups for long charts.
7. If tenants and SREs both need speed, two sort orders, not one compromise key.

At work you will recognise this problem when Grafana is slow and read_bytes ≈ table size. That is not “need more CPU.” That is the wrong primary key.

Practice the idea¶

Compare two sort keys in the ClickHouse ORDER BY explorer, then run the ClickHouse lab against two real tables. Use rows or marks read—not elapsed time alone—to explain the result. The slow-query incident is the final diagnostic pass.

Check your understanding¶

Observability cluster, 500 million events/day, queries:

1. Service owners: WHERE service = ? AND timestamp > now()-1h GROUP BY endpoint (90% of QPS).
2. Support: WHERE customer_id = ? AND timestamp > now()-1d (9%).
3. Exec wall: WHERE timestamp > now()-1h GROUP BY service (1%).

Shard count will be 4 in six months. Pick PARTITION BY, ORDER BY, sharding key, and what you do for (2). Say what happens to BigCorp (8% of events) and what EXPLAIN indexes = 1 should show for (1).