fixed

Author: lgbo-ustc

backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHSuite.scala

@@ -468,7 +468,13 @@ class GlutenClickHouseTPCHSuite extends MergeTreeSuite {
                 |""".stripMargin
     spark.sql(sql)
     sql = """
-            | select * from cross_join_t as t1 full join cross_join_t as t2 limit 10
+            |select * from (
+            | select a as a1, b as b1, c as c1 from cross_join_t
+            |) as t1 full join (
+            | select a as a2, b as b2, c as c2 from cross_join_t
+            |) as t2
+            |order by a1, b1, c1, a2, b2, c2
+            |limit 10
             |""".stripMargin
     compareResultsAgainstVanillaSpark(sql, true, { _ => })
     spark.sql("drop table cross_join_t")

cpp-ch/local-engine/Common/CHUtil.cpp

@@ -1048,16 +1048,25 @@ UInt64 MemoryUtil::getMemoryRSS()
     return rss * sysconf(_SC_PAGESIZE);
 }
 
-
-void JoinUtil::reorderJoinOutput(DB::QueryPlan & plan, DB::Names cols)
+void JoinUtil::adjustJoinOutput(DB::QueryPlan & plan, DB::Names cols)
 {
-    ActionsDAG project{plan.getCurrentHeader()->getNamesAndTypesList()};
-    NamesWithAliases project_cols;
+    auto header = plan.getCurrentHeader();
+    std::unordered_map<String, const DB::ActionsDAG::Node *> name_to_node;
+    ActionsDAG project;
+    for (const auto & col : header->getColumnsWithTypeAndName())
+    {
+        const auto * node = &(project.addInput(col));
+        name_to_node[col.name] = node;
+    }
     for (const auto & col : cols)
     {
-        project_cols.emplace_back(NameWithAlias(col, col));
+        const auto it = name_to_node.find(col);
+        if (it == name_to_node.end())
+        {
+            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Column {} not found in header", col);
+        }
+        project.addOrReplaceInOutputs(*(it->second));
     }
-    project.project(project_cols);
     QueryPlanStepPtr project_step = std::make_unique<ExpressionStep>(plan.getCurrentHeader(), std::move(project));
     project_step->setStepDescription("Reorder Join Output");
     plan.addStep(std::move(project_step));

cpp-ch/local-engine/Common/CHUtil.h

@@ -42,8 +42,6 @@ static const String MERGETREE_INSERT_WITHOUT_LOCAL_STORAGE = "mergetree.insert_w
 static const String MERGETREE_MERGE_AFTER_INSERT = "mergetree.merge_after_insert";
 static const std::string DECIMAL_OPERATIONS_ALLOW_PREC_LOSS = "spark.sql.decimalOperations.allowPrecisionLoss";
 static const std::string TIMER_PARSER_POLICY = "spark.sql.legacy.timeParserPolicy";
-// static constexpr auto CROSS_REL_CONST_KEY_COLUMN = "__CROSS_REL_CONST_KEY_COLUMN__";
-
 
 static const std::unordered_set<String> BOOL_VALUE_SETTINGS{
     MERGETREE_MERGE_AFTER_INSERT, MERGETREE_INSERT_WITHOUT_LOCAL_STORAGE, DECIMAL_OPERATIONS_ALLOW_PREC_LOSS};
@@ -254,7 +252,8 @@ class JoinUtil
     static constexpr auto CROSS_REL_LEFT_CONST_KEY_COLUMN = "__CROSS_REL_LEFT_CONST_KEY_COLUMN__";
     static constexpr auto CROSS_REL_RIGHT_CONST_KEY_COLUMN = "__CROSS_REL_RIGHT_CONST_KEY_COLUMN__";
 
-    static void reorderJoinOutput(DB::QueryPlan & plan, DB::Names cols);
+    // Keep necessarily columns and reorder them according to cols
+    static void adjustJoinOutput(DB::QueryPlan & plan, DB::Names cols);
     static std::pair<DB::JoinKind, DB::JoinStrictness>
     getJoinKindAndStrictness(substrait::JoinRel_JoinType join_type, bool is_existence_join);
     static std::pair<DB::JoinKind, DB::JoinStrictness> getCrossJoinKindAndStrictness(substrait::CrossRel_JoinType join_type);

cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp

@@ -189,15 +189,7 @@ DB::QueryPlanPtr CrossRelParser::parseJoin(const substrait::CrossRel & join, DB:
             right->getCurrentHeader()->dumpNames());
     }
 
-    Names after_join_names;
-    auto left_names = left->getCurrentHeader()->getNames();
-    after_join_names.insert(after_join_names.end(), left_names.begin(), left_names.end());
-    auto right_name = table_join->columnsFromJoinedTable().getNames();
-    after_join_names.insert(after_join_names.end(), right_name.begin(), right_name.end());
-
-    auto left_header = left->getCurrentHeader();
-    auto right_header = right->getCurrentHeader();
-
+    Names after_join_names = collectOutputColumnsName(*left, *right);
 
     if (table_join->kind() != JoinKind::Cross)
     {
@@ -223,15 +215,7 @@ DB::QueryPlanPtr CrossRelParser::parseJoin(const substrait::CrossRel & join, DB:
         extra_plan_holder.emplace_back(std::move(right));
 
         addPostFilter(*query_plan, join);
-        Names cols;
-        for (auto after_join_name : after_join_names)
-        {
-            if (BlockUtil::VIRTUAL_ROW_COUNT_COLUMN == after_join_name)
-                continue;
-
-            cols.emplace_back(after_join_name);
-        }
-        JoinUtil::reorderJoinOutput(*query_plan, cols);
+        JoinUtil::adjustJoinOutput(*query_plan, after_join_names);
     }
     else
     {
@@ -255,7 +239,7 @@ DB::QueryPlanPtr CrossRelParser::parseJoin(const substrait::CrossRel & join, DB:
 
         query_plan = std::make_unique<QueryPlan>();
         query_plan->unitePlans(std::move(join_step), {std::move(plans)});
-        JoinUtil::reorderJoinOutput(*query_plan, after_join_names);
+        JoinUtil::adjustJoinOutput(*query_plan, after_join_names);
     }
 
     return query_plan;
@@ -357,6 +341,26 @@ void CrossRelParser::addConvertStep(TableJoin & table_join, DB::QueryPlan & left
     }
 }
 
+DB::Names CrossRelParser::collectOutputColumnsName(const DB::QueryPlan & left, const DB::QueryPlan & right)
+{
+    Names join_result_names;
+    auto is_unused_column = [](const String & name)
+    {
+        return name == JoinUtil::CROSS_REL_LEFT_CONST_KEY_COLUMN || name == JoinUtil::CROSS_REL_RIGHT_CONST_KEY_COLUMN
+            || name == BlockUtil::VIRTUAL_ROW_COUNT_COLUMN;
+    };
+    for (auto & col : left.getCurrentHeader()->getColumnsWithTypeAndName())
+    {
+        if (!is_unused_column(col.name))
+            join_result_names.emplace_back(col.name);
+    }
+    for (auto & col : right.getCurrentHeader()->getColumnsWithTypeAndName())
+    {
+        if (!is_unused_column(col.name))
+            join_result_names.emplace_back(col.name);
+    }
+    return join_result_names;
+}
 
 void registerCrossRelParser(RelParserFactory & factory)
 {

cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.h

@@ -67,6 +67,7 @@ class CrossRelParser : public RelParser
         bool allow_mixed_condition);
 
     void addConstJoinKeys(DB::QueryPlan & left, DB::QueryPlan & right);
+    DB::Names collectOutputColumnsName(const DB::QueryPlan & left, const DB::QueryPlan & right);
 };
 
 }

cpp-ch/local-engine/Parser/RelParsers/JoinRelParser.cpp

@@ -358,7 +358,7 @@ DB::QueryPlanPtr JoinRelParser::parseJoin(const substrait::JoinRel & join, DB::Q
         }
     }
 
-    JoinUtil::reorderJoinOutput(*query_plan, after_join_names);
+    JoinUtil::adjustJoinOutput(*query_plan, after_join_names);
     /// Need to project the right table column into boolean type
     if (join_opt_info.is_existence_join)
         existenceJoinPostProject(*query_plan, left_names);