Fix total rpc time in stats (#5307)

src/graph/context/Iterator.cpp

@@ -62,7 +62,7 @@ void GetNeighborsIter::goToFirstEdge() {
          ++currentRow_) {
       colIdx_ = currentDs_->colLowerBound + 1;
       while (colIdx_ < currentDs_->colUpperBound && !valid_) {
-        const auto& currentCol = currentRow_->operator[](colIdx_);
+        const auto& currentCol = (*currentRow_)[colIdx_];
         if (!currentCol.isList() || currentCol.getList().empty()) {
           ++colIdx_;
           continue;
@@ -229,7 +229,7 @@ void GetNeighborsIter::next() {
 
   while (++edgeIdx_ > -1) {
     if (edgeIdx_ < edgeIdxUpperBound_) {
-      const auto& currentEdge = currentCol_->operator[](edgeIdx_);
+      const auto& currentEdge = (*currentCol_)[edgeIdx_];
       if (!currentEdge.isList()) {
         continue;
       }
@@ -366,10 +366,10 @@ const Value& GetNeighborsIter::getTagProp(const std::string& tag, const std::str
   auto& row = *currentRow_;
   if (tag == "*") {
     for (auto& index : currentDs_->tagPropsMap) {
-      auto propIndex = index.second.propIndices.find(prop);
-      if (propIndex != index.second.propIndices.end()) {
+      auto propIndexIter = index.second.propIndices.find(prop);
+      if (propIndexIter != index.second.propIndices.end()) {
         colId = index.second.colIdx;
-        propId = propIndex->second;
+        propId = propIndexIter->second;
         DCHECK_GT(row.size(), colId);
         if (row[colId].empty()) {
           continue;
@@ -393,12 +393,12 @@ const Value& GetNeighborsIter::getTagProp(const std::string& tag, const std::str
     if (index == tagPropIndices.end()) {
       return Value::kEmpty;
     }
-    auto propIndex = index->second.propIndices.find(prop);
-    if (propIndex == index->second.propIndices.end()) {
+    auto propIndexIter = index->second.propIndices.find(prop);
+    if (propIndexIter == index->second.propIndices.end()) {
       return Value::kEmpty;
     }
     colId = index->second.colIdx;
-    propId = propIndex->second;
+    propId = propIndexIter->second;
     DCHECK_GT(row.size(), colId);
     if (row[colId].empty()) {
       return Value::kEmpty;
@@ -427,8 +427,7 @@ const Value& GetNeighborsIter::getEdgeProp(const std::string& edge, const std::s
   }
   auto index = currentDs_->edgePropsMap.find(currentEdge);
   if (index == currentDs_->edgePropsMap.end()) {
-    DLOG(INFO) << "No edge found: " << edge;
-    DLOG(INFO) << "Current edge: " << currentEdge;
+    DLOG(INFO) << "No edge found: " << edge << " Current edge: " << currentEdge;
     return Value::kEmpty;
   }
   auto propIndex = index->second.propIndices.find(prop);
@@ -466,10 +465,9 @@ Value GetNeighborsIter::getVertex(const std::string& name) {
     Tag tag;
     tag.name = tagProp.first;
     for (size_t i = 0; i < propList.size(); ++i) {
-      if (tagPropNameList[i] == nebula::kTag) {
-        continue;
+      if (tagPropNameList[i] != nebula::kTag) {
+        tag.props.emplace(tagPropNameList[i], propList[i]);
       }
-      tag.props.emplace(tagPropNameList[i], propList[i]);
     }
     vertex.tags.emplace_back(std::move(tag));
   }

src/graph/executor/algo/BatchShortestPath.cpp

@@ -9,9 +9,12 @@
 #include "sys/sysinfo.h"
 
 using nebula::storage::StorageClient;
+
 DECLARE_uint32(num_path_thread);
+
 namespace nebula {
 namespace graph {
+
 folly::Future<Status> BatchShortestPath::execute(const HashSet& startVids,
                                                  const HashSet& endVids,
                                                  DataSet* result) {
@@ -550,12 +553,13 @@ size_t BatchShortestPath::splitTask(const HashSet& startVids, const HashSet& end
       ++count;
     }
   }
-  std::stringstream ss;
-  ss << "{\n"
-     << "startVids' size : " << startVidsSize << " endVids's size : " << endVidsSize;
-  ss << " thread num : " << threadNum;
-  ss << " start blocks : " << startSlices << " end blocks : " << endSlices << "\n}";
-  stats_->emplace(folly::sformat("split task "), ss.str());
+  folly::dynamic obj = folly::dynamic::object();
+  obj.insert("startVids' size", startVidsSize);
+  obj.insert("endVids's size", endVidsSize);
+  obj.insert("thread num", threadNum);
+  obj.insert("start blocks", startSlices);
+  obj.insert("end blocks", endSlices);
+  stats_->emplace("split task", folly::toPrettyJson(obj));
   return startSlices * endSlices;
 }
 

src/graph/executor/algo/BatchShortestPath.h

@@ -9,6 +9,7 @@
 
 namespace nebula {
 namespace graph {
+
 class BatchShortestPath final : public ShortestPathBase {
  public:
   BatchShortestPath(const ShortestPath* node,

src/graph/executor/algo/ShortestPathBase.cpp

@@ -173,13 +173,17 @@ void ShortestPathBase::addStats(RpcResponse& resp,
     if (result.vertices_ref().has_value()) {
       size = (*result.vertices_ref()).size();
     }
-    auto info = util::collectRespProfileData(result.result, hostLatency[i], size, timeInUSec);
+    auto info = util::collectRespProfileData(result.result, hostLatency[i], size);
     stats.push_back(std::move(info));
   }
 
+  folly::dynamic stepObj = folly::dynamic::object();
+  stepObj.insert("total_rpc_time", folly::sformat("{}(us)", timeInUSec));
+  stepObj.insert("storage", stats);
+
   auto key = folly::sformat("{}step[{}]", reverse ? "reverse " : "", stepNum);
   statsLock_.lock();
-  stats_->emplace(key, folly::toPrettyJson(stats));
+  stats_->emplace(key, folly::toPrettyJson(stepObj));
   statsLock_.unlock();
 }
 
@@ -188,12 +192,16 @@ void ShortestPathBase::addStats(PropRpcResponse& resp, int64_t timeInUSec) const
   auto& hostLatency = resp.hostLatency();
   for (size_t i = 0; i < hostLatency.size(); ++i) {
     const auto& result = resp.responses()[i].get_result();
-    auto info = util::collectRespProfileData(result, hostLatency[i], 0, timeInUSec);
+    auto info = util::collectRespProfileData(result, hostLatency[i], 0);
     stats.push_back(std::move(info));
   }
 
+  folly::dynamic propObj = folly::dynamic::object();
+  propObj.insert("storage", stats);
+  propObj.insert("total_rpc_time", folly::sformat("{}(us)", timeInUSec));
+
   statsLock_.lock();
-  stats_->emplace("get_prop", folly::toPrettyJson(stats));
+  stats_->emplace("get_prop", folly::toPrettyJson(propObj));
   statsLock_.unlock();
 }
 

src/graph/executor/algo/ShortestPathBase.h

@@ -13,8 +13,10 @@ using nebula::storage::StorageRpcResponse;
 using nebula::storage::cpp2::GetNeighborsResponse;
 using RpcResponse = StorageRpcResponse<GetNeighborsResponse>;
 using PropRpcResponse = StorageRpcResponse<nebula::storage::cpp2::GetPropResponse>;
+
 namespace nebula {
 namespace graph {
+
 class ShortestPathBase {
  public:
   using HashSet = robin_hood::unordered_flat_set<Value, std::hash<Value>>;

src/graph/executor/algo/ShortestPathExecutor.cpp

@@ -13,8 +13,10 @@
 using nebula::storage::StorageClient;
 
 DEFINE_uint32(num_path_thread, 0, "number of concurrent threads when do shortest path");
+
 namespace nebula {
 namespace graph {
+
 folly::Future<Status> ShortestPathExecutor::execute() {
   // MemoryTrackerVerified
   SCOPED_TIMER(&execTime_);

src/graph/executor/algo/SingleShortestPath.cpp

@@ -10,6 +10,7 @@ using nebula::storage::StorageClient;
 
 namespace nebula {
 namespace graph {
+
 folly::Future<Status> SingleShortestPath::execute(const HashSet& startVids,
                                                   const HashSet& endVids,
                                                   DataSet* result) {

src/graph/executor/algo/SingleShortestPath.h

@@ -9,6 +9,7 @@
 
 namespace nebula {
 namespace graph {
+
 class SingleShortestPath final : public ShortestPathBase {
  public:
   using HashSet = robin_hood::unordered_flat_set<Value, std::hash<Value>>;