ClickHouse / ClickHouse/ClickHouse
max_parallel_replicas=2 gives incorrect results when joining to a non-replicated table
- Dominant language
- C++
- Stars
- 49.9k
- Forks
- 9k
- Avg merge
- 21h 32m
- Merged PRs (30d)
- 515
Description
(you don't have to strictly follow this form)
**Description**
When max_parallel_replicas=2, doing a distributed join with a replicated table with sampling, and a non-replicated table (with identical contents on all nodes) misses some rows from the **non-replicated** table
My use case is to try and use all available nodes by creating a carefully-chosen sampling key and sorting by it. This gives the expected performance increase, and correct results, when not joining to another table. The problem only happens with joins.
**How to reproduce**
* The problem appears on Version 20.11.4 revision 54442
* cluster configuration: 2 shards with 2 replicas each .. let me know if a config.xml extract is needed.
* settings:
set prefer_localhost_replica=0;
set max_parallel_replicas=2;
* Sample SQL:
create database test on cluster test;
create table test.large_sharded_replicated on cluster test
(sample_key UInt32, shard_column Int32, foreign_key Int32)
engine = ReplicatedMergeTree('/clickhouse/tables/replicated/{shard}/tmp_large_sharded_replicated', '{replica}')
sample by sample_key
order by (sample_key);
create table test.large_sharded_replicated_dist on cluster test as test.large_sharded_replicated
engine = Distributed(test, test, large_sharded_replicated, shard_column);
create table test.medium_nonsharded on cluster test
(foreign_key Int32, foreign_value Int32)
engine = MergeTree()
order by (foreign_key);
insert into test.large_sharded_replicated_dist (sample_key, shard_column, foreign_key)
values
(0, 1, 10), (0, 2, 10), (0x80000000, 1, 10), (0x80000000, 2, 10);
-- Run this on all four nodes
insert into test.medium_nonsharded values (10, 100);
select * from test.large_sharded_replicated_dist
left outer join test.medium_nonsharded using (foreign_key);
```
┌─sample_key─┬─shard_column─┬─foreign_key─┬─foreign_value─┐
│ 0 │ 2 │ 10 │ 100 │
└────────────┴──────────────┴─────────────┴───────────────┘
┌─sample_key─┬─shard_column─┬─foreign_key─┬─foreign_value─┐
│ 2147483648 │ 2 │ 10 │ 0 │
└────────────┴──────────────┴─────────────┴───────────────┘
┌─sample_key─┬─shard_column─┬─foreign_key─┬─foreign_value─┐
│ 0 │ 1 │ 10 │ 100 │
└────────────┴──────────────┴─────────────┴───────────────┘
┌─sample_key─┬─shard_column─┬─foreign_key─┬─foreign_value─┐
│ 2147483648 │ 1 │ 10 │ 0 │
└────────────┴──────────────┴─────────────┴───────────────┘
```
**Expected behavior**
The above example shows that all rows of large_sharded_replicated are being scanned ; and that each node returns a single row, as expected based on the values of sample_key and shard_column. What should happen is that foreign_value should be 100 on every row, because medium_nonsharded shouldn't be affected by sharding or replication.
**Error message and/or stacktrace**
I think this is the relevant message (set medium_nonsharded) that appears on the nodes that do the join incorrectly:
```
test.medium_nonsharded (ed269f10-1dc7-4cc7-94f6-0b6ce4a5739b) (SelectExecutor): Will use no data on this replica because parallel replicas processing has been requested (the setting 'max_parallel_replicas') but the table does not support sampling and this replica is not the first.
```
**Additional context**
This problem seems to have been introduced in 1bcf22d, which fixes incorrect results coming from a distributed query where the main table has no sampling.
I don't know enough about the ClickHouse code base to suggest a fix. I would expect that the "use_sampling" logic should only apply to tables that were distributed in the original query. Even for a table that does support sampling but isn't used as a distributed table, sampling shouldn't be applied. But that information may be lost by the time the replicas receive the query.
Contributor guide
Assessment
This issue has not been assessed yet.