Return variable if not empty (#1233)

* Add return sentence. * Add test. * Address comment and add more tests. * Rebase and fix conflict. * Update.
vesoft-inc · Nov 21, 2019 · 317203f · 317203f
1 parent 5ace754
commit 317203f
Show file tree

Hide file tree

Showing 51 changed files with 435 additions and 68 deletions.
diff --git a/src/graph/AlterEdgeExecutor.cpp b/src/graph/AlterEdgeExecutor.cpp
@@ -57,7 +57,7 @@ void AlterEdgeExecutor::execute() {
         }
 
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/AlterTagExecutor.cpp b/src/graph/AlterTagExecutor.cpp
@@ -58,7 +58,7 @@ void AlterTagExecutor::execute() {
         }
 
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/AssignmentExecutor.cpp b/src/graph/AssignmentExecutor.cpp
@@ -27,9 +27,9 @@ Status AssignmentExecutor::prepare() {
         DCHECK(onError_);
         onError_(std::move(s));
     };
-    auto onFinish = [this] () {
+    auto onFinish = [this] (Executor::ProcessControl ctr) {
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(ctr);
     };
     auto onResult = [this] (std::unique_ptr<InterimResult> result) {
         ectx()->variableHolder()->add(*var_, std::move(result));

diff --git a/src/graph/BalanceExecutor.cpp b/src/graph/BalanceExecutor.cpp
@@ -56,7 +56,7 @@ void BalanceExecutor::balanceLeader() {
             return;
         }
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
@@ -101,7 +101,7 @@ void BalanceExecutor::balanceData(bool isStop) {
         resp_->set_rows(std::move(rows));
 
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
@@ -172,7 +172,7 @@ void BalanceExecutor::showBalancePlan() {
         rows.back().set_columns(std::move(row));
         resp_->set_rows(std::move(rows));
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/CMakeLists.txt b/src/graph/CMakeLists.txt
@@ -49,6 +49,7 @@ nebula_add_library(
     FindPathExecutor.cpp
     LimitExecutor.cpp
     GroupByExecutor.cpp
+    ReturnExecutor.cpp
 )
 
 nebula_add_library(

diff --git a/src/graph/ConfigExecutor.cpp b/src/graph/ConfigExecutor.cpp
@@ -78,7 +78,7 @@ void ConfigExecutor::showVariables() {
         resp_->set_rows(std::move(rows));
 
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
@@ -154,7 +154,7 @@ void ConfigExecutor::setVariables() {
         resp_ = std::make_unique<cpp2::ExecutionResponse>();
 
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
@@ -209,7 +209,7 @@ void ConfigExecutor::getVariables() {
         resp_->set_rows(std::move(rows));
 
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/CreateEdgeExecutor.cpp b/src/graph/CreateEdgeExecutor.cpp
@@ -58,7 +58,7 @@ void CreateEdgeExecutor::execute() {
         }
 
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/CreateSpaceExecutor.cpp b/src/graph/CreateSpaceExecutor.cpp
@@ -54,7 +54,7 @@ void CreateSpaceExecutor::execute() {
             return;
         }
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/CreateTagExecutor.cpp b/src/graph/CreateTagExecutor.cpp
@@ -58,7 +58,7 @@ void CreateTagExecutor::execute() {
         }
 
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/DeleteVertexExecutor.cpp b/src/graph/DeleteVertexExecutor.cpp
@@ -103,7 +103,7 @@ void DeleteVertexExecutor::deleteVertex() {
             return;
         }
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
         return;
     };
 

diff --git a/src/graph/DescribeEdgeExecutor.cpp b/src/graph/DescribeEdgeExecutor.cpp
@@ -57,7 +57,7 @@ void DescribeEdgeExecutor::execute() {
 
         resp_->set_rows(std::move(rows));
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/DescribeSpaceExecutor.cpp b/src/graph/DescribeSpaceExecutor.cpp
@@ -48,7 +48,7 @@ void DescribeSpaceExecutor::execute() {
         rows.back().set_columns(std::move(row));
         resp_->set_rows(std::move(rows));
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/DescribeTagExecutor.cpp b/src/graph/DescribeTagExecutor.cpp
@@ -57,7 +57,7 @@ void DescribeTagExecutor::execute() {
 
         resp_->set_rows(std::move(rows));
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/DownloadExecutor.cpp b/src/graph/DownloadExecutor.cpp
@@ -74,7 +74,7 @@ void DownloadExecutor::execute() {
         }
         resp_ = std::make_unique<cpp2::ExecutionResponse>();
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/DropEdgeExecutor.cpp b/src/graph/DropEdgeExecutor.cpp
@@ -38,7 +38,7 @@ void DropEdgeExecutor::execute() {
             return;
         }
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/DropSpaceExecutor.cpp b/src/graph/DropSpaceExecutor.cpp
@@ -42,7 +42,7 @@ void DropSpaceExecutor::execute() {
             ectx()->rctx()->session()->setSpace("", -1);
         }
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/DropTagExecutor.cpp b/src/graph/DropTagExecutor.cpp
@@ -38,7 +38,7 @@ void DropTagExecutor::execute() {
             return;
         }
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/ExecutionPlan.cpp b/src/graph/ExecutionPlan.cpp
@@ -38,7 +38,8 @@ void ExecutionPlan::execute() {
     }
 
     // Prepared
-    auto onFinish = [this] () {
+    auto onFinish = [this] (Executor::ProcessControl ctr) {
+        UNUSED(ctr);
         this->onFinish();
     };
     auto onError = [this] (Status s) {

diff --git a/src/graph/Executor.cpp b/src/graph/Executor.cpp
@@ -46,6 +46,7 @@
 #include "graph/FindPathExecutor.h"
 #include "graph/LimitExecutor.h"
 #include "graph/GroupByExecutor.h"
+#include "graph/ReturnExecutor.h"
 
 namespace nebula {
 namespace graph {
@@ -159,6 +160,9 @@ std::unique_ptr<Executor> Executor::makeExecutor(Sentence *sentence) {
         case Sentence::Kind::kLimit:
             executor = std::make_unique<LimitExecutor>(sentence, ectx());
             break;
+        case Sentence::Kind::kReturn:
+            executor = std::make_unique<ReturnExecutor>(sentence, ectx());
+            break;
         case Sentence::Kind::kUnknown:
             LOG(ERROR) << "Sentence kind unknown";
             return nullptr;

diff --git a/src/graph/Executor.h b/src/graph/Executor.h
@@ -45,10 +45,15 @@ class Executor : public cpp::NonCopyable, public cpp::NonMovable {
 
     virtual const char* name() const = 0;
 
+    enum ProcessControl : uint8_t {
+        kNext = 0,
+        kReturn,
+    };
+
     /**
      * Set callback to be invoked when this executor is finished(normally).
      */
-    void setOnFinish(std::function<void()> onFinish) {
+    void setOnFinish(std::function<void(ProcessControl)> onFinish) {
         onFinish_ = onFinish;
     }
     /**
@@ -102,8 +107,8 @@ class Executor : public cpp::NonCopyable, public cpp::NonMovable {
     }
 
 protected:
-    ExecutionContext                            *ectx_;
-    std::function<void()>                       onFinish_;
+    ExecutionContext                           *ectx_;
+    std::function<void(ProcessControl)>         onFinish_;
     std::function<void(Status)>                 onError_;
 };
 

diff --git a/src/graph/FetchExecutor.cpp b/src/graph/FetchExecutor.cpp
@@ -103,7 +103,7 @@ void FetchExecutor::onEmptyInputs() {
     } else if (resp_ == nullptr) {
         resp_ = std::make_unique<cpp2::ExecutionResponse>();
     }
-    onFinish_();
+    onFinish_(Executor::ProcessControl::kNext);
 }
 
 Status FetchExecutor::getOutputSchema(
@@ -153,7 +153,7 @@ void FetchExecutor::finishExecution(std::unique_ptr<RowSetWriter> rsWriter) {
         }
     }
     DCHECK(onFinish_);
-    onFinish_();
+    onFinish_(Executor::ProcessControl::kNext);
 }
 }  // namespace graph
 }  // namespace nebula
diff --git a/src/graph/FindPathExecutor.cpp b/src/graph/FindPathExecutor.cpp
@@ -173,7 +173,7 @@ void FindPathExecutor::execute() {
 void FindPathExecutor::getNeighborsAndFindPath() {
     // We meet the dead end.
     if (fromVids_.empty() || toVids_.empty()) {
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
         return;
     }
 
@@ -270,7 +270,7 @@ void FindPathExecutor::findPath() {
     // if frontiersF meets frontiersT, we found an even path
     if (!intersect.empty()) {
         if (shortest_ && targetNotFound_.empty()) {
-            onFinish_();
+            onFinish_(Executor::ProcessControl::kNext);
             return;
         }
         for (auto intersectId : intersect) {
@@ -280,7 +280,7 @@ void FindPathExecutor::findPath() {
 
     if (isFinalStep() ||
          (shortest_ && targetNotFound_.empty())) {
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
         return;
     } else {
         VLOG(2) << "Current step:" << currentStep_;

diff --git a/src/graph/GoExecutor.cpp b/src/graph/GoExecutor.cpp
@@ -581,7 +581,7 @@ void GoExecutor::finishExecution(RpcResponse &&rpcResp) {
         }
     }
     DCHECK(onFinish_);
-    onFinish_();
+    onFinish_(Executor::ProcessControl::kNext);
 }
 
 StatusOr<std::vector<storage::cpp2::PropDef>> GoExecutor::getStepOutProps() {
@@ -796,7 +796,7 @@ void GoExecutor::onEmptyInputs() {
     } else if (resp_ == nullptr) {
         resp_ = std::make_unique<cpp2::ExecutionResponse>();
     }
-    onFinish_();
+    onFinish_(Executor::ProcessControl::kNext);
 }
 
 

diff --git a/src/graph/GroupByExecutor.cpp b/src/graph/GroupByExecutor.cpp
@@ -183,7 +183,7 @@ void GroupByExecutor::execute() {
 
     if (rows_.empty()) {
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
         return;
     }
 
@@ -219,7 +219,7 @@ void GroupByExecutor::execute() {
     }
 
     DCHECK(onFinish_);
-    onFinish_();
+    onFinish_(Executor::ProcessControl::kNext);
 }
 
 

diff --git a/src/graph/IngestExecutor.cpp b/src/graph/IngestExecutor.cpp
@@ -62,7 +62,7 @@ void IngestExecutor::execute() {
         }
         resp_ = std::make_unique<cpp2::ExecutionResponse>();
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/InsertEdgeExecutor.cpp b/src/graph/InsertEdgeExecutor.cpp
@@ -230,7 +230,7 @@ void InsertEdgeExecutor::execute() {
             return;
         }
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/InsertVertexExecutor.cpp b/src/graph/InsertVertexExecutor.cpp
@@ -212,7 +212,7 @@ void InsertVertexExecutor::execute() {
             return;
         }
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {

diff --git a/src/graph/InterimResult.h b/src/graph/InterimResult.h
@@ -63,7 +63,7 @@ class InterimResult final {
         return rsReader_->schema();
     }
 
-    std::vector<std::string> getColNames() {
+    std::vector<std::string> getColNames() const {
         // Once getColNames called, colNames_ would be invalid
         return std::move(colNames_);
     }

diff --git a/src/graph/LimitExecutor.cpp b/src/graph/LimitExecutor.cpp
@@ -33,14 +33,15 @@ void LimitExecutor::execute() {
     FLOG_INFO("Executing Limit: %s", sentence_->toString().c_str());
     if (inputs_ == nullptr || count_ == 0) {
         DCHECK(onFinish_);
-        onFinish_();
+        onFinish_(Executor::ProcessControl::kNext);
         return;
     }
 
     auto ret = inputs_->getRows();
     if (!ret.ok()) {
-        DCHECK(onFinish_);
-        onFinish_();
+        LOG(ERROR) << "Get rows failed: " << ret.status();
+        DCHECK(onError_);
+        onError_(std::move(ret).status());
         return;
     }
     auto inRows = std::move(ret).value();
@@ -61,7 +62,7 @@ void LimitExecutor::execute() {
     }
 
     DCHECK(onFinish_);
-    onFinish_();
+    onFinish_(Executor::ProcessControl::kNext);
 }
-Original file line number
+Diff line change
@@ Expand Up / @@ -103,7 +103,7 @@ void DeleteVertexExecutor::deleteVertex() { @@
                 return;
             }
             DCHECK(onFinish_);
-            onFinish_();
+            onFinish_(Executor::ProcessControl::kNext);
             return;
         };
@@ Expand Down @@