[GLUTEN-7145][CH][PART]refactor for rel parsers

cpp-ch/local-engine/CMakeLists.txt

@@ -47,6 +47,7 @@ add_subdirectory(proto)
 add_headers_and_sources(builder Builder)
 add_headers_and_sources(join Join)
 add_headers_and_sources(parser Parser)
+add_headers_and_sources(parser Parser/RelParsers)
 add_headers_and_sources(rewriter Rewriter)
 add_headers_and_sources(storages Storages)
 add_headers_and_sources(storages Storages/Output)

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

@@ -48,7 +48,7 @@
 #include <Functions/registerFunctions.h>
 #include <IO/SharedThreadPools.h>
 #include <Interpreters/JIT/CompiledExpressionCache.h>
-#include <Parser/RelParser.h>
+#include <Parser/RelParsers/RelParser.h>
 #include <Parser/SerializedPlanParser.h>
 #include <Parser/SubstraitParserUtils.h>
 #include <Planner/PlannerActionsVisitor.h>

cpp-ch/local-engine/Join/BroadCastJoinBuilder.cpp

@@ -19,7 +19,7 @@
 #include <Compression/CompressedReadBuffer.h>
 #include <Interpreters/TableJoin.h>
 #include <Join/StorageJoinFromReadBuffer.h>
-#include <Parser/JoinRelParser.h>
+#include <Parser/RelParsers/JoinRelParser.h>
 #include <Parser/TypeParser.h>
 #include <QueryPipeline/ProfileInfo.h>
 #include <Shuffle/ShuffleReader.h>

cpp-ch/local-engine/Parser/AggregateFunctionParser.cpp

@@ -20,7 +20,7 @@
 #include <DataTypes/DataTypeAggregateFunction.h>
 #include <DataTypes/DataTypeTuple.h>
 #include <Functions/FunctionHelpers.h>
-#include <Parser/RelParser.h>
+#include <Parser/RelParsers/RelParser.h>
 #include <Parser/TypeParser.h>
 #include <Common/CHUtil.h>
 #include <Common/Exception.h>

cpp-ch/local-engine/Parser/AggregateRelParser.h → cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.h

@@ -16,7 +16,7 @@
  */
 #pragma once
 #include <Parser/AggregateFunctionParser.h>
-#include <Parser/RelParser.h>
+#include <Parser/RelParsers/RelParser.h>
 #include <Poco/Logger.h>
 #include <Common/logger_useful.h>
 
@@ -30,7 +30,7 @@ class AggregateRelParser : public RelParser
     ~AggregateRelParser() override = default;
     DB::QueryPlanPtr
     parse(DB::QueryPlanPtr query_plan, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_) override;
-    const substrait::Rel & getSingleInput(const substrait::Rel & rel) override { return rel.aggregate().input(); }
+    std::optional<const substrait::Rel *> getSingleInput(const substrait::Rel & rel) override { return &rel.aggregate().input(); }
 
 private:
     struct AggregateInfo

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

@@ -15,6 +15,7 @@
  * limitations under the License.
  */
 #include "CrossRelParser.h"
+#include <optional>
 
 #include <Interpreters/CollectJoinOnKeysVisitor.h>
 #include <Interpreters/GraceHashJoin.h>
@@ -73,25 +74,27 @@ CrossRelParser::parse(DB::QueryPlanPtr /*query_plan*/, const substrait::Rel & /*
     throw Exception(ErrorCodes::LOGICAL_ERROR, "join node has 2 inputs, can't call parse().");
 }
 
-const substrait::Rel & CrossRelParser::getSingleInput(const substrait::Rel & /*rel*/)
-{
-    throw Exception(ErrorCodes::LOGICAL_ERROR, "join node has 2 inputs, can't call getSingleInput().");
-}
-
-DB::QueryPlanPtr CrossRelParser::parseOp(const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack)
+std::vector<const substrait::Rel *> CrossRelParser::getInputs(const substrait::Rel & rel)
 {
     const auto & join = rel.cross();
     if (!join.has_left() || !join.has_right())
     {
         throw Exception(ErrorCodes::BAD_ARGUMENTS, "left table or right table is missing.");
     }
 
-    rel_stack.push_back(&rel);
-    auto left_plan = getPlanParser()->parseOp(join.left(), rel_stack);
-    auto right_plan = getPlanParser()->parseOp(join.right(), rel_stack);
-    rel_stack.pop_back();
+    return {&join.left(), &join.right()};
+}
+std::optional<const substrait::Rel *> CrossRelParser::getSingleInput(const substrait::Rel & /*rel*/)
+{
+    throw Exception(ErrorCodes::LOGICAL_ERROR, "join node has 2 inputs, can't call getSingleInput().");
+}
 
-    return parseJoin(join, std::move(left_plan), std::move(right_plan));
+DB::QueryPlanPtr
+CrossRelParser::parse(std::vector<DB::QueryPlanPtr> & input_plans_, const substrait::Rel & rel, std::list<const substrait::Rel *> &)
+{
+    assert(input_plans_.size() == 2);
+    const auto & join = rel.cross();
+    return parseJoin(join, std::move(input_plans_[0]), std::move(input_plans_[1]));
 }
 
 void CrossRelParser::renamePlanColumns(DB::QueryPlan & left, DB::QueryPlan & right, const StorageJoinFromReadBuffer & storage_join)
@@ -194,7 +197,8 @@ DB::QueryPlanPtr CrossRelParser::parseJoin(const substrait::CrossRel & join, DB:
     else
     {
         JoinPtr hash_join = std::make_shared<HashJoin>(table_join, right->getCurrentDataStream().header.cloneEmpty());
-        QueryPlanStepPtr join_step = std::make_unique<DB::JoinStep>(left->getCurrentDataStream(), right->getCurrentDataStream(), hash_join, 8192, 1, false);
+        QueryPlanStepPtr join_step
+            = std::make_unique<DB::JoinStep>(left->getCurrentDataStream(), right->getCurrentDataStream(), hash_join, 8192, 1, false);
         join_step->setStepDescription("CROSS_JOIN");
         steps.emplace_back(join_step.get());
         std::vector<QueryPlanPtr> plans;

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

@@ -17,7 +17,8 @@
 #pragma once
 
 #include <memory>
-#include <Parser/RelParser.h>
+#include <optional>
+#include <Parser/RelParsers/RelParser.h>
 #include <substrait/algebra.pb.h>
 
 namespace DB
@@ -37,12 +38,13 @@ class CrossRelParser : public RelParser
     explicit CrossRelParser(SerializedPlanParser * plan_paser_);
     ~CrossRelParser() override = default;
 
+    DB::QueryPlanPtr parse(
+        std::vector<DB::QueryPlanPtr> & input_plans_, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_) override;
     DB::QueryPlanPtr
-    parse(DB::QueryPlanPtr query_plan, const substrait::Rel & sort_rel, std::list<const substrait::Rel *> & rel_stack_) override;
+    parse(DB::QueryPlanPtr query_plan, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_) override;
 
-    DB::QueryPlanPtr parseOp(const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack) override;
-
-    const substrait::Rel & getSingleInput(const substrait::Rel & rel) override;
+    std::vector<const substrait::Rel *> getInputs(const substrait::Rel & rel) override;
+    std::optional<const substrait::Rel *> getSingleInput(const substrait::Rel & rel) override;
 
 private:
     std::unordered_map<std::string, std::string> & function_mapping;
@@ -55,7 +57,11 @@ class CrossRelParser : public RelParser
     void addConvertStep(TableJoin & table_join, DB::QueryPlan & left, DB::QueryPlan & right);
     void addPostFilter(DB::QueryPlan & query_plan, const substrait::CrossRel & join);
     bool applyJoinFilter(
-        DB::TableJoin & table_join, const substrait::CrossRel & join_rel, DB::QueryPlan & left, DB::QueryPlan & right, bool allow_mixed_condition);
+        DB::TableJoin & table_join,
+        const substrait::CrossRel & join_rel,
+        DB::QueryPlan & left,
+        DB::QueryPlan & right,
+        bool allow_mixed_condition);
 };
 
 }

cpp-ch/local-engine/Parser/ExpandRelParser.cpp → cpp-ch/local-engine/Parser/RelParsers/ExpandRelParser.cpp

@@ -20,7 +20,7 @@
 #include <Core/ColumnWithTypeAndName.h>
 #include <Operator/ExpandStep.h>
 #include <Parser/ExpandField.h>
-#include <Parser/RelParser.h>
+#include <Parser/RelParsers/RelParser.h>
 #include <Parser/SerializedPlanParser.h>
 #include <Processors/QueryPlan/QueryPlan.h>
 #include <Common/logger_useful.h>

cpp-ch/local-engine/Parser/ExpandRelParser.h → cpp-ch/local-engine/Parser/RelParsers/ExpandRelParser.h

@@ -15,7 +15,8 @@
  * limitations under the License.
  */
 #pragma once
-#include <Parser/RelParser.h>
+#include <optional>
+#include <Parser/RelParsers/RelParser.h>
 
 
 namespace local_engine
@@ -29,6 +30,6 @@ class ExpandRelParser : public RelParser
     DB::QueryPlanPtr
     parse(DB::QueryPlanPtr query_plan, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_) override;
 
-    const substrait::Rel & getSingleInput(const substrait::Rel & rel) override { return rel.expand().input(); }
+    std::optional<const substrait::Rel *> getSingleInput(const substrait::Rel & rel) override { return &rel.expand().input(); }
 };
 }

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

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <memory>
+#include <optional>
+#include <Parser/SerializedPlanParser.h>
+#include <Processors/QueryPlan/LimitStep.h>
+#include "RelParser.h"
+namespace local_engine
+{
+class FetchRelParser : public RelParser
+{
+public:
+    explicit FetchRelParser(SerializedPlanParser * plan_parser_) : RelParser(plan_parser_) { }
+    ~FetchRelParser() override = default;
+
+    DB::QueryPlanPtr parse(DB::QueryPlanPtr query_plan, const substrait::Rel & rel, std::list<const substrait::Rel *> &)
+    {
+        const auto & limit = rel.fetch();
+        auto limit_step = std::make_unique<DB::LimitStep>(query_plan->getCurrentDataStream(), limit.count(), limit.offset());
+        limit_step->setStepDescription("LIMIT");
+        steps.push_back(limit_step.get());
+        query_plan->addStep(std::move(limit_step));
+        return query_plan;
+    }
+    std::optional<const substrait::Rel *> getSingleInput(const substrait::Rel & rel) override { return &rel.fetch().input(); }
+};
+
+void registerFetchRelParser(RelParserFactory & factory)
+{
+    auto builder = [](SerializedPlanParser * plan_parser_) { return std::make_unique<FetchRelParser>(plan_parser_); };
+    factory.registerBuilder(substrait::Rel::RelTypeCase::kFetch, builder);
+}
+
+}

cpp-ch/local-engine/Parser/FilterRelParser.h → cpp-ch/local-engine/Parser/RelParsers/FilterRelParser.h

@@ -17,7 +17,8 @@
 
 #pragma once
 
-#include <Parser/RelParser.h>
+#include <optional>
+#include <Parser/RelParsers/RelParser.h>
 #include <Poco/Logger.h>
 #include <Common/logger_useful.h>
 
@@ -29,11 +30,12 @@ class FilterRelParser : public RelParser
     explicit FilterRelParser(SerializedPlanParser * plan_paser_);
     ~FilterRelParser() override = default;
 
-    const substrait::Rel & getSingleInput(const substrait::Rel & rel) override { return rel.filter().input(); }
+    std::optional<const substrait::Rel *> getSingleInput(const substrait::Rel & rel) override { return &rel.filter().input(); }
 
     DB::QueryPlanPtr
     parse(DB::QueryPlanPtr query_plan, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_) override;
+
 private:
     // Poco::Logger * logger = &Poco::Logger::get("ProjectRelParser");
-};   
+};
 }

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

@@ -15,6 +15,7 @@
  * limitations under the License.
  */
 #include "JoinRelParser.h"
+#include <optional>
 
 #include <Core/Block.h>
 #include <Functions/FunctionFactory.h>
@@ -27,10 +28,10 @@
 #include <Interpreters/TableJoin.h>
 #include <Join/BroadCastJoinBuilder.h>
 #include <Join/StorageJoinFromReadBuffer.h>
+#include <Operator/EarlyStopStep.h>
 #include <Parser/AdvancedParametersParseUtil.h>
 #include <Parser/SerializedPlanParser.h>
 #include <Parsers/ASTIdentifier.h>
-#include <Operator/EarlyStopStep.h>
 #include <Processors/QueryPlan/ExpressionStep.h>
 #include <Processors/QueryPlan/FilterStep.h>
 #include <Processors/QueryPlan/JoinStep.h>
@@ -56,8 +57,8 @@ namespace local_engine
 {
 std::shared_ptr<DB::TableJoin> createDefaultTableJoin(substrait::JoinRel_JoinType join_type, bool is_existence_join, ContextPtr & context)
 {
-    auto table_join = std::make_shared<TableJoin>(
-        context->getSettingsRef(), context->getGlobalTemporaryVolume(), context->getTempDataOnDisk());
+    auto table_join
+        = std::make_shared<TableJoin>(context->getSettingsRef(), context->getGlobalTemporaryVolume(), context->getTempDataOnDisk());
 
     std::pair<DB::JoinKind, DB::JoinStrictness> kind_and_strictness = JoinUtil::getJoinKindAndStrictness(join_type, is_existence_join);
     table_join->setKind(kind_and_strictness.first);
@@ -79,25 +80,27 @@ JoinRelParser::parse(DB::QueryPlanPtr /*query_plan*/, const substrait::Rel & /*r
     throw Exception(ErrorCodes::LOGICAL_ERROR, "join node has 2 inputs, can't call parse().");
 }
 
-const substrait::Rel & JoinRelParser::getSingleInput(const substrait::Rel & /*rel*/)
-{
-    throw Exception(ErrorCodes::LOGICAL_ERROR, "join node has 2 inputs, can't call getSingleInput().");
-}
-
-DB::QueryPlanPtr JoinRelParser::parseOp(const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack)
+std::vector<const substrait::Rel *> JoinRelParser::getInputs(const substrait::Rel & rel)
 {
     const auto & join = rel.join();
     if (!join.has_left() || !join.has_right())
     {
         throw Exception(ErrorCodes::BAD_ARGUMENTS, "left table or right table is missing.");
     }
 
-    rel_stack.push_back(&rel);
-    auto left_plan = getPlanParser()->parseOp(join.left(), rel_stack);
-    auto right_plan = getPlanParser()->parseOp(join.right(), rel_stack);
-    rel_stack.pop_back();
+    return {&join.left(), &join.right()};
+}
+std::optional<const substrait::Rel *> JoinRelParser::getSingleInput(const substrait::Rel & /*rel*/)
+{
+    throw Exception(ErrorCodes::LOGICAL_ERROR, "join node has 2 inputs, can't call getSingleInput().");
+}
 
-    return parseJoin(join, std::move(left_plan), std::move(right_plan));
+DB::QueryPlanPtr JoinRelParser::parse(
+    std::vector<DB::QueryPlanPtr> & input_plans_, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_)
+{
+    assert(input_plans_.size() == 2);
+    const auto & join = rel.join();
+    return parseJoin(join, std::move(input_plans_[0]), std::move(input_plans_[1]));
 }
 
 std::unordered_set<DB::JoinTableSide> JoinRelParser::extractTableSidesFromExpression(
@@ -282,13 +285,22 @@ DB::QueryPlanPtr JoinRelParser::parseJoin(const substrait::JoinRel & join, DB::Q
                 auto input_header = left->getCurrentDataStream().header;
                 DB::ActionsDAG filter_is_not_null_dag{input_header.getColumnsWithTypeAndName()};
                 // when is_null_aware_anti_join is true, there is only one join key
-                const auto * key_field = filter_is_not_null_dag.getInputs()[join.expression().scalar_function().arguments().at(0).value().selection().direct_reference().struct_field().field()];
+                const auto * key_field = filter_is_not_null_dag.getInputs()[join.expression()
+                                                                                .scalar_function()
+                                                                                .arguments()
+                                                                                .at(0)
+                                                                                .value()
+                                                                                .selection()
+                                                                                .direct_reference()
+                                                                                .struct_field()
+                                                                                .field()];
 
                 auto result_node = filter_is_not_null_dag.tryFindInOutputs(key_field->result_name);
                 // add a function isNotNull to filter the null key on the left side
                 const auto * cond_node = plan_parser->toFunctionNode(filter_is_not_null_dag, "isNotNull", {result_node});
                 filter_is_not_null_dag.addOrReplaceInOutputs(*cond_node);
-                auto filter_step = std::make_unique<FilterStep>(left->getCurrentDataStream(), std::move(filter_is_not_null_dag), cond_node->result_name, true);
+                auto filter_step = std::make_unique<FilterStep>(
+                    left->getCurrentDataStream(), std::move(filter_is_not_null_dag), cond_node->result_name, true);
                 left->addStep(std::move(filter_step));
             }
             // other case: is_empty_hash_table, don't need to handle
@@ -342,8 +354,7 @@ DB::QueryPlanPtr JoinRelParser::parseJoin(const substrait::JoinRel & join, DB::Q
             = couldRewriteToMultiJoinOnClauses(table_join->getOnlyClause(), join_on_clauses, join, left_header, right_header);
         if (is_multi_join_on_clauses && join_config.prefer_multi_join_on_clauses && join_opt_info.right_table_rows > 0
             && join_opt_info.partitions_num > 0
-            && join_opt_info.right_table_rows / join_opt_info.partitions_num
-                < join_config.multi_join_on_clauses_build_side_rows_limit)
+            && join_opt_info.right_table_rows / join_opt_info.partitions_num < join_config.multi_join_on_clauses_build_side_rows_limit)
         {
             query_plan = buildMultiOnClauseHashJoin(table_join, std::move(left), std::move(right), join_on_clauses);
         }

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

@@ -20,7 +20,7 @@
 #include <unordered_set>
 #include <Core/Joins.h>
 #include <Interpreters/TableJoin.h>
-#include <Parser/RelParser.h>
+#include <Parser/RelParsers/RelParser.h>
 #include <substrait/algebra.pb.h>
 
 namespace DB
@@ -42,9 +42,11 @@ class JoinRelParser : public RelParser
     DB::QueryPlanPtr
     parse(DB::QueryPlanPtr query_plan, const substrait::Rel & sort_rel, std::list<const substrait::Rel *> & rel_stack_) override;
 
-    DB::QueryPlanPtr parseOp(const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack) override;
+    DB::QueryPlanPtr parse(
+        std::vector<DB::QueryPlanPtr> & input_plans_, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_) override;
 
-    const substrait::Rel & getSingleInput(const substrait::Rel & rel) override;
+    std::vector<const substrait::Rel *> getInputs(const substrait::Rel & rel) override;
+    std::optional<const substrait::Rel *> getSingleInput(const substrait::Rel & rel) override;
 
 private:
     std::unordered_map<std::string, std::string> & function_mapping;
@@ -69,8 +71,8 @@ class JoinRelParser : public RelParser
 
     void existenceJoinPostProject(DB::QueryPlan & plan, const DB::Names & left_input_cols);
 
-    static std::unordered_set<DB::JoinTableSide> extractTableSidesFromExpression(
-        const substrait::Expression & expr, const DB::Block & left_header, const DB::Block & right_header);
+    static std::unordered_set<DB::JoinTableSide>
+    extractTableSidesFromExpression(const substrait::Expression & expr, const DB::Block & left_header, const DB::Block & right_header);
 
     bool couldRewriteToMultiJoinOnClauses(
         const DB::TableJoin::JoinOnClause & prefix_clause,