From 454ce3654d0ba5ca80dc92e9604c44275055ec95 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 19 Oct 2026 13:48:56 +0200 Subject: [PATCH 01/14] Make `object_storage_cluster_join_mode=allow` keep initiator-local joins. Frontport of #2228. `local` stays as a deprecated alias of `allow`. The right table is joined on the initiator, including with the old analyzer, and `global` broadcasts that table without requiring the analyzer. Co-authored-by: Cursor --- docs/en/antalya/swarm.md | 11 +- .../sql-reference/distribution-on-cluster.md | 2 +- src/Core/Settings.cpp | 9 +- src/Core/SettingsEnums.h | 4 +- src/Interpreters/InterpreterSelectQuery.cpp | 8 + src/Storages/IStorageCluster.cpp | 473 ++++++++++++------ src/Storages/IStorageCluster.h | 11 + .../integration/test_database_iceberg/test.py | 2 +- tests/integration/test_s3_cluster/test.py | 21 +- .../test_cluster_joins.py | 2 +- .../test_remote_initiator.py | 9 +- ...385_s3_cluster_join_order_by_all.reference | 10 + .../05385_s3_cluster_join_order_by_all.sql | 49 ++ 13 files changed, 424 insertions(+), 187 deletions(-) create mode 100644 tests/queries/0_stateless/05385_s3_cluster_join_order_by_all.reference create mode 100644 tests/queries/0_stateless/05385_s3_cluster_join_order_by_all.sql diff --git a/docs/en/antalya/swarm.md b/docs/en/antalya/swarm.md index 5b40c5dc704fb..b47ad8ea77888 100644 --- a/docs/en/antalya/swarm.md +++ b/docs/en/antalya/swarm.md @@ -281,14 +281,13 @@ the same node while the cluster is stable, which helps local caches. See ### Joins against local tables {#joins} A swarm query can `JOIN` a table that exists only on the initiator. Workers do not have that table, -so you must choose how the join is rewritten with `object_storage_cluster_join_mode` -(requires `allow_experimental_analyzer = 1` unless the value is `allow`): +so you must choose how the join is rewritten with `object_storage_cluster_join_mode`: | Value | Behaviour | | --- | --- | -| `allow` (default) | Leave the join as written. Use this when both sides are object-storage cluster scans, or when workers can see the right-hand table. | -| `local` | Send only the object-storage scan to the swarm; perform the join on the initiator. | -| `global` | Read the right-hand side on the initiator first and ship it to workers as a temporary table. | +| `allow` (default) | Send only the object-storage scan to the swarm; perform the join on the initiator. The right table does not need to exist on workers. | +| `local` | Deprecated alias of `allow`. | +| `global` | Read the right-hand side on the initiator first and ship it to workers as a temporary table. `INNER`, `LEFT`, and `CROSS` run on the workers. `RIGHT` and `FULL` stay on the initiator. | ```sql SELECT * @@ -467,5 +466,5 @@ connections drain. | `object_storage_max_nodes` | Cap on how many hosts from that cluster run a given query. `0` = all. | | `object_storage_remote_initiator` | Run the cluster query on a random swarm node instead of coordinating from the client-facing initiator. | | `object_storage_remote_initiator_cluster` | Cluster used only to pick that remote initiator. When empty, `object_storage_cluster` is used. | -| `object_storage_cluster_join_mode` | How to rewrite `JOIN`s when workers cannot see the right-hand table: `allow`, `local`, `global`. | +| `object_storage_cluster_join_mode` | How to rewrite `JOIN`s when workers cannot see the right-hand table: `allow` (initiator-local join), `local` (deprecated alias of `allow`), `global`. | | `lock_object_storage_task_distribution_ms` | How long a free worker waits before stealing another node's files. Higher values favor cache locality. Default `500`. | diff --git a/docs/en/sql-reference/distribution-on-cluster.md b/docs/en/sql-reference/distribution-on-cluster.md index 82e4c3fcae949..2cb2597fd78ce 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` the nodes read the left table and the join runs on the initiator. `local` is a deprecated alias of `allow`. 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. 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 47cef0c8dcc3a..64b85e10be2f7 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -2306,10 +2306,15 @@ 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. +- `allow` — Default. 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`. - `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. See [Distributed JOIN](../../sql-reference/statements/select/join.md#object-storage-cluster-join-mode). diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h index 17cc35dd7f84e..af88efe055d5a 100644 --- a/src/Core/SettingsEnums.h +++ b/src/Core/SettingsEnums.h @@ -171,9 +171,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 df906bbd51b04..f7ab3caf4e851 100644 --- a/src/Interpreters/InterpreterSelectQuery.cpp +++ b/src/Interpreters/InterpreterSelectQuery.cpp @@ -229,6 +229,7 @@ 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 SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; } namespace ServerSetting @@ -815,6 +816,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/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index c66fc0b4a7709..24df15c37fc59 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -42,10 +42,18 @@ #include #include #include -#include -#include #include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include @@ -176,49 +184,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 */ @@ -269,7 +234,117 @@ class CollectUsedColumnsForSourceVisitor : public InDepthQueryTreeVisitorWithCon bool collect_columns_from_other_sources; }; -}; +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 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); + } +} + +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; + } + + 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 @@ -282,7 +357,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 @@ -305,101 +380,24 @@ 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); - - if (info.has_join || info.has_cross_join || info.has_local_columns_in_where) - { - 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) - { - const auto & join_node = query_node.getJoinTreeNode(); - query_tree_distributed = join_node->as()->getLeftTableExpressionNode()->clone(); - } - 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, std::static_pointer_cast(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.getJoinTreeNodeTyped(), std::static_pointer_cast(query_tree_distributed)); - } - - query_to_send = queryNodeToDistributedSelectQuery(modified_query_tree); - } + auto info = getQueryJoinInfo(query_info, context); + if (!needsInitiatorLocalJoin(info)) + return; + rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context); return; } case ObjectStorageClusterJoinMode::GLOBAL: { + if (!query_info.query_tree) + return; + 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(); @@ -418,8 +416,6 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( return; } - case ObjectStorageClusterJoinMode::ALLOW: // Do nothing special - return; } } @@ -686,9 +682,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, @@ -883,23 +881,22 @@ IStorageCluster::QueryTreeInfo IStorageCluster::getQueryTreeInfo(QueryTreeNodePt QueryTreeInfo info; auto & query_node = query_tree->as(); - if (const auto & join_node = query_node.getJoinTreeNode()) - { - 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.getJoinTreeNode(); + if (!join_tree) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); - 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) + if (join_tree->getNodeType() == QueryTreeNodeType::JOIN) + info.has_join = true; + else if (join_tree->getNodeType() == QueryTreeNodeType::CROSS_JOIN) + info.has_cross_join = true; + + auto left_table_expression = extractLeftTableExpression(query_node.getJoinTreeNodeTyped()); + 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()) @@ -915,23 +912,179 @@ IStorageCluster::QueryTreeInfo IStorageCluster::getQueryTreeInfo(QueryTreeNodePt 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 (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` and `buildQueryToReadColumnsFromTableExpression` + /// (same helpers the planner uses when wrapping `IStorageCluster` in a subquery). + if (query_info.query_tree && query_info.query_tree->as()) + { + auto modified_query_tree = query_info.query_tree->clone(); + auto & query_node = modified_query_tree->as(); + auto join_tree = query_node.getJoinTreeNode(); + if (!join_tree) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); + + auto left_table_expression = extractLeftTableExpression(query_node.getJoinTreeNodeTyped()); + 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); + + 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`, and `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(); + } - if (object_storage_cluster_join_mode == ObjectStorageClusterJoinMode::LOCAL) + 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; + 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 WHERE, including `GLOBAL IN (_subqueryN)`, which remotes cannot resolve. + if (astContainsInTableIdentifier(select_query->where()) || astContainsInTableIdentifier(select_query->prewhere()) + || astContainsSubquery(select_query->where()) || astContainsSubquery(select_query->prewhere())) + { + select_query->setExpression(ASTSelectQuery::Expression::PREWHERE, {}); + select_query->setExpression(ASTSelectQuery::Expression::WHERE, {}); + } + } + 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; + 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]; + + if (object_storage_cluster_join_mode != ObjectStorageClusterJoinMode::GLOBAL + && needsInitiatorLocalJoin(getQueryJoinInfo(query_info, context))) + { + return QueryProcessingStage::Enum::FetchColumns; } /// Only a follower reached by another node's cluster function (SECONDARY_QUERY) just fetches diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 97019fed87605..c592a259e12b0 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); + /// Prepare the `SELECT ... FROM f(...)` query (`f` is a table function) for the other nodes of the cluster: add the /// structure and format arguments so that the nodes do not infer the schema again, and turn a plain table /// function (`url`, `s3`, ...) that `parallel_replicas_for_cluster_engines` converted into this cluster @@ -174,6 +177,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 02d09468d9214..713fe544f49a8 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3133,7 +3133,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 1aa1d56cc648c..aaff4f0d99b0f 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -1466,7 +1466,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'. @@ -1500,7 +1501,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}; """ ) @@ -1523,7 +1524,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}; """ ) @@ -1541,7 +1542,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}; """ ) @@ -1560,7 +1561,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}; """ ) @@ -1578,7 +1579,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}; """ ) @@ -1594,7 +1595,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())) @@ -1609,7 +1610,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())) @@ -1626,7 +1627,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" @@ -1642,7 +1643,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/05385_s3_cluster_join_order_by_all.reference b/tests/queries/0_stateless/05385_s3_cluster_join_order_by_all.reference new file mode 100644 index 0000000000000..b91dd8da35db6 --- /dev/null +++ b/tests/queries/0_stateless/05385_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/05385_s3_cluster_join_order_by_all.sql b/tests/queries/0_stateless/05385_s3_cluster_join_order_by_all.sql new file mode 100644 index 0000000000000..1996d196369b9 --- /dev/null +++ b/tests/queries/0_stateless/05385_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/05385_' || 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/05385_' || 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/05385_' || 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/05385_' || 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/05385_' || 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; From 86b85863a3c65d308e2b7988165c6ad583f316cf Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 19 Oct 2026 17:32:27 +0200 Subject: [PATCH 02/14] Keep old-analyzer `RIGHT` and `FULL` joins on the initiator. `object_storage_cluster_join_mode = 'global'` was marking every join `GLOBAL`, so each worker emitted the preserved right rows again. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 65 ++++++++++++++++++- ...ght_full_join_stays_on_initiator.reference | 11 ++++ ...ter_right_full_join_stays_on_initiator.sql | 43 ++++++++++++ 3 files changed, 117 insertions(+), 2 deletions(-) create mode 100644 tests/queries/0_stateless/05386_s3_cluster_right_full_join_stays_on_initiator.reference create mode 100644 tests/queries/0_stateless/05386_s3_cluster_right_full_join_stays_on_initiator.sql diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 24df15c37fc59..34c7f1bba3e14 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -309,6 +309,47 @@ void rewriteASTInFunctionsToGlobalIn(ASTPtr & node) } } +bool selectHasRightOrFullJoin(const ASTSelectQuery & select_query) +{ + const auto tables = select_query.tables(); + if (!tables) + return false; + + for (const auto & child : tables->children) + { + const auto & tables_element = child->as(); + if (!tables_element.table_join) + continue; + + const auto & table_join = tables_element.table_join->as(); + if (isRightOrFull(table_join.kind)) + return true; + } + return false; +} + +bool queryHasRightOrFullJoin(const ASTPtr & query) +{ + if (!query) + return false; + + if (const auto * select_query = query->as()) + return selectHasRightOrFullJoin(*select_query); + + if (const auto * union_query = query->as()) + { + if (!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(); @@ -325,6 +366,13 @@ void rewriteASTJoinsToGlobal(ASTPtr & query) return; } + /// `RIGHT` and `FULL` preserve rows from the broadcast side. Running them on every worker + /// emits those rows once per worker. They stay on the initiator, as they do with the analyzer. + /// A preserved-side join also keeps the rest of the join tree there: an earlier `INNER`/`LEFT` + /// cannot be shipped on its own once a later `RIGHT`/`FULL` must see the full left stream. + if (selectHasRightOrFullJoin(*select_query)) + return; + if (auto tables = select_query->tables()) { auto & tables_in_select_query = tables->as(); @@ -392,8 +440,17 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( } case ObjectStorageClusterJoinMode::GLOBAL: { + /// Old analyzer. `RIGHT`/`FULL` were left local by `rewriteASTJoinsToGlobal`. Ship only the + /// left scan; the initiator runs the join. `INNER`/`LEFT`/`CROSS` stay in `query_to_send`. if (!query_info.query_tree) + { + if (queryHasRightOrFullJoin(query_to_send)) + { + auto info = getQueryJoinInfoFromAST(query_to_send); + rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context); + } return; + } auto info = getQueryTreeInfo(query_info.query_tree, context); @@ -1080,9 +1137,13 @@ 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` ships `INNER`/`LEFT`/`CROSS`. `RIGHT`/`FULL` preserve broadcast rows, so even in + /// `global` the initiator fetches left columns and joins locally. + const bool keep_join_on_initiator = object_storage_cluster_join_mode != ObjectStorageClusterJoinMode::GLOBAL + || queryHasRightOrFullJoin(query_info.query); - if (object_storage_cluster_join_mode != ObjectStorageClusterJoinMode::GLOBAL - && needsInitiatorLocalJoin(getQueryJoinInfo(query_info, context))) + if (keep_join_on_initiator && needsInitiatorLocalJoin(join_info)) { return QueryProcessingStage::Enum::FetchColumns; } 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; From 2f934a85da624c67c7cf4cd06e8e98f781d707a3 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 19 Oct 2026 17:34:16 +0200 Subject: [PATCH 03/14] Send the stripped query when a remote initiator runs an initiator-local join. `StorageDistributed::read` was still analyzing the original `query_info`, so the JOIN and its right table were planned on a host that does not have that table. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 68 ++++++++++++++++--- ...te_initiator_keeps_stripped_join.reference | 8 +++ ...r_remote_initiator_keeps_stripped_join.sql | 42 ++++++++++++ 3 files changed, 108 insertions(+), 10 deletions(-) create mode 100644 tests/queries/0_stateless/05387_s3_cluster_remote_initiator_keeps_stripped_join.reference create mode 100644 tests/queries/0_stateless/05387_s3_cluster_remote_initiator_keeps_stripped_join.sql diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 34c7f1bba3e14..76a7e845f1c38 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -282,6 +282,25 @@ 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) @@ -585,10 +604,48 @@ void IStorageCluster::readFromRemoteInitiator( ClusterPtr remote_initiator_cluster, const String & remote_initiator_cluster_name) { + /// `allow`/`local` (and initiator-local `RIGHT`/`FULL`) rewrite only `query_to_send`. + /// Clone that AST before `convertToRemote` wraps its table function in `remote()`. + /// `StorageDistributed::read` must analyze the stripped query. The original `query_info` + /// still contains the JOIN, and replacing only the left table there sends the right table + /// to a host that does not have it. + 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; + /// The syntax result describes the original JOIN. Shard pruning must not reuse it. + 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->getJoinTreeNode()) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected a table scan after stripping the initiator-local JOIN"); + + modified_query_info.table_expression = extractLeftTableExpression(query_node->getJoinTreeNodeTyped()); + } + 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); @@ -798,16 +855,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) 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; From 25437aa30754c59e71976ef62488b375a6a49abc Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 19 Oct 2026 17:35:52 +0200 Subject: [PATCH 04/14] Do not push left-table predicates under initiator-local `RIGHT` and `FULL` joins. `removeJoin` kept those predicates on the shard scan, which changes which right rows are unmatched before the initiator applies the original predicate. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 41 ++++++++++++------- ...ster_right_join_where_not_pushed.reference | 4 ++ ...s3_cluster_right_join_where_not_pushed.sql | 36 ++++++++++++++++ 3 files changed, 67 insertions(+), 14 deletions(-) create mode 100644 tests/queries/0_stateless/05388_s3_cluster_right_join_where_not_pushed.reference create mode 100644 tests/queries/0_stateless/05388_s3_cluster_right_join_where_not_pushed.sql diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 76a7e845f1c38..0d83af52188d7 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -1081,23 +1081,32 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( CollectUsedColumnsForSourceVisitor collector(left_table_expression, context); collector.visit(modified_query_tree); - 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); + /// A left-only predicate applied before a `RIGHT` or `FULL` join changes which right rows are + /// unmatched. The initiator still has the original predicate, so the shard scan must not. + const bool preserves_right_side = queryHasRightOrFullJoin(query_to_send); + if (!preserves_right_side) + { + 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(); + if (!preserves_right_side) + { + rewritten_query_node.getPrewhere() = query_node.getPrewhere(); + rewritten_query_node.getWhere() = query_node.getWhere(); + } query_to_send = queryNodeToDistributedSelectQuery(rewritten_query_tree); return; @@ -1124,6 +1133,9 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( throw Exception(ErrorCodes::LOGICAL_ERROR, "Query is not analyzed: no syntax analyzer result"); TreeRewriterResult rewriter_result = *query_info.syntax_analyzer_result; + /// `removeJoin` drops the join, so the kind has to be read first. A left-only `WHERE` kept + /// below a `RIGHT` or `FULL` join changes which right rows are unmatched. + const bool preserves_right_side = selectHasRightOrFullJoin(*select_query); if (!removeJoin(*select_query, rewriter_result, context)) throw Exception(ErrorCodes::LOGICAL_ERROR, "Failed to strip JOIN from query sent to cluster nodes"); @@ -1136,7 +1148,8 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( select_query->order_by_all = false; /// `removeJoin` keeps left-table WHERE, including `GLOBAL IN (_subqueryN)`, which remotes cannot resolve. - if (astContainsInTableIdentifier(select_query->where()) || astContainsInTableIdentifier(select_query->prewhere()) + if (preserves_right_side + || astContainsInTableIdentifier(select_query->where()) || astContainsInTableIdentifier(select_query->prewhere()) || astContainsSubquery(select_query->where()) || astContainsSubquery(select_query->prewhere())) { select_query->setExpression(ASTSelectQuery::Expression::PREWHERE, {}); 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; From 11ba8eeb9e16d4d3377a7f4620c081a167541fed Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 19 Oct 2026 17:36:24 +0200 Subject: [PATCH 05/14] Clear `GROUP BY` modifiers when an initiator-local `IN` subquery strips the clause. A shard query with `WITH TOTALS`, `ROLLUP`, `CUBE`, or `GROUPING SETS` and no `GROUP BY` is rejected before the initiator can aggregate. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 6 +++++ ...in_subquery_group_by_with_totals.reference | 3 +++ ...uster_in_subquery_group_by_with_totals.sql | 27 +++++++++++++++++++ 3 files changed, 36 insertions(+) create mode 100644 tests/queries/0_stateless/05389_s3_cluster_in_subquery_group_by_with_totals.reference create mode 100644 tests/queries/0_stateless/05389_s3_cluster_in_subquery_group_by_with_totals.sql diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 0d83af52188d7..a7188fded6912 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -1178,6 +1178,12 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( select_query->setExpression(ASTSelectQuery::Expression::WHERE, {}); select_query->setExpression(ASTSelectQuery::Expression::GROUP_BY, {}); select_query->group_by_all = false; + /// Same flags `removeJoin` clears. A leftover `WITH TOTALS`/`ROLLUP`/`CUBE`/`GROUPING SETS` + /// with no `GROUP BY` makes the re-analyzed shard query fail. + 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; 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; From 91c56bb73ca11a4eb3a800372aa19d019be86c94 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 19 Oct 2026 17:37:20 +0200 Subject: [PATCH 06/14] Treat a column-less subquery table as initiator-local under the analyzer. `WHERE k IN (SELECT 1 FROM dim)` names `dim` without projecting a column, so workers were asked to resolve a table that exists only on the initiator. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 41 ++++++++++++++++++- ...ster_in_subquery_without_columns.reference | 6 +++ ...s3_cluster_in_subquery_without_columns.sql | 33 +++++++++++++++ 3 files changed, 79 insertions(+), 1 deletion(-) create mode 100644 tests/queries/0_stateless/05390_s3_cluster_in_subquery_without_columns.reference create mode 100644 tests/queries/0_stateless/05390_s3_cluster_in_subquery_without_columns.sql diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index a7188fded6912..b1b76148eaeb3 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -234,6 +234,37 @@ class CollectUsedColumnsForSourceVisitor : public InDepthQueryTreeVisitorWithCon bool collect_columns_from_other_sources; }; +/// A subquery can name a table and project no column from it (`SELECT 1 FROM dim`). +/// Column collection misses that, and the subquery is then sent to workers that cannot resolve `dim`. +class CollectForeignTableExpressionsVisitor : public InDepthQueryTreeVisitorWithContext +{ +public: + using Base = InDepthQueryTreeVisitorWithContext; + using Base::Base; + + explicit CollectForeignTableExpressionsVisitor(QueryTreeNodePtr left_table_, ContextPtr context) + : Base(context) + , left_table(std::move(left_table_)) + { + } + + void enterImpl(QueryTreeNodePtr & node) + { + const auto node_type = node->getNodeType(); + if (node_type != QueryTreeNodeType::TABLE && node_type != QueryTreeNodeType::TABLE_FUNCTION) + return; + + if (node.get() != left_table.get()) + found = true; + } + + bool hasForeignTable() const { return found; } + +private: + QueryTreeNodePtr left_table; + bool found = false; +}; + bool astContainsSubquery(const ASTPtr & node) { if (!node) @@ -1010,7 +1041,15 @@ 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()) + // `SELECT 1 FROM initiator_only_dim` has a table and no column from it, so also look at + // table expressions. + CollectForeignTableExpressionsVisitor foreign_tables(left_table_expression, context); + if (query_node.hasPrewhere()) + foreign_tables.visit(query_node.getPrewhere()); + if (query_node.hasWhere()) + foreign_tables.visit(query_node.getWhere()); + + if (!collector_where.getColumns().empty() || foreign_tables.hasForeignTable()) info.has_local_columns_in_where = true; } 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; From 87c2d9a4940c3224336716ac762fbec0a5c7ec46 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 19 Oct 2026 17:45:07 +0200 Subject: [PATCH 07/14] Shrink the initiator-local join fixes. Fold foreign-table detection into the existing column visitor, use one `RIGHT`/`FULL` check, and drop a left predicate before the existing filter rewrite. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 162 ++++++++++--------------------- 1 file changed, 52 insertions(+), 110 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index b1b76148eaeb3..3405246737940 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -206,6 +206,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; @@ -226,43 +232,14 @@ 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; -}; - -/// A subquery can name a table and project no column from it (`SELECT 1 FROM dim`). -/// Column collection misses that, and the subquery is then sent to workers that cannot resolve `dim`. -class CollectForeignTableExpressionsVisitor : public InDepthQueryTreeVisitorWithContext -{ -public: - using Base = InDepthQueryTreeVisitorWithContext; - using Base::Base; - - explicit CollectForeignTableExpressionsVisitor(QueryTreeNodePtr left_table_, ContextPtr context) - : Base(context) - , left_table(std::move(left_table_)) - { - } - - void enterImpl(QueryTreeNodePtr & node) - { - const auto node_type = node->getNodeType(); - if (node_type != QueryTreeNodeType::TABLE && node_type != QueryTreeNodeType::TABLE_FUNCTION) - return; - - if (node.get() != left_table.get()) - found = true; - } - - bool hasForeignTable() const { return found; } - -private: - QueryTreeNodePtr left_table; - bool found = false; + bool has_foreign_table = false; }; bool astContainsSubquery(const ASTPtr & node) @@ -359,44 +336,33 @@ void rewriteASTInFunctionsToGlobalIn(ASTPtr & node) } } -bool selectHasRightOrFullJoin(const ASTSelectQuery & select_query) -{ - const auto tables = select_query.tables(); - if (!tables) - return false; - - for (const auto & child : tables->children) - { - const auto & tables_element = child->as(); - if (!tables_element.table_join) - continue; - - const auto & table_join = tables_element.table_join->as(); - if (isRightOrFull(table_join.kind)) - return true; - } - return false; -} - bool queryHasRightOrFullJoin(const ASTPtr & query) { if (!query) return false; if (const auto * select_query = query->as()) - return selectHasRightOrFullJoin(*select_query); - - if (const auto * union_query = query->as()) { - if (!union_query->list_of_selects) + const auto tables = select_query->tables(); + if (!tables) return false; - for (const auto & child : union_query->list_of_selects->children) + for (const auto & child : tables->children) { - if (queryHasRightOrFullJoin(child)) + const auto & element = child->as(); + if (element.table_join && isRightOrFull(element.table_join->as().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; } @@ -416,11 +382,8 @@ void rewriteASTJoinsToGlobal(ASTPtr & query) return; } - /// `RIGHT` and `FULL` preserve rows from the broadcast side. Running them on every worker - /// emits those rows once per worker. They stay on the initiator, as they do with the analyzer. - /// A preserved-side join also keeps the rest of the join tree there: an earlier `INNER`/`LEFT` - /// cannot be shipped on its own once a later `RIGHT`/`FULL` must see the full left stream. - if (selectHasRightOrFullJoin(*select_query)) + /// `RIGHT`/`FULL` preserve broadcast rows, so the whole join stays on the initiator. + if (queryHasRightOrFullJoin(query)) return; if (auto tables = select_query->tables()) @@ -490,8 +453,7 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( } case ObjectStorageClusterJoinMode::GLOBAL: { - /// Old analyzer. `RIGHT`/`FULL` were left local by `rewriteASTJoinsToGlobal`. Ship only the - /// left scan; the initiator runs the join. `INNER`/`LEFT`/`CROSS` stay in `query_to_send`. + /// Old analyzer: ship the left scan only. `INNER`/`LEFT`/`CROSS` stay in `query_to_send`. if (!query_info.query_tree) { if (queryHasRightOrFullJoin(query_to_send)) @@ -635,11 +597,8 @@ void IStorageCluster::readFromRemoteInitiator( ClusterPtr remote_initiator_cluster, const String & remote_initiator_cluster_name) { - /// `allow`/`local` (and initiator-local `RIGHT`/`FULL`) rewrite only `query_to_send`. - /// Clone that AST before `convertToRemote` wraps its table function in `remote()`. - /// `StorageDistributed::read` must analyze the stripped query. The original `query_info` - /// still contains the JOIN, and replacing only the left table there sends the right table - /// to a host that does not have it. + /// 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) @@ -653,7 +612,6 @@ void IStorageCluster::readFromRemoteInitiator( if (query_for_remote) { modified_query_info.query = query_for_remote; - /// The syntax result describes the original JOIN. Shard pruning must not reuse it. modified_query_info.syntax_analyzer_result.reset(); if (storage_and_context.context->getSettingsRef()[Setting::allow_experimental_analyzer]) @@ -672,9 +630,7 @@ void IStorageCluster::readFromRemoteInitiator( modified_query_info.table_expression = extractLeftTableExpression(query_node->getJoinTreeNodeTyped()); } else - { modified_query_info.query_tree.reset(); - } } auto new_storage_snapshot = storage_and_context.storage->getStorageSnapshot(storage_snapshot->metadata, storage_and_context.context); @@ -1038,18 +994,9 @@ IStorageCluster::QueryTreeInfo IStorageCluster::getQueryTreeInfo(QueryTreeNodePt if (query_node.hasWhere()) collector_where.visit(query_node.getWhere()); - // 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. - // `SELECT 1 FROM initiator_only_dim` has a table and no column from it, so also look at - // table expressions. - CollectForeignTableExpressionsVisitor foreign_tables(left_table_expression, context); - if (query_node.hasPrewhere()) - foreign_tables.visit(query_node.getPrewhere()); - if (query_node.hasWhere()) - foreign_tables.visit(query_node.getWhere()); - - if (!collector_where.getColumns().empty() || foreign_tables.hasForeignTable()) + // `WHERE` must be rewritten on the remote if it uses another table, including + // `WHERE x IN (SELECT 1 FROM local_table)`, which has no column from that table. + if (!collector_where.getColumns().empty() || collector_where.hasForeignTable()) info.has_local_columns_in_where = true; } @@ -1120,32 +1067,30 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( CollectUsedColumnsForSourceVisitor collector(left_table_expression, context); collector.visit(modified_query_tree); - /// A left-only predicate applied before a `RIGHT` or `FULL` join changes which right rows are - /// unmatched. The initiator still has the original predicate, so the shard scan must not. - const bool preserves_right_side = queryHasRightOrFullJoin(query_to_send); - if (!preserves_right_side) + /// A left-only predicate under `RIGHT`/`FULL` changes which right rows are unmatched. + if (queryHasRightOrFullJoin(query_to_send)) { - 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); + query_node.getPrewhere() = {}; + 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(); - if (!preserves_right_side) - { - rewritten_query_node.getPrewhere() = query_node.getPrewhere(); - rewritten_query_node.getWhere() = query_node.getWhere(); - } + rewritten_query_node.getPrewhere() = query_node.getPrewhere(); + rewritten_query_node.getWhere() = query_node.getWhere(); query_to_send = queryNodeToDistributedSelectQuery(rewritten_query_tree); return; @@ -1172,9 +1117,8 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( throw Exception(ErrorCodes::LOGICAL_ERROR, "Query is not analyzed: no syntax analyzer result"); TreeRewriterResult rewriter_result = *query_info.syntax_analyzer_result; - /// `removeJoin` drops the join, so the kind has to be read first. A left-only `WHERE` kept - /// below a `RIGHT` or `FULL` join changes which right rows are unmatched. - const bool preserves_right_side = selectHasRightOrFullJoin(*select_query); + /// 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); if (!removeJoin(*select_query, rewriter_result, context)) throw Exception(ErrorCodes::LOGICAL_ERROR, "Failed to strip JOIN from query sent to cluster nodes"); @@ -1217,8 +1161,7 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( select_query->setExpression(ASTSelectQuery::Expression::WHERE, {}); select_query->setExpression(ASTSelectQuery::Expression::GROUP_BY, {}); select_query->group_by_all = false; - /// Same flags `removeJoin` clears. A leftover `WITH TOTALS`/`ROLLUP`/`CUBE`/`GROUPING SETS` - /// with no `GROUP BY` makes the re-analyzed shard query fail. + /// `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; @@ -1244,8 +1187,7 @@ QueryProcessingStage::Enum IStorageCluster::getQueryProcessingStage( { auto object_storage_cluster_join_mode = context->getSettingsRef()[Setting::object_storage_cluster_join_mode]; const auto join_info = getQueryJoinInfo(query_info, context); - /// `global` ships `INNER`/`LEFT`/`CROSS`. `RIGHT`/`FULL` preserve broadcast rows, so even in - /// `global` the initiator fetches left columns and joins locally. + /// `global` still fetches columns for `RIGHT`/`FULL`: those joins stay on the initiator. const bool keep_join_on_initiator = object_storage_cluster_join_mode != ObjectStorageClusterJoinMode::GLOBAL || queryHasRightOrFullJoin(query_info.query); From c6d4d54435c1a0443a2eb7ae17958c7881bd7435 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 20 Oct 2026 11:16:30 +0200 Subject: [PATCH 08/14] Keep `PREWHERE` on the left scan of an initiator-local `RIGHT` or `FULL` join. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 18 ++++++------- ...luster_right_join_prewhere_stays.reference | 2 ++ ...1_s3_cluster_right_join_prewhere_stays.sql | 25 +++++++++++++++++++ 3 files changed, 34 insertions(+), 11 deletions(-) create mode 100644 tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.reference create mode 100644 tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 3405246737940..4144155abeb30 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -1067,12 +1067,10 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( CollectUsedColumnsForSourceVisitor collector(left_table_expression, context); collector.visit(modified_query_tree); - /// A left-only predicate under `RIGHT`/`FULL` changes which right rows are unmatched. + /// 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.getPrewhere() = {}; query_node.getWhere() = {}; - } if (query_node.getPrewhere()) removeExpressionsThatDoNotDependOnTableIdentifiers(query_node.getPrewhere(), left_table_expression, context); @@ -1130,14 +1128,12 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( select_query->group_by_all = false; select_query->order_by_all = false; - /// `removeJoin` keeps left-table WHERE, including `GLOBAL IN (_subqueryN)`, which remotes cannot resolve. - if (preserves_right_side - || astContainsInTableIdentifier(select_query->where()) || astContainsInTableIdentifier(select_query->prewhere()) - || astContainsSubquery(select_query->where()) || astContainsSubquery(select_query->prewhere())) - { - select_query->setExpression(ASTSelectQuery::Expression::PREWHERE, {}); + /// `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) { diff --git a/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.reference b/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.reference new file mode 100644 index 0000000000000..59a33abe0e25d --- /dev/null +++ b/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.reference @@ -0,0 +1,2 @@ +1 L a +0 b diff --git a/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql b/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql new file mode 100644 index 0000000000000..807e2d4c649aa --- /dev/null +++ b/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql @@ -0,0 +1,25 @@ +-- Tags: no-fasttest +-- no-fasttest: uses the S3 mock server on localhost:11111 + +-- `PREWHERE` filters the left table before a `RIGHT` join. Dropping it from the shard +-- scan keeps the other left row, so the matched right row is emitted twice. + +CREATE TEMPORARY TABLE t_dim_05391 (k UInt64, name String); +INSERT INTO t_dim_05391 VALUES (1, 'a'), (2, 'b'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05391_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') +SELECT k, v FROM +( + SELECT 1 AS k, 'L' AS v + UNION ALL SELECT 1, 'M' +) +SETTINGS s3_truncate_on_insert = 1; + +SELECT t1.k, t1.v, t2.name +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05391_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +PREWHERE t1.v = 'L' +RIGHT JOIN t_dim_05391 AS t2 ON t1.k = t2.k +ORDER BY t2.name, t1.v +SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'allow'; + +DROP TEMPORARY TABLE t_dim_05391; From 88e9e0cb58ef1e3f2a65e99dce85356a49b09b5f Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 20 Oct 2026 11:16:50 +0200 Subject: [PATCH 09/14] Strip an ordinary join that follows `ARRAY JOIN` from the shard query. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 12 +++++++++++ ..._s3_cluster_array_join_then_join.reference | 2 ++ .../05392_s3_cluster_array_join_then_join.sql | 21 +++++++++++++++++++ 3 files changed, 35 insertions(+) create mode 100644 tests/queries/0_stateless/05392_s3_cluster_array_join_then_join.reference create mode 100644 tests/queries/0_stateless/05392_s3_cluster_array_join_then_join.sql diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 4144155abeb30..ff6dba4342179 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -1117,6 +1117,18 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( 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"); diff --git a/tests/queries/0_stateless/05392_s3_cluster_array_join_then_join.reference b/tests/queries/0_stateless/05392_s3_cluster_array_join_then_join.reference new file mode 100644 index 0000000000000..f3a529a194b61 --- /dev/null +++ b/tests/queries/0_stateless/05392_s3_cluster_array_join_then_join.reference @@ -0,0 +1,2 @@ +x a +y a diff --git a/tests/queries/0_stateless/05392_s3_cluster_array_join_then_join.sql b/tests/queries/0_stateless/05392_s3_cluster_array_join_then_join.sql new file mode 100644 index 0000000000000..8657bc2fa51e1 --- /dev/null +++ b/tests/queries/0_stateless/05392_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_05392 (k UInt64, name String); +INSERT INTO t_dim_05392 VALUES (1, 'a'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05392_' || 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/05392_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, arr Array(String)') AS t1 +ARRAY JOIN arr AS elem +INNER JOIN t_dim_05392 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_05392; From 82e50dab2cc783919bad0f5b049c075c11fa8322 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 20 Oct 2026 11:17:48 +0200 Subject: [PATCH 10/14] Keep `PASTE` joins on the initiator. Co-authored-by: Cursor --- .../sql-reference/distribution-on-cluster.md | 2 +- src/Planner/PlannerJoinTree.cpp | 9 ++++-- src/Storages/IStorageCluster.cpp | 13 ++++++--- ...er_paste_join_stays_on_initiator.reference | 4 +++ ..._cluster_paste_join_stays_on_initiator.sql | 28 +++++++++++++++++++ 5 files changed, 49 insertions(+), 7 deletions(-) create mode 100644 tests/queries/0_stateless/05393_s3_cluster_paste_join_stays_on_initiator.reference create mode 100644 tests/queries/0_stateless/05393_s3_cluster_paste_join_stays_on_initiator.sql diff --git a/docs/en/sql-reference/distribution-on-cluster.md b/docs/en/sql-reference/distribution-on-cluster.md index 2cb2597fd78ce..98e54843529e6 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` the nodes read the left table and the join runs on the initiator. `local` is a deprecated alias of `allow`. 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` the nodes read the left table and the join runs on the initiator. `local` is a deprecated alias of `allow`. 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/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 75f160ddc5151..03977b68b9be9 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -3596,6 +3596,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; @@ -3627,6 +3628,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; @@ -3715,7 +3718,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 @@ -3728,7 +3732,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 ff6dba4342179..d0aa67012d410 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -350,8 +350,13 @@ bool queryHasRightOrFullJoin(const ASTPtr & query) for (const auto & child : tables->children) { const auto & element = child->as(); - if (element.table_join && isRightOrFull(element.table_join->as().kind)) - return true; + 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; } @@ -382,7 +387,7 @@ void rewriteASTJoinsToGlobal(ASTPtr & query) return; } - /// `RIGHT`/`FULL` preserve broadcast rows, so the whole join stays on the initiator. + /// `RIGHT`/`FULL` preserve broadcast rows, and `PASTE` pairs by position, so the join stays on the initiator. if (queryHasRightOrFullJoin(query)) return; @@ -1195,7 +1200,7 @@ QueryProcessingStage::Enum IStorageCluster::getQueryProcessingStage( { 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`: those joins stay on the initiator. + /// `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); diff --git a/tests/queries/0_stateless/05393_s3_cluster_paste_join_stays_on_initiator.reference b/tests/queries/0_stateless/05393_s3_cluster_paste_join_stays_on_initiator.reference new file mode 100644 index 0000000000000..60d24e9a0d1e1 --- /dev/null +++ b/tests/queries/0_stateless/05393_s3_cluster_paste_join_stays_on_initiator.reference @@ -0,0 +1,4 @@ +old +1 +analyzer +1 diff --git a/tests/queries/0_stateless/05393_s3_cluster_paste_join_stays_on_initiator.sql b/tests/queries/0_stateless/05393_s3_cluster_paste_join_stays_on_initiator.sql new file mode 100644 index 0000000000000..9b00ab15d8f43 --- /dev/null +++ b/tests/queries/0_stateless/05393_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_05393 (name String); +INSERT INTO t_dim_05393 VALUES ('a'); + +INSERT INTO FUNCTION s3('http://localhost:11111/test/05393_' || 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/05393_' || currentDatabase() || '_*.tsv', 'test', 'testtest', 'TSV', 'k UInt64') AS t1 +PASTE JOIN t_dim_05393 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/05393_' || currentDatabase() || '_*.tsv', 'test', 'testtest', 'TSV', 'k UInt64') AS t1 +PASTE JOIN t_dim_05393 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_05393; From ac820e248b649580c6488633d4e3eb2b3f1d9160 Mon Sep 17 00:00:00 2001 From: blau-ai Date: Tue, 20 Oct 2026 16:12:12 +0000 Subject: [PATCH 11/14] Move PREWHERE after the JOIN in 05391 to fix the syntax error PREWHERE is parsed after the table/JOIN section, so the query `FROM ... AS t1 PREWHERE ... RIGHT JOIN ...` failed with SYNTAX_ERROR at the RIGHT keyword (every run of 05391 failed in CI). Reorder to `FROM ... AS t1 RIGHT JOIN ... ON ... PREWHERE ...`, matching the PREWHERE-after-JOIN grammar; the intended behaviour and reference output are unchanged. Co-Authored-By: Claude Opus 4.8 --- .../0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql b/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql index 807e2d4c649aa..60fb240fe8b00 100644 --- a/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql +++ b/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql @@ -17,8 +17,8 @@ SETTINGS s3_truncate_on_insert = 1; SELECT t1.k, t1.v, t2.name FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05391_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 -PREWHERE t1.v = 'L' RIGHT JOIN t_dim_05391 AS t2 ON t1.k = t2.k +PREWHERE t1.v = 'L' ORDER BY t2.name, t1.v SETTINGS allow_experimental_analyzer = 0, object_storage_cluster_join_mode = 'allow'; From dbdae80d09afe8d85c5718a57694a25f429a70a7 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Tue, 20 Oct 2026 20:59:10 +0200 Subject: [PATCH 12/14] Do not wrap `query_info.query` in place when sending to a remote initiator With the old analyzer and `object_storage_cluster_join_mode='global'`, an `INNER` join leaves `query_to_send` pointing at `query_info.query`. `convertToRemote` then wrapped the left table function in `remote` in place, moving its alias inside, and `StorageDistributed::read` analyzed that query and threw `ALIAS_REQUIRED`. Pass `convertToRemote` a copy instead. Fixes `test_remote_initiator.py::test_remote_initiator_global_join_without_analyzer[s3]`. https://github.com/Altinity/ClickHouse/pull/2609#issuecomment-6295505075 https://altinity-build-artifacts.s3.amazonaws.com/json.html?PR=2609&sha=ac820e248b649580c6488633d4e3eb2b3f1d9160&name_0=PR&name_1=Integration%20tests%20(amd_asan_ubsan%2C%20targeted) --- src/Storages/IStorageCluster.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index d0aa67012d410..e0b8c42d32280 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -609,7 +609,9 @@ void IStorageCluster::readFromRemoteInitiator( if (query_for_remote) stripObjectStorageRemoteInitiatorSettings(query_for_remote); - auto storage_and_context = convertToRemote(remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); + /// `convertToRemote` wraps the left table function in place and moves its alias inside `remote`. + /// `query_to_send` may be `query_info.query`, which `StorageDistributed` analyzes with its JOINs. + auto storage_and_context = convertToRemote(remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send->clone()); auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); auto modified_query_info = query_info; modified_query_info.cluster = src_distributed->getCluster(); From 07395ad6f2481e900a70573dd0ebe27f35a46680 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Tue, 20 Oct 2026 20:59:10 +0200 Subject: [PATCH 13/14] Use `Parquet` in 05391 so the S3 storage supports `PREWHERE` Object storage supports `PREWHERE` only for formats that do. With `TSV` the query failed with `Storage S3 doesn't support PREWHERE` (`ILLEGAL_PREWHERE`) before reaching the join rewrite. https://github.com/Altinity/ClickHouse/pull/2609#issuecomment-6295505075 https://altinity-build-artifacts.s3.amazonaws.com/json.html?PR=2609&sha=ac820e248b649580c6488633d4e3eb2b3f1d9160&name_0=PR&name_1=Stateless%20tests%20(amd_asan_ubsan%2C%20distributed%20plan%2C%20parallel%2C%20selected%20tests) --- .../05391_s3_cluster_right_join_prewhere_stays.sql | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql b/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql index 60fb240fe8b00..f3aa5c0a3799e 100644 --- a/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql +++ b/tests/queries/0_stateless/05391_s3_cluster_right_join_prewhere_stays.sql @@ -7,7 +7,7 @@ CREATE TEMPORARY TABLE t_dim_05391 (k UInt64, name String); INSERT INTO t_dim_05391 VALUES (1, 'a'), (2, 'b'); -INSERT INTO FUNCTION s3('http://localhost:11111/test/05391_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') +INSERT INTO FUNCTION s3('http://localhost:11111/test/05391_' || currentDatabase() || '.parquet', 'test', 'testtest', 'Parquet', 'k UInt64, v String') SELECT k, v FROM ( SELECT 1 AS k, 'L' AS v @@ -16,7 +16,7 @@ SELECT k, v FROM SETTINGS s3_truncate_on_insert = 1; SELECT t1.k, t1.v, t2.name -FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05391_' || currentDatabase() || '.tsv', 'test', 'testtest', 'TSV', 'k UInt64, v String') AS t1 +FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/05391_' || currentDatabase() || '.parquet', 'test', 'testtest', 'Parquet', 'k UInt64, v String') AS t1 RIGHT JOIN t_dim_05391 AS t2 ON t1.k = t2.k PREWHERE t1.v = 'L' ORDER BY t2.name, t1.v From 2a081d60ca42a6bd8bdf9464cbd69dfa9aac08d6 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:26:47 +0200 Subject: [PATCH 14/14] Pin `serialize_query_plan = 0` in the analyzer query of 05387 With `serialize_query_plan = 1` and `prefer_localhost_replica = 0`, the remote initiator receives a serialized plan whose `s3Cluster` arguments are formatted with the secret hidden, so the S3 read fails with 403. This is not specific to the remote initiator: a plain `remote(..., s3(...))` with credentials fails the same way. https://github.com/Altinity/ClickHouse/pull/2609#issuecomment-6314330017 https://altinity-build-artifacts.s3.amazonaws.com/json.html?PR=2609&sha=07395ad6f2481e900a70573dd0ebe27f35a46680&name_0=PR&name_1=Stateless%20tests%20(amd_asan_ubsan%2C%20distributed%20plan%2C%20parallel%2C%20selected%20tests) https://github.com/Altinity/ClickHouse/pull/2609 --- .../05387_s3_cluster_remote_initiator_keeps_stripped_join.sql | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 index b3ed8624ea7ab..615156e6c8882 100644 --- 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 @@ -37,6 +37,8 @@ SETTINGS allow_experimental_analyzer = 1, object_storage_cluster_join_mode = 'allow', object_storage_remote_initiator = 1, - object_storage_remote_initiator_cluster = 'test_shard_localhost'; + object_storage_remote_initiator_cluster = 'test_shard_localhost', + -- A serialized plan carries the `s3Cluster` arguments with the secret hidden. + serialize_query_plan = 0; DROP TEMPORARY TABLE t_dim_05387;