From cba62d6caad51925b3ec2caf46a034c7d109f645 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 4 Sep 2026 13:30:55 +0200 Subject: [PATCH 1/5] 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/5] 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/5] 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"] From 3df7433ba7243005eb6c91d010130a6e70a8d825 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 8 Sep 2026 14:41:09 +0200 Subject: [PATCH 4/5] Widen the global RIGHT JOIN rewrite of Distributed tables Port of the review fixes made for the `master` pull request https://github.com/ClickHouse/ClickHouse/pull/118611 `tryRewriteGlobalRightJoinAsLeftJoin` only recognised a narrow shape of the query it was meant to fix, so a `GLOBAL RIGHT JOIN` whose left table is a sharded `Distributed` table still emitted the rows preserved from the broadcast side once per shard in these cases: * `USING` and `NATURAL` joins were skipped, because only `ON` was accepted. A `USING` key stores its sides positionally, so the key pairs are now swapped together with the table expressions, and a key that does not hold a plain column per side leaves the query alone. * The rewrite required the right side to be a sharded `Distributed` table too. Only the left table fans the query out across shards, so the gate is now on the left table alone, and a local table, a one shard `Distributed` table, a subquery or a table function on the right is handled as well. `Any`, `Semi` and `Anti` join over to the swapped sides the same way `All` does and are accepted; `Asof` and `RightAny` do not mirror and stay out. * Only the topmost join was examined. Join trees are left deep, so the walk now descends to the join that reads the leftmost table. A `RIGHT` or `FULL` join that survives the rewrite, such as the one in `(A INNER C) RIGHT B`, cannot be swapped without moving a join into the right table expression. Those now read the leftmost table through a subquery, which keeps the join on the initiator, the only remaining way to emit the preserved rows once. `getDistributedStorageFromTableExpression` also missed `remote` and `clusterAllReplicas` on the left, because it only looked at `TableNode`. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 121 +++++- ...58_distributed_global_right_join.reference | 384 ++++++++++++++++++ .../05058_distributed_global_right_join.sql | 331 +++++++++++++++ 3 files changed, 816 insertions(+), 20 deletions(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 5d42f9a69662..46e71b6cc3c3 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -37,6 +37,7 @@ #include #include #include +#include #include #include #include @@ -161,41 +162,101 @@ namespace const StorageDistributed * getDistributedStorageFromTableExpression(const QueryTreeNodePtr & table_expression) { const auto * table_node = table_expression->as(); - if (!table_node) - return nullptr; + if (table_node) + return typeid_cast(table_node->getStorage().get()); + + const auto * table_function_node = table_expression->as(); + if (table_function_node) + return typeid_cast(table_function_node->getStorage().get()); + + return nullptr; +} - return typeid_cast(table_node->getStorage().get()); +bool isGlobalJoin(const JoinNode & join_node, const Settings & settings) +{ + const auto distributed_product_mode = settings[Setting::distributed_product_mode]; + return join_node.getLocality() == JoinLocality::Global + || distributed_product_mode == DistributedProductMode::GLOBAL + || (distributed_product_mode != DistributedProductMode::LOCAL && settings[Setting::prefer_global_in_and_join]); } void tryRewriteGlobalRightJoinAsLeftJoin(QueryNode & query_node, const ContextPtr & context) { + /** Join trees are left deep, so the join that reads the leftmost table is the deepest one, and it is + * the only one whose sides can be swapped without moving a join into the right table expression. + */ auto * join_node = query_node.getJoinTree()->as(); - if (!join_node - || join_node->getKind() != JoinKind::Right - || join_node->getStrictness() != JoinStrictness::All - || !join_node->isOnJoinExpression()) + while (join_node) + { + auto * deeper_join_node = join_node->getLeftTableExpression()->as(); + if (!deeper_join_node) + break; + join_node = deeper_join_node; + } + + if (!join_node || join_node->getKind() != JoinKind::Right || !join_node->hasJoinExpression()) 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) + /** These strictnesses mirror when both the table expressions and the kind are flipped. + * `Asof` does not: its last key is an inequality, and swapping the sides reverses its direction. + * `RightAny` does not either, because the strictness itself names the side to take a row from, + * and that name does not follow the tables across the swap. + */ + const auto strictness = join_node->getStrictness(); + if (strictness != JoinStrictness::All && strictness != JoinStrictness::Any + && strictness != JoinStrictness::Semi && strictness != JoinStrictness::Anti) return; + if (!isGlobalJoin(*join_node, context->getSettingsRef())) + return; + + /// Only the left table fans the query out across shards, so only its shard count decides whether + /// the rows of the preserved side get emitted more than once. What the right side is does not matter. 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) + if (!left_storage || left_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. + /** A `JOIN USING` key records its sides positionally, the left one first. The join condition, the + * `USING (a AS b)` clause shipped to the shards and the key supertype all read that order, so the + * sides have to be swapped together with the table expressions. A key that does not hold a plain + * column per side is not swappable that way, so leave such a query alone. `NATURAL` needs no separate + * handling: the analyzer has already turned it into `USING` by now. + */ + std::vector using_key_sides; + if (join_node->isUsingJoinExpression()) + { + for (const auto & using_key : join_node->getJoinExpression()->as().getNodes()) + { + auto * using_column = using_key->as(); + if (!using_column || !using_column->hasExpression()) + return; + + auto * key_sides = using_column->getExpression()->as(); + if (!key_sides || key_sides->getNodes().size() != 2) + return; + + for (const auto & side : key_sides->getNodes()) + { + const auto * side_column = side->as(); + if (!side_column || side_column->hasExpression()) + return; + } + + using_key_sides.push_back(key_sides); + } + } + + /** A `GLOBAL RIGHT JOIN` cannot run with the left table sharded and the right side broadcast. + * Every shard would independently emit the rows of the complete right side that the kind preserves. + * Swap the inputs before choosing the table expression that will execute the query, so the preserved + * side moves out of the broadcast position. It then keeps running on the shards if it is a sharded + * `Distributed` table of its own, and falls back to the initiator otherwise, which is slower but is + * the only way to emit those rows once. * Projection nodes are already resolved and keep the user-visible column order unchanged. */ std::swap(join_node->getLeftTableExpression(), join_node->getRightTableExpression()); + for (auto * key_sides : using_key_sides) + std::swap(key_sides->getNodes()[0], key_sides->getNodes()[1]); join_node->setKind(JoinKind::Left); } @@ -2049,6 +2110,7 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, int first_join_pos = -1; int last_right_join_pos = -1; bool is_cross_join = false; + bool has_global_join_preserving_broadcast_rows = false; /// For each table, table function, query, union table expressions prepare before query plan build for (size_t i = 0; i < table_expressions_stack_size; ++i) { @@ -2078,6 +2140,13 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, if (join_node.getLocality() == JoinLocality::Global) is_global_join = true; + /// Rows of the right side are preserved by these kinds, and that side is broadcast whole to + /// every shard. `tryRewriteGlobalRightJoinAsLeftJoin` swaps the sides where it can, so a join + /// still standing here would emit those rows once per shard. + if ((join_kind == JoinKind::Right || join_kind == JoinKind::Full) + && isGlobalJoin(join_node, planner_context->getQueryContext()->getSettingsRef())) + has_global_join_preserving_broadcast_rows = true; + // save join positions for later check if (first_join_pos < 0 && (join_kind == JoinKind::Left || join_kind == JoinKind::Inner || join_kind == JoinKind::Right)) first_join_pos = static_cast(i); @@ -2158,7 +2227,8 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, * 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). + * NOT be wrapped (wrapping breaks tests like 03577_server_constant_folding), except for the + * join trees that stay wrong when the join runs per shard, handled below. */ bool should_wrap_left_table = false; const bool has_multiple_tables = table_expressions_stack.size() > 1; @@ -2181,6 +2251,17 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, // Only wrap if it's specifically IStorageCluster, not StorageDistributed or other remote storages should_wrap_left_table = (dynamic_cast(storage.get()) != nullptr); } + + /** Reading the leftmost table through a subquery keeps the join on the initiator instead of running + * it on every shard, which is the only way left to emit the preserved rows once. It costs the + * distributed execution of the join, and shard specific values such as `shardNum` stop varying, + * so do it only for the join trees that are wrong without it. + */ + if (!should_wrap_left_table && has_global_join_preserving_broadcast_rows) + { + const auto * left_storage = getDistributedStorageFromTableExpression(left_table_expression); + should_wrap_left_table = left_storage && left_storage->getShardCount() > 1; + } } auto left_table_expression_query_plan = buildQueryPlanForTableExpression( diff --git a/tests/queries/0_stateless/05058_distributed_global_right_join.reference b/tests/queries/0_stateless/05058_distributed_global_right_join.reference index 3277112362bc..2e645c47b92b 100644 --- a/tests/queries/0_stateless/05058_distributed_global_right_join.reference +++ b/tests/queries/0_stateless/05058_distributed_global_right_join.reference @@ -34,3 +34,387 @@ k1 v1 k2 v2 2 b 2 B \N \N 3 C \N \N 3 C +using_initiator +k v1 v2 +1 a A +1 a A +1 a A +1 a A +2 b B +2 b B +2 b B +2 b B +3 \N C +3 \N C +using_global +k v1 v2 +1 a A +1 a A +1 a A +1 a A +2 b B +2 b B +2 b B +2 b B +3 \N C +3 \N C +using_distributed_product_mode +k v1 v2 +1 a A +1 a A +1 a A +1 a A +2 b B +2 b B +2 b B +2 b B +3 \N C +3 \N C +using_multiple_keys_global +k k2 v1 v2 +1 1 a A +1 1 a A +1 1 a A +1 1 a A +2 2 b B +2 2 b B +2 2 b B +2 2 b B +3 3 \N C +3 3 \N C +natural_initiator +k v1 v2 +1 a A +1 a A +1 a A +1 a A +2 b B +2 b B +2 b B +2 b B +3 \N C +3 \N C +natural_global +k v1 v2 +1 a A +1 a A +1 a A +1 a A +2 b B +2 b B +2 b B +2 b B +3 \N C +3 \N C +using_alias_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 +3 \N 3 C +3 \N 3 C +using_alias_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 +3 \N 3 C +3 \N 3 C +using_left_join_global +k v1 v2 +1 a A +1 a A +1 a A +1 a A +2 b B +2 b B +2 b B +2 b B +4 d \N +4 d \N +right_is_local_table_initiator +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +\N \N 3 C +right_is_local_table_global +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +\N \N 3 C +right_is_one_shard_initiator +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +\N \N 3 C +right_is_one_shard_global +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +\N \N 3 C +right_is_subquery_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 +right_is_subquery_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 +right_is_table_function_initiator +k1 v1 n +1 a 1 +1 a 1 +2 b 2 +2 b 2 +4 d 4 +4 d 4 +\N \N 0 +\N \N 3 +right_is_table_function_global +k1 v1 n +1 a 1 +1 a 1 +2 b 2 +2 b 2 +4 d 4 +4 d 4 +\N \N 0 +\N \N 3 +any_initiator +k2 v2 +1 A +2 B +3 C +any_global +k2 v2 +1 A +2 B +3 C +semi_initiator +k2 v2 +1 A +2 B +semi_global +k2 v2 +1 A +2 B +anti_initiator +k2 v2 +3 C +anti_global +k2 v2 +3 C +left_is_one_shard_initiator +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +\N \N 3 C +\N \N 3 C +left_is_one_shard_global +k1 v1 k2 v2 +1 a 1 A +1 a 1 A +2 b 2 B +2 b 2 B +\N \N 3 C +\N \N 3 C +nested_right_then_inner_initiator +k1 v1 k2 v2 k3 v3 +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N 3 C 3 Z +nested_right_then_inner_global +k1 v1 k2 v2 k3 v3 +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N 3 C 3 Z +nested_right_then_right_initiator +k1 v1 k2 v2 k3 v3 +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N \N \N 4 W +\N \N \N \N 4 W +nested_right_then_right_global +k1 v1 k2 v2 k3 v3 +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N 3 C 3 Z +\N \N \N \N 4 W +\N \N \N \N 4 W +nested_inner_then_right_initiator +k1 v1 k2 v2 k3 v3 +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +\N \N 3 C \N \N +\N \N 3 C \N \N +nested_inner_then_right_global +k1 v1 k2 v2 k3 v3 +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +1 a 1 A 1 X +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +2 b 2 B 2 Y +\N \N 3 C \N \N +\N \N 3 C \N \N +full_join_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 +4 d \N \N +4 d \N \N +\N \N 3 C +\N \N 3 C +full_join_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 +4 d \N \N +4 d \N \N +\N \N 3 C +\N \N 3 C +shard_num_is_untouched_without_right_join +s +1 +2 diff --git a/tests/queries/0_stateless/05058_distributed_global_right_join.sql b/tests/queries/0_stateless/05058_distributed_global_right_join.sql index b43495a419e7..620d9931cbc8 100644 --- a/tests/queries/0_stateless/05058_distributed_global_right_join.sql +++ b/tests/queries/0_stateless/05058_distributed_global_right_join.sql @@ -49,6 +49,337 @@ GLOBAL RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 ORDER BY ALL FORMAT TSVWithNames; +-- The same rewrite must happen for the USING carrier, and for NATURAL, which the analyzer turns into USING. +DROP TABLE IF EXISTS shared_left_local_05058; +DROP TABLE IF EXISTS shared_right_local_05058; +DROP TABLE IF EXISTS shared_left_distributed_05058; +DROP TABLE IF EXISTS shared_right_distributed_05058; + +CREATE TABLE shared_left_local_05058 (k UInt32, k2 UInt32, v1 String) +ENGINE = MergeTree +ORDER BY k; + +CREATE TABLE shared_right_local_05058 (k UInt32, k2 Int64, v2 String) +ENGINE = MergeTree +ORDER BY k; + +CREATE TABLE shared_left_distributed_05058 AS shared_left_local_05058 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), shared_left_local_05058); + +CREATE TABLE shared_right_distributed_05058 AS shared_right_local_05058 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), shared_right_local_05058); + +INSERT INTO shared_left_local_05058 VALUES (1, 1, 'a'), (2, 2, 'b'), (4, 4, 'd'); +INSERT INTO shared_right_local_05058 VALUES (1, 1, 'A'), (2, 2, 'B'), (3, 3, 'C'); + +SELECT 'using_initiator'; +SELECT k, v1, v2 +FROM (SELECT * FROM shared_left_distributed_05058) AS l +RIGHT JOIN (SELECT * FROM shared_right_distributed_05058) AS r USING (k) +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'using_global'; +SELECT k, v1, v2 +FROM shared_left_distributed_05058 AS l +GLOBAL RIGHT JOIN shared_right_distributed_05058 AS r USING (k) +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'using_distributed_product_mode'; +SELECT k, v1, v2 +FROM shared_left_distributed_05058 AS l +RIGHT JOIN shared_right_distributed_05058 AS r USING (k) +ORDER BY ALL +SETTINGS distributed_product_mode = 'global' +FORMAT TSVWithNames; + +-- Two keys at once, the second one needing a common supertype. +SELECT 'using_multiple_keys_global'; +SELECT k, k2, v1, v2 +FROM shared_left_distributed_05058 AS l +GLOBAL RIGHT JOIN shared_right_distributed_05058 AS r USING (k, k2) +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'natural_initiator'; +SELECT * +FROM (SELECT k, v1 FROM shared_left_distributed_05058) AS l +NATURAL RIGHT JOIN (SELECT k, v2 FROM shared_right_distributed_05058) AS r +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'natural_global'; +SELECT * +FROM (SELECT k, v1 FROM shared_left_distributed_05058) AS l +GLOBAL NATURAL RIGHT JOIN (SELECT k, v2 FROM shared_right_distributed_05058) AS r +ORDER BY ALL +FORMAT TSVWithNames; + +-- USING (a AS b) takes the key from the left table as `a` and from the right one as `b`. +SELECT 'using_alias_initiator'; +SELECT * +FROM (SELECT * FROM left_distributed_05058) AS l +RIGHT JOIN (SELECT * FROM right_distributed_05058) AS r USING (k1 AS k2) +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'using_alias_global'; +SELECT * +FROM left_distributed_05058 AS l +GLOBAL RIGHT JOIN right_distributed_05058 AS r USING (k1 AS k2) +ORDER BY ALL +FORMAT TSVWithNames; + +-- LEFT JOIN is not rewritten and must keep working. +SELECT 'using_left_join_global'; +SELECT k, v1, v2 +FROM shared_left_distributed_05058 AS l +GLOBAL LEFT JOIN shared_right_distributed_05058 AS r USING (k) +ORDER BY ALL +FORMAT TSVWithNames; + +-- Only the left table fans the query out, so the right side may be anything at all. +-- Every result below is paired with the same join forced onto the initiator, which defines the correct rows. +DROP TABLE IF EXISTS right_one_shard_05058; + +CREATE TABLE right_one_shard_05058 AS right_local_05058 +ENGINE = Distributed('test_shard_localhost', currentDatabase(), right_local_05058); + +SELECT 'right_is_local_table_initiator'; +SELECT k1, v1, k2, v2 +FROM (SELECT * FROM left_distributed_05058) AS l +RIGHT JOIN right_local_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'right_is_local_table_global'; +SELECT k1, v1, k2, v2 +FROM left_distributed_05058 AS l +GLOBAL RIGHT JOIN right_local_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'right_is_one_shard_initiator'; +SELECT k1, v1, k2, v2 +FROM (SELECT * FROM left_distributed_05058) AS l +RIGHT JOIN right_one_shard_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'right_is_one_shard_global'; +SELECT k1, v1, k2, v2 +FROM left_distributed_05058 AS l +GLOBAL RIGHT JOIN right_one_shard_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'right_is_subquery_initiator'; +SELECT k1, v1, k2, v2 +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; + +SELECT 'right_is_subquery_global'; +SELECT k1, v1, k2, v2 +FROM left_distributed_05058 AS l +GLOBAL RIGHT JOIN (SELECT * FROM right_distributed_05058) AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'right_is_table_function_initiator'; +SELECT k1, v1, n +FROM (SELECT * FROM left_distributed_05058) AS l +RIGHT JOIN (SELECT number AS n FROM numbers(5)) AS r ON l.k1 = r.n +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'right_is_table_function_global'; +SELECT k1, v1, n +FROM left_distributed_05058 AS l +GLOBAL RIGHT JOIN (SELECT number AS n FROM numbers(5)) AS r ON l.k1 = r.n +ORDER BY ALL +FORMAT TSVWithNames; + +-- ANY, SEMI and ANTI mirror into their LEFT counterparts, so they are rewritten as well. +-- The duplicate key on the left makes ANY and ALL return a different number of rows. +DROP TABLE IF EXISTS dup_left_local_05058; +DROP TABLE IF EXISTS dup_left_distributed_05058; + +CREATE TABLE dup_left_local_05058 (k1 UInt32, v1 String) +ENGINE = MergeTree +ORDER BY k1; + +CREATE TABLE dup_left_distributed_05058 AS dup_left_local_05058 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), dup_left_local_05058); + +INSERT INTO dup_left_local_05058 VALUES (1, 'a1'), (1, 'a2'), (2, 'b'), (4, 'd'); + +SELECT 'any_initiator'; +SELECT k2, v2 +FROM (SELECT * FROM dup_left_distributed_05058) AS l +RIGHT ANY JOIN right_local_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'any_global'; +SELECT k2, v2 +FROM dup_left_distributed_05058 AS l +GLOBAL RIGHT ANY JOIN right_local_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'semi_initiator'; +SELECT k2, v2 +FROM (SELECT * FROM dup_left_distributed_05058) AS l +RIGHT SEMI JOIN right_local_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'semi_global'; +SELECT k2, v2 +FROM dup_left_distributed_05058 AS l +GLOBAL RIGHT SEMI JOIN right_local_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'anti_initiator'; +SELECT k2, v2 +FROM (SELECT * FROM dup_left_distributed_05058) AS l +RIGHT ANTI JOIN right_local_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'anti_global'; +SELECT k2, v2 +FROM dup_left_distributed_05058 AS l +GLOBAL RIGHT ANTI JOIN right_local_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +-- A single shard on the left never duplicates anything, so nothing is rewritten there. +DROP TABLE IF EXISTS left_one_shard_05058; + +CREATE TABLE left_one_shard_05058 AS left_local_05058 +ENGINE = Distributed('test_shard_localhost', currentDatabase(), left_local_05058); + +SELECT 'left_is_one_shard_initiator'; +SELECT k1, v1, k2, v2 +FROM (SELECT * FROM left_one_shard_05058) AS l +RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'left_is_one_shard_global'; +SELECT k1, v1, k2, v2 +FROM left_one_shard_05058 AS l +GLOBAL RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +DROP TABLE left_one_shard_05058; + +-- A RIGHT join anywhere in the tree emits the preserved rows once per shard, not only at the root. +-- The third table is a superset of the keys, otherwise the outer join filters out exactly the +-- duplicated rows and the test passes while the bug is still there. +DROP TABLE IF EXISTS third_local_05058; +DROP TABLE IF EXISTS third_distributed_05058; + +CREATE TABLE third_local_05058 (k3 UInt32, v3 String) +ENGINE = MergeTree +ORDER BY k3; + +CREATE TABLE third_distributed_05058 AS third_local_05058 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), third_local_05058); + +INSERT INTO third_local_05058 VALUES (1, 'X'), (2, 'Y'), (3, 'Z'), (4, 'W'); + +SELECT 'nested_right_then_inner_initiator'; +SELECT k1, v1, k2, v2, k3, v3 +FROM (SELECT * FROM left_distributed_05058) AS l +RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +INNER JOIN third_distributed_05058 AS t ON r.k2 = t.k3 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'nested_right_then_inner_global'; +SELECT k1, v1, k2, v2, k3, v3 +FROM left_distributed_05058 AS l +GLOBAL RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +GLOBAL INNER JOIN third_distributed_05058 AS t ON r.k2 = t.k3 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'nested_right_then_right_initiator'; +SELECT k1, v1, k2, v2, k3, v3 +FROM (SELECT * FROM left_distributed_05058) AS l +RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +RIGHT JOIN third_distributed_05058 AS t ON r.k2 = t.k3 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'nested_right_then_right_global'; +SELECT k1, v1, k2, v2, k3, v3 +FROM left_distributed_05058 AS l +GLOBAL RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +GLOBAL RIGHT JOIN third_distributed_05058 AS t ON r.k2 = t.k3 +ORDER BY ALL +FORMAT TSVWithNames; + +-- Here the RIGHT join sits at the root and its left side is another join, so the sides cannot be +-- swapped. The join runs on the initiator instead. +SELECT 'nested_inner_then_right_initiator'; +SELECT k1, v1, k2, v2, k3, v3 +FROM (SELECT * FROM left_distributed_05058) AS l +INNER JOIN third_distributed_05058 AS t ON l.k1 = t.k3 +RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'nested_inner_then_right_global'; +SELECT k1, v1, k2, v2, k3, v3 +FROM left_distributed_05058 AS l +GLOBAL INNER JOIN third_distributed_05058 AS t ON l.k1 = t.k3 +GLOBAL RIGHT JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +-- FULL JOIN preserves the broadcast side too, and cannot be repaired by swapping. +SELECT 'full_join_initiator'; +SELECT k1, v1, k2, v2 +FROM (SELECT * FROM left_distributed_05058) AS l +FULL JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +SELECT 'full_join_global'; +SELECT k1, v1, k2, v2 +FROM left_distributed_05058 AS l +GLOBAL FULL JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +-- Trees without a RIGHT or FULL join keep running on the shards, so shardNum still varies there. +SELECT 'shard_num_is_untouched_without_right_join'; +SELECT DISTINCT shardNum() AS s +FROM left_distributed_05058 AS l +GLOBAL INNER JOIN right_distributed_05058 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +DROP TABLE third_distributed_05058; +DROP TABLE third_local_05058; + +DROP TABLE dup_left_distributed_05058; +DROP TABLE dup_left_local_05058; +DROP TABLE right_one_shard_05058; + +DROP TABLE shared_left_distributed_05058; +DROP TABLE shared_right_distributed_05058; +DROP TABLE shared_left_local_05058; +DROP TABLE shared_right_local_05058; + DROP TABLE left_distributed_05058; DROP TABLE right_distributed_05058; DROP TABLE left_local_05058; From be7d6edbeef1decbad24cea812982cf048dbdb89 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 9 Sep 2026 13:09:29 +0200 Subject: [PATCH 5/5] Look through wrapper storages when rewriting a global RIGHT JOIN of Distributed tables. An `Alias`, `Buffer`, `MaterializedView` or `StorageProxy` in front of a sharded `Distributed` table still fans the query out across shards, so unmatched right rows would otherwise be duplicated once per shard. Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 35 ++++++++++--- src/Storages/StorageBuffer.h | 3 +- ...tributed_global_right_join_alias.reference | 24 +++++++++ ...38_distributed_global_right_join_alias.sql | 52 +++++++++++++++++++ ...ributed_global_right_join_buffer.reference | 24 +++++++++ ...9_distributed_global_right_join_buffer.sql | 52 +++++++++++++++++++ 6 files changed, 182 insertions(+), 8 deletions(-) create mode 100644 tests/queries/0_stateless/05138_distributed_global_right_join_alias.reference create mode 100644 tests/queries/0_stateless/05138_distributed_global_right_join_alias.sql create mode 100644 tests/queries/0_stateless/05139_distributed_global_right_join_buffer.reference create mode 100644 tests/queries/0_stateless/05139_distributed_global_right_join_buffer.sql diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 46e71b6cc3c3..031e3fd7a6ca 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -25,11 +25,14 @@ #include #include #include +#include +#include #include #include #include #include #include +#include #include #include #include @@ -161,13 +164,33 @@ namespace const StorageDistributed * getDistributedStorageFromTableExpression(const QueryTreeNodePtr & table_expression) { - const auto * table_node = table_expression->as(); - if (table_node) - return typeid_cast(table_node->getStorage().get()); + StoragePtr storage; + if (const auto * table_node = table_expression->as()) + storage = table_node->getStorage(); + else if (const auto * table_function_node = table_expression->as()) + storage = table_function_node->getStorage(); + else + return nullptr; - const auto * table_function_node = table_expression->as(); - if (table_function_node) - return typeid_cast(table_function_node->getStorage().get()); + /// `Alias`, `MaterializedView`, `Buffer` and `StorageProxy` (for example `lazy_load_tables`) + /// forward `read` to a nested storage. If that nested storage is `Distributed`, the join still + /// fans out across shards, so look through the wrappers before deciding. + for (size_t i = 0; storage && i < 16; ++i) + { + if (const auto * distributed = typeid_cast(storage.get())) + return distributed; + + if (const auto * proxy = dynamic_cast(storage.get())) + storage = proxy->getNested(); + else if (const auto * alias = storage->as()) + storage = alias->tryGetTargetTable(); + else if (const auto * materialized_view = storage->as()) + storage = materialized_view->tryGetTargetTable(); + else if (const auto * buffer = storage->as()) + storage = buffer->getDestinationTable(); + else + break; + } return nullptr; } diff --git a/src/Storages/StorageBuffer.h b/src/Storages/StorageBuffer.h index aad37e3badfd..b1120111ca9e 100644 --- a/src/Storages/StorageBuffer.h +++ b/src/Storages/StorageBuffer.h @@ -88,6 +88,7 @@ friend class BufferSink; size_t max_block_size, size_t num_streams) override; bool isRemote() const override; + StoragePtr getDestinationTable() const; bool supportsParallelInsert() const override { return true; } @@ -200,8 +201,6 @@ friend class BufferSink; void backgroundFlush(); void reschedule(size_t min_delay); - StoragePtr getDestinationTable() const; - BackgroundSchedulePool & bg_pool; BackgroundSchedulePoolTaskHolder flush_handle; diff --git a/tests/queries/0_stateless/05138_distributed_global_right_join_alias.reference b/tests/queries/0_stateless/05138_distributed_global_right_join_alias.reference new file mode 100644 index 000000000000..cd2e4ce4bb53 --- /dev/null +++ b/tests/queries/0_stateless/05138_distributed_global_right_join_alias.reference @@ -0,0 +1,24 @@ +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 +alias_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/05138_distributed_global_right_join_alias.sql b/tests/queries/0_stateless/05138_distributed_global_right_join_alias.sql new file mode 100644 index 000000000000..ff92cdca3a16 --- /dev/null +++ b/tests/queries/0_stateless/05138_distributed_global_right_join_alias.sql @@ -0,0 +1,52 @@ +-- Tags: distributed + +SET enable_analyzer = 1; +SET join_use_nulls = 1; + +DROP TABLE IF EXISTS left_local_05138; +DROP TABLE IF EXISTS right_local_05138; +DROP TABLE IF EXISTS left_distributed_05138; +DROP TABLE IF EXISTS right_distributed_05138; +DROP TABLE IF EXISTS left_alias_05138; + +CREATE TABLE left_local_05138 (k1 UInt32, v1 String) +ENGINE = MergeTree +ORDER BY k1; + +CREATE TABLE right_local_05138 (k2 UInt32, v2 String) +ENGINE = MergeTree +ORDER BY k2; + +CREATE TABLE left_distributed_05138 AS left_local_05138 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), left_local_05138); + +CREATE TABLE right_distributed_05138 AS right_local_05138 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), right_local_05138); + +CREATE TABLE left_alias_05138 +ENGINE = Alias('left_distributed_05138'); + +INSERT INTO left_local_05138 VALUES (1, 'a'), (2, 'b'), (4, 'd'); +INSERT INTO right_local_05138 VALUES (1, 'A'), (2, 'B'), (3, 'C'); + +SELECT 'initiator'; +SELECT * +FROM (SELECT * FROM left_alias_05138) AS l +RIGHT JOIN (SELECT * FROM right_distributed_05138) AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +-- An `Alias` of a sharded `Distributed` table still fans the query out across shards, +-- so the rewrite must look through it the same way as a bare `Distributed` table. +SELECT 'alias_global'; +SELECT * +FROM left_alias_05138 AS l +GLOBAL RIGHT JOIN right_distributed_05138 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +DROP TABLE left_alias_05138; +DROP TABLE left_distributed_05138; +DROP TABLE right_distributed_05138; +DROP TABLE left_local_05138; +DROP TABLE right_local_05138; diff --git a/tests/queries/0_stateless/05139_distributed_global_right_join_buffer.reference b/tests/queries/0_stateless/05139_distributed_global_right_join_buffer.reference new file mode 100644 index 000000000000..514e988378d5 --- /dev/null +++ b/tests/queries/0_stateless/05139_distributed_global_right_join_buffer.reference @@ -0,0 +1,24 @@ +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 +buffer_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/05139_distributed_global_right_join_buffer.sql b/tests/queries/0_stateless/05139_distributed_global_right_join_buffer.sql new file mode 100644 index 000000000000..ccdc9a895b3a --- /dev/null +++ b/tests/queries/0_stateless/05139_distributed_global_right_join_buffer.sql @@ -0,0 +1,52 @@ +-- Tags: distributed + +SET enable_analyzer = 1; +SET join_use_nulls = 1; + +DROP TABLE IF EXISTS left_local_05139; +DROP TABLE IF EXISTS right_local_05139; +DROP TABLE IF EXISTS left_distributed_05139; +DROP TABLE IF EXISTS right_distributed_05139; +DROP TABLE IF EXISTS left_buffer_05139; + +CREATE TABLE left_local_05139 (k1 UInt32, v1 String) +ENGINE = MergeTree +ORDER BY k1; + +CREATE TABLE right_local_05139 (k2 UInt32, v2 String) +ENGINE = MergeTree +ORDER BY k2; + +CREATE TABLE left_distributed_05139 AS left_local_05139 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), left_local_05139); + +CREATE TABLE right_distributed_05139 AS right_local_05139 +ENGINE = Distributed('test_cluster_two_shards_localhost', currentDatabase(), right_local_05139); + +CREATE TABLE left_buffer_05139 AS left_local_05139 +ENGINE = Buffer(currentDatabase(), left_distributed_05139, 1, 1, 1, 1, 1, 1, 1); + +INSERT INTO left_local_05139 VALUES (1, 'a'), (2, 'b'), (4, 'd'); +INSERT INTO right_local_05139 VALUES (1, 'A'), (2, 'B'), (3, 'C'); + +SELECT 'initiator'; +SELECT * +FROM (SELECT * FROM left_buffer_05139) AS l +RIGHT JOIN (SELECT * FROM right_distributed_05139) AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +-- A `Buffer` whose destination is a sharded `Distributed` table still fans the query out +-- across shards, so the rewrite must look through it the same way as a bare `Distributed` table. +SELECT 'buffer_global'; +SELECT * +FROM left_buffer_05139 AS l +GLOBAL RIGHT JOIN right_distributed_05139 AS r ON l.k1 = r.k2 +ORDER BY ALL +FORMAT TSVWithNames; + +DROP TABLE left_buffer_05139; +DROP TABLE left_distributed_05139; +DROP TABLE right_distributed_05139; +DROP TABLE left_local_05139; +DROP TABLE right_local_05139;