add Subgraph executor & refactor subgraph (#660)

* add subgraph executor

* add subgraph excutor & refactor subgraph

* fix error

* fix error second

* fix miss edge

* fix test case

* remove useless code

* change to hash(Edge)

* change hash(edge)

* change hash(string) to hash(Value)

* address comment

* address comment

* remove useless code

src/executor/CMakeLists.txt

@@ -35,6 +35,7 @@ nebula_add_library(
     algo/ProduceSemiShortestPathExecutor.cpp
     algo/ProduceAllPathsExecutor.cpp
     algo/CartesianProductExecutor.cpp
+    algo/SubgraphExecutor.cpp
     admin/SwitchSpaceExecutor.cpp
     admin/CreateUserExecutor.cpp
     admin/DropUserExecutor.cpp

src/executor/Executor.cpp

@@ -49,6 +49,7 @@
 #include "executor/algo/ConjunctPathExecutor.h"
 #include "executor/algo/ProduceAllPathsExecutor.h"
 #include "executor/algo/CartesianProductExecutor.h"
+#include "executor/algo/SubgraphExecutor.h"
 #include "executor/logic/LoopExecutor.h"
 #include "executor/logic/PassThroughExecutor.h"
 #include "executor/logic/SelectExecutor.h"
@@ -418,6 +419,9 @@ Executor *Executor::makeExecutor(QueryContext *qctx, const PlanNode *node) {
         case PlanNode::Kind::kCartesianProduct: {
             return pool->add(new CartesianProductExecutor(node, qctx));
         }
+        case PlanNode::Kind::kSubgraph: {
+            return pool->add(new SubgraphExecutor(node, qctx));
+        }
         case PlanNode::Kind::kAddGroup: {
             return pool->add(new AddGroupExecutor(node, qctx));
         }

src/executor/algo/SubgraphExecutor.cpp

+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include "executor/algo/SubgraphExecutor.h"
+
+#include "planner/Algo.h"
+
+namespace nebula {
+namespace graph {
+
+folly::Future<Status> SubgraphExecutor::execute() {
+    SCOPED_TIMER(&execTime_);
+    auto* subgraph = asNode<Subgraph>(node());
+    DataSet ds;
+    ds.colNames = subgraph->colNames();
+
+    uint32_t steps = subgraph->steps();
+    const auto& currentStepVal = ectx_->getValue(subgraph->currentStepVar());
+    DCHECK(currentStepVal.isInt());
+    auto currentStep = currentStepVal.getInt();
+    VLOG(1) << "Current Step is: " << currentStep << " Total Steps is: " << steps;
+
+    if (currentStep == steps) {
+        oneMoreStep();
+        return finish(ResultBuilder().value(Value(std::move(ds))).finish());
+    }
+
+    VLOG(1) << "input: " << subgraph->inputVar() << " output: " << node()->outputVar();
+    auto iter = ectx_->getResult(subgraph->inputVar()).iter();
+    DCHECK(iter && iter->isGetNeighborsIter());
+    ds.rows.reserve(iter->size());
+    if (currentStep == 1) {
+        for (; iter->valid(); iter->next()) {
+            const auto& src = iter->getColumn(nebula::kVid);
+            historyVids_.emplace(src);
+        }
+        iter->reset();
+    }
+    for (; iter->valid(); iter->next()) {
+        const auto& dst = iter->getEdgeProp("*", nebula::kDst);
+        if (historyVids_.emplace(dst).second) {
+            Row row;
+            row.values.emplace_back(std::move(dst));
+            ds.rows.emplace_back(std::move(row));
+        }
+    }
+
+    VLOG(1) << "Next step vid is : " << ds;
+    return finish(ResultBuilder().value(Value(std::move(ds))).finish());
+}
+
+void SubgraphExecutor::oneMoreStep() {
+    auto* subgraph = asNode<Subgraph>(node());
+    auto output = subgraph->oneMoreStepOutput();
+    VLOG(1) << "OneMoreStep Input: " << subgraph->inputVar() << " Output: " << output;
+    auto iter = ectx_->getResult(subgraph->inputVar()).iter();
+    DCHECK(iter && iter->isGetNeighborsIter());
+
+    ResultBuilder builder;
+    builder.value(iter->valuePtr());
+    while (iter->valid()) {
+        const auto& dst = iter->getEdgeProp("*", nebula::kDst);
+        if (historyVids_.find(dst) == historyVids_.end()) {
+            iter->unstableErase();
+        } else {
+            iter->next();
+        }
+    }
+    iter->reset();
+    builder.iter(std::move(iter));
+    ectx_->setResult(output, builder.finish());
+}
+
+}   // namespace graph
+}   // namespace nebula

src/executor/algo/SubgraphExecutor.h

+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#ifndef EXECUTOR_ALGO_SUBGRAPHEXECUTOR_H_
+#define EXECUTOR_ALGO_SUBGRAPHEXECUTOR_H_
+
+#include "executor/Executor.h"
+
+namespace nebula {
+namespace graph {
+class SubgraphExecutor : public Executor {
+public:
+    SubgraphExecutor(const PlanNode* node, QueryContext* qctx)
+        : Executor("SubgraphExecutor", node, qctx) {}
+
+    folly::Future<Status> execute() override;
+
+private:
+    void oneMoreStep();
+
+private:
+    std::unordered_set<Value>   historyVids_;
+};
+
+}   // namespace graph
+}   // namespace nebula
+#endif  // EXECUTOR_ALGO_SUBGRAPHEXECUTOR_H_

src/executor/query/DataCollectExecutor.cpp

@@ -62,11 +62,14 @@ Status DataCollectExecutor::collectSubgraph(const std::vector<std::string>& vars
     ds.colNames = std::move(colNames_);
     // the subgraph not need duplicate vertices or edges, so dedup here directly
     std::unordered_set<Value> uniqueVids;
-    std::unordered_set<std::tuple<Value, std::string, int64_t, Value>> uniqueEdges;
-    for (auto& var : vars) {
-        auto& hist = ectx_->getHistory(var);
-        for (auto& result : hist) {
-            auto iter = result.iter();
+    std::unordered_set<std::tuple<Value, EdgeType, EdgeRanking, Value>> uniqueEdges;
+    for (auto i = vars.begin(); i != vars.end(); ++i) {
+        const auto& hist = ectx_->getHistory(*i);
+        for (auto j = hist.begin(); j != hist.end(); ++j) {
+            if (i == vars.begin() && j == hist.end() - 1) {
+                continue;
+            }
+            auto iter = (*j).iter();
             if (!iter->isGetNeighborsIter()) {
                 std::stringstream msg;
                 msg << "Iterator should be kind of GetNeighborIter, but was: " << iter->kind();
@@ -85,14 +88,14 @@ Status DataCollectExecutor::collectSubgraph(const std::vector<std::string>& vars
                 }
             }
             auto originEdges = gnIter->getEdges();
-            for (auto& e : originEdges.values) {
-                if (!e.isEdge()) {
+            for (auto& edge : originEdges.values) {
+                if (!edge.isEdge()) {
                     continue;
                 }
-                auto edgeKey = std::make_tuple(
-                    e.getEdge().src, e.getEdge().name, e.getEdge().ranking, e.getEdge().dst);
+                const auto& e = edge.getEdge();
+                auto edgeKey = std::make_tuple(e.src, e.type, e.ranking, e.dst);
                 if (uniqueEdges.emplace(std::move(edgeKey)).second) {
-                    edges.emplace_back(std::move(e));
+                    edges.emplace_back(std::move(edge));
                 }
             }
             ds.rows.emplace_back(Row({std::move(vertices), std::move(edges)}));

src/executor/test/DataCollectTest.cpp

@@ -87,12 +87,16 @@ protected:
 
             List datasets;
             datasets.values.emplace_back(std::move(ds1));
-            datasets.values.emplace_back(std::move(ds2));
 
             ResultBuilder builder;
             builder.value(Value(std::move(datasets))).iter(Iterator::Kind::kGetNeighbors);
             qctx_->symTable()->newVariable("input_datasets");
             qctx_->ectx()->setResult("input_datasets", builder.finish());
+            qctx_->ectx()->setResult("input_datasets",
+                                     ResultBuilder()
+                                         .value(Value(std::move(ds2)))
+                                         .iter(Iterator::Kind::kGetNeighbors)
+                                         .finish());
         }
         {
             DataSet ds;
@@ -115,6 +119,11 @@ protected:
                             "_edge:+edge1:prop1:prop2:_dst:_rank",
                             "_expr"};
             qctx_->symTable()->newVariable("empty_get_neighbors");
+            qctx_->ectx()->setResult("empty_get_neighbors",
+                                     ResultBuilder()
+                                         .value(Value(ds))
+                                         .iter(Iterator::Kind::kGetNeighbors)
+                                         .finish());
             qctx_->ectx()->setResult("empty_get_neighbors",
                                      ResultBuilder()
                                          .value(Value(std::move(ds)))
@@ -140,8 +149,8 @@ TEST_F(DataCollectTest, CollectSubgraph) {
 
     DataSet expected;
     expected.colNames = {"_vertices", "_edges"};
-    auto& input = qctx_->ectx()->getResult("input_datasets");
-    auto iter = input.iter();
+    auto& hist = qctx_->ectx()->getHistory("input_datasets");
+    auto iter = hist[0].iter();
     auto* gNIter = static_cast<GetNeighborsIter*>(iter.get());
     Row row;
     std::unordered_set<Value> vids;

src/planner/Algo.h

@@ -139,6 +139,45 @@ private:
     std::vector<std::vector<std::string>> allColNames_;
 };
 
+class Subgraph final : public SingleInputNode {
+public:
+    static Subgraph* make(QueryContext* qctx,
+                          PlanNode* input,
+                          const std::string& oneMoreStepOutput,
+                          const std::string& currentStepVar,
+                          uint32_t steps) {
+        return qctx->objPool()->add(
+            new Subgraph(qctx, input, oneMoreStepOutput, currentStepVar, steps));
+    }
+
+    const std::string& oneMoreStepOutput() const {
+        return oneMoreStepOutput_;
+    }
+
+    const std::string& currentStepVar() const {
+        return currentStepVar_;
+    }
+
+    uint32_t steps() const {
+        return steps_;
+    }
+
+private:
+    Subgraph(QueryContext* qctx,
+             PlanNode* input,
+             const std::string& oneMoreStepOutput,
+             const std::string& currentStepVar,
+             uint32_t steps)
+        : SingleInputNode(qctx, Kind::kSubgraph, input),
+          oneMoreStepOutput_(oneMoreStepOutput),
+          currentStepVar_(currentStepVar),
+          steps_(steps) {}
+
+    std::string oneMoreStepOutput_;
+    std::string currentStepVar_;
+    uint32_t steps_;
+};
+
 }  // namespace graph
 }  // namespace nebula
 #endif  // PLANNER_ALGO_H_

src/planner/PlanNode.cpp

@@ -207,6 +207,8 @@ const char* PlanNode::toString(PlanNode::Kind kind) {
             return "ProduceAllPaths";
         case Kind::kCartesianProduct:
             return "CartesianProduct";
+        case Kind::kSubgraph:
+            return "Subgraph";
         // Group and Zone
         case Kind::kAddGroup:
             return "AddGroup";

src/planner/PlanNode.h

@@ -118,6 +118,7 @@ public:
         kConjunctPath,
         kProduceAllPaths,
         kCartesianProduct,
+        kSubgraph,
         // zone related
         kAddGroup,
         kDropGroup,

src/validator/GetSubgraphValidator.cpp

@@ -14,6 +14,7 @@
 #include "parser/TraverseSentences.h"
 #include "planner/Logic.h"
 #include "planner/Query.h"
+#include "planner/Algo.h"
 
 namespace nebula {
 namespace graph {
@@ -101,54 +102,6 @@ Status GetSubgraphValidator::validateBothInOutBound(BothInOutClause* out) {
     return Status::OK();
 }
 
-/*
- * 1 steps   history: collectVid{0}
- * 2 steps   history: collectVid{-1} collectVid{0}
- * 3 steps   history: collectVid{-2} collectVid{-1} collectVid{0}
- * ...
- * n steps   history: collectVid{-n+1} ...  collectVid{-1} collectVid{0}
- */
-Expression* GetSubgraphValidator::buildFilterCondition(int64_t step) {
-    // where *._dst IN startVids(*._dst IN runtimeStartVar_) OR *._dst IN collectVid{0}[0][0] OR
-    // *._dst IN collectVid{-1}[0][0] OR ... *._dst IN collectVid{-step+1}[0][0]
-    if (step == 1) {
-        Expression* left = nullptr;
-        if (!runtimeStartVar_.empty()) {
-            auto* startDataSet = new VersionedVariableExpression(new std::string(runtimeStartVar_),
-                                                                 new ConstantExpression(0));
-            auto* runtimeStartList = new SubscriptExpression(
-                new SubscriptExpression(startDataSet, new ConstantExpression(0)),
-                new ConstantExpression(0));
-            left = new RelationalExpression(Expression::Kind::kRelIn,
-                                            new EdgeDstIdExpression(new std::string("*")),
-                                            runtimeStartList);
-        } else {
-            left = new RelationalExpression(Expression::Kind::kRelIn,
-                                            new EdgeDstIdExpression(new std::string("*")),
-                                            new SetExpression(startVidList_.release()));
-        }
-        auto* lastestVidsDataSet = new VersionedVariableExpression(new std::string(collectVar_),
-                                                                   new ConstantExpression(0));
-        auto* lastestVidsList = new SubscriptExpression(
-            new SubscriptExpression(lastestVidsDataSet, new ConstantExpression(0)),
-            new ConstantExpression(0));
-        auto* right = new RelationalExpression(Expression::Kind::kRelIn,
-                                               new EdgeDstIdExpression(new std::string("*")),
-                                               lastestVidsList);
-        return new LogicalExpression(Expression::Kind::kLogicalOr, left, right);
-    }
-    auto* historyVidsDataSet = new VersionedVariableExpression(new std::string(collectVar_),
-                                                               new ConstantExpression(1 - step));
-    auto* historyVidsList = new SubscriptExpression(
-        new SubscriptExpression(historyVidsDataSet, new ConstantExpression(0)),
-        new ConstantExpression(0));
-    auto* left = new RelationalExpression(
-        Expression::Kind::kRelIn, new EdgeDstIdExpression(new std::string("*")), historyVidsList);
-    auto* right = buildFilterCondition(step - 1);
-    auto* result = new LogicalExpression(Expression::Kind::kLogicalOr, left, right);
-    return result;
-}
-
 StatusOr<GetNeighbors::EdgeProps> GetSubgraphValidator::buildEdgeProps() {
     if (edgeTypes_.empty()) {
         auto allEdgePropResult = buildAllEdgeProp();
@@ -243,39 +196,19 @@ Status GetSubgraphValidator::zeroStep(PlanNode* depend, const std::string& input
 
 Status GetSubgraphValidator::toPlan() {
     auto& space = vctx_->whichSpace();
-    // gn <- filter <- DataCollect
-    //  |
-    // loop(step) -> Agg(collect) -> project -> gn -> bodyStart
     auto* bodyStart = StartNode::make(qctx_);
 
     std::string startVidsVar;
-    SingleInputNode* collectRunTimeStartVids = nullptr;
-
+    PlanNode* loopDep = nullptr;
     if (!from_.vids.empty() && from_.originalSrc == nullptr) {
         buildConstantInput(from_, startVidsVar);
     } else {
-        PlanNode* dedupStartVid = buildRuntimeInput(from_, projectStartVid_);
-        startVidsVar = dedupStartVid->outputVar();
-        // collect runtime startVids
-        auto var = vctx_->anonVarGen()->getVar();
-        auto* column = new VariablePropertyExpression(new std::string(var), new std::string(kVid));
-        auto* func = new AggregateExpression(new std::string("COLLECT_SET"), column, true);
-        qctx_->objPool()->add(func);
-        collectRunTimeStartVids =
-            Aggregate::make(qctx_,
-                            dedupStartVid,
-                            {},
-                            {func});
-        collectRunTimeStartVids->setInputVar(dedupStartVid->outputVar());
-        collectRunTimeStartVids->setColNames({kVid});
-        runtimeStartVar_ = collectRunTimeStartVids->outputVar();
+        loopDep = buildRuntimeInput(from_, projectStartVid_);
+        startVidsVar = loopDep->outputVar();
     }
 
     if (steps_.steps == 0) {
-        if (collectRunTimeStartVids == nullptr) {
-            return zeroStep(bodyStart, startVidsVar);
-        }
-        return zeroStep(collectRunTimeStartVids, startVidsVar);
+        return zeroStep(loopDep == nullptr ? bodyStart : loopDep, startVidsVar);
     }
 
     auto vertexPropsResult = buildVertexProp();
@@ -288,51 +221,19 @@ Status GetSubgraphValidator::toPlan() {
     NG_RETURN_IF_ERROR(edgePropsResult);
     gn->setEdgeProps(
         std::make_unique<std::vector<storage::cpp2::EdgeProp>>(*edgePropsResult.value()));
-    gn->setEdgeDirection(storage::cpp2::EdgeDirection::BOTH);
     gn->setInputVar(startVidsVar);
 
-    auto* projectVids = projectDstVidsFromGN(gn, startVidsVar);
+    auto oneMoreStepOutput = vctx_->anonVarGen()->getVar();
+    auto* subgraph = Subgraph::make(qctx_, gn, oneMoreStepOutput, loopSteps_, steps_.steps + 1);
+    subgraph->setOutputVar(startVidsVar);
+    subgraph->setColNames({nebula::kVid});
 
-    auto var = vctx_->anonVarGen()->getVar();
-    auto* column = new VariablePropertyExpression(new std::string(var), new std::string(kVid));
-    auto* func = new AggregateExpression(new std::string("COLLECT_SET"), column, true);
-    qctx_->objPool()->add(func);
-    auto* collect =
-        Aggregate::make(qctx_,
-                        projectVids,
-                        {},
-                        {func});
-    collect->setInputVar(projectVids->outputVar());
-    collect->setColNames({kVid});
-    collectVar_ = collect->outputVar();
+    auto* loopCondition = buildNStepLoopCondition(steps_.steps + 1);
+    auto* loop = Loop::make(qctx_, loopDep, subgraph, loopCondition);
 
-    // TODO(jmq) add condition when gn get empty result
-    auto* condition = buildNStepLoopCondition(steps_.steps);
-    auto* loop = Loop::make(qctx_, collectRunTimeStartVids, collect, condition);
-
-    vertexPropsResult = buildVertexProp();
-    NG_RETURN_IF_ERROR(vertexPropsResult);
-    auto edgeProps = std::make_unique<std::vector<storage::cpp2::EdgeProp>>();
-    auto* gn1 = GetNeighbors::make(qctx_, loop, space.id);
-    gn1->setSrc(from_.src);
-    gn1->setVertexProps(std::make_unique<std::vector<storage::cpp2::VertexProp>>(
-        std::move(vertexPropsResult).value()));
-    auto allEdgePropResult = buildAllEdgeProp();
-    NG_RETURN_IF_ERROR(allEdgePropResult);
-    gn1->setEdgeProps(std::make_unique<std::vector<storage::cpp2::EdgeProp>>(
-        std::move(allEdgePropResult).value()));
-    gn1->setEdgeDirection(storage::cpp2::EdgeDirection::BOTH);
-    gn1->setInputVar(projectVids->outputVar());
-
-    auto* filter =
-        Filter::make(qctx_, gn1, qctx_->objPool()->add(buildFilterCondition(steps_.steps)));
-    filter->setInputVar(gn1->outputVar());
-    filter->setColNames({kVid});
-
-    // datacollect
-    std::vector<std::string> collects = {gn->outputVar(), filter->outputVar()};
+    std::vector<std::string> collects = {gn->outputVar(), oneMoreStepOutput};
     auto* dc =
-        DataCollect::make(qctx_, filter, DataCollect::CollectKind::kSubgraph, std::move(collects));
+        DataCollect::make(qctx_, loop, DataCollect::CollectKind::kSubgraph, std::move(collects));
     dc->setColNames({"_vertices", "_edges"});
     root_ = dc;
     tail_ = projectStartVid_ != nullptr ? projectStartVid_ : loop;

src/validator/GetSubgraphValidator.h

@@ -29,8 +29,6 @@ private:
 
     Status validateBothInOutBound(BothInOutClause* out);
 
-    Expression* buildFilterCondition(int64_t step);
-
     StatusOr<GetNeighbors::EdgeProps> buildEdgeProps();
 
     Status zeroStep(PlanNode* depend, const std::string& inputVar);
@@ -44,8 +42,6 @@ private:
 
 private:
     std::unordered_set<EdgeType>                edgeTypes_;
-    std::string                                 collectVar_;
-    std::string                                 runtimeStartVar_;
 };
 }  // namespace graph
 }  // namespace nebula

src/validator/TraversalValidator.cpp

@@ -199,13 +199,11 @@ PlanNode* TraversalValidator::buildRuntimeInput(Starts& starts, PlanNode*& proje
 Expression* TraversalValidator::buildNStepLoopCondition(uint32_t steps) const {
     VLOG(1) << "steps: " << steps;
     // ++loopSteps{0} <= steps
-    auto loopSteps = vctx_->anonVarGen()->getVar();
-    qctx_->ectx()->setValue(loopSteps, 0);
+    qctx_->ectx()->setValue(loopSteps_, 0);
     return qctx_->objPool()->add(new RelationalExpression(
         Expression::Kind::kRelLE,
-        new UnaryExpression(
-            Expression::Kind::kUnaryIncr,
-            new VersionedVariableExpression(new std::string(loopSteps), new ConstantExpression(0))),
+        new UnaryExpression(Expression::Kind::kUnaryIncr,
+                            new VariableExpression(new std::string(loopSteps_))),
         new ConstantExpression(static_cast<int32_t>(steps))));
 }
 

src/validator/TraversalValidator.h

@@ -47,6 +47,7 @@ protected:
 protected:
     TraversalValidator(Sentence* sentence, QueryContext* qctx) : Validator(sentence, qctx) {
         startVidList_.reset(new ExpressionList());
+        loopSteps_ = vctx_->anonVarGen()->getVar();
     }
 
     Status validateStarts(const VerticesClause* clause, Starts& starts);
@@ -68,6 +69,7 @@ protected:
     Steps                 steps_;
     std::string           srcVidColName_;
     PlanNode*             projectStartVid_{nullptr};
+    std::string           loopSteps_;
 
     std::unique_ptr<ExpressionList>  startVidList_;
 };

src/validator/test/GetSubgraphValidatorTest.cpp

@@ -24,13 +24,9 @@ TEST_F(GetSubgraphValidatorTest, Base) {
         std::string query = "GET SUBGRAPH FROM \"1\"";
         std::vector<PlanNode::Kind> expected = {
             PK::kDataCollect,
-            PK::kFilter,
-            PK::kGetNeighbors,
             PK::kLoop,
             PK::kStart,
-            PK::kAggregate,
-            PK::kDedup,
-            PK::kProject,
+            PK::kSubgraph,
             PK::kGetNeighbors,
             PK::kStart,
         };
@@ -40,13 +36,9 @@ TEST_F(GetSubgraphValidatorTest, Base) {
         std::string query = "GET SUBGRAPH 3 STEPS FROM \"1\"";
         std::vector<PlanNode::Kind> expected = {
             PK::kDataCollect,
-            PK::kFilter,
-            PK::kGetNeighbors,
             PK::kLoop,
             PK::kStart,
-            PK::kAggregate,
-            PK::kDedup,
-            PK::kProject,
+            PK::kSubgraph,
             PK::kGetNeighbors,
             PK::kStart,
         };
@@ -56,13 +48,9 @@ TEST_F(GetSubgraphValidatorTest, Base) {
         std::string query = "GET SUBGRAPH FROM \"1\" BOTH like";
         std::vector<PlanNode::Kind> expected = {
             PK::kDataCollect,
-            PK::kFilter,
-            PK::kGetNeighbors,
             PK::kLoop,
             PK::kStart,
-            PK::kAggregate,
-            PK::kDedup,
-            PK::kProject,
+            PK::kSubgraph,
             PK::kGetNeighbors,
             PK::kStart,
         };
@@ -72,13 +60,9 @@ TEST_F(GetSubgraphValidatorTest, Base) {
         std::string query = "GET SUBGRAPH FROM \"1\", \"2\" IN like";
         std::vector<PlanNode::Kind> expected = {
             PK::kDataCollect,
-            PK::kFilter,
-            PK::kGetNeighbors,
             PK::kLoop,
             PK::kStart,
-            PK::kAggregate,
-            PK::kDedup,
-            PK::kProject,
+            PK::kSubgraph,
             PK::kGetNeighbors,
             PK::kStart,
         };
@@ -92,19 +76,14 @@ TEST_F(GetSubgraphValidatorTest, Input) {
             "GO FROM \"1\" OVER like YIELD like._src AS src | GET SUBGRAPH FROM $-.src";
         std::vector<PlanNode::Kind> expected = {
             PK::kDataCollect,
-            PK::kFilter,
-            PK::kGetNeighbors,
             PK::kLoop,
-            PK::kAggregate,
-            PK::kAggregate,
-            PK::kDedup,
             PK::kDedup,
+            PK::kSubgraph,
             PK::kProject,
-            PK::kProject,
-            PK::kProject,
-            PK::kGetNeighbors,
             PK::kGetNeighbors,
+            PK::kProject,
             PK::kStart,
+            PK::kGetNeighbors,
             PK::kStart,
         };
         EXPECT_TRUE(checkResult(query, expected));
@@ -114,19 +93,14 @@ TEST_F(GetSubgraphValidatorTest, Input) {
             "$a = GO FROM \"1\" OVER like YIELD like._src AS src; GET SUBGRAPH FROM $a.src";
         std::vector<PlanNode::Kind> expected = {
             PK::kDataCollect,
-            PK::kFilter,
-            PK::kGetNeighbors,
             PK::kLoop,
-            PK::kAggregate,
-            PK::kAggregate,
-            PK::kDedup,
             PK::kDedup,
+            PK::kSubgraph,
             PK::kProject,
-            PK::kProject,
-            PK::kProject,
-            PK::kGetNeighbors,
             PK::kGetNeighbors,
+            PK::kProject,
             PK::kStart,
+            PK::kGetNeighbors,
             PK::kStart,
         };
         EXPECT_TRUE(checkResult(query, expected));
@@ -155,7 +129,6 @@ TEST_F(GetSubgraphValidatorTest, Input) {
         std::vector<PlanNode::Kind> expected = {
             PK::kAggregate,
             PK::kGetVertices,
-            PK::kAggregate,
             PK::kDedup,
             PK::kProject,
             PK::kProject,
@@ -170,7 +143,6 @@ TEST_F(GetSubgraphValidatorTest, Input) {
         std::vector<PlanNode::Kind> expected = {
             PK::kAggregate,
             PK::kGetVertices,
-            PK::kAggregate,
             PK::kDedup,
             PK::kProject,
             PK::kProject,