Merge branch 'project-pineapples' into add-gnuplot-example

[🍍 < MG] Fix mgbench access control benchmark

src/expr/interpret/frame.hpp

@@ -34,6 +34,7 @@ class Frame {
   const TypedValue &at(const Symbol &symbol) const { return elems_.at(symbol.position()); }
 
   auto &elems() { return elems_; }
+  const auto &elems() const { return elems_; }
 
   utils::MemoryResource *GetMemoryResource() const { return elems_.get_allocator().GetMemoryResource(); }
 

src/io/local_transport/local_system.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

src/io/simulator/simulator.hpp

@@ -41,7 +41,7 @@ class Simulator {
   Io<SimulatorTransport> Register(Address address) {
     std::uniform_int_distribution<uint64_t> seed_distrib;
     uint64_t seed = seed_distrib(rng_);
-    return Io{SimulatorTransport{simulator_handle_, address, seed}, address};
+    return Io{SimulatorTransport(simulator_handle_, address, seed), address};
   }
 
   void IncrementServerCountAndWaitForQuiescentState(Address address) {
@@ -50,8 +50,12 @@ class Simulator {
 
   SimulatorStats Stats() { return simulator_handle_->Stats(); }
 
+  std::shared_ptr<SimulatorHandle> GetSimulatorHandle() const { return simulator_handle_; }
+
   std::function<bool()> GetSimulatorTickClosure() {
-    std::function<bool()> tick_closure = [handle_copy = simulator_handle_] { return handle_copy->MaybeTickSimulator(); };
+    std::function<bool()> tick_closure = [handle_copy = simulator_handle_] {
+      return handle_copy->MaybeTickSimulator();
+    };
     return tick_closure;
   }
 };

src/io/simulator/simulator_transport.hpp

@@ -26,7 +26,7 @@ using memgraph::io::Time;
 
 class SimulatorTransport {
   std::shared_ptr<SimulatorHandle> simulator_handle_;
-  const Address address_;
+  Address address_;
   std::mt19937 rng_;
 
  public:
@@ -36,7 +36,9 @@ class SimulatorTransport {
   template <Message RequestT, Message ResponseT>
   ResponseFuture<ResponseT> Request(Address to_address, Address from_address, RequestT request,
                                     std::function<void()> notification, Duration timeout) {
-    std::function<bool()> tick_simulator = [handle_copy = simulator_handle_] { return handle_copy->MaybeTickSimulator(); };
+    std::function<bool()> tick_simulator = [handle_copy = simulator_handle_] {
+      return handle_copy->MaybeTickSimulator();
+    };
 
     return simulator_handle_->template SubmitRequest<RequestT, ResponseT>(
         to_address, from_address, std::move(request), timeout, std::move(tick_simulator), std::move(notification));

src/memgraph.cpp

@@ -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
@@ -640,6 +640,8 @@ int main(int argc, char **argv) {
   memgraph::machine_manager::MachineManager<memgraph::io::local_transport::LocalTransport> mm{io, config, coordinator};
   std::jthread mm_thread([&mm] { mm.Run(); });
 
+  auto rr_factory = std::make_unique<memgraph::query::v2::LocalRequestRouterFactory>(io);
+
   memgraph::query::v2::InterpreterContext interpreter_context{
       (memgraph::storage::v3::Shard *)(nullptr),
       {.query = {.allow_load_csv = FLAGS_allow_load_csv},
@@ -650,7 +652,7 @@ int main(int argc, char **argv) {
        .stream_transaction_conflict_retries = FLAGS_stream_transaction_conflict_retries,
        .stream_transaction_retry_interval = std::chrono::milliseconds(FLAGS_stream_transaction_retry_interval)},
       FLAGS_data_directory,
-      std::move(io),
+      std::move(rr_factory),
       mm.CoordinatorAddress()};
 
   SessionData session_data{&interpreter_context};

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();
@@ -704,7 +709,6 @@ PullPlan::PullPlan(const std::shared_ptr<CachedPlan> plan, const Parameters &par
   ctx_.request_router = request_router;
   ctx_.edge_ids_alloc = &interpreter_context->edge_ids_alloc;
 }
-
 std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::PullMultiple(AnyStream *stream, std::optional<int> n,
                                                                         const std::vector<Symbol> &output_symbols,
                                                                         std::map<std::string, TypedValue> *summary) {
@@ -732,10 +736,7 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::PullMultiple(AnyStrea
   }
 
   // Returns true if a result was pulled.
-  const auto pull_result = [&]() -> bool {
-    cursor_->PullMultiple(multi_frame_, ctx_);
-    return !multi_frame_.HasInvalidFrame();
-  };
+  const auto pull_result = [&]() -> bool { return cursor_->PullMultiple(multi_frame_, ctx_); };
 
   const auto stream_values = [&output_symbols, &stream](const Frame &frame) {
     // TODO: The streamed values should also probably use the above memory.
@@ -755,13 +756,14 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::PullMultiple(AnyStrea
   int i = 0;
   if (has_unsent_results_ && !output_symbols.empty()) {
     // stream unsent results from previous pull
-
-    auto iterator_for_valid_frame_only = multi_frame_.GetValidFramesReader();
-    for (const auto &frame : iterator_for_valid_frame_only) {
+    for (auto &frame : multi_frame_.GetValidFramesConsumer()) {
       stream_values(frame);
+      frame.MakeInvalid();
       ++i;
+      if (i == n) {
+        break;
+      }
     }
-    multi_frame_.MakeAllFramesInvalid();
   }
 
   for (; !n || i < n;) {
@@ -770,13 +772,17 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::PullMultiple(AnyStrea
     }
 
     if (!output_symbols.empty()) {
-      auto iterator_for_valid_frame_only = multi_frame_.GetValidFramesReader();
-      for (const auto &frame : iterator_for_valid_frame_only) {
+      for (auto &frame : multi_frame_.GetValidFramesConsumer()) {
         stream_values(frame);
+        frame.MakeInvalid();
         ++i;
+        if (i == n) {
+          break;
+        }
       }
+    } else {
+      multi_frame_.MakeAllFramesInvalid();
     }
-    multi_frame_.MakeAllFramesInvalid();
   }
 
   // If we finished because we streamed the requested n results,
@@ -811,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
@@ -906,34 +911,24 @@ using RWType = plan::ReadWriteTypeChecker::RWType;
 
 InterpreterContext::InterpreterContext(storage::v3::Shard *db, const InterpreterConfig config,
                                        const std::filesystem::path & /*data_directory*/,
-                                       io::Io<io::local_transport::LocalTransport> io,
+                                       std::unique_ptr<RequestRouterFactory> request_router_factory,
                                        coordinator::Address coordinator_addr)
-    : db(db), config(config), io{std::move(io)}, coordinator_address{coordinator_addr} {}
+    : db(db),
+      config(config),
+      coordinator_address{coordinator_addr},
+      request_router_factory_{std::move(request_router_factory)} {}
 
 Interpreter::Interpreter(InterpreterContext *interpreter_context) : interpreter_context_(interpreter_context) {
   MG_ASSERT(interpreter_context_, "Interpreter context must not be NULL");
 
-  // TODO(tyler) make this deterministic so that it can be tested.
-  auto random_uuid = boost::uuids::uuid{boost::uuids::random_generator()()};
-  auto query_io = interpreter_context_->io.ForkLocal(random_uuid);
+  request_router_ =
+      interpreter_context_->request_router_factory_->CreateRequestRouter(interpreter_context_->coordinator_address);
 
-  request_router_ = std::make_unique<RequestRouter<io::local_transport::LocalTransport>>(
-      coordinator::CoordinatorClient<io::local_transport::LocalTransport>(
-          query_io, interpreter_context_->coordinator_address, std::vector{interpreter_context_->coordinator_address}),
-      std::move(query_io));
   // Get edge ids
-  coordinator::CoordinatorWriteRequests requests{coordinator::AllocateEdgeIdBatchRequest{.batch_size = 1000000}};
-  io::rsm::WriteRequest<coordinator::CoordinatorWriteRequests> ww;
-  ww.operation = requests;
-  auto resp = interpreter_context_->io
-                  .Request<io::rsm::WriteRequest<coordinator::CoordinatorWriteRequests>,
-                           io::rsm::WriteResponse<coordinator::CoordinatorWriteResponses>>(
-                      interpreter_context_->coordinator_address, ww)
-                  .Wait();
-  if (resp.HasValue()) {
-    const auto alloc_edge_id_reps =
-        std::get<coordinator::AllocateEdgeIdBatchResponse>(resp.GetValue().message.write_return);
-    interpreter_context_->edge_ids_alloc = {alloc_edge_id_reps.low, alloc_edge_id_reps.high};
+  const auto edge_ids_alloc_min_max_pair =
+      request_router_->AllocateInitialEdgeIds(interpreter_context_->coordinator_address);
+  if (edge_ids_alloc_min_max_pair) {
+    interpreter_context_->edge_ids_alloc = {edge_ids_alloc_min_max_pair->first, edge_ids_alloc_min_max_pair->second};
   }
 }
 

src/query/v2/interpreter.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
@@ -16,7 +16,6 @@
 
 #include "coordinator/coordinator.hpp"
 #include "coordinator/coordinator_client.hpp"
-#include "io/local_transport/local_transport.hpp"
 #include "io/transport.hpp"
 #include "query/v2/auth_checker.hpp"
 #include "query/v2/bindings/cypher_main_visitor.hpp"
@@ -172,7 +171,8 @@ struct PreparedQuery {
 struct InterpreterContext {
   explicit InterpreterContext(storage::v3::Shard *db, InterpreterConfig config,
                               const std::filesystem::path &data_directory,
-                              io::Io<io::local_transport::LocalTransport> io, coordinator::Address coordinator_addr);
+                              std::unique_ptr<RequestRouterFactory> request_router_factory,
+                              coordinator::Address coordinator_addr);
 
   storage::v3::Shard *db;
 
@@ -188,26 +188,24 @@ struct InterpreterContext {
   const InterpreterConfig config;
   IdAllocator edge_ids_alloc;
 
-  // TODO (antaljanosbenjamin) Figure out an abstraction for io::Io to make it possible to construct an interpreter
-  // context with a simulator transport without templatizing it.
-  io::Io<io::local_transport::LocalTransport> io;
   coordinator::Address coordinator_address;
+  std::unique_ptr<RequestRouterFactory> request_router_factory_;
 
   storage::v3::LabelId NameToLabelId(std::string_view label_name) {
-    return storage::v3::LabelId::FromUint(query_id_mapper.NameToId(label_name));
+    return storage::v3::LabelId::FromUint(query_id_mapper_.NameToId(label_name));
   }
 
   storage::v3::PropertyId NameToPropertyId(std::string_view property_name) {
-    return storage::v3::PropertyId::FromUint(query_id_mapper.NameToId(property_name));
+    return storage::v3::PropertyId::FromUint(query_id_mapper_.NameToId(property_name));
   }
 
   storage::v3::EdgeTypeId NameToEdgeTypeId(std::string_view edge_type_name) {
-    return storage::v3::EdgeTypeId::FromUint(query_id_mapper.NameToId(edge_type_name));
+    return storage::v3::EdgeTypeId::FromUint(query_id_mapper_.NameToId(edge_type_name));
   }
 
  private:
   // TODO Replace with local map of labels, properties and edge type ids
-  storage::v3::NameIdMapper query_id_mapper;
+  storage::v3::NameIdMapper query_id_mapper_;
 };
 
 /// Function that is used to tell all active interpreters that they should stop
@@ -297,12 +295,15 @@ class Interpreter final {
   void Abort();
 
   const RequestRouterInterface *GetRequestRouter() const { return request_router_.get(); }
+  void InstallSimulatorTicker(std::function<bool()> &&tick_simulator) {
+    request_router_->InstallSimulatorTicker(tick_simulator);
+  }
 
  private:
   struct QueryExecution {
-    std::optional<PreparedQuery> prepared_query;
     utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize};
     utils::ResourceWithOutOfMemoryException execution_memory_with_exception{&execution_memory};
+    std::optional<PreparedQuery> prepared_query;
 
     std::map<std::string, TypedValue> summary;
     std::vector<Notification> notifications;

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>);
@@ -54,15 +57,17 @@ bool MultiFrame::HasInvalidFrame() const noexcept {
 
 // NOLINTNEXTLINE (bugprone-exception-escape)
 void MultiFrame::DefragmentValidFrames() noexcept {
-  /*
-  from: https://en.cppreference.com/w/cpp/algorithm/remove
-  "Removing is done by shifting (by means of copy assignment (until C++11)move assignment (since C++11)) the elements
-  in the range in such a way that the elements that are not to be removed appear in the beginning of the range.
-  Relative order of the elements that remain is preserved and the physical size of the container is unchanged."
-  */
-
-  // NOLINTNEXTLINE (bugprone-unused-return-value)
-  std::remove_if(frames_.begin(), frames_.end(), [](auto &frame) { return !frame.IsValid(); });
+  static constexpr auto kIsValid = [](const FrameWithValidity &frame) { return frame.IsValid(); };
+  static constexpr auto kIsInvalid = [](const FrameWithValidity &frame) { return !frame.IsValid(); };
+  auto first_invalid_frame = std::find_if(frames_.begin(), frames_.end(), kIsInvalid);
+  auto following_first_valid = std::find_if(first_invalid_frame, frames_.end(), kIsValid);
+  while (first_invalid_frame != frames_.end() && following_first_valid != frames_.end()) {
+    std::swap(*first_invalid_frame, *following_first_valid);
+    first_invalid_frame++;
+    first_invalid_frame = std::find_if(first_invalid_frame, frames_.end(), kIsInvalid);
+    following_first_valid++;
+    following_first_valid = std::find_if(following_first_valid, frames_.end(), kIsValid);
+  }
 }
 
 ValidFramesReader MultiFrame::GetValidFramesReader() { return ValidFramesReader{*this}; }

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;
@@ -33,6 +37,7 @@ class MultiFrame {
   MultiFrame(size_t size_of_frame, size_t number_of_frames, utils::MemoryResource *execution_memory);
   ~MultiFrame() = default;
 
+  // Assigning and moving the MultiFrame is not allowed if any accessor from the above ones are alive.
   MultiFrame(const MultiFrame &other);
   MultiFrame(MultiFrame &&other) noexcept;
   MultiFrame &operator=(const MultiFrame &other) = delete;
@@ -97,9 +102,9 @@ class ValidFramesReader {
 
   ~ValidFramesReader() = default;
   ValidFramesReader(const ValidFramesReader &other) = delete;
-  ValidFramesReader(ValidFramesReader &&other) noexcept = delete;
+  ValidFramesReader(ValidFramesReader &&other) noexcept = default;
   ValidFramesReader &operator=(const ValidFramesReader &other) = delete;
-  ValidFramesReader &operator=(ValidFramesReader &&other) noexcept = delete;
+  ValidFramesReader &operator=(ValidFramesReader &&other) noexcept = default;
 
   struct Iterator {
     using iterator_category = std::forward_iterator_tag;
@@ -147,9 +152,9 @@ class ValidFramesModifier {
 
   ~ValidFramesModifier() = default;
   ValidFramesModifier(const ValidFramesModifier &other) = delete;
-  ValidFramesModifier(ValidFramesModifier &&other) noexcept = delete;
+  ValidFramesModifier(ValidFramesModifier &&other) noexcept = default;
   ValidFramesModifier &operator=(const ValidFramesModifier &other) = delete;
-  ValidFramesModifier &operator=(ValidFramesModifier &&other) noexcept = delete;
+  ValidFramesModifier &operator=(ValidFramesModifier &&other) noexcept = default;
 
   struct Iterator {
     using iterator_category = std::forward_iterator_tag;
@@ -202,9 +207,9 @@ class ValidFramesConsumer {
 
   ~ValidFramesConsumer() noexcept;
   ValidFramesConsumer(const ValidFramesConsumer &other) = delete;
-  ValidFramesConsumer(ValidFramesConsumer &&other) noexcept = delete;
+  ValidFramesConsumer(ValidFramesConsumer &&other) noexcept = default;
   ValidFramesConsumer &operator=(const ValidFramesConsumer &other) = delete;
-  ValidFramesConsumer &operator=(ValidFramesConsumer &&other) noexcept = delete;
+  ValidFramesConsumer &operator=(ValidFramesConsumer &&other) noexcept = default;
 
   struct Iterator {
     using iterator_category = std::forward_iterator_tag;
@@ -256,9 +261,9 @@ class InvalidFramesPopulator {
   ~InvalidFramesPopulator() = default;
 
   InvalidFramesPopulator(const InvalidFramesPopulator &other) = delete;
-  InvalidFramesPopulator(InvalidFramesPopulator &&other) noexcept = delete;
+  InvalidFramesPopulator(InvalidFramesPopulator &&other) noexcept = default;
   InvalidFramesPopulator &operator=(const InvalidFramesPopulator &other) = delete;
-  InvalidFramesPopulator &operator=(InvalidFramesPopulator &&other) noexcept = delete;
+  InvalidFramesPopulator &operator=(InvalidFramesPopulator &&other) noexcept = default;
 
   struct Iterator {
     using iterator_category = std::forward_iterator_tag;

src/query/v2/plan/cost_estimator.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

src/query/v2/plan/operator.lcp

@@ -72,7 +72,21 @@ class Cursor {
   /// @throws QueryRuntimeException if something went wrong with execution
   virtual bool Pull(Frame &, ExecutionContext &) = 0;
 
-  virtual void PullMultiple(MultiFrame &, ExecutionContext &) { LOG_FATAL("PullMultipleIsNotImplemented"); }
+  /// Run an iteration of a @c LogicalOperator with MultiFrame.
+  ///
+  /// Since operators may be chained, the iteration may pull results from
+  /// multiple operators.
+  ///
+  /// @param MultiFrame May be read from or written to while performing the
+  ///     iteration.
+  /// @param ExecutionContext Used to get the position of symbols in frame and
+  ///     other information.
+  /// @return True if the operator was able to populate at least one Frame on the MultiFrame,
+  ///     thus if an operator returns true, that means there is at least one valid Frame in the
+  ///     MultiFrame.
+  ///
+  /// @throws QueryRuntimeException if something went wrong with execution
+  virtual bool PullMultiple(MultiFrame &, ExecutionContext &) {MG_ASSERT(false, "PullMultipleIsNotImplemented"); return false; }
 
   /// Resets the Cursor to its initial state.
   virtual void Reset() = 0;
@@ -335,7 +349,7 @@ and false on every following Pull.")
    class OnceCursor : public Cursor {
     public:
      OnceCursor() {}
-     void PullMultiple(MultiFrame &, ExecutionContext &) override;
+     bool PullMultiple(MultiFrame &, ExecutionContext &) override;
      bool Pull(Frame &, ExecutionContext &) override;
      void Shutdown() override;
      void Reset() override;
@@ -1162,6 +1176,7 @@ a boolean value.")
     public:
      FilterCursor(const Filter &, utils::MemoryResource *);
      bool Pull(Frame &, ExecutionContext &) override;
+     bool PullMultiple(MultiFrame &, ExecutionContext &) override;
      void Shutdown() override;
      void Reset() override;
 
@@ -1213,7 +1228,7 @@ RETURN clause) the Produce's pull succeeds exactly once.")
     public:
      ProduceCursor(const Produce &, utils::MemoryResource *);
      bool Pull(Frame &, ExecutionContext &) override;
-     void PullMultiple(MultiFrame &, ExecutionContext &) override;
+     bool PullMultiple(MultiFrame &, ExecutionContext &) override;
      void Shutdown() override;
      void Reset() override;
 
@@ -1261,6 +1276,7 @@ Has a flag for using DETACH DELETE when deleting vertices.")
     public:
      DeleteCursor(const Delete &, utils::MemoryResource *);
      bool Pull(Frame &, ExecutionContext &) override;
+     bool PullMultiple(MultiFrame &, ExecutionContext &) override;
      void Shutdown() override;
      void Reset() override;
 
@@ -1554,6 +1570,7 @@ edge lists).")
      EdgeUniquenessFilterCursor(const EdgeUniquenessFilter &,
                                 utils::MemoryResource *);
      bool Pull(Frame &, ExecutionContext &) override;
+     bool PullMultiple(MultiFrame &, ExecutionContext &) override;
      void Shutdown() override;
      void Reset() override;
 

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

@@ -597,6 +597,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/request_router.hpp

@@ -24,14 +24,18 @@
 #include <stdexcept>
 #include <thread>
 #include <unordered_map>
+#include <variant>
 #include <vector>
 
+#include <boost/uuid/uuid.hpp>
+
 #include "coordinator/coordinator.hpp"
 #include "coordinator/coordinator_client.hpp"
 #include "coordinator/coordinator_rsm.hpp"
 #include "coordinator/shard_map.hpp"
 #include "io/address.hpp"
 #include "io/errors.hpp"
+#include "io/local_transport/local_transport.hpp"
 #include "io/notifier.hpp"
 #include "io/rsm/raft.hpp"
 #include "io/rsm/rsm_client.hpp"
@@ -113,7 +117,10 @@ 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;
 };
 
@@ -139,7 +146,7 @@ class RequestRouter : public RequestRouterInterface {
 
   ~RequestRouter() override {}
 
-  void InstallSimulatorTicker(std::function<bool()> tick_simulator) {
+  void InstallSimulatorTicker(std::function<bool()> tick_simulator) override {
     notifier_.InstallSimulatorTicker(tick_simulator);
   }
 
@@ -224,7 +231,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());
 
@@ -715,6 +722,23 @@ class RequestRouter : public RequestRouterInterface {
     edge_types_.StoreMapping(std::move(id_to_name));
   }
 
+  std::optional<std::pair<uint64_t, uint64_t>> AllocateInitialEdgeIds(io::Address coordinator_address) override {
+    coordinator::CoordinatorWriteRequests requests{coordinator::AllocateEdgeIdBatchRequest{.batch_size = 1000000}};
+
+    io::rsm::WriteRequest<coordinator::CoordinatorWriteRequests> ww;
+    ww.operation = requests;
+    auto resp =
+        io_.template Request<io::rsm::WriteRequest<coordinator::CoordinatorWriteRequests>,
+                             io::rsm::WriteResponse<coordinator::CoordinatorWriteResponses>>(coordinator_address, ww)
+            .Wait();
+    if (resp.HasValue()) {
+      const auto alloc_edge_id_reps =
+          std::get<coordinator::AllocateEdgeIdBatchResponse>(resp.GetValue().message.write_return);
+      return std::make_pair(alloc_edge_id_reps.low, alloc_edge_id_reps.high);
+    }
+    return {};
+  }
+
   ShardMap shards_map_;
   storage::v3::NameIdMapper properties_;
   storage::v3::NameIdMapper edge_types_;
@@ -726,4 +750,66 @@ class RequestRouter : public RequestRouterInterface {
   io::Notifier notifier_ = {};
   // TODO(kostasrim) Add batch prefetching
 };
+
+class RequestRouterFactory {
+ public:
+  RequestRouterFactory() = default;
+  RequestRouterFactory(const RequestRouterFactory &) = delete;
+  RequestRouterFactory &operator=(const RequestRouterFactory &) = delete;
+  RequestRouterFactory(RequestRouterFactory &&) = delete;
+  RequestRouterFactory &operator=(RequestRouterFactory &&) = delete;
+
+  virtual ~RequestRouterFactory() = default;
+
+  virtual std::unique_ptr<RequestRouterInterface> CreateRequestRouter(
+      const coordinator::Address &coordinator_address) const = 0;
+};
+
+class LocalRequestRouterFactory : public RequestRouterFactory {
+  using LocalTransportIo = io::Io<io::local_transport::LocalTransport>;
+  LocalTransportIo &io_;
+
+ public:
+  explicit LocalRequestRouterFactory(LocalTransportIo &io) : io_(io) {}
+
+  std::unique_ptr<RequestRouterInterface> CreateRequestRouter(
+      const coordinator::Address &coordinator_address) const override {
+    using TransportType = io::local_transport::LocalTransport;
+
+    auto query_io = io_.ForkLocal(boost::uuids::uuid{boost::uuids::random_generator()()});
+    auto local_transport_io = io_.ForkLocal(boost::uuids::uuid{boost::uuids::random_generator()()});
+
+    return std::make_unique<RequestRouter<TransportType>>(
+        coordinator::CoordinatorClient<TransportType>(query_io, coordinator_address, {coordinator_address}),
+        std::move(local_transport_io));
+  }
+};
+
+class SimulatedRequestRouterFactory : public RequestRouterFactory {
+  io::simulator::Simulator *simulator_;
+
+ public:
+  explicit SimulatedRequestRouterFactory(io::simulator::Simulator &simulator) : simulator_(&simulator) {}
+
+  std::unique_ptr<RequestRouterInterface> CreateRequestRouter(
+      const coordinator::Address &coordinator_address) const override {
+    using TransportType = io::simulator::SimulatorTransport;
+    auto actual_transport_handle = simulator_->GetSimulatorHandle();
+
+    boost::uuids::uuid random_uuid;
+    io::Address unique_local_addr_query;
+
+    // The simulated RR should not introduce stochastic behavior.
+    random_uuid = boost::uuids::uuid{3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
+    unique_local_addr_query = {.unique_id = boost::uuids::uuid{4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}};
+
+    auto io = simulator_->Register(unique_local_addr_query);
+    auto query_io = io.ForkLocal(random_uuid);
+
+    return std::make_unique<RequestRouter<TransportType>>(
+        coordinator::CoordinatorClient<TransportType>(query_io, coordinator_address, {coordinator_address}),
+        std::move(io));
+  }
+};
+
 }  // namespace memgraph::query::v2

src/query/v2/requests.hpp

@@ -25,6 +25,7 @@
 #include "storage/v3/id_types.hpp"
 #include "storage/v3/property_value.hpp"
 #include "storage/v3/result.hpp"
+#include "utils/fnv.hpp"
 
 namespace memgraph::msgs {
 
@@ -579,3 +580,48 @@ using WriteResponses = std::variant<CreateVerticesResponse, DeleteVerticesRespon
                                     CreateExpandResponse, DeleteEdgesResponse, UpdateEdgesResponse, CommitResponse>;
 
 }  // namespace memgraph::msgs
+
+namespace std {
+
+template <>
+struct hash<memgraph::msgs::Value>;
+
+template <>
+struct hash<memgraph::msgs::VertexId> {
+  size_t operator()(const memgraph::msgs::VertexId &id) const {
+    using LabelId = memgraph::storage::v3::LabelId;
+    using Value = memgraph::msgs::Value;
+    return memgraph::utils::HashCombine<LabelId, std::vector<Value>, std::hash<LabelId>,
+                                        memgraph::utils::FnvCollection<std::vector<Value>, Value>>{}(id.first.id,
+                                                                                                     id.second);
+  }
+};
+
+template <>
+struct hash<memgraph::msgs::Value> {
+  size_t operator()(const memgraph::msgs::Value &value) const {
+    using Type = memgraph::msgs::Value::Type;
+    switch (value.type) {
+      case Type::Null:
+        return std::hash<size_t>{}(0U);
+      case Type::Bool:
+        return std::hash<bool>{}(value.bool_v);
+      case Type::Int64:
+        return std::hash<int64_t>{}(value.int_v);
+      case Type::Double:
+        return std::hash<double>{}(value.double_v);
+      case Type::String:
+        return std::hash<std::string>{}(value.string_v);
+      case Type::List:
+        LOG_FATAL("Add hash for lists");
+      case Type::Map:
+        LOG_FATAL("Add hash for maps");
+      case Type::Vertex:
+        LOG_FATAL("Add hash for vertices");
+      case Type::Edge:
+        LOG_FATAL("Add hash for edges");
+    }
+  }
+};
+
+}  // namespace std

src/storage/v3/request_helper.cpp

@@ -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
@@ -321,12 +321,15 @@ EdgeFiller InitializeEdgeFillerFunction(const msgs::ExpandOneRequest &req) {
         value_properties.insert(std::make_pair(prop_key, FromPropertyValueToValue(std::move(prop_val))));
       }
       using EdgeWithAllProperties = msgs::ExpandOneResultRow::EdgeWithAllProperties;
-      EdgeWithAllProperties edges{ToMsgsVertexId(edge.From()), msgs::EdgeType{edge.EdgeType()}, edge.Gid().AsUint(),
-                                  std::move(value_properties)};
+
       if (is_in_edge) {
-        result_row.in_edges_with_all_properties.push_back(std::move(edges));
+        result_row.in_edges_with_all_properties.push_back(
+            EdgeWithAllProperties{ToMsgsVertexId(edge.From()), msgs::EdgeType{edge.EdgeType()}, edge.Gid().AsUint(),
+                                  std::move(value_properties)});
       } else {
-        result_row.out_edges_with_all_properties.push_back(std::move(edges));
+        result_row.out_edges_with_all_properties.push_back(
+            EdgeWithAllProperties{ToMsgsVertexId(edge.To()), msgs::EdgeType{edge.EdgeType()}, edge.Gid().AsUint(),
+                                  std::move(value_properties)});
       }
       return {};
     };
@@ -346,12 +349,15 @@ EdgeFiller InitializeEdgeFillerFunction(const msgs::ExpandOneRequest &req) {
         value_properties.emplace_back(FromPropertyValueToValue(std::move(property_result.GetValue())));
       }
       using EdgeWithSpecificProperties = msgs::ExpandOneResultRow::EdgeWithSpecificProperties;
-      EdgeWithSpecificProperties edges{ToMsgsVertexId(edge.From()), msgs::EdgeType{edge.EdgeType()},
-                                       edge.Gid().AsUint(), std::move(value_properties)};
+
       if (is_in_edge) {
-        result_row.in_edges_with_specific_properties.push_back(std::move(edges));
+        result_row.in_edges_with_specific_properties.push_back(
+            EdgeWithSpecificProperties{ToMsgsVertexId(edge.From()), msgs::EdgeType{edge.EdgeType()},
+                                       edge.Gid().AsUint(), std::move(value_properties)});
       } else {
-        result_row.out_edges_with_specific_properties.push_back(std::move(edges));
+        result_row.out_edges_with_specific_properties.push_back(
+            EdgeWithSpecificProperties{ToMsgsVertexId(edge.To()), msgs::EdgeType{edge.EdgeType()}, edge.Gid().AsUint(),
+                                       std::move(value_properties)});
       }
       return {};
     };

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/simulation/CMakeLists.txt

@@ -17,7 +17,7 @@ function(add_simulation_test test_cpp)
   # requires unique logical target names
   set_target_properties(${target_name} PROPERTIES OUTPUT_NAME ${exec_name})
 
-  target_link_libraries(${target_name} mg-storage-v3 mg-communication mg-utils mg-io mg-io-simulator mg-coordinator mg-query-v2)
+  target_link_libraries(${target_name} mg-communication mg-utils mg-io mg-io-simulator mg-coordinator mg-query-v2 mg-storage-v3)
   target_link_libraries(${target_name} Boost::headers)
   target_link_libraries(${target_name} gtest gtest_main gmock rapidcheck rapidcheck_gtest)
 
@@ -32,4 +32,5 @@ add_simulation_test(trial_query_storage/query_storage_test.cpp)
 add_simulation_test(sharded_map.cpp)
 add_simulation_test(shard_rsm.cpp)
 add_simulation_test(cluster_property_test.cpp)
+add_simulation_test(cluster_property_test_cypher_queries.cpp)
 add_simulation_test(request_router.cpp)

tests/simulation/cluster_property_test_cypher_queries.cpp

@@ -0,0 +1,64 @@
+// 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
+// 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.
+
+// This test serves as an example of a property-based model test.
+// It generates a cluster configuration and a set of operations to
+// apply against both the real system and a greatly simplified model.
+
+#include <chrono>
+
+#include <gtest/gtest.h>
+#include <rapidcheck.h>
+#include <rapidcheck/gtest.h>
+#include <spdlog/cfg/env.h>
+
+#include "generated_operations.hpp"
+#include "io/simulator/simulator_config.hpp"
+#include "io/time.hpp"
+#include "storage/v3/shard_manager.hpp"
+#include "test_cluster.hpp"
+
+namespace memgraph::tests::simulation {
+
+using io::Duration;
+using io::Time;
+using io::simulator::SimulatorConfig;
+using storage::v3::kMaximumCronInterval;
+
+RC_GTEST_PROP(RandomClusterConfig, HappyPath, (ClusterConfig cluster_config, NonEmptyOpVec ops, uint64_t rng_seed)) {
+  spdlog::cfg::load_env_levels();
+
+  SimulatorConfig sim_config{
+      .drop_percent = 0,
+      .perform_timeouts = false,
+      .scramble_messages = true,
+      .rng_seed = rng_seed,
+      .start_time = Time::min(),
+      .abort_time = Time::max(),
+  };
+
+  std::vector<std::string> queries = {"CREATE (n:test_label{property_1: 0, property_2: 0});", "MATCH (n) RETURN n;"};
+
+  auto [sim_stats_1, latency_stats_1] = RunClusterSimulationWithQueries(sim_config, cluster_config, queries);
+  auto [sim_stats_2, latency_stats_2] = RunClusterSimulationWithQueries(sim_config, cluster_config, queries);
+
+  if (latency_stats_1 != latency_stats_2) {
+    spdlog::error("simulator stats diverged across runs");
+    spdlog::error("run 1 simulator stats: {}", sim_stats_1);
+    spdlog::error("run 2 simulator stats: {}", sim_stats_2);
+    spdlog::error("run 1 latency:\n{}", latency_stats_1.SummaryTable());
+    spdlog::error("run 2 latency:\n{}", latency_stats_2.SummaryTable());
+    RC_ASSERT(latency_stats_1 == latency_stats_2);
+    RC_ASSERT(sim_stats_1 == sim_stats_2);
+  }
+}
+
+}  // namespace memgraph::tests::simulation

tests/simulation/simulation_interpreter.hpp

@@ -0,0 +1,93 @@
+// 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
+// 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.
+
+#include "io/simulator/simulator_handle.hpp"
+#include "machine_manager/machine_config.hpp"
+#include "machine_manager/machine_manager.hpp"
+#include "query/v2/config.hpp"
+#include "query/v2/discard_value_stream.hpp"
+#include "query/v2/frontend/ast/ast.hpp"
+#include "query/v2/interpreter.hpp"
+#include "query/v2/request_router.hpp"
+
+#include <string>
+#include <vector>
+
+// TODO(gvolfing)
+// -How to set up the entire raft cluster with the QE. Also provide abrstraction for that.
+// -Pass an argument to the setup to determine, how many times the retry of a query should happen.
+
+namespace memgraph::io::simulator {
+
+class SimulatedInterpreter {
+  using ResultStream = query::v2::DiscardValueResultStream;
+
+ public:
+  explicit SimulatedInterpreter(std::unique_ptr<query::v2::InterpreterContext> interpreter_context)
+      : interpreter_context_(std::move(interpreter_context)) {
+    interpreter_ = std::make_unique<memgraph::query::v2::Interpreter>(interpreter_context_.get());
+  }
+
+  SimulatedInterpreter(const SimulatedInterpreter &) = delete;
+  SimulatedInterpreter &operator=(const SimulatedInterpreter &) = delete;
+  SimulatedInterpreter(SimulatedInterpreter &&) = delete;
+  SimulatedInterpreter &operator=(SimulatedInterpreter &&) = delete;
+  ~SimulatedInterpreter() = default;
+
+  void InstallSimulatorTicker(Simulator &simulator) {
+    interpreter_->InstallSimulatorTicker(simulator.GetSimulatorTickClosure());
+  }
+
+  std::vector<ResultStream> RunQueries(const std::vector<std::string> &queries) {
+    std::vector<ResultStream> results;
+    results.reserve(queries.size());
+
+    for (const auto &query : queries) {
+      results.emplace_back(RunQuery(query));
+    }
+    return results;
+  }
+
+ private:
+  ResultStream RunQuery(const std::string &query) {
+    ResultStream stream;
+
+    std::map<std::string, memgraph::storage::v3::PropertyValue> params;
+    const std::string *username = nullptr;
+
+    interpreter_->Prepare(query, params, username);
+    interpreter_->PullAll(&stream);
+
+    return stream;
+  }
+
+  std::unique_ptr<query::v2::InterpreterContext> interpreter_context_;
+  std::unique_ptr<query::v2::Interpreter> interpreter_;
+};
+
+SimulatedInterpreter SetUpInterpreter(Address coordinator_address, Simulator &simulator) {
+  auto rr_factory = std::make_unique<memgraph::query::v2::SimulatedRequestRouterFactory>(simulator);
+
+  auto interpreter_context = std::make_unique<memgraph::query::v2::InterpreterContext>(
+      nullptr,
+      memgraph::query::v2::InterpreterConfig{.query = {.allow_load_csv = true},
+                                             .execution_timeout_sec = 600,
+                                             .replication_replica_check_frequency = std::chrono::seconds(1),
+                                             .default_kafka_bootstrap_servers = "",
+                                             .default_pulsar_service_url = "",
+                                             .stream_transaction_conflict_retries = 30,
+                                             .stream_transaction_retry_interval = std::chrono::milliseconds(500)},
+      std::filesystem::path("mg_data"), std::move(rr_factory), coordinator_address);
+
+  return SimulatedInterpreter(std::move(interpreter_context));
+}
+
+}  // namespace memgraph::io::simulator

tests/simulation/test_cluster.hpp

@@ -36,6 +36,8 @@
 #include "utils/print_helpers.hpp"
 #include "utils/variant_helpers.hpp"
 
+#include "simulation_interpreter.hpp"
+
 namespace memgraph::tests::simulation {
 
 using coordinator::Coordinator;
@@ -279,4 +281,65 @@ std::pair<SimulatorStats, LatencyHistogramSummaries> RunClusterSimulation(const
   return std::make_pair(stats, histo);
 }
 
+std::pair<SimulatorStats, LatencyHistogramSummaries> RunClusterSimulationWithQueries(
+    const SimulatorConfig &sim_config, const ClusterConfig &cluster_config, const std::vector<std::string> &queries) {
+  spdlog::info("========================== NEW SIMULATION ==========================");
+
+  auto simulator = Simulator(sim_config);
+
+  auto machine_1_addr = Address::TestAddress(1);
+  auto cli_addr = Address::TestAddress(2);
+  auto cli_addr_2 = Address::TestAddress(3);
+
+  Io<SimulatorTransport> cli_io = simulator.Register(cli_addr);
+  Io<SimulatorTransport> cli_io_2 = simulator.Register(cli_addr_2);
+
+  auto coordinator_addresses = std::vector{
+      machine_1_addr,
+  };
+
+  ShardMap initialization_sm = TestShardMap(cluster_config.shards - 1, cluster_config.replication_factor);
+
+  auto mm_1 = MkMm(simulator, coordinator_addresses, machine_1_addr, initialization_sm);
+  Address coordinator_address = mm_1.CoordinatorAddress();
+
+  auto mm_thread_1 = std::jthread(RunMachine, std::move(mm_1));
+  simulator.IncrementServerCountAndWaitForQuiescentState(machine_1_addr);
+
+  auto detach_on_error = DetachIfDropped{.handle = mm_thread_1};
+
+  // TODO(tyler) clarify addresses of coordinator etc... as it's a mess
+
+  CoordinatorClient<SimulatorTransport> coordinator_client(cli_io, coordinator_address, {coordinator_address});
+  WaitForShardsToInitialize(coordinator_client);
+
+  auto simulated_interpreter = io::simulator::SetUpInterpreter(coordinator_address, simulator);
+  simulated_interpreter.InstallSimulatorTicker(simulator);
+
+  auto query_results = simulated_interpreter.RunQueries(queries);
+
+  // We have now completed our workload without failing any assertions, so we can
+  // disable detaching the worker thread, which will cause the mm_thread_1 jthread
+  // to be joined when this function returns.
+  detach_on_error.detach = false;
+
+  simulator.ShutDown();
+
+  mm_thread_1.join();
+
+  SimulatorStats stats = simulator.Stats();
+
+  spdlog::info("total messages:     {}", stats.total_messages);
+  spdlog::info("dropped messages:   {}", stats.dropped_messages);
+  spdlog::info("timed out requests: {}", stats.timed_out_requests);
+  spdlog::info("total requests:     {}", stats.total_requests);
+  spdlog::info("total responses:    {}", stats.total_responses);
+  spdlog::info("simulator ticks:    {}", stats.simulator_ticks);
+
+  auto histo = cli_io_2.ResponseLatencies();
+
+  spdlog::info("========================== SUCCESS :) ==========================");
+  return std::make_pair(stats, histo);
+}
+
 }  // namespace memgraph::tests::simulation

tests/unit/mock_helpers.hpp

@@ -41,7 +41,9 @@ 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));
 };
 
@@ -59,7 +61,7 @@ class MockedLogicalOperator : public plan::LogicalOperator {
 class MockedCursor : public plan::Cursor {
  public:
   MOCK_METHOD(bool, Pull, (Frame &, expr::ExecutionContext &));
-  MOCK_METHOD(void, PullMultiple, (MultiFrame &, expr::ExecutionContext &));
+  MOCK_METHOD(bool, PullMultiple, (MultiFrame &, expr::ExecutionContext &));
   MOCK_METHOD(void, Reset, ());
   MOCK_METHOD(void, Shutdown, ());
 };

tests/unit/query_v2_common.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

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,8 +123,13 @@ 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 {};
+  }
+
+  void InstallSimulatorTicker(std::function<bool()> tick_simulator) override {}
   const std::vector<coordinator::SchemaProperty> &GetSchemaForLabel(storage::v3::LabelId /*label*/) const override {
     static std::vector<coordinator::SchemaProperty> schema;
     return schema;

tests/unit/query_v2_plan.cpp

@@ -86,7 +86,7 @@ class TestPlanner : public ::testing::Test {};
 
 using PlannerTypes = ::testing::Types<Planner>;
 
-TYPED_TEST_CASE(TestPlanner, PlannerTypes);
+TYPED_TEST_SUITE(TestPlanner, PlannerTypes);
 
 TYPED_TEST(TestPlanner, MatchFilterPropIsNotNull) {
   const char *prim_label_name = "prim_label_one";

tools/plot/.gitignore

@@ -0,0 +1,2 @@
+*.dat
+*.out

tools/plot/pool_size_batch_size_query_latency.gnuplot

@@ -0,0 +1,7 @@
+set dgrid3d 30,30
+set hidden3d
+set label "pool size"   at 8, 11000, 0
+set label "multiframe size"  at 20,6000,0
+set label "execution time (ms)"  at 14,0,5000
+splot "frames1.dat" u 1:2:3 with lines
+pause mouse close

tools/plot/thread_shard_scanall_latency.gnuplot

@@ -0,0 +1,6 @@
+set dgrid3d 30,30
+set hidden3d
+set label "shards"   at 20,-1,10000
+set label "threads"  at -10,5,10000
+splot "data.dat" u 1:2:3 with lines
+pause mouse close