Set start-up flag to a fix value for now

The FLAGS_query_vertex_count_to_expand_existing flag has specific
meaning for v2 related things. In v3 we king of hijacked the semantic
mening of indexing beacuse of the label-based indexing.
This caused the rule based planner to pick a not so efficient plan for
execution. with this quick fix the generated plan is correct, however
the flag is useless at this form. This will be reworked in subsequent
commits.

src/coordinator/shard_map.hpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -128,6 +128,9 @@ struct LabelSpace {
   // Maps between the smallest primary key stored in the shard and the shard
   std::map<PrimaryKey, ShardMetadata> shards;
   size_t replication_factor;
+  // TODO
+  // Stub value. Should be replaced once the shard-split logic is in place.
+  int64_t split_threshold{10000};
 
   friend std::ostream &operator<<(std::ostream &in, const LabelSpace &label_space) {
     using utils::print_helpers::operator<<;

src/query/v2/interpreter.cpp

@@ -64,7 +64,11 @@
 #include "utils/tsc.hpp"
 #include "utils/variant_helpers.hpp"
 
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_bool(use_multi_frame, false, "Whether to use MultiFrame or not");
+
 namespace EventCounter {
+
 extern Event ReadQuery;
 extern Event WriteQuery;
 extern Event ReadWriteQuery;
@@ -74,6 +78,7 @@ extern const Event LabelPropertyIndexCreated;
 
 extern const Event StreamsCreated;
 extern const Event TriggersCreated;
+
 }  // namespace EventCounter
 
 namespace memgraph::query::v2 {
@@ -688,7 +693,7 @@ PullPlan::PullPlan(const std::shared_ptr<CachedPlan> plan, const Parameters &par
     : plan_(plan),
       cursor_(plan->plan().MakeCursor(execution_memory)),
       frame_(plan->symbol_table().max_position(), execution_memory),
-      multi_frame_(plan->symbol_table().max_position(), kNumberOfFramesInMultiframe, execution_memory),
+      multi_frame_(plan->symbol_table().max_position(), FLAGS_default_multi_frame_size, execution_memory),
       memory_limit_(memory_limit) {
   ctx_.db_accessor = dba;
   ctx_.symbol_table = plan->symbol_table();
@@ -812,8 +817,7 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::PullMultiple(AnyStrea
 std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *stream, std::optional<int> n,
                                                                 const std::vector<Symbol> &output_symbols,
                                                                 std::map<std::string, TypedValue> *summary) {
-  auto should_pull_multiple = false;  // TODO on the long term, we will only use PullMultiple
-  if (should_pull_multiple) {
+  if (FLAGS_use_multi_frame) {
     return PullMultiple(stream, n, output_symbols, summary);
   }
   // Set up temporary memory for a single Pull. Initial memory comes from the

src/query/v2/multiframe.cpp

@@ -17,6 +17,9 @@
 #include "query/v2/bindings/frame.hpp"
 #include "utils/pmr/vector.hpp"
 
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_uint64(default_multi_frame_size, 100, "Default size of MultiFrame");
+
 namespace memgraph::query::v2 {
 
 static_assert(std::forward_iterator<ValidFramesReader::Iterator>);

src/query/v2/multiframe.hpp

@@ -13,10 +13,14 @@
 
 #include <iterator>
 
+#include <gflags/gflags.h>
+
 #include "query/v2/bindings/frame.hpp"
 
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DECLARE_uint64(default_multi_frame_size);
+
 namespace memgraph::query::v2 {
-constexpr uint64_t kNumberOfFramesInMultiframe = 1000;  // TODO have it configurable
 
 class ValidFramesConsumer;
 class ValidFramesModifier;

src/query/v2/plan/operator.cpp

@@ -218,6 +218,7 @@ class DistributedCreateNodeCursor : public Cursor {
   }
 
   std::vector<msgs::NewVertex> NodeCreationInfoToRequest(ExecutionContext &context, Frame &frame) {
+    primary_keys_.clear();
     std::vector<msgs::NewVertex> requests;
     msgs::PrimaryKey pk;
     msgs::NewVertex rqst;
@@ -227,22 +228,27 @@ class DistributedCreateNodeCursor : public Cursor {
     ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, nullptr,
                                   storage::v3::View::NEW);
     if (const auto *node_info_properties = std::get_if<PropertiesMapList>(&node_info_.properties)) {
-      for (const auto &[key, value_expression] : *node_info_properties) {
+      for (const auto &[property, value_expression] : *node_info_properties) {
         TypedValue val = value_expression->Accept(evaluator);
-        if (context.request_router->IsPrimaryKey(primary_label, key)) {
-          rqst.primary_key.push_back(TypedValueToValue(val));
-          pk.push_back(TypedValueToValue(val));
+        auto msgs_value = TypedValueToValue(val);
+        if (context.request_router->IsPrimaryProperty(primary_label, property)) {
+          rqst.primary_key.push_back(msgs_value);
+          pk.push_back(std::move(msgs_value));
+        } else {
+          rqst.properties.emplace_back(property, std::move(msgs_value));
         }
       }
     } else {
       auto property_map = evaluator.Visit(*std::get<ParameterLookup *>(node_info_.properties)).ValueMap();
-      for (const auto &[key, value] : property_map) {
-        auto key_str = std::string(key);
-        auto property_id = context.request_router->NameToProperty(key_str);
-        if (context.request_router->IsPrimaryKey(primary_label, property_id)) {
-          rqst.primary_key.push_back(TypedValueToValue(value));
-          pk.push_back(TypedValueToValue(value));
-        }
+      for (const auto &[property, typed_value] : property_map) {
+        auto property_str = std::string(property);
+        auto property_id = context.request_router->NameToProperty(property_str);
+        auto msgs_value = TypedValueToValue(typed_value);
+        if (context.request_router->IsPrimaryProperty(primary_label, property_id)) {
+          rqst.primary_key.push_back(msgs_value);
+          pk.push_back(std::move(msgs_value));
+        } else
+          rqst.properties.emplace_back(property_id, std::move(msgs_value));
       }
     }
 
@@ -268,6 +274,7 @@ class DistributedCreateNodeCursor : public Cursor {
   }
 
   std::vector<msgs::NewVertex> NodeCreationInfoToRequests(ExecutionContext &context, MultiFrame &multi_frame) {
+    primary_keys_.clear();
     std::vector<msgs::NewVertex> requests;
     auto multi_frame_modifier = multi_frame.GetValidFramesModifier();
     for (auto &frame : multi_frame_modifier) {
@@ -280,22 +287,27 @@ class DistributedCreateNodeCursor : public Cursor {
       ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, nullptr,
                                     storage::v3::View::NEW);
       if (const auto *node_info_properties = std::get_if<PropertiesMapList>(&node_info_.properties)) {
-        for (const auto &[key, value_expression] : *node_info_properties) {
+        for (const auto &[property, value_expression] : *node_info_properties) {
           TypedValue val = value_expression->Accept(evaluator);
-          if (context.request_router->IsPrimaryKey(primary_label, key)) {
-            rqst.primary_key.push_back(TypedValueToValue(val));
-            pk.push_back(TypedValueToValue(val));
+          auto msgs_value = TypedValueToValue(val);
+          if (context.request_router->IsPrimaryProperty(primary_label, property)) {
+            rqst.primary_key.push_back(msgs_value);
+            pk.push_back(std::move(msgs_value));
+          } else {
+            rqst.properties.emplace_back(property, std::move(msgs_value));
           }
         }
       } else {
         auto property_map = evaluator.Visit(*std::get<ParameterLookup *>(node_info_.properties)).ValueMap();
-        for (const auto &[key, value] : property_map) {
-          auto key_str = std::string(key);
-          auto property_id = context.request_router->NameToProperty(key_str);
-          if (context.request_router->IsPrimaryKey(primary_label, property_id)) {
-            rqst.primary_key.push_back(TypedValueToValue(value));
-            pk.push_back(TypedValueToValue(value));
-          }
+        for (const auto &[property, typed_value] : property_map) {
+          auto property_str = std::string(property);
+          auto property_id = context.request_router->NameToProperty(property_str);
+          auto msgs_value = TypedValueToValue(typed_value);
+          if (context.request_router->IsPrimaryProperty(primary_label, property_id)) {
+            rqst.primary_key.push_back(msgs_value);
+            pk.push_back(std::move(msgs_value));
+          } else
+            rqst.properties.emplace_back(property_id, std::move(msgs_value));
         }
       }
 
@@ -497,7 +509,7 @@ class DistributedScanAllAndFilterCursor : public Cursor {
 
     if (!own_multi_frame_.has_value()) {
       own_multi_frame_.emplace(MultiFrame(output_multi_frame.GetFirstFrame().elems().size(),
-                                          kNumberOfFramesInMultiframe, output_multi_frame.GetMemoryResource()));
+                                          FLAGS_default_multi_frame_size, output_multi_frame.GetMemoryResource()));
       own_frames_consumer_.emplace(own_multi_frame_->GetValidFramesConsumer());
       own_frames_it_ = own_frames_consumer_->begin();
     }
@@ -693,7 +705,7 @@ class DistributedScanByPrimaryKeyCursor : public Cursor {
   void EnsureOwnMultiFrameIsGood(MultiFrame &output_multi_frame) {
     if (!own_multi_frame_.has_value()) {
       own_multi_frame_.emplace(MultiFrame(output_multi_frame.GetFirstFrame().elems().size(),
-                                          kNumberOfFramesInMultiframe, output_multi_frame.GetMemoryResource()));
+                                          FLAGS_default_multi_frame_size, output_multi_frame.GetMemoryResource()));
       own_frames_consumer_.emplace(own_multi_frame_->GetValidFramesConsumer());
       own_frames_it_ = own_frames_consumer_->begin();
     }
@@ -1335,15 +1347,14 @@ bool ContainsSameEdge(const TypedValue &a, const TypedValue &b) {
 }
 
 bool IsExpansionOk(Frame &frame, const Symbol &expand_symbol, const std::vector<Symbol> &previous_symbols) {
-  const auto &expand_value = frame[expand_symbol];
-  for (const auto &previous_symbol : previous_symbols) {
-    const auto &previous_value = frame[previous_symbol];
-    // This shouldn't raise a TypedValueException, because the planner
-    // makes sure these are all of the expected type. In case they are not
-    // an error should be raised long before this code is executed.
-    if (ContainsSameEdge(previous_value, expand_value)) return false;
-  }
-  return true;
+  // This shouldn't raise a TypedValueException, because the planner
+  // makes sure these are all of the expected type. In case they are not
+  // an error should be raised long before this code is executed.
+  return std::ranges::all_of(previous_symbols,
+                             [&frame, &expand_value = frame[expand_symbol]](const auto &previous_symbol) {
+                               const auto &previous_value = frame[previous_symbol];
+                               return !ContainsSameEdge(previous_value, expand_value);
+                             });
 }
 
 }  // namespace
@@ -1595,6 +1606,7 @@ class AggregateCursor : public Cursor {
         ExpressionEvaluator evaluator(&frame, context->symbol_table, context->evaluation_context,
                                       context->request_router, storage::v3::View::NEW);
         ProcessOne(frame, &evaluator);
+        frame.MakeInvalid();
       }
     }
 
@@ -2201,7 +2213,7 @@ class UnwindCursor : public Cursor {
 
     if (!own_multi_frame_.has_value()) {
       own_multi_frame_.emplace(MultiFrame(output_multi_frame.GetFirstFrame().elems().size(),
-                                          kNumberOfFramesInMultiframe, output_multi_frame.GetMemoryResource()));
+                                          FLAGS_default_multi_frame_size, output_multi_frame.GetMemoryResource()));
       own_frames_consumer_.emplace(own_multi_frame_->GetValidFramesConsumer());
       own_frames_it_ = own_frames_consumer_->begin();
     }
@@ -2960,27 +2972,16 @@ class DistributedCreateExpandCursor : public Cursor {
       const auto &v1 = v1_value.ValueVertex();
       const auto &v2 = OtherVertex(frame);
 
-      // Set src and dest vertices
-      // TODO(jbajic) Currently we are only handling scenario where vertices
-      // are matched
-      const auto set_vertex = [&context](const auto &vertex, auto &vertex_id) {
-        vertex_id.first = vertex.PrimaryLabel();
-        for (const auto &[key, val] : vertex.Properties()) {
-          if (context.request_router->IsPrimaryKey(vertex_id.first.id, key)) {
-            vertex_id.second.push_back(val);
-          }
-        }
-      };
       std::invoke([&]() {
         switch (edge_info.direction) {
           case EdgeAtom::Direction::IN: {
-            set_vertex(v2, request.src_vertex);
-            set_vertex(v1, request.dest_vertex);
+            request.src_vertex = v2.Id();
+            request.dest_vertex = v1.Id();
             break;
           }
           case EdgeAtom::Direction::OUT: {
-            set_vertex(v1, request.src_vertex);
-            set_vertex(v2, request.dest_vertex);
+            request.src_vertex = v1.Id();
+            request.dest_vertex = v2.Id();
             break;
           }
           case EdgeAtom::Direction::BOTH:
@@ -3122,6 +3123,9 @@ class DistributedExpandCursor : public Cursor {
       auto &vertex = vertex_value.ValueVertex();
       msgs::ExpandOneRequest request;
       request.direction = DirectionToMsgsDirection(self_.common_.direction);
+      std::transform(self_.common_.edge_types.begin(), self_.common_.edge_types.end(),
+                     std::back_inserter(request.edge_types),
+                     [](const storage::v3::EdgeTypeId edge_type_id) { return msgs::EdgeType{edge_type_id}; });
       // to not fetch any properties of the edges
       request.edge_properties.emplace();
       request.src_vertices.push_back(vertex.Id());
@@ -3262,6 +3266,9 @@ class DistributedExpandCursor : public Cursor {
 
     msgs::ExpandOneRequest request;
     request.direction = DirectionToMsgsDirection(self_.common_.direction);
+    std::transform(self_.common_.edge_types.begin(), self_.common_.edge_types.end(),
+                   std::back_inserter(request.edge_types),
+                   [](const storage::v3::EdgeTypeId edge_type_id) { return msgs::EdgeType{edge_type_id}; });
     // to not fetch any properties of the edges
     request.edge_properties.emplace();
     for (const auto &frame : own_multi_frame_->GetValidFramesReader()) {
@@ -3375,7 +3382,7 @@ class DistributedExpandCursor : public Cursor {
   void EnsureOwnMultiFrameIsGood(MultiFrame &output_multi_frame) {
     if (!own_multi_frame_.has_value()) {
       own_multi_frame_.emplace(MultiFrame(output_multi_frame.GetFirstFrame().elems().size(),
-                                          kNumberOfFramesInMultiframe, output_multi_frame.GetMemoryResource()));
+                                          FLAGS_default_multi_frame_size, output_multi_frame.GetMemoryResource()));
       own_frames_consumer_.emplace(own_multi_frame_->GetValidFramesConsumer());
       own_frames_it_ = own_frames_consumer_->begin();
     }

src/query/v2/plan/rewrite/index_lookup.hpp

@@ -17,6 +17,7 @@
 #pragma once
 
 #include <algorithm>
+#include <limits>
 #include <memory>
 #include <optional>
 #include <unordered_map>
@@ -561,8 +562,12 @@ class IndexLookupRewriter final : public HierarchicalLogicalOperatorVisitor {
   // `max_vertex_count` controls, whether no operator should be created if the
   // vertex count in the best index exceeds this number. In such a case,
   // `nullptr` is returned and `input` is not chained.
-  std::unique_ptr<ScanAll> GenScanByIndex(const ScanAll &scan,
-                                          const std::optional<int64_t> &max_vertex_count = std::nullopt) {
+  // std::unique_ptr<ScanAll> GenScanByIndex(const ScanAll &scan, const std::optional<int64_t> &max_vertex_count =
+  // std::nullopt) {
+  std::unique_ptr<ScanAll> GenScanByIndex(const ScanAll &scan, std::optional<int64_t> max_vertex_count = std::nullopt) {
+    // debug (gvolfing)
+    max_vertex_count = std::numeric_limits<int64_t>::max();
+
     const auto &input = scan.input();
     const auto &node_symbol = scan.output_symbol_;
     const auto &view = scan.view_;
@@ -597,6 +602,9 @@ class IndexLookupRewriter final : public HierarchicalLogicalOperatorVisitor {
                      [](const auto &schema_elem) { return schema_elem.property_id; });
 
       for (const auto &property_filter : property_filters) {
+        if (property_filter.property_filter->type_ != PropertyFilter::Type::EQUAL) {
+          continue;
+        }
         const auto &property_id = db_->NameToProperty(property_filter.property_filter->property_.name);
         if (std::find(schema_properties.begin(), schema_properties.end(), property_id) != schema_properties.end()) {
           pk_temp.emplace_back(std::make_pair(property_filter.expression, property_filter));

src/query/v2/plan/vertex_count_cache.hpp

@@ -38,12 +38,15 @@ class VertexCountCache {
   auto NameToProperty(const std::string &name) { return request_router_->NameToProperty(name); }
   auto NameToEdgeType(const std::string &name) { return request_router_->NameToEdgeType(name); }
 
-  int64_t VerticesCount() { return 1; }
+  int64_t VerticesCount() { return request_router_->GetApproximateVertexCount(); }
 
-  int64_t VerticesCount(storage::v3::LabelId /*label*/) { return 1; }
+  int64_t VerticesCount(storage::v3::LabelId label) { return request_router_->GetApproximateVertexCount(label); }
 
-  int64_t VerticesCount(storage::v3::LabelId /*label*/, storage::v3::PropertyId /*property*/) { return 1; }
+  int64_t VerticesCount(storage::v3::LabelId label, storage::v3::PropertyId property) {
+    return request_router_->GetApproximateVertexCount(label, property);
+  }
 
+  // TODO(gvolfing) check if we actually use these overloads...
   int64_t VerticesCount(storage::v3::LabelId /*label*/, storage::v3::PropertyId /*property*/,
                         const storage::v3::PropertyValue & /*value*/) {
     return 1;

src/query/v2/request_router.hpp

@@ -117,11 +117,15 @@ class RequestRouterInterface {
   virtual std::optional<storage::v3::EdgeTypeId> MaybeNameToEdgeType(const std::string &name) const = 0;
   virtual std::optional<storage::v3::LabelId> MaybeNameToLabel(const std::string &name) const = 0;
   virtual bool IsPrimaryLabel(storage::v3::LabelId label) const = 0;
-  virtual bool IsPrimaryKey(storage::v3::LabelId primary_label, storage::v3::PropertyId property) const = 0;
+  virtual bool IsPrimaryProperty(storage::v3::LabelId primary_label, storage::v3::PropertyId property) const = 0;
 
   virtual std::optional<std::pair<uint64_t, uint64_t>> AllocateInitialEdgeIds(io::Address coordinator_address) = 0;
   virtual void InstallSimulatorTicker(std::function<bool()> tick_simulator) = 0;
   virtual const std::vector<coordinator::SchemaProperty> &GetSchemaForLabel(storage::v3::LabelId label) const = 0;
+
+  virtual int64_t GetApproximateVertexCount() const = 0;
+  virtual int64_t GetApproximateVertexCount(storage::v3::LabelId label) const = 0;
+  virtual int64_t GetApproximateVertexCount(storage::v3::LabelId label, storage::v3::PropertyId property) const = 0;
 };
 
 // TODO(kostasrim)rename this class template
@@ -231,7 +235,7 @@ class RequestRouter : public RequestRouterInterface {
     return edge_types_.IdToName(id.AsUint());
   }
 
-  bool IsPrimaryKey(storage::v3::LabelId primary_label, storage::v3::PropertyId property) const override {
+  bool IsPrimaryProperty(storage::v3::LabelId primary_label, storage::v3::PropertyId property) const override {
     const auto schema_it = shards_map_.schemas.find(primary_label);
     MG_ASSERT(schema_it != shards_map_.schemas.end(), "Invalid primary label id: {}", primary_label.AsUint());
 
@@ -415,6 +419,30 @@ class RequestRouter : public RequestRouterInterface {
     return shards_map_.GetLabelId(name);
   }
 
+  int64_t GetApproximateVertexCount() const override {
+    int64_t vertex_count = 0;
+
+    for (const auto &label_space : shards_map_.label_spaces) {
+      const auto split_threshold = label_space.second.split_threshold;
+      const auto shard_count = static_cast<int64_t>(label_space.second.shards.size());
+      vertex_count += split_threshold * shard_count;
+    }
+
+    return vertex_count;
+  }
+
+  int64_t GetApproximateVertexCount(storage::v3::LabelId label) const override {
+    const auto &label_space = shards_map_.label_spaces.at(label);
+    return label_space.split_threshold * label_space.shards.size();
+  }
+
+  int64_t GetApproximateVertexCount(storage::v3::LabelId label, storage::v3::PropertyId /*property*/) const override {
+    // TODO(gvolfing)
+    // Once we have reliable metadata to approximate the
+    // vertex count -based on properties- rework this function.
+    return GetApproximateVertexCount(label);
+  }
+
  private:
   std::vector<ShardRequestState<msgs::CreateVerticesRequest>> RequestsForCreateVertices(
       const std::vector<msgs::NewVertex> &new_vertices) {

tests/mgbench/compare_results.py

@@ -14,7 +14,6 @@
 import argparse
 import json
 
-
 FIELDS = [
     {
         "name": "throughput",
@@ -85,39 +84,32 @@ def compare_results(results_from, results_to, fields):
                 if group == "__import__":
                     continue
                 for scenario, summary_to in scenarios.items():
-                    summary_from = recursive_get(
-                        results_from, dataset, variant, group, scenario,
-                        value={})
-                    if len(summary_from) > 0 and \
-                            summary_to["count"] != summary_from["count"] or \
-                            summary_to["num_workers"] != \
-                            summary_from["num_workers"]:
+                    summary_from = recursive_get(results_from, dataset, variant, group, scenario, value={})
+                    if (
+                        len(summary_from) > 0
+                        and summary_to["count"] != summary_from["count"]
+                        or summary_to["num_workers"] != summary_from["num_workers"]
+                    ):
                         raise Exception("Incompatible results!")
-                    testcode = "/".join([dataset, variant, group, scenario,
-                                         "{:02d}".format(
-                                             summary_to["num_workers"])])
+                    testcode = "/".join([dataset, variant, group, scenario, "{:02d}".format(summary_to["num_workers"])])
                     row = {}
                     performance_changed = False
                     for field in fields:
                         key = field["name"]
                         if key in summary_to:
-                            row[key] = compute_diff(
-                                summary_from.get(key, None),
-                                summary_to[key])
+                            row[key] = compute_diff(summary_from.get(key, None), summary_to[key])
                         elif key in summary_to["database"]:
                             row[key] = compute_diff(
-                                recursive_get(summary_from, "database", key,
-                                              value=None),
-                                summary_to["database"][key])
+                                recursive_get(summary_from, "database", key, value=None), summary_to["database"][key]
+                            )
                         else:
                             row[key] = compute_diff(
-                                recursive_get(summary_from, "metadata", key,
-                                              "average", value=None),
-                                summary_to["metadata"][key]["average"])
-                        if "diff" not in row[key] or \
-                                ("diff_treshold" in field and
-                                 abs(row[key]["diff"]) >=
-                                 field["diff_treshold"]):
+                                recursive_get(summary_from, "metadata", key, "average", value=None),
+                                summary_to["metadata"][key]["average"],
+                            )
+                        if "diff" not in row[key] or (
+                            "diff_treshold" in field and abs(row[key]["diff"]) >= field["diff_treshold"]
+                        ):
                             performance_changed = True
                     if performance_changed:
                         ret[testcode] = row
@@ -130,29 +122,36 @@ def generate_remarkup(fields, data):
         ret += "<table>\n"
         ret += "  <tr>\n"
         ret += "    <th>Testcode</th>\n"
-        ret += "\n".join(map(lambda x: "    <th>{}</th>".format(
-            x["name"].replace("_", " ").capitalize()), fields)) + "\n"
+        ret += (
+            "\n".join(
+                map(
+                    lambda x: "    <th>{}</th>".format(x["name"].replace("_", " ").capitalize()),
+                    fields,
+                )
+            )
+            + "\n"
+        )
         ret += "  </tr>\n"
         for testcode in sorted(data.keys()):
             ret += "  <tr>\n"
             ret += "    <td>{}</td>\n".format(testcode)
             for field in fields:
-                result = data[testcode][field["name"]]
-                value = result["value"] * field["scaling"]
-                if "diff" in result:
-                    diff = result["diff"]
-                    arrow = "arrow-up" if diff >= 0 else "arrow-down"
-                    if not (field["positive_diff_better"] ^ (diff >= 0)):
-                        color = "green"
+                result = data[testcode].get(field["name"])
+                if result != None:
+                    value = result["value"] * field["scaling"]
+                    if "diff" in result:
+                        diff = result["diff"]
+                        arrow = "arrow-up" if diff >= 0 else "arrow-down"
+                        if not (field["positive_diff_better"] ^ (diff >= 0)):
+                            color = "green"
+                        else:
+                            color = "red"
+                        sign = "{{icon {} color={}}}".format(arrow, color)
+                        ret += '    <td bgcolor="{}">{:.3f}{} ({:+.2%})</td>\n'.format(
+                            color, value, field["unit"], diff
+                        )
                     else:
-                        color = "red"
-                    sign = "{{icon {} color={}}}".format(arrow, color)
-                    ret += "    <td>{:.3f}{} //({:+.2%})// {}</td>\n".format(
-                        value, field["unit"], diff, sign)
-                else:
-                    ret += "    <td>{:.3f}{} //(new)// " \
-                           "{{icon plus color=blue}}</td>\n".format(
-                               value, field["unit"])
+                        ret += '<td bgcolor="blue">{:.3f}{} //(new)// </td>\n'.format(value, field["unit"])
             ret += "  </tr>\n"
         ret += "</table>\n"
     else:
@@ -161,11 +160,14 @@ def generate_remarkup(fields, data):
 
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser(
-        description="Compare results of multiple benchmark runs.")
-    parser.add_argument("--compare", action="append", nargs=2,
-                        metavar=("from", "to"),
-                        help="compare results between `from` and `to` files")
+    parser = argparse.ArgumentParser(description="Compare results of multiple benchmark runs.")
+    parser.add_argument(
+        "--compare",
+        action="append",
+        nargs=2,
+        metavar=("from", "to"),
+        help="compare results between `from` and `to` files",
+    )
     parser.add_argument("--output", default="", help="output file name")
     args = parser.parse_args()
 

tests/mgbench/dataset_creator_unwind.py

@@ -0,0 +1,139 @@
+# Copyright 2022 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import argparse
+import random
+
+import helpers
+
+# Explaination of datasets:
+#   - empty_only_index: contains index; contains no data
+#   - small: contains index; contains data (small dataset)
+#
+# Datamodel is as follow:
+#
+#                               ┌──────────────┐
+#                               │ Permission   │
+#         ┌────────────────┐    │  Schema:uuid │   ┌────────────┐
+#         │:IS_FOR_IDENTITY├────┤  Index:name  ├───┤:IS_FOR_FILE│
+#         └┬───────────────┘    └──────────────┘   └────────────┤
+#          │                                                    │
+#   ┌──────▼──────────────┐                                  ┌──▼────────────────┐
+#   │  Identity           │                                  │ File              │
+#   │   Schema:uuid       │                                  │  Schema:uuid      │
+#   │   Index:email       │                                  │  Index:name       │
+#   └─────────────────────┘                                  │  Index:platformId │
+#                                                            └───────────────────┘
+#
+#   - File: attributes: ["uuid", "name", "platformId"]
+#   - Permission: attributes: ["uuid", "name"]
+#   - Identity: attributes: ["uuid", "email"]
+#
+# Indexes:
+#   - File: [File(uuid), File(platformId), File(name)]
+#   - Permission: [Permission(uuid), Permission(name)]
+#   - Identity: [Identity(uuid), Identity(email)]
+#
+# Edges:
+#   - (:Permission)-[:IS_FOR_FILE]->(:File)
+#   - (:Permission)-[:IS_FOR_IDENTITYR]->(:Identity)
+#
+# AccessControl specific: uuid is the schema
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--number_of_identities", type=int, default=10)
+    parser.add_argument("--number_of_files", type=int, default=10)
+    parser.add_argument("--percentage_of_permissions", type=float, default=1.0)
+    parser.add_argument("--filename", default="dataset.cypher")
+
+    args = parser.parse_args()
+
+    number_of_identities = args.number_of_identities
+    number_of_files = args.number_of_files
+    percentage_of_permissions = args.percentage_of_permissions
+    filename = args.filename
+
+    assert number_of_identities >= 0
+    assert number_of_files >= 0
+    assert percentage_of_permissions > 0.0 and percentage_of_permissions <= 1.0
+    assert filename != ""
+
+    with open(filename, "w") as f:
+        f.write("MATCH (n) DETACH DELETE n;\n")
+
+        # Create the indexes
+        f.write("CREATE INDEX ON :File;\n")
+        f.write("CREATE INDEX ON :Permission;\n")
+        f.write("CREATE INDEX ON :Identity;\n")
+        f.write("CREATE INDEX ON :File(platformId);\n")
+        f.write("CREATE INDEX ON :File(name);\n")
+        f.write("CREATE INDEX ON :Permission(name);\n")
+        f.write("CREATE INDEX ON :Identity(email);\n")
+
+        # Create extra index: in distributed, this will be the schema
+        f.write("CREATE INDEX ON :File(uuid);\n")
+        f.write("CREATE INDEX ON :Permission(uuid);\n")
+        f.write("CREATE INDEX ON :Identity(uuid);\n")
+
+        uuid = 1
+
+        # Create the nodes File
+        f.write("UNWIND [")
+        for index in range(0, number_of_files):
+            if index != 0:
+                f.write(",")
+            f.write(f' {{uuid: {uuid}, platformId: "platform_id", name: "name_file_{uuid}"}}')
+            uuid += 1
+        f.write("] AS props CREATE (:File {uuid: props.uuid, platformId: props.platformId, name: props.name});\n")
+
+        identities = []
+        f.write("UNWIND [")
+        # Create the nodes Identity
+        for index in range(0, number_of_identities):
+            if index != 0:
+                f.write(",")
+            f.write(f' {{uuid: {uuid}, name: "mail_{uuid}@something.com"}}')
+            uuid += 1
+        f.write("] AS props CREATE (:Identity {uuid: props.uuid, name: props.name});\n")
+
+        f.write("UNWIND [")
+        created = 0
+        for outer_index in range(0, number_of_files):
+            for inner_index in range(0, number_of_identities):
+
+                file_uuid = outer_index + 1
+                identity_uuid = number_of_files + inner_index + 1
+
+                if random.random() <= percentage_of_permissions:
+
+                    if created > 0:
+                        f.write(",")
+
+                    f.write(
+                        f' {{permUuid: {uuid}, permName: "name_permission_{uuid}", fileUuid: {file_uuid}, identityUuid: {identity_uuid}}}'
+                    )
+                    created += 1
+                    uuid += 1
+
+                    if created == 5000:
+                        f.write(
+                            "] AS props MATCH (file:File {uuid:props.fileUuid}), (identity:Identity {uuid: props.identityUuid}) CREATE (permission:Permission {uuid: props.permUuid, name: props.permName}) CREATE (permission)-[: IS_FOR_FILE]->(file) CREATE (permission)-[: IS_FOR_IDENTITY]->(identity);\nUNWIND ["
+                        )
+                        created = 0
+        f.write(
+            "] AS props MATCH (file:File {uuid:props.fileUuid}), (identity:Identity {uuid: props.identityUuid}) CREATE (permission:Permission {uuid: props.permUuid, name: props.permName}) CREATE (permission)-[: IS_FOR_FILE]->(file) CREATE (permission)-[: IS_FOR_IDENTITY]->(identity);\n"
+        )
+
+
+if __name__ == "__main__":
+    main()

tests/mgbench/datasets.py

@@ -353,7 +353,7 @@ class AccessControl(Dataset):
 
     def benchmark__create__vertex(self):
         self.next_value_idx += 1
-        query = (f"CREATE (:File {{uuid: {self.next_value_idx}}});", {})
+        query = ("CREATE (:File {uuid: $uuid})", {"uuid": self.next_value_idx})
         return query
 
     def benchmark__create__edges(self):
@@ -379,6 +379,24 @@ class AccessControl(Dataset):
         return query
 
     def benchmark__match__match_all_vertices_with_edges(self):
-        self.next_value_idx += 1
         query = ("MATCH (permission:Permission)-[e:IS_FOR_FILE]->(file:File) RETURN *", {})
         return query
+
+    def benchmark__match__match_users_with_permission_for_files(self):
+        file_uuid_1 = self._get_random_uuid("File")
+        file_uuid_2 = self._get_random_uuid("File")
+        min_file_uuid = min(file_uuid_1, file_uuid_2)
+        max_file_uuid = max(file_uuid_1, file_uuid_2)
+        query = (
+            "MATCH (f:File)<-[ff:IS_FOR_FILE]-(p:Permission)-[fi:IS_FOR_IDENTITY]->(i:Identity) WHERE f.uuid >= $min_file_uuid  AND f.uuid <= $max_file_uuid RETURN *",
+            {"min_file_uuid": min_file_uuid, "max_file_uuid": max_file_uuid},
+        )
+        return query
+
+    def benchmark__match__match_users_with_permission_for_specific_file(self):
+        file_uuid = self._get_random_uuid("File")
+        query = (
+            "MATCH (f:File {uuid: $file_uuid})<-[ff:IS_FOR_FILE]-(p:Permission)-[fi:IS_FOR_IDENTITY]->(i:Identity) RETURN *",
+            {"file_uuid": file_uuid},
+        )
+        return query

tests/mgbench/runners.py

@@ -68,6 +68,15 @@ class Memgraph:
         self._cleanup()
         atexit.unregister(self._cleanup)
 
+    # Returns None if string_value is not true or false, casing doesn't matter
+    def _get_bool_value(self, string_value):
+        lower_string_value = string_value.lower()
+        if lower_string_value == "true":
+            return True
+        if lower_string_value == "false":
+            return False
+        return None
+
     def _get_args(self, **kwargs):
         data_directory = os.path.join(self._directory.name, "memgraph")
         if self._memgraph_version >= (0, 50, 0):
@@ -83,7 +92,13 @@ class Memgraph:
             args_list = self._extra_args.split(" ")
             assert len(args_list) % 2 == 0
             for i in range(0, len(args_list), 2):
-                kwargs[args_list[i]] = args_list[i + 1]
+                key = args_list[i]
+                value = args_list[i + 1]
+                maybe_bool_value = self._get_bool_value(value)
+                if maybe_bool_value is not None:
+                    kwargs[key] = maybe_bool_value
+                else:
+                    kwargs[key] = value
 
         return _convert_args_to_flags(self._memgraph_binary, **kwargs)
 

tests/mgbench/splitfiles/accesscontrol_large.shard_configuration

@@ -1,8 +1,12 @@
-4
+8
 uuid
 email
 name
 platformId
+permUuid
+permName
+fileUuid
+identityUuid
 2
 IS_FOR_IDENTITY
 IS_FOR_FILE

tests/mgbench/splitfiles/accesscontrol_medium.shard_configuration

@@ -1,8 +1,12 @@
-4
+8
 uuid
 email
 name
 platformId
+permUuid
+permName
+fileUuid
+identityUuid
 2
 IS_FOR_IDENTITY
 IS_FOR_FILE

tests/mgbench/splitfiles/accesscontrol_small.shard_configuration

@@ -1,8 +1,12 @@
-4
+8
 uuid
 email
 name
 platformId
+permUuid
+permName
+fileUuid
+identityUuid
 2
 IS_FOR_IDENTITY
 IS_FOR_FILE

tests/unit/mock_helpers.hpp

@@ -41,10 +41,14 @@ class MockedRequestRouter : public RequestRouterInterface {
   MOCK_METHOD(std::optional<storage::v3::EdgeTypeId>, MaybeNameToEdgeType, (const std::string &), (const));
   MOCK_METHOD(std::optional<storage::v3::LabelId>, MaybeNameToLabel, (const std::string &), (const));
   MOCK_METHOD(bool, IsPrimaryLabel, (storage::v3::LabelId), (const));
-  MOCK_METHOD(bool, IsPrimaryKey, (storage::v3::LabelId, storage::v3::PropertyId), (const));
+  MOCK_METHOD(bool, IsPrimaryProperty, (storage::v3::LabelId, storage::v3::PropertyId), (const));
   MOCK_METHOD((std::optional<std::pair<uint64_t, uint64_t>>), AllocateInitialEdgeIds, (io::Address));
   MOCK_METHOD(void, InstallSimulatorTicker, (std::function<bool()>));
   MOCK_METHOD(const std::vector<coordinator::SchemaProperty> &, GetSchemaForLabel, (storage::v3::LabelId), (const));
+  MOCK_METHOD(int64_t, GetApproximateVertexCount, (), (const));
+  MOCK_METHOD(int64_t, GetApproximateVertexCount, (storage::v3::LabelId label), (const));
+  MOCK_METHOD(int64_t, GetApproximateVertexCount, (storage::v3::LabelId label, storage::v3::PropertyId property),
+              (const));
 };
 
 class MockedLogicalOperator : public plan::LogicalOperator {

tests/unit/query_v2_create_node_multiframe.cpp

@@ -58,7 +58,7 @@ TEST(CreateNodeTest, CreateNodeCursor) {
   MockedRequestRouter router;
   EXPECT_CALL(router, CreateVertices(_)).Times(1).WillOnce(Return(std::vector<msgs::CreateVerticesResponse>{}));
   EXPECT_CALL(router, IsPrimaryLabel(_)).WillRepeatedly(Return(true));
-  EXPECT_CALL(router, IsPrimaryKey(_, _)).WillRepeatedly(Return(true));
+  EXPECT_CALL(router, IsPrimaryProperty(_, _)).WillRepeatedly(Return(true));
   auto context = MakeContext(ast, symbol_table, &router, &id_alloc);
   auto multi_frame = CreateMultiFrame(context.symbol_table.max_position());
   cursor->PullMultiple(multi_frame, context);

tests/unit/query_v2_expression_evaluator.cpp

@@ -123,7 +123,7 @@ class MockedRequestRouter : public RequestRouterInterface {
 
   bool IsPrimaryLabel(LabelId label) const override { return true; }
 
-  bool IsPrimaryKey(LabelId primary_label, PropertyId property) const override { return true; }
+  bool IsPrimaryProperty(LabelId primary_label, PropertyId property) const override { return true; }
 
   std::optional<std::pair<uint64_t, uint64_t>> AllocateInitialEdgeIds(io::Address coordinator_address) override {
     return {};
@@ -135,6 +135,13 @@ class MockedRequestRouter : public RequestRouterInterface {
     return schema;
   };
 
+  // TODO(gvolfing) once the real implementation is done make sure these are solved as well.
+  int64_t GetApproximateVertexCount() const override { return 1; }
+  int64_t GetApproximateVertexCount(storage::v3::LabelId label) const override { return 1; }
+  int64_t GetApproximateVertexCount(storage::v3::LabelId label, storage::v3::PropertyId property) const override {
+    return 1;
+  }
+
  private:
   void SetUpNameIdMappers() {
     std::unordered_map<uint64_t, std::string> id_to_name;