> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-mintlify-fbfa8bee.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# Répliques parallèles
> Dans ce guide, nous verrons d’abord comment ClickHouse répartit une requête entre plusieurs shards au moyen de tables distribuées, puis comment une requête peut exploiter plusieurs répliques lors de son exécution.
export const Image = ({img, alt, size}) => {
return
;
};
## Introduction
ClickHouse traite les requêtes extrêmement rapidement, mais comment ces requêtes
sont-elles réparties et parallélisées sur plusieurs serveurs ?
> Dans ce guide, nous verrons d'abord comment ClickHouse répartit une requête sur
> plusieurs shards via des tables distribuées, puis comment une requête peut tirer parti
> de plusieurs répliques pour son exécution.
## Architecture shardée
Dans une architecture shared-nothing, les clusters sont généralement répartis en
plusieurs shards, chaque shard contenant un sous-ensemble de l'ensemble des données. Une
table distribuée se place au-dessus de ces shards et fournit une vue unifiée de
l'intégralité des données.
Les lectures peuvent être envoyées à la table locale. L'exécution de la requête n'aura lieu
que sur le shard spécifié, ou bien elle peut être envoyée à la table distribuée, auquel
cas chaque shard exécutera la requête donnée. Le serveur sur lequel la table distribuée
a été interrogée agrégera les données et renverra la réponse au client :
La figure ci-dessus illustre ce qui se passe lorsqu'un client interroge une table distribuée :
-
La requête SELECT est envoyée à une table distribuée sur un nœud choisi arbitrairement
(via une stratégie round-robin ou après avoir été redirigée vers un serveur spécifique
par un répartiteur de charge). Ce nœud va alors jouer le rôle de coordinateur.
-
Le nœud localise chaque shard qui doit exécuter la requête
à l'aide des informations spécifiées par la table distribuée, puis la requête est
envoyée à chaque shard.
-
Chaque shard lit, filtre et agrège les données localement, puis
renvoie un état fusionnable au coordinateur.
-
Le nœud coordinateur fusionne les données, puis renvoie la réponse
au client.
Lorsque l'on ajoute des réplicas à l'ensemble, le processus reste globalement le même, la seule
différence étant qu'une seule réplique de chaque shard exécutera la requête.
Cela signifie qu'un plus grand nombre de requêtes peut alors être traité en parallèle.
## Architecture non shardée
ClickHouse Cloud présente une architecture très différente de celle décrite ci-dessus.
(Voir ["Architecture de ClickHouse Cloud"](/fr/products/cloud/reference/architecture)
pour plus de détails). Avec la séparation du calcul et du stockage, ainsi qu'une capacité de stockage virtuellement
infinie, le besoin de shards devient moins important.
La figure ci-dessous montre l'architecture de ClickHouse Cloud :
Cette architecture permet d'ajouter et de supprimer des répliques presque
instantanément, ce qui garantit une très forte scalabilité du cluster. Le cluster
ClickHouse Keeper (illustré à droite) garantit l'existence d'une source de vérité unique
pour les métadonnées. Les répliques peuvent récupérer les métadonnées depuis le cluster ClickHouse Keeper
et conservent toutes les mêmes données. Les données elles-mêmes sont stockées dans
le stockage objet, et le cache SSD permet d'accélérer les requêtes.
Mais comment distribuer désormais l'exécution des requêtes sur plusieurs serveurs ? Dans une
architecture shardée, c'était assez évident puisque chaque shard pouvait effectivement
exécuter une requête sur un sous-ensemble des données. Comment cela fonctionne-t-il lorsqu'il n'y a pas de sharding ?
## Présentation des répliques parallèles
Pour paralléliser l'exécution des requêtes sur plusieurs serveurs, nous devons d'abord être
en mesure de désigner l'un de nos serveurs comme coordinateur. Le coordinateur est celui
qui crée la liste des tâches à exécuter, veille à ce qu'elles soient toutes
exécutées et agrégées, puis à ce que le résultat soit renvoyé au client. Comme
dans la plupart des systèmes distribués, ce sera le rôle du nœud qui reçoit la
requête initiale. Nous devons également définir l'unité de travail. Dans une architecture shardée,
l'unité de travail est le shard, c'est-à-dire un sous-ensemble des données. Avec les répliques parallèles, nous
utiliserons une petite portion de la table, appelée [granules](/fr/guides/clickhouse/data-modelling/sparse-primary-indexes#data-is-organized-into-granules-for-parallel-data-processing),
comme unité de travail.
Voyons maintenant comment cela fonctionne en pratique à l'aide de la figure ci-dessous :
Avec les répliques parallèles :
-
La requête du client est envoyée à un nœud après être passée par un
répartiteur de charge. Ce nœud devient le coordinateur de cette requête.
-
Le nœud analyse l'index de chaque partie et sélectionne les bonnes parties et
granules à traiter.
-
Le coordinateur répartit la charge de travail en un ensemble de granules pouvant être
attribués à différentes répliques.
-
Chaque ensemble de granules est traité par les répliques correspondantes, puis un
état fusionnable est envoyé au coordinateur une fois le traitement terminé.
-
Enfin, le coordinateur fusionne tous les résultats des répliques puis
renvoie la réponse au client.
Les étapes ci-dessus décrivent le fonctionnement théorique des répliques parallèles.
Cependant, en pratique, de nombreux facteurs peuvent empêcher cette logique
de fonctionner parfaitement :
-
Certaines répliques peuvent être indisponibles.
-
La réplication dans ClickHouse est asynchrone ; à un instant donné, certaines répliques peuvent ne pas
avoir les mêmes parties.
-
Il faut également gérer la tail latency.
-
Le cache du système de fichiers varie d'une réplique à l'autre en fonction de
l'activité sur chacune d'elles, ce qui signifie qu'une attribution aléatoire des tâches pourrait
entraîner des performances moins bonnes en raison de la localité du cache.
Nous verrons dans les sections suivantes comment ces facteurs sont pris en compte.
### Annonces
Pour répondre aux points (1) et (2) de la liste ci-dessus, nous avons introduit le concept d'annonce. Visualisons son fonctionnement à l'aide de la figure ci-dessous :
-
La requête du client est envoyée à un nœud après être passée par un répartiteur de charge. Ce nœud devient le coordinateur de cette requête.
-
Le nœud coordinateur envoie ensuite une requête pour récupérer les annonces de
toutes les répliques du cluster. Les répliques peuvent avoir des vues légèrement différentes
de l'ensemble actuel des parts d'une table. Il faut donc
collecter ces informations afin d'éviter des décisions de planification incorrectes.
-
Le nœud coordinateur utilise ensuite les annonces pour définir un ensemble de
granules pouvant être attribuées aux différentes répliques. Ici, par exemple,
on peut voir qu'aucune granule de la part 3 n'a été attribuée à la réplique 2,
car cette réplique n'a pas indiqué cette part dans son annonce.
Notez également qu'aucune tâche n'a été attribuée à la réplique 3, car la
réplique n'a fourni aucune annonce.
-
Une fois que chaque réplique a traité la requête sur son sous-ensemble de granules
et que l'état fusionnable a été renvoyé au coordinateur, celui-ci
fusionne les résultats et renvoie la réponse au client.
### Coordination dynamique
Pour remédier au problème de latence des requêtes les plus lentes, nous avons ajouté la coordination dynamique. Cela signifie
que toutes les granules ne sont pas envoyées à une réplique en une seule requête, mais que chaque réplique
peut demander une nouvelle tâche (un ensemble de granules à traiter) au
coordinateur. Le coordinateur fournit alors à la réplique l'ensemble de granules en fonction de
l'annonce reçue.
Supposons que nous soyons à l'étape du processus où toutes les répliques ont envoyé
une annonce avec toutes les parts.
La figure ci-dessous illustre le fonctionnement de la coordination dynamique :
-
Les répliques indiquent au nœud coordinateur qu'elles sont en mesure de traiter
des tâches, et peuvent également préciser la quantité de travail qu'elles peuvent prendre en charge.
-
Le coordinateur attribue des tâches aux répliques.
-
Les répliques 1 et 2 terminent leur tâche très rapidement. Elles
demandent alors une autre tâche au nœud coordinateur.
-
Le coordinateur attribue de nouvelles tâches aux répliques 1 et 2.
-
Toutes les répliques ont maintenant terminé le traitement de leur tâche. Elles
demandent davantage de tâches.
-
Le coordinateur, à l'aide des annonces, vérifie quelles tâches restent à
traiter, mais il n'en reste aucune.
-
Le coordinateur indique aux répliques que tout a été traité.
Il va maintenant fusionner tous les états fusionnables et répondre à la requête.
### Gestion de la localité du cache
Le dernier problème potentiel concerne la manière de préserver la localité du cache. Si la requête
est exécutée plusieurs fois, comment garantir que la même tâche soit envoyée à la
même réplique ? Dans l’exemple précédent, les tâches suivantes étaient attribuées :
|
Réplique 1 |
Réplique 2 |
Réplique 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 |
Pour garantir que les mêmes tâches soient attribuées aux mêmes répliques et puissent
bénéficier du cache, deux mécanismes entrent en jeu. On calcule un hachage de la part et de l’ensemble de
granules (autrement dit, d’une tâche). On applique ensuite un modulo sur le nombre de répliques pour
attribuer la tâche.
Sur le papier, cela semble pertinent, mais en pratique, une charge soudaine sur une réplique ou une
dégradation du réseau peut introduire de la tail latency si la même réplique est
systématiquement utilisée pour exécuter certaines tâches. Si `max_parallel_replicas` est inférieur
au nombre de répliques, des répliques aléatoires sont alors sélectionnées pour l’exécution des requêtes.
### Vol de tâches
si une réplique traite les tâches plus lentement que les autres, celles-ci essaieront de
« voler » les tâches qui, en principe, lui sont attribuées par hachage afin de réduire la
tail latency.
### Limitations
Cette fonctionnalité présente des limitations connues, dont les principales sont documentées dans
cette section.
Si vous trouvez un ticket qui ne figure pas parmi les limitations indiquées ci-dessous et
soupçonnez les répliques parallèles d’en être la cause, veuillez ouvrir un ticket sur GitHub avec
le label `comp-parallel-replicas`.
| Limitation | Description |
| ------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Requêtes complexes | Actuellement, les répliques parallèles fonctionnent assez bien pour les requêtes simples. Des couches de complexité comme les CTE, les sous-requêtes, les JOIN, les requêtes à structure non plate, etc. peuvent avoir un impact négatif sur les performances des requêtes. |
| Petites requêtes | Si vous exécutez une requête qui ne traite pas beaucoup de lignes, l’exécuter sur plusieurs répliques peut ne pas améliorer les performances, étant donné que le temps réseau nécessaire à la coordination entre les répliques peut entraîner des cycles supplémentaires dans l’exécution de la requête. Vous pouvez limiter ces problèmes en utilisant le paramètre : [`parallel_replicas_min_number_of_rows_per_replica`](/fr/reference/settings/session-settings#parallel_replicas_min_number_of_rows_per_replica). |
| Les répliques parallèles sont désactivés avec FINAL | |
| Les projections ne sont pas utilisées avec les répliques parallèles | |
| Données à forte cardinalité et agrégation complexe | Une agrégation à forte cardinalité qui doit envoyer beaucoup de données peut ralentir considérablement vos requêtes. |
| Compatibilité avec le nouvel analyseur | Le nouvel analyseur peut considérablement ralentir ou accélérer l’exécution des requêtes dans certains scénarios. |
## Paramètres liés aux répliques parallèles
| Paramètre | Description |
| -------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `enable_parallel_replicas` | `0` : désactivé
`1` : activé
`2` : force l'utilisation d'une réplique parallèle et lèvera une exception si ce n'est pas le cas. |
| `cluster_for_parallel_replicas` | Nom du cluster à utiliser pour la réplication parallèle ; si vous utilisez ClickHouse Cloud, utilisez `default`. |
| `max_parallel_replicas` | Nombre maximal de répliques à utiliser pour exécuter la requête sur plusieurs répliques. Si un nombre inférieur au nombre de répliques du cluster est spécifié, les nœuds seront sélectionnés aléatoirement. Cette valeur peut également être surallouée pour tenir compte de la mise à l'échelle horizontale. |
| `parallel_replicas_min_number_of_rows_per_replica` | Permet de limiter le nombre de répliques utilisées en fonction du nombre de lignes à traiter. Le nombre de répliques utilisées est défini par :
`estimated rows to read` / `min_number_of_rows_per_replica`. |
| `enable_analyzer` | L'exécution des requêtes avec des répliques parallèles n'est prise en charge que lorsque l'analyseur est activé |
## Investiguer les problèmes liés aux répliques parallèles
Vous pouvez vérifier quels paramètres sont utilisés pour chaque requête dans la
table [`system.query_log`](/fr/reference/system-tables/query_log). Vous pouvez
également consulter la table [`system.events`](/fr/reference/system-tables/events)
pour voir tous les événements survenus sur le serveur, et vous pouvez utiliser la
table function [`clusterAllReplicas`](/fr/reference/functions/table-functions/cluster) pour voir les tables sur tous les répliques
(si vous utilisez ClickHouse Cloud, utilisez `default`).
```sql title="Query" theme={null}
SELECT
hostname(),
*
FROM clusterAllReplicas('default', system.events)
WHERE event ILIKE '%ParallelReplicas%'
```
```response title="Response" theme={null}
┌─hostname()───────────────────────┬─event──────────────────────────────────────────┬─value─┬─description──────────────────────────────────────────────────────────────────────────────────────────┐
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasHandleRequestMicroseconds │ 438 │ Time spent processing requests for marks from replicas │
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasHandleAnnouncementMicroseconds │ 558 │ Time spent processing replicas announcements │
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasReadUnassignedMarks │ 240 │ Sum across all replicas of how many unassigned marks were scheduled │
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasReadAssignedForStealingMarks │ 4 │ Sum across all replicas of how many of scheduled marks were assigned for stealing by consistent hash │
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasStealingByHashMicroseconds │ 5 │ Time spent collecting segments meant for stealing by hash │
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasProcessingPartsMicroseconds │ 5 │ Time spent processing data parts │
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasStealingLeftoversMicroseconds │ 3 │ Time spent collecting orphaned segments │
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasUsedCount │ 2 │ Number of replicas used to execute a query with task-based parallel replicas │
│ c-crimson-vd-86-server-rdhnsx3-0 │ ParallelReplicasAvailableCount │ 6 │ Number of replicas available to execute a query with task-based parallel replicas │
└──────────────────────────────────┴────────────────────────────────────────────────┴───────┴──────────────────────────────────────────────────────────────────────────────────────────────────────┘
┌─hostname()───────────────────────┬─event──────────────────────────────────────────┬─value─┬─description──────────────────────────────────────────────────────────────────────────────────────────┐
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasHandleRequestMicroseconds │ 698 │ Time spent processing requests for marks from replicas │
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasHandleAnnouncementMicroseconds │ 644 │ Time spent processing replicas announcements │
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasReadUnassignedMarks │ 190 │ Sum across all replicas of how many unassigned marks were scheduled │
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasReadAssignedForStealingMarks │ 54 │ Sum across all replicas of how many of scheduled marks were assigned for stealing by consistent hash │
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasStealingByHashMicroseconds │ 8 │ Time spent collecting segments meant for stealing by hash │
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasProcessingPartsMicroseconds │ 4 │ Time spent processing data parts │
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasStealingLeftoversMicroseconds │ 2 │ Time spent collecting orphaned segments │
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasUsedCount │ 2 │ Number of replicas used to execute a query with task-based parallel replicas │
│ c-crimson-vd-86-server-e9kp5f0-0 │ ParallelReplicasAvailableCount │ 6 │ Number of replicas available to execute a query with task-based parallel replicas │
└──────────────────────────────────┴────────────────────────────────────────────────┴───────┴──────────────────────────────────────────────────────────────────────────────────────────────────────┘
┌─hostname()───────────────────────┬─event──────────────────────────────────────────┬─value─┬─description──────────────────────────────────────────────────────────────────────────────────────────┐
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasHandleRequestMicroseconds │ 620 │ Time spent processing requests for marks from replicas │
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasHandleAnnouncementMicroseconds │ 656 │ Time spent processing replicas announcements │
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasReadUnassignedMarks │ 1 │ Sum across all replicas of how many unassigned marks were scheduled │
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasReadAssignedForStealingMarks │ 1 │ Sum across all replicas of how many of scheduled marks were assigned for stealing by consistent hash │
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasStealingByHashMicroseconds │ 4 │ Time spent collecting segments meant for stealing by hash │
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasProcessingPartsMicroseconds │ 3 │ Time spent processing data parts │
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasStealingLeftoversMicroseconds │ 1 │ Time spent collecting orphaned segments │
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasUsedCount │ 2 │ Number of replicas used to execute a query with task-based parallel replicas │
│ c-crimson-vd-86-server-ybtm18n-0 │ ParallelReplicasAvailableCount │ 12 │ Number of replicas available to execute a query with task-based parallel replicas │
└──────────────────────────────────┴────────────────────────────────────────────────┴───────┴──────────────────────────────────────────────────────────────────────────────────────────────────────┘
┌─hostname()───────────────────────┬─event──────────────────────────────────────────┬─value─┬─description──────────────────────────────────────────────────────────────────────────────────────────┐
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasHandleRequestMicroseconds │ 696 │ Time spent processing requests for marks from replicas │
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasHandleAnnouncementMicroseconds │ 717 │ Time spent processing replicas announcements │
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasReadUnassignedMarks │ 2 │ Sum across all replicas of how many unassigned marks were scheduled │
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasReadAssignedForStealingMarks │ 2 │ Sum across all replicas of how many of scheduled marks were assigned for stealing by consistent hash │
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasStealingByHashMicroseconds │ 10 │ Time spent collecting segments meant for stealing by hash │
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasProcessingPartsMicroseconds │ 6 │ Time spent processing data parts │
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasStealingLeftoversMicroseconds │ 2 │ Time spent collecting orphaned segments │
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasUsedCount │ 2 │ Number of replicas used to execute a query with task-based parallel replicas │
│ c-crimson-vd-86-server-16j1ncj-0 │ ParallelReplicasAvailableCount │ 12 │ Number of replicas available to execute a query with task-based parallel replicas │
└──────────────────────────────────┴────────────────────────────────────────────────┴───────┴──────────────────────────────────────────────────────────────────────────────────────────────────────┘
```
La table [`system.text_log`](/fr/reference/system-tables/text_log) contient également des informations sur l’exécution de requêtes utilisant des répliques parallèles :
```sql title="Query" theme={null}
SELECT message
FROM clusterAllReplicas('default', system.text_log)
WHERE query_id = 'ad40c712-d25d-45c4-b1a1-a28ba8d4019c'
ORDER BY event_time_microseconds ASC
```
```response title="Response" theme={null}
┌─message────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ (from 54.218.178.249:59198) SELECT * FROM session_events WHERE type='type2' LIMIT 10 SETTINGS allow_experimental_parallel_reading_from_replicas=2; (stage: Complete) │
│ Query SELECT __table1.clientId AS clientId, __table1.sessionId AS sessionId, __table1.pageId AS pageId, __table1.timestamp AS timestamp, __table1.type AS type FROM default.session_events AS __table1 WHERE __table1.type = 'type2' LIMIT _CAST(10, 'UInt64') SETTINGS allow_experimental_parallel_reading_from_replicas = 2 to stage Complete │
│ Access granted: SELECT(clientId, sessionId, pageId, timestamp, type) ON default.session_events │
│ Query SELECT __table1.clientId AS clientId, __table1.sessionId AS sessionId, __table1.pageId AS pageId, __table1.timestamp AS timestamp, __table1.type AS type FROM default.session_events AS __table1 WHERE __table1.type = 'type2' LIMIT _CAST(10, 'UInt64') to stage WithMergeableState only analyze │
│ Access granted: SELECT(clientId, sessionId, pageId, timestamp, type) ON default.session_events │
│ Query SELECT __table1.clientId AS clientId, __table1.sessionId AS sessionId, __table1.pageId AS pageId, __table1.timestamp AS timestamp, __table1.type AS type FROM default.session_events AS __table1 WHERE __table1.type = 'type2' LIMIT _CAST(10, 'UInt64') from stage FetchColumns to stage WithMergeableState only analyze │
│ Query SELECT __table1.clientId AS clientId, __table1.sessionId AS sessionId, __table1.pageId AS pageId, __table1.timestamp AS timestamp, __table1.type AS type FROM default.session_events AS __table1 WHERE __table1.type = 'type2' LIMIT _CAST(10, 'UInt64') SETTINGS allow_experimental_parallel_reading_from_replicas = 2 to stage WithMergeableState only analyze │
│ Access granted: SELECT(clientId, sessionId, pageId, timestamp, type) ON default.session_events │
│ Query SELECT __table1.clientId AS clientId, __table1.sessionId AS sessionId, __table1.pageId AS pageId, __table1.timestamp AS timestamp, __table1.type AS type FROM default.session_events AS __table1 WHERE __table1.type = 'type2' LIMIT _CAST(10, 'UInt64') SETTINGS allow_experimental_parallel_reading_from_replicas = 2 from stage FetchColumns to stage WithMergeableState only analyze │
│ Query SELECT __table1.clientId AS clientId, __table1.sessionId AS sessionId, __table1.pageId AS pageId, __table1.timestamp AS timestamp, __table1.type AS type FROM default.session_events AS __table1 WHERE __table1.type = 'type2' LIMIT _CAST(10, 'UInt64') SETTINGS allow_experimental_parallel_reading_from_replicas = 2 from stage WithMergeableState to stage Complete │
│ The number of replicas requested (100) is bigger than the real number available in the cluster (6). Will use the latter number to execute the query. │
│ Initial request from replica 4: 2 parts: [part all_0_2_1 with ranges [(0, 182)], part all_3_3_0 with ranges [(0, 62)]]----------
Received from 4 replica
│
│ Reading state is fully initialized: part all_0_2_1 with ranges [(0, 182)] in replicas [4]; part all_3_3_0 with ranges [(0, 62)] in replicas [4] │
│ Sent initial requests: 1 Replicas count: 6 │
│ Initial request from replica 2: 2 parts: [part all_0_2_1 with ranges [(0, 182)], part all_3_3_0 with ranges [(0, 62)]]----------
Received from 2 replica
│
│ Sent initial requests: 2 Replicas count: 6 │
│ Handling request from replica 4, minimal marks size is 240 │
│ Going to respond to replica 4 with 1 parts: [part all_0_2_1 with ranges [(128, 182)]]. Finish: false; mine_marks=0, stolen_by_hash=54, stolen_rest=0 │
│ Initial request from replica 1: 2 parts: [part all_0_2_1 with ranges [(0, 182)], part all_3_3_0 with ranges [(0, 62)]]----------
Received from 1 replica
│
│ Sent initial requests: 3 Replicas count: 6 │
│ Handling request from replica 4, minimal marks size is 240 │
│ Going to respond to replica 4 with 2 parts: [part all_0_2_1 with ranges [(0, 128)], part all_3_3_0 with ranges [(0, 62)]]. Finish: false; mine_marks=0, stolen_by_hash=0, stolen_rest=190 │
│ Initial request from replica 0: 2 parts: [part all_0_2_1 with ranges [(0, 182)], part all_3_3_0 with ranges [(0, 62)]]----------
Received from 0 replica
│
│ Sent initial requests: 4 Replicas count: 6 │
│ Initial request from replica 5: 2 parts: [part all_0_2_1 with ranges [(0, 182)], part all_3_3_0 with ranges [(0, 62)]]----------
Received from 5 replica
│
│ Sent initial requests: 5 Replicas count: 6 │
│ Handling request from replica 2, minimal marks size is 240 │
│ Going to respond to replica 2 with 0 parts: []. Finish: true; mine_marks=0, stolen_by_hash=0, stolen_rest=0 │
│ Initial request from replica 3: 2 parts: [part all_0_2_1 with ranges [(0, 182)], part all_3_3_0 with ranges [(0, 62)]]----------
Received from 3 replica
│
│ Sent initial requests: 6 Replicas count: 6 │
│ Total rows to read: 2000000 │
│ Handling request from replica 5, minimal marks size is 240 │
│ Going to respond to replica 5 with 0 parts: []. Finish: true; mine_marks=0, stolen_by_hash=0, stolen_rest=0 │
│ Handling request from replica 0, minimal marks size is 240 │
│ Going to respond to replica 0 with 0 parts: []. Finish: true; mine_marks=0, stolen_by_hash=0, stolen_rest=0 │
│ Handling request from replica 1, minimal marks size is 240 │
│ Going to respond to replica 1 with 0 parts: []. Finish: true; mine_marks=0, stolen_by_hash=0, stolen_rest=0 │
│ Handling request from replica 3, minimal marks size is 240 │
│ Going to respond to replica 3 with 0 parts: []. Finish: true; mine_marks=0, stolen_by_hash=0, stolen_rest=0 │
│ (c-crimson-vd-86-server-rdhnsx3-0.c-crimson-vd-86-server-headless.ns-crimson-vd-86.svc.cluster.local:9000) Cancelling query because enough data has been read │
│ Read 81920 rows, 5.16 MiB in 0.013166 sec., 6222087.194288318 rows/sec., 391.63 MiB/sec. │
│ Coordination done: Statistics: replica 0 - {requests: 2 marks: 0 assigned_to_me: 0 stolen_by_hash: 0 stolen_unassigned: 0}; replica 1 - {requests: 2 marks: 0 assigned_to_me: 0 stolen_by_hash: 0 stolen_unassigned: 0}; replica 2 - {requests: 2 marks: 0 assigned_to_me: 0 stolen_by_hash: 0 stolen_unassigned: 0}; replica 3 - {requests: 2 marks: 0 assigned_to_me: 0 stolen_by_hash: 0 stolen_unassigned: 0}; replica 4 - {requests: 3 marks: 244 assigned_to_me: 0 stolen_by_hash: 54 stolen_unassigned: 190}; replica 5 - {requests: 2 marks: 0 assigned_to_me: 0 stolen_by_hash: 0 stolen_unassigned: 0} │
│ Peak memory usage (for query): 1.81 MiB. │
│ Processed in 0.024095586 sec. │
└────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
```
Enfin, vous pouvez également utiliser `EXPLAIN PIPELINE`. Il met en évidence la manière dont ClickHouse
va exécuter une requête et quelles ressources seront utilisées pour
son exécution. Prenons la requête suivante comme exemple :
```sql theme={null}
SELECT count(), uniq(pageId) , min(timestamp), max(timestamp)
FROM session_events
WHERE type='type3'
GROUP BY toYear(timestamp) LIMIT 10
```
Voyons le pipeline de requête sans réplica parallèle :
```sql title="EXPLAIN PIPELINE (without parallel replica)" theme={null}
EXPLAIN PIPELINE graph = 1, compact = 0
SELECT count(), uniq(pageId) , min(timestamp), max(timestamp)
FROM session_events
WHERE type='type3'
GROUP BY toYear(timestamp)
LIMIT 10
SETTINGS allow_experimental_parallel_reading_from_replicas=0
FORMAT TSV;
```
Et maintenant avec la réplique parallèle :
```sql title="EXPLAIN PIPELINE (with parallel replica)" theme={null}
EXPLAIN PIPELINE graph = 1, compact = 0
SELECT count(), uniq(pageId) , min(timestamp), max(timestamp)
FROM session_events
WHERE type='type3'
GROUP BY toYear(timestamp)
LIMIT 10
SETTINGS allow_experimental_parallel_reading_from_replicas=2
FORMAT TSV;
```