Impl of Scan with filters (unfinished)

Changed ScanAll request return type to be friendlier for use.

src/query/v2/plan/operator.cpp

@@ -2703,10 +2703,27 @@ bool Foreach::Accept(HierarchicalLogicalOperatorVisitor &visitor) {
   return visitor.PostVisit(*this);
 }
 
+/*
+#NoCommit #JBA we will need to also modify the request sent (attributes?) + the client of storage + the storage side
+for the string version we need to stringify the expression
+perhaps not needed now for customer?
+Note: operators are not yet connected to anything are they? For now I think the plan still use the non-distrib version
+*/
+
 class DistributedScanAllCursor : public Cursor {
  public:
-  explicit DistributedScanAllCursor(Symbol output_symbol, UniqueCursorPtr input_cursor, const char *op_name)
-      : output_symbol_(output_symbol), input_cursor_(std::move(input_cursor)), op_name_(op_name) {}
+  explicit DistributedScanAllCursor(
+      Symbol output_symbol, UniqueCursorPtr input_cursor, const char *op_name,
+      std::optional<storage::v3::LabelId> label_to_filter, std::optional<storage::v3::PropertyId> property_to_filter,
+      std::optional<std::vector<std::pair<storage::v3::PropertyId, Expression *>>> property_value_pairs_to_filter)
+      : output_symbol_(output_symbol),
+        input_cursor_(std::move(input_cursor)),
+        op_name_(op_name),
+        label_to_filter_(label_to_filter),
+        property_to_filter_(property_to_filter),
+        property_value_pairs_to_filter_(property_value_pairs_to_filter) {
+    ResetExecutionState();
+  }
 
   using VertexAccessor = accessors::VertexAccessor;
 
@@ -2744,6 +2761,47 @@ class DistributedScanAllCursor : public Cursor {
     current_batch.clear();
     current_vertex_it = current_batch.end();
     request_state_ = msgs::ExecutionState<msgs::ScanVerticesRequest>{};
+
+    /*
+    Many questions here:
+    Where are all attributes of ScanVerticesRequest init?
+
+    Then why vector of request? What do we expect?
+    How do we build the vector? What are the several elements corresponding to?
+    Where do we iterate on those requests?
+
+    Convert the Expression* to string version of ast? or something?
+    We gonna have to break down the expression at some point somewhere
+
+    Is there a way to Ut something? Perhaps we can UT the creatiuon of request and compare we exactly get the
+    ScanVerticesRequest we want
+    */
+    auto request = msgs::ScanVerticesRequest{};
+    if (label_to_filter_.has_value()) {
+      request.label_to_filter_ = msgs::Label{.id = label_to_filter_.value()};
+    }
+    if (property_to_filter_.has_value()) {
+      request.property_to_filter_ = property_to_filter_;
+    }
+    if (property_value_pairs_to_filter_.has_value()) {
+      auto res = std::vector<std::pair<storage::v3::PropertyId, std::string>>{};  // #NoCommit not int
+      res.reserve(property_value_pairs_to_filter_->size());
+      std::transform(
+          property_value_pairs_to_filter_->begin(), property_value_pairs_to_filter_->end(), std::back_inserter(res),
+          [](auto &property_value_pair) {
+            return std::make_pair(
+                property_value_pair.first,
+                std::string("") /* ToString(property_value_pair.second)*/);  // #NoCommit call expression.tostring? see
+                                                                             // the json thing?
+            // #NoCommit we need to replace the node by a special character (sequence?) that the storage will
+            // identify as a node to know where is the symbol on the frame #NoCommit not bool but something
+            // with property_value_pair.second?
+          });
+
+      request.property_value_pairs_to_filter_ = res;
+    }
+
+    request_state_.requests.emplace_back(request);
   }
 
   void Reset() override {
@@ -2758,6 +2816,9 @@ class DistributedScanAllCursor : public Cursor {
   std::vector<VertexAccessor> current_batch;
   decltype(std::vector<VertexAccessor>().begin()) current_vertex_it;
   msgs::ExecutionState<msgs::ScanVerticesRequest> request_state_;
+  std::optional<storage::v3::LabelId> label_to_filter_;
+  std::optional<storage::v3::PropertyId> property_to_filter_;
+  std::optional<std::vector<std::pair<storage::v3::PropertyId, Expression *>>> property_value_pairs_to_filter_;
 };
 
 class DistributedCreateNodeCursor : public Cursor {
@@ -2824,4 +2885,62 @@ class DistributedCreateNodeCursor : public Cursor {
   std::vector<NodeCreationInfo> nodes_info_;
   msgs::ExecutionState<msgs::CreateVerticesRequest> state_;
 };
+
+class DistributedUnwindCursor : public Cursor {
+ public:
+  DistributedUnwindCursor(const Unwind &self, utils::MemoryResource *mem)
+      : self_(self), input_cursor_(self.input_->MakeCursor(mem)), input_value_(mem) {}
+
+  bool Pull(Frame &frame, ExecutionContext &context) override {
+    SCOPED_PROFILE_OP("Unwind");
+    while (true) {
+      if (MustAbort(context)) {
+        throw HintedAbortError();
+      }
+      // if we reached the end of our list of values
+      // pull from the input
+      if (input_value_it_ == input_value_.end()) {
+        if (!input_cursor_->Pull(frame, context)) {
+          return false;
+        }
+
+        // successful pull from input, initialize value and iterator
+        ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, context.db_accessor,
+                                      storage::v3::View::OLD);
+        TypedValue input_value = self_.input_expression_->Accept(evaluator);
+        if (input_value.type() != TypedValue::Type::List) {
+          throw QueryRuntimeException("Argument of UNWIND must be a list, but '{}' was provided.", input_value.type());
+        }
+        // Copy the evaluted input_value_list to our vector.
+        input_value_ = input_value.ValueList();
+        input_value_it_ = input_value_.begin();
+      }
+
+      // if we reached the end of our list of values goto back to top
+      if (input_value_it_ == input_value_.end()) {
+        continue;
+      }
+
+      frame[self_.output_symbol_] = *input_value_it_++;
+      return true;
+    }
+  }
+
+  void Shutdown() override { input_cursor_->Shutdown(); }
+
+  void Reset() override {
+    input_cursor_->Reset();
+    input_value_.clear();
+    input_value_it_ = input_value_.end();
+  }
+
+ private:
+  const Unwind &self_;
+  const UniqueCursorPtr input_cursor_;
+  // typed values we are unwinding and yielding
+  utils::pmr::vector<TypedValue> input_value_;
+  // current position in input_value_
+  decltype(input_value_)::iterator input_value_it_ = input_value_.end();
+};
+
 }  // namespace memgraph::query::v2::plan

src/query/v2/requests.hpp

@@ -376,13 +376,19 @@ struct OrderBy {
 
 enum class StorageView { OLD = 0, NEW = 1 };
 
+// #NoCommit #JBA to modify
 struct ScanVerticesRequest {
   Hlc transaction_id;
   VertexId start_id;
   std::optional<std::vector<PropertyId>> props_to_return;
-  std::optional<std::vector<std::string>> filter_expressions;
+  std::optional<std::vector<std::string>> filter_expressions;  // #NoCommit what the use of that exactly?
   std::optional<size_t> batch_limit;
   StorageView storage_view;
+
+  std::optional<Label> label_to_filter_;
+  std::optional<PropertyId> property_to_filter_;
+  std::optional<std::vector<std::pair<storage::v3::PropertyId, std::string>>> property_value_pairs_to_filter_;
+  // #NoCommit not bool but the string version of the ast? to double check
 };
 
 struct ScanResultRow {

tests/unit/CMakeLists.txt

@@ -433,7 +433,3 @@ target_link_libraries(${test_prefix}local_transport mg-io)
 # Test MachineManager with LocalTransport
 add_unit_test(machine_manager.cpp)
 target_link_libraries(${test_prefix}machine_manager mg-io mg-coordinator mg-storage-v3)
-
-# Test ExpressionEvaluator with expressions
-add_unit_test(storage_v3_expr_usage.cpp)
-target_link_libraries(${test_prefix}storage_v3_expr_usage mg-io mg-expr mg-storage-v3)

tests/unit/storage_v3_expr_usage.cpp

@@ -1,260 +0,0 @@
-// 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.
-
-#include <chrono>
-#include <limits>
-#include <thread>
-
-#include <gtest/gtest.h>
-
-#include <coordinator/coordinator.hpp>
-#include <coordinator/coordinator_client.hpp>
-#include <coordinator/hybrid_logical_clock.hpp>
-#include <coordinator/shard_map.hpp>
-#include <io/local_transport/local_system.hpp>
-#include <io/local_transport/local_transport.hpp>
-#include <io/rsm/rsm_client.hpp>
-#include <io/transport.hpp>
-#include <machine_manager/machine_config.hpp>
-#include <machine_manager/machine_manager.hpp>
-#include "common/types.hpp"
-#include "exceptions.hpp"
-#include "io/rsm/rsm_client.hpp"
-#include "parser/opencypher/parser.hpp"
-#include "storage/v3/bindings/ast/ast.hpp"
-#include "storage/v3/bindings/bindings.hpp"
-#include "storage/v3/bindings/cypher_main_visitor.hpp"
-#include "storage/v3/bindings/db_accessor.hpp"
-#include "storage/v3/bindings/eval.hpp"
-#include "storage/v3/bindings/frame.hpp"
-#include "storage/v3/bindings/symbol_generator.hpp"
-#include "storage/v3/bindings/symbol_table.hpp"
-#include "storage/v3/bindings/typed_value.hpp"
-#include "storage/v3/id_types.hpp"
-#include "storage/v3/key_store.hpp"
-#include "storage/v3/property_value.hpp"
-#include "storage/v3/schemas.hpp"
-#include "storage/v3/shard.hpp"
-#include "utils/string.hpp"
-
-namespace memgraph::storage::v3::test {
-
-class ExpressionEvaluatorUsageTest : public ::testing::Test {
- protected:
-  LabelId primary_label{LabelId::FromInt(1)};
-  PropertyId primary_property{PropertyId::FromInt(2)};
-  PrimaryKey min_pk{PropertyValue(0)};
-
-  Shard db{primary_label, min_pk, std::nullopt};
-  Shard::Accessor storage_dba{db.Access(GetNextHlc())};
-  DbAccessor dba{&storage_dba};
-
-  AstStorage storage;
-  memgraph::utils::MonotonicBufferResource mem{1024};
-  EvaluationContext ctx{.memory = &mem};
-  SymbolTable symbol_table_;
-
-  Frame frame_{128};
-  ExpressionEvaluator eval{&frame_, symbol_table_, ctx, &dba, View::NEW};
-
-  coordinator::Hlc last_hlc{0, io::Time{}};
-
-  void SetUp() override {
-    db.StoreMapping({{1, "label"}, {2, "property"}});
-    ASSERT_TRUE(
-        db.CreateSchema(primary_label, {storage::v3::SchemaProperty{primary_property, common::SchemaType::INT}}));
-  }
-
-  std::vector<PropertyId> NamesToProperties(const std::vector<std::string> &property_names) {
-    std::vector<PropertyId> properties;
-    properties.reserve(property_names.size());
-    for (const auto &name : property_names) {
-      properties.push_back(dba.NameToProperty(name));
-    }
-    return properties;
-  }
-
-  std::vector<LabelId> NamesToLabels(const std::vector<std::string> &label_names) {
-    std::vector<LabelId> labels;
-    labels.reserve(label_names.size());
-    for (const auto &name : label_names) {
-      labels.push_back(dba.NameToLabel(name));
-    }
-    return labels;
-  }
-
-  Identifier *CreateIdentifierWithValue(std::string name, const TypedValue &value) {
-    auto *id = storage.Create<Identifier>(name, true);
-    auto symbol = symbol_table_.CreateSymbol(name, true);
-    id->MapTo(symbol);
-    frame_[symbol] = value;
-    return id;
-  }
-
-  template <class TExpression>
-  auto Eval(TExpression *expr) {
-    ctx.properties = NamesToProperties(storage.properties_);
-    ctx.labels = NamesToLabels(storage.labels_);
-    auto value = expr->Accept(eval);
-    EXPECT_EQ(value.GetMemoryResource(), &mem) << "ExpressionEvaluator must use the MemoryResource from "
-                                                  "EvaluationContext for allocations!";
-    return value;
-  }
-
-  coordinator::Hlc GetNextHlc() {
-    ++last_hlc.logical_id;
-    last_hlc.coordinator_wall_clock += std::chrono::seconds(1);
-    return last_hlc;
-  }
-};
-
-class StrippedQuery {
- public:
-  /**
-   * Strips the input query and stores stripped query, stripped arguments and
-   * stripped query hash.
-   *
-   * @param query Input query.
-   */
-  explicit StrippedQuery(const std::string &query);
-
-  /**
-   * Copy constructor is deleted because we don't want to make unnecessary
-   * copies of this object (copying of string and vector could be expensive)
-   */
-  StrippedQuery(const StrippedQuery &other) = delete;
-  StrippedQuery &operator=(const StrippedQuery &other) = delete;
-
-  /**
-   * Move is allowed operation because it is not expensive and we can
-   * move the object after it was created.
-   */
-  StrippedQuery(StrippedQuery &&other) = default;
-  StrippedQuery &operator=(StrippedQuery &&other) = default;
-
-  const std::string &query() const { return query_; }
-  const auto &original_query() const { return original_; }
-  const auto &literals() const { return literals_; }
-  const auto &named_expressions() const { return named_exprs_; }
-  const auto &parameters() const { return parameters_; }
-  uint64_t hash() const { return hash_; }
-
- private:
-  // Return len of matched keyword if something is matched, otherwise 0.
-  int MatchKeyword(int start) const;
-  int MatchString(int start) const;
-  int MatchSpecial(int start) const;
-  int MatchDecimalInt(int start) const;
-  int MatchOctalInt(int start) const;
-  int MatchHexadecimalInt(int start) const;
-  int MatchReal(int start) const;
-  int MatchParameter(int start) const;
-  int MatchEscapedName(int start) const;
-  int MatchUnescapedName(int start) const;
-  int MatchWhitespaceAndComments(int start) const;
-
-  // Original query.
-  std::string original_;
-
-  // Stripped query.
-  std::string query_;
-
-  // Token positions of stripped out literals mapped to their values.
-  // TODO: Parameters class really doesn't provide anything interesting. This
-  // could be changed to std::unordered_map, but first we need to rewrite (or
-  // get rid of) hardcoded queries which expect Parameters.
-  Parameters literals_;
-
-  // Token positions of query parameters mapped to their names.
-  std::unordered_map<int, std::string> parameters_;
-
-  // Token positions of nonaliased named expressions in return statement mapped
-  // to their original (unstripped) string.
-  std::unordered_map<int, std::string> named_exprs_;
-
-  // Hash based on the stripped query.
-  uint64_t hash_;
-};
-
-TEST_F(ExpressionEvaluatorUsageTest, DummyExample) {
-  // dummy
-  const std::string expr1{"2+1-3+2"};
-
-  // Parse stuff
-  memgraph::frontend::opencypher::Parser<frontend::opencypher::ParserOpTag::EXPRESSION> parser(expr1);
-  expr::ParsingContext pc;
-  CypherMainVisitor visitor(pc, &storage);
-
-  auto *ast = parser.tree();
-  auto ladida = visitor.visit(ast);
-  auto res1 = Eval(std::any_cast<Expression *>(ladida));
-  EXPECT_EQ(res1.ValueInt(), 2);
-}
-
-TEST_F(ExpressionEvaluatorUsageTest, PropertyLookup) {
-  db.StoreMapping({{1, "label"}, {2, "property"}, {3, "prop2"}});
-  const auto prop2 = PropertyId::FromUint(3);
-
-  auto v1 = *dba.InsertVertexAndValidate(primary_label, {}, {{primary_property, PropertyValue(1)}});
-  ASSERT_TRUE(v1.SetPropertyAndValidate(prop2, PropertyValue(5)).HasValue());
-
-  auto v2 = *dba.InsertVertexAndValidate(primary_label, {}, {{primary_property, PropertyValue(2)}});
-  ASSERT_TRUE(v2.SetPropertyAndValidate(prop2, PropertyValue(5)).HasValue());
-
-  auto v3 = *dba.InsertVertexAndValidate(primary_label, {}, {{primary_property, PropertyValue(3)}});
-  ASSERT_TRUE(v3.SetPropertyAndValidate(prop2, PropertyValue(5)).HasValue());
-
-  // Property filtering
-  // const std::string expr1{"n.prop2 > 0"};
-  const std::string expr1{"node.prop2 > 0 AND node.prop2 < 10"};
-
-  // Parse stuff
-  memgraph::frontend::opencypher::Parser<memgraph::frontend::opencypher::ParserOpTag::EXPRESSION> parser(expr1);
-  expr::ParsingContext pc;
-  CypherMainVisitor visitor(pc, &storage);
-
-  auto *ast = parser.tree();
-  auto expr = visitor.visit(ast);
-
-  static constexpr const char *node_name = "node";
-  static constexpr const char *edge_name = "edge";
-
-  static Identifier node_identifier = Identifier(std::string(node_name), false);
-  bool is_node_identifier_present = false;
-  static Identifier edge_identifier = Identifier(std::string(edge_name), false);
-  bool is_edge_identifier_present = false;
-
-  std::vector<Identifier *> identifiers;
-
-  if (expr1.find(node_name) != std::string::npos) {
-    is_node_identifier_present = true;
-    identifiers.push_back(&node_identifier);
-  }
-  if (expr1.find(edge_name) != std::string::npos) {
-    is_edge_identifier_present = true;
-    identifiers.push_back(&edge_identifier);
-  }
-
-  expr::SymbolGenerator symbol_generator(&symbol_table_, identifiers);
-  (std::any_cast<Expression *>(expr))->Accept(symbol_generator);
-
-  if (is_node_identifier_present) {
-    frame_[symbol_table_.at(node_identifier)] = v2;  // vertex accessor
-  }
-  if (is_edge_identifier_present) {
-    frame_[symbol_table_.at(edge_identifier)] = v1;  // edge accessor
-  }
-
-  auto res1 = Eval(std::any_cast<Expression *>(expr));
-  ASSERT_TRUE(res1.ValueBool());
-}
-
-}  // namespace memgraph::storage::v3::test