---
title: "Introducing multi-stage distributed query execution in ClickHouse Cloud"
description: "ClickHouse Cloud introduced multi-stage distributed query execution to improve scaling for large joins and high-cardinality aggregations across multi-node clusters. Prior execution models relied on physical sharding or parallel replicas, which lacked the ability to freely repartition intermediate data between query stages. By splitting query plans into stages connected by exchange operators, worker nodes can shuffle, broadcast, scatter, or gather intermediate data during execution. In TPC-H benchmarks at scale factor 100 on eight nodes, total query time dropped from 117.6 seconds to 54.7 seconds, with scan-dominated queries showing near-linear 7.4x speedups. While currently in private preview with a rule-based planner, future work includes developing a cost-based optimizer to select optimal aggregation and join strategies automatically."
---

# Introducing multi-stage distributed query execution in ClickHouse Cloud

[Clickhouse](https://yomu.fyi/company/clickhouse) · Alexander Gololobov · May 27, 2026

**Type:** Announcement

## Summary

ClickHouse Cloud introduced multi-stage distributed query execution to improve scaling for large joins and high-cardinality aggregations across multi-node clusters. Prior execution models relied on physical sharding or parallel replicas, which lacked the ability to freely repartition intermediate data between query stages. By splitting query plans into stages connected by exchange operators, worker nodes can shuffle, broadcast, scatter, or gather intermediate data during execution. In TPC-H benchmarks at scale factor 100 on eight nodes, total query time dropped from 117.6 seconds to 54.7 seconds, with scan-dominated queries showing near-linear 7.4x speedups. While currently in private preview with a rule-based planner, future work includes developing a cost-based optimizer to select optimal aggregation and join strategies automatically.

## Context

In shared-nothing ClickHouse deployments and ClickHouse Cloud parallel replicas, queries could fan out work across nodes but could not freely repartition intermediate results between execution stages. As a result, distributed joins required building hash tables from the full right side on every node or copying data across shards, while high-cardinality aggregations required a single coordinator node to merge all partial results.

## Approach / What changed

ClickHouse Cloud added multi-stage distributed query execution, splitting query plans into stages running in parallel across worker nodes. Intermediate data is dynamically redistributed between stages using exchange operators: ShuffleExchange to repartition rows by join or GROUP BY keys, BroadcastExchange to replicate small inputs across workers, ScatterExchange to distribute rows randomly, and GatherExchange to send outputs to the coordinator.

## Takeaways

- In TPC-H scale factor 100 testing on 8 Graviton nodes, multi-stage distributed execution achieved a 2.1x overall speedup, including a 7.4x near-linear speedup on scan-heavy Q01.
- Exchange operators like ShuffleExchange and BroadcastExchange eliminate the need for every worker node to build full right-side hash tables during large joins.
- The initial implementation uses a rule-based planner and does not yet fully support runtime filters across distributed joins, which can cause regressions on queries like Q02.

**Tags:** [Architecture](https://yomu.fyi/topic/architecture), [Performance](https://yomu.fyi/topic/performance), [Scalability](https://yomu.fyi/topic/scalability)

- Source: [Clickhouse](https://clickhouse.com/blog/multi-stage-distributed-query-execution-clickhouse-cloud)
- Source URL: https://clickhouse.com/blog/multi-stage-distributed-query-execution-clickhouse-cloud
- Ingested by Yomu: 2026-08-28T04:00:56.015Z

[Read original post](https://clickhouse.com/blog/multi-stage-distributed-query-execution-clickhouse-cloud)
