Make expand one work with MultiFrame

src/coordinator/shard_map.cpp

@@ -228,7 +228,7 @@ Hlc ShardMap::IncrementShardMapVersion() noexcept {
   return shard_map_version;
 }
 
-// TODO(antaljanosbenjamin) use a single map for all name id 
+// TODO(antaljanosbenjamin) use a single map for all name id
 // mapping and a single counter to maintain the next id
 std::unordered_map<uint64_t, std::string> ShardMap::IdToNames() {
   std::unordered_map<uint64_t, std::string> id_to_names;
@@ -455,7 +455,7 @@ Shards ShardMap::GetShardsForRange(const LabelName &label_name, const PrimaryKey
   return shards;
 }
 
-Shard ShardMap::GetShardForKey(const LabelName &label_name, const PrimaryKey &key) const {
+const Shard &ShardMap::GetShardForKey(const LabelName &label_name, const PrimaryKey &key) const {
   MG_ASSERT(labels.contains(label_name));
 
   LabelId label_id = labels.at(label_name);
@@ -468,7 +468,7 @@ Shard ShardMap::GetShardForKey(const LabelName &label_name, const PrimaryKey &ke
   return std::prev(label_space.shards.upper_bound(key))->second;
 }
 
-Shard ShardMap::GetShardForKey(const LabelId &label_id, const PrimaryKey &key) const {
+const Shard &ShardMap::GetShardForKey(const LabelId &label_id, const PrimaryKey &key) const {
   MG_ASSERT(label_spaces.contains(label_id));
 
   const auto &label_space = label_spaces.at(label_id);

src/coordinator/shard_map.hpp

@@ -160,9 +160,9 @@ struct ShardMap {
 
   Shards GetShardsForRange(const LabelName &label_name, const PrimaryKey &start_key, const PrimaryKey &end_key) const;
 
-  Shard GetShardForKey(const LabelName &label_name, const PrimaryKey &key) const;
+  const Shard &GetShardForKey(const LabelName &label_name, const PrimaryKey &key) const;
 
-  Shard GetShardForKey(const LabelId &label_id, const PrimaryKey &key) const;
+  const Shard &GetShardForKey(const LabelId &label_id, const PrimaryKey &key) const;
 
   PropertyMap AllocatePropertyIds(const std::vector<PropertyName> &new_properties);
 

src/expr/interpret/frame.hpp

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

src/query/v2/bindings/frame.hpp

@@ -13,9 +13,15 @@
 
 #include "query/v2/bindings/bindings.hpp"
 
-#include "query/v2/bindings/typed_value.hpp"
 #include "expr/interpret/frame.hpp"
+#include "query/v2/bindings/typed_value.hpp"
+#include "utils/pmr/vector.hpp"
 
 namespace memgraph::query::v2 {
 using Frame = memgraph::expr::Frame<TypedValue>;
-} // namespace memgraph::query::v2
+
+struct MultiFrame {
+  utils::pmr::vector<Frame> frames = utils::pmr::vector<Frame>(0, Frame{1}, utils::NewDeleteResource());
+  size_t valid_frames{0};
+};
+}  // namespace memgraph::query::v2

src/query/v2/interpreter.cpp

@@ -148,7 +148,7 @@ Callback HandleAuthQuery(AuthQuery *auth_query, AuthQueryHandler *auth, const Pa
   // Empty frame for evaluation of password expression. This is OK since
   // password should be either null or string literal and it's evaluation
   // should not depend on frame.
-  expr::Frame<TypedValue> frame(0);
+  Frame frame(0);
   SymbolTable symbol_table;
   EvaluationContext evaluation_context;
   // TODO: MemoryResource for EvaluationContext, it should probably be passed as
@@ -315,7 +315,7 @@ Callback HandleAuthQuery(AuthQuery *auth_query, AuthQueryHandler *auth, const Pa
 Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &parameters,
                                 InterpreterContext *interpreter_context, msgs::ShardRequestManagerInterface *manager,
                                 std::vector<Notification> *notifications) {
-  expr::Frame<TypedValue> frame(0);
+  Frame frame(0);
   SymbolTable symbol_table;
   EvaluationContext evaluation_context;
   // TODO: MemoryResource for EvaluationContext, it should probably be passed as
@@ -450,7 +450,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
 
 Callback HandleSettingQuery(SettingQuery *setting_query, const Parameters &parameters,
                             msgs::ShardRequestManagerInterface *manager) {
-  expr::Frame<TypedValue> frame(0);
+  Frame frame(0);
   SymbolTable symbol_table;
   EvaluationContext evaluation_context;
   // TODO: MemoryResource for EvaluationContext, it should probably be passed as
@@ -656,11 +656,15 @@ struct PullPlan {
   std::optional<plan::ProfilingStatsWithTotalTime> Pull(AnyStream *stream, std::optional<int> n,
                                                         const std::vector<Symbol> &output_symbols,
                                                         std::map<std::string, TypedValue> *summary);
+  std::optional<plan::ProfilingStatsWithTotalTime> PullMultiple(AnyStream *stream, std::optional<int> n,
+                                                                const std::vector<Symbol> &output_symbols,
+                                                                std::map<std::string, TypedValue> *summary);
 
  private:
   std::shared_ptr<CachedPlan> plan_ = nullptr;
   plan::UniqueCursorPtr cursor_ = nullptr;
-  expr::Frame<TypedValue> frame_;
+  Frame frame_;
+  MultiFrame multi_frame_;
   ExecutionContext ctx_;
   std::optional<size_t> memory_limit_;
 
@@ -684,6 +688,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_({.frames = utils::pmr::vector<Frame>(1000, frame_, execution_memory), .valid_frames = 0}),
       memory_limit_(memory_limit) {
   ctx_.db_accessor = dba;
   ctx_.symbol_table = plan->symbol_table();
@@ -703,6 +708,9 @@ PullPlan::PullPlan(const std::shared_ptr<CachedPlan> plan, const Parameters &par
 std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *stream, std::optional<int> n,
                                                                 const std::vector<Symbol> &output_symbols,
                                                                 std::map<std::string, TypedValue> *summary) {
+  if (!output_symbols.empty() && output_symbols[0].name() == "mmm") {
+    return PullMultiple(stream, n, output_symbols, summary);
+  }
   // Set up temporary memory for a single Pull. Initial memory comes from the
   // stack. 256 KiB should fit on the stack and should be more than enough for a
   // single `Pull`.
@@ -789,6 +797,106 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *strea
   return GetStatsWithTotalTime(ctx_);
 }
 
+std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::PullMultiple(AnyStream *stream, std::optional<int> n,
+                                                                        const std::vector<Symbol> &output_symbols,
+                                                                        std::map<std::string, TypedValue> *summary) {
+  // Set up temporary memory for a single Pull. Initial memory comes from the
+  // stack. 256 KiB should fit on the stack and should be more than enough for a
+  // single `Pull`.
+  MG_ASSERT(!n.has_value(), "should pull all!");
+  static constexpr size_t stack_size = 256UL * 1024UL;
+  char stack_data[stack_size];
+  utils::ResourceWithOutOfMemoryException resource_with_exception;
+  utils::MonotonicBufferResource monotonic_memory(&stack_data[0], stack_size, &resource_with_exception);
+  // We can throw on every query because a simple queries for deleting will use only
+  // the stack allocated buffer.
+  // Also, we want to throw only when the query engine requests more memory and not the storage
+  // so we add the exception to the allocator.
+  // TODO (mferencevic): Tune the parameters accordingly.
+  utils::PoolResource pool_memory(128, 1024, &monotonic_memory);
+  std::optional<utils::LimitedMemoryResource> maybe_limited_resource;
+
+  if (memory_limit_) {
+    maybe_limited_resource.emplace(&pool_memory, *memory_limit_);
+    ctx_.evaluation_context.memory = &*maybe_limited_resource;
+  } else {
+    ctx_.evaluation_context.memory = &pool_memory;
+  }
+
+  // Returns true if a result was pulled.
+  const auto pull_result = [&]() -> bool {
+    cursor_->PullMultiple(multi_frame_, ctx_);
+    return multi_frame_.valid_frames > 0;
+  };
+
+  const auto stream_values = [&output_symbols, &stream](Frame &frame) {
+    // TODO: The streamed values should also probably use the above memory.
+    std::vector<TypedValue> values;
+    values.reserve(output_symbols.size());
+
+    for (const auto &symbol : output_symbols) {
+      values.emplace_back(frame[symbol]);
+    }
+
+    stream->Result(values);
+  };
+
+  // Get the execution time of all possible result pulls and streams.
+  utils::Timer timer;
+
+  int i = 0;
+  if (has_unsent_results_ && !output_symbols.empty()) {
+    // stream unsent results from previous pull
+    for (auto frame_index = 0U; frame_index < multi_frame_.valid_frames; ++frame_index) {
+      stream_values(multi_frame_.frames[frame_index]);
+      ++i;
+    }
+    multi_frame_.valid_frames = 0;
+  }
+
+  for (; !n || i < n;) {
+    if (!pull_result()) {
+      break;
+    }
+
+    if (!output_symbols.empty()) {
+      for (auto frame_index = 0U; frame_index < multi_frame_.valid_frames; ++frame_index) {
+        stream_values(multi_frame_.frames[frame_index]);
+        ++i;
+      }
+    }
+    multi_frame_.valid_frames = 0;
+  }
+
+  // If we finished because we streamed the requested n results,
+  // we try to pull the next result to see if there is more.
+  // If there is additional result, we leave the pulled result in the frame
+  // and set the flag to true.
+  has_unsent_results_ = i == n && pull_result();
+
+  execution_time_ += timer.Elapsed();
+
+  if (has_unsent_results_) {
+    return std::nullopt;
+  }
+  summary->insert_or_assign("plan_execution_time", execution_time_.count());
+  // We are finished with pulling all the data, therefore we can send any
+  // metadata about the results i.e. notifications and statistics
+  const bool is_any_counter_set =
+      std::any_of(ctx_.execution_stats.counters.begin(), ctx_.execution_stats.counters.end(),
+                  [](const auto &counter) { return counter > 0; });
+  if (is_any_counter_set) {
+    std::map<std::string, TypedValue> stats;
+    for (size_t i = 0; i < ctx_.execution_stats.counters.size(); ++i) {
+      stats.emplace(ExecutionStatsKeyToString(ExecutionStats::Key(i)), ctx_.execution_stats.counters[i]);
+    }
+    summary->insert_or_assign("stats", std::move(stats));
+  }
+  cursor_->Shutdown();
+  ctx_.profile_execution_time = execution_time_;
+  return GetStatsWithTotalTime(ctx_);
+}
+
 using RWType = plan::ReadWriteTypeChecker::RWType;
 }  // namespace
 
@@ -882,7 +990,7 @@ PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string,
   //                                 TriggerContextCollector *trigger_context_collector = nullptr) {
   auto *cypher_query = utils::Downcast<CypherQuery>(parsed_query.query);
 
-  expr::Frame<TypedValue> frame(0);
+  Frame frame(0);
   SymbolTable symbol_table;
   EvaluationContext evaluation_context;
   evaluation_context.timestamp = QueryTimestamp();
@@ -1026,7 +1134,7 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
 
   auto *cypher_query = utils::Downcast<CypherQuery>(parsed_inner_query.query);
   MG_ASSERT(cypher_query, "Cypher grammar should not allow other queries in PROFILE");
-  expr::Frame<TypedValue> frame(0);
+  Frame frame(0);
   SymbolTable symbol_table;
   EvaluationContext evaluation_context;
   evaluation_context.timestamp = QueryTimestamp();
@@ -1056,7 +1164,7 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
                          if (!stats_and_total_time) {
                            stats_and_total_time = PullPlan(plan, parameters, true, dba, interpreter_context,
                                                            execution_memory, shard_request_manager, memory_limit)
-                                                      .Pull(stream, {}, {}, summary);
+                                                      .PullMultiple(stream, {}, {}, summary);
                            pull_plan = std::make_shared<PullPlanVector>(ProfilingStatsToTable(*stats_and_total_time));
                          }
 

src/query/v2/plan/operator.cpp

@@ -14,6 +14,7 @@
 #include <algorithm>
 #include <cstdint>
 #include <limits>
+#include <mutex>
 #include <queue>
 #include <random>
 #include <string>
@@ -42,6 +43,7 @@
 #include "query/v2/shard_request_manager.hpp"
 #include "storage/v3/conversions.hpp"
 #include "storage/v3/property_value.hpp"
+#include "typed_value.hpp"
 #include "utils/algorithm.hpp"
 #include "utils/csv_parsing.hpp"
 #include "utils/event_counter.hpp"
@@ -123,6 +125,8 @@ namespace memgraph::query::v2::plan {
 
 namespace {
 
+constexpr size_t kMultiFrameHeight = 1000;
+
 // Custom equality function for a vector of typed values.
 // Used in unordered_maps in Aggregate and Distinct operators.
 struct TypedValueVectorEqual {
@@ -217,7 +221,7 @@ class DistributedCreateNodeCursor : public Cursor {
       if (const auto *node_info_properties = std::get_if<PropertiesMapList>(&node_info->properties)) {
         for (const auto &[key, value_expression] : *node_info_properties) {
           TypedValue val = value_expression->Accept(evaluator);
-          if (context.shard_request_manager->IsPrimaryKey(primary_label, key)) {
+          if (context.shard_request_manager->IsPrimaryProperty(primary_label, key)) {
             rqst.primary_key.push_back(TypedValueToValue(val));
             pk.push_back(TypedValueToValue(val));
           }
@@ -227,7 +231,7 @@ class DistributedCreateNodeCursor : public Cursor {
         for (const auto &[key, value] : property_map) {
           auto key_str = std::string(key);
           auto property_id = context.shard_request_manager->NameToProperty(key_str);
-          if (context.shard_request_manager->IsPrimaryKey(primary_label, property_id)) {
+          if (context.shard_request_manager->IsPrimaryProperty(primary_label, property_id)) {
             rqst.primary_key.push_back(TypedValueToValue(value));
             pk.push_back(TypedValueToValue(value));
           }
@@ -257,13 +261,27 @@ class DistributedCreateNodeCursor : public Cursor {
 bool Once::OnceCursor::Pull(Frame &, ExecutionContext &context) {
   SCOPED_PROFILE_OP("Once");
 
-  if (!did_pull_) {
-    did_pull_ = true;
+  if (pull_count_ < 1) {
+    pull_count_++;
     return true;
   }
   return false;
 }
 
+void Once::OnceCursor::PullMultiple(MultiFrame &multi_frame, ExecutionContext &context) {
+  SCOPED_PROFILE_OP("OnceMF");
+
+  if (pull_count_ < 1) {
+    MG_ASSERT(!multi_frame.frames.empty());
+    multi_frame.valid_frames = 1;
+    auto *memory_resource = multi_frame.frames[0].GetMemoryResource();
+    for (auto &value : multi_frame.frames[0].elems()) {
+      value = TypedValue{memory_resource};
+    }
+    pull_count_++;
+  }
+}
+
 UniqueCursorPtr Once::MakeCursor(utils::MemoryResource *mem) const {
   EventCounter::IncrementCounter(EventCounter::OnceOperator);
 
@@ -274,7 +292,7 @@ WITHOUT_SINGLE_INPUT(Once);
 
 void Once::OnceCursor::Shutdown() {}
 
-void Once::OnceCursor::Reset() { did_pull_ = false; }
+void Once::OnceCursor::Reset() { pull_count_ = 0; }
 
 CreateNode::CreateNode(const std::shared_ptr<LogicalOperator> &input, const NodeCreationInfo &node_info)
     : input_(input ? input : std::make_shared<Once>()), node_info_(node_info) {}
@@ -385,10 +403,42 @@ class DistributedScanAllAndFilterCursor : public Cursor {
 
   using VertexAccessor = accessors::VertexAccessor;
 
-  bool MakeRequest(msgs::ShardRequestManagerInterface &shard_manager, ExecutionContext &context) {
+  bool MakeRequest(Frame &frame, ExecutionContext &context) {
+    auto &shard_request_manager = *context.shard_request_manager;
+    std::vector<msgs::VertexId> scanned_vertices;
+    if (property_expression_pair_.has_value()) {
+      MG_ASSERT(label_);
+      MG_ASSERT(shard_request_manager.IsPrimaryKey(*label_, property_expression_pair_->first));
+      ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context,
+                                    context.shard_request_manager, storage::v3::View::OLD);
+      auto key = property_expression_pair_->second->Accept(evaluator);
+      scanned_vertices.emplace_back(msgs::Label{.id = *label_}, std::vector<msgs::Value>{TypedValueToValue(key)});
+    }
     {
       SCOPED_REQUEST_WAIT_PROFILE;
-      current_batch = shard_manager.Request(request_state_);
+      current_batch = shard_request_manager.Request(request_state_, std::move(scanned_vertices));
+    }
+    current_vertex_it = current_batch.begin();
+    return !current_batch.empty();
+  }
+
+  bool MakeRequest(MultiFrame &multi_frame, ExecutionContext &context) {
+    auto &shard_request_manager = *context.shard_request_manager;
+    std::vector<msgs::VertexId> scanned_vertices;
+    if (property_expression_pair_.has_value()) {
+      MG_ASSERT(label_.has_value());
+      MG_ASSERT(shard_request_manager.IsPrimaryKey(*label_, property_expression_pair_->first));
+      for (auto frame_index = 0U; frame_index < multi_frame.valid_frames; ++frame_index) {
+        auto &frame = multi_frame.frames[frame_index];
+        ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context,
+                                      context.shard_request_manager, storage::v3::View::OLD);
+        auto key = property_expression_pair_->second->Accept(evaluator);
+        scanned_vertices.emplace_back(msgs::Label{.id = *label_}, std::vector<msgs::Value>{TypedValueToValue(key)});
+      }
+    }
+    {
+      SCOPED_REQUEST_WAIT_PROFILE;
+      current_batch = shard_request_manager.Request(request_state_, std::move(scanned_vertices));
     }
     current_vertex_it = current_batch.begin();
     return !current_batch.empty();
@@ -413,7 +463,7 @@ class DistributedScanAllAndFilterCursor : public Cursor {
       request_state_.label = label_.has_value() ? std::make_optional(shard_manager.LabelToName(*label_)) : std::nullopt;
 
       if (current_vertex_it == current_batch.end() &&
-          (request_state_.state == State::COMPLETED || !MakeRequest(shard_manager, context))) {
+          (request_state_.state == State::COMPLETED || !MakeRequest(frame, context))) {
         ResetExecutionState();
         continue;
       }
@@ -424,6 +474,75 @@ class DistributedScanAllAndFilterCursor : public Cursor {
     }
   }
 
+  void PullMultiple(MultiFrame &output_frames, ExecutionContext &context) override {
+    SCOPED_PROFILE_OP(op_name_);
+
+    EnsureBufferIsGood(output_frames);
+
+    using State = msgs::ExecutionState<msgs::ScanVerticesRequest>;
+    auto &shard_manager = *context.shard_request_manager;
+    // difference between lookup by id vs by label-property index:
+    // - by id: we are sure we only get at most 1 vertex by a single "lookup value" and it is easy to skip not matched
+    // vertices
+    // - by label-property index: we can get multiple vertices by a single "lookup value" and it might be complicated to
+    // skip
+    // Maybe it is best to separate these different functionalities, because if the default scan all behavior is
+    // considered also, they can behave very differently in terms of whether they are producing more or less rows.
+    const auto is_scan_by_id = property_expression_pair_.has_value();
+    MG_ASSERT(!is_scan_by_id);
+
+    while (true) {
+      if (buffer_.valid_frames == consumed_from_buffer_) {
+        consumed_from_buffer_ = 0;
+        buffer_.valid_frames = 0;
+        input_cursor_->PullMultiple(buffer_, context);
+        if (buffer_.valid_frames == 0) {
+          break;
+        }
+      }
+      MG_ASSERT(buffer_.valid_frames >= consumed_from_buffer_);
+
+      for (auto input_frame_index = consumed_from_buffer_; input_frame_index < buffer_.valid_frames;
+           input_frame_index++) {
+        auto &input_frame = buffer_.frames[input_frame_index];
+        request_state_.label =
+            label_.has_value() ? std::make_optional(shard_manager.LabelToName(*label_)) : std::nullopt;
+
+        if (current_vertex_it == current_batch.end()) {
+          MG_ASSERT(MakeRequest(input_frame, context));
+        }
+
+        for (auto output_frame_index = output_frames.valid_frames;
+             output_frame_index < output_frames.frames.size() && current_vertex_it != current_batch.end();
+             ++output_frame_index) {
+          auto &output_frame = output_frames.frames[output_frames.valid_frames++];
+          output_frame = input_frame;
+          output_frame[output_symbol_] = TypedValue(std::move(*current_vertex_it));
+          ++current_vertex_it;
+        }
+
+        if (current_vertex_it == current_batch.end()) {
+          MG_ASSERT(request_state_.state == State::COMPLETED, "For now we don't support pagination");
+          consumed_from_buffer_++;
+          ResetExecutionState();
+        }
+        if (output_frames.frames.size() == output_frames.valid_frames) {
+          return;
+        }
+      }
+    }
+  }
+
+  void EnsureBufferIsGood(const MultiFrame &output_frames) {
+    if (buffer_.frames.empty()) {
+      const auto &first_frame = output_frames.frames[0];
+      buffer_.frames = utils::pmr::vector<Frame>{
+          kMultiFrameHeight, Frame{static_cast<int64_t>(first_frame.elems().size()), first_frame.GetMemoryResource()},
+          first_frame.GetMemoryResource()};
+    }
+    MG_ASSERT(output_frames.frames[0].elems().size() == buffer_.frames[0].elems().size());
+  }
+
   void Shutdown() override { input_cursor_->Shutdown(); }
 
   void ResetExecutionState() {
@@ -447,6 +566,8 @@ class DistributedScanAllAndFilterCursor : public Cursor {
   std::optional<storage::v3::LabelId> label_;
   std::optional<std::pair<storage::v3::PropertyId, Expression *>> property_expression_pair_;
   std::optional<std::vector<Expression *>> filter_expressions_;
+  MultiFrame buffer_;
+  size_t consumed_from_buffer_{0};
 };
 
 ScanAll::ScanAll(const std::shared_ptr<LogicalOperator> &input, Symbol output_symbol, storage::v3::View view)
@@ -766,6 +887,19 @@ bool Produce::ProduceCursor::Pull(Frame &frame, ExecutionContext &context) {
   return false;
 }
 
+void Produce::ProduceCursor::PullMultiple(MultiFrame &multi_frame, ExecutionContext &context) {
+  SCOPED_PROFILE_OP("ProduceMF");
+
+  input_cursor_->PullMultiple(multi_frame, context);
+
+  for (auto row_index = 0U; row_index < multi_frame.valid_frames; row_index++) {
+    // Produce should always yield the latest results.
+    ExpressionEvaluator evaluator(&multi_frame.frames[row_index], context.symbol_table, context.evaluation_context,
+                                  context.shard_request_manager, storage::v3::View::NEW);
+    for (auto *named_expr : self_.named_expressions_) named_expr->Accept(evaluator);
+  }
+};
+
 void Produce::ProduceCursor::Shutdown() { input_cursor_->Shutdown(); }
 
 void Produce::ProduceCursor::Reset() { input_cursor_->Reset(); }
@@ -2450,7 +2584,7 @@ class DistributedCreateExpandCursor : public Cursor {
       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.shard_request_manager->IsPrimaryKey(vertex_id.first.id, key)) {
+          if (context.shard_request_manager->IsPrimaryProperty(vertex_id.first.id, key)) {
             vertex_id.second.push_back(val);
           }
         }
@@ -2524,17 +2658,18 @@ class DistributedExpandCursor : public Cursor {
           throw std::runtime_error("EdgeDirection Both not implemented");
       }
     };
-    msgs::ExpandOneRequest request;
-    // to not fetch any properties of the edges
-    request.edge_properties.emplace();
-    request.src_vertices.push_back(get_dst_vertex(edge, direction));
-    request.direction = (direction == EdgeAtom::Direction::IN) ? msgs::EdgeDirection::OUT : msgs::EdgeDirection::IN;
-    msgs::ExecutionState<msgs::ExpandOneRequest> request_state;
-    auto result_rows = context.shard_request_manager->Request(request_state, std::move(request));
-    MG_ASSERT(result_rows.size() == 1);
-    auto &result_row = result_rows.front();
+    // msgs::ExpandOneRequest request;
+    // // to not fetch any properties of the edges
+    // request.edge_properties.emplace();
+    // request.src_vertices.push_back(get_dst_vertex(edge, direction));
+    // request.direction = (direction == EdgeAtom::Direction::IN) ? msgs::EdgeDirection::OUT : msgs::EdgeDirection::IN;
+    // msgs::ExecutionState<msgs::ExpandOneRequest> request_state;
+    // auto result_rows = context.shard_request_manager->Request(request_state, std::move(request));
+    // MG_ASSERT(result_rows.size() == 1);
+    // auto &result_row = result_rows.front();
     frame[self_.common_.node_symbol] = accessors::VertexAccessor(
-        msgs::Vertex{result_row.src_vertex}, result_row.src_vertex_properties, context.shard_request_manager);
+        msgs::Vertex{get_dst_vertex(edge, direction)}, std::vector<std::pair<msgs::PropertyId, msgs::Value>>{},
+        context.shard_request_manager);
   }
 
   bool InitEdges(Frame &frame, ExecutionContext &context) {
@@ -2609,6 +2744,87 @@ class DistributedExpandCursor : public Cursor {
     }
   }
 
+  void InitEdgesMultiple(ExecutionContext &context) {
+    TypedValue &vertex_value = buffer_.frames[consumed_from_buffer_][self_.input_symbol_];
+
+    if (vertex_value.IsNull()) {
+      return;
+    }
+
+    ExpectType(self_.input_symbol_, vertex_value, TypedValue::Type::Vertex);
+    auto &vertex = vertex_value.ValueVertex();
+
+    const auto convert_edges = [&vertex, &context](
+                                   std::vector<msgs::ExpandOneResultRow::EdgeWithSpecificProperties> &&edge_messages,
+                                   const EdgeAtom::Direction direction) {
+      std::vector<EdgeAccessor> edge_accessors;
+      edge_accessors.reserve(edge_messages.size());
+
+      switch (direction) {
+        case EdgeAtom::Direction::IN: {
+          for (auto &edge : edge_messages) {
+            edge_accessors.emplace_back(msgs::Edge{std::move(edge.other_end), vertex.Id(), {}, {edge.gid}, edge.type},
+                                        context.shard_request_manager);
+          }
+          break;
+        }
+        case EdgeAtom::Direction::OUT: {
+          for (auto &edge : edge_messages) {
+            edge_accessors.emplace_back(msgs::Edge{vertex.Id(), std::move(edge.other_end), {}, {edge.gid}, edge.type},
+                                        context.shard_request_manager);
+          }
+          break;
+        }
+        case EdgeAtom::Direction::BOTH: {
+          LOG_FATAL("Must indicate exact expansion direction here");
+        }
+      }
+      return edge_accessors;
+    };
+
+    auto *result_row = vertex_id_to_result_row[vertex.Id()];
+    current_in_edges_.clear();
+    current_in_edges_ =
+        convert_edges(std::move(result_row->in_edges_with_specific_properties), EdgeAtom::Direction::IN);
+    current_in_edge_it_ = current_in_edges_.begin();
+    current_out_edges_ =
+        convert_edges(std::move(result_row->out_edges_with_specific_properties), EdgeAtom::Direction::OUT);
+    current_out_edge_it_ = current_out_edges_.begin();
+    vertex_id_to_result_row.erase(vertex.Id());
+  }
+
+  void PullEdgesFromStorage(ExecutionContext &context) {
+    // Input Vertex could be null if it is created by a failed optional match. In
+    // those cases we skip that input pull and continue with the next.
+
+    msgs::ExpandOneRequest request;
+    request.direction = DirectionToMsgsDirection(self_.common_.direction);
+    // to not fetch any properties of the edges
+    request.edge_properties.emplace();
+    for (auto frame_index = 0; frame_index < buffer_.valid_frames; ++frame_index) {
+      auto &frame = buffer_.frames[frame_index];
+      TypedValue &vertex_value = frame[self_.input_symbol_];
+
+      // Null check due to possible failed optional match.
+      MG_ASSERT(!vertex_value.IsNull());
+
+      ExpectType(self_.input_symbol_, vertex_value, TypedValue::Type::Vertex);
+      auto &vertex = vertex_value.ValueVertex();
+      request.src_vertices.push_back(vertex.Id());
+    }
+
+    msgs::ExecutionState<msgs::ExpandOneRequest> request_state;
+    result_rows_ = std::invoke([&context, &request_state, &request]() mutable {
+      SCOPED_REQUEST_WAIT_PROFILE;
+      return context.shard_request_manager->Request(request_state, std::move(request));
+    });
+    MG_ASSERT(result_rows_.size() == buffer_.valid_frames);
+    vertex_id_to_result_row.clear();
+    for (auto &row : result_rows_) {
+      vertex_id_to_result_row[row.src_vertex.id] = &row;
+    }
+  }
+
   bool Pull(Frame &frame, ExecutionContext &context) override {
     SCOPED_PROFILE_OP("DistributedExpand");
     // A helper function for expanding a node from an edge.
@@ -2644,6 +2860,77 @@ class DistributedExpandCursor : public Cursor {
     }
   }
 
+  void PullMultiple(MultiFrame &output_frames, ExecutionContext &context) override {
+    SCOPED_PROFILE_OP("DistributedExpandMF");
+    MG_ASSERT(!self_.common_.existing_node);
+    EnsureBufferIsGood(output_frames);
+    // A helper function for expanding a node from an edge.
+
+    while (true) {
+      if (MustAbort(context)) throw HintedAbortError();
+      if (consumed_from_buffer_ == buffer_.valid_frames) {
+        buffer_.valid_frames = 0;
+
+        if (input_cursor_->PullMultiple(buffer_, context); buffer_.valid_frames == 0) {
+          break;
+        }
+
+        PullEdgesFromStorage(context);
+        InitEdgesMultiple(context);
+      }
+
+      while (consumed_from_buffer_ < buffer_.valid_frames) {
+        auto &input_frame = buffer_.frames[consumed_from_buffer_];
+
+        for (auto output_frame_index = output_frames.valid_frames;
+             output_frame_index < output_frames.frames.size() && current_in_edge_it_ != current_in_edges_.end();
+             ++output_frame_index) {
+          auto &edge = *current_in_edge_it_;
+          ++current_in_edge_it_;
+          auto &output_frame = output_frames.frames[output_frames.valid_frames++];
+          output_frame = input_frame;
+          output_frame[self_.common_.edge_symbol] = edge;
+          PullDstVertex(output_frame, context, EdgeAtom::Direction::IN);
+        }
+
+        for (auto output_frame_index = output_frames.valid_frames;
+             output_frame_index < output_frames.frames.size() && current_out_edge_it_ != current_out_edges_.end();
+             ++output_frame_index) {
+          auto &edge = *current_out_edge_it_;
+          ++current_out_edge_it_;
+          // if (self_.common_.direction == EdgeAtom::Direction::BOTH && edge.IsCycle()) {
+          //   continue;
+          // };
+          auto &output_frame = output_frames.frames[output_frames.valid_frames++];
+          output_frame = input_frame;
+          output_frame[self_.common_.edge_symbol] = edge;
+          PullDstVertex(output_frame, context, EdgeAtom::Direction::OUT);
+        }
+        if (current_in_edge_it_ == current_in_edges_.end() && current_out_edge_it_ == current_out_edges_.end()) {
+          consumed_from_buffer_++;
+          if (consumed_from_buffer_ == buffer_.valid_frames) {
+            return;
+          }
+
+          InitEdgesMultiple(context);
+        }
+        if (output_frames.frames.size() == output_frames.valid_frames) {
+          return;
+        }
+      }
+    }
+  }
+
+  void EnsureBufferIsGood(const MultiFrame &output_frames) {
+    if (buffer_.frames.empty()) {
+      const auto &first_frame = output_frames.frames[0];
+      buffer_.frames = utils::pmr::vector<Frame>{
+          kMultiFrameHeight, Frame{static_cast<int64_t>(first_frame.elems().size()), first_frame.GetMemoryResource()},
+          first_frame.GetMemoryResource()};
+    }
+    MG_ASSERT(output_frames.frames[0].elems().size() == buffer_.frames[0].elems().size());
+  }
+
   void Shutdown() override { input_cursor_->Shutdown(); }
 
   void Reset() override {
@@ -2661,6 +2948,11 @@ class DistributedExpandCursor : public Cursor {
   std::vector<EdgeAccessor> current_out_edges_;
   std::vector<EdgeAccessor>::iterator current_in_edge_it_;
   std::vector<EdgeAccessor>::iterator current_out_edge_it_;
+  MultiFrame buffer_;
+  size_t consumed_from_buffer_{0U};
+  std::vector<msgs::ExpandOneResultRow> result_rows_;
+  // This won't work if any vertex id is duplicated in the input
+  std::unordered_map<msgs::VertexId, msgs::ExpandOneResultRow *> vertex_id_to_result_row;
 };
 
 }  // namespace memgraph::query::v2::plan

src/query/v2/plan/operator.lcp

@@ -71,6 +71,8 @@ class Cursor {
   /// @throws QueryRuntimeException if something went wrong with execution
   virtual bool Pull(Frame &, ExecutionContext &) = 0;
 
+  virtual void PullMultiple(MultiFrame &, ExecutionContext &) { LOG_FATAL("PullMultipleIsNotImplemented"); }
+
   /// Resets the Cursor to its initial state.
   virtual void Reset() = 0;
 
@@ -333,11 +335,12 @@ and false on every following Pull.")
     public:
      OnceCursor() {}
      bool Pull(Frame &, ExecutionContext &) override;
+     void PullMultiple(MultiFrame &, ExecutionContext &) override;
      void Shutdown() override;
      void Reset() override;
 
     private:
-     bool did_pull_{false};
+     size_t pull_count_{false};
    };
    cpp<#)
   (:serialize (:slk))
@@ -1207,6 +1210,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;
      void Shutdown() override;
      void Reset() override;
 

src/query/v2/plan/vertex_count_cache.hpp

@@ -9,7 +9,6 @@
 // by the Apache License, Version 2.0, included in the file
 // licenses/APL.txt.
 
-/// @file
 #pragma once
 
 #include <optional>
@@ -55,7 +54,9 @@ class VertexCountCache {
   // For now return true if label is primary label
   bool LabelIndexExists(storage::v3::LabelId label) { return shard_request_manager_->IsPrimaryLabel(label); }
 
-  bool LabelPropertyIndexExists(storage::v3::LabelId /*label*/, storage::v3::PropertyId /*property*/) { return false; }
+  bool LabelPropertyIndexExists(storage::v3::LabelId label, storage::v3::PropertyId property) {
+    return shard_request_manager_->IsPrimaryKey(label, property);
+  }
 
   msgs::ShardRequestManagerInterface *shard_request_manager_;
 };

src/query/v2/requests.hpp

@@ -24,6 +24,7 @@
 #include "coordinator/hybrid_logical_clock.hpp"
 #include "storage/v3/id_types.hpp"
 #include "storage/v3/property_value.hpp"
+#include "utils/fnv.hpp"
 
 namespace memgraph::msgs {
 
@@ -338,6 +339,7 @@ struct ScanVerticesRequest {
   Hlc transaction_id;
   // This should be optional
   VertexId start_id;
+  std::vector<VertexId> scanned_vertices;
   //  The empty optional means return all of the properties, while an empty list means do not return any properties
   std::optional<std::vector<PropertyId>> props_to_return;
   // expression that determines if vertex is returned or not
@@ -561,3 +563,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/query/v2/shard_request_manager.hpp

@@ -115,7 +115,8 @@ class ShardRequestManagerInterface {
 
   virtual void StartTransaction() = 0;
   virtual void Commit() = 0;
-  virtual std::vector<VertexAccessor> Request(ExecutionState<ScanVerticesRequest> &state) = 0;
+  virtual std::vector<VertexAccessor> Request(ExecutionState<ScanVerticesRequest> &state,
+                                              std::vector<VertexId> &&scanned_vertices) = 0;
   virtual std::vector<CreateVerticesResponse> Request(ExecutionState<CreateVerticesRequest> &state,
                                                       std::vector<NewVertex> new_vertices) = 0;
   virtual std::vector<ExpandOneResultRow> Request(ExecutionState<ExpandOneRequest> &state,
@@ -130,6 +131,7 @@ class ShardRequestManagerInterface {
   virtual const std::string &LabelToName(memgraph::storage::v3::LabelId label) const = 0;
   virtual const std::string &EdgeTypeToName(memgraph::storage::v3::EdgeTypeId type) const = 0;
   virtual bool IsPrimaryLabel(LabelId label) const = 0;
+  virtual bool IsPrimaryProperty(LabelId primary_label, PropertyId property) const = 0;
   virtual bool IsPrimaryKey(LabelId primary_label, PropertyId property) const = 0;
 };
 
@@ -235,7 +237,9 @@ class ShardRequestManager : public ShardRequestManagerInterface {
     return edge_types_.IdToName(id.AsUint());
   }
 
-  bool IsPrimaryKey(LabelId primary_label, PropertyId property) const override {
+  bool IsPrimaryLabel(LabelId label) const override { return shards_map_.label_spaces.contains(label); }
+
+  bool IsPrimaryProperty(LabelId primary_label, 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());
 
@@ -244,11 +248,17 @@ class ShardRequestManager : public ShardRequestManagerInterface {
            }) != schema_it->second.end();
   }
 
-  bool IsPrimaryLabel(LabelId label) const override { return shards_map_.label_spaces.contains(label); }
+  bool IsPrimaryKey(LabelId primary_label, 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());
+
+    return schema_it->second.size() == 1 && schema_it->second[0].property_id == property;
+  }
 
   // TODO(kostasrim) Simplify return result
-  std::vector<VertexAccessor> Request(ExecutionState<ScanVerticesRequest> &state) override {
-    MaybeInitializeExecutionState(state);
+  std::vector<VertexAccessor> Request(ExecutionState<ScanVerticesRequest> &state,
+                                      std::vector<VertexId> &&scanned_vertices) override {
+    MaybeInitializeExecutionState(state, std::move(scanned_vertices));
     std::vector<ScanVerticesResponse> responses;
 
     SendAllRequests(state);
@@ -397,8 +407,8 @@ class ShardRequestManager : public ShardRequestManagerInterface {
 
     for (auto &new_vertex : new_vertices) {
       MG_ASSERT(!new_vertex.label_ids.empty(), "This is error!");
-      auto shard = shards_map_.GetShardForKey(new_vertex.label_ids[0].id,
-                                              storage::conversions::ConvertPropertyVector(new_vertex.primary_key));
+      const auto &shard = shards_map_.GetShardForKey(
+          new_vertex.label_ids[0].id, storage::conversions::ConvertPropertyVector(new_vertex.primary_key));
       if (!per_shard_request_table.contains(shard)) {
         CreateVerticesRequest create_v_rqst{.transaction_id = transaction_id_};
         per_shard_request_table.insert(std::pair(shard, std::move(create_v_rqst)));
@@ -430,9 +440,9 @@ class ShardRequestManager : public ShardRequestManagerInterface {
     };
 
     for (auto &new_expand : new_expands) {
-      const auto shard_src_vertex = shards_map_.GetShardForKey(
+      const auto &shard_src_vertex = shards_map_.GetShardForKey(
           new_expand.src_vertex.first.id, storage::conversions::ConvertPropertyVector(new_expand.src_vertex.second));
-      const auto shard_dest_vertex = shards_map_.GetShardForKey(
+      const auto &shard_dest_vertex = shards_map_.GetShardForKey(
           new_expand.dest_vertex.first.id, storage::conversions::ConvertPropertyVector(new_expand.dest_vertex.second));
 
       ensure_shard_exists_in_table(shard_src_vertex);
@@ -451,30 +461,47 @@ class ShardRequestManager : public ShardRequestManagerInterface {
     state.state = ExecutionState<CreateExpandRequest>::EXECUTING;
   }
 
-  void MaybeInitializeExecutionState(ExecutionState<ScanVerticesRequest> &state) {
+  void MaybeInitializeExecutionState(ExecutionState<ScanVerticesRequest> &state,
+                                     std::vector<VertexId> &&scanned_vertices) {
     ThrowIfStateCompleted(state);
     if (ShallNotInitializeState(state)) {
       return;
     }
 
-    std::vector<coordinator::Shards> multi_shards;
-    state.transaction_id = transaction_id_;
-    if (!state.label) {
-      multi_shards = shards_map_.GetAllShards();
+    if (!scanned_vertices.empty()) {
+      MG_ASSERT(state.label.has_value(), "Fill out the label member of the state!");
+      std::map<Shard, ScanVerticesRequest> shards_and_requests;
+      for (auto &vertex_id : scanned_vertices) {
+        const auto &shard = shards_map_.GetShardForKey(vertex_id.first.id,
+                                                       storage::conversions::ConvertPropertyVector(vertex_id.second));
+        shards_and_requests[shard].scanned_vertices.push_back(std::move(vertex_id));
+      }
+
+      for (auto &[shard, request] : shards_and_requests) {
+        state.shard_cache.push_back(shard);
+        request.transaction_id = transaction_id_;
+        state.requests.push_back(std::move(request));
+      }
     } else {
-      const auto label_id = shards_map_.GetLabelId(*state.label);
-      MG_ASSERT(label_id);
-      MG_ASSERT(IsPrimaryLabel(*label_id));
-      multi_shards = {shards_map_.GetShardsForLabel(*state.label)};
-    }
-    for (auto &shards : multi_shards) {
-      for (auto &[key, shard] : shards) {
-        MG_ASSERT(!shard.empty());
-        state.shard_cache.push_back(std::move(shard));
-        ScanVerticesRequest rqst;
-        rqst.transaction_id = transaction_id_;
-        rqst.start_id.second = storage::conversions::ConvertValueVector(key);
-        state.requests.push_back(std::move(rqst));
+      std::vector<coordinator::Shards> multi_shards;
+      state.transaction_id = transaction_id_;
+      if (!state.label) {
+        multi_shards = shards_map_.GetAllShards();
+      } else {
+        const auto label_id = shards_map_.GetLabelId(*state.label);
+        MG_ASSERT(label_id);
+        MG_ASSERT(IsPrimaryLabel(*label_id));
+        multi_shards = {shards_map_.GetShardsForLabel(*state.label)};
+      }
+      for (auto &shards : multi_shards) {
+        for (auto &[key, shard] : shards) {
+          MG_ASSERT(!shard.empty());
+          state.shard_cache.push_back(std::move(shard));
+          ScanVerticesRequest rqst;
+          rqst.transaction_id = transaction_id_;
+          rqst.start_id.second = storage::conversions::ConvertValueVector(key);
+          state.requests.push_back(std::move(rqst));
+        }
       }
     }
     state.state = ExecutionState<ScanVerticesRequest>::EXECUTING;
@@ -493,16 +520,16 @@ class ShardRequestManager : public ShardRequestManagerInterface {
     top_level_rqst_template.src_vertices.clear();
     state.requests.clear();
     for (auto &vertex : request.src_vertices) {
-      auto shard =
+      const auto &shard =
           shards_map_.GetShardForKey(vertex.first.id, storage::conversions::ConvertPropertyVector(vertex.second));
       if (!per_shard_request_table.contains(shard)) {
         per_shard_request_table.insert(std::pair(shard, top_level_rqst_template));
-        state.shard_cache.push_back(shard);
       }
-      per_shard_request_table[shard].src_vertices.push_back(vertex);
+      per_shard_request_table[shard].src_vertices.push_back(std::move(vertex));
     }
 
     for (auto &[shard, rqst] : per_shard_request_table) {
+      state.shard_cache.push_back(shard);
       state.requests.push_back(std::move(rqst));
     }
     state.state = ExecutionState<ExpandOneRequest>::EXECUTING;

src/storage/v3/edge.hpp

@@ -36,6 +36,7 @@ struct Edge {
   bool deleted;
   // uint8_t PAD;
   // uint16_t PAD;
+  // uint32_t PAD;
 
   Delta *delta;
 };

src/storage/v3/request_helper.cpp

@@ -59,6 +59,11 @@ std::vector<Element> OrderByElements(Shard::Accessor &acc, DbAccessor &dba, Vert
 VerticesIterable::Iterator GetStartVertexIterator(VerticesIterable &vertex_iterable,
                                                   const std::vector<PropertyValue> &start_ids, const View view) {
   auto it = vertex_iterable.begin();
+
+  if (start_ids.empty()) {
+    return it;
+  }
+
   while (it != vertex_iterable.end()) {
     if (const auto &vertex = *it; start_ids <= vertex.PrimaryKey(view).GetValue()) {
       break;

src/storage/v3/shard_rsm.cpp

@@ -836,10 +836,18 @@ msgs::ReadResponses ShardRsm::HandleRead(msgs::ScanVerticesRequest &&req) {
   const auto start_id = ConvertPropertyVector(std::move(req.start_id.second));
   uint64_t sample_counter{0};
   auto vertex_iterable = acc.Vertices(view);
-  if (!req.order_bys.empty()) {
+  if (!req.scanned_vertices.empty()) {
+    for (auto &scanned_vertex_id : req.scanned_vertices) {
+      if (auto maybe_vertex = acc.FindVertex(ConvertPropertyVector(std::move(scanned_vertex_id.second)), view);
+          maybe_vertex.has_value()) {
+        emplace_scan_result(*maybe_vertex);
+      }
+    }
+
+  } else if (!req.order_bys.empty()) {
     const auto ordered = OrderByElements(acc, dba, vertex_iterable, req.order_bys);
     // we are traversing Elements
-    auto it = GetStartOrderedElementsIterator(ordered, start_id, View(req.storage_view));
+    auto it = GetStartOrderedElementsIterator(ordered, start_id, view);
     for (; it != ordered.end(); ++it) {
       emplace_scan_result(it->vertex_acc);
       ++sample_counter;

tests/simulation/sharded_map.cpp

@@ -294,7 +294,7 @@ int main() {
       client_shard_map = hlc_response.fresher_shard_map.value();
     }
 
-    auto target_shard = client_shard_map.GetShardForKey(label_name, compound_key);
+    const auto &target_shard = client_shard_map.GetShardForKey(label_name, compound_key);
 
     // Determine which shard to send the requests to. This should be a more proper client cache in the "real" version.
     auto storage_client_opt = DetermineShardLocation(target_shard, a_addrs, shard_a_client, b_addrs, shard_b_client);

tests/simulation/test_cluster.hpp

@@ -182,7 +182,7 @@ void ExecuteOp(msgs::ShardRequestManager<SimulatorTransport> &shard_request_mana
                std::set<CompoundKey> &correctness_model, ScanAll scan_all) {
   msgs::ExecutionState<msgs::ScanVerticesRequest> request{.label = "test_label"};
 
-  auto results = shard_request_manager.Request(request);
+  auto results = shard_request_manager.Request(request, {});
 
   RC_ASSERT(results.size() == correctness_model.size());
 

tests/unit/high_density_shard_create_scan.cpp

@@ -196,7 +196,7 @@ void ExecuteOp(msgs::ShardRequestManager<LocalTransport> &shard_request_manager,
                std::set<CompoundKey> &correctness_model, ScanAll scan_all) {
   msgs::ExecutionState<msgs::ScanVerticesRequest> request{.label = "test_label"};
 
-  auto results = shard_request_manager.Request(request);
+  auto results = shard_request_manager.Request(request, {});
 
   MG_ASSERT(results.size() == correctness_model.size());
 

tests/unit/machine_manager.cpp

@@ -113,7 +113,7 @@ template <typename ShardRequestManager>
 void TestScanAll(ShardRequestManager &shard_request_manager) {
   msgs::ExecutionState<msgs::ScanVerticesRequest> state{.label = kLabelName};
 
-  auto result = shard_request_manager.Request(state);
+  auto result = shard_request_manager.Request(state, {});
   EXPECT_EQ(result.size(), 2);
 }