title_wip

[Edit this page](https://github.com/ClickHouse/clickhouse-docs/blob/main/docs/guides/best-practices/query-parallelism.md)

ClickHouse is [built for speed](/docs/concepts/why-clickhouse-is-so-fast). It executes queries in a highly parallel fashion, using all available CPU cores, distributing data across processing lanes, and often pushing hardware close to its limits.

This guide walks through how query parallelism works in ClickHouse and how you can tune or monitor it to improve performance on large workloads.

We use an aggregation query on the [uk\_price\_paid\_simple](/docs/parts) dataset to illustrate key concepts.

## Step-by-step: How ClickHouse parallelizes an aggregation query[​](#step-by-step-how-clickHouse-parallelizes-an-aggregation-query "Direct link to Step-by-step: How ClickHouse parallelizes an aggregation query")

When ClickHouse ① runs an aggregation query with a filter on the table's primary key, it ② loads the primary index into memory to ③ identify which granules need to be processed, and which can be safely skipped:

![Index analysis](/docs/assets/images/query-parallelism_01-04167e71e6fb8ce3a9b12a2c11256d4a.gif)

### Distributing work across processing lanes[​](#distributing-work-across-processing-lanes "Direct link to Distributing work across processing lanes")

The selected data is then [dynamically](#load-balancing-across-processing-lanes) distributed across `n` parallel [processing lanes](/docs/academic_overview#4-2-multi-core-parallelization), which stream and process the data [block](/docs/development/architecture#block) by block into the final result:

![4 parallel processing lanes](/docs/assets/images/query-parallelism_02-82fba4a0fd87b980b12a61b59f4533f4.gif)

  
  

The number of `n` parallel processing lanes is controlled by the [`max_threads`](/docs/operations/settings/settings#max_threads) setting, which by default matches the number of cores (threads) of a single CPU available to ClickHouse on the server. In the example above, we assume `4` cores.

On a machine with `8` cores, query processing throughput would roughly double (but memory usage would also increase accordingly), as more lanes process data in parallel:

![8 parallel processing lanes](/docs/assets/images/query-parallelism_03-2120da7d58c366bd09295e0bc6bc9b20.gif)

  
  

Efficient lane distribution is key to maximizing CPU utilization and reducing total query time.

### Processing queries on sharded tables[​](#processing-queries-on-sharded-tables "Direct link to Processing queries on sharded tables")

When table data is distributed across multiple servers as [shards](/docs/shards), each server processes its shard in parallel. Within each server, the local data is handled using parallel processing lanes, just as described above:

![Distributed lanes](/docs/assets/images/query-parallelism_04-fdd68c179689c7410287062a72440de8.gif)

  
  

The server that initially receives the query collects all sub-results from the shards and combines them into the final global result.

Distributing query load across shards allows horizontal scaling of parallelism, especially for high-throughput environments.

ClickHouse Cloud uses parallel replicas instead of shards

In ClickHouse Cloud, this same parallelism is achieved through [parallel replicas](https://clickhouse.com/docs/deployment-guides/parallel-replicas), which function similarly to shards in shared-nothing clusters. Each ClickHouse Cloud replica—a stateless compute node—processes a portion of the data in parallel and contributes to the final result, just like an independent shard would.

## Monitoring query parallelism[​](#monitoring-query-parallelism "Direct link to Monitoring query parallelism")

Use these tools to verify that your query fully utilizes available CPU resources and to diagnose when it doesn't.

We're running this on a test server with 59 CPU cores, which allows ClickHouse to fully showcase its query parallelism.

To observe how the example query is executed, we can instruct the ClickHouse server to return all trace-level log entries during the aggregation query. For this demonstration, we removed the query's predicate—otherwise, only 3 granules would be processed, which isn't enough data for ClickHouse to make use of more than a few parallel processing lanes:

```
SELECT
   max(price)
FROM
   uk.uk_price_paid_simple
SETTINGS send_logs_level='trace';
```

```
① <Debug> ...: 3609 marks to read from 3 ranges
② <Trace> ...: Spreading mark ranges among streams
② <Debug> ...: Reading approx. 29564928 rows with 59 streams
```

We can see that

- ① ClickHouse needs to read 3,609 granules (indicated as marks in the trace logs) across 3 data ranges.
- ② With 59 CPU cores, it distributes this work across 59 parallel processing streams—one per lane.

Alternatively, we can use the [EXPLAIN](/docs/sql-reference/statements/explain#explain-pipeline) clause to inspect the [physical operator plan](/docs/academic_overview#4-2-multi-core-parallelization)—also known as the "query pipeline"—for the aggregation query:

```
EXPLAIN PIPELINE
SELECT
   max(price)
FROM
   uk.uk_price_paid_simple;
```

```
    ┌─explain───────────────────────────────────────────────────────────────────────────┐
 1. │ (Expression)                                                                      │
 2. │ ExpressionTransform × 59                                                          │
 3. │   (Aggregating)                                                                   │
 4. │   Resize 59 → 59                                                                  │
 5. │     AggregatingTransform × 59                                                     │
 6. │       StrictResize 59 → 59                                                        │
 7. │         (Expression)                                                              │
 8. │         ExpressionTransform × 59                                                  │
 9. │           (ReadFromMergeTree)                                                     │
10. │           MergeTreeSelect(pool: PrefetchedReadPool, algorithm: Thread) × 59 0 → 1 │
    └───────────────────────────────────────────────────────────────────────────────────┘
```

Note: Read the operator plan above from bottom to top. Each line represents a stage in the physical execution plan, starting with reading data from storage at the bottom and ending with the final processing steps at the top. Operators marked with `× 59` are executed concurrently on non-overlapping data regions across 59 parallel processing lanes. This reflects the value of `max_threads` and illustrates how each stage of the query is parallelized across CPU cores.

ClickHouse's [embedded web UI](/docs/interfaces/http) (available at the `/play` endpoint) can render the physical plan from above as a graphical visualization. In this example, we set `max_threads` to `4` to keep the visualization compact, showing just 4 parallel processing lanes:

Note: Read the visualization from left to right. Each row represents a parallel processing lane that streams data block by block, applying transformations such as filtering, aggregation, and final processing stages. In this example, you can see four parallel lanes corresponding to the `max_threads = 4` setting.

### Load balancing across processing lanes[​](#load-balancing-across-processing-lanes "Direct link to Load balancing across processing lanes")

Note that the `Resize` operators in the physical plan above [repartition and redistribute](/docs/academic_overview#4-2-multi-core-parallelization) data block streams across processing lanes to keep them evenly utilized. This rebalancing is especially important when data ranges vary in how many rows match the query predicates, otherwise, some lanes may become overloaded while others sit idle. By redistributing the work, faster lanes effectively help out slower ones, optimizing overall query runtime.

## Why max\_threads isn't always respected[​](#why-max-threads-isnt-always-respected "Direct link to Why max_threads isn't always respected")