diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index bd4609af5963..6811076923fd 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -2130,14 +2130,14 @@ ClickHouse applies this setting when the query contains the product of object st Restrictions: -- Only applied for JOIN subqueries. -- Only if the FROM section uses a object storage cluster function or table. +- 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: -- `local` — Replaces the database and table in the subquery with local ones for the destination server (shard), leaving the normal `IN`/`JOIN.` -- `global` — Replaces the `IN`/`JOIN` query with `GLOBAL IN`/`GLOBAL JOIN.` Right table executes first and is added to the secondary query as temporay table. -- `allow` — Default value. Allows the use of these types of subqueries. +- `global` — Replaces the `IN`/`JOIN` query with `GLOBAL IN`/`GLOBAL JOIN`. Right table executes first and is added to the secondary query as a temporary table. +- `allow` — Default value. Reads the left object-storage table on cluster nodes and executes `JOIN` / local `IN` on the initiator, so the right table does not need to exist on remote nodes. +- `local` — Deprecated legacy alias of `allow`. )", 0) \ \ DECLARE(UInt64, max_concurrent_queries_for_all_users, 0, R"( diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h index b4df59f05af0..3240e2f88b1c 100644 --- a/src/Core/SettingsEnums.h +++ b/src/Core/SettingsEnums.h @@ -168,9 +168,9 @@ DECLARE_SETTING_ENUM(DistributedProductMode) /// The setting for executing object storage cluster function or table JOIN sections. enum class ObjectStorageClusterJoinMode : uint8_t { - LOCAL, /// Convert to local query + LOCAL, /// Legacy alias of ALLOW: initiator-local join GLOBAL, /// Convert to global query - ALLOW /// Enable + ALLOW /// Initiator-local join when the right table is not on remote nodes }; DECLARE_SETTING_ENUM(ObjectStorageClusterJoinMode) diff --git a/src/Interpreters/InterpreterSelectQuery.cpp b/src/Interpreters/InterpreterSelectQuery.cpp index 26937355446f..de2df6742294 100644 --- a/src/Interpreters/InterpreterSelectQuery.cpp +++ b/src/Interpreters/InterpreterSelectQuery.cpp @@ -215,6 +215,8 @@ namespace Setting extern const SettingsBool enable_lazy_columns_replication; extern const SettingsBool serialize_string_in_memory_with_zero_byte; extern const SettingsBool use_hive_partitioning; + extern const SettingsBool allow_experimental_analyzer; + extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; } namespace ServerSetting @@ -758,6 +760,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 3012c7bff735..dabba25c948c 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -30,14 +30,20 @@ #include #include #include -#include -#include #include #include -#include -#include #include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include @@ -112,49 +118,6 @@ void ReadFromCluster::createExtension(const ActionsDAG::Node * predicate) 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 */ @@ -205,7 +168,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 @@ -218,7 +291,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 @@ -241,94 +314,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) - { - auto join_node = query_node.getJoinTree(); - query_tree_distributed = join_node->as()->getLeftTableExpression()->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, 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); - } - - query_node.getOrderByNode() = std::make_shared(); - query_node.getGroupByNode() = std::make_shared(); - - if (query_tree_distributed) - { - // Left only table function to send on cluster nodes - modified_query_tree = modified_query_tree->cloneAndReplace(query_node.getJoinTree(), query_tree_distributed); - } - - query_to_send = queryNodeToDistributedSelectQuery(modified_query_tree); - } + 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(); @@ -347,8 +350,6 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( return; } - case ObjectStorageClusterJoinMode::ALLOW: // Do nothing special - return; } } @@ -462,9 +463,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, @@ -657,23 +660,22 @@ IStorageCluster::QueryTreeInfo IStorageCluster::getQueryTreeInfo(QueryTreeNodePt QueryTreeInfo info; auto & query_node = query_tree->as(); - if (auto join_node = query_node.getJoinTree()) - { - if (join_node->getNodeType() == QueryTreeNodeType::JOIN) - info.has_join = true; - else if (join_node->getNodeType() == QueryTreeNodeType::CROSS_JOIN) - info.has_cross_join = true; - } + auto join_tree = query_node.getJoinTree(); + if (!join_tree) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); - 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(join_tree); + if (!left_table_expression) throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); if (query_node.hasWhere() || query_node.hasPrewhere()) { - CollectUsedColumnsForSourceVisitor collector_where(table_function_node, context, true); + CollectUsedColumnsForSourceVisitor collector_where(left_table_expression, context, true); if (query_node.hasPrewhere()) collector_where.visit(query_node.getPrewhere()); if (query_node.hasWhere()) @@ -689,23 +691,164 @@ 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; +} + +IStorageCluster::QueryTreeInfo IStorageCluster::getQueryJoinInfoFromAST(const ASTPtr & query) +{ + QueryTreeInfo info; + if (!query) + return info; + + const ASTSelectQuery * select_query = query->as(); + if (!select_query) + { + 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); - if (object_storage_cluster_join_mode != ObjectStorageClusterJoinMode::ALLOW) + return getQueryJoinInfoFromAST(query_info.query); +} + +void IStorageCluster::rewriteQueryForInitiatorLocalJoin( + ASTPtr & query_to_send, + const SelectQueryInfo & query_info, + const QueryTreeInfo & info, + const ContextPtr & context) +{ + /// Analyzer path: reuse extractLeftTableExpression + buildQueryToReadColumnsFromTableExpression + /// (same helpers the planner uses when wrapping IStorageCluster in a subquery). + if (query_info.query_tree) { - 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 modified_query_tree = query_info.query_tree->clone(); + auto & query_node = modified_query_tree->as(); + auto join_tree = query_node.getJoinTree(); + if (!join_tree) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); + + auto left_table_expression = extractLeftTableExpression(join_tree); + if (!left_table_expression) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't find table or table function node"); + + CollectUsedColumnsForSourceVisitor collector(left_table_expression, context); + collector.visit(modified_query_tree); + + if (query_node.getPrewhere()) + removeExpressionsThatDoNotDependOnTableIdentifiers(query_node.getPrewhere(), left_table_expression, context); + if (query_node.getWhere()) + removeExpressionsThatDoNotDependOnTableIdentifiers(query_node.getWhere(), left_table_expression, context); + + auto rewritten_query_tree = buildQueryToReadColumnsFromTableExpression( + collector.getColumns(), left_table_expression, context); + auto & rewritten_query_node = rewritten_query_tree->as(); + rewritten_query_node.getPrewhere() = query_node.getPrewhere(); + rewritten_query_node.getWhere() = query_node.getWhere(); + + query_to_send = queryNodeToDistributedSelectQuery(rewritten_query_tree); + return; + } + + /// Old interpreter: reuse removeJoin used by StorageDistributed / StorageMerge / StorageWindowView. + if (!query_to_send) + return; - if (object_storage_cluster_join_mode == ObjectStorageClusterJoinMode::LOCAL) + query_to_send = query_to_send->clone(); + if (auto * union_query = query_to_send->as()) + { + if (union_query->list_of_selects && union_query->list_of_selects->children.size() == 1) + query_to_send = union_query->list_of_selects->children[0]->clone(); + } + + auto * select_query = query_to_send->as(); + if (!select_query) + return; + + if (select_query->hasJoin()) + { + if (!query_info.syntax_analyzer_result) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Query is not analyzed: no syntax analyzer result"); + + TreeRewriterResult rewriter_result = *query_info.syntax_analyzer_result; + if (!removeJoin(*select_query, rewriter_result, context)) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Failed to strip JOIN from query sent to cluster nodes"); + + /// `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; } /// Initiator executes query on remote node. diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 9613f9549562..fa4131679e94 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -63,13 +63,19 @@ class IStorageCluster : public IStorage const String & getOriginalClusterName() const { return cluster_name; } virtual String getClusterName(ContextPtr /* context */) const { return getOriginalClusterName(); } + /// Old interpreter: rewrite JOIN / IN to GLOBAL JOIN / GLOBAL IN so GlobalSubqueriesVisitor can broadcast right tables. + static void rewriteASTForGlobalJoin(ASTPtr & query); + protected: virtual void updateQueryToSendIfNeeded( ASTPtr & /*query*/, const StorageSnapshotPtr & /*storage_snapshot*/, const ContextPtr & /*context*/, bool /*make_cluster_function*/) {} - void updateQueryWithJoinToSendIfNeeded(ASTPtr & query_to_send, SelectQueryInfo query_info, const ContextPtr & context); + void updateQueryWithJoinToSendIfNeeded( + ASTPtr & query_to_send, + SelectQueryInfo query_info, + const ContextPtr & context); virtual void updateConfigurationIfNeeded(ContextPtr /* context */) {} @@ -127,6 +133,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 6c949cc73330..0181cf2b1c27 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -1432,7 +1432,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 00978fb7231c..cebaea88bb9a 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -1077,8 +1077,9 @@ def test_remote_no_hedged(started_cluster): assert TSV(pure_s3) == TSV(s3_distributed) -@pytest.mark.parametrize("join_mode", ["local", "global"]) -def test_joins(started_cluster, join_mode): +@pytest.mark.parametrize("join_mode", ["allow", "local", "global"]) +@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'. @@ -1112,7 +1113,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}; """ ) @@ -1135,7 +1136,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}; """ ) @@ -1153,7 +1154,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}; """ ) @@ -1172,7 +1173,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}; """ ) @@ -1190,7 +1191,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}; """ ) @@ -1206,7 +1207,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())) @@ -1221,7 +1222,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())) @@ -1238,7 +1239,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" @@ -1254,7 +1255,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 82e9d6c3c572..ab2d34311403 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 763836d21f60..77891e65fb44 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"