Add rocksdb 8.1.1

Add subqueries

Co-authored-by: Bruno Sačarić <bruno.sacaric@gmail.com>

CMakeLists.txt

@@ -279,7 +279,7 @@ if (MG_ENTERPRISE)
   add_definitions(-DMG_ENTERPRISE)
 endif()
 
-set(ENABLE_JEMALLOC OFF)
+set(ENABLE_JEMALLOC ON)
 
 if (ASAN)
   message(WARNING "Disabling jemalloc as it doesn't work well with ASAN")

libs/.gitignore

@@ -6,3 +6,4 @@
 !__main.cpp
 !pulsar.patch
 !antlr4.10.1.patch
+!rocksdb8.1.1.patch

libs/rocksdb8.1.1.patch

@@ -0,0 +1,13 @@
+diff --git a/CMakeLists.txt b/CMakeLists.txt
+index 598c728..816c705 100644
+--- a/CMakeLists.txt
++++ b/CMakeLists.txt
+@@ -1242,7 +1242,7 @@ if(NOT WIN32 OR ROCKSDB_INSTALL_ON_WINDOWS)
+
+   if(ROCKSDB_BUILD_SHARED)
+     install(
+-      TARGETS ${ROCKSDB_SHARED_LIB}
++      TARGETS ${ROCKSDB_SHARED_LIB} OPTIONAL
+       EXPORT RocksDBTargets
+       COMPONENT runtime
+       ARCHIVE DESTINATION "${CMAKE_INSTALL_LIBDIR}"

libs/setup.sh

@@ -192,10 +192,10 @@ cd json
 file_get_try_double "${primary_urls[nlohmann]}" "${secondary_urls[nlohmann]}"
 cd ..
 
-rocksdb_tag="v6.14.6" # (2020-10-14)
+rocksdb_tag="v8.1.1" # (2023-04-21)
 repo_clone_try_double "${primary_urls[rocksdb]}" "${secondary_urls[rocksdb]}" "rocksdb" "$rocksdb_tag" true
 pushd rocksdb
-git apply ../rocksdb.patch
+git apply ../rocksdb8.1.1.patch
 popd
 
 # mgclient

src/query/interpreter.cpp

@@ -1256,7 +1256,7 @@ PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string,
     contains_csv = true;
   }
   // If this is LOAD CSV query, use PoolResource without MonotonicMemoryResource as we want to reuse allocated memory
-  auto use_monotonic_memory = false;
+  auto use_monotonic_memory = !contains_csv;
   auto plan = CypherQueryToPlan(parsed_query.stripped_query.hash(), std::move(parsed_query.ast_storage), cypher_query,
                                 parsed_query.parameters,
                                 parsed_query.is_cacheable ? &interpreter_context->plan_cache : nullptr, dba);
@@ -1385,7 +1385,7 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
     contains_csv = true;
   }
   // If this is LOAD CSV query, use PoolResource without MonotonicMemoryResource as we want to reuse allocated memory
-  auto use_monotonic_memory = false;
+  auto use_monotonic_memory = !contains_csv;
 
   MG_ASSERT(cypher_query, "Cypher grammar should not allow other queries in PROFILE");
   Frame frame(0);
@@ -2743,12 +2743,12 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
       }
 
       if (const auto &clauses = cypher_query->single_query_->clauses_;
-          true || std::any_of(clauses.begin(), clauses.end(),
-                              [](const auto *clause) { return clause->GetTypeInfo() == LoadCsv::kType; })) {
+          std::any_of(clauses.begin(), clauses.end(),
+                      [](const auto *clause) { return clause->GetTypeInfo() == LoadCsv::kType; })) {
         // Using PoolResource without MonotonicMemoryResouce for LOAD CSV reduces memory usage.
         // QueryExecution MemoryResource is mostly used for allocations done on Frame and storing `row`s
-        query_executions_[query_executions_.size() - 1] = std::make_unique<QueryExecution>(utils::PoolResource(
-            128, kExecutionPoolMaxBlockSize, utils::NewDeleteResource(), utils::NewDeleteResource()));
+        query_executions_[query_executions_.size() - 1] = std::make_unique<QueryExecution>(
+            utils::PoolResource(8, kExecutionPoolMaxBlockSize, utils::NewDeleteResource(), utils::NewDeleteResource()));
         query_execution_ptr = &query_executions_.back();
       }
     }

src/query/plan/operator.cpp

@@ -53,7 +53,6 @@
 #include "utils/likely.hpp"
 #include "utils/logging.hpp"
 #include "utils/memory.hpp"
-#include "utils/on_scope_exit.hpp"
 #include "utils/pmr/deque.hpp"
 #include "utils/pmr/list.hpp"
 #include "utils/pmr/unordered_map.hpp"
@@ -4529,14 +4528,7 @@ class CallProcedureCursor : public Cursor {
       // evaluation of an expression.
       // TODO: This will probably need to be changed when we add support for
       // generator like procedures which yield a new result on each invocation.
-
-      // 1. problem je sto se koristi ova memorija
-      // 2. problem je sto je to memorija od EvalContexta
-      // 3. problem je sto se koristi i memorija od QueryExecutiona
-
-      // utils::PoolResource poolResource{128, 1024, utils::NewDeleteResource(), utils::NewDeleteResource()};
       auto *memory = context.evaluation_context.memory;
-      // utils::OnScopeExit scope_exit([&poolResource] { poolResource.Release(); });
       auto memory_limit = EvaluateMemoryLimit(&evaluator, self_->memory_limit_, self_->memory_scale_);
       auto graph = mgp_graph::WritableGraph(*context.db_accessor, graph_view, context);
       CallCustomProcedure(self_->procedure_name_, *proc, self_->arguments_, graph, &evaluator, memory, memory_limit,

src/query/procedure/mg_procedure_impl.cpp

@@ -1504,7 +1504,6 @@ mgp_error mgp_result_new_record(mgp_result *res, mgp_result_record **result) {
 
 mgp_error mgp_result_record_insert(mgp_result_record *record, const char *field_name, mgp_value *val) {
   return WrapExceptions([=] {
-    // MEMORY RESOURCE: QueryExecution memory resource
     auto *memory = record->values.get_allocator().GetMemoryResource();
     // Validate field_name & val satisfy the procedure's result signature.
     MG_ASSERT(record->signature, "Expected to have a valid signature");

src/query/procedure/py_module.cpp

@@ -14,7 +14,6 @@
 #include <datetime.h>
 #include <pyerrors.h>
 #include <array>
-#include <optional>
 #include <sstream>
 #include <stdexcept>
 #include <string>
@@ -861,7 +860,7 @@ py::Object MgpListToPyTuple(mgp_list *list, PyObject *py_graph) {
 }
 
 namespace {
-std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Object py_record, mgp_memory *memory) {
+std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Object py_record) {
   py::Object py_mgp(PyImport_ImportModule("mgp"));
   if (!py_mgp) return py::FetchError();
   auto record_cls = py_mgp.GetAttr("Record");
@@ -903,9 +902,8 @@ std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Obj
     if (!field_name) return py::FetchError();
     auto *val = PyTuple_GetItem(item, 1);
     if (!val) return py::FetchError();
-    // ova memorija je od query executiona
-    // mgp_memory memory{result->rows.get_allocator().GetMemoryResource()};
-    mgp_value *field_val = PyObjectToMgpValueWithPythonExceptions(val, memory);
+    mgp_memory memory{result->rows.get_allocator().GetMemoryResource()};
+    mgp_value *field_val = PyObjectToMgpValueWithPythonExceptions(val, &memory);
     if (field_val == nullptr) {
       return py::FetchError();
     }
@@ -923,20 +921,14 @@ std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Obj
   return std::nullopt;
 }
 
-std::optional<py::ExceptionInfo> AddMultipleRecordsFromPython(mgp_result *result, py::Object py_seq,
-                                                              mgp_memory *memory) {
+std::optional<py::ExceptionInfo> AddMultipleRecordsFromPython(mgp_result *result, py::Object py_seq) {
   Py_ssize_t len = PySequence_Size(py_seq.Ptr());
   if (len == -1) return py::FetchError();
-
-  for (Py_ssize_t i = 0, curr_loc = 0; i < len; ++i, ++curr_loc) {
-    py::Object py_record(PySequence_GetItem(py_seq.Ptr(), curr_loc));
+  for (Py_ssize_t i = 0; i < len; ++i) {
+    py::Object py_record(PySequence_GetItem(py_seq.Ptr(), i));
     if (!py_record) return py::FetchError();
-    auto maybe_exc = AddRecordFromPython(result, py_record, memory);
+    auto maybe_exc = AddRecordFromPython(result, py_record);
     if (maybe_exc) return maybe_exc;
-    if (i && i % 100000 == 0) {
-      PySequence_DelSlice(py_seq.Ptr(), 0, 100000);
-      curr_loc = -1;
-    }
   }
   return std::nullopt;
 }
@@ -970,7 +962,6 @@ std::function<void()> PyObjectCleanup(py::Object &py_object) {
 
 void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *graph, mgp_result *result,
                          mgp_memory *memory) {
-  // ovaj *memory tu je memorija od EvalContexta
   auto gil = py::EnsureGIL();
 
   auto error_to_msg = [](const std::optional<py::ExceptionInfo> &exc_info) -> std::optional<std::string> {
@@ -988,9 +979,9 @@ void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *gra
     auto py_res = py_cb.Call(py_graph, py_args);
     if (!py_res) return py::FetchError();
     if (PySequence_Check(py_res.Ptr())) {
-      return AddMultipleRecordsFromPython(result, py_res, memory);
+      return AddMultipleRecordsFromPython(result, py_res);
     } else {
-      return AddRecordFromPython(result, py_res, memory);
+      return AddRecordFromPython(result, py_res);
     }
   };
 
@@ -1036,9 +1027,9 @@ void CallPythonTransformation(const py::Object &py_cb, mgp_messages *msgs, mgp_g
     auto py_res = py_cb.Call(py_graph, py_messages);
     if (!py_res) return py::FetchError();
     if (PySequence_Check(py_res.Ptr())) {
-      return AddMultipleRecordsFromPython(result, py_res, memory);
+      return AddMultipleRecordsFromPython(result, py_res);
     }
-    return AddRecordFromPython(result, py_res, memory);
+    return AddRecordFromPython(result, py_res);
   };
 
   // It is *VERY IMPORTANT* to note that this code takes great care not to keep
@@ -1140,7 +1131,6 @@ PyObject *PyQueryModuleAddProcedure(PyQueryModule *self, PyObject *cb, bool is_w
     PyErr_SetString(PyExc_ValueError, "Procedure name is not a valid identifier");
     return nullptr;
   }
-  //
   auto *memory = self->module->procedures.get_allocator().GetMemoryResource();
   mgp_proc proc(name,
                 [py_cb](mgp_list *args, mgp_graph *graph, mgp_result *result, mgp_memory *memory) {

src/storage/README.md

@@ -0,0 +1,3 @@
+# Storage Modes
+
+* `ON_DISK_TRANSACTIONAL`

tests/mgbench/.gitignore

@@ -4,3 +4,4 @@
 *.report
 *.sysinfo
 *.json
+*.html

tests/mgbench/benchmark.py

@@ -643,7 +643,7 @@ if __name__ == "__main__":
                         ret = client.execute(
                             queries=get_queries(func, count),
                             num_workers=benchmark_context.num_workers_for_benchmark,
-                            time_dependent_execution=benchmark_context.time_depended_execution,
+                            time_dependent_execution=benchmark_context.time_dependent_execution,
                         )[0]
                     else:
                         ret = client.execute(

tests/setup.sh

@@ -34,14 +34,7 @@ set -u
 PYTHON_MINOR=$(python3 -c 'import sys; print(sys.version_info[:][1])')
 
 # install pulsar-client
-# NOTE (2021-11-15): PyPi doesn't contain pulsar-client for Python 3.9 so we have to use
-# our manually built wheel file. When they update the repository, pulsar-client can be
-# added as a regular PIP dependancy
-if [ $PYTHON_MINOR -lt 9 ]; then
-  pip --timeout 1000 install "pulsar-client==2.8.1"
-else
-  pip --timeout 1000 install https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/pulsar_client-2.8.1-cp39-cp39-manylinux_2_5_x86_64.manylinux1_x86_64.whl
-fi
+pip --timeout 1000 install "pulsar-client==3.1.0"
 
 for pkg in "${PIP_DEPS[@]}"; do
     pip --timeout 1000 install "$pkg"