diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 7d1eb58af169..6c80af15b999 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 @@ -37,6 +40,7 @@ #include #include #include +#include #include #include #include @@ -105,6 +109,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; @@ -117,6 +122,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; @@ -130,6 +136,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; @@ -156,6 +163,127 @@ namespace ErrorCodes namespace { +const StorageDistributed * getDistributedStorageFromTableExpression(const QueryTreeNodePtr & table_expression) +{ + 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; + + /// `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; +} + +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(); + 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; + + /** 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()); + if (!left_storage || left_storage->getShardCount() < 2) + return; + + /** 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); +} + /// 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 @@ -2130,7 +2258,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; @@ -2150,11 +2281,13 @@ 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; 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) { @@ -2178,10 +2311,19 @@ 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; + /// 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); @@ -2248,21 +2390,33 @@ 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 - * NOT be wrapped (wrapping breaks tests like 03577_server_constant_folding). + * 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), except for the + * join trees that stay wrong when the join runs per shard, handled below. */ 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(); @@ -2274,6 +2428,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/src/Storages/StorageBuffer.h b/src/Storages/StorageBuffer.h index 4685aa736cd5..30322f03e2ff 100644 --- a/src/Storages/StorageBuffer.h +++ b/src/Storages/StorageBuffer.h @@ -88,6 +88,9 @@ friend class BufferSink; size_t max_block_size, size_t num_streams) override; bool isRemote() const override; + + StoragePtr getDestinationTable() const; + bool readsFromOtherTables() const override { return static_cast(destination_id); } bool supportsParallelInsert() const override { return true; } @@ -201,8 +204,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/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..2e645c47b92b --- /dev/null +++ b/tests/queries/0_stateless/05058_distributed_global_right_join.reference @@ -0,0 +1,420 @@ +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 +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 new file mode 100644 index 000000000000..620d9931cbc8 --- /dev/null +++ b/tests/queries/0_stateless/05058_distributed_global_right_join.sql @@ -0,0 +1,386 @@ +-- 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; + +-- 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; +DROP TABLE right_local_05058; 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;