add slice delete of python records

CMakeLists.txt

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

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 = !contains_csv;
+  auto use_monotonic_memory = false;
   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 = !contains_csv;
+  auto use_monotonic_memory = false;
 
   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_;
-          std::any_of(clauses.begin(), clauses.end(),
-                      [](const auto *clause) { return clause->GetTypeInfo() == LoadCsv::kType; })) {
+          true || 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(8, kExecutionPoolMaxBlockSize, utils::NewDeleteResource(), utils::NewDeleteResource()));
+        query_executions_[query_executions_.size() - 1] = std::make_unique<QueryExecution>(utils::PoolResource(
+            128, kExecutionPoolMaxBlockSize, utils::NewDeleteResource(), utils::NewDeleteResource()));
         query_execution_ptr = &query_executions_.back();
       }
     }

src/query/plan/operator.cpp

@@ -53,6 +53,7 @@
 #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"
@@ -4528,7 +4529,14 @@ 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,6 +1504,7 @@ 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,6 +14,7 @@
 #include <datetime.h>
 #include <pyerrors.h>
 #include <array>
+#include <optional>
 #include <sstream>
 #include <stdexcept>
 #include <string>
@@ -860,7 +861,7 @@ py::Object MgpListToPyTuple(mgp_list *list, PyObject *py_graph) {
 }
 
 namespace {
-std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Object py_record) {
+std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Object py_record, mgp_memory *memory) {
   py::Object py_mgp(PyImport_ImportModule("mgp"));
   if (!py_mgp) return py::FetchError();
   auto record_cls = py_mgp.GetAttr("Record");
@@ -902,8 +903,9 @@ 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();
-    mgp_memory memory{result->rows.get_allocator().GetMemoryResource()};
-    mgp_value *field_val = PyObjectToMgpValueWithPythonExceptions(val, &memory);
+    // ova memorija je od query executiona
+    // mgp_memory memory{result->rows.get_allocator().GetMemoryResource()};
+    mgp_value *field_val = PyObjectToMgpValueWithPythonExceptions(val, memory);
     if (field_val == nullptr) {
       return py::FetchError();
     }
@@ -921,14 +923,20 @@ 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) {
+std::optional<py::ExceptionInfo> AddMultipleRecordsFromPython(mgp_result *result, py::Object py_seq,
+                                                              mgp_memory *memory) {
   Py_ssize_t len = PySequence_Size(py_seq.Ptr());
   if (len == -1) return py::FetchError();
-  for (Py_ssize_t i = 0; i < len; ++i) {
-    py::Object py_record(PySequence_GetItem(py_seq.Ptr(), i));
+
+  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));
     if (!py_record) return py::FetchError();
-    auto maybe_exc = AddRecordFromPython(result, py_record);
+    auto maybe_exc = AddRecordFromPython(result, py_record, memory);
     if (maybe_exc) return maybe_exc;
+    if (i && i % 100000 == 0) {
+      PySequence_DelSlice(py_seq.Ptr(), 0, 100000);
+      curr_loc = -1;
+    }
   }
   return std::nullopt;
 }
@@ -962,6 +970,7 @@ 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> {
@@ -979,9 +988,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);
+      return AddMultipleRecordsFromPython(result, py_res, memory);
     } else {
-      return AddRecordFromPython(result, py_res);
+      return AddRecordFromPython(result, py_res, memory);
     }
   };
 
@@ -1027,9 +1036,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);
+      return AddMultipleRecordsFromPython(result, py_res, memory);
     }
-    return AddRecordFromPython(result, py_res);
+    return AddRecordFromPython(result, py_res, memory);
   };
 
   // It is *VERY IMPORTANT* to note that this code takes great care not to keep
@@ -1131,6 +1140,7 @@ 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) {