Cherry pick v2.6.0 1018 (#3136)

* Pass graph profile param into storage and cleanup graph storage client interfaces (#3026) * Cleanup graph storage client interfaces * Fix compile * Format * Fix shadow compile error * Fix storage core when exit (#3050) * although I don't have time to check why it works, but this do fix the issue * fix ut * Accumulate toss bug fix during test. (#3091) * add some debug info * accumulate bug fix for TOSS Co-authored-by: Doodle <[email protected]> * Support more validation when create space on an empty zone (#3065) Co-authored-by: Yee <[email protected]> Co-authored-by: Doodle <[email protected]> * small tck fix (#3087) Co-authored-by: jie.wang <[email protected]> * cascading deletion and addition (#3059) Co-authored-by: Doodle <[email protected]> * fix toss switch not set properly (#3119) Co-authored-by: Yee <[email protected]> Co-authored-by: Doodle <[email protected]> Co-authored-by: Yee <[email protected]> Co-authored-by: Doodle <[email protected]> Co-authored-by: [email protected] <[email protected]> Co-authored-by: yaphet <[email protected]> Co-authored-by: kyle.cao <[email protected]> Co-authored-by: jie.wang <[email protected]>
vesoft-inc · Oct 19, 2021 · 650776b · 650776b
1 parent 9243bba
commit 650776b
Show file tree

Hide file tree

Showing 34 changed files with 535 additions and 394 deletions.
diff --git a/src/clients/storage/GraphStorageClient.cpp b/src/clients/storage/GraphStorageClient.cpp
diff --git a/src/clients/storage/GraphStorageClient.h b/src/clients/storage/GraphStorageClient.h
@@ -16,6 +16,9 @@
 namespace nebula {
 namespace storage {
 
+template <typename T>
+using StorageRpcRespFuture = folly::SemiFuture<StorageRpcResponse<T>>;
+
 /**
  * A wrapper class for GraphStorageServiceAsyncClient thrift API
  *
@@ -24,19 +27,32 @@ namespace storage {
 class GraphStorageClient : public StorageClientBase<cpp2::GraphStorageServiceAsyncClient> {
   FRIEND_TEST(StorageClientTest, LeaderChangeTest);
 
-  using Parent = StorageClientBase<cpp2::GraphStorageServiceAsyncClient>;
-
  public:
+  struct CommonRequestParam {
+    GraphSpaceID space;
+    SessionID session;
+    ExecutionPlanID plan;
+    bool profile{false};
+    bool useExperimentalFeature{false};
+    folly::EventBase* evb{nullptr};
+
+    CommonRequestParam(GraphSpaceID space_,
+                       SessionID sess,
+                       ExecutionPlanID plan_,
+                       bool profile_ = false,
+                       bool experimental = false,
+                       folly::EventBase* evb_ = nullptr);
+
+    cpp2::RequestCommon toReqCommon() const;
+  };
+
   GraphStorageClient(std::shared_ptr<folly::IOThreadPoolExecutor> ioThreadPool,
                      meta::MetaClient* metaClient)
-      : Parent(ioThreadPool, metaClient) {}
+      : StorageClientBase<cpp2::GraphStorageServiceAsyncClient>(ioThreadPool, metaClient) {}
   virtual ~GraphStorageClient() {}
 
-  folly::SemiFuture<StorageRpcResponse<cpp2::GetNeighborsResponse>> getNeighbors(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
-      bool profile,
+  StorageRpcRespFuture<cpp2::GetNeighborsResponse> getNeighbors(
+      const CommonRequestParam& param,
       std::vector<std::string> colNames,
       // The first column has to be the VertexID
       const std::vector<Row>& vertices,
@@ -50,110 +66,70 @@ class GraphStorageClient : public StorageClientBase<cpp2::GraphStorageServiceAsy
       bool random = false,
       const std::vector<cpp2::OrderBy>& orderBy = std::vector<cpp2::OrderBy>(),
       int64_t limit = std::numeric_limits<int64_t>::max(),
-      const Expression* filter = nullptr,
-      folly::EventBase* evb = nullptr);
+      const Expression* filter = nullptr);
 
-  folly::SemiFuture<StorageRpcResponse<cpp2::GetPropResponse>> getProps(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
+  StorageRpcRespFuture<cpp2::GetPropResponse> getProps(
+      const CommonRequestParam& param,
       const DataSet& input,
       const std::vector<cpp2::VertexProp>* vertexProps,
       const std::vector<cpp2::EdgeProp>* edgeProps,
       const std::vector<cpp2::Expr>* expressions,
       bool dedup = false,
       const std::vector<cpp2::OrderBy>& orderBy = std::vector<cpp2::OrderBy>(),
       int64_t limit = std::numeric_limits<int64_t>::max(),
-      const Expression* filter = nullptr,
-      folly::EventBase* evb = nullptr);
+      const Expression* filter = nullptr);
 
-  folly::SemiFuture<StorageRpcResponse<cpp2::ExecResponse>> addVertices(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
+  StorageRpcRespFuture<cpp2::ExecResponse> addVertices(
+      const CommonRequestParam& param,
       std::vector<cpp2::NewVertex> vertices,
       std::unordered_map<TagID, std::vector<std::string>> propNames,
-      bool ifNotExists,
-      folly::EventBase* evb = nullptr);
-
-  folly::SemiFuture<StorageRpcResponse<cpp2::ExecResponse>> addEdges(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
-      std::vector<cpp2::NewEdge> edges,
-      std::vector<std::string> propNames,
-      bool ifNotExists,
-      folly::EventBase* evb = nullptr,
-      bool useToss = false);
-
-  folly::SemiFuture<StorageRpcResponse<cpp2::ExecResponse>> deleteEdges(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
-      std::vector<storage::cpp2::EdgeKey> edges,
-      folly::EventBase* evb = nullptr);
-
-  folly::SemiFuture<StorageRpcResponse<cpp2::ExecResponse>> deleteVertices(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
-      std::vector<Value> ids,
-      folly::EventBase* evb = nullptr);
-
-  folly::SemiFuture<StorageRpcResponse<cpp2::ExecResponse>> deleteTags(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
-      std::vector<cpp2::DelTags> delTags,
-      folly::EventBase* evb = nullptr);
+      bool ifNotExists);
+
+  StorageRpcRespFuture<cpp2::ExecResponse> addEdges(const CommonRequestParam& param,
+                                                    std::vector<cpp2::NewEdge> edges,
+                                                    std::vector<std::string> propNames,
+                                                    bool ifNotExists);
+
+  StorageRpcRespFuture<cpp2::ExecResponse> deleteEdges(const CommonRequestParam& param,
+                                                       std::vector<storage::cpp2::EdgeKey> edges);
+
+  StorageRpcRespFuture<cpp2::ExecResponse> deleteVertices(const CommonRequestParam& param,
+                                                          std::vector<Value> ids);
+
+  StorageRpcRespFuture<cpp2::ExecResponse> deleteTags(const CommonRequestParam& param,
+                                                      std::vector<cpp2::DelTags> delTags);
 
   folly::Future<StatusOr<storage::cpp2::UpdateResponse>> updateVertex(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
+      const CommonRequestParam& param,
       Value vertexId,
       TagID tagId,
       std::vector<cpp2::UpdatedProp> updatedProps,
       bool insertable,
       std::vector<std::string> returnProps,
-      std::string condition,
-      folly::EventBase* evb = nullptr);
+      std::string condition);
 
   folly::Future<StatusOr<storage::cpp2::UpdateResponse>> updateEdge(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
+      const CommonRequestParam& param,
       storage::cpp2::EdgeKey edgeKey,
       std::vector<cpp2::UpdatedProp> updatedProps,
       bool insertable,
       std::vector<std::string> returnProps,
-      std::string condition,
-      folly::EventBase* evb = nullptr,
-      bool useExperimentalFeature = false);
+      std::string condition);
 
   folly::Future<StatusOr<cpp2::GetUUIDResp>> getUUID(GraphSpaceID space,
                                                      const std::string& name,
                                                      folly::EventBase* evb = nullptr);
 
-  folly::SemiFuture<StorageRpcResponse<cpp2::LookupIndexResp>> lookupIndex(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
-      bool profile,
+  StorageRpcRespFuture<cpp2::LookupIndexResp> lookupIndex(
+      const CommonRequestParam& param,
       const std::vector<storage::cpp2::IndexQueryContext>& contexts,
       bool isEdge,
       int32_t tagOrEdge,
       const std::vector<std::string>& returnCols,
-      int64_t limit,
-      folly::EventBase* evb = nullptr);
+      int64_t limit);
 
-  folly::SemiFuture<StorageRpcResponse<cpp2::GetNeighborsResponse>> lookupAndTraverse(
-      GraphSpaceID space,
-      SessionID session,
-      ExecutionPlanID plan,
-      cpp2::IndexSpec indexSpec,
-      cpp2::TraverseSpec traverseSpec,
-      folly::EventBase* evb = nullptr);
+  StorageRpcRespFuture<cpp2::GetNeighborsResponse> lookupAndTraverse(
+      const CommonRequestParam& param, cpp2::IndexSpec indexSpec, cpp2::TraverseSpec traverseSpec);
 
   folly::Future<StatusOr<cpp2::ScanEdgeResponse>> scanEdge(cpp2::ScanEdgeRequest req,
                                                            folly::EventBase* evb = nullptr);
@@ -178,10 +154,6 @@ class GraphStorageClient : public StorageClientBase<cpp2::GraphStorageServiceAsy
 
   StatusOr<std::function<const VertexID&(const cpp2::DelTags&)>> getIdFromDelTags(
       GraphSpaceID space) const;
-
-  cpp2::RequestCommon makeRequestCommon(SessionID sessionId,
-                                        ExecutionPlanID planId,
-                                        bool profile = false);
 };
 
 }  // namespace storage

diff --git a/src/clients/storage/InternalStorageClient.cpp b/src/clients/storage/InternalStorageClient.cpp
@@ -95,7 +95,7 @@ void InternalStorageClient::chainAddEdges(cpp2::AddEdgesRequest& directReq,
   }
   HostAddr& leader = optLeader.value();
   leader.port += kInternalPortOffset;
-  VLOG(1) << "leader host: " << leader;
+  VLOG(2) << "leader host: " << leader;
 
   cpp2::ChainAddEdgesRequest chainReq = makeChainAddReq(directReq, termId, optVersion);
   auto resp = getResponse(

diff --git a/src/graph/executor/mutate/DeleteExecutor.cpp b/src/graph/executor/mutate/DeleteExecutor.cpp
@@ -12,6 +12,8 @@
 #include "graph/util/SchemaUtil.h"
 #include "graph/util/ScopedTimer.h"
 
+using nebula::storage::GraphStorageClient;
+
 namespace nebula {
 namespace graph {
 
@@ -61,10 +63,12 @@ folly::Future<Status> DeleteVerticesExecutor::deleteVertices() {
   }
   auto spaceId = spaceInfo.id;
   time::Duration deleteVertTime;
+  auto plan = qctx()->plan();
+  GraphStorageClient::CommonRequestParam param(
+      spaceId, qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
   return qctx()
       ->getStorageClient()
-      ->deleteVertices(
-          spaceId, qctx()->rctx()->session()->id(), qctx()->plan()->id(), std::move(vertices))
+      ->deleteVertices(param, std::move(vertices))
       .via(runner())
       .ensure([deleteVertTime]() {
         VLOG(1) << "Delete vertices time: " << deleteVertTime.elapsedInUSec() << "us";
@@ -115,10 +119,12 @@ folly::Future<Status> DeleteTagsExecutor::deleteTags() {
 
   auto spaceId = spaceInfo.id;
   time::Duration deleteTagTime;
+  auto plan = qctx()->plan();
+  GraphStorageClient::CommonRequestParam param(
+      spaceId, qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
   return qctx()
       ->getStorageClient()
-      ->deleteTags(
-          spaceId, qctx()->rctx()->session()->id(), qctx()->plan()->id(), std::move(delTags))
+      ->deleteTags(param, std::move(delTags))
       .via(runner())
       .ensure([deleteTagTime]() {
         VLOG(1) << "Delete vertices time: " << deleteTagTime.elapsedInUSec() << "us";
@@ -198,10 +204,12 @@ folly::Future<Status> DeleteEdgesExecutor::deleteEdges() {
 
   auto spaceId = spaceInfo.id;
   time::Duration deleteEdgeTime;
+  auto plan = qctx()->plan();
+  GraphStorageClient::CommonRequestParam param(
+      spaceId, qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
   return qctx()
       ->getStorageClient()
-      ->deleteEdges(
-          spaceId, qctx()->rctx()->session()->id(), qctx()->plan()->id(), std::move(edgeKeys))
+      ->deleteEdges(param, std::move(edgeKeys))
       .via(runner())
       .ensure([deleteEdgeTime]() {
         VLOG(1) << "Delete edge time: " << deleteEdgeTime.elapsedInUSec() << "us";

diff --git a/src/graph/executor/mutate/InsertExecutor.cpp b/src/graph/executor/mutate/InsertExecutor.cpp
@@ -11,6 +11,8 @@
 #include "graph/service/GraphFlags.h"
 #include "graph/util/ScopedTimer.h"
 
+using nebula::storage::GraphStorageClient;
+
 namespace nebula {
 namespace graph {
 
@@ -21,14 +23,12 @@ folly::Future<Status> InsertVerticesExecutor::insertVertices() {
 
   auto *ivNode = asNode<InsertVertices>(node());
   time::Duration addVertTime;
+  auto plan = qctx()->plan();
+  GraphStorageClient::CommonRequestParam param(
+      ivNode->getSpace(), qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
   return qctx()
       ->getStorageClient()
-      ->addVertices(ivNode->getSpace(),
-                    qctx()->rctx()->session()->id(),
-                    qctx()->plan()->id(),
-                    ivNode->getVertices(),
-                    ivNode->getPropNames(),
-                    ivNode->getIfNotExists())
+      ->addVertices(param, ivNode->getVertices(), ivNode->getPropNames(), ivNode->getIfNotExists())
       .via(runner())
       .ensure([addVertTime]() {
         VLOG(1) << "Add vertices time: " << addVertTime.elapsedInUSec() << "us";
@@ -47,16 +47,13 @@ folly::Future<Status> InsertEdgesExecutor::insertEdges() {
 
   auto *ieNode = asNode<InsertEdges>(node());
   time::Duration addEdgeTime;
+  auto plan = qctx()->plan();
+  GraphStorageClient::CommonRequestParam param(
+      ieNode->getSpace(), qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
+  param.useExperimentalFeature = FLAGS_enable_experimental_feature;
   return qctx()
       ->getStorageClient()
-      ->addEdges(ieNode->getSpace(),
-                 qctx()->rctx()->session()->id(),
-                 qctx()->plan()->id(),
-                 ieNode->getEdges(),
-                 ieNode->getPropNames(),
-                 ieNode->getIfNotExists(),
-                 nullptr,
-                 FLAGS_enable_experimental_feature)
+      ->addEdges(param, ieNode->getEdges(), ieNode->getPropNames(), ieNode->getIfNotExists())
       .via(runner())
       .ensure(
           [addEdgeTime]() { VLOG(1) << "Add edge time: " << addEdgeTime.elapsedInUSec() << "us"; })

diff --git a/src/graph/executor/mutate/UpdateExecutor.cpp b/src/graph/executor/mutate/UpdateExecutor.cpp
@@ -12,6 +12,8 @@
 #include "graph/util/SchemaUtil.h"
 #include "graph/util/ScopedTimer.h"
 
+using nebula::storage::GraphStorageClient;
+
 namespace nebula {
 namespace graph {
 
@@ -46,11 +48,13 @@ folly::Future<Status> UpdateVertexExecutor::execute() {
   auto *uvNode = asNode<UpdateVertex>(node());
   yieldNames_ = uvNode->getYieldNames();
   time::Duration updateVertTime;
+  auto plan = qctx()->plan();
+  auto sess = qctx()->rctx()->session();
+  GraphStorageClient::CommonRequestParam param(
+      uvNode->getSpaceId(), sess->id(), plan->id(), plan->isProfileEnabled());
   return qctx()
       ->getStorageClient()
-      ->updateVertex(uvNode->getSpaceId(),
-                     qctx()->rctx()->session()->id(),
-                     qctx()->plan()->id(),
+      ->updateVertex(param,
                      uvNode->getVId(),
                      uvNode->getTagId(),
                      uvNode->getUpdatedProps(),
@@ -96,18 +100,18 @@ folly::Future<Status> UpdateEdgeExecutor::execute() {
   yieldNames_ = ueNode->getYieldNames();
 
   time::Duration updateEdgeTime;
+  auto plan = qctx()->plan();
+  GraphStorageClient::CommonRequestParam param(
+      ueNode->getSpaceId(), qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
+  param.useExperimentalFeature = FLAGS_enable_experimental_feature;
   return qctx()
       ->getStorageClient()
-      ->updateEdge(ueNode->getSpaceId(),
-                   qctx()->rctx()->session()->id(),
-                   qctx()->plan()->id(),
+      ->updateEdge(param,
                    edgeKey,
                    ueNode->getUpdatedProps(),
                    ueNode->getInsertable(),
                    ueNode->getReturnProps(),
-                   ueNode->getCondition(),
-                   nullptr,
-                   FLAGS_enable_experimental_feature)
+                   ueNode->getCondition())
       .via(runner())
       .ensure([updateEdgeTime]() {
         VLOG(1) << "Update edge time: " << updateEdgeTime.elapsedInUSec() << "us";

diff --git a/src/graph/executor/query/GetEdgesExecutor.cpp b/src/graph/executor/query/GetEdgesExecutor.cpp
@@ -66,10 +66,12 @@ folly::Future<Status> GetEdgesExecutor::getEdges() {
   }
 
   time::Duration getPropsTime;
+  GraphStorageClient::CommonRequestParam param(ge->space(),
+                                               qctx()->rctx()->session()->id(),
+                                               qctx()->plan()->id(),
+                                               qctx()->plan()->isProfileEnabled());
   return DCHECK_NOTNULL(client)
-      ->getProps(ge->space(),
-                 qctx()->rctx()->session()->id(),
-                 qctx()->plan()->id(),
+      ->getProps(param,
                  std::move(edges),
                  nullptr,
                  ge->props(),

diff --git a/src/graph/executor/query/GetNeighborsExecutor.cpp b/src/graph/executor/query/GetNeighborsExecutor.cpp
@@ -43,11 +43,12 @@ folly::Future<Status> GetNeighborsExecutor::execute() {
   time::Duration getNbrTime;
   GraphStorageClient* storageClient = qctx_->getStorageClient();
   QueryExpressionContext qec(qctx()->ectx());
+  GraphStorageClient::CommonRequestParam param(gn_->space(),
+                                               qctx()->rctx()->session()->id(),
+                                               qctx()->plan()->id(),
+                                               qctx()->plan()->isProfileEnabled());
   return storageClient
-      ->getNeighbors(gn_->space(),
-                     qctx()->rctx()->session()->id(),
-                     qctx()->plan()->id(),
-                     qctx()->plan()->isProfileEnabled(),
+      ->getNeighbors(param,
                      std::move(reqDs.colNames),
                      std::move(reqDs.rows),
                      gn_->edgeTypes(),