The INNER ANY JOIN between s3Cluster and icebergS3Cluster table functions with object_storage_cluster_join_mode='global' returns duplicate rows

[alsugiliazova]

I have this data

SELECT *
FROM s3Cluster(replicated_cluster, 'http://minio:9000/warehouse/data3/data/**.parquet', 'admin', 'password')

1. │ true        │     1000 │     456.78 │ Alice      │ 2024-01-01 12:00:00.000000 │ 2024-01-01 │ 1970-01-01 12:00:00.000000 │ 2024-01-01 12:00:00.000000 │        1000 │    456.78 │      456.78 │
2. │ false       │     2000 │     456.78 │ Bob        │ 2024-01-01 12:00:00.000000 │ 2024-01-01 │ 1970-01-01 12:00:00.000000 │ 2024-01-01 12:00:00.000000 │        2000 │    456.78 │      456.78 │
3. │ true        │     1000 │     456.78 │ Alice      │ 2024-01-01 12:00:00.000000 │ 2024-01-01 │ 1970-01-01 12:00:00.000000 │ 2024-01-01 12:00:00.000000 │        1000 │    456.78 │      456.78 │
4. │ false       │     2000 │     456.78 │ Bob        │ 2024-01-01 12:00:00.000000 │ 2024-01-01 │ 1970-01-01 12:00:00.000000 │ 2024-01-01 12:00:00.000000 │        2000 │    456.78 │      456.78 │
5. │ true        │     1000 │     456.78 │ Alice      │ 2024-01-01 12:00:00.000000 │ 2024-01-01 │ 1970-01-01 12:00:00.000000 │ 2024-01-01 12:00:00.000000 │        1000 │    456.78 │      456.78 │
6. │ false       │     2000 │     456.78 │ Bob        │ 2024-01-01 12:00:00.000000 │ 2024-01-01 │ 1970-01-01 12:00:00.000000 │ 2024-01-01 12:00:00.000000 │        2000 │    456.78 │      456.78 │
7. │ true        │     1000 │     456.78 │ Alice      │ 2024-01-01 12:00:00.000000 │ 2024-01-01 │ 1970-01-01 12:00:00.000000 │ 2024-01-01 12:00:00.000000 │        1000 │    456.78 │      456.78 │
8. │ false       │     2000 │     456.78 │ Bob        │ 2024-01-01 12:00:00.000000 │ 2024-01-01 │ 1970-01-01 12:00:00.000000 │ 2024-01-01 12:00:00.000000 │        2000 │    456.78 │      456.78 │
   └─────────────┴──────────┴────────────┴────────────┴────────────────────────────┴────────────┴────────────────────────────┴────────────────────────────┴─────────────┴───────────┴─────────────┘

8 rows in set. Elapsed: 0.119 sec. 

SELECT *
FROM icebergS3Cluster(replicated_cluster, 'http://minio:9000/warehouse/data1', 'admin', 'password')

1. │ true        │     1000 │     456.78 │ Alice      │ 2024-01-01 13:00:00.000000 │ 2024-01-01 │ 43200000000 │ 2024-01-01 12:00:00.000000 │        1000 │    456.78 │      456.78 │
2. │ false       │     2000 │     456.78 │ Bob        │ 2024-01-01 13:00:00.000000 │ 2024-01-01 │ 43200000000 │ 2024-01-01 12:00:00.000000 │        2000 │    456.78 │      456.78 │
3. │ true        │     1000 │     456.78 │ Alice      │ 2024-01-01 13:00:00.000000 │ 2024-01-01 │ 43200000000 │ 2024-01-01 12:00:00.000000 │        1000 │    456.78 │      456.78 │
4. │ false       │     2000 │     456.78 │ Bob        │ 2024-01-01 13:00:00.000000 │ 2024-01-01 │ 43200000000 │ 2024-01-01 12:00:00.000000 │        2000 │    456.78 │      456.78 │
5. │ true        │     1000 │     456.78 │ Alice      │ 2024-01-01 13:00:00.000000 │ 2024-01-01 │ 43200000000 │ 2024-01-01 12:00:00.000000 │        1000 │    456.78 │      456.78 │
6. │ false       │     2000 │     456.78 │ Bob        │ 2024-01-01 13:00:00.000000 │ 2024-01-01 │ 43200000000 │ 2024-01-01 12:00:00.000000 │        2000 │    456.78 │      456.78 │
7. │ true        │     1000 │     456.78 │ Alice      │ 2024-01-01 13:00:00.000000 │ 2024-01-01 │ 43200000000 │ 2024-01-01 12:00:00.000000 │        1000 │    456.78 │      456.78 │
8. │ false       │     2000 │     456.78 │ Bob        │ 2024-01-01 13:00:00.000000 │ 2024-01-01 │ 43200000000 │ 2024-01-01 12:00:00.000000 │        2000 │    456.78 │      456.78 │
   └─────────────┴──────────┴────────────┴────────────┴────────────────────────────┴────────────┴─────────────┴────────────────────────────┴─────────────┴───────────┴─────────────┘

8 rows in set. Elapsed: 0.132 sec. 

I also have merge tree tables with same exact data and expected join result is calculated using those merge tree tables:

SELECT *
FROM table_3898404a_3a8b_11f1_995b_de7b9eea348f AS t1
ANY INNER JOIN table_38c3eb3c_3a8b_11f1_995b_de7b9eea348f AS t2 ON (t1.boolean_col = t2.boolean_col) AND (t1.string_col = t2.string_col)
ORDER BY tuple(*) ASC
FORMAT TabSeparated

Query id: 46d4e498-66ab-4038-a7cb-21572550a760

false	2000	456.78	Bob	2024-01-01 12:00:00.000000	2024-01-01	1970-01-01 12:00:00.000000	2024-01-01 12:00:00.000000	2000	456.78	456.78	false	2000	456.78	Bob	2024-01-01 13:00:00.000000	2024-01-01	43200000000	2024-01-01 12:00:00.000000	2000	456.78	456.78
true	1000	456.78	Alice	2024-01-01 12:00:00.000000	2024-01-01	1970-01-01 12:00:00.000000	2024-01-01 12:00:00.000000	1000	456.78	456.78	true	1000	456.78	Alice	2024-01-01 13:00:00.000000	2024-01-01	43200000000	2024-01-01 12:00:00.000000	1000	456.78	456.78

2 rows in set. Elapsed: 0.010 sec. 

The same join but with object_storage_cluster_join_mode = 'global', object_storage_cluster = 'replicated_cluster' returnes 6 rows instead of 2:

SELECT *
FROM s3Cluster(replicated_cluster, 'http://minio:9000/warehouse/data3/data/**.parquet', 'admin', 'password') AS t1
ANY INNER JOIN icebergS3Cluster(replicated_cluster, 'http://minio:9000/warehouse/data1', 'admin', 'password') AS t2 ON (t1.boolean_col = t2.boolean_col) AND (t1.string_col = t2.string_col)
ORDER BY tuple(*) ASC
SETTINGS object_storage_cluster_join_mode = 'global', object_storage_cluster = 'replicated_cluster'
FORMAT TabSeparated

Query id: af48112f-995c-47fe-b5e1-a5466955f9be

false	2000	456.78	Bob	2024-01-01 12:00:00.000000	2024-01-01	1970-01-01 12:00:00.000000	2024-01-01 12:00:00.000000	2000	456.78	456.78	false	2000	456.78	Bob	2024-01-01 13:00:00.000000	2024-01-01	43200000000	2024-01-01 12:00:00.000000	2000	456.78	456.78
false	2000	456.78	Bob	2024-01-01 12:00:00.000000	2024-01-01	1970-01-01 12:00:00.000000	2024-01-01 12:00:00.000000	2000	456.78	456.78	false	2000	456.78	Bob	2024-01-01 13:00:00.000000	2024-01-01	43200000000	2024-01-01 12:00:00.000000	2000	456.78	456.78
false	2000	456.78	Bob	2024-01-01 12:00:00.000000	2024-01-01	1970-01-01 12:00:00.000000	2024-01-01 12:00:00.000000	2000	456.78	456.78	false	2000	456.78	Bob	2024-01-01 13:00:00.000000	2024-01-01	43200000000	2024-01-01 12:00:00.000000	2000	456.78	456.78
true	1000	456.78	Alice	2024-01-01 12:00:00.000000	2024-01-01	1970-01-01 12:00:00.000000	2024-01-01 12:00:00.000000	1000	456.78	456.78	true	1000	456.78	Alice	2024-01-01 13:00:00.000000	2024-01-01	43200000000	2024-01-01 12:00:00.000000	1000	456.78	456.78
true	1000	456.78	Alice	2024-01-01 12:00:00.000000	2024-01-01	1970-01-01 12:00:00.000000	2024-01-01 12:00:00.000000	1000	456.78	456.78	true	1000	456.78	Alice	2024-01-01 13:00:00.000000	2024-01-01	43200000000	2024-01-01 12:00:00.000000	1000	456.78	456.78
true	1000	456.78	Alice	2024-01-01 12:00:00.000000	2024-01-01	1970-01-01 12:00:00.000000	2024-01-01 12:00:00.000000	1000	456.78	456.78	true	1000	456.78	Alice	2024-01-01 13:00:00.000000	2024-01-01	43200000000	2024-01-01 12:00:00.000000	1000	456.78	456.78

6 rows in set. Elapsed: 0.427 sec. 

[alsugiliazova]

Root Cause Analysis: Issue #1661 — ANY INNER JOIN duplicates with object_storage_cluster_join_mode='global'

Issue: Altinity/ClickHouse #1661

Branch: feature/antalya-26.1/json_part2 (PR #1527 — Cluster Joins part 2)

Confirmed defect

Severity: High — correctness failure, returns wrong number of rows under realistic workload.

Title: ANY INNER JOIN deduplication is per-shard instead of global in object_storage_cluster_join_mode='global'

Impact: Queries using ANY INNER JOIN (or other ANY strictness variants) between cluster table functions with object_storage_cluster_join_mode='global' return num_shards × expected_rows instead of expected_rows.

Affected file/function: src/Storages/IStorageCluster.cpp — IStorageCluster::getQueryProcessingStage

Proof sketch

1. ANY INNER JOIN in ClickHouse's HashJoin uses used_flags.setUsedOnce() to output only one row per unique join key — the first left row that matches a right key claims that key; all subsequent left rows with the same key are filtered out (src/Interpreters/HashJoin/HashJoinMethodsImpl.h, lines 438–448).

2. In LOCAL mode, getQueryProcessingStage returns FetchColumns when a join is detected (line 654). This means shards only send raw left-side data back to the initiator, and the join executes once on the initiator. The setUsedOnce tracking works globally — correct result.

3. In GLOBAL mode, there is no special handling in getQueryProcessingStage — it falls through to WithMergeableState (line 661). This means each shard independently executes the full ANY INNER JOIN query (with the right side materialized into a temp table). Each shard has its own local used_flags, so setUsedOnce fires independently on every shard. Each shard that has left-side rows matching a key produces one output row for that key.

4. Results from all shards are concatenated by the initiator (ReadFromCluster::initializePipeline, line 534) with no re-application of ANY deduplication.

5. With a 3-shard cluster where each shard has left-side rows for both keys (Alice/true and Bob/false): each shard outputs 2 rows → total 6 rows instead of 2.

Reproduction steps

1. Set up a 3-node replicated_cluster.
2. Create s3/iceberg parquet data with duplicated rows across multiple files (e.g., 4 files × 2 rows each).
3. Run SELECT * FROM s3Cluster(...) AS t1 ANY INNER JOIN icebergS3Cluster(...) AS t2 ON ... SETTINGS object_storage_cluster_join_mode = 'global'.
4. Observe 6 rows returned instead of the expected 2.

Likely fix direction

Handle GLOBAL mode in getQueryProcessingStage to avoid pushing ANY-strictness joins down to shards — either return FetchColumns (same as LOCAL), or add a post-join deduplication step on the initiator, or rewrite ANY to ALL in the shard query and re-apply ANY semantics on the initiator.

Regression test direction

Add a test that verifies ANY INNER JOIN between cluster table functions with object_storage_cluster_join_mode='global' returns the same row count as the equivalent single-node MergeTree join.

Affected subsystem and blast radius

Affects all ANY-strictness joins (ANY INNER, ANY LEFT) executed via object_storage_cluster_join_mode='global' on any cluster table function (s3Cluster, icebergS3Cluster, etc.). The LOCAL and ALLOW modes are not affected.

Coverage summary

• Scope reviewed: IStorageCluster::getQueryProcessingStage, IStorageCluster::updateQueryWithJoinToSendIfNeeded (GLOBAL path), buildQueryTreeForShard, rewriteJoinToGlobalJoin, ReadFromCluster::initializePipeline, HashJoin ANY INNER implementation.
• Categories failed: Distributed join correctness for ANY strictness in GLOBAL mode.
• Categories passed: GLOBAL join rewrite mechanics (temp table materialization, locality marking); LOCAL mode join correctness; query tree cloning and replacement; external table propagation.
• Assumptions/limits: Static code-path analysis only; cluster shard count (3) inferred from the 6-vs-2 row ratio in the issue report.

[ianton-ru]

Join over cluster function reuses algorithm of join over distributed tables.
Join over distributed tables has the same issue in upstream.
On upstream version 26.2.4.23:
Cluster 'shards' has three shard, each with one replica.

create table t1 on cluster 'shards' (k1 UInt32, v1 String) engine=MergeTree() order by k1;
create table t2 on cluster 'shards' (k2 UInt32, v2 String) engine=MergeTree() order by k2;
create table d1 on cluster 'shards' as t1 engine=Distributed('shards', default, t1, rand());
create table d2 on cluster 'shards' as t2 engine=Distributed('shards', default, t2, rand());
insert into d1 values (1, 'a'),(2, 'b'),(3, 'c'),(1, 'a'),(2, 'b'),(3, 'c'),(1, 'a'),(2, 'b'),(3, 'c');
insert into d2 values (1, 'A'),(2, 'B'),(3, 'C'),(1, 'A'),(2, 'B'),(3, 'C'),(1, 'A'),(2, 'B'),(3, 'C');

# Correct result, :
select * from (select * from d1) as d1 any inner join (select * from d2) as d2 on d1.k1=d2.k2 order by all
   ┌─k1─┬─v1─┬─k2─┬─v2─┐
1. │  1 │ a  │  1 │ A  │
2. │  2 │ b  │  2 │ B  │
3. │  3 │ c  │  3 │ C  │
   └────┴────┴────┴────┘

# Broken result with global join:
select * from d1 any inner join d2 on d1.k1=d2.k2 order by all settings prefer_global_in_and_join=1
   ┌─k1─┬─v1─┬─k2─┬─v2─┐
1. │  1 │ a  │  1 │ A  │
2. │  1 │ a  │  1 │ A  │
3. │  1 │ a  │  1 │ A  │
4. │  2 │ b  │  2 │ B  │
5. │  2 │ b  │  2 │ B  │
6. │  2 │ b  │  2 │ B  │
7. │  3 │ c  │  3 │ C  │
8. │  3 │ c  │  3 │ C  │
   └────┴────┴────┴────┘

# different setting  distributed_product_mode='global', actually has here the same effect as prefer_global_in_and_join=1
select * from d1 any inner join d2 on d1.k1=d2.k2 order by all settings distributed_product_mode='global'
   ┌─k1─┬─v1─┬─k2─┬─v2─┐
1. │  1 │ a  │  1 │ A  │
2. │  1 │ a  │  1 │ A  │
3. │  1 │ a  │  1 │ A  │
4. │  2 │ b  │  2 │ B  │
5. │  2 │ b  │  2 │ B  │
6. │  2 │ b  │  2 │ B  │
7. │  3 │ c  │  3 │ C  │
8. │  3 │ c  │  3 │ C  │
   └────┴────┴────┴────┘

See ClickHouse#65478 for example

Issue confirmed, but more global than current PR.