Fix Edges filter by vertex

Summary:
This diff fixes the storage part of the bug that happen for the
following queries if all nodes end up on the non-master machine.
```
CREATE (a:T{c: True})-[:X{x: 2.5}]->(:A:B)-[:Y]->()-[:Z{r: 1}]->(a)

MATCH (:T{c: True})-[a:X{x: 2.5}]->(node:A:B)-[:Y]->()-[:Z{r: 1}]->() RETURN a AS node, node AS a
```

The current state of the code doesn't work completely, it's still missing the
query execution fix.

Reviewers: teon.banek, ipaljak, vkasljevic

Reviewed By: teon.banek

Subscribers: mferencevic, pullbot

Differential Revision: https://phabricator.memgraph.io/D1629

src/query/plan/distributed_ops.cpp

@@ -968,7 +968,9 @@ class DistributedExpandCursor : public query::plan::Cursor {
         } else {
           in_edges_.emplace(vertex.in(&self_->edge_types_));
         }
-        in_edges_it_.emplace(in_edges_->begin());
+        if (in_edges_) {
+          in_edges_it_.emplace(in_edges_->begin());
+        }
       }
 
       if (direction == EdgeAtom::Direction::OUT ||
@@ -985,7 +987,9 @@ class DistributedExpandCursor : public query::plan::Cursor {
         } else {
           out_edges_.emplace(vertex.out(&self_->edge_types_));
         }
-        out_edges_it_.emplace(out_edges_->begin());
+        if (out_edges_) {
+          out_edges_it_.emplace(out_edges_->begin());
+        }
       }
 
       return true;

src/query/plan/operator.cpp

@@ -548,7 +548,9 @@ bool Expand::ExpandCursor::InitEdges(Frame &frame, Context &context) {
       } else {
         in_edges_.emplace(vertex.in(&self_.edge_types_));
       }
-      in_edges_it_.emplace(in_edges_->begin());
+      if (in_edges_) {
+        in_edges_it_.emplace(in_edges_->begin());
+      }
     }
 
     if (direction == EdgeAtom::Direction::OUT ||
@@ -565,7 +567,9 @@ bool Expand::ExpandCursor::InitEdges(Frame &frame, Context &context) {
       } else {
         out_edges_.emplace(vertex.out(&self_.edge_types_));
       }
-      out_edges_it_.emplace(out_edges_->begin());
+      if (out_edges_) {
+        out_edges_it_.emplace(out_edges_->begin());
+      }
     }
 
     return true;

src/storage/distributed/edges.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <experimental/optional>
 #include <utility>
 #include <vector>
 
@@ -49,7 +50,7 @@ class Edges {
      */
     Iterator(std::vector<Element>::const_iterator position,
              std::vector<Element>::const_iterator end,
-             storage::VertexAddress vertex,
+             std::experimental::optional<storage::VertexAddress> vertex,
              const std::vector<storage::EdgeType> *edge_types)
         : position_(position),
           end_(end),
@@ -79,17 +80,17 @@ class Edges {
     std::vector<Element>::const_iterator end_;
 
     // Optional predicates. If set they define which edges are skipped by the
-    // iterator. Only one can be not-null in the current implementation.
-    storage::VertexAddress vertex_{nullptr};
+    // iterator.
+    std::experimental::optional<storage::VertexAddress> vertex_;
     // For edge types we use a vector pointer because it's optional.
     const std::vector<storage::EdgeType> *edge_types_ = nullptr;
 
     /** Helper function that skips edges that don't satisfy the predicate
      * present in this iterator. */
     void update_position() {
-      if (vertex_.local()) {
-        position_ = std::find_if(position_,
-                                 end_, [v = this->vertex_](const Element &e) {
+      if (vertex_) {
+        position_ = std::find_if(position_, end_,
+                                 [v = this->vertex_.value()](const Element &e) {
                                    return e.vertex == v;
                                  });
       }
@@ -143,7 +144,7 @@ class Edges {
    * @param edge_types - The edge types at least one of which must be matched.
    * If nullptr edges are not filtered on type.
    */
-  auto begin(storage::VertexAddress vertex,
+  auto begin(std::experimental::optional<storage::VertexAddress> vertex,
              const std::vector<storage::EdgeType> *edge_types) const {
     if (edge_types && edge_types->empty()) edge_types = nullptr;
     return Iterator(storage_.begin(), storage_.end(), vertex, edge_types);

src/storage/distributed/vertex_accessor.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <experimental/optional>
 #include <limits>
 #include <set>
 #include <vector>
@@ -107,7 +108,7 @@ class VertexAccessor final : public RecordAccessor<Vertex> {
    */
   auto in(const std::vector<storage::EdgeType> *edge_types) const {
     return MakeAccessorIterator(
-        current().in_.begin(storage::VertexAddress(nullptr), edge_types),
+        current().in_.begin(std::experimental::nullopt, edge_types),
         current().in_.end(), false, address(), db_accessor());
   }
 
@@ -140,7 +141,7 @@ class VertexAccessor final : public RecordAccessor<Vertex> {
    */
   auto out(const std::vector<storage::EdgeType> *edge_types) const {
     return MakeAccessorIterator(
-        current().out_.begin(storage::VertexAddress(nullptr), edge_types),
+        current().out_.begin(std::experimental::nullopt, edge_types),
         current().out_.end(), true, address(), db_accessor());
   }
 

src/storage/single_node/edges.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <experimental/optional>
 #include <utility>
 #include <vector>
 
@@ -77,7 +78,7 @@ class Edges {
     std::vector<Element>::const_iterator end_;
 
     // Optional predicates. If set they define which edges are skipped by the
-    // iterator. Only one can be not-null in the current implementation.
+    // iterator.
     mvcc::VersionList<Vertex> *vertex_{nullptr};
     // For edge types we use a vector pointer because it's optional.
     const std::vector<storage::EdgeType> *edge_types_ = nullptr;