diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index dd4ef0a462a1..5d42f9a69662 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; @@ -116,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; @@ -129,6 +131,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 +158,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 +2020,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; @@ -1996,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; @@ -2024,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; @@ -2094,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"] 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;