Fix filtering of edges by edge type and destination

Summary:
`Edges` class didn't properly filter edges when given both edge type
and destination arguments, which causes weird bugs in query execution (wrong
edge getting matched in `Expand` with existing node flag set). This is probably
because we didn't support using both filters at the same time, but the API and
documentation for `VertexAccessor::in` and `VertexAccessor::out` functions
didn't reflect that.

Reviewers: teon.banek, msantl

Reviewed By: msantl

Subscribers: pullbot

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

src/storage/distributed/edges.hpp

@@ -88,15 +88,18 @@ class Edges {
     /** Helper function that skips edges that don't satisfy the predicate
      * present in this iterator. */
     void update_position() {
-      if (vertex_) {
-        position_ = std::find_if(position_, end_,
-                                 [v = this->vertex_.value()](const Element &e) {
-                                   return e.vertex == v;
-                                 });
-      }
-      if (edge_types_) {
+      if (vertex_ && edge_types_) {
         position_ = std::find_if(position_, end_, [this](const Element &e) {
-          return utils::Contains(*edge_types_, e.edge_type);
+          return e.vertex == this->vertex_ &&
+                 utils::Contains(*this->edge_types_, e.edge_type);
+        });
+      } else if (vertex_) {
+        position_ = std::find_if(position_, end_, [this](const Element &e) {
+          return e.vertex == this->vertex_;
+        });
+      } else if (edge_types_) {
+        position_ = std::find_if(position_, end_, [this](const Element &e) {
+          return utils::Contains(*this->edge_types_, e.edge_type);
         });
       }
     }

src/storage/single_node/edges.hpp

@@ -86,15 +86,18 @@ class Edges {
     /** Helper function that skips edges that don't satisfy the predicate
      * present in this iterator. */
     void update_position() {
-      if (vertex_) {
-        position_ = std::find_if(position_,
-                                 end_, [v = this->vertex_](const Element &e) {
-                                   return e.vertex == v;
-                                 });
-      }
-      if (edge_types_) {
+      if (vertex_ && edge_types_) {
         position_ = std::find_if(position_, end_, [this](const Element &e) {
-          return utils::Contains(*edge_types_, e.edge_type);
+          return e.vertex == this->vertex_ &&
+                 utils::Contains(*this->edge_types_, e.edge_type);
+        });
+      } else if (vertex_) {
+        position_ = std::find_if(position_, end_, [this](const Element &e) {
+          return e.vertex == this->vertex_;
+        });
+      } else if (edge_types_) {
+        position_ = std::find_if(position_, end_, [this](const Element &e) {
+          return utils::Contains(*this->edge_types_, e.edge_type);
         });
       }
     }

src/storage/single_node_ha/edges.hpp

@@ -86,15 +86,18 @@ class Edges {
     /** Helper function that skips edges that don't satisfy the predicate
      * present in this iterator. */
     void update_position() {
-      if (vertex_) {
-        position_ = std::find_if(position_,
-                                 end_, [v = this->vertex_](const Element &e) {
-                                   return e.vertex == v;
-                                 });
-      }
-      if (edge_types_) {
+      if (vertex_ && edge_types_) {
         position_ = std::find_if(position_, end_, [this](const Element &e) {
-          return utils::Contains(*edge_types_, e.edge_type);
+          return e.vertex == this->vertex_ &&
+                 utils::Contains(*this->edge_types_, e.edge_type);
+        });
+      } else if (vertex_) {
+        position_ = std::find_if(position_, end_, [this](const Element &e) {
+          return e.vertex == this->vertex_;
+        });
+      } else if (edge_types_) {
+        position_ = std::find_if(position_, end_, [this](const Element &e) {
+          return utils::Contains(*this->edge_types_, e.edge_type);
         });
       }
     }

tests/unit/CMakeLists.txt

@@ -116,6 +116,12 @@ target_link_libraries(${test_prefix}durability mg-single-node kvstore_dummy_lib)
 add_unit_test(dynamic_bitset.cpp)
 target_link_libraries(${test_prefix}dynamic_bitset mg-single-node kvstore_dummy_lib)
 
+add_unit_test(edges_distributed.cpp)
+target_link_libraries(${test_prefix}edges_distributed mg-distributed kvstore_dummy_lib)
+
+add_unit_test(edges_single_node.cpp)
+target_link_libraries(${test_prefix}edges_single_node mg-single-node kvstore_dummy_lib)
+
 add_unit_test(gid.cpp)
 target_link_libraries(${test_prefix}gid mg-distributed kvstore_dummy_lib)
 

tests/unit/edges_distributed.cpp

@@ -0,0 +1,125 @@
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "distributed/coordination.hpp"
+#include "storage/distributed/edge.hpp"
+#include "storage/distributed/vertex.hpp"
+#include "transactions/distributed/engine_master.hpp"
+#include "utils/algorithm.hpp"
+
+#include "test_coordination.hpp"
+
+#include "storage/distributed/edges.hpp"
+
+TEST(Edges, Filtering) {
+  Edges edges;
+
+  TestMasterCoordination coordination;
+  tx::EngineMaster tx_engine(&coordination);
+  auto tx = tx_engine.Begin();
+
+  int64_t vertex_gid = 0;
+
+  mvcc::VersionList<Vertex> v0(*tx, vertex_gid, vertex_gid);
+  storage::VertexAddress va0{&v0};
+  vertex_gid++;
+
+  mvcc::VersionList<Vertex> v1(*tx, vertex_gid, vertex_gid);
+  storage::VertexAddress va1{&v1};
+  vertex_gid++;
+
+  mvcc::VersionList<Vertex> v2(*tx, vertex_gid, vertex_gid);
+  storage::VertexAddress va2{&v2};
+  vertex_gid++;
+
+  mvcc::VersionList<Vertex> v3(*tx, vertex_gid, vertex_gid);
+  storage::VertexAddress va3{&v3};
+  vertex_gid++;
+
+  storage::EdgeType t1{1};
+  storage::EdgeType t2{2};
+
+  int64_t edge_gid = 0;
+
+  mvcc::VersionList<Edge> e1(*tx, edge_gid, edge_gid, va0, va1, t1);
+  storage::EdgeAddress ea1(&e1);
+  edges.emplace(va1, ea1, t1);
+  edge_gid++;
+
+  mvcc::VersionList<Edge> e2(*tx, edge_gid, edge_gid, va0, va2, t1);
+  storage::EdgeAddress ea2(&e2);
+  edges.emplace(va2, ea2, t2);
+  edge_gid++;
+
+  mvcc::VersionList<Edge> e3(*tx, edge_gid, edge_gid, va0, va3, t1);
+  storage::EdgeAddress ea3(&e3);
+  edges.emplace(va3, ea3, t1);
+  edge_gid++;
+
+  mvcc::VersionList<Edge> e4(*tx, edge_gid, edge_gid, va0, va1, t1);
+  storage::EdgeAddress ea4(&e4);
+  edges.emplace(va1, ea4, t2);
+  edge_gid++;
+
+  mvcc::VersionList<Edge> e5(*tx, edge_gid, edge_gid, va0, va2, t1);
+  storage::EdgeAddress ea5(&e5);
+  edges.emplace(va2, ea5, t1);
+  edge_gid++;
+
+  mvcc::VersionList<Edge> e6(*tx, edge_gid, edge_gid, va0, va3, t1);
+  storage::EdgeAddress ea6(&e6);
+  edges.emplace(va3, ea6, t2);
+  edge_gid++;
+
+  auto edge_addresses =
+      [edges](std::experimental::optional<storage::VertexAddress> dest,
+              std::vector<storage::EdgeType> *edge_types) {
+        std::vector<storage::EdgeAddress> ret;
+        for (auto it = edges.begin(dest, edge_types); it != edges.end(); ++it)
+          ret.push_back(it->edge);
+        return ret;
+      };
+
+  {  // no filtering
+    EXPECT_THAT(edge_addresses(std::experimental::nullopt, nullptr),
+                ::testing::UnorderedElementsAre(ea1, ea2, ea3, ea4, ea5, ea6));
+  }
+  {
+    // filter by node
+    EXPECT_THAT(edge_addresses(va1, nullptr),
+                ::testing::UnorderedElementsAre(ea1, ea4));
+    EXPECT_THAT(edge_addresses(va2, nullptr),
+                ::testing::UnorderedElementsAre(ea2, ea5));
+    EXPECT_THAT(edge_addresses(va3, nullptr),
+                ::testing::UnorderedElementsAre(ea3, ea6));
+  }
+
+  {
+    // filter by edge type
+    std::vector<storage::EdgeType> f1{t1};
+    std::vector<storage::EdgeType> f2{t2};
+    std::vector<storage::EdgeType> f3{t1, t2};
+
+    EXPECT_THAT(edge_addresses(std::experimental::nullopt, &f1),
+                ::testing::UnorderedElementsAre(ea1, ea3, ea5));
+    EXPECT_THAT(edge_addresses(std::experimental::nullopt, &f2),
+                ::testing::UnorderedElementsAre(ea2, ea4, ea6));
+    EXPECT_THAT(edge_addresses(std::experimental::nullopt, &f3),
+                ::testing::UnorderedElementsAre(ea1, ea2, ea3, ea4, ea5, ea6));
+  }
+
+  {
+    // filter by both node and edge type
+    std::vector<storage::EdgeType> f1{t1};
+    std::vector<storage::EdgeType> f2{t2};
+
+    EXPECT_THAT(edge_addresses(va1, &f1), ::testing::UnorderedElementsAre(ea1));
+    EXPECT_THAT(edge_addresses(va1, &f2), ::testing::UnorderedElementsAre(ea4));
+    EXPECT_THAT(edge_addresses(va2, &f1), ::testing::UnorderedElementsAre(ea5));
+    EXPECT_THAT(edge_addresses(va2, &f2), ::testing::UnorderedElementsAre(ea2));
+    EXPECT_THAT(edge_addresses(va3, &f1), ::testing::UnorderedElementsAre(ea3));
+    EXPECT_THAT(edge_addresses(va3, &f2), ::testing::UnorderedElementsAre(ea6));
+  }
+
+  tx_engine.Abort(*tx);
+}

tests/unit/edges_single_node.cpp

@@ -0,0 +1,96 @@
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "storage/single_node/edge.hpp"
+#include "storage/single_node/vertex.hpp"
+#include "transactions/single_node/engine.hpp"
+#include "utils/algorithm.hpp"
+
+#include "storage/single_node/edges.hpp"
+
+TEST(Edges, Filtering) {
+  Edges edges;
+
+  tx::Engine tx_engine;
+  auto tx = tx_engine.Begin();
+
+  int64_t vertex_gid = 0;
+  mvcc::VersionList<Vertex> v0(*tx, vertex_gid++);
+  mvcc::VersionList<Vertex> v1(*tx, vertex_gid++);
+  mvcc::VersionList<Vertex> v2(*tx, vertex_gid++);
+  mvcc::VersionList<Vertex> v3(*tx, vertex_gid++);
+
+  storage::EdgeType t1{1};
+  storage::EdgeType t2{2};
+
+  int64_t edge_gid = 0;
+  mvcc::VersionList<Edge> e1(*tx, edge_gid++, &v0, &v1, t1);
+  edges.emplace(&v1, &e1, t1);
+
+  mvcc::VersionList<Edge> e2(*tx, edge_gid++, &v0, &v2, t2);
+  edges.emplace(&v2, &e2, t2);
+
+  mvcc::VersionList<Edge> e3(*tx, edge_gid++, &v0, &v3, t1);
+  edges.emplace(&v3, &e3, t1);
+
+  mvcc::VersionList<Edge> e4(*tx, edge_gid++, &v0, &v1, t2);
+  edges.emplace(&v1, &e4, t2);
+
+  mvcc::VersionList<Edge> e5(*tx, edge_gid++, &v0, &v2, t1);
+  edges.emplace(&v2, &e5, t1);
+
+  mvcc::VersionList<Edge> e6(*tx, edge_gid++, &v0, &v3, t2);
+  edges.emplace(&v3, &e6, t2);
+
+  auto edge_addresses = [edges](mvcc::VersionList<Vertex> *dest,
+                                std::vector<storage::EdgeType> *edge_types) {
+    std::vector<mvcc::VersionList<Edge> *> ret;
+    for (auto it = edges.begin(dest, edge_types); it != edges.end(); ++it)
+      ret.push_back(it->edge);
+    return ret;
+  };
+
+  {  // no filtering
+    EXPECT_THAT(edge_addresses(nullptr, nullptr),
+                ::testing::UnorderedElementsAre(&e1, &e2, &e3, &e4, &e5, &e6));
+  }
+
+  {
+    // filter by node
+    EXPECT_THAT(edge_addresses(&v1, nullptr),
+                ::testing::UnorderedElementsAre(&e1, &e4));
+    EXPECT_THAT(edge_addresses(&v2, nullptr),
+                ::testing::UnorderedElementsAre(&e2, &e5));
+    EXPECT_THAT(edge_addresses(&v3, nullptr),
+                ::testing::UnorderedElementsAre(&e3, &e6));
+  }
+
+  {
+    // filter by edge type
+    std::vector<storage::EdgeType> f1{t1};
+    std::vector<storage::EdgeType> f2{t2};
+    std::vector<storage::EdgeType> f3{t1, t2};
+
+    EXPECT_THAT(edge_addresses(nullptr, &f1),
+                ::testing::UnorderedElementsAre(&e1, &e3, &e5));
+    EXPECT_THAT(edge_addresses(nullptr, &f2),
+                ::testing::UnorderedElementsAre(&e2, &e4, &e6));
+    EXPECT_THAT(edge_addresses(nullptr, &f3),
+                ::testing::UnorderedElementsAre(&e1, &e2, &e3, &e4, &e5, &e6));
+  }
+
+  {
+    // filter by both node and edge type
+    std::vector<storage::EdgeType> f1{t1};
+    std::vector<storage::EdgeType> f2{t2};
+
+    EXPECT_THAT(edge_addresses(&v1, &f1), ::testing::UnorderedElementsAre(&e1));
+    EXPECT_THAT(edge_addresses(&v1, &f2), ::testing::UnorderedElementsAre(&e4));
+    EXPECT_THAT(edge_addresses(&v2, &f1), ::testing::UnorderedElementsAre(&e5));
+    EXPECT_THAT(edge_addresses(&v2, &f2), ::testing::UnorderedElementsAre(&e2));
+    EXPECT_THAT(edge_addresses(&v3, &f1), ::testing::UnorderedElementsAre(&e3));
+    EXPECT_THAT(edge_addresses(&v3, &f2), ::testing::UnorderedElementsAre(&e6));
+  }
+
+  tx_engine.Abort(*tx);
+}