Address code review comments

Author: novikd

src/Analyzer/QueryNode.h

@@ -647,6 +647,8 @@ class QueryNode final : public IQueryTreeNode
 
     void addCorrelatedColumn(const QueryTreeNodePtr & correlated_column);
 
+    /// Returns result type of projection expression if query is correlated
+    /// or throws an exception otherwise.
     DataTypePtr getResultType() const override;
 
     QueryTreeNodeType getNodeType() const override

src/Interpreters/JoinInfo.cpp

@@ -356,6 +356,16 @@ JoinActionRef JoinActionRef::deserialize(ReadBuffer & in, const ActionsDAGRawPtr
     return res;
 }
 
+JoinActionRef JoinActionRef::clone(const ActionsDAG * actions_dag_) const
+{
+    return JoinActionRef{actions_dag_, column_name};
+}
+
+JoinActionRef::JoinActionRef(const ActionsDAG * actions_dag_, const String & column_name_)
+    : actions_dag(actions_dag_)
+    , column_name(column_name_)
+{}
+
 void JoinPredicate::serialize(WriteBuffer & out, const JoinActionRef::ActionsDAGRawPtrs & dags) const
 {
     serializePredicateOperator(op, out);
@@ -435,6 +445,40 @@ JoinCondition JoinCondition::deserialize(ReadBuffer & in, const JoinActionRef::A
     };
 }
 
+JoinCondition JoinCondition::clone(const JoinExpressionActions & expression_actions) const
+{
+    JoinCondition copy;
+
+    copy.predicates.reserve(predicates.size());
+    for (const auto & predicate : predicates)
+    {
+        copy.predicates.emplace_back(
+            predicate.left_node.clone(expression_actions.left_pre_join_actions.get()),
+            predicate.right_node.clone(expression_actions.right_pre_join_actions.get()),
+            predicate.op);
+    }
+
+    copy.left_filter_conditions.reserve(left_filter_conditions.size());
+    for (const auto & condition: left_filter_conditions)
+    {
+        copy.left_filter_conditions.emplace_back(condition.clone(expression_actions.left_pre_join_actions.get()));
+    }
+
+    copy.right_filter_conditions.reserve(right_filter_conditions.size());
+    for (const auto & condition: right_filter_conditions)
+    {
+        copy.right_filter_conditions.emplace_back(condition.clone(expression_actions.right_pre_join_actions.get()));
+    }
+
+    copy.residual_conditions.reserve(residual_conditions.size());
+    for (const auto & condition: residual_conditions)
+    {
+        copy.residual_conditions.emplace_back(condition.clone(expression_actions.post_join_actions.get()));
+    }
+
+    return copy;
+}
+
 void JoinExpression::serialize(WriteBuffer & out, const JoinActionRef::ActionsDAGRawPtrs & dags) const
 {
     UInt8 is_using_flag = is_using ? 1 : 0;
@@ -473,6 +517,20 @@ JoinExpression JoinExpression::deserialize(ReadBuffer & in, const JoinActionRef:
     return {std::move(condition), std::move(disjunctive_conditions), bool(is_using_flag)};
 }
 
+JoinExpression JoinExpression::clone(const JoinExpressionActions & expression_copy) const
+{
+    JoinExpression copy;
+    copy.condition = condition.clone(expression_copy);
+
+    copy.disjunctive_conditions.reserve(disjunctive_conditions.size());
+    for (const auto & disjunctive_condition : disjunctive_conditions)
+        copy.disjunctive_conditions.emplace_back(disjunctive_condition.clone(expression_copy));
+
+    copy.is_using = is_using;
+
+    return copy;
+}
+
 void JoinInfo::serialize(WriteBuffer & out, const JoinActionRef::ActionsDAGRawPtrs & dags) const
 {
     expression.serialize(out, dags);

src/Interpreters/JoinInfo.h

@@ -77,7 +77,7 @@ struct JoinExpressionActions
 
     JoinExpressionActions clone() const
     {
-        return JoinExpressionActions(
+        return  JoinExpressionActions(
             std::make_unique<ActionsDAG>(left_pre_join_actions->clone()),
             std::make_unique<ActionsDAG>(right_pre_join_actions->clone()),
             std::make_unique<ActionsDAG>(post_join_actions->clone())
@@ -117,7 +117,11 @@ class JoinActionRef
     void serialize(WriteBuffer & out, const ActionsDAGRawPtrs & dags) const;
     static JoinActionRef deserialize(ReadBuffer & in, const ActionsDAGRawPtrs & dags);
 
+    JoinActionRef clone(const ActionsDAG * actions_dag_) const;
+
 private:
+    JoinActionRef(const ActionsDAG * actions_dag_, const String & column_name_);
+
     const ActionsDAG * actions_dag = nullptr;
     String column_name;
 };
@@ -148,32 +152,10 @@ struct JoinCondition
     /// Unlike the join predicates, these conditions can be arbitrary expressions.
     std::vector<JoinActionRef> residual_conditions;
 
-    void fixReferences(const JoinExpressionActions & expression_actions)
-    {
-        for (auto & predicate : predicates)
-        {
-            predicate.left_node = JoinActionRef(&expression_actions.left_pre_join_actions->findInOutputs(predicate.left_node.getColumnName()), expression_actions.left_pre_join_actions.get());
-            predicate.right_node = JoinActionRef(&expression_actions.right_pre_join_actions->findInOutputs(predicate.right_node.getColumnName()), expression_actions.right_pre_join_actions.get());
-        }
-
-        for (auto & condition: left_filter_conditions)
-        {
-            condition = JoinActionRef(&expression_actions.left_pre_join_actions->findInOutputs(condition.getColumnName()), expression_actions.left_pre_join_actions.get());
-        }
-
-        for (auto & condition: right_filter_conditions)
-        {
-            condition = JoinActionRef(&expression_actions.right_pre_join_actions->findInOutputs(condition.getColumnName()), expression_actions.right_pre_join_actions.get());
-        }
-
-        for (auto & condition: residual_conditions)
-        {
-            condition = JoinActionRef(&expression_actions.post_join_actions->findInOutputs(condition.getColumnName()), expression_actions.post_join_actions.get());
-        }
-    }
-
     void serialize(WriteBuffer & out, const JoinActionRef::ActionsDAGRawPtrs & dags) const;
     static JoinCondition deserialize(ReadBuffer & in, const JoinActionRef::ActionsDAGRawPtrs & dags);
+
+    JoinCondition clone(const JoinExpressionActions & expression_actions) const;
 };
 
 struct JoinExpression
@@ -192,6 +174,8 @@ struct JoinExpression
 
     void serialize(WriteBuffer & out, const JoinActionRef::ActionsDAGRawPtrs & dags) const;
     static JoinExpression deserialize(ReadBuffer & in, const JoinActionRef::ActionsDAGRawPtrs & dags);
+
+    JoinExpression clone(const JoinExpressionActions & expression_copy) const;
 };
 
 struct JoinInfo
@@ -208,6 +192,11 @@ struct JoinInfo
     /// The locality of the join (e.g., LOCAL, GLOBAL)
     JoinLocality locality;
 
+    JoinInfo clone(const JoinExpressionActions & expression_actions) const
+    {
+        return JoinInfo{ expression.clone(expression_actions), kind, strictness, locality};
+    }
+
     void serialize(WriteBuffer & out, const JoinActionRef::ActionsDAGRawPtrs & dags) const;
     static JoinInfo deserialize(ReadBuffer & in, const JoinActionRef::ActionsDAGRawPtrs & dags);
 };

src/Processors/QueryPlan/JoinStepLogical.cpp

@@ -892,10 +892,7 @@ std::unique_ptr<IQueryPlanStep> JoinStepLogical::deserialize(Deserialization & c
 QueryPlanStepPtr JoinStepLogical::clone() const
 {
     auto new_expression_actions = expression_actions.clone();
-    auto new_join_info = join_info;
-    new_join_info.expression.condition.fixReferences(new_expression_actions);
-    for (auto & condition : new_join_info.expression.disjunctive_conditions)
-        condition.fixReferences(new_expression_actions);
+    auto new_join_info = join_info.clone(new_expression_actions);
 
     auto result_step = std::make_unique<JoinStepLogical>(
         getInputHeaders().front(), getInputHeaders().back(),

tests/queries/0_stateless/03447_analyzer_correlated_subqueries_tpc_h.sql

@@ -100,6 +100,7 @@ INSERT INTO lineitem SELECT * FROM generateRandom() LIMIT 1;
 
 set enable_analyzer = 1;
 set allow_experimental_correlated_subqueries = 1;
+SET enable_parallel_replicas = 0;
 
 -- Q2
 SELECT