Every Millisecond Counts

By Jordi Villar A deep dive into squeezing sub-second latency out of ClickHouse queries.

Imagine stepping into a new project with a singular, high-stakes objective: optimize ClickHouse performance so that the largest enterprise clients aren't staring at loading screens for minutes.

To start, you identify the "worst offender"—the most resource-heavy query—knowing that fixing it provides the highest ROI. The initial stats were staggering:

Initial Query Profile:

• Rows Processed: 1.96 billion
• Data Volume: 198.69 GB
• Peak RAM Usage: 23.05 GiB

This wasn't a standalone query; it was one of twelve running in parallel to generate metrics and sparklines for the application's primary dashboard. For the end-user, this meant a frustratingly long wait immediately after logging in.

Over the next four months, the goal was to move from "minutes" to "sub-second." The secret wasn't a single "silver bullet" but rather the magic compounding effect of basic optimizations.

---

The Struggle with ReplacingMergeTree

The architecture is straightforward: events stream from Kafka directly into ClickHouse. However, there is a catch: events are mutable. Data can be updated or deleted at any point in history.

This creates a fundamental conflict. ClickHouse is designed for immutable storage, yet we are receiving updates to historical data at rates equal to or faster than new arrivals. To handle this, we relied on the ReplacingMergeTree engine and the FINAL modifier.

The challenges were twofold:

• FINAL consumes the vast majority of processing time.
• Certain metrics require scanning a client's entire event history, which grows indefinitely.

Interestingly, the sheer volume of incoming data often masked progress. An optimization that reduced a read from $200\text{GB} \rightarrow 150\text{GB}$ might appear as $210\text{GB}$ a week later simply because the dataset grew.

---

Fixing the "Weird" Partition Key

Initially, the table used a standard time-based partition: ~~PARTITION BY toYYYYMM(event_created_at)~~

While intuitive, this was the wrong choice for this specific workload. Because events arrive in batches that span multiple months, each insert created numerous files. This triggered a "snowball effect," increasing the number of files the system had to read and putting immense pressure on the merge process.

Furthermore, the parallelism of FINAL depends on how keys are distributed. With time-based partitioning, a single client's data is scattered across every partition, often forcing ClickHouse to use a single thread for the final merge.

The Solution: A Counterintuitive Shift

We pivoted to a partitioning strategy based on a modulo of the customer's client ID:

PARTITION BY (client_customer_id % 36)

(Note: client_customer_id refers to the end-customer's ID, not our direct client's ID, and is not used in query filters.)

Why this worked:

1. Controlled Inserts: It limits the creation to a maximum of 36 parts per insert.
2. Balanced Load: Partitions are distributed evenly, preventing "whale" clients from creating massive, slow partitions.
3. Maximized Parallelism: It ensures FINAL can run across multiple threads without needing to calculate complex range intersections.

This allowed us to disable expensive pipeline steps like split_parts_ranges_into_intersecting_and_non_intersecting_final. We sacrificed partition pruning (since we don't filter by this ID), but for queries reading full histories, the trade-off was a massive win.

---

Refining the ORDER BY Clause

While analyzing event_type contributions, we realized some types were being processed despite contributing nothing to the final metrics.

Since the sorting key is immutable after table creation, we couldn't just run an ALTER command. We had to create a new table and rewrite the entire dataset to move event_type to an earlier position in the key.

The Results:

Note: The latency improvement ($\sim 15\%$) didn't perfectly match the row reduction ($\sim 7\%$) because ClickHouse filters by granules. If a granule contains even one relevant row, the entire granule must be read.

---

Eliminating Expensive Joins

The query contained several joins, many of which were "ghost joins"—subqueries gated by parameters that returned nothing if the branch wasn't needed:

ANY LEFT JOIN (
    SELECT ... 
    WHERE {parameter_gate} = true
)

While these were relatively cheap, they still added milliseconds to parsing and planning. However, one join was a genuine performance killer: a real join against a derived table used to fetch intermediate fields for metric calculations.

-- The expensive culprit
ANY LEFT JOIN (
    SELECT 
        client_id, 
        internal_id, 
        latest_a_field, 
        latest_another_field 
    FROM table_derived_from_original_
)

By identifying and removing these unnecessary overheads, we continued the march toward that sub-second goal.