ClickHouse / ClickHouse/ClickHouse

parallel_distributed_insert_select=2 falls back to single-node writes when SELECT has a GROUP BY, even for Summing/AggregatingMergeTree

Open
#98,787 2 comments 2 reactions 1 assignee Claimed by @devcrafter View on GitHub
comp-parallel-replicas feature
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.