Share memory for vertex and edge value (#1565)

src/common/datatypes/Edge.h

@@ -20,6 +20,7 @@ struct Edge {
   std::string name;
   EdgeRanking ranking;
   std::unordered_map<std::string, Value> props;
+  std::atomic<size_t> refcnt{1};
 
   Edge() {}
   Edge(Edge&& v) noexcept
@@ -44,6 +45,13 @@ struct Edge {
         ranking(std::move(r)),
         props(std::move(p)) {}
 
+  size_t ref() {
+    return ++refcnt;
+  }
+  size_t unref() {
+    return --refcnt;
+  }
+
   void clear();
 
   void __clear() {
@@ -57,6 +65,10 @@ struct Edge {
 
   bool operator==(const Edge& rhs) const;
 
+  bool operator!=(const Edge& rhs) const {
+    return !(*this == rhs);
+  }
+
   void format() {
     if (type < 0) {
       reverse();

src/common/datatypes/Path.h

@@ -91,7 +91,7 @@ struct Step {
     if (dst != rhs.dst) {
       return dst < rhs.dst;
     }
-    if (type != rhs.dst) {
+    if (type != rhs.type) {
       return type < rhs.type;
     }
     if (ranking != rhs.ranking) {

src/common/datatypes/Value.cpp

@@ -463,9 +463,9 @@ void Value::setVertex(Vertex&& v) {
   setV(std::move(v));
 }
 
-void Value::setVertex(std::unique_ptr<Vertex>&& v) {
+void Value::setVertex(Vertex* v) {
   clear();
-  setV(std::move(v));
+  setV(v);
 }
 
 void Value::setEdge(const Edge& v) {
@@ -478,9 +478,9 @@ void Value::setEdge(Edge&& v) {
   setE(std::move(v));
 }
 
-void Value::setEdge(std::unique_ptr<Edge>&& v) {
+void Value::setEdge(Edge* v) {
   clear();
-  setE(std::move(v));
+  setE(v);
 }
 
 void Value::setPath(const Path& v) {
@@ -620,7 +620,7 @@ const Vertex& Value::getVertex() const {
 
 const Vertex* Value::getVertexPtr() const {
   CHECK_EQ(type_, Type::VERTEX);
-  return value_.vVal.get();
+  return value_.vVal;
 }
 
 const Edge& Value::getEdge() const {
@@ -630,7 +630,7 @@ const Edge& Value::getEdge() const {
 
 const Edge* Value::getEdgePtr() const {
   CHECK_EQ(type_, Type::EDGE);
-  return value_.eVal.get();
+  return value_.eVal;
 }
 
 const Path& Value::getPath() const {
@@ -945,11 +945,23 @@ void Value::clearSlow() {
       break;
     }
     case Type::VERTEX: {
-      destruct(value_.vVal);
+      if (value_.vVal) {
+        if (value_.vVal->unref() == 0) {
+          delete value_.vVal;
+        }
+        value_.vVal = nullptr;
+        type_ = Type::__EMPTY__;
+      }
       break;
     }
     case Type::EDGE: {
-      destruct(value_.eVal);
+      if (value_.eVal) {
+        if (value_.eVal->unref() == 0) {
+          delete value_.eVal;
+        }
+        value_.eVal = nullptr;
+        type_ = Type::__EMPTY__;
+      }
       break;
     }
     case Type::PATH: {
@@ -1026,12 +1038,18 @@ Value& Value::operator=(Value&& rhs) noexcept {
       break;
     }
     case Type::VERTEX: {
-      setV(std::move(rhs.value_.vVal));
-      break;
+      value_.vVal = rhs.value_.vVal;
+      type_ = Type::VERTEX;
+      rhs.value_.vVal = nullptr;
+      rhs.type_ = Type::__EMPTY__;
+      return *this;
     }
     case Type::EDGE: {
-      setE(std::move(rhs.value_.eVal));
-      break;
+      value_.eVal = rhs.value_.eVal;
+      type_ = Type::EDGE;
+      rhs.value_.eVal = nullptr;
+      rhs.type_ = Type::__EMPTY__;
+      return *this;
     }
     case Type::PATH: {
       setP(std::move(rhs.value_.pVal));
@@ -1246,44 +1264,36 @@ void Value::setDT(DateTime&& v) {
   new (std::addressof(value_.dtVal)) DateTime(std::move(v));
 }
 
-void Value::setV(const std::unique_ptr<Vertex>& v) {
-  type_ = Type::VERTEX;
-  new (std::addressof(value_.vVal)) std::unique_ptr<Vertex>(new Vertex(*v));
-}
-
-void Value::setV(std::unique_ptr<Vertex>&& v) {
+void Value::setV(Vertex* v) {
   type_ = Type::VERTEX;
-  new (std::addressof(value_.vVal)) std::unique_ptr<Vertex>(std::move(v));
+  value_.vVal = v;
+  value_.vVal->ref();
 }
 
 void Value::setV(const Vertex& v) {
   type_ = Type::VERTEX;
-  new (std::addressof(value_.vVal)) std::unique_ptr<Vertex>(new Vertex(v));
+  new (std::addressof(value_.vVal)) Vertex*(new Vertex(v));
 }
 
 void Value::setV(Vertex&& v) {
   type_ = Type::VERTEX;
-  new (std::addressof(value_.vVal)) std::unique_ptr<Vertex>(new Vertex(std::move(v)));
+  new (std::addressof(value_.vVal)) Vertex*(new Vertex(std::move(v)));
 }
 
-void Value::setE(const std::unique_ptr<Edge>& v) {
+void Value::setE(Edge* v) {
   type_ = Type::EDGE;
-  new (std::addressof(value_.eVal)) std::unique_ptr<Edge>(new Edge(*v));
-}
-
-void Value::setE(std::unique_ptr<Edge>&& v) {
-  type_ = Type::EDGE;
-  new (std::addressof(value_.eVal)) std::unique_ptr<Edge>(std::move(v));
+  value_.eVal = v;
+  value_.eVal->ref();
 }
 
 void Value::setE(const Edge& v) {
   type_ = Type::EDGE;
-  new (std::addressof(value_.eVal)) std::unique_ptr<Edge>(new Edge(v));
+  new (std::addressof(value_.eVal)) Edge*(new Edge(v));
 }
 
 void Value::setE(Edge&& v) {
   type_ = Type::EDGE;
-  new (std::addressof(value_.eVal)) std::unique_ptr<Edge>(new Edge(std::move(v)));
+  new (std::addressof(value_.eVal)) Edge*(new Edge(std::move(v)));
 }
 
 void Value::setP(const std::unique_ptr<Path>& v) {
@@ -1902,9 +1912,15 @@ Value Value::equal(const Value& v) const {
       return getDateTime() == v.getDateTime();
     }
     case Value::Type::VERTEX: {
+      if (value_.vVal == v.value_.vVal) {
+        return true;
+      }
       return getVertex() == v.getVertex();
     }
     case Value::Type::EDGE: {
+      if (value_.eVal == v.value_.eVal) {
+        return true;
+      }
       return getEdge() == v.getEdge();
     }
     case Value::Type::PATH: {
@@ -2737,9 +2753,15 @@ bool Value::equals(const Value& rhs) const {
       return getDateTime() == rhs.getDateTime();
     }
     case Value::Type::VERTEX: {
+      if (value_.vVal == rhs.value_.vVal) {
+        return true;
+      }
       return getVertex() == rhs.getVertex();
     }
     case Value::Type::EDGE: {
+      if (value_.eVal == rhs.value_.eVal) {
+        return true;
+      }
       return getEdge() == rhs.getEdge();
     }
     case Value::Type::PATH: {

src/common/datatypes/Value.h

@@ -256,10 +256,10 @@ struct Value {
   void setDateTime(DateTime&& v);
   void setVertex(const Vertex& v);
   void setVertex(Vertex&& v);
-  void setVertex(std::unique_ptr<Vertex>&& v);
+  void setVertex(Vertex* v);
   void setEdge(const Edge& v);
   void setEdge(Edge&& v);
-  void setEdge(std::unique_ptr<Edge>&& v);
+  void setEdge(Edge* v);
   void setPath(const Path& v);
   void setPath(Path&& v);
   void setPath(std::unique_ptr<Path>&& v);
@@ -391,8 +391,8 @@ struct Value {
     Date dVal;
     Time tVal;
     DateTime dtVal;
-    std::unique_ptr<Vertex> vVal;
-    std::unique_ptr<Edge> eVal;
+    Vertex* vVal;
+    Edge* eVal;
     std::unique_ptr<Path> pVal;
     std::unique_ptr<List> lVal;
     std::unique_ptr<Map> mVal;
@@ -437,13 +437,11 @@ struct Value {
   void setDT(const DateTime& v);
   void setDT(DateTime&& v);
   // Vertex value
-  void setV(const std::unique_ptr<Vertex>& v);
-  void setV(std::unique_ptr<Vertex>&& v);
+  void setV(Vertex* v);
   void setV(const Vertex& v);
   void setV(Vertex&& v);
   // Edge value
-  void setE(const std::unique_ptr<Edge>& v);
-  void setE(std::unique_ptr<Edge>&& v);
+  void setE(Edge* v);
   void setE(const Edge& v);
   void setE(Edge&& v);
   // Path value

src/common/datatypes/ValueOps-inl.h

@@ -368,21 +368,21 @@ void Cpp2Ops<nebula::Value>::read(Protocol* proto, nebula::Value* obj) {
       }
       case 9: {
         if (readState.fieldType == apache::thrift::protocol::T_STRUCT) {
-          obj->setVertex(nebula::Vertex());
-          auto ptr = std::make_unique<nebula::Vertex>();
-          Cpp2Ops<nebula::Vertex>::read(proto, ptr.get());
-          obj->setVertex(std::move(ptr));
+          auto* ptr = new nebula::Vertex();
+          Cpp2Ops<nebula::Vertex>::read(proto, ptr);
+          obj->setVertex(ptr);
+          ptr->unref();
         } else {
           proto->skip(readState.fieldType);
         }
         break;
       }
       case 10: {
         if (readState.fieldType == apache::thrift::protocol::T_STRUCT) {
-          obj->setEdge(nebula::Edge());
-          auto ptr = std::make_unique<nebula::Edge>();
-          Cpp2Ops<nebula::Edge>::read(proto, ptr.get());
-          obj->setEdge(std::move(ptr));
+          auto* ptr = new nebula::Edge();
+          Cpp2Ops<nebula::Edge>::read(proto, ptr);
+          obj->setEdge(ptr);
+          ptr->unref();
         } else {
           proto->skip(readState.fieldType);
         }

src/common/datatypes/Vertex.h

@@ -61,12 +61,20 @@ struct Tag {
 struct Vertex {
   Value vid;
   std::vector<Tag> tags;
+  std::atomic<size_t> refcnt{1};
 
   Vertex() = default;
   Vertex(const Vertex& v) : vid(v.vid), tags(v.tags) {}
   Vertex(Vertex&& v) noexcept : vid(std::move(v.vid)), tags(std::move(v.tags)) {}
   Vertex(Value id, std::vector<Tag> t) : vid(std::move(id)), tags(std::move(t)) {}
 
+  size_t ref() {
+    return ++refcnt;
+  }
+  size_t unref() {
+    return --refcnt;
+  }
+
   void clear() {
     vid.clear();
     tags.clear();
@@ -89,6 +97,10 @@ struct Vertex {
     return vid == rhs.vid && tags == rhs.tags;
   }
 
+  bool operator!=(const Vertex& rhs) const {
+    return vid != rhs.vid || tags != rhs.tags;
+  }
+
   bool operator<(const Vertex& rhs) const;
 
   bool contains(const Value& key) const;

src/graph/context/Iterator.cpp

@@ -439,12 +439,15 @@ const Value& GetNeighborsIter::getEdgeProp(const std::string& edge, const std::s
   return currentEdge_->values[propIndex->second];
 }
 
-Value GetNeighborsIter::getVertex(const std::string& name) const {
+Value GetNeighborsIter::getVertex(const std::string& name) {
   UNUSED(name);
   if (!valid()) {
     return Value::kNullValue;
   }
   auto vidVal = getColumn(0);
+  if (!prevVertex_.empty() && prevVertex_.getVertex().vid == vidVal) {
+    return prevVertex_;
+  }
 
   Vertex vertex;
   vertex.vid = vidVal;
@@ -470,7 +473,8 @@ Value GetNeighborsIter::getVertex(const std::string& name) const {
     }
     vertex.tags.emplace_back(std::move(tag));
   }
-  return Value(std::move(vertex));
+  prevVertex_ = Value(std::move(vertex));
+  return prevVertex_;
 }
 
 std::vector<Value> GetNeighborsIter::vids() {
@@ -735,7 +739,7 @@ StatusOr<std::size_t> SequentialIter::getColumnIndex(const std::string& col) con
   return index->second;
 }
 
-Value SequentialIter::getVertex(const std::string& name) const {
+Value SequentialIter::getVertex(const std::string& name) {
   return getColumn(name);
 }
 
@@ -852,7 +856,7 @@ const Value& PropIter::getProp(const std::string& name, const std::string& prop)
   }
 }
 
-Value PropIter::getVertex(const std::string& name) const {
+Value PropIter::getVertex(const std::string& name) {
   UNUSED(name);
   if (!valid()) {
     return Value::kNullValue;