Every Millisecond Counts

On this pageReplacingMergeTree Is No FunThe Weird Partition KeyOrder By Slightly ChangedRemoving an Expensive JoinReading Fewer ColumnsTemplatingAggressive MergesReading Fewer RowsOk But WhyImagine you join a new project with a clear goal, improve ClickHouse performance, so our biggest customers can run their queries in a reasonable amount of time. This is something you’ve done lots of times before. You ask for the most expensive query since it’s usually the one that will have the most impact and a few low-hanging fruit improvements. You run it and it takes more than a minute to complete. 1 row in set. Elapsed: 85.715 sec. Processed 1.96 billion rows, 198.69 GB Peak memory usage: 23.05 GiB. The query is one of the twelve being run in parallel, each computing its own metric or sparkline. And they are in the first screen a client lands on when accessing the application. Imagine a client looking at a spinner for at least a couple of minutes right after logging in. This is a note on all the changes, learnings, and optimizations we’ve made during the last four months to bring that query to sub-second latency. Changes were simple and incremental. None of them was a clever trick, just basic improvements compounding on each other. ReplacingMergeTree Is No Fun The setup is simple, events get written to a ClickHouse table directly from Kafka. Events are mutable, any event at any given point in time can be updated or deleted1. So, events changing past history are being received at similar or even faster rates than new events. Exactly what a database based on immutable storage is not for. We are forced to use ReplacingMergeTree engine and make heavy use of FINAL, that’s where most of the time goes. On top of that, there are a few metrics that require going through the entire history of a client’s events to be computed. History that is always increasing since they keep their activity and generate events every day. For this reason, some numbers in this post may vary. An optimization that took a query from 200GB to 150GB could later be seen reading 210GB due to the amount of data received between one optimization and the next2. The Weird Partition Key The table was originally partitioned by month: PARTITION BY toYYYYMM(event_created_at) This is the obvious choice, but the wrong one in many ways for this case. Events come in batches, and those batches impact any point in time. A single batch will contain events from multiple months, writing a lot of files per insert and causing the usual snowball effect: increasing the number of files real-time queries have to read, adding pressure to merges, etc. Also, FINAL merges rows together, and its parallelism depends a lot on how keys are distributed across partitions and how they overlap, the worst case scenario is a single thread doing the final merge and taking a lot of time. Partitioning by time means that a client’s events are spread across every partition in the table, so ClickHouse has to spend a considerable amount of time building the pipeline and splitting ranges to reduce intersections. An example of the performance impact of this partitioning causing FINAL to be performed by a single thread: FINAL performance using a single thread FINAL performance using the right parallelism To solve both issues while gaining a bit of performance, we applied the following partitioning strategy: PARTITION BY (client_customer_id % 36) Note that client_customer_id is not our client’s id but our client’s client id, we don’t use this column in our queries. The strategy is counterintuitive3 but it works well for our use case: Keeps insert performance under control by “only” creating 36 parts per insert Partitions are equally distributed. There are no huge partitions containing whale clients and slowing down queries for every client landing in the same partition Ensures FINAL parallelism without requiring ClickHouse to compute ranges intersections. Which means that we can disable split_parts_ranges_into_intersecting_and_non_intersecting_final and split_intersecting_parts_ranges_into_layers_final saving a few extra seconds on the data pipeline building step We give up partition pruning entirely, since we never filter by client_customer_id. There is a cost, but not for metrics that have to read a client’s full history anyway. Partitioning by month was not pruning anything either. And as mentioned above, we gain a lot in parallelism Order By Slightly Changed While checking how every event_type contributes to each metric, we discovered that a few of them didn’t contribute but we were still reading and processing them. They were ~16% of the rows on the table. Not a huge difference but an obvious change that could improve query performance. Promoting event_type to an earlier position is not an ALTER. The sorting key is fixed when the table is created, so this meant a new table and a full rewrite. Behaviour stayed the same, since event_type was already in the key and only its position moved, so ReplacingMergeTree still collapses exactly the rows it collapsed before. After the change, we saw a significant improvement in query performance: Before4 1 row in set. Elapsed: 13.43 sec. Processed 256.67 million rows, 18.82 GiB Peak memory usage: 18.78 GiB. After 1 row in set. Elapsed: 11.44 sec. Processed 237.86 million rows, 15.67 GiB Peak memory usage: 14.04 GiB. The query latency improved by ~15% while reading ~7% fewer rows. That explains why we don’t see the full ~16%: the sorting key filters out granules, and if a granule contains even one row of an event_type we are interested in, we have to read the remaining rows from that granule. Removing an Expensive Join The query had a few expensive joins, almost all of them behind a trick that let us skip the real table without removing the join. The right side is a subquery gated by a parameter, so when the branch isn’t needed the filter is false and it returns nothing: ANY LEFT JOIN ( SELECT ... FROM table_2 WHERE {has_currency:Boolean} = 1 AND ... ) USING ... Executing that costs a few milliseconds, parsing and planning it a few more. That’s what the templating section is about. There was a join though that was being performed against a real table on every single call. The join was simply retrieving a couple of fields used to compute some intermediate values used for the metrics. ANY LEFT JOIN ( SELECT client_id, internal_id, latest_a_field, latest_another_field FROM table_derived_from_original_table FINAL PREWHERE client_id IN {client_ids:Array(UInt64)} ) USING client_id, internal_id A second FINAL on a table derived from the original one to compute the latest values of the two fields we were interested in. ClickHouse builds the hash table for the right side and puts it in memory, the join is relatively fast but the cost is not only reading but also in memory footprint. The fix was easier than it seems. The intermediate column computed with these two values could be precomputed at application level and sent as part of the event. The queries are now simpler and the join can be removed entirely. 1 row in set. Elapsed: 2.46 sec. Processed 49.86 million rows, 5.60 GiB Peak memory usage: 1.09 GiB Memory went down almost 13x, and query time improved by almost 5x. Reading Fewer Columns We have seen how precomputing a column at application level can save us a lot of memory and CPU. Why should we stop there? Turns out that some of the metrics, especially the ones that require to read full client’s history, were using 9 columns. This is not a big deal with row-based databases but ClickHouse is columnar and reading a large number of columns means accessing extra data that could be avoided otherwise. After precomputing the metrics contribution for each row at application level5, the impact was clear. 1 row in set. Elapsed: 1.38 s. Processed 49.69 million rows, 3.46 GiB Peak memory usage: 582.40 MiB The 9 columns came to roughly 38 bytes per row. Replacing them with an 8-byte column saves 1.5GB of reads on its own. Add the extra files you avoid reading, indices and marks, plus the seeks you save. And you get another ~2x improvement. Templating At this point, we have managed to go from 13.43s to 1.38s, that’s almost a 10x improvement by just compounding optimizations that we managed to apply in a few weeks. During the process, we have observed a few things that need to be addressed with a more aggressive approach. One of the biggest bottlenecks we identified was related to how queries are being executed. We were using nested ClickHouse parameterized views, three levels deep generating huge queries that were re-parsed and re-planned on every single request. Those views were passing parameters and rendering some optional branches like the joins mentioned above. No matter what we tried, the branches were still there consuming CPU even if it was not being used. Measuring the impact was easy. Building a minimal version of the query and running it against a client without data. Execution is almost zero, so everything left is parse, AST, and pipeline building6: Metric Original Simplified Delta metric_1 0.448s 0.029s 420ms metric_2 1.345s 0.051s 1.3s 420ms of fixed overhead on a query whose average execution time was 740ms. More than half of the time was spent on preparing the query. So we replaced the views with flat query templates. dbt compiles each metric into a single self-contained statement (from the same macros that build the parameterized views, so there’s one source of truth) leaving second-stage markers with bracket delimiters, which pass through dbt’s own curly-brace Jinja untouched: SELECT [% if has_currency %] {currency:String} as display_currency, coalesce(nullIf(exchange_rate, 0), 1) [% else %] 'USD' as display_currency, 1 [% endif %] AS exchange_rate, money_field_in_usd * exchange_rate as money_field_in_display_currency FROM original_table FINAL PREWHERE client_id IN {client_ids:Array(UInt64)} [% if has_currency %] ANY LEFT JOIN ( SELECT date, exchange_rate