diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index d73c8fb5a46f..b4a51e20505d 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -5549,6 +5549,14 @@ Serialize query plan for distributed processing )", 0) \ DECLARE(Bool, correlated_subqueries_substitute_equivalent_expressions, true, R"( Use filter expressions to inference equivalent expressions and substitute them instead of creating a CROSS JOIN. +)", 0) \ + DECLARE(DecorrelationJoinKind, correlated_subqueries_default_join_kind, DecorrelationJoinKind::RIGHT, R"( +Controls the kind of joins in the decorrelated query plan. The default value is `right`, which means that decorrelated plan will contain RIGHT JOINs with subquery input on the right side. + +Possible values: + +- `left` - Decorrelation process will produce LEFT JOINs and input table will appear on the left side. +- `right` - Decorrelation process will produce RIGHT JOINs and input table will appear on the right side. )", 0) \ \ DECLARE(UInt64, regexp_max_matches_per_row, 1000, R"( diff --git a/src/Core/Settings.h b/src/Core/Settings.h index 7f284dd78b4e..97b7bfa0e07f 100644 --- a/src/Core/Settings.h +++ b/src/Core/Settings.h @@ -107,7 +107,8 @@ class WriteBuffer; M(CLASS_NAME, UInt64Auto) \ M(CLASS_NAME, URI) \ M(CLASS_NAME, VectorSearchFilterStrategy) \ - M(CLASS_NAME, GeoToH3ArgumentOrder) + M(CLASS_NAME, GeoToH3ArgumentOrder) \ + M(CLASS_NAME, DecorrelationJoinKind) \ COMMON_SETTINGS_SUPPORTED_TYPES(Settings, DECLARE_SETTING_TRAIT) diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 5bd62541ff60..cf4ec9b4db47 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -39,6 +39,10 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() /// controls new feature and it's 'true' by default, use 'false' as previous_value). /// It's used to implement `compatibility` setting (see https://github.com/ClickHouse/ClickHouse/issues/35972) /// Note: please check if the key already exists to prevent duplicate entries. + addSettingsChanges(settings_changes_history, "25.8.28", + { + {"correlated_subqueries_default_join_kind", "left", "right", "New setting. Default join kind for decorrelated query plan."}, + }); addSettingsChanges(settings_changes_history, "25.8.16.10002", { {"allow_local_data_lakes", false, false, "New setting to guard local data lake engines and table functions"}, diff --git a/src/Core/SettingsEnums.cpp b/src/Core/SettingsEnums.cpp index 7a586f51168e..e740b6e726e7 100644 --- a/src/Core/SettingsEnums.cpp +++ b/src/Core/SettingsEnums.cpp @@ -356,6 +356,12 @@ IMPLEMENT_SETTING_ENUM( {"local", SearchOrphanedPartsDisks::LOCAL}, {"none", SearchOrphanedPartsDisks::NONE}}) +IMPLEMENT_SETTING_ENUM( + DecorrelationJoinKind, + ErrorCodes::BAD_ARGUMENTS, + {{"left", DecorrelationJoinKind::LEFT}, + {"right", DecorrelationJoinKind::RIGHT}}) + IMPLEMENT_SETTING_ENUM( IcebergMetadataLogLevel, ErrorCodes::BAD_ARGUMENTS, diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h index 18873a0790ae..532340e281e4 100644 --- a/src/Core/SettingsEnums.h +++ b/src/Core/SettingsEnums.h @@ -459,6 +459,14 @@ enum class SearchOrphanedPartsDisks : uint8_t DECLARE_SETTING_ENUM(SearchOrphanedPartsDisks) +enum class DecorrelationJoinKind : uint8_t +{ + LEFT = 0, + RIGHT, +}; + +DECLARE_SETTING_ENUM(DecorrelationJoinKind) + enum class IcebergMetadataLogLevel : uint8_t { None = 0, diff --git a/src/Planner/PlannerCorrelatedSubqueries.cpp b/src/Planner/PlannerCorrelatedSubqueries.cpp index 42055989829f..090b707c73af 100644 --- a/src/Planner/PlannerCorrelatedSubqueries.cpp +++ b/src/Planner/PlannerCorrelatedSubqueries.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include @@ -27,12 +28,18 @@ #include #include +#include +#include #include #include #include #include #include +#include +#include +#include + #include #include #include @@ -53,8 +60,11 @@ extern const int LOGICAL_ERROR; namespace Setting { -extern const SettingsBool join_use_nulls; extern const SettingsBool correlated_subqueries_substitute_equivalent_expressions; +extern const SettingsDecorrelationJoinKind correlated_subqueries_default_join_kind; +extern const SettingsBool join_use_nulls; +extern const SettingsMaxThreads max_threads; +extern const SettingsNonZeroUInt64 max_block_size; } @@ -185,36 +195,6 @@ struct DecorrelationContext std::vector equivalence_class_stack; }; -namespace -{ - -void projectCorrelatedColumns( - QueryPlan & lhs_plan, - const ColumnIdentifiers & correlated_column_identifiers) -{ - ActionsDAG project_only_correlated_columns_actions; - - NameSet correlated_column_identifiers_set(correlated_column_identifiers.begin(), correlated_column_identifiers.end()); - - const auto & lhs_plan_header = lhs_plan.getCurrentHeader(); - - auto & outputs = project_only_correlated_columns_actions.getOutputs(); - for (const auto & column : lhs_plan_header->getColumnsWithTypeAndName()) - { - const auto * input_node = &project_only_correlated_columns_actions.addInput(column); - if (correlated_column_identifiers_set.contains(column.name)) - { - outputs.push_back(input_node); - } - } - - lhs_plan.addStep(std::make_unique( - lhs_plan_header, - std::move(project_only_correlated_columns_actions))); -} - -} - /// Correlated subquery is represented by implicit dependent join operator. /// This function builds a query plan to evaluate correlated subquery by /// pushing dependent join down and replacing it with CROSS JOIN. @@ -225,12 +205,12 @@ QueryPlan decorrelateQueryPlan( { if (!context.correlated_plan_steps[node]) { + /// The rest of the query plan doesn't use any correlated columns. const auto & settings = context.planner_context->getQueryContext()->getSettingsRef(); - auto decorrelated_plan_header = node->step->getOutputHeader(); - if (settings[Setting::correlated_subqueries_substitute_equivalent_expressions]) { + const auto & decorrelated_plan_header = node->step->getOutputHeader(); ActionsDAG dag(decorrelated_plan_header->getNamesAndTypesList()); auto & outputs = dag.getOutputs(); @@ -238,6 +218,7 @@ QueryPlan decorrelateQueryPlan( for (const auto * output : outputs) decorrelated_nodes_names[output->result_name] = output; + /// Find possible renamings for all correlated columns std::vector> expression_renamings; for (const auto & correlated_column_identifier : context.correlated_subquery.correlated_column_identifiers) { @@ -256,6 +237,8 @@ QueryPlan decorrelateQueryPlan( } } + /// If all columns from outer query have equivalent expressions in the current subplan, + /// we can safely replace them and avoid introduction of CROSS JOIN. if (context.correlated_subquery.correlated_column_identifiers.size() == expression_renamings.size()) { for (const auto & [from, to] : expression_renamings) @@ -269,29 +252,44 @@ QueryPlan decorrelateQueryPlan( } } - /// The rest of the query plan doesn't use any correlated columns. - auto lhs_plan = context.query_plan.clone(); + QueryPlan lhs_plan = context.correlated_query_plan.extractSubplan(node); + QueryPlan rhs_plan; + + auto default_join_kind = settings[Setting::correlated_subqueries_default_join_kind]; + context.query_plan.addStep(std::make_unique(context.query_plan.getCurrentHeader())); + + auto buffer_header = std::make_shared(); + for (const auto & column : context.correlated_subquery.correlated_column_identifiers) + buffer_header->insert(context.query_plan.getCurrentHeader()->getByName(column)); - projectCorrelatedColumns(lhs_plan, context.correlated_subquery.correlated_column_identifiers); + rhs_plan.addStep(std::make_unique( + buffer_header, + context.query_plan.getRootNode(), + context.correlated_subquery.correlated_column_identifiers)); + rhs_plan.getRootNode()->step->setStepDescription("Input for " + context.correlated_subquery.action_node_name); + + if (default_join_kind == DecorrelationJoinKind::LEFT) + std::swap(lhs_plan, rhs_plan); auto lhs_plan_header = lhs_plan.getCurrentHeader(); + auto rhs_plan_header = rhs_plan.getCurrentHeader(); ColumnsWithTypeAndName output_columns_and_types; output_columns_and_types.insert_range(output_columns_and_types.cend(), lhs_plan_header->getColumnsWithTypeAndName()); - output_columns_and_types.insert_range(output_columns_and_types.cend(), decorrelated_plan_header->getColumnsWithTypeAndName()); + output_columns_and_types.insert_range(output_columns_and_types.cend(), rhs_plan_header->getColumnsWithTypeAndName()); JoinExpressionActions join_expression_actions( lhs_plan_header->getColumnsWithTypeAndName(), - decorrelated_plan_header->getColumnsWithTypeAndName(), + rhs_plan_header->getColumnsWithTypeAndName(), output_columns_and_types); Names output_columns; output_columns.insert_range(output_columns.cend(), lhs_plan_header->getNames()); - output_columns.insert_range(output_columns.cend(), node->step->getOutputHeader()->getNames()); + output_columns.insert_range(output_columns.cend(), rhs_plan_header->getNames()); auto decorrelated_join = std::make_unique( lhs_plan_header, - /*right_header_=*/decorrelated_plan_header, + /*right_header_=*/rhs_plan_header, JoinInfo{ .expression = {}, .kind = JoinKind::Cross, @@ -305,12 +303,12 @@ QueryPlan decorrelateQueryPlan( SortingStep::Settings(settings)); decorrelated_join->setStepDescription("JOIN to evaluate correlated expression"); - /// Add CROSS JOIN + /// Add CROSS JOIN to combine data streams from left and right plans. QueryPlan result_plan; std::vector plans; plans.emplace_back(std::make_unique(std::move(lhs_plan))); - plans.emplace_back(std::make_unique(context.correlated_query_plan.extractSubplan(node))); + plans.emplace_back(std::make_unique(std::move(rhs_plan))); result_plan.unitePlans(std::move(decorrelated_join), {std::move(plans)}); @@ -523,34 +521,52 @@ void buildExistsResultExpression( QueryPlan buildLogicalJoin( const PlannerContextPtr & planner_context, - QueryPlan left_plan, - QueryPlan right_plan, + QueryPlan input_stream_plan, + QueryPlan decorrelated_plan, const CorrelatedSubquery & correlated_subquery ) { - const auto & lhs_plan_header = left_plan.getCurrentHeader(); - const auto & rhs_plan_header = right_plan.getCurrentHeader(); + auto lhs_plan_header = decorrelated_plan.getCurrentHeader(); + auto rhs_plan_header = input_stream_plan.getCurrentHeader(); - ColumnsWithTypeAndName output_columns_and_types; - output_columns_and_types.insert_range(output_columns_and_types.cend(), lhs_plan_header->getColumnsWithTypeAndName()); - output_columns_and_types.emplace_back(rhs_plan_header->getByName(correlated_subquery.action_node_name)); + using ColumnNameGetter = std::function; + ColumnNameGetter get_lhs_column_name = [&](const String & column_name) -> String { + return fmt::format("{}.{}", correlated_subquery.action_node_name, column_name); + }; + ColumnNameGetter get_rhs_column_name = [&](const String & column_name) -> String { + return column_name; + }; - JoinExpressionActions join_expression_actions( - lhs_plan_header->getColumnsWithTypeAndName(), - rhs_plan_header->getColumnsWithTypeAndName(), - output_columns_and_types); + auto lhs_plan = std::move(decorrelated_plan); + auto rhs_plan = std::move(input_stream_plan); Names output_columns; - output_columns.insert_range(output_columns.cend(), lhs_plan_header->getNames()); + output_columns.insert_range(output_columns.cend(), rhs_plan_header->getNames()); output_columns.push_back(correlated_subquery.action_node_name); + ColumnsWithTypeAndName output_columns_and_types; + output_columns_and_types.insert_range(output_columns_and_types.cend(), rhs_plan_header->getColumnsWithTypeAndName()); + output_columns_and_types.emplace_back(lhs_plan_header->getByName(correlated_subquery.action_node_name)); + const auto & settings = planner_context->getQueryContext()->getSettingsRef(); + if (settings[Setting::correlated_subqueries_default_join_kind] == DecorrelationJoinKind::LEFT) + { + std::swap(lhs_plan, rhs_plan); + std::swap(lhs_plan_header, rhs_plan_header); + std::swap(get_lhs_column_name, get_rhs_column_name); + } + + JoinExpressionActions join_expression_actions( + lhs_plan_header->getColumnsWithTypeAndName(), + rhs_plan_header->getColumnsWithTypeAndName(), + output_columns_and_types); + std::vector predicates; for (const auto & column_name : correlated_subquery.correlated_column_identifiers) { - const auto * left_node = &join_expression_actions.left_pre_join_actions->findInOutputs(column_name); - const auto * right_node = &join_expression_actions.right_pre_join_actions->findInOutputs(fmt::format("{}.{}", correlated_subquery.action_node_name, column_name)); + const auto * left_node = &join_expression_actions.left_pre_join_actions->findInOutputs(get_lhs_column_name(column_name)); + const auto * right_node = &join_expression_actions.right_pre_join_actions->findInOutputs(get_rhs_column_name(column_name)); JoinPredicate predicate{ .left_node = JoinActionRef(left_node, join_expression_actions.left_pre_join_actions.get()), @@ -561,7 +577,9 @@ QueryPlan buildLogicalJoin( predicates.emplace_back(std::move(predicate)); } - /// Add LEFT OUTER JOIN + auto join_kind_to_use = settings[Setting::correlated_subqueries_default_join_kind] == DecorrelationJoinKind::RIGHT ? JoinKind::Right : JoinKind::Left; + + /// Add ANY OUTER JOIN auto result_join = std::make_unique( lhs_plan_header, rhs_plan_header, @@ -575,7 +593,7 @@ QueryPlan buildLogicalJoin( }, .disjunctive_conditions = {} }, - .kind = JoinKind::Left, + .kind = join_kind_to_use, .strictness = JoinStrictness::Any, .locality = JoinLocality::Local }, @@ -589,8 +607,8 @@ QueryPlan buildLogicalJoin( QueryPlan result_plan; std::vector plans; - plans.emplace_back(std::make_unique(std::move(left_plan))); - plans.emplace_back(std::make_unique(std::move(right_plan))); + plans.emplace_back(std::make_unique(std::move(lhs_plan))); + plans.emplace_back(std::make_unique(std::move(rhs_plan))); result_plan.unitePlans(std::move(result_join), {std::move(plans)}); return result_plan; diff --git a/src/Planner/Utils.cpp b/src/Planner/Utils.cpp index 783e946240bf..5d2653f1ad71 100644 --- a/src/Planner/Utils.cpp +++ b/src/Planner/Utils.cpp @@ -670,4 +670,25 @@ bool optimizePlanForExists(QueryPlan & query_plan) return false; } +QueryPlanStepPtr projectOnlyUsedColumns( + const SharedHeader & stream_header, + const ColumnIdentifiers & used_column_identifiers) +{ + ActionsDAG project_only_used_columns_actions; + + NameSet used_column_identifiers_set(used_column_identifiers.begin(), used_column_identifiers.end()); + + auto & outputs = project_only_used_columns_actions.getOutputs(); + for (const auto & column : stream_header->getColumnsWithTypeAndName()) + { + const auto * input_node = &project_only_used_columns_actions.addInput(column); + if (used_column_identifiers_set.contains(column.name)) + outputs.push_back(input_node); + } + + auto step = std::make_unique(stream_header, std::move(project_only_used_columns_actions)); + step->setStepDescription("Project only used columns"); + return step; +} + } diff --git a/src/Planner/Utils.h b/src/Planner/Utils.h index cd91e7abe020..4a0c013baa54 100644 --- a/src/Planner/Utils.h +++ b/src/Planner/Utils.h @@ -115,4 +115,9 @@ ActionsDAG::NodeRawConstPtrs getConjunctsList(ActionsDAG::Node * predicate); /// Returns true if the query always returns at least 1 row. bool optimizePlanForExists(QueryPlan & query_plan); +/// Create ExpressionStep that projects only used columns +QueryPlanStepPtr projectOnlyUsedColumns( + const SharedHeader & stream_header, + const ColumnIdentifiers & used_column_identifiers); + } diff --git a/src/Processors/QueryPlan/CommonSubplanReferenceStep.cpp b/src/Processors/QueryPlan/CommonSubplanReferenceStep.cpp new file mode 100644 index 000000000000..c8b2cbc48d4d --- /dev/null +++ b/src/Processors/QueryPlan/CommonSubplanReferenceStep.cpp @@ -0,0 +1,21 @@ +#include + +#include +#include + +namespace DB +{ + +namespace ErrorCodes +{ + +extern const int NOT_IMPLEMENTED; + +} + +void CommonSubplanReferenceStep::initializePipeline(QueryPipelineBuilder &, const BuildQueryPipelineSettings &) +{ + throw Exception(ErrorCodes::NOT_IMPLEMENTED, "CommonSubplanReference cannot be used to build pipeline"); +} + +} diff --git a/src/Processors/QueryPlan/CommonSubplanReferenceStep.h b/src/Processors/QueryPlan/CommonSubplanReferenceStep.h new file mode 100644 index 000000000000..21222d0cc0a8 --- /dev/null +++ b/src/Processors/QueryPlan/CommonSubplanReferenceStep.h @@ -0,0 +1,45 @@ +#pragma once + +#include +#include + +namespace DB +{ + +using ColumnIdentifier = std::string; +using ColumnIdentifiers = std::vector; + +class CommonSubplanReferenceStep : public ISourceStep +{ +public: + explicit CommonSubplanReferenceStep( + const SharedHeader & header_, + QueryPlan::Node * subplan_root_, + ColumnIdentifiers columns_to_use_) + : ISourceStep(header_) + , subplan_root(subplan_root_) + , columns_to_use(std::move(columns_to_use_)) + {} + + String getName() const override { return "CommonSubplanReference"; } + + void initializePipeline(QueryPipelineBuilder &, const BuildQueryPipelineSettings &) override; + + QueryPlanStepPtr clone() const override + { + return std::make_unique(getOutputHeader(), subplan_root, columns_to_use); + } + + QueryPlan::Node * getSubplanReferenceRoot() const { return subplan_root; } + + const ColumnIdentifiers & getColumnsToUse() const { return columns_to_use; } + + ColumnIdentifiers extractColumnsToUse() { return std::move(columns_to_use); } + +private: + QueryPlan::Node * subplan_root = nullptr; + + ColumnIdentifiers columns_to_use; +}; + +} diff --git a/src/Processors/QueryPlan/CommonSubplanStep.cpp b/src/Processors/QueryPlan/CommonSubplanStep.cpp new file mode 100644 index 000000000000..6c547ae5f0bf --- /dev/null +++ b/src/Processors/QueryPlan/CommonSubplanStep.cpp @@ -0,0 +1,45 @@ +#include + +#include +#include + +namespace DB +{ + +namespace ErrorCodes +{ + +extern const int NOT_IMPLEMENTED; + +} + +namespace +{ + +constexpr ITransformingStep::Traits getTraits() +{ + return ITransformingStep::Traits + { + { + .returns_single_stream = false, + .preserves_number_of_streams = true, + .preserves_sorting = true, + }, + { + .preserves_number_of_rows = true, + } + }; +} + +} + +CommonSubplanStep::CommonSubplanStep(const SharedHeader & header_) + : ITransformingStep(header_, header_, getTraits()) +{} + +void CommonSubplanStep::transformPipeline(QueryPipelineBuilder &, const BuildQueryPipelineSettings &) +{ + throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Subplan cannot be used to build pipeline"); +} + +} diff --git a/src/Processors/QueryPlan/CommonSubplanStep.h b/src/Processors/QueryPlan/CommonSubplanStep.h new file mode 100644 index 000000000000..446431ebfcf5 --- /dev/null +++ b/src/Processors/QueryPlan/CommonSubplanStep.h @@ -0,0 +1,32 @@ +#pragma once + +#include +#include +#include + +namespace DB +{ + +class CommonSubplanStep : public ITransformingStep +{ +public: + explicit CommonSubplanStep(const SharedHeader & header_); + + CommonSubplanStep(const CommonSubplanStep &) = default; + + String getName() const override { return "CommonSubplan"; } + + void transformPipeline(QueryPipelineBuilder &, const BuildQueryPipelineSettings &) override; + + void updateOutputHeader() override + { + output_header = input_headers.front(); + } + + QueryPlanStepPtr clone() const override + { + return std::make_unique(*this); + } +}; + +} diff --git a/src/Processors/QueryPlan/Optimizations/Optimizations.h b/src/Processors/QueryPlan/Optimizations/Optimizations.h index 44a1e8366b39..3a1fd2681fd9 100644 --- a/src/Processors/QueryPlan/Optimizations/Optimizations.h +++ b/src/Processors/QueryPlan/Optimizations/Optimizations.h @@ -145,6 +145,8 @@ void optimizeJoinByShards(QueryPlan::Node & root); void optimizeDistinctInOrder(QueryPlan::Node & node, QueryPlan::Nodes &); void updateQueryConditionCache(const Stack & stack, const QueryPlanOptimizationSettings & optimization_settings); bool optimizeVectorSearchSecondPass(QueryPlan::Node & root, Stack & stack, QueryPlan::Nodes & nodes, const Optimization::ExtraSettings &); +void materializeQueryPlanReferences(QueryPlan::Node & node, QueryPlan::Nodes & nodes); +void optimizeUnusedCommonSubplans(QueryPlan::Node & node); // Should be called once the query plan tree structure is finalized, i.e. no nodes addition, deletion or pushing down should happen after that call. // Since those hashes are used for join optimization, the calculation performed before join optimization. diff --git a/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp b/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp index 515960834316..de9ee0d0db81 100644 --- a/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp +++ b/src/Processors/QueryPlan/Optimizations/filterPushDown.cpp @@ -5,23 +5,25 @@ #include -#include -#include -#include -#include #include -#include -#include -#include #include +#include +#include #include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include #include -#include #include -#include -#include + #include #include @@ -696,6 +698,9 @@ size_t tryPushDownFilter(QueryPlan::Node * parent_node, QueryPlan::Nodes & nodes return 1; } + if (auto updated_steps = simplePushDownOverStep(parent_node, false, nodes, child)) + return updated_steps; + return 0; } diff --git a/src/Processors/QueryPlan/Optimizations/materializeQueryPlanReferences.cpp b/src/Processors/QueryPlan/Optimizations/materializeQueryPlanReferences.cpp new file mode 100644 index 000000000000..1342a5fc827b --- /dev/null +++ b/src/Processors/QueryPlan/Optimizations/materializeQueryPlanReferences.cpp @@ -0,0 +1,69 @@ +#include +#include +#include +#include +#include +#include + +#include + +namespace DB +{ + +namespace ErrorCodes +{ + +extern const int LOGICAL_ERROR; + +} + +namespace QueryPlanOptimizations +{ + +void materializeQueryPlanReferences(QueryPlan::Node & node, QueryPlan::Nodes & nodes) +{ + auto * subplan_reference = typeid_cast(node.step.get()); + if (!subplan_reference) + return; + + auto columns_to_use = subplan_reference->extractColumnsToUse(); + + QueryPlan::cloneSubplanAndReplace(&node, subplan_reference->getSubplanReferenceRoot(), nodes); + + auto * common_subplan = typeid_cast(node.step.get()); + if (!common_subplan) + throw Exception( + ErrorCodes::LOGICAL_ERROR, + "Expected CommonSubplanReferenceStep to reference CommonSubplanStep, but got {}", + node.step->getName()); + + if (node.children.size() != 1) + throw Exception( + ErrorCodes::LOGICAL_ERROR, + "Expected CommonSubplanStep to have exactly one child, but got {}", + node.children.size()); + + node.step = projectOnlyUsedColumns(common_subplan->getInputHeaders().front(), columns_to_use); +} + +void optimizeUnusedCommonSubplans(QueryPlan::Node & node) +{ + auto * common_subplan = typeid_cast(node.step.get()); + if (!common_subplan) + return; + + if (node.children.size() != 1) + throw Exception( + ErrorCodes::LOGICAL_ERROR, + "Expected CommonSubplanStep to have exactly one child, but got {}", + node.children.size()); + + auto * child = node.children[0]; + + node.step = std::move(child->step); + node.children = std::move(child->children); +} + +} + +} diff --git a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp index cc14c7de52ec..6a6d62f8c67c 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp @@ -120,6 +120,31 @@ void optimizeTreeFirstPass(const QueryPlanOptimizationSettings & optimization_se } } +template +void traverseQueryPlan(Stack & stack, QueryPlan::Node & root, Func && on_enter) +{ + stack.clear(); + stack.push_back({.node = &root}); + + while (!stack.empty()) + { + auto & frame = stack.back(); + + if (frame.next_child == 0) + on_enter(*frame.node); + + if (frame.next_child < frame.node->children.size()) + { + auto next_frame = Frame{.node = frame.node->children[frame.next_child]}; + ++frame.next_child; + stack.push_back(next_frame); + continue; + } + + stack.pop_back(); + } +} + void optimizeTreeSecondPass( const QueryPlanOptimizationSettings & optimization_settings, QueryPlan::Node & root, QueryPlan::Nodes & nodes, QueryPlan & query_plan) { @@ -163,6 +188,18 @@ void optimizeTreeSecondPass( stack.pop_back(); } + /// Materialize subplan references before other optimizations. + traverseQueryPlan(stack, root, [&](auto & frame_node) + { + materializeQueryPlanReferences(frame_node, nodes); + }); + + /// Remove CommonSubplanSteps (they must be not used at that point). + traverseQueryPlan(stack, root, [&](auto & frame_node) + { + optimizeUnusedCommonSubplans(frame_node); + }); + calculateHashTableCacheKeys(root); stack.push_back({.node = &root}); diff --git a/src/Processors/QueryPlan/QueryPlan.cpp b/src/Processors/QueryPlan/QueryPlan.cpp index 0567953d8614..4ee902b82b3a 100644 --- a/src/Processors/QueryPlan/QueryPlan.cpp +++ b/src/Processors/QueryPlan/QueryPlan.cpp @@ -93,7 +93,7 @@ void QueryPlan::unitePlans(QueryPlanStepPtr step, std::vectorgetName(), - root->step->getName(), + plans[i]->root->step->getName(), plan_header->dumpStructure(), step_header->dumpStructure()); } @@ -662,9 +662,10 @@ QueryPlan QueryPlan::extractSubplan(Node * subplan_root) return new_plan; } -QueryPlan QueryPlan::clone() const +void QueryPlan::cloneInplace(Node * node_to_replace, Node * subplan_root) { - QueryPlan result; + if (!subplan_root) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot clone subplan in place because subplan root is null"); struct Frame { @@ -673,10 +674,7 @@ QueryPlan QueryPlan::clone() const std::vector children = {}; }; - result.nodes.emplace_back(Node{ .step = {}, .children = {} }); - result.root = &result.nodes.back(); - - std::vector nodes_to_process{ Frame{ .node = root, .clone = result.root } }; + std::vector nodes_to_process{ Frame{ .node = subplan_root, .clone = node_to_replace } }; while (!nodes_to_process.empty()) { @@ -692,19 +690,68 @@ QueryPlan QueryPlan::clone() const size_t next_child = frame.children.size(); auto * child = frame.node->children[next_child]; - result.nodes.emplace_back(Node{ .step = {} }); - result.nodes.back().children.reserve(child->children.size()); - auto * child_clone = &result.nodes.back(); + nodes.emplace_back(Node{ .step = {} }); + nodes.back().children.reserve(child->children.size()); + auto * child_clone = &nodes.back(); frame.children.push_back(child_clone); nodes_to_process.push_back(Frame{ .node = child, .clone = child_clone }); } } +} + +QueryPlan QueryPlan::clone() const +{ + QueryPlan result; + result.nodes.emplace_back(Node{ .step = {}, .children = {} }); + auto * current_subplan_copy_root = &result.nodes.back(); + + result.cloneInplace(current_subplan_copy_root, root); + result.root = current_subplan_copy_root; return result; } +void QueryPlan::cloneSubplanAndReplace(Node * node_to_replace, Node * subplan_root, Nodes & nodes) +{ + if (!subplan_root) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot clone subplan in place because subplan root is null"); + + struct Frame + { + Node * node; + Node * clone; + std::vector children = {}; + }; + + std::vector nodes_to_process{ Frame{ .node = subplan_root, .clone = node_to_replace } }; + + while (!nodes_to_process.empty()) + { + auto & frame = nodes_to_process.back(); + if (frame.children.size() == frame.node->children.size()) + { + frame.clone->step = frame.node->step->clone(); + frame.clone->children = std::move(frame.children); + nodes_to_process.pop_back(); + } + else + { + size_t next_child = frame.children.size(); + auto * child = frame.node->children[next_child]; + + nodes.emplace_back(Node{ .step = {} }); + nodes.back().children.reserve(child->children.size()); + auto * child_clone = &nodes.back(); + + frame.children.push_back(child_clone); + + nodes_to_process.push_back(Frame{ .node = child, .clone = child_clone }); + } + } +} + void QueryPlan::replaceNodeWithPlan(Node * node, QueryPlanPtr plan) { diff --git a/src/Processors/QueryPlan/QueryPlan.h b/src/Processors/QueryPlan/QueryPlan.h index 2eb384b1cf6c..bab24b697786 100644 --- a/src/Processors/QueryPlan/QueryPlan.h +++ b/src/Processors/QueryPlan/QueryPlan.h @@ -140,8 +140,11 @@ class QueryPlan void replaceNodeWithPlan(Node * node, QueryPlanPtr plan); QueryPlan extractSubplan(Node * subplan_root); + void cloneInplace(Node * node_to_replace, Node * subplan_root); QueryPlan clone() const; + static void cloneSubplanAndReplace(Node * node_to_replace, Node * subplan_root, Nodes & nodes); + private: struct SerializationFlags; diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.cpp b/src/Processors/QueryPlan/ReadFromMergeTree.cpp index 757a37046efd..fb1c2144dfc9 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.cpp +++ b/src/Processors/QueryPlan/ReadFromMergeTree.cpp @@ -2409,7 +2409,27 @@ Pipe ReadFromMergeTree::groupStreamsByPartition(AnalysisResult & result, std::op QueryPlanStepPtr ReadFromMergeTree::clone() const { - return std::make_unique(*this); + AnalysisResultPtr analysis_result_copy; + if (analyzed_result_ptr) + analysis_result_copy = std::make_shared(*analyzed_result_ptr); + + return std::make_unique( + prepared_parts, + mutations_snapshot, + all_column_names, + data, + query_info, + storage_snapshot, + context, + block_size.max_block_size_rows, + requested_num_streams, + max_block_numbers_to_read, + log, + std::move(analysis_result_copy), + is_parallel_reading_from_replicas, + all_ranges_callback, + read_task_callback, + number_of_current_replica); } void ReadFromMergeTree::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.h b/src/Processors/QueryPlan/ReadFromMergeTree.h index 73f523772b47..83608bfce723 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.h +++ b/src/Processors/QueryPlan/ReadFromMergeTree.h @@ -132,6 +132,11 @@ class ReadFromMergeTree final : public SourceStepWithFilter UInt64 selected_rows = 0; bool has_exact_ranges = false; + AnalysisResult() = default; + + AnalysisResult(const AnalysisResult &) = default; + AnalysisResult(AnalysisResult &&) noexcept = default; + bool readFromProjection() const { return !parts_with_ranges.empty() && parts_with_ranges.front().data_part->isProjectionPart(); } void checkLimits(const Settings & settings, const SelectQueryInfo & query_info_) const; }; diff --git a/tests/queries/0_stateless/03545_analyzer_correlated_subqueries_use_equivalence_classes.sql b/tests/queries/0_stateless/03545_analyzer_correlated_subqueries_use_equivalence_classes.sql index 001ffd9e3a29..148064f6b4a7 100644 --- a/tests/queries/0_stateless/03545_analyzer_correlated_subqueries_use_equivalence_classes.sql +++ b/tests/queries/0_stateless/03545_analyzer_correlated_subqueries_use_equivalence_classes.sql @@ -2,9 +2,10 @@ SET enable_analyzer = 1; SET allow_experimental_correlated_subqueries = 1; SET enable_parallel_replicas = 0; SET query_plan_join_swap_table = 0; -- Changes query plan +SET correlated_subqueries_default_join_kind = 'left'; EXPLAIN actions = 1 -SELECT +SELECT (SELECT count() FROM diff --git a/tests/queries/0_stateless/03545_analyzer_correlated_subqueries_use_equivalence_classes_2.sql b/tests/queries/0_stateless/03545_analyzer_correlated_subqueries_use_equivalence_classes_2.sql index f128465853a4..babf94551b3f 100644 --- a/tests/queries/0_stateless/03545_analyzer_correlated_subqueries_use_equivalence_classes_2.sql +++ b/tests/queries/0_stateless/03545_analyzer_correlated_subqueries_use_equivalence_classes_2.sql @@ -3,6 +3,7 @@ SET allow_experimental_correlated_subqueries = 1; SET enable_parallel_replicas = 0; SET correlated_subqueries_substitute_equivalent_expressions=1; SET query_plan_join_swap_table = false; +SET correlated_subqueries_default_join_kind = 'left'; CREATE TABLE a(c1 Int64, c2 Int64, c3 Int64, c4 Int64) ENGINE = MergeTree() ORDER BY (); CREATE TABLE b(c1 Int64, c2 Int64, c3 Int64, c4 Int64) ENGINE = MergeTree() ORDER BY (); diff --git a/tests/queries/0_stateless/03545_analyzer_correlated_subquery_exists_asterisk.reference b/tests/queries/0_stateless/03545_analyzer_correlated_subquery_exists_asterisk.reference index 767f446f0d06..303e65d68c13 100644 --- a/tests/queries/0_stateless/03545_analyzer_correlated_subquery_exists_asterisk.reference +++ b/tests/queries/0_stateless/03545_analyzer_correlated_subquery_exists_asterisk.reference @@ -42,14 +42,17 @@ Positions: 0 Strictness: ALL Algorithm: HashJoin Clauses: [(__table1.i1) = (__table4.i2)] - Expression (( + Change column names to column identifiers)) - Actions: INPUT : 0 -> i1 Int64 : 0 - ALIAS i1 :: 0 -> __table1.i1 Int64 : 1 - Positions: 1 - ReadFromMergeTree (default.test) - ReadType: Default - Parts: 1 - Granules: 1 + Expression (Project only used columns) + Actions: INPUT :: 0 -> __table1.i1 Int64 : 0 + Positions: 0 + Expression (Change column names to column identifiers) + Actions: INPUT : 0 -> i1 Int64 : 0 + ALIAS i1 :: 0 -> __table1.i1 Int64 : 1 + Positions: 1 + ReadFromMergeTree (default.test) + ReadType: Default + Parts: 1 + Granules: 1 Expression Actions: INPUT :: 0 -> __table4.i1 Int64 : 0 INPUT :: 1 -> __table4.i2 Int64 : 1 diff --git a/tests/queries/0_stateless/03545_analyzer_correlated_subquery_exists_asterisk.sql b/tests/queries/0_stateless/03545_analyzer_correlated_subquery_exists_asterisk.sql index 6cb95b408a85..952c9cdaf60e 100644 --- a/tests/queries/0_stateless/03545_analyzer_correlated_subquery_exists_asterisk.sql +++ b/tests/queries/0_stateless/03545_analyzer_correlated_subquery_exists_asterisk.sql @@ -1,6 +1,7 @@ SET enable_analyzer = 1; SET allow_experimental_correlated_subqueries = 1; SET enable_parallel_replicas = 0; +SET correlated_subqueries_default_join_kind = 'left'; -- Disable table swaps during query planning SET query_plan_join_swap_table = false; diff --git a/tests/queries/0_stateless/03701_analyzer_correlated_subquery_plan_reference.reference b/tests/queries/0_stateless/03701_analyzer_correlated_subquery_plan_reference.reference new file mode 100644 index 000000000000..910c496e9718 --- /dev/null +++ b/tests/queries/0_stateless/03701_analyzer_correlated_subquery_plan_reference.reference @@ -0,0 +1,100 @@ +Expression ((Project names + Projection)) +Actions: INPUT :: 0 -> count() UInt64 : 0 +Positions: 0 + Aggregating + Keys: + Aggregates: + count() + Function: count() → UInt64 + Arguments: none + Skip merging: 0 + Expression (Before GROUP BY) + Positions: + Filter (WHERE) + Filter column: and(exists(__table2), equals(__table1.y, 1_UInt8)) (removed) + Actions: INPUT :: 0 -> __table1.y Int32 : 0 + INPUT : 1 -> exists(__table2) UInt8 : 1 + INPUT :: 2 -> __table1.x Int32 : 2 + ALIAS exists(__table2) :: 1 -> and(exists(__table2), equals(__table1.y, 1_UInt8)) UInt8 : 3 + Positions: 3 + Expression + Actions: INPUT :: 0 -> __table1.x Int32 : 0 + INPUT :: 1 -> __table1.y Int32 : 1 + INPUT :: 2 -> exists(__table2) UInt8 : 2 + Positions: 0 1 2 + Join (JOIN to generate result stream) + Type: LEFT + Strictness: ANY + Algorithm: ConcurrentHashJoin + Clauses: [(__table1.x) = (exists(__table2).__table1.x)] + Expression (WHERE) + Actions: INPUT :: 0 -> __table1.x Int32 : 0 + INPUT :: 1 -> __table1.y Int32 : 1 + Positions: 0 1 + Expression ((WHERE + Change column names to column identifiers)) + Actions: INPUT : 0 -> x Int32 : 0 + INPUT : 1 -> y Int32 : 1 + ALIAS x :: 0 -> __table1.x Int32 : 2 + ALIAS y :: 1 -> __table1.y Int32 : 0 + Positions: 2 0 + ReadFromMergeTree (default.t) + ReadType: Default + Parts: 1 + Granules: 1 + Prewhere info + Need filter: 1 + Prewhere filter + Prewhere filter column: equals(__table1.y, 1_UInt8) (removed) + Actions: INPUT : 0 -> y Int32 : 0 + COLUMN Const(UInt8) -> 1_UInt8 UInt8 : 1 + FUNCTION equals(y : 0, 1_UInt8 :: 1) -> equals(__table1.y, 1_UInt8) UInt8 : 2 + Positions: 0 2 + Expression (Create result for always true EXISTS expression) + Actions: INPUT :: 0 -> __table4.number UInt64 : 0 + INPUT : 1 -> __table1.x Int32 : 1 + COLUMN Const(UInt8) -> exists(__table2) UInt8 : 2 + ALIAS __table1.x :: 1 -> exists(__table2).__table1.x Int32 : 3 + FUNCTION materialize(exists(__table2) :: 2) -> materialize(exists(__table2)) UInt8 : 1 + ALIAS materialize(exists(__table2)) :: 1 -> exists(__table2) UInt8 : 2 + Positions: 3 2 + Filter (WHERE) + Filter column: notEquals(__table4.number, __table1.x) (removed) + Actions: INPUT : 0 -> __table4.number UInt64 : 0 + INPUT : 1 -> __table1.x Int32 : 1 + FUNCTION notEquals(__table4.number : 0, __table1.x : 1) -> notEquals(__table4.number, __table1.x) UInt8 : 2 + Positions: 2 0 1 + Expression + Actions: INPUT :: 0 -> __table1.x Int32 : 0 + INPUT :: 1 -> __table4.number UInt64 : 1 + Positions: 0 1 + Join (JOIN to evaluate correlated expression) + Type: CROSS + Strictness: ALL + Algorithm: HashJoin + Expression (Project only used columns) + Actions: INPUT :: 0 -> __table1.x Int32 : 0 + INPUT :: 1 -> __table1.y Int32 : 1 + Positions: 0 + Expression ((WHERE + Change column names to column identifiers)) + Actions: INPUT : 0 -> x Int32 : 0 + INPUT : 1 -> y Int32 : 1 + ALIAS x :: 0 -> __table1.x Int32 : 2 + ALIAS y :: 1 -> __table1.y Int32 : 0 + Positions: 2 0 + ReadFromMergeTree (default.t) + ReadType: Default + Parts: 1 + Granules: 1 + Prewhere info + Need filter: 1 + Prewhere filter + Prewhere filter column: equals(__table1.y, 1_UInt8) (removed) + Actions: INPUT : 0 -> y Int32 : 0 + COLUMN Const(UInt8) -> 1_UInt8 UInt8 : 1 + FUNCTION equals(y : 0, 1_UInt8 :: 1) -> equals(__table1.y, 1_UInt8) UInt8 : 2 + Positions: 0 2 + Expression (Change column names to column identifiers) + Actions: INPUT : 0 -> number UInt64 : 0 + ALIAS number :: 0 -> __table4.number UInt64 : 1 + Positions: 1 + ReadFromSystemNumbers diff --git a/tests/queries/0_stateless/03701_analyzer_correlated_subquery_plan_reference.sql b/tests/queries/0_stateless/03701_analyzer_correlated_subquery_plan_reference.sql new file mode 100644 index 000000000000..2a2cf96b11d0 --- /dev/null +++ b/tests/queries/0_stateless/03701_analyzer_correlated_subquery_plan_reference.sql @@ -0,0 +1,23 @@ +SET enable_analyzer = 1; +SET query_plan_join_swap_table = false; +SET enable_parallel_replicas = 0; +SET correlated_subqueries_default_join_kind = 'left'; + +CREATE TABLE t(x Int, y Int) ORDER BY () +AS SELECT number as x, number % 2 as y FROM numbers(100); + +EXPLAIN actions = 1 +SELECT + count() +FROM + t n +WHERE + EXISTS ( + SELECT + * + FROM + numbers(10) + WHERE + number != n.x + ) + AND n.y = 1;