Parallel Replicas

[Edit this page](https://github.com/ClickHouse/clickhouse-docs/blob/main/docs/deployment-guides/parallel-replicas.mdx)

## Introduction[​](#introduction "Direct link to Introduction")

ClickHouse processes queries extremely quickly, but how are these queries
distributed and parallelized across multiple servers?

> In this guide, we will first discuss how ClickHouse distributes a query across
> multiple shards via distributed tables, and then how a query can leverage
> multiple replicas for its execution.

## Sharded architecture[​](#sharded-architecture "Direct link to Sharded architecture")

In a shared-nothing architecture, clusters are commonly split into
multiple shards, with each shard containing a subset of the overall data. A
distributed table sits on top of these shards, providing a unified view of the
complete data.

Reads can be sent to the local table. Query execution will occur only
on the specified shard, or it can be sent to the distributed table, and in that
case, each shard will execute the given queries. The server where the distributed
table was queried will aggregate the data and respond to the client:

The figure above visualizes what happens when a client queries a distributed table:

1. The select query is sent to a distributed table on a node arbitrarily
   (via a round-robin strategy or after being routed to a specific server
   by a load balancer). This node is now going to act as a coordinator.- The node will locate each shard that needs to execute the query
     via the information specified by the distributed table, and the query is
     sent to each shard.- Each shard reads, filters, and aggregates the data locally and then
       sends back a mergeable state to the coordinator.- The coordinating node merges the data and then sends the response back
         to the client.

When we add replicas into the mix, the process is fairly similar, with the only
difference being that only a single replica from each shard will execute the query.
This means that more queries can then be processed in parallel.

## Non-sharded architecture[​](#non-sharded-architecture "Direct link to Non-sharded architecture")

ClickHouse Cloud has a very different architecture to the one presented above.
(See ["ClickHouse Cloud Architecture"](https://clickhouse.com/docs/cloud/reference/architecture)
for more details). With separation of compute and storage, and with virtually an
infinite amount of storage, the need for shards becomes less important.

The figure below shows the ClickHouse Cloud architecture:

This architecture allows us to be able to add and remove replicas nearly
instantaneously, ensuring a very high cluster scalability. The ClickHouse
Keeper cluster (shown right) ensures that we have a single source of truth
for the metadata. Replicas can fetch the metadata from the ClickHouse Keeper
cluster and all maintain the same data. The data themselves are stored in
object storage, and the SSD cache allows us to speed up queries.

But how can we now distribute query execution across multiple servers? In a
sharded architecture, it was fairly obvious given that each shard could actually
execute a query on a subset of the data. How does it work when there is no sharding?

## Introducing parallel replicas[​](#introducing-parallel-replicas "Direct link to Introducing parallel replicas")

To parallelize query execution through multiple servers, we first need to be
able to assign one of our servers as a coordinator. The coordinator is the one
that creates the list of tasks that need to be executed, ensures they're all
executed, aggregated and that the result is returned to the client. Like
in most distributed systems, this will be the role of the node that receives the
initial query. We also need to define the unit of work. In a sharded architecture,
the unit of work is the shard, a subset of the data. With parallel replicas we
will use a small portion of the table, called [granules](/docs/guides/best-practices/sparse-primary-indexes#data-is-organized-into-granules-for-parallel-data-processing),
as the unit of work.

Now, let's see how it works in practice with the help of the figure below:

With parallel replicas:

1. The query from the client is sent to one node after going through a load
   balancer. This node becomes the coordinator for this query.- The node analyzes the index of each part, and selects the right parts and
     granules to process.- The coordinator splits the workload into a set of granules that can be
       assigned to different replicas.- Each set of granules gets processed by the corresponding replicas and a
         mergeable state is sent to the coordinator when they're finished.- Finally, the coordinator merges all the results from the replicas and
           then returns the response to the client.

The steps above outline how parallel replicas work in theory.
However, in practice, there are a lot of factors that can prevent such logic
from working perfectly:

1. Some replicas may be unavailable.- Replication in ClickHouse is asynchronous, some replicas might not
     have the same parts at some point in time.- Tail latency between replicas needs to be handled somehow.- The filesystem cache varies from replica to replica based on the
         activity on each replica, meaning that a random task assignment might
         lead to less optimal performance given the cache locality.

We explore how these factors are overcome in the following sections.

### Announcements[​](#announcements "Direct link to Announcements")

To address (1) and (2) from the list above, we introduced the concept of an
announcement. Let's visualize how it works using the figure below:

1. The query from the client is sent to one node after going through a load
   balancer. The node becomes the coordinator for this query.- The coordinating node sends a request to get announcements from
     all the replicas in the cluster. Replicas may have slightly different
     views of the current set of parts for a table. As a result we need to
     collect this information to avoid incorrect scheduling decisions.- The coordinating node then uses the announcements to define a set of
       granules that can be assigned to the different replicas. Here for example,
       we can see that no granules from part 3 have been assigned to replica 2
       because this replica didn't provide this part in its announcement.
       Also note that no tasks were assigned to replica 3 because the
       replica didn't provide an announcement.- After each replica has processed the query on their subset of granules
         and the mergeable state has been sent back to the coordinator, the
         coordinator merges the results and the response is sent to the client.

### Dynamic coordination[​](#dynamic-coordination "Direct link to Dynamic coordination")

To address the issue of tail latency, we added dynamic coordination. This means
that all the granules aren't sent to a replica in one request, but each replica
will be able to request a new task (a set of granules to be processed) to the
coordinator. The coordinator will give the replica the set of granules based on
the announcement received.

Let's assume that we're at the stage in the process where all replicas have sent
an announcement with all parts.

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.