When the clickhouse distributed table query, why only read very little data from the node，How did this do？

[ChaoHsupin]

Company or project name

No response

Question

After testing, I found that when the distributed table query data of Clickhouse is not simply, it is not simply to gather all data from other nodes to nodes that initiate requests.

In fact, there are very few data read from other nodes，But I did not find related articles to explain its principle or algorithm.

Do you have related articles or blogs?

[den-crane]

Explanation from chatgpt, it's close to the reality):

---

In ClickHouse, distributed tables are used to handle data that is spread across multiple servers (shards). The system allows for efficient querying and processing by distributing the workload across these shards and then combining the results. Here's a breakdown of how this works, especially concerning aggregate functions:

1. Distributed Table Structure

• A distributed table is a virtual table that does not store data itself. Instead, it references physical tables (often replicated tables) on multiple shards.
• When a query is executed against a distributed table, ClickHouse automatically distributes the query to the underlying shards, which contain the actual data.

2. Aggregate Function States

• Aggregate functions in ClickHouse can operate in multiple stages: state calculation and finalization.
• State Calculation: When an aggregate function is applied, it calculates an intermediate state (e.g., sum, average, count) rather than the final result. This intermediate state can be merged with other states later.
• Finalization: The final result is obtained by finalizing the merged states (e.g., computing the average from the sum and count).

3. Pre-calculation on Shards

• When a query involves an aggregation, each shard independently calculates the aggregate function states for its portion of the data.
• This distributed pre-calculation reduces the amount of data that needs to be transferred across the network, as only the intermediate states (instead of raw data) need to be sent to the initiator node.

4. Finalization at the Initiator

• The initiator node is the server that initially receives the query.
• After receiving the pre-calculated aggregate states from all shards, the initiator node merges these states.
• The initiator then finalizes the aggregate function by computing the final result from the merged states (e.g., summing counts and computing the final average).

5. Example Workflow

1. A query requesting an aggregate, like SELECT AVG(value) FROM distributed_table, is issued.
2. The query is distributed to all shards.
3. Each shard calculates the sum and count for value and returns these states to the initiator.
4. The initiator merges the sums and counts from all shards.
5. The initiator finalizes the AVG by dividing the merged sum by the merged count.

6. Benefits

• Efficiency: By performing most of the heavy lifting on the shards and only sending essential data back to the initiator, ClickHouse minimizes network overhead and speeds up query processing.
• Scalability: This approach allows ClickHouse to scale horizontally, effectively handling large datasets spread across multiple nodes.

This distributed approach is particularly powerful for large-scale data processing, making ClickHouse a strong choice for analytics on massive datasets.

[ChaoHsupin]

Explanation from chatgpt, it's close to the reality):

In ClickHouse, distributed tables are used to handle data that is spread across multiple servers (shards). The system allows for efficient querying and processing by distributing the workload across these shards and then combining the results. Here's a breakdown of how this works, especially concerning aggregate functions:

1. Distributed Table Structure

• A distributed table is a virtual table that does not store data itself. Instead, it references physical tables (often replicated tables) on multiple shards.
• When a query is executed against a distributed table, ClickHouse automatically distributes the query to the underlying shards, which contain the actual data.

2. Aggregate Function States

• Aggregate functions in ClickHouse can operate in multiple stages: state calculation and finalization.
• State Calculation: When an aggregate function is applied, it calculates an intermediate state (e.g., sum, average, count) rather than the final result. This intermediate state can be merged with other states later.
• Finalization: The final result is obtained by finalizing the merged states (e.g., computing the average from the sum and count).

3. Pre-calculation on Shards

• When a query involves an aggregation, each shard independently calculates the aggregate function states for its portion of the data.
• This distributed pre-calculation reduces the amount of data that needs to be transferred across the network, as only the intermediate states (instead of raw data) need to be sent to the initiator node.

4. Finalization at the Initiator

• The initiator node is the server that initially receives the query.
• After receiving the pre-calculated aggregate states from all shards, the initiator node merges these states.
• The initiator then finalizes the aggregate function by computing the final result from the merged states (e.g., summing counts and computing the final average).

5. Example Workflow

1. A query requesting an aggregate, like SELECT AVG(value) FROM distributed_table, is issued.
2. The query is distributed to all shards.
3. Each shard calculates the sum and count for value and returns these states to the initiator.
4. The initiator merges the sums and counts from all shards.
5. The initiator finalizes the AVG by dividing the merged sum by the merged count.

6. Benefits

• Efficiency: By performing most of the heavy lifting on the shards and only sending essential data back to the initiator, ClickHouse minimizes network overhead and speeds up query processing.
• Scalability: This approach allows ClickHouse to scale horizontally, effectively handling large datasets spread across multiple nodes.

This distributed approach is particularly powerful for large-scale data processing, making ClickHouse a strong choice for analytics on massive datasets.

Thanks for you explain and examples, helped me very much！

I have a question, I hope to get your help. For example, the SQL：

SELECT * FROM monitor.qunhe_log_all ORDER BY timestamp DESC LIMIT 50

The distributed query below is to query 50 in each piece, and then merge on query node, and then sort to take the top 50? Or, when the nodes scan data, they will communicate with each other's current data status, and then only send a small amount of effective data until the query node.

[den-crane]

The distributed query below is to query 50 in each piece, and then merge on query node, and then sort to take the top 50?

Like this. Each node reads data, finds 50 rows, sends them back to the query-initator, query-initator sort them again and leave only 50 rows.

Or, when the nodes scan data, they will communicate with each other's current data status, and then only send a small amount of effective data until the query node.

No, shards don't communicate. They are not aware about each other. They behave like a single (standalone) Clickhouse server, they execute usual query, there is no difference if they receive a query from Distributed table or from go/java/clickhouse-client client. Distributed table (query-initator) communicates with shards as a simple client, using the same TCP protocol and Native format the same way as go-library (go-lang client).

[ChaoHsupin]

@den-crane Thanks for your help!