Stable 25.8 Backport of #85455 - Propagate query plan optimizations to correlated subquery input

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"(

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)

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"},

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,

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,

src/Planner/PlannerCorrelatedSubqueries.cpp

@@ -7,6 +7,7 @@
 #include <Common/typeid_cast.h>
 
 #include <Core/Joins.h>
+#include <Core/QueryProcessingStage.h>
 #include <Core/Settings.h>
 
 #include <DataTypes/DataTypeNullable.h>
@@ -27,12 +28,18 @@
 #include <Planner/Utils.h>
 
 #include <Processors/QueryPlan/AggregatingStep.h>
+#include <Processors/QueryPlan/CommonSubplanReferenceStep.h>
+#include <Processors/QueryPlan/CommonSubplanStep.h>
 #include <Processors/QueryPlan/ExpressionStep.h>
 #include <Processors/QueryPlan/FilterStep.h>
 #include <Processors/QueryPlan/JoinStepLogical.h>
 #include <Processors/QueryPlan/LimitStep.h>
 #include <Processors/QueryPlan/UnionStep.h>
 
+#include <Storages/ColumnsDescription.h>
+#include <Storages/ConstraintsDescription.h>
+#include <Storages/IStorage.h>
+
 #include <memory>
 #include <string_view>
 #include <unordered_map>
@@ -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<EquivalenceClasses> 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<ExpressionStep>(
-        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,19 +205,20 @@ 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();
 
             std::unordered_map<std::string_view, const ActionsDAG::Node *> decorrelated_nodes_names;
             for (const auto * output : outputs)
                 decorrelated_nodes_names[output->result_name] = output;
 
+            /// Find possible renamings for all correlated columns
             std::vector<std::pair<const ActionsDAG::Node *, const String &>> 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<CommonSubplanStep>(context.query_plan.getCurrentHeader()));
+
+        auto buffer_header = std::make_shared<Block>();
+        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<CommonSubplanReferenceStep>(
+            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<JoinStepLogical>(
             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<QueryPlanPtr> plans;
         plans.emplace_back(std::make_unique<QueryPlan>(std::move(lhs_plan)));
-        plans.emplace_back(std::make_unique<QueryPlan>(context.correlated_query_plan.extractSubplan(node)));
+        plans.emplace_back(std::make_unique<QueryPlan>(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<String(const String &)>;
+    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<JoinPredicate> 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<JoinStepLogical>(
         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<QueryPlanPtr> plans;
-    plans.emplace_back(std::make_unique<QueryPlan>(std::move(left_plan)));
-    plans.emplace_back(std::make_unique<QueryPlan>(std::move(right_plan)));
+    plans.emplace_back(std::make_unique<QueryPlan>(std::move(lhs_plan)));
+    plans.emplace_back(std::make_unique<QueryPlan>(std::move(rhs_plan)));
 
     result_plan.unitePlans(std::move(result_join), {std::move(plans)});
     return result_plan;

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<ExpressionStep>(stream_header, std::move(project_only_used_columns_actions));
+    step->setStepDescription("Project only used columns");
+    return step;
+}
+
 }

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);
+
 }

src/Processors/QueryPlan/CommonSubplanReferenceStep.cpp

@@ -0,0 +1,21 @@
+#include <Processors/QueryPlan/CommonSubplanReferenceStep.h>
+
+#include <QueryPipeline/Pipe.h>
+#include <QueryPipeline/QueryPipelineBuilder.h>
+
+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");
+}
+
+}