diff --git a/docs/en/sql-reference/distribution-on-cluster.md b/docs/en/sql-reference/distribution-on-cluster.md index c1b9a4bdcfaae..3e7151839959a 100644 --- a/docs/en/sql-reference/distribution-on-cluster.md +++ b/docs/en/sql-reference/distribution-on-cluster.md @@ -26,6 +26,6 @@ Receiving new tasks can be resumed with the command `SYSTEM START SWARM MODE`. ## Joins {#joins} -[`object_storage_cluster_join_mode`](/operations/settings/settings#object_storage_cluster_join_mode) chooses where a `JOIN` of a `*Cluster` table function, or of a table with `object_storage_cluster` set, is executed, and which table is sent to the nodes. With `allow` and `local` the nodes read the left table and the join runs on the initiator. With `global`, an `INNER`, `LEFT`, or `CROSS` join runs on the nodes and the initiator sends the right table (every table after the first, for `CROSS JOIN`). A `RIGHT` or `FULL` join stays on the initiator. +[`object_storage_cluster_join_mode`](/operations/settings/settings#object_storage_cluster_join_mode) chooses where a `JOIN` of a `*Cluster` table function, or of a table with `object_storage_cluster` set, is executed, and which table is sent to the nodes. With `allow` and `local` the nodes read the left table and the join runs on the initiator. With `global`, an `INNER`, `LEFT`, or `CROSS` join runs on the nodes and the initiator sends the right table (every table after the first, for `CROSS JOIN`). A `RIGHT`, `FULL`, or `PASTE` join stays on the initiator. The same rules for a `Distributed` table, including which side of a `GLOBAL` `RIGHT` join is sent, are described in [Distributed JOIN](/sql-reference/statements/select/join.md#distributed-join). diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index c3f0a39e28740..6ec28e0250177 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -2131,11 +2131,16 @@ Possible values: DECLARE(ObjectStorageClusterJoinMode, object_storage_cluster_join_mode, ObjectStorageClusterJoinMode::ALLOW, R"( Where a [JOIN](../../sql-reference/statements/select/join.md#distributed-join) runs when its left table is a cluster table function (`s3Cluster`, `icebergCluster`, and the other `*Cluster` functions) or a table engine with [`object_storage_cluster`](#object_storage_cluster) set. +Restrictions: + +- Only applied for `JOIN` and for `IN`/`WHERE` subqueries that reference other tables. +- Only if the FROM section uses an object storage cluster function or table. + Possible values: -- `allow` — Default. Nodes read the left table and return its rows. The join runs on the initiator. The right table is read there and is not sent to the nodes. -- `local` — The join runs on the initiator. Nodes read the left table and return its rows. The right table is read on the initiator and is not sent to the nodes. -- `global` — `INNER`, `LEFT`, and `CROSS` run on the nodes. The initiator reads the right table once and sends it to each node. For `CROSS JOIN`, every table after the first is sent. `RIGHT` and `FULL` stay on the initiator, so unmatched rows of the preserved side are returned once. +- `global` — Replaces the `IN`/`JOIN` query with `GLOBAL IN`/`GLOBAL JOIN`. Right table executes first and is added to the secondary query as a temporary table. +- `allow` — Default value. Reads the left object-storage table on cluster nodes and executes `JOIN` / local `IN` on the initiator, so the right table does not need to exist on remote nodes. +- `local` — Deprecated legacy alias of `allow`. See [Distributed JOIN](../../sql-reference/statements/select/join.md#object-storage-cluster-join-mode). )", 0) \ diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h index a17f433ba6ecc..19fec342d5a1f 100644 --- a/src/Core/SettingsEnums.h +++ b/src/Core/SettingsEnums.h @@ -168,9 +168,9 @@ DECLARE_SETTING_ENUM(DistributedProductMode) /// The setting for executing object storage cluster function or table JOIN sections. enum class ObjectStorageClusterJoinMode : uint8_t { - LOCAL, /// Convert to local query + LOCAL, /// Legacy alias of ALLOW: initiator-local join GLOBAL, /// Convert to global query - ALLOW /// Enable + ALLOW /// Initiator-local join when the right table is not on remote nodes }; DECLARE_SETTING_ENUM(ObjectStorageClusterJoinMode) diff --git a/src/Interpreters/InterpreterSelectQuery.cpp b/src/Interpreters/InterpreterSelectQuery.cpp index 913306be42e3a..c8557fb2d6705 100644 --- a/src/Interpreters/InterpreterSelectQuery.cpp +++ b/src/Interpreters/InterpreterSelectQuery.cpp @@ -216,6 +216,8 @@ namespace Setting extern const SettingsBool enable_lazy_columns_replication; extern const SettingsBool serialize_string_in_memory_with_zero_byte; extern const SettingsBool use_hive_partitioning; + extern const SettingsBool allow_experimental_analyzer; + extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; } namespace ServerSetting @@ -788,6 +790,13 @@ InterpreterSelectQuery::InterpreterSelectQuery( joined_tables.rewriteDistributedInAndJoins(query_ptr); + if (storage && !settings[Setting::allow_experimental_analyzer] + && settings[Setting::object_storage_cluster_join_mode] == ObjectStorageClusterJoinMode::GLOBAL + && dynamic_cast(storage.get())) + { + IStorageCluster::rewriteASTForGlobalJoin(query_ptr); + } + max_streams = getMaxThreadsForAvailableMemory( settings[Setting::max_threads], settings[Setting::max_threads_min_free_memory_per_thread]); ASTSelectQuery & query = getSelectQuery(); diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 1633d044ab470..d205cf7553f7b 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -2289,6 +2289,7 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, size_t joins_count = 0; bool is_full_join = false; bool is_right_join = false; + bool is_paste_join = false; bool is_global_join = false; bool is_right_join_with_remote_table = false; int first_join_pos = -1; @@ -2320,6 +2321,8 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, is_full_join = true; if (join_kind == JoinKind::Right) is_right_join = true; + if (join_kind == JoinKind::Paste) + is_paste_join = true; if (join_node.getLocality() == JoinLocality::Global) is_global_join = true; @@ -2408,7 +2411,8 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, * 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. + * preserved right side would make every shard emit the same unmatched rows. `PASTE JOIN` stays + * there too: each shard would pair its own rows with the broadcast right side. * * `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 @@ -2421,7 +2425,8 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, == ObjectStorageClusterJoinMode::GLOBAL && joins_count > 0 && !is_right_join - && !is_full_join; + && !is_full_join + && !is_paste_join; if (has_multiple_tables && !use_global_join) { diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 56eea64cd5217..deb8fb69ab6ae 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -37,14 +37,20 @@ #include #include #include -#include -#include #include #include -#include -#include #include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include @@ -175,49 +181,6 @@ void ReadFromCluster::createExtension() namespace { -/* -Helping class to find in query tree first node of required type -*/ -class SearcherVisitor : public InDepthQueryTreeVisitorWithContext -{ -public: - using Base = InDepthQueryTreeVisitorWithContext; - using Base::Base; - - explicit SearcherVisitor(std::unordered_set types_, size_t entry_, ContextPtr context) - : Base(context) - , types(types_) - , entry(entry_) {} - - bool needChildVisit(QueryTreeNodePtr & /*parent*/, QueryTreeNodePtr & /*child*/) - { - return getSubqueryDepth() <= 2 && !passed_node && !current_entry; - } - - void enterImpl(QueryTreeNodePtr & node) - { - if (passed_node) - return; - - auto node_type = node->getNodeType(); - - if (types.contains(node_type)) - { - ++current_entry; - if (current_entry == entry) - passed_node = node; - } - } - - QueryTreeNodePtr getNode() const { return passed_node; } - -private: - std::unordered_set types; - size_t entry; - size_t current_entry = 0; - QueryTreeNodePtr passed_node; -}; - /* Helping class to find all used columns with specific source */ @@ -240,6 +203,12 @@ class CollectUsedColumnsForSourceVisitor : public InDepthQueryTreeVisitorWithCon { auto node_type = node->getNodeType(); + /// `SELECT 1 FROM dim` names a table and projects no column. Column collection misses it. + if (collect_columns_from_other_sources + && node != source + && (node_type == QueryTreeNodeType::TABLE || node_type == QueryTreeNodeType::TABLE_FUNCTION)) + has_foreign_table = true; + if (node_type != QueryTreeNodeType::COLUMN) return; @@ -260,15 +229,185 @@ class CollectUsedColumnsForSourceVisitor : public InDepthQueryTreeVisitorWithCon } const NamesAndTypes & getColumns() const { return columns; } + bool hasForeignTable() const { return has_foreign_table; } private: std::unordered_set names; QueryTreeNodePtr source; NamesAndTypes columns; bool collect_columns_from_other_sources; + bool has_foreign_table = false; }; -}; +bool astContainsSubquery(const ASTPtr & node) +{ + if (!node) + return false; + + if (node->as() || node->as() || node->as()) + return true; + + for (const auto & child : node->children) + { + if (astContainsSubquery(child)) + return true; + } + + return false; +} + +bool astContainsInTableIdentifier(const ASTPtr & node) +{ + if (!node) + return false; + + if (const auto * function = node->as()) + { + if (isNameOfInFunction(function->name) && function->arguments && function->arguments->children.size() >= 2) + { + const auto & rhs = function->arguments->children[1]; + /// GLOBAL IN is rewritten to an external table (`_subqueryN`) as `ASTTableIdentifier`. + /// `as` is an exact typeid match and does not see that subclass. + if (rhs && (rhs->as() || rhs->as())) + return true; + } + } + + for (const auto & child : node->children) + { + if (astContainsInTableIdentifier(child)) + return true; + } + + return false; +} + +bool astIsNestedSelect(const ASTPtr & node) +{ + return node && (node->as() || node->as() || node->as()); +} + +void stripObjectStorageRemoteInitiatorSettings(ASTPtr & query) +{ + auto * select_query = query ? query->as() : nullptr; + if (!select_query) + return; + + auto query_settings = select_query->settings(); + if (!query_settings) + return; + + auto & settings_ast = query_settings->as(); + bool settings_changed = false; + for (const auto * setting_name : {"object_storage_remote_initiator", "object_storage_remote_initiator_cluster"}) + settings_changed |= settings_ast.changes.removeSetting(setting_name); + + if (settings_changed && settings_ast.changes.empty()) + select_query->setExpression(ASTSelectQuery::Expression::SETTINGS, {}); +} + +void rewriteASTInFunctionsToGlobalIn(ASTPtr & node) +{ + if (!node) + return; + + if (auto * function = node->as()) + { + if (isNameOfLocalInFunction(function->name) && function->arguments && function->arguments->children.size() >= 2) + { + const auto & rhs = function->arguments->children[1]; + if (rhs + && (rhs->as() || rhs->as() || rhs->as() + || rhs->as() || rhs->as())) + { + function->name = getGlobalInFunctionNameForLocalInFunctionName(function->name); + } + } + } + + for (auto & child : node->children) + { + if (astIsNestedSelect(child)) + continue; + rewriteASTInFunctionsToGlobalIn(child); + } +} + +bool queryHasRightOrFullJoin(const ASTPtr & query) +{ + if (!query) + return false; + + if (const auto * select_query = query->as()) + { + const auto tables = select_query->tables(); + if (!tables) + return false; + + for (const auto & child : tables->children) + { + const auto & element = child->as(); + if (element.table_join) + { + const auto kind = element.table_join->as().kind; + /// `PASTE` pairs by position, so a per-shard join would emit one pair per shard. + if (isRightOrFull(kind) || isPaste(kind)) + return true; + } + } + return false; + } + + const auto * union_query = query->as(); + if (!union_query || !union_query->list_of_selects) + return false; + + for (const auto & child : union_query->list_of_selects->children) + if (queryHasRightOrFullJoin(child)) + return true; + return false; +} + +void rewriteASTJoinsToGlobal(ASTPtr & query) +{ + ASTSelectQuery * select_query = query->as(); + if (!select_query) + { + if (auto * union_query = query->as()) + { + if (union_query->list_of_selects) + { + for (auto & child : union_query->list_of_selects->children) + rewriteASTJoinsToGlobal(child); + } + } + return; + } + + /// `RIGHT`/`FULL` preserve broadcast rows, and `PASTE` pairs by position, so the join stays on the initiator. + if (queryHasRightOrFullJoin(query)) + return; + + if (auto tables = select_query->tables()) + { + auto & tables_in_select_query = tables->as(); + for (auto & child : tables_in_select_query.children) + { + auto & tables_element = child->as(); + if (tables_element.table_join) + tables_element.table_join->as().locality = JoinLocality::Global; + } + } + + rewriteASTInFunctionsToGlobalIn(query); +} + +} + +void IStorageCluster::rewriteASTForGlobalJoin(ASTPtr & query) +{ + rewriteASTJoinsToGlobal(query); +} /* Try to make subquery to send on nodes @@ -281,7 +420,7 @@ Converts localtable as t ON s3.key == t.key -to (object_storage_cluster_join_mode='local') +to (object_storage_cluster_join_mode='allow' or 'local') SELECT s3.c1, s3.c2, s3.key FROM @@ -304,99 +443,32 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( auto object_storage_cluster_join_mode = context->getSettingsRef()[Setting::object_storage_cluster_join_mode]; switch (object_storage_cluster_join_mode) { - case ObjectStorageClusterJoinMode::LOCAL: + case ObjectStorageClusterJoinMode::LOCAL: /// Legacy alias of `allow` + case ObjectStorageClusterJoinMode::ALLOW: { - if (!context->getSettingsRef()[Setting::allow_experimental_analyzer]) - throw Exception(ErrorCodes::NOT_IMPLEMENTED, - "object_storage_cluster_join_mode!='allow' is not supported without allow_experimental_analyzer=true"); - - auto info = getQueryTreeInfo(query_info.query_tree, context); + auto info = getQueryJoinInfo(query_info, context); + if (!needsInitiatorLocalJoin(info)) + return; - if (info.has_join || info.has_cross_join || info.has_local_columns_in_where) + rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context); + return; + } + case ObjectStorageClusterJoinMode::GLOBAL: + { + /// Old analyzer: ship the left scan only. `INNER`/`LEFT`/`CROSS` stay in `query_to_send`. + if (!query_info.query_tree) { - auto modified_query_tree = query_info.query_tree->clone(); - - SearcherVisitor left_table_expression_searcher({QueryTreeNodeType::TABLE, QueryTreeNodeType::TABLE_FUNCTION}, 1, context); - left_table_expression_searcher.visit(modified_query_tree); - auto table_function_node = left_table_expression_searcher.getNode(); - if (!table_function_node) - throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find left table function node"); - - QueryTreeNodePtr query_tree_distributed; - - auto & query_node = modified_query_tree->as(); - - if (info.has_join) + if (queryHasRightOrFullJoin(query_to_send)) { - auto join_node = query_node.getJoinTree(); - query_tree_distributed = join_node->as()->getLeftTableExpression()->clone(); + auto info = getQueryJoinInfoFromAST(query_to_send); + rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context); } - else if (info.has_cross_join) - { - SearcherVisitor join_searcher({QueryTreeNodeType::CROSS_JOIN}, 1, context); - join_searcher.visit(modified_query_tree); - auto cross_join_node = join_searcher.getNode(); - if (!cross_join_node) - throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find CROSS JOIN node"); - // CrossJoinNode contains vector of nodes. 0 is left expression, always exists. - query_tree_distributed = cross_join_node->as()->getTableExpressions()[0]->clone(); - } - - // Find add used columns from table function to make proper projection list - // Need to do before changing WHERE condition - CollectUsedColumnsForSourceVisitor collector(table_function_node, context); - collector.visit(modified_query_tree); - const auto & columns = collector.getColumns(); - - if (columns.empty()) - { - auto column_nodes_to_select = std::make_shared(); - column_nodes_to_select->getNodes().reserve(1); - column_nodes_to_select->getNodes().emplace_back(std::make_shared(1)); - query_node.getProjectionNode() = column_nodes_to_select; - } - else - { - query_node.resolveProjectionColumns(columns); - auto column_nodes_to_select = std::make_shared(); - column_nodes_to_select->getNodes().reserve(columns.size()); - for (auto & column : columns) - column_nodes_to_select->getNodes().emplace_back(std::make_shared(column, table_function_node)); - query_node.getProjectionNode() = column_nodes_to_select; - } - - if (info.has_local_columns_in_where) - { - if (query_node.getPrewhere()) - removeExpressionsThatDoNotDependOnTableIdentifiers(query_node.getPrewhere(), table_function_node, context); - if (query_node.getWhere()) - removeExpressionsThatDoNotDependOnTableIdentifiers(query_node.getWhere(), table_function_node, context); - } - - if (query_node.getPrewhere()) - removeExpressionsThatAreUnsafeToDuplicate(query_node.getPrewhere(), context); - if (query_node.getWhere()) - removeExpressionsThatAreUnsafeToDuplicate(query_node.getWhere(), context); - - query_node.getOrderByNode() = std::make_shared(); - query_node.getGroupByNode() = std::make_shared(); - - if (query_tree_distributed) - { - // Left only table function to send on cluster nodes - modified_query_tree = modified_query_tree->cloneAndReplace(query_node.getJoinTree(), query_tree_distributed); - } - - query_to_send = queryNodeToDistributedSelectQuery(modified_query_tree); + return; } - return; - } - case ObjectStorageClusterJoinMode::GLOBAL: - { auto info = getQueryTreeInfo(query_info.query_tree, context); - if (info.has_join || info.has_cross_join || info.has_local_columns_in_where) + if (needsInitiatorLocalJoin(info)) { auto modified_query_tree = query_info.query_tree->clone(); @@ -415,8 +487,6 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( return; } - case ObjectStorageClusterJoinMode::ALLOW: // Do nothing special - return; } } @@ -529,10 +599,44 @@ void IStorageCluster::readFromRemoteInitiator( ClusterPtr remote_initiator_cluster, const String & remote_initiator_cluster_name) { + /// Clone the stripped query before `convertToRemote` wraps it. `StorageDistributed` must not + /// analyze the original `query_info`, which still contains the JOIN. + const bool query_was_rewritten = query_to_send.get() != query_info.query.get(); + ASTPtr query_for_remote = query_was_rewritten ? query_to_send->clone() : ASTPtr{}; + if (query_for_remote) + stripObjectStorageRemoteInitiatorSettings(query_for_remote); + auto storage_and_context = convertToRemote(remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); auto modified_query_info = query_info; modified_query_info.cluster = src_distributed->getCluster(); + + if (query_for_remote) + { + modified_query_info.query = query_for_remote; + modified_query_info.syntax_analyzer_result.reset(); + + if (storage_and_context.context->getSettingsRef()[Setting::allow_experimental_analyzer]) + { + InterpreterSelectQueryAnalyzer interpreter( + query_for_remote, + storage_and_context.context, + SelectQueryOptions(processed_stage).analyze()); + modified_query_info.query_tree = interpreter.getQueryTree(); + modified_query_info.planner_context = interpreter.getPlanner().getPlannerContext(); + + auto * query_node = modified_query_info.query_tree->as(); + if (!query_node || !query_node->getJoinTree()) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected a table scan after stripping the initiator-local JOIN"); + + modified_query_info.table_expression = extractLeftTableExpression(query_node->getJoinTree()); + } + else + { + modified_query_info.query_tree.reset(); + } + } + auto new_storage_snapshot = storage_and_context.storage->getStorageSnapshot(storage_snapshot->metadata, storage_and_context.context); storage_and_context.storage->read( query_plan, column_names, new_storage_snapshot, modified_query_info, storage_and_context.context, processed_stage, max_block_size, num_streams); @@ -677,9 +781,11 @@ void IStorageCluster::read( auto this_ptr = std::static_pointer_cast(shared_from_this()); - std::optional external_tables = std::nullopt; + std::optional external_tables; if (query_info.planner_context && query_info.planner_context->getMutableQueryContext()) external_tables = query_info.planner_context->getMutableQueryContext()->getExternalTables(); + if (!external_tables || external_tables->empty()) + external_tables = context->getExternalTables(); auto reading = std::make_unique( column_names, @@ -734,16 +840,7 @@ IStorageCluster::RemoteCallVariables IStorageCluster::convertToRemote( if (!select_query) throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected SELECT query"); - auto query_settings = select_query->settings(); - if (query_settings) - { - auto & settings_ast = query_settings->as(); - bool settings_changed = false; - for (const auto & setting_to_remove : settings_to_remove) - settings_changed |= settings_ast.changes.removeSetting(setting_to_remove); - if (settings_changed && settings_ast.changes.empty()) - select_query->setExpression(ASTSelectQuery::Expression::SETTINGS, {}); - } + stripObjectStorageRemoteInitiatorSettings(query_to_send); ASTTableExpression * table_expression = extractTableExpressionASTPtrFromSelectQuery(query_to_send); if (!table_expression) @@ -874,23 +971,22 @@ IStorageCluster::QueryTreeInfo IStorageCluster::getQueryTreeInfo(QueryTreeNodePt QueryTreeInfo info; auto & query_node = query_tree->as(); - if (auto join_node = query_node.getJoinTree()) - { - if (join_node->getNodeType() == QueryTreeNodeType::JOIN) - info.has_join = true; - else if (join_node->getNodeType() == QueryTreeNodeType::CROSS_JOIN) - info.has_cross_join = true; - } + auto join_tree = query_node.getJoinTree(); + if (!join_tree) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); + + if (join_tree->getNodeType() == QueryTreeNodeType::JOIN) + info.has_join = true; + else if (join_tree->getNodeType() == QueryTreeNodeType::CROSS_JOIN) + info.has_cross_join = true; - SearcherVisitor left_table_expression_searcher({QueryTreeNodeType::TABLE, QueryTreeNodeType::TABLE_FUNCTION}, 1, context); - left_table_expression_searcher.visit(query_tree); - auto table_function_node = left_table_expression_searcher.getNode(); - if (!table_function_node) + auto left_table_expression = extractLeftTableExpression(join_tree); + if (!left_table_expression) throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); if (query_node.hasWhere() || query_node.hasPrewhere()) { - CollectUsedColumnsForSourceVisitor collector_where(table_function_node, context, true); + CollectUsedColumnsForSourceVisitor collector_where(left_table_expression, context, true); if (query_node.hasPrewhere()) collector_where.visit(query_node.getPrewhere()); if (query_node.hasWhere()) @@ -899,30 +995,213 @@ IStorageCluster::QueryTreeInfo IStorageCluster::getQueryTreeInfo(QueryTreeNodePt // SELECT x FROM datalake.table WHERE x IN local.table. // Need to modify 'WHERE' on remote node if it contains columns from other sources // because remote node might not have those sources. - if (!collector_where.getColumns().empty()) + // `WHERE x IN (SELECT 1 FROM local_table)` has no column from that table. + if (!collector_where.getColumns().empty() || collector_where.hasForeignTable()) info.has_local_columns_in_where = true; } return info; } -QueryProcessingStage::Enum IStorageCluster::getQueryProcessingStage( - ContextPtr context, QueryProcessingStage::Enum to_stage, const StorageSnapshotPtr &, SelectQueryInfo & query_info) const +bool IStorageCluster::needsInitiatorLocalJoin(const QueryTreeInfo & info) { - auto object_storage_cluster_join_mode = context->getSettingsRef()[Setting::object_storage_cluster_join_mode]; + return info.has_join || info.has_cross_join || info.has_local_columns_in_where; +} - if (object_storage_cluster_join_mode != ObjectStorageClusterJoinMode::ALLOW) +IStorageCluster::QueryTreeInfo IStorageCluster::getQueryJoinInfoFromAST(const ASTPtr & query) +{ + QueryTreeInfo info; + if (!query) + return info; + + const ASTSelectQuery * select_query = query->as(); + if (!select_query) { - if (!context->getSettingsRef()[Setting::allow_experimental_analyzer]) - throw Exception(ErrorCodes::NOT_IMPLEMENTED, - "object_storage_cluster_join_mode!='allow' is not supported without allow_experimental_analyzer=true"); + if (const auto * union_query = query->as()) + { + if (union_query->list_of_selects && union_query->list_of_selects->children.size() == 1) + select_query = union_query->list_of_selects->children[0]->as(); + } + } + if (!select_query) + return info; - if (object_storage_cluster_join_mode == ObjectStorageClusterJoinMode::LOCAL) + if (select_query->hasJoin()) + info.has_join = true; + + if (astContainsSubquery(select_query->where()) || astContainsSubquery(select_query->prewhere()) + || astContainsInTableIdentifier(select_query->where()) || astContainsInTableIdentifier(select_query->prewhere())) + info.has_local_columns_in_where = true; + + return info; +} + +IStorageCluster::QueryTreeInfo IStorageCluster::getQueryJoinInfo(const SelectQueryInfo & query_info, const ContextPtr & context) +{ + if (query_info.query_tree && query_info.query_tree->as()) + return getQueryTreeInfo(query_info.query_tree, context); + + return getQueryJoinInfoFromAST(query_info.query); +} + +void IStorageCluster::rewriteQueryForInitiatorLocalJoin( + ASTPtr & query_to_send, + const SelectQueryInfo & query_info, + const QueryTreeInfo & info, + const ContextPtr & context) +{ + /// Analyzer path: reuse extractLeftTableExpression + buildQueryToReadColumnsFromTableExpression + /// (same helpers the planner uses when wrapping IStorageCluster in a subquery). + if (query_info.query_tree) + { + auto modified_query_tree = query_info.query_tree->clone(); + auto & query_node = modified_query_tree->as(); + auto join_tree = query_node.getJoinTree(); + if (!join_tree) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); + + auto left_table_expression = extractLeftTableExpression(join_tree); + if (!left_table_expression) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); + + CollectUsedColumnsForSourceVisitor collector(left_table_expression, context); + collector.visit(modified_query_tree); + + /// A left-only `WHERE` under `RIGHT`/`FULL` changes which right rows are unmatched. + /// `PREWHERE` filters the left scan before the join, so it stays on that scan. + if (queryHasRightOrFullJoin(query_to_send)) + query_node.getWhere() = {}; + + if (query_node.getPrewhere()) + removeExpressionsThatDoNotDependOnTableIdentifiers(query_node.getPrewhere(), left_table_expression, context); + if (query_node.getWhere()) + removeExpressionsThatDoNotDependOnTableIdentifiers(query_node.getWhere(), left_table_expression, context); + + /// The wrap predicate runs on remote nodes. Drop conjuncts such as `dictGet` and `hostName` + /// that must stay on the initiator. + if (query_node.getPrewhere()) + removeExpressionsThatAreUnsafeToDuplicate(query_node.getPrewhere(), context); + if (query_node.getWhere()) + removeExpressionsThatAreUnsafeToDuplicate(query_node.getWhere(), context); + + auto rewritten_query_tree = buildQueryToReadColumnsFromTableExpression( + collector.getColumns(), left_table_expression, context); + auto & rewritten_query_node = rewritten_query_tree->as(); + rewritten_query_node.getPrewhere() = query_node.getPrewhere(); + rewritten_query_node.getWhere() = query_node.getWhere(); + + query_to_send = queryNodeToDistributedSelectQuery(rewritten_query_tree); + return; + } + + /// Old interpreter: reuse removeJoin used by StorageDistributed / StorageMerge / StorageWindowView. + if (!query_to_send) + return; + + query_to_send = query_to_send->clone(); + if (auto * union_query = query_to_send->as()) + { + if (union_query->list_of_selects && union_query->list_of_selects->children.size() == 1) + query_to_send = union_query->list_of_selects->children[0]->clone(); + } + + auto * select_query = query_to_send->as(); + if (!select_query) + return; + + if (select_query->hasJoin()) + { + if (!query_info.syntax_analyzer_result) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Query is not analyzed: no syntax analyzer result"); + + TreeRewriterResult rewriter_result = *query_info.syntax_analyzer_result; + /// Read the kind before `removeJoin` drops it. Left-only `WHERE` under `RIGHT`/`FULL` changes unmatched rows. + const bool preserves_right_side = queryHasRightOrFullJoin(query_to_send); + /// `removeJoin` only looks at the second table element. An `ARRAY JOIN` there hides a later join. + if (auto tables = select_query->tables()) + { + auto & children = tables->children; + for (size_t i = 1; i < children.size();) + { + if (children[i]->as().array_join) + children.erase(children.begin() + i); + else + ++i; + } + } + if (!removeJoin(*select_query, rewriter_result, context)) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Failed to strip JOIN from query sent to cluster nodes"); + + /// `removeJoin` drops the ORDER BY and GROUP BY expressions but leaves `order_by_all` and + /// `group_by_all` set. The non-join branch below clears both flags. `IStorageCluster::read` + /// re-analyzes this AST with `TreeRewriter`, which still honors them: `ORDER BY ALL` + /// dereferences a null `orderBy`, and `GROUP BY ALL` aggregates the stripped left columns + /// on the shards before the initiator join. + select_query->group_by_all = false; + select_query->order_by_all = false; + + /// `removeJoin` keeps left-table predicates. A left `WHERE` under `RIGHT`/`FULL` changes unmatched rows. + /// A subquery cannot be resolved on the shards. `PREWHERE` stays: it filters the left scan before the join. + if (preserves_right_side || astContainsInTableIdentifier(select_query->where()) || astContainsSubquery(select_query->where())) + select_query->setExpression(ASTSelectQuery::Expression::WHERE, {}); + if (astContainsInTableIdentifier(select_query->prewhere()) || astContainsSubquery(select_query->prewhere())) + select_query->setExpression(ASTSelectQuery::Expression::PREWHERE, {}); + } + else if (info.has_local_columns_in_where) + { + if (!query_info.syntax_analyzer_result) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Query is not analyzed: no syntax analyzer result"); + + auto select_expression_list = make_intrusive(); + const auto & required_columns = query_info.syntax_analyzer_result->required_source_columns; + if (required_columns.empty()) + { + select_expression_list->children.push_back(make_intrusive(Field{static_cast(1)})); + } + else { - auto info = getQueryTreeInfo(query_info.query_tree, context); - if (info.has_join || info.has_cross_join || info.has_local_columns_in_where) - return QueryProcessingStage::Enum::FetchColumns; + select_expression_list->children.reserve(required_columns.size()); + for (const auto & column : required_columns) + select_expression_list->children.push_back(make_intrusive(column.name)); } + select_query->setExpression(ASTSelectQuery::Expression::SELECT, std::move(select_expression_list)); + select_query->setExpression(ASTSelectQuery::Expression::PREWHERE, {}); + select_query->setExpression(ASTSelectQuery::Expression::WHERE, {}); + select_query->setExpression(ASTSelectQuery::Expression::GROUP_BY, {}); + select_query->group_by_all = false; + /// `removeJoin` clears these. A leftover modifier with no `GROUP BY` rejects the shard query. + select_query->group_by_with_totals = false; + select_query->group_by_with_rollup = false; + select_query->group_by_with_cube = false; + select_query->group_by_with_grouping_sets = false; + select_query->setExpression(ASTSelectQuery::Expression::HAVING, {}); + select_query->setExpression(ASTSelectQuery::Expression::ORDER_BY, {}); + select_query->order_by_all = false; + } + + /// FetchColumns on remotes must not apply initiator-only LIMIT / WINDOW / QUALIFY. + select_query->setExpression(ASTSelectQuery::Expression::WINDOW, {}); + select_query->setExpression(ASTSelectQuery::Expression::QUALIFY, {}); + select_query->setExpression(ASTSelectQuery::Expression::LIMIT_BY_OFFSET, {}); + select_query->setExpression(ASTSelectQuery::Expression::LIMIT_BY_LENGTH, {}); + select_query->setExpression(ASTSelectQuery::Expression::LIMIT_BY, {}); + select_query->setExpression(ASTSelectQuery::Expression::LIMIT_OFFSET, {}); + select_query->setExpression(ASTSelectQuery::Expression::LIMIT_LENGTH, {}); + select_query->setExpression(ASTSelectQuery::Expression::INTERPOLATE, {}); +} + +QueryProcessingStage::Enum IStorageCluster::getQueryProcessingStage( + ContextPtr context, QueryProcessingStage::Enum to_stage, const StorageSnapshotPtr &, SelectQueryInfo & query_info) const +{ + auto object_storage_cluster_join_mode = context->getSettingsRef()[Setting::object_storage_cluster_join_mode]; + const auto join_info = getQueryJoinInfo(query_info, context); + /// `global` still fetches columns for `RIGHT`/`FULL`/`PASTE`: those joins stay on the initiator. + const bool keep_join_on_initiator = object_storage_cluster_join_mode != ObjectStorageClusterJoinMode::GLOBAL + || queryHasRightOrFullJoin(query_info.query); + + if (keep_join_on_initiator && needsInitiatorLocalJoin(join_info)) + { + return QueryProcessingStage::Enum::FetchColumns; } /// Initiator executes query on remote node. diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 664d1c4980817..6feb8116f7dc6 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -63,6 +63,9 @@ class IStorageCluster : public IStorage const String & getOriginalClusterName() const { return cluster_name; } virtual String getClusterName(ContextPtr /* context */) const { return getOriginalClusterName(); } + /// Old interpreter: rewrite JOIN / IN to GLOBAL JOIN / GLOBAL IN so GlobalSubqueriesVisitor can broadcast right tables. + static void rewriteASTForGlobalJoin(ASTPtr & query); + protected: virtual void updateBeforeRead(const ContextPtr &) {} virtual void updateQueryToSendIfNeeded( @@ -70,7 +73,10 @@ class IStorageCluster : public IStorage const StorageSnapshotPtr & /*storage_snapshot*/, const ContextPtr & /*context*/, bool /*make_cluster_function*/) {} - void updateQueryWithJoinToSendIfNeeded(ASTPtr & query_to_send, SelectQueryInfo query_info, const ContextPtr & context); + void updateQueryWithJoinToSendIfNeeded( + ASTPtr & query_to_send, + SelectQueryInfo query_info, + const ContextPtr & context); virtual void updateConfigurationIfNeeded(ContextPtr /* context */) {} @@ -167,6 +173,14 @@ class IStorageCluster : public IStorage }; static QueryTreeInfo getQueryTreeInfo(QueryTreeNodePtr query_tree, ContextPtr context); + static QueryTreeInfo getQueryJoinInfoFromAST(const ASTPtr & query); + static QueryTreeInfo getQueryJoinInfo(const SelectQueryInfo & query_info, const ContextPtr & context); + static bool needsInitiatorLocalJoin(const QueryTreeInfo & info); + static void rewriteQueryForInitiatorLocalJoin( + ASTPtr & query_to_send, + const SelectQueryInfo & query_info, + const QueryTreeInfo & info, + const ContextPtr & context); }; diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 5270a2b5a3bd1..af2f93cf87439 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -2629,7 +2629,7 @@ def create_namespace(suffix): # TODO - turn on after merge alternative syntax -@pytest.mark.parametrize("join_mode", ["local", "global"]) +@pytest.mark.parametrize("join_mode", ["allow", "local", "global"]) def _test_cluster_joins(started_cluster, join_mode): node = started_cluster.instances["node1"] diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index 5afb360ff3da8..745420b5a81d9 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -1467,7 +1467,8 @@ def test_global_join_executes_on_shards(started_cluster): @pytest.mark.parametrize("join_mode", ["local", "global"]) -def test_joins(started_cluster, join_mode): +@pytest.mark.parametrize("allow_experimental_analyzer", [0, 1]) +def test_joins(started_cluster, join_mode, allow_experimental_analyzer): node = started_cluster.instances["s0_0_0"] # Table join_table only exists on the node 's0_0_0'. @@ -1501,7 +1502,7 @@ def test_joins(started_cluster, join_mode): join_table AS t2 ON t1.value = t2.id ORDER BY t1.name - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) @@ -1524,7 +1525,7 @@ def test_joins(started_cluster, join_mode): 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') AS t1 ON t1.value = t2.id ORDER BY t1.name - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) @@ -1542,7 +1543,7 @@ def test_joins(started_cluster, join_mode): ON t1.value = t2.id WHERE (t1.value % 2) ORDER BY t1.name - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) @@ -1561,7 +1562,7 @@ def test_joins(started_cluster, join_mode): ON t1.value = t2.id WHERE (t2.id % 2) ORDER BY t1.name - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) @@ -1579,7 +1580,7 @@ def test_joins(started_cluster, join_mode): ON t1.value = t2.id WHERE (t1.value % 2) AND ((t2.id % 3) == 2) ORDER BY t1.name - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) @@ -1595,7 +1596,7 @@ def test_joins(started_cluster, join_mode): 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') WHERE value IN (SELECT id FROM join_table) ORDER BY name - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) res = list(map(str.split, result6.splitlines())) @@ -1610,7 +1611,7 @@ def test_joins(started_cluster, join_mode): 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') WHERE value GLOBAL IN (SELECT id FROM join_table) ORDER BY name - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) res = list(map(str.split, result6.splitlines())) @@ -1627,7 +1628,7 @@ def test_joins(started_cluster, join_mode): join_table AS t2 ON 1 GROUP BY ALL - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) assert result7.strip() == "625" @@ -1643,7 +1644,7 @@ def test_joins(started_cluster, join_mode): join_table AS t2 ON 1 GROUP BY ALL - SETTINGS object_storage_cluster_join_mode='{join_mode}'; + SETTINGS object_storage_cluster_join_mode='{join_mode}', allow_experimental_analyzer={allow_experimental_analyzer}; """ ) res = list(map(str.split, result8.splitlines())) diff --git a/tests/integration/test_storage_iceberg_with_spark/test_cluster_joins.py b/tests/integration/test_storage_iceberg_with_spark/test_cluster_joins.py index 82e9d6c3c5720..ab2d34311403d 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_cluster_joins.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_cluster_joins.py @@ -6,7 +6,7 @@ execute_spark_query_general, ) -@pytest.mark.parametrize("join_mode", ["local", "global"]) +@pytest.mark.parametrize("join_mode", ["allow", "local", "global"]) @pytest.mark.parametrize("storage_type", ["s3", "azure"]) def test_cluster_joins(started_cluster_iceberg_with_spark, storage_type, join_mode): instance = started_cluster_iceberg_with_spark.instances["node1"] diff --git a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py index 763836d21f607..77891e65fb449 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_remote_initiator.py @@ -105,7 +105,7 @@ def flush_logs(): @pytest.mark.parametrize("storage_type", ["s3"]) -def test_remote_initiator_after_with_join_old_analyzer(started_cluster_iceberg_with_spark, storage_type): +def test_remote_initiator_global_join_without_analyzer(started_cluster_iceberg_with_spark, storage_type): instance = started_cluster_iceberg_with_spark.instances["node1"] spark = started_cluster_iceberg_with_spark.spark_session TABLE_NAME = "test_remote_initiator_after_with_join_old_analyzer_table_" + get_uuid_str() @@ -144,13 +144,14 @@ def execute_spark_query(query: str): instance.query(f"CREATE TABLE {TABLE2_NAME} (tag INT, number2 INT) ENGINE=Memory") instance.query(f"INSERT INTO {TABLE2_NAME} VALUES (1, 2)") - assert "object_storage_cluster_join_mode!='allow' is not supported without allow_experimental_analyzer=true" in instance.query_and_get_error(f""" - SELECT * + res = instance.query(f""" + SELECT t1.tag, t1.number, t2.number2 FROM {TABLE_NAME} AS t1 JOIN {TABLE2_NAME} AS t2 USING (tag) SETTINGS object_storage_remote_initiator=1, object_storage_remote_initiator_cluster='cluster_simple', - object_storage_cluster_join_mode='local', + object_storage_cluster_join_mode='global', allow_experimental_analyzer=0 """) + assert res == "1\t1\t2\n" diff --git a/tests/queries/0_stateless/05027_s3_cluster_join_order_by_all.reference b/tests/queries/0_stateless/05027_s3_cluster_join_order_by_all.reference new file mode 100644 index 0000000000000..b91dd8da35db6 --- /dev/null +++ b/tests/queries/0_stateless/05027_s3_cluster_join_order_by_all.reference @@ -0,0 +1,10 @@ +1 x a +1 x a +2 y b +x 2 +y 1 +1 x a +1 x a +2 y b +x 2 +y 1 diff --git a/tests/queries/0_stateless/05027_s3_cluster_join_order_by_all.sql b/tests/queries/0_stateless/05027_s3_cluster_join_order_by_all.sql new file mode 100644 index 0000000000000..29f087aea0624 --- /dev/null +++ b/tests/queries/0_stateless/05027_s3_cluster_join_order_by_all.sql @@ -0,0 +1,49 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- Old analyzer, `s3Cluster` joined to a table that exists only as far as the initiator is concerned. +-- `ORDER BY ALL` must not null-dereference `orderBy` after the join is stripped, and `GROUP BY ALL` +-- must count joined rows. Two identical left rows for key 1 make a shard-side `GROUP BY` visible: +-- it would report `x` once instead of twice. + +DROP TABLE IF EXISTS t_dim; + +CREATE TABLE t_dim (k UInt64, name String) ENGINE = Memory; +INSERT INTO t_dim VALUES (1, 'a'), (2, 'b'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05027_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') +SELECT k, v FROM +( + SELECT 1 AS k, 'x' AS v + UNION ALL SELECT 1, 'x' + UNION ALL SELECT 2, 'y' +) +SETTINGS s3_truncate_on_insert = 1; + +SELECT t1.k, t1.v, t2.name +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05027_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +INNER JOIN t_dim AS t2 ON t1.k = t2.k +ORDER BY ALL +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'allow'; + +SELECT t1.v, count() +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05027_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +INNER JOIN t_dim AS t2 ON t1.k = t2.k +GROUP BY ALL +ORDER BY ALL +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'allow'; + +SELECT t1.k, t1.v, t2.name +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05027_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +INNER JOIN t_dim AS t2 ON t1.k = t2.k +ORDER BY ALL +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'local'; + +SELECT t1.v, count() +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05027_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +INNER JOIN t_dim AS t2 ON t1.k = t2.k +GROUP BY ALL +ORDER BY ALL +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'local'; + +DROP TABLE t_dim; diff --git a/tests/queries/0_stateless/05386_s3_cluster_right_full_join_stays_on_initiator.reference b/tests/queries/0_stateless/05386_s3_cluster_right_full_join_stays_on_initiator.reference new file mode 100644 index 0000000000000..fd24f8d572f31 --- /dev/null +++ b/tests/queries/0_stateless/05386_s3_cluster_right_full_join_stays_on_initiator.reference @@ -0,0 +1,11 @@ +RIGHT +1 x a +1 x a +0 b +FULL +1 x a +1 x a +0 b +INNER +1 x a +1 x a diff --git a/tests/queries/0_stateless/05386_s3_cluster_right_full_join_stays_on_initiator.sql b/tests/queries/0_stateless/05386_s3_cluster_right_full_join_stays_on_initiator.sql new file mode 100644 index 0000000000000..33b378515b66d --- /dev/null +++ b/tests/queries/0_stateless/05386_s3_cluster_right_full_join_stays_on_initiator.sql @@ -0,0 +1,43 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- Old analyzer, `object_storage_cluster_join_mode = 'global'`. +-- `RIGHT` and `FULL` preserve the right side, so they must run on the initiator. +-- Two shards and one file: a shard that reads no files would still emit every preserved +-- right row if the join were broadcast. The right table is temporary, so only the +-- initiator session can see it. + +DROP TABLE IF EXISTS t_dim_05386; +CREATE TEMPORARY TABLE t_dim_05386 (k UInt64, name String); +INSERT INTO t_dim_05386 VALUES (1, 'a'), (2, 'b'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05386_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') +SELECT k, v FROM +( + SELECT 1 AS k, 'x' AS v + UNION ALL SELECT 1, 'x' +) +SETTINGS s3_truncate_on_insert = 1; + +SELECT 'RIGHT'; +SELECT t1.k, t1.v, t2.name +FROM s3Cluster('test_cluster_two_shards', 'http://localhost:11111/test/05386_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +RIGHT JOIN t_dim_05386 AS t2 ON t1.k = t2.k +ORDER BY t2.name, t1.v +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'global'; + +SELECT 'FULL'; +SELECT t1.k, t1.v, t2.name +FROM s3Cluster('test_cluster_two_shards', 'http://localhost:11111/test/05386_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +FULL JOIN t_dim_05386 AS t2 ON t1.k = t2.k +ORDER BY t2.name, t1.v +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'global'; + +SELECT 'INNER'; +SELECT t1.k, t1.v, t2.name +FROM s3Cluster('test_cluster_two_shards', 'http://localhost:11111/test/05386_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +INNER JOIN t_dim_05386 AS t2 ON t1.k = t2.k +ORDER BY t1.v +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'global'; + +DROP TEMPORARY TABLE t_dim_05386; diff --git a/tests/queries/0_stateless/05387_s3_cluster_remote_initiator_keeps_stripped_join.reference b/tests/queries/0_stateless/05387_s3_cluster_remote_initiator_keeps_stripped_join.reference new file mode 100644 index 0000000000000..422813a4b9365 --- /dev/null +++ b/tests/queries/0_stateless/05387_s3_cluster_remote_initiator_keeps_stripped_join.reference @@ -0,0 +1,8 @@ +old +1 x a +1 x a +2 y b +analyzer +1 x a +1 x a +2 y b diff --git a/tests/queries/0_stateless/05387_s3_cluster_remote_initiator_keeps_stripped_join.sql b/tests/queries/0_stateless/05387_s3_cluster_remote_initiator_keeps_stripped_join.sql new file mode 100644 index 0000000000000..b3ed8624ea7ab --- /dev/null +++ b/tests/queries/0_stateless/05387_s3_cluster_remote_initiator_keeps_stripped_join.sql @@ -0,0 +1,42 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- `object_storage_remote_initiator` analyzes the query on another host. +-- `allow` strips the JOIN into `query_to_send`; that stripped query must be what the remote host +-- runs. The right table is temporary, so it exists only in this session. + +CREATE TEMPORARY TABLE t_dim_05387 (k UInt64, name String); +INSERT INTO t_dim_05387 VALUES (1, 'a'), (2, 'b'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05387_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') +SELECT k, v FROM +( + SELECT 1 AS k, 'x' AS v + UNION ALL SELECT 1, 'x' + UNION ALL SELECT 2, 'y' +) +SETTINGS s3_truncate_on_insert = 1; + +SELECT 'old'; +SELECT t1.k, t1.v, t2.name +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05387_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +INNER JOIN t_dim_05387 AS t2 ON t1.k = t2.k +ORDER BY t1.k, t1.v +SETTINGS + allow_experimental_analyzer = 0, + object_storage_cluster_join_mode = 'allow', + object_storage_remote_initiator = 1, + object_storage_remote_initiator_cluster = 'test_shard_localhost'; + +SELECT 'analyzer'; +SELECT t1.k, t1.v, t2.name +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05387_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +INNER JOIN t_dim_05387 AS t2 ON t1.k = t2.k +ORDER BY t1.k, t1.v +SETTINGS + allow_experimental_analyzer = 1, + object_storage_cluster_join_mode = 'allow', + object_storage_remote_initiator = 1, + object_storage_remote_initiator_cluster = 'test_shard_localhost'; + +DROP TEMPORARY TABLE t_dim_05387; diff --git a/tests/queries/0_stateless/05388_s3_cluster_right_join_where_not_pushed.reference b/tests/queries/0_stateless/05388_s3_cluster_right_join_where_not_pushed.reference new file mode 100644 index 0000000000000..7b8528f2c4eab --- /dev/null +++ b/tests/queries/0_stateless/05388_s3_cluster_right_join_where_not_pushed.reference @@ -0,0 +1,4 @@ +RIGHT +\N \N 2 unmatched +INNER +1 L 1 matched diff --git a/tests/queries/0_stateless/05388_s3_cluster_right_join_where_not_pushed.sql b/tests/queries/0_stateless/05388_s3_cluster_right_join_where_not_pushed.sql new file mode 100644 index 0000000000000..8f51219c83651 --- /dev/null +++ b/tests/queries/0_stateless/05388_s3_cluster_right_join_where_not_pushed.sql @@ -0,0 +1,36 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- A left-only predicate must not run before a `RIGHT` join. With `join_use_nulls = 1`, +-- `WHERE isNull(t1.k)` is true for unmatched right rows. Filtering the matching left row +-- on the shard turns the matched right row into another unmatched row. + +CREATE TEMPORARY TABLE t_dim_05388 (k Nullable(Int32), name String); +INSERT INTO t_dim_05388 VALUES (1, 'matched'), (2, 'unmatched'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05388_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k Nullable(Int32), v String') +SELECT k, v FROM +( + SELECT CAST(1, 'Nullable(Int32)') AS k, 'L' AS v + UNION ALL SELECT CAST(1, 'Nullable(Int32)'), 'M' + UNION ALL SELECT CAST(NULL, 'Nullable(Int32)'), 'N' +) +SETTINGS s3_truncate_on_insert = 1; + +SELECT 'RIGHT'; +SELECT t1.k, t1.v, t2.k, t2.name +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05388_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k Nullable(Int32), v String') AS t1 +RIGHT JOIN t_dim_05388 AS t2 ON t1.k = t2.k +WHERE isNull(t1.k) +ORDER BY t2.k, t1.v +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'allow', join_use_nulls = 1; + +SELECT 'INNER'; +SELECT t1.k, t1.v, t2.k, t2.name +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05388_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k Nullable(Int32), v String') AS t1 +INNER JOIN t_dim_05388 AS t2 ON t1.k = t2.k +WHERE t1.v = 'L' +ORDER BY t1.v +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'allow', join_use_nulls = 1; + +DROP TEMPORARY TABLE t_dim_05388; diff --git a/tests/queries/0_stateless/05389_s3_cluster_in_subquery_group_by_with_totals.reference b/tests/queries/0_stateless/05389_s3_cluster_in_subquery_group_by_with_totals.reference new file mode 100644 index 0000000000000..1fe64875bcf72 --- /dev/null +++ b/tests/queries/0_stateless/05389_s3_cluster_in_subquery_group_by_with_totals.reference @@ -0,0 +1,3 @@ +1 2 + +0 2 diff --git a/tests/queries/0_stateless/05389_s3_cluster_in_subquery_group_by_with_totals.sql b/tests/queries/0_stateless/05389_s3_cluster_in_subquery_group_by_with_totals.sql new file mode 100644 index 0000000000000..e181bc06529bc --- /dev/null +++ b/tests/queries/0_stateless/05389_s3_cluster_in_subquery_group_by_with_totals.sql @@ -0,0 +1,27 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- No explicit JOIN. The `IN` subquery references a temporary table, so the shard query drops +-- `GROUP BY`. The `WITH TOTALS` flag has to be dropped with it; otherwise the shard query is +-- rejected and the initiator never sees the rows. + +CREATE TEMPORARY TABLE t_dim_05389 (k UInt64); +INSERT INTO t_dim_05389 VALUES (1); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05389_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') +SELECT k, v FROM +( + SELECT 1 AS k, 'x' AS v + UNION ALL SELECT 1, 'x' + UNION ALL SELECT 2, 'y' +) +SETTINGS s3_truncate_on_insert = 1; + +SELECT t1.k, count() +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05389_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +WHERE t1.k IN (SELECT k FROM t_dim_05389) +GROUP BY t1.k WITH TOTALS +ORDER BY t1.k +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'allow'; + +DROP TEMPORARY TABLE t_dim_05389; diff --git a/tests/queries/0_stateless/05390_s3_cluster_in_subquery_without_columns.reference b/tests/queries/0_stateless/05390_s3_cluster_in_subquery_without_columns.reference new file mode 100644 index 0000000000000..499ca679a67b1 --- /dev/null +++ b/tests/queries/0_stateless/05390_s3_cluster_in_subquery_without_columns.reference @@ -0,0 +1,6 @@ +no-column +1 x +1 x +column +1 x +1 x diff --git a/tests/queries/0_stateless/05390_s3_cluster_in_subquery_without_columns.sql b/tests/queries/0_stateless/05390_s3_cluster_in_subquery_without_columns.sql new file mode 100644 index 0000000000000..6eb61b2d654bc --- /dev/null +++ b/tests/queries/0_stateless/05390_s3_cluster_in_subquery_without_columns.sql @@ -0,0 +1,33 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- Analyzer. `SELECT 1 FROM t_dim` references a temporary table and no column of it. +-- That still has to stay on the initiator; workers cannot resolve the temporary table. + +CREATE TEMPORARY TABLE t_dim_05390 (k UInt64, name String); +INSERT INTO t_dim_05390 VALUES (1, 'a'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05390_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') +SELECT k, v FROM +( + SELECT 1 AS k, 'x' AS v + UNION ALL SELECT 1, 'x' + UNION ALL SELECT 2, 'y' +) +SETTINGS s3_truncate_on_insert = 1; + +SELECT 'no-column'; +SELECT t1.k, t1.v +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05390_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +WHERE t1.k IN (SELECT 1 FROM t_dim_05390 LIMIT 1) +ORDER BY t1.k, t1.v +SETTINGS allow_experimental_analyzer = 1, object_storage_cluster_join_mode = 'allow'; + +SELECT 'column'; +SELECT t1.k, t1.v +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05390_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +WHERE t1.k IN (SELECT k FROM t_dim_05390) +ORDER BY t1.k, t1.v +SETTINGS allow_experimental_analyzer = 1, object_storage_cluster_join_mode = 'allow'; + +DROP TEMPORARY TABLE t_dim_05390; diff --git a/tests/queries/0_stateless/05391_s3_cluster_array_join_then_join.reference b/tests/queries/0_stateless/05391_s3_cluster_array_join_then_join.reference new file mode 100644 index 0000000000000..f3a529a194b61 --- /dev/null +++ b/tests/queries/0_stateless/05391_s3_cluster_array_join_then_join.reference @@ -0,0 +1,2 @@ +x a +y a diff --git a/tests/queries/0_stateless/05391_s3_cluster_array_join_then_join.sql b/tests/queries/0_stateless/05391_s3_cluster_array_join_then_join.sql new file mode 100644 index 0000000000000..2450690874cbb --- /dev/null +++ b/tests/queries/0_stateless/05391_s3_cluster_array_join_then_join.sql @@ -0,0 +1,21 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- `FROM t ARRAY JOIN arr INNER JOIN dim` is one join. The shard scan must drop the join +-- instead of throwing because the second table element is the `ARRAY JOIN`. + +CREATE TEMPORARY TABLE t_dim_05391 (k UInt64, name String); +INSERT INTO t_dim_05391 VALUES (1, 'a'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05391_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, arr Array(String)') +SELECT 1, ['x', 'y'] +SETTINGS s3_truncate_on_insert = 1; + +SELECT elem, t2.name +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05391_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, arr Array(String)') AS t1 +ARRAY JOIN arr AS elem +INNER JOIN t_dim_05391 AS t2 ON t1.k = t2.k +ORDER BY elem +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'allow'; + +DROP TEMPORARY TABLE t_dim_05391; diff --git a/tests/queries/0_stateless/05392_s3_cluster_paste_join_stays_on_initiator.reference b/tests/queries/0_stateless/05392_s3_cluster_paste_join_stays_on_initiator.reference new file mode 100644 index 0000000000000..60d24e9a0d1e1 --- /dev/null +++ b/tests/queries/0_stateless/05392_s3_cluster_paste_join_stays_on_initiator.reference @@ -0,0 +1,4 @@ +old +1 +analyzer +1 diff --git a/tests/queries/0_stateless/05392_s3_cluster_paste_join_stays_on_initiator.sql b/tests/queries/0_stateless/05392_s3_cluster_paste_join_stays_on_initiator.sql new file mode 100644 index 0000000000000..7977ba0e60e3c --- /dev/null +++ b/tests/queries/0_stateless/05392_s3_cluster_paste_join_stays_on_initiator.sql @@ -0,0 +1,28 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- `PASTE` pairs rows by position. With `global`, a one-row right side paired on each +-- shard that has left rows would return one row per such shard. Several files make +-- both shards nonempty, so the initiator must return a single row. + +CREATE TEMPORARY TABLE t_dim_05392 (name String); +INSERT INTO t_dim_05392 VALUES ('a'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05392_' || currentDatabase() || '_{_partition_id}.tsv', 'test', 'testtest', 'TSV', 'k UInt64') +PARTITION BY k +SELECT number AS k FROM numbers(16) +SETTINGS s3_truncate_on_insert = 1; + +SELECT 'old'; +SELECT count() +FROM s3Cluster('test_cluster_two_shards', 'http://localhost:11111/test/05392_' || currentDatabase() || '_*.tsv', 'test', 'testtest', 'TSV', 'k UInt64') AS t1 +PASTE JOIN t_dim_05392 AS t2 +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'global', lock_object_storage_task_distribution_ms = 600000; + +SELECT 'analyzer'; +SELECT count() +FROM s3Cluster('test_cluster_two_shards', 'http://localhost:11111/test/05392_' || currentDatabase() || '_*.tsv', 'test', 'testtest', 'TSV', 'k UInt64') AS t1 +PASTE JOIN t_dim_05392 AS t2 +SETTINGS allow_experimental_analyzer = 1, object_storage_cluster_join_mode = 'global', lock_object_storage_task_distribution_ms = 600000; + +DROP TEMPORARY TABLE t_dim_05392;