From cba62d6caad51925b3ec2caf46a034c7d109f645 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 4 Sep 2026 13:30:55 +0200 Subject: [PATCH 1/3] Add a regression test for global RIGHT JOIN on Distributed tables The expected result matches initiator-side join semantics; the test currently fails because shards emit extra unmatched right-side rows. Related: https://github.com/Altinity/ClickHouse/issues/1664 Co-authored-by: Cursor --- ...58_distributed_global_right_join.reference | 36 ++++++++++++ .../05058_distributed_global_right_join.sql | 55 +++++++++++++++++++ 2 files changed, 91 insertions(+) create mode 100644 tests/queries/0_stateless/05058_distributed_global_right_join.reference create mode 100644 tests/queries/0_stateless/05058_distributed_global_right_join.sql diff --git a/tests/queries/0_stateless/05058_distributed_global_right_join.reference b/tests/queries/0_stateless/05058_distributed_global_right_join.reference new file mode 100644 index 000000000000..3277112362bc --- /dev/null +++ b/tests/queries/0_stateless/05058_distributed_global_right_join.reference @@ -0,0 +1,36 @@ +initiator +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +2 b 2 B +2 b 2 B +\N \N 3 C +\N \N 3 C +distributed_product_mode +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +2 b 2 B +2 b 2 B +\N \N 3 C +\N \N 3 C +explicit_global +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +2 b 2 B +2 b 2 B +\N \N 3 C +\N \N 3 C diff --git a/tests/queries/0_stateless/05058_distributed_global_right_join.sql b/tests/queries/0_stateless/05058_distributed_global_right_join.sql new file mode 100644 index 000000000000..b43495a419e7 --- /dev/null +++ b/tests/queries/0_stateless/05058_distributed_global_right_join.sql @@ -0,0 +1,55 @@ +-- Tags: distributed + +SET enable_analyzer = 1; +SET join_use_nulls = 1; + +DROP TABLE IF EXISTS left_local_05058; +DROP TABLE IF EXISTS right_local_05058; +DROP TABLE IF EXISTS left_distributed_05058; +DROP TABLE IF EXISTS right_distributed_05058; + +CREATE TABLE left_local_05058 (k1 UInt32, v1 String) +ENGINE = MergeTree +ORDER BY k1; + +CREATE TABLE right_local_05058 (k2 UInt32, v2 String) +ENGINE = MergeTree +ORDER BY k2; + +CREATE TABLE left_distributed_05058 AS left_local_05058 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), left_local_05058); + +CREATE TABLE right_distributed_05058 AS right_local_05058 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), right_local_05058); + +INSERT INTO left_local_05058 VALUES (1, 'a'), (2, 'b'), (4, 'd'); +INSERT INTO right_local_05058 VALUES (1, 'A'), (2, 'B'), (3, 'C'); + +-- The subqueries force the join to run on the initiator and define the correct result. +SELECT 'initiator'; +SELECT * +FROM (SELECT * FROM left_distributed_05058) AS l +RIGHT JOIN (SELECT * FROM right_distributed_05058) AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +-- The global join must produce the same rows and preserve the original column order. +SELECT 'distributed_product_mode'; +SELECT * +FROM left_distributed_05058 AS l +RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +SETTINGS distributed_product_mode = 'global' +FORMAT TSVWithNames; + +SELECT 'explicit_global'; +SELECT * +FROM left_distributed_05058 AS l +GLOBAL RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +DROP TABLE left_distributed_05058; +DROP TABLE right_distributed_05058; +DROP TABLE left_local_05058; +DROP TABLE right_local_05058; From 3ec16e41cc603914946dc85bd40e9f4a92fff9ef Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 4 Sep 2026 14:03:46 +0200 Subject: [PATCH 2/3] Rewrite global RIGHT JOIN of Distributed tables as LEFT JOIN Broadcast the original left table and keep the preserved right side sharded, so unmatched right rows are not duplicated on every shard. Related: https://github.com/Altinity/ClickHouse/issues/1664 Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 48 ++++++++++++++++++++++++++++++++- 1 file changed, 47 insertions(+), 1 deletion(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index dd4ef0a462a1..101b81a412d7 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -104,6 +104,7 @@ namespace Setting extern const SettingsMap additional_table_filters; extern const SettingsUInt64 allow_experimental_parallel_reading_from_replicas; extern const SettingsBool async_socket_for_remote; + extern const SettingsDistributedProductMode distributed_product_mode; extern const SettingsBool empty_result_for_aggregation_by_empty_set; extern const SettingsBool enable_unaligned_array_join; extern const SettingsBool join_use_nulls; @@ -129,6 +130,7 @@ namespace Setting extern const SettingsBool parallel_replicas_for_non_replicated_merge_tree; extern const SettingsUInt64 parallel_replicas_min_number_of_rows_per_replica; extern const SettingsUInt64 parallel_replica_offset; + extern const SettingsBool prefer_global_in_and_join; extern const SettingsBool optimize_move_to_prewhere; extern const SettingsBool optimize_move_to_prewhere_if_final; extern const SettingsBool use_concurrency_control; @@ -155,6 +157,47 @@ namespace ErrorCodes namespace { +const StorageDistributed * getDistributedStorageFromTableExpression(const QueryTreeNodePtr & table_expression) +{ + const auto * table_node = table_expression->as(); + if (!table_node) + return nullptr; + + return typeid_cast(table_node->getStorage().get()); +} + +void tryRewriteGlobalRightJoinAsLeftJoin(QueryNode & query_node, const ContextPtr & context) +{ + auto * join_node = query_node.getJoinTree()->as(); + if (!join_node + || join_node->getKind() != JoinKind::Right + || join_node->getStrictness() != JoinStrictness::All + || !join_node->isOnJoinExpression()) + return; + + const auto & settings = context->getSettingsRef(); + const auto distributed_product_mode = settings[Setting::distributed_product_mode]; + const bool is_global = join_node->getLocality() == JoinLocality::Global + || distributed_product_mode == DistributedProductMode::GLOBAL + || (distributed_product_mode != DistributedProductMode::LOCAL && settings[Setting::prefer_global_in_and_join]); + if (!is_global) + return; + + const auto * left_storage = getDistributedStorageFromTableExpression(join_node->getLeftTableExpression()); + const auto * right_storage = getDistributedStorageFromTableExpression(join_node->getRightTableExpression()); + if (!left_storage || !right_storage || left_storage->getShardCount() < 2 || right_storage->getShardCount() < 2) + return; + + /** A `GLOBAL RIGHT JOIN` cannot run with the left table sharded and the right table broadcast. + * Every shard would independently emit unmatched rows from the complete right table. + * Swap the inputs before choosing the `Distributed` table that will execute the query, so the + * preserved side stays sharded and the original left table is broadcast instead. + * Projection nodes are already resolved and keep the user-visible column order unchanged. + */ + std::swap(join_node->getLeftTableExpression(), join_node->getRightTableExpression()); + join_node->setKind(JoinKind::Left); +} + /// Check if current user has privileges to SELECT columns from table /// Throws an exception if access to any column from `column_names` is not granted /// If `column_names` is empty, check access to any columns and return names of accessible columns @@ -1976,7 +2019,10 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, const ColumnIdentifierSet & outer_scope_columns, PlannerContextPtr & planner_context) { - const QueryTreeNodePtr & join_tree_node = query_node->as().getJoinTree(); + auto & query_node_typed = query_node->as(); + tryRewriteGlobalRightJoinAsLeftJoin(query_node_typed, planner_context->getQueryContext()); + + const QueryTreeNodePtr & join_tree_node = query_node_typed.getJoinTree(); auto table_expressions_stack = buildTableExpressionsStack(join_tree_node); size_t table_expressions_stack_size = table_expressions_stack.size(); bool is_single_table_expression = table_expressions_stack_size == 1; From 5ca17e59261261cb4e7f5234a4540e4d68b7014f Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 4 Sep 2026 15:20:08 +0200 Subject: [PATCH 3/3] Skip wrapping IStorageCluster on GLOBAL INNER/LEFT joins After #94748 the planner always wrapped the left cluster source, so object_storage_cluster_join_mode='global' never saw the JOIN and ran on the initiator. Keep wrapping for allow/local and for RIGHT/FULL to avoid duplicated unmatched rows. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 33 +++++++++++++------ tests/integration/test_s3_cluster/test.py | 39 +++++++++++++++++++++++ 2 files changed, 63 insertions(+), 9 deletions(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 101b81a412d7..5d42f9a69662 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -117,6 +117,7 @@ namespace Setting extern const SettingsUInt64 max_parser_depth; extern const SettingsUInt64 max_query_size; extern const SettingsNonZeroUInt64 max_parallel_replicas; + extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; extern const SettingsFloat max_streams_to_max_threads_ratio; extern const SettingsMaxThreads max_threads; extern const SettingsUInt64 max_threads_min_free_memory_per_thread; @@ -2042,6 +2043,7 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, size_t joins_count = 0; bool is_full_join = false; + bool is_right_join = false; bool is_global_join = false; bool is_right_join_with_remote_table = false; int first_join_pos = -1; @@ -2070,6 +2072,8 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, if (join_kind == JoinKind::Full) is_full_join = true; + if (join_kind == JoinKind::Right) + is_right_join = true; if (join_node.getLocality() == JoinLocality::Global) is_global_join = true; @@ -2140,21 +2144,32 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, */ auto left_table_expression = table_expressions_stack.front(); - /** If the leftmost table uses IStorageCluster (e.g., s3Cluster, hdfsCluster) - * and there are multiple tables (indicating a JOIN), we must wrap it in a subquery. - * This prevents IStorageCluster from receiving the full JOIN query, which it cannot handle. + /** If the leftmost table uses `IStorageCluster` (e.g., `s3Cluster`, `hdfsCluster`) + * and there are multiple tables (indicating a JOIN), we normally wrap it in a subquery. + * This prevents `IStorageCluster` from receiving the full JOIN query, which it cannot handle. * - * IStorageCluster is a simple storage that just forwards queries to remote nodes. - * Unlike StorageDistributed, it cannot decompose and handle JOINs across multiple tables, + * `IStorageCluster` is a simple storage that just forwards queries to remote nodes. + * Unlike `StorageDistributed`, it cannot decompose and handle JOINs across multiple tables, * because remote nodes don't have access to other tables in the JOIN. * - * StorageDistributed has sophisticated query planning logic to handle JOINs and should + * With `object_storage_cluster_join_mode = 'global'`, `IStorageCluster` must receive a + * supported full query so it can materialize the right side and send a `GLOBAL JOIN` to + * remote nodes. `RIGHT JOIN` and `FULL JOIN` stay on the initiator because broadcasting their + * preserved right side would make every shard emit the same unmatched rows. + * + * `StorageDistributed` has sophisticated query planning logic to handle JOINs and should * NOT be wrapped (wrapping breaks tests like 03577_server_constant_folding). */ bool should_wrap_left_table = false; - bool has_multiple_tables = table_expressions_stack.size() > 1; - - if (has_multiple_tables) + const bool has_multiple_tables = table_expressions_stack.size() > 1; + const bool use_global_join + = planner_context->getQueryContext()->getSettingsRef()[Setting::object_storage_cluster_join_mode] + == ObjectStorageClusterJoinMode::GLOBAL + && joins_count > 0 + && !is_right_join + && !is_full_join; + + if (has_multiple_tables && !use_global_join) { // Get the actual storage to check its type auto * table_node = left_table_expression->as(); diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index ff13f493571d..6f241b0b1adb 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -1078,6 +1078,45 @@ def test_remote_no_hedged(started_cluster): assert TSV(pure_s3) == TSV(s3_distributed) +def test_global_join_executes_on_shards(started_cluster): + node = started_cluster.instances["s0_0_0"] + + node.query("DROP TABLE IF EXISTS join_table SYNC") + node.query( + """ + CREATE TABLE join_table ( + id UInt32, + name String + ) ENGINE=Memory() + """ + ) + + query = f""" + SELECT t1.name, t2.name FROM + s3Cluster('cluster_simple', + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') AS t1 + JOIN join_table AS t2 ON t1.value = t2.id + """ + + global_pipeline = node.query( + f"EXPLAIN PIPELINE {query} SETTINGS object_storage_cluster_join_mode='global'" + ) + assert "JoiningTransform" not in global_pipeline + + allow_pipeline = node.query( + f"EXPLAIN PIPELINE {query} SETTINGS object_storage_cluster_join_mode='allow'" + ) + assert "JoiningTransform" in allow_pipeline + + for join_kind in ("RIGHT", "FULL"): + outer_join_query = query.replace("JOIN join_table", f"{join_kind} JOIN join_table") + outer_join_pipeline = node.query( + f"EXPLAIN PIPELINE {outer_join_query} SETTINGS object_storage_cluster_join_mode='global'" + ) + assert "JoiningTransform" in outer_join_pipeline + + @pytest.mark.parametrize("join_mode", ["local", "global"]) def test_joins(started_cluster, join_mode): node = started_cluster.instances["s0_0_0"]