ClickHouse / ClickHouse/ClickHouse
parallel_distributed_insert_select=2 falls back to single-node writes when SELECT has a GROUP BY, even for Summing/AggregatingMergeTree
- Dominant language
- C++
- Stars
- 49.9k
- Forks
- 9k
- Avg merge
- 21h 32m
- Merged PRs (30d)
- 515
Description
### Company or project name
_No response_
### Use case
Inserts like `INSERT INTO ... SELECT ... GROUP BY` with `parallel_distributed_insert_select=2` and `enable_parallel_replicas=1` fall back to single-node writes.
However, when the destination table is a `SummingMergeTree` or an `AggregatingMergeTree`, partial aggregates from each replica are semantically correct: the background merges will eventually combine them into the correct final state.
## Reproduction
```sql
CREATE TABLE t1 ON CLUSTER `{cluster}` (ts DateTime64(3), value UInt64)
ENGINE = ReplicatedMergeTree ORDER BY ts;
INSERT INTO t1 SELECT * FROM generateRandom('ts DateTime64(3), value UInt64') LIMIT 1000000000;
CREATE TABLE t2_smt ON CLUSTER `{cluster}` (date Date, sum_value UInt64)
ENGINE = ReplicatedSummingMergeTree() ORDER BY date;
```
### Without `GROUP BY` — both nodes write
```sql
INSERT INTO t2_smt
SETTINGS parallel_distributed_insert_select = 2, enable_parallel_replicas = 1, cluster_for_parallel_replicas = 'cluster_1S_2R'
SELECT * FROM t1;
SELECT hostname(), read_rows, written_rows
FROM clusterAllReplicas(`{cluster}`, system.query_log)
-- replace with the id returned by the INSERT above
WHERE initial_query_id =
AND type = 2;
```
```
┌─hostname()────┬──read_rows─┬─written_rows─┐
│ clickhouse-01 │ 1000000000 │ 663342728 │
│ clickhouse-02 │ 336657272 │ 336657272 │
└───────────────┴────────────┴──────────────┘
```
### With `GROUP BY` — only the coordinator writes
```sql
INSERT INTO t2_smt
SETTINGS parallel_distributed_insert_select = 2, enable_parallel_replicas = 1, cluster_for_parallel_replicas = 'cluster_1S_2R'
SELECT toDate(ts) AS date, sum(value) AS sum_value FROM t1 GROUP BY ALL;
SELECT hostname(), read_rows, written_rows
FROM clusterAllReplicas(`{cluster}`, system.query_log)
-- replace with the id returned by the INSERT above
WHERE initial_query_id =
AND type = 2;
```
```
┌─hostname()────┬──read_rows─┬─written_rows─┐
│ clickhouse-01 │ 1000000000 │ 49711 │
│ clickhouse-02 │ 336657272 │ 0 │ ← replica read but wrote nothing
└───────────────┴────────────┴──────────────┘
```
The same behaviour is observed with `ReplicatedAggregatingMergeTree`.
```sql
CREATE TABLE t2_amt ON CLUSTER `{cluster}` (`date` Date, `sum_value` SimpleAggregateFunction(sum,UInt64))
ENGINE = ReplicatedAggregatingMergeTree()
ORDER BY date;
```
### Describe the solution you'd like
When the destination table is a `SummingMergeTree` or an `AggregatingMergeTree`, nodes involved in the distributed insert select should write their partial aggregates.
It seems like [`buildInsertSelectPipelineParallelReplicas`](https://github.com/ClickHouse/ClickHouse/blob/c88c91a252e09806f0deca594cad55a52a11e42f/src/Interpreters/InterpreterInsertQuery.cpp#L653) calls [`isInsertSelectTrivialEnoughForDistributedExecution`](https://github.com/ClickHouse/ClickHouse/blob/c88c91a252e09806f0deca594cad55a52a11e42f/src/Interpreters/InterpreterInsertQuery.cpp#L613), which is agnostic of the destination table.
In case the destination table is a `SummingMergeTree` or `AggregatingMergeTree`, relax the restriction on `GROUP BY` and aggregate functions, and allow the distributed write to proceed.
### Describe alternatives you've considered
_No response_
### Additional context
_No response_
Contributor guide
Assessment
This issue has not been assessed yet.