Merge branch 'master' into T635-add-vertex-degree-to-query-planner

src/CMakeLists.txt

@@ -16,10 +16,7 @@ add_subdirectory(slk)
 add_subdirectory(rpc)
 add_subdirectory(license)
 add_subdirectory(auth)
-
-if(MG_ENTERPRISE)
-        add_subdirectory(audit)
-endif()
+add_subdirectory(audit)
 
 string(TOLOWER ${CMAKE_BUILD_TYPE} lower_build_type)
 
@@ -36,12 +33,7 @@ set(mg_single_node_v2_sources
 )
 
 set(mg_single_node_v2_libs stdc++fs Threads::Threads
-        mg-telemetry mg-query mg-communication mg-memory mg-utils mg-auth mg-license mg-settings mg-glue)
-
-if(MG_ENTERPRISE)
-        # These are enterprise subsystems
-        set(mg_single_node_v2_libs ${mg_single_node_v2_libs} mg-audit)
-endif()
+        mg-telemetry mg-query mg-communication mg-memory mg-utils mg-auth mg-license mg-settings mg-glue mg-audit)
 
 # memgraph main executable
 add_executable(memgraph ${mg_single_node_v2_sources})

src/memgraph.cpp

@@ -34,6 +34,7 @@
 #include <spdlog/sinks/dist_sink.h>
 #include <spdlog/sinks/stdout_color_sinks.h>
 
+#include "audit/log.hpp"
 #include "auth/models.hpp"
 #include "communication/bolt/v1/constants.hpp"
 #include "communication/http/server.hpp"
@@ -98,10 +99,6 @@
 #include "auth/auth.hpp"
 #include "glue/auth.hpp"
 
-#ifdef MG_ENTERPRISE
-#include "audit/log.hpp"
-#endif
-
 constexpr const char *kMgUser = "MEMGRAPH_USER";
 constexpr const char *kMgPassword = "MEMGRAPH_PASSWORD";
 constexpr const char *kMgPassfile = "MEMGRAPH_PASSFILE";
@@ -473,31 +470,30 @@ struct SessionData {
 DEFINE_string(auth_user_or_role_name_regex, memgraph::glue::kDefaultUserRoleRegex.data(),
               "Set to the regular expression that each user or role name must fulfill.");
 
-void InitFromCypherlFile(memgraph::query::InterpreterContext &ctx, std::string cypherl_file_path
-#ifdef MG_ENTERPRISE
-                         ,
-                         memgraph::audit::Log *audit_log
-#endif
-) {
+void InitFromCypherlFile(memgraph::query::InterpreterContext &ctx, std::string cypherl_file_path,
+                         memgraph::audit::Log *audit_log = nullptr) {
   memgraph::query::Interpreter interpreter(&ctx);
   std::ifstream file(cypherl_file_path);
-  if (file.is_open()) {
-    std::string line;
-    while (std::getline(file, line)) {
-      if (!line.empty()) {
-        auto results = interpreter.Prepare(line, {}, {});
-        memgraph::query::DiscardValueResultStream stream;
-        interpreter.Pull(&stream, {}, results.qid);
 
-#ifdef MG_ENTERPRISE
-        if (memgraph::license::global_license_checker.IsEnterpriseValidFast()) {
-          audit_log->Record("", "", line, {});
-        }
-#endif
+  if (!file.is_open()) {
+    spdlog::trace("Could not find init file {}", cypherl_file_path);
+    return;
+  }
+
+  std::string line;
+  while (std::getline(file, line)) {
+    if (!line.empty()) {
+      auto results = interpreter.Prepare(line, {}, {});
+      memgraph::query::DiscardValueResultStream stream;
+      interpreter.Pull(&stream, {}, results.qid);
+
+      if (audit_log) {
+        audit_log->Record("", "", line, {});
       }
     }
-    file.close();
   }
+
+  file.close();
 }
 
 namespace memgraph::metrics {
@@ -945,10 +941,12 @@ int main(int argc, char **argv) {
   interpreter_context.auth_checker = &auth_checker;
 
   if (!FLAGS_init_file.empty()) {
-    spdlog::info("Running init file.");
+    spdlog::info("Running init file...");
 #ifdef MG_ENTERPRISE
     if (memgraph::license::global_license_checker.IsEnterpriseValidFast()) {
       InitFromCypherlFile(interpreter_context, FLAGS_init_file, &audit_log);
+    } else {
+      InitFromCypherlFile(interpreter_context, FLAGS_init_file);
     }
 #else
     InitFromCypherlFile(interpreter_context, FLAGS_init_file);
@@ -1095,6 +1093,8 @@ int main(int argc, char **argv) {
 #ifdef MG_ENTERPRISE
     if (memgraph::license::global_license_checker.IsEnterpriseValidFast()) {
       InitFromCypherlFile(interpreter_context, FLAGS_init_data_file, &audit_log);
+    } else {
+      InitFromCypherlFile(interpreter_context, FLAGS_init_data_file);
     }
 #else
     InitFromCypherlFile(interpreter_context, FLAGS_init_data_file);

src/query/plan/operator.cpp

@@ -4648,7 +4648,7 @@ LoadCsv::LoadCsv(std::shared_ptr<LogicalOperator> input, Expression *file, bool
   MG_ASSERT(file_, "Something went wrong - '{}' member file_ shouldn't be a nullptr", __func__);
 }
 
-bool LoadCsv::Accept(HierarchicalLogicalOperatorVisitor &visitor) { return false; };
+ACCEPT_WITH_INPUT(LoadCsv)
 
 class LoadCsvCursor;
 
@@ -4699,14 +4699,12 @@ TypedValue CsvRowToTypedMap(csv::Reader::Row &row, csv::Reader::Header header) {
 class LoadCsvCursor : public Cursor {
   const LoadCsv *self_;
   const UniqueCursorPtr input_cursor_;
-  bool input_is_once_;
+  bool did_pull_;
   std::optional<csv::Reader> reader_{};
 
  public:
   LoadCsvCursor(const LoadCsv *self, utils::MemoryResource *mem)
-      : self_(self), input_cursor_(self_->input_->MakeCursor(mem)) {
-    input_is_once_ = dynamic_cast<Once *>(self_->input_.get());
-  }
+      : self_(self), input_cursor_(self_->input_->MakeCursor(mem)), did_pull_{false} {}
 
   bool Pull(Frame &frame, ExecutionContext &context) override {
     SCOPED_PROFILE_OP("LoadCsv");
@@ -4722,14 +4720,14 @@ class LoadCsvCursor : public Cursor {
       reader_ = MakeReader(&context.evaluation_context);
     }
 
-    bool input_pulled = input_cursor_->Pull(frame, context);
+    if (input_cursor_->Pull(frame, context)) {
+      if (did_pull_) {
+        throw QueryRuntimeException(
+            "LOAD CSV can be executed only once, please check if the cardinality of the operator before LOAD CSV is 1");
+      }
+      did_pull_ = true;
+    }
 
-    // If the input is Once, we have to keep going until we read all the rows,
-    // regardless of whether the pull on Once returned false.
-    // If we have e.g. MATCH(n) LOAD CSV ... AS x SET n.name = x.name, then we
-    // have to read at most cardinality(n) rows (but we can read less and stop
-    // pulling MATCH).
-    if (!input_is_once_ && !input_pulled) return false;
     auto row = reader_->GetNextRow(context.evaluation_context.memory);
     if (!row) {
       return false;

src/query/plan/pretty_print.cpp

@@ -874,11 +874,27 @@ bool PlanToJsonVisitor::PreVisit(query::plan::CallProcedure &op) {
 bool PlanToJsonVisitor::PreVisit(query::plan::LoadCsv &op) {
   json self;
   self["name"] = "LoadCsv";
-  self["file"] = ToJson(op.file_);
-  self["with_header"] = op.with_header_;
-  self["ignore_bad"] = op.ignore_bad_;
-  self["delimiter"] = ToJson(op.delimiter_);
-  self["quote"] = ToJson(op.quote_);
+
+  if (op.file_) {
+    self["file"] = ToJson(op.file_);
+  }
+
+  if (op.with_header_) {
+    self["with_header"] = op.with_header_;
+  }
+
+  if (op.ignore_bad_) {
+    self["ignore_bad"] = op.ignore_bad_;
+  }
+
+  if (op.delimiter_) {
+    self["delimiter"] = ToJson(op.delimiter_);
+  }
+
+  if (op.quote_) {
+    self["quote"] = ToJson(op.quote_);
+  }
+
   self["row_variable"] = ToJson(op.row_var_);
 
   op.input_->Accept(*this);

src/query/plan/rewrite/index_lookup.hpp

@@ -504,6 +504,16 @@ class IndexLookupRewriter final : public HierarchicalLogicalOperatorVisitor {
     return true;
   }
 
+  bool PreVisit(LoadCsv &op) override {
+    prev_ops_.push_back(&op);
+    return true;
+  }
+
+  bool PostVisit(LoadCsv & /*op*/) override {
+    prev_ops_.pop_back();
+    return true;
+  }
+
   std::shared_ptr<LogicalOperator> new_root_;
 
  private:

src/utils/csv_parsing.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

src/utils/csv_parsing.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/e2e/CMakeLists.txt

@@ -29,6 +29,14 @@ function(copy_e2e_cpp_files TARGET_PREFIX FILE_NAME)
                 DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/${FILE_NAME})
 endfunction()
 
+function(copy_e2e_files TARGET_PREFIX FILE_NAME)
+        add_custom_target(memgraph__e2e__${TARGET_PREFIX}__${FILE_NAME} ALL
+                COMMAND ${CMAKE_COMMAND} -E copy
+                ${CMAKE_CURRENT_SOURCE_DIR}/${FILE_NAME}
+                ${CMAKE_CURRENT_BINARY_DIR}/${FILE_NAME}
+                DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/${FILE_NAME})
+endfunction()
+
 add_subdirectory(fine_grained_access)
 add_subdirectory(server)
 add_subdirectory(replication)
@@ -47,6 +55,8 @@ add_subdirectory(python_query_modules_reloading)
 add_subdirectory(analyze_graph)
 add_subdirectory(transaction_queue)
 add_subdirectory(mock_api)
+add_subdirectory(load_csv)
+add_subdirectory(init_file_flags)
 
 copy_e2e_python_files(pytest_runner pytest_runner.sh "")
 file(COPY ${CMAKE_CURRENT_SOURCE_DIR}/memgraph-selfsigned.crt DESTINATION ${CMAKE_CURRENT_BINARY_DIR})

tests/e2e/init_file_flags/CMakeLists.txt

@@ -0,0 +1,12 @@
+function(copy_init_file_flags_e2e_python_files FILE_NAME)
+copy_e2e_files(init_file_flags ${FILE_NAME})
+endfunction()
+
+function(copy_init_file_flags_e2e_files FILE_NAME)
+    copy_e2e_files(init_file_flags ${FILE_NAME})
+endfunction()
+
+copy_init_file_flags_e2e_python_files(init_file_setup.py)
+copy_init_file_flags_e2e_python_files(init_data_file_setup.py)
+
+copy_init_file_flags_e2e_files(init_file.cypherl)

tests/e2e/init_file_flags/init_data_file_setup.py

@@ -0,0 +1,27 @@
+# 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.
+
+import sys
+
+import pytest
+from gqlalchemy import Memgraph
+
+
+def test_given_init_data_file_when_memgraph_started_then_node_is_created():
+    mg = Memgraph("localhost", 7687)
+
+    result = next(mg.execute_and_fetch("MATCH (n) RETURN count(n) AS cnt"))["cnt"]
+
+    assert result == 1
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/init_file_flags/init_file.cypherl

@@ -0,0 +1 @@
+CREATE (n);

tests/e2e/init_file_flags/init_file_setup.py

@@ -0,0 +1,27 @@
+# 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.
+
+import sys
+
+import pytest
+from gqlalchemy import Memgraph
+
+
+def test_given_init_file_when_memgraph_started_then_node_is_created():
+    mg = Memgraph("localhost", 7687)
+
+    result = next(mg.execute_and_fetch("MATCH (n) RETURN count(n) AS cnt"))["cnt"]
+
+    assert result == 1
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/init_file_flags/workloads.yaml

@@ -0,0 +1,32 @@
+init_file_cluster: &init_file_cluster
+  cluster:
+    main:
+      args: [
+        "--bolt-port", "7687",
+        "--log-level=TRACE",
+        "--init-file=init_file_flags/init_file.cypherl"
+      ]
+      log_file: "init-file-flags-e2e.log"
+      validation_queries: []
+
+init_data_file_cluster: &init_data_file_cluster
+  cluster:
+    main:
+      args: [
+        "--bolt-port", "7687",
+        "--log-level=TRACE",
+        "--init-data-file=init_file_flags/init_file.cypherl"
+      ]
+      log_file: "init-data-file-flags-e2e.log"
+      validation_queries: []
+
+workloads:
+  - name: "Init file flags"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["init_file_flags/init_file_setup.py"]
+    <<: *init_file_cluster
+
+  - name: "Init data file flags"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["init_file_flags/init_data_file_setup.py"]
+    <<: *init_data_file_cluster

tests/e2e/load_csv/CMakeLists.txt

@@ -0,0 +1,10 @@
+function(copy_load_csv_e2e_python_files FILE_NAME)
+    copy_e2e_python_files(load_csv ${FILE_NAME})
+endfunction()
+
+function(copy_load_csv_e2e_files FILE_NAME)
+    copy_e2e_python_files(load_csv ${FILE_NAME})
+endfunction()
+
+copy_load_csv_e2e_python_files(load_csv.py)
+copy_load_csv_e2e_files(simple.csv)

tests/e2e/load_csv/load_csv.py

@@ -0,0 +1,56 @@
+# 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 os
+import sys
+from pathlib import Path
+
+import pytest
+from gqlalchemy import Memgraph
+from mgclient import DatabaseError
+
+SIMPLE_CSV_FILE = "simple.csv"
+
+
+def get_file_path(file: str) -> str:
+    return os.path.join(Path(__file__).parent.absolute(), file)
+
+
+def test_given_two_rows_in_db_when_load_csv_after_match_then_throw_exception():
+    memgraph = Memgraph("localhost", 7687)
+
+    with pytest.raises(DatabaseError):
+        next(
+            memgraph.execute_and_fetch(
+                f"""MATCH (n) LOAD CSV
+            FROM '{get_file_path(SIMPLE_CSV_FILE)}' WITH HEADER AS row
+            CREATE (:Person {{name: row.name}})
+            """
+            )
+        )
+
+
+def test_given_one_row_in_db_when_load_csv_after_match_then_pass():
+    memgraph = Memgraph("localhost", 7687)
+
+    results = memgraph.execute_and_fetch(
+        f"""MATCH (n {{prop: 1}}) LOAD CSV
+        FROM '{get_file_path(SIMPLE_CSV_FILE)}' WITH HEADER AS row
+        CREATE (:Person {{name: row.name}})
+        RETURN n
+        """
+    )
+
+    assert len(list(results)) == 4
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/load_csv/simple.csv

@@ -0,0 +1,5 @@
+id,name
+1,Joseph
+2,Peter
+3,Ella
+4,Joe

tests/e2e/load_csv/workloads.yaml

@@ -0,0 +1,15 @@
+load_csv_cluster: &load_csv_cluster
+  cluster:
+    main:
+      args: ["--bolt-port", "7687", "--log-level=TRACE"]
+      log_file: "load_csv_log_file.txt"
+      setup_queries:
+        - "CREATE (n {prop: 1});"
+        - "CREATE (n {prop: 2});"
+      validation_queries: []
+
+workloads:
+  - name: "MATCH + LOAD CSV"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["load_csv/load_csv.py"]
+    <<: *load_csv_cluster

tests/setup.sh

@@ -1,6 +1,7 @@
 #!/bin/bash
 
 # shellcheck disable=1091
+
 set -Eeuo pipefail
 
 DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"

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