delete pathbaseexecutor

vesoft-inc · Mar 21, 2023 · 7dc2b72 · 7dc2b72
1 parent cd67eae
commit 7dc2b72
Show file tree

Hide file tree

Showing 7 changed files with 105 additions and 282 deletions.
diff --git a/src/graph/executor/CMakeLists.txt b/src/graph/executor/CMakeLists.txt
@@ -6,7 +6,6 @@ nebula_add_library(
  executor_obj OBJECT
  Executor.cpp
  StorageAccessExecutor.cpp
- PathBaseExecutor.cpp
  logic/LoopExecutor.cpp
  logic/PassThroughExecutor.cpp
  logic/StartExecutor.cpp

diff --git a/src/graph/executor/PathBaseExecutor.cpp b/src/graph/executor/PathBaseExecutor.cpp
diff --git a/src/graph/executor/PathBaseExecutor.h b/src/graph/executor/PathBaseExecutor.h
diff --git a/src/graph/executor/StorageAccessExecutor.cpp b/src/graph/executor/StorageAccessExecutor.cpp
@@ -8,11 +8,13 @@
 
 #include "graph/context/Iterator.h"
 #include "graph/context/QueryExpressionContext.h"
+#include "graph/service/GraphFlags.h"
 #include "graph/util/SchemaUtil.h"
 #include "graph/util/Utils.h"
 #include "interface/gen-cpp2/meta_types.h"
 
 using apache::thrift::optional_field_ref;
+using nebula::storage::StorageClient;
 
 namespace nebula {
 namespace graph {
@@ -148,5 +150,85 @@ StatusOr<std::vector<Value>> StorageAccessExecutor::buildRequestListByVidType(It
  return internal::buildRequestList<std::string>(space, exprCtx, iter, expr, dedup, isCypher);
 }
 
+bool StorageAccessExecutor::hasSameEdge(const std::vector<Value> &edgeList, const Edge &edge) {
+ for (auto &leftEdge : edgeList) {
+ if (!leftEdge.isEdge()) {
+ continue;
+ }
+ if (edge.keyEqual(leftEdge.getEdge())) {
+ return true;
+ }
+ }
+ return false;
+}
+
+folly::Future<std::vector<Value>> StorageAccessExecutor::getProps(
+ const std::vector<Value> &vids, const std::vector<VertexProp> *vertexPropPtr) {
+ nebula::DataSet vertices({kVid});
+ vertices.rows.reserve(vids.size());
+ for (auto &vid : vids) {
+ vertices.emplace_back(Row({vid}));
+ }
+ StorageClient *storageClient = qctx_->getStorageClient();
+ StorageClient::CommonRequestParam param(qctx_->rctx()->session()->space().id,
+ qctx_->rctx()->session()->id(),
+ qctx_->plan()->id(),
+ qctx_->plan()->isProfileEnabled());
+ return DCHECK_NOTNULL(storageClient)
+ ->getProps(
+ param, std::move(vertices), vertexPropPtr, nullptr, nullptr, false, {}, -1, nullptr)
+ .via(runner())
+ .thenValue([this](PropRpcResponse &&resp) {
+ addStats(resp);
+ return handlePropResp(std::move(resp));
+ });
+}
+
+std::vector<Value> StorageAccessExecutor::handlePropResp(PropRpcResponse &&resps) {
+ std::vector<Value> vertices;
+ auto result = handleCompleteness(resps, FLAGS_accept_partial_success);
+ if (!result.ok()) {
+ LOG(WARNING) << "GetProp partial fail";
+ return vertices;
+ }
+ nebula::DataSet v;
+ for (auto &resp : resps.responses()) {
+ if (resp.props_ref().has_value()) {
+ if (UNLIKELY(!v.append(std::move(*resp.props_ref())))) {
+ // it's impossible according to the interface
+ LOG(WARNING) << "Heterogeneous props dataset";
+ }
+ } else {
+ LOG(WARNING) << "GetProp partial success";
+ }
+ }
+ auto val = std::make_shared<Value>(std::move(v));
+ auto iter = std::make_unique<PropIter>(val);
+ vertices.reserve(iter->size());
+ for (; iter->valid(); iter->next()) {
+ vertices.emplace_back(iter->getVertex());
+ }
+ return vertices;
+}
+
+void StorageAccessExecutor::addGetNeighborStats(RpcResponse &resp, size_t stepNum, bool reverse) {
+ folly::dynamic stats = folly::dynamic::array();
+ auto &hostLatency = resp.hostLatency();
+ for (size_t i = 0; i < hostLatency.size(); ++i) {
+ size_t size = 0u;
+ auto &result = resp.responses()[i];
+ if (result.vertices_ref().has_value()) {
+ size = (*result.vertices_ref()).size();
+ }
+ auto info = util::collectRespProfileData(result.result, hostLatency[i], size);
+ stats.push_back(std::move(info));
+ }
+
+ auto key = folly::sformat("{}step[{}]", reverse ? "reverse " : "", stepNum);
+ statsLock_.lock();
+ otherStats_.emplace(key, folly::toPrettyJson(stats));
+ statsLock_.unlock();
+}
+
 } // namespace graph
 } // namespace nebula