Parallel Replicas

The figure below visualizes how dynamic coordination works:

1. Replicas let the coordinator node know that they're able to process
   tasks, they can also specify how much work they can process.- The coordinator assigns tasks to the replicas.

1. Replica 1 and 2 are able to finish their task very quickly. They
   will request another task from the coordinator node.- The coordinator assigns new tasks to replica 1 and 2.

1. All the replicas are now done with the processing of their task. They
   request more tasks.- The coordinator, using the announcements, checks what tasks remain to
     be process, but there are no remaining tasks.- The coordinator tells the replicas that everything has been processed.
       It will now merge all the mergeable states and respond to the query.

### Managing cache locality[​](#managing-cache-locality "Direct link to Managing cache locality")

The last remaining potential issue is how we handle cache locality. If the query
is executed multiple times, how can we ensure the same task gets routed to the
same replica? In the previous example, we had the following tasks assigned:

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Replica 1 Replica 2 Replica 3|  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | Part 1 g1, g6, g7 g2, g4, g5 g3|  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | | Part 2 g1 g2, g4, g5 g3|  |  |  |  | | --- | --- | --- | --- | | Part 3 g1, g6 g2, g4, g5 g3 | | | | | | | | | | | | | | | |

To ensure that the same tasks are assigned to the same replicas and can
benefit from the cache, two things happen. A hash of the part + set of
granules (a task) gets computed. A modulo of the number of replicas for the task
assignment gets applied.

On paper this sounds good, but in reality, a sudden load on one replica, or a
network degradation, can introduce tail latency if the same replica is
consistently used for executing certain tasks. If `max_parallel_replicas` is less
than the number of replicas, then random replicas are picked for query execution.

### Task stealing[​](#task-stealing "Direct link to Task stealing")

if some replica processes tasks slower than others, other replicas will try to
'steal' tasks that in principle belong to that replica by hash to reduce the
tail latency.

### Limitations[​](#limitations "Direct link to Limitations")

This feature has known limitations, of which the major ones are documented in
this section.

Note

If you find an issue which isn't one of the limitations given below, and
suspect parallel replica to be the cause, please open an issue on GitHub using
the label `comp-parallel-replicas`.

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Limitation Description|  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | Complex queries Currently parallel replica works fairly well for simple queries. Complexity layers like CTEs, subqueries, JOINs, non-flat query, etc. can have a negative impact on query performance.|  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | Small queries If you're executing a query that doesn't process a lot of rows, executing it on multiple replicas might not yield a better performance time given that the network time for the coordination between replicas can lead to additional cycles in the query execution. You can limit these issues by using the setting: [`parallel_replicas_min_number_of_rows_per_replica`](/docs/operations/settings/settings#parallel_replicas_min_number_of_rows_per_replica).| Parallel replicas are disabled with FINAL |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | | Projections aren't used together with Parallel replicas |  |  |  |  | | --- | --- | --- | --- | | High Cardinality data and complex aggregation High cardinality aggregation that needs to send much data can significantly slow down your queries.|  |  | | --- | --- | | Compatibility with the new analyzer The new analyzer might significantly slow down or speed up query execution in specific scenarios. | | | | | | | | | | | | | |

## Settings related to parallel replicas[​](#settings-related-to-parallel-replicas "Direct link to Settings related to parallel replicas")

|  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Setting Description|  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `enable_parallel_replicas` `0`: disabled  `1`: enabled  `2`: Force the usage of parallel replica, will throw an exception if not used.| `cluster_for_parallel_replicas` The cluster name to use for parallel replication; if you're using ClickHouse Cloud, use `default`.| `max_parallel_replicas` Maximum number of replicas to use for the query execution on multiple replicas, if a number lower than the number of replicas in the cluster is specified, nodes will be selected randomly. This value can also be overcommitted to account for horizontal scaling.|  |  |  |  | | --- | --- | --- | --- | | `parallel_replicas_min_number_of_rows_per_replica` Help limiting the number of replicas used based on the number of rows that need to be processed the number of replicas used is defined by:   `estimated rows to read` / `min_number_of_rows_per_replica`.| `enable_analyzer` Query execution with parallel replicas is supported only with enabled analyzer | | | | | | | | | | | |

## Investigating issues with parallel replicas[​](#investigating-issues-with-parallel-replicas "Direct link to Investigating issues with parallel replicas")

You can check what settings are being used for each query in the
[`system.query_log`](/docs/operations/system-tables/query_log) table. You can
also look at the [`system.events`](/docs/operations/system-tables/events)
table to see all the events that have occured on the server, and you can use the
[`clusterAllReplicas`](/docs/sql-reference/table-functions/cluster) table function to see the tables on all the replicas
(if you're a cloud user, use `default`).

```
SELECT
   hostname(),
   *
FROM clusterAllReplicas('default', system.events)
WHERE event ILIKE '%ParallelReplicas%'
```

Response