tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use one common allocator for queries

Browse Source
This commit is contained in:
Gareth Lloyd 2024-03-05 20:06:47 +00:00
parent be66f03cc8
commit 37f11a75d4
8 changed files with 138 additions and 218 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
CMakeLists.txt
View File

@ -300,6 +300,19 @@ endif()
option(ENABLE_JEMALLOC "Use jemalloc" ON) option(ENABLE_JEMALLOC "Use jemalloc" ON)
option(MG_MEMORY_PROFILE "If build should be setup for memory profiling" OFF)
if (MG_MEMORY_PROFILE AND ENABLE_JEMALLOC)
message(STATUS "Jemalloc has been disabled because MG_MEMORY_PROFILE is enabled")
set(ENABLE_JEMALLOC OFF)
endif ()
if (MG_MEMORY_PROFILE AND ASAN)
message(STATUS "ASAN has been disabled because MG_MEMORY_PROFILE is enabled")
set(ASAN OFF)
endif ()
if (MG_MEMORY_PROFILE)
add_compile_definitions(MG_MEMORY_PROFILE)
endif ()
if (ASAN) if (ASAN)
message(WARNING "Disabling jemalloc as it doesn't work well with ASAN") message(WARNING "Disabling jemalloc as it doesn't work well with ASAN")
set(ENABLE_JEMALLOC OFF) set(ENABLE_JEMALLOC OFF)

182
src/query/interpreter.cpp
View File

@ -246,27 +246,6 @@ std::optional<std::string> GetOptionalStringValue(query::Expression *expression,
return {}; return {};
}; };
bool IsAllShortestPathsQuery(const std::vector<memgraph::query::Clause *> &clauses) {
for (const auto &clause : clauses) {
if (clause->GetTypeInfo() != Match::kType) {
continue;
}
auto *match_clause = utils::Downcast<Match>(clause);
for (const auto &pattern : match_clause->patterns_) {
for (const auto &atom : pattern->atoms_) {
if (atom->GetTypeInfo() != EdgeAtom::kType) {
continue;
}
auto *edge_atom = utils::Downcast<EdgeAtom>(atom);
if (edge_atom->type_ == EdgeAtom::Type::ALL_SHORTEST_PATHS) {
return true;
}
}
}
}
return false;
}
inline auto convertFromCoordinatorToReplicationMode(const CoordinatorQuery::SyncMode &sync_mode) inline auto convertFromCoordinatorToReplicationMode(const CoordinatorQuery::SyncMode &sync_mode)
-> replication_coordination_glue::ReplicationMode { -> replication_coordination_glue::ReplicationMode {
switch (sync_mode) { switch (sync_mode) {
@ -1670,8 +1649,7 @@ struct PullPlan {
std::shared_ptr<QueryUserOrRole> user_or_role, std::atomic<TransactionStatus> *transaction_status, std::shared_ptr<QueryUserOrRole> user_or_role, std::atomic<TransactionStatus> *transaction_status,
std::shared_ptr<utils::AsyncTimer> tx_timer, std::shared_ptr<utils::AsyncTimer> tx_timer,
TriggerContextCollector *trigger_context_collector = nullptr, TriggerContextCollector *trigger_context_collector = nullptr,
std::optional<size_t> memory_limit = {}, bool use_monotonic_memory = true, std::optional<size_t> memory_limit = {}, FrameChangeCollector *frame_change_collector_ = nullptr);
FrameChangeCollector *frame_change_collector_ = nullptr);
std::optional<plan::ProfilingStatsWithTotalTime> Pull(AnyStream *stream, std::optional<int> n, std::optional<plan::ProfilingStatsWithTotalTime> Pull(AnyStream *stream, std::optional<int> n,
const std::vector<Symbol> &output_symbols, const std::vector<Symbol> &output_symbols,
@ -1696,26 +1674,17 @@ struct PullPlan {
// we have to keep track of any unsent results from previous `PullPlan::Pull` // we have to keep track of any unsent results from previous `PullPlan::Pull`
// manually by using this flag. // manually by using this flag.
bool has_unsent_results_ = false; bool has_unsent_results_ = false;
// In the case of LOAD CSV, we want to use only PoolResource without MonotonicMemoryResource
// to reuse allocated memory. As LOAD CSV is processing row by row
// it is possible to reduce memory usage significantly if MemoryResource deals with memory allocation
// can reuse memory that was allocated on processing the first row on all subsequent rows.
// This flag signals to `PullPlan::Pull` which MemoryResource to use
bool use_monotonic_memory_;
}; };
PullPlan::PullPlan(const std::shared_ptr<PlanWrapper> plan, const Parameters &parameters, const bool is_profile_query, PullPlan::PullPlan(const std::shared_ptr<PlanWrapper> plan, const Parameters &parameters, const bool is_profile_query,
DbAccessor *dba, InterpreterContext *interpreter_context, utils::MemoryResource *execution_memory, DbAccessor *dba, InterpreterContext *interpreter_context, utils::MemoryResource *execution_memory,
std::shared_ptr<QueryUserOrRole> user_or_role, std::atomic<TransactionStatus> *transaction_status, std::shared_ptr<QueryUserOrRole> user_or_role, std::atomic<TransactionStatus> *transaction_status,
std::shared_ptr<utils::AsyncTimer> tx_timer, TriggerContextCollector *trigger_context_collector, std::shared_ptr<utils::AsyncTimer> tx_timer, TriggerContextCollector *trigger_context_collector,
const std::optional<size_t> memory_limit, bool use_monotonic_memory, const std::optional<size_t> memory_limit, FrameChangeCollector *frame_change_collector)
FrameChangeCollector *frame_change_collector)
: plan_(plan), : plan_(plan),
cursor_(plan->plan().MakeCursor(execution_memory)), cursor_(plan->plan().MakeCursor(execution_memory)),
frame_(plan->symbol_table().max_position(), execution_memory), frame_(plan->symbol_table().max_position(), execution_memory),
memory_limit_(memory_limit), memory_limit_(memory_limit) {
use_monotonic_memory_(use_monotonic_memory) {
ctx_.db_accessor = dba; ctx_.db_accessor = dba;
ctx_.symbol_table = plan->symbol_table(); ctx_.symbol_table = plan->symbol_table();
ctx_.evaluation_context.timestamp = QueryTimestamp(); ctx_.evaluation_context.timestamp = QueryTimestamp();
@ -1741,6 +1710,7 @@ PullPlan::PullPlan(const std::shared_ptr<PlanWrapper> plan, const Parameters &pa
ctx_.is_profile_query = is_profile_query; ctx_.is_profile_query = is_profile_query;
ctx_.trigger_context_collector = trigger_context_collector; ctx_.trigger_context_collector = trigger_context_collector;
ctx_.frame_change_collector = frame_change_collector; ctx_.frame_change_collector = frame_change_collector;
ctx_.evaluation_context.memory = execution_memory;
} }
std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *stream, std::optional<int> n, std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *stream, std::optional<int> n,
@ -1764,31 +1734,6 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *strea
} }
}}; }};
// 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`.
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};
std::optional<utils::PoolResource> pool_memory;
static constexpr auto kMaxBlockPerChunks = 128;
if (!use_monotonic_memory_) {
pool_memory.emplace(kMaxBlockPerChunks, kExecutionPoolMaxBlockSize, &resource_with_exception,
&resource_with_exception);
} else {
// 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.
pool_memory.emplace(kMaxBlockPerChunks, 1024, &monotonic_memory, &resource_with_exception);
}
ctx_.evaluation_context.memory = &*pool_memory;
// Returns true if a result was pulled. // Returns true if a result was pulled.
const auto pull_result = [&]() -> bool { return cursor_->Pull(frame_, ctx_); }; const auto pull_result = [&]() -> bool { return cursor_->Pull(frame_, ctx_); };
@ -1910,7 +1855,7 @@ PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper,
std::function<void()> handler; std::function<void()> handler;
if (query_upper == "BEGIN") { if (query_upper == "BEGIN") {
ResetInterpreter(); // ResetInterpreter();
// TODO: Evaluate doing move(extras). Currently the extras is very small, but this will be important if it ever // TODO: Evaluate doing move(extras). Currently the extras is very small, but this will be important if it ever
// becomes large. // becomes large.
handler = [this, extras = extras] { handler = [this, extras = extras] {
@ -1988,30 +1933,6 @@ inline static void TryCaching(const AstStorage &ast_storage, FrameChangeCollecto
} }
} }
bool IsLoadCsvQuery(const std::vector<memgraph::query::Clause *> &clauses) {
return std::any_of(clauses.begin(), clauses.end(),
[](memgraph::query::Clause const *clause) { return clause->GetTypeInfo() == LoadCsv::kType; });
}
bool IsCallBatchedProcedureQuery(const std::vector<memgraph::query::Clause *> &clauses) {
EvaluationContext evaluation_context;
return std::ranges::any_of(clauses, [&evaluation_context](memgraph::query::Clause *clause) -> bool {
if (!(clause->GetTypeInfo() == CallProcedure::kType)) return false;
auto *call_procedure_clause = utils::Downcast<CallProcedure>(clause);
const auto &maybe_found = memgraph::query::procedure::FindProcedure(
procedure::gModuleRegistry, call_procedure_clause->procedure_name_, evaluation_context.memory);
if (!maybe_found) {
throw QueryRuntimeException("There is no procedure named '{}'.", call_procedure_clause->procedure_name_);
}
const auto &[module, proc] = *maybe_found;
if (!proc->info.is_batched) return false;
spdlog::trace("Using PoolResource for batched query procedure");
return true;
});
}
PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string, TypedValue> *summary, PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string, TypedValue> *summary,
InterpreterContext *interpreter_context, CurrentDB &current_db, InterpreterContext *interpreter_context, CurrentDB &current_db,
utils::MemoryResource *execution_memory, std::vector<Notification> *notifications, utils::MemoryResource *execution_memory, std::vector<Notification> *notifications,
@ -2031,7 +1952,6 @@ PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string,
spdlog::info("Running query with memory limit of {}", utils::GetReadableSize(*memory_limit)); spdlog::info("Running query with memory limit of {}", utils::GetReadableSize(*memory_limit));
} }
auto clauses = cypher_query->single_query_->clauses_; auto clauses = cypher_query->single_query_->clauses_;
bool contains_csv = false;
if (std::any_of(clauses.begin(), clauses.end(), if (std::any_of(clauses.begin(), clauses.end(),
[](const auto *clause) { return clause->GetTypeInfo() == LoadCsv::kType; })) { [](const auto *clause) { return clause->GetTypeInfo() == LoadCsv::kType; })) {
notifications->emplace_back( notifications->emplace_back(
@ -2039,13 +1959,8 @@ PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string,
"It's important to note that the parser parses the values as strings. It's up to the user to " "It's important to note that the parser parses the values as strings. It's up to the user to "
"convert the parsed row values to the appropriate type. This can be done using the built-in " "convert the parsed row values to the appropriate type. This can be done using the built-in "
"conversion functions such as ToInteger, ToFloat, ToBoolean etc."); "conversion functions such as ToInteger, ToFloat, ToBoolean etc.");
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 && !IsCallBatchedProcedureQuery(clauses) && !IsAllShortestPathsQuery(clauses);
MG_ASSERT(current_db.execution_db_accessor_, "Cypher query expects a current DB transaction"); MG_ASSERT(current_db.execution_db_accessor_, "Cypher query expects a current DB transaction");
auto *dba = auto *dba =
&*current_db &*current_db
@ -2084,7 +1999,7 @@ PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string,
current_db.trigger_context_collector_ ? &*current_db.trigger_context_collector_ : nullptr; current_db.trigger_context_collector_ ? &*current_db.trigger_context_collector_ : nullptr;
auto pull_plan = std::make_shared<PullPlan>( auto pull_plan = std::make_shared<PullPlan>(
plan, parsed_query.parameters, false, dba, interpreter_context, execution_memory, std::move(user_or_role), plan, parsed_query.parameters, false, dba, interpreter_context, execution_memory, std::move(user_or_role),
transaction_status, std::move(tx_timer), trigger_context_collector, memory_limit, use_monotonic_memory, transaction_status, std::move(tx_timer), trigger_context_collector, memory_limit,
frame_change_collector->IsTrackingValues() ? frame_change_collector : nullptr); frame_change_collector->IsTrackingValues() ? frame_change_collector : nullptr);
return PreparedQuery{std::move(header), std::move(parsed_query.required_privileges), return PreparedQuery{std::move(header), std::move(parsed_query.required_privileges),
[pull_plan = std::move(pull_plan), output_symbols = std::move(output_symbols), summary]( [pull_plan = std::move(pull_plan), output_symbols = std::move(output_symbols), summary](
@ -2198,18 +2113,6 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
auto *cypher_query = utils::Downcast<CypherQuery>(parsed_inner_query.query); auto *cypher_query = utils::Downcast<CypherQuery>(parsed_inner_query.query);
bool contains_csv = false;
auto clauses = cypher_query->single_query_->clauses_;
if (std::any_of(clauses.begin(), clauses.end(),
[](const auto *clause) { return clause->GetTypeInfo() == LoadCsv::kType; })) {
contains_csv = true;
}
// If this is LOAD CSV, BatchedProcedure or AllShortest query, use PoolResource without MonotonicMemoryResource as we
// want to reuse allocated memory
auto use_monotonic_memory =
!contains_csv && !IsCallBatchedProcedureQuery(clauses) && !IsAllShortestPathsQuery(clauses);
MG_ASSERT(cypher_query, "Cypher grammar should not allow other queries in PROFILE"); MG_ASSERT(cypher_query, "Cypher grammar should not allow other queries in PROFILE");
EvaluationContext evaluation_context; EvaluationContext evaluation_context;
evaluation_context.timestamp = QueryTimestamp(); evaluation_context.timestamp = QueryTimestamp();
@ -2243,14 +2146,14 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
// We want to execute the query we are profiling lazily, so we delay // We want to execute the query we are profiling lazily, so we delay
// the construction of the corresponding context. // the construction of the corresponding context.
stats_and_total_time = std::optional<plan::ProfilingStatsWithTotalTime>{}, stats_and_total_time = std::optional<plan::ProfilingStatsWithTotalTime>{},
pull_plan = std::shared_ptr<PullPlanVector>(nullptr), transaction_status, use_monotonic_memory, pull_plan = std::shared_ptr<PullPlanVector>(nullptr), transaction_status, frame_change_collector,
frame_change_collector, tx_timer = std::move(tx_timer)]( tx_timer = std::move(tx_timer)](AnyStream *stream,
AnyStream *stream, std::optional<int> n) mutable -> std::optional<QueryHandlerResult> { std::optional<int> n) mutable -> std::optional<QueryHandlerResult> {
// No output symbols are given so that nothing is streamed. // No output symbols are given so that nothing is streamed.
if (!stats_and_total_time) { if (!stats_and_total_time) {
stats_and_total_time = stats_and_total_time =
PullPlan(plan, parameters, true, dba, interpreter_context, execution_memory, std::move(user_or_role), PullPlan(plan, parameters, true, dba, interpreter_context, execution_memory, std::move(user_or_role),
transaction_status, std::move(tx_timer), nullptr, memory_limit, use_monotonic_memory, transaction_status, std::move(tx_timer), nullptr, memory_limit,
frame_change_collector->IsTrackingValues() ? frame_change_collector : nullptr) frame_change_collector->IsTrackingValues() ? frame_change_collector : nullptr)
.Pull(stream, {}, {}, summary); .Pull(stream, {}, {}, summary);
pull_plan = std::make_shared<PullPlanVector>(ProfilingStatsToTable(*stats_and_total_time)); pull_plan = std::make_shared<PullPlanVector>(ProfilingStatsToTable(*stats_and_total_time));
@ -4213,6 +4116,7 @@ PreparedQuery PrepareShowDatabasesQuery(ParsedQuery parsed_query, InterpreterCon
std::optional<uint64_t> Interpreter::GetTransactionId() const { return current_transaction_; } std::optional<uint64_t> Interpreter::GetTransactionId() const { return current_transaction_; }
void Interpreter::BeginTransaction(QueryExtras const &extras) { void Interpreter::BeginTransaction(QueryExtras const &extras) {
ResetInterpreter();
const auto prepared_query = PrepareTransactionQuery("BEGIN", extras); const auto prepared_query = PrepareTransactionQuery("BEGIN", extras);
prepared_query.query_handler(nullptr, {}); prepared_query.query_handler(nullptr, {});
} }
@ -4247,12 +4151,12 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
const auto upper_case_query = utils::ToUpperCase(query_string); const auto upper_case_query = utils::ToUpperCase(query_string);
const auto trimmed_query = utils::Trim(upper_case_query); const auto trimmed_query = utils::Trim(upper_case_query);
if (trimmed_query == "BEGIN" || trimmed_query == "COMMIT" || trimmed_query == "ROLLBACK") { if (trimmed_query == "BEGIN" || trimmed_query == "COMMIT" || trimmed_query == "ROLLBACK") {
auto resource = utils::MonotonicBufferResource(kExecutionMemoryBlockSize); if (trimmed_query == "BEGIN") {
auto prepared_query = PrepareTransactionQuery(trimmed_query, extras); ResetInterpreter();
auto &query_execution = }
query_executions_.emplace_back(QueryExecution::Create(std::move(resource), std::move(prepared_query))); auto &query_execution = query_executions_.emplace_back(QueryExecution::Create());
std::optional<int> qid = query_execution->prepared_query = PrepareTransactionQuery(trimmed_query, extras);
in_explicit_transaction_ ? static_cast<int>(query_executions_.size() - 1) : std::optional<int>{}; auto qid = in_explicit_transaction_ ? static_cast<int>(query_executions_.size() - 1) : std::optional<int>{};
return {query_execution->prepared_query->header, query_execution->prepared_query->privileges, qid, {}}; return {query_execution->prepared_query->header, query_execution->prepared_query->privileges, qid, {}};
} }
@ -4282,35 +4186,8 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
ParseQuery(query_string, params, &interpreter_context_->ast_cache, interpreter_context_->config.query); ParseQuery(query_string, params, &interpreter_context_->ast_cache, interpreter_context_->config.query);
auto parsing_time = parsing_timer.Elapsed().count(); auto parsing_time = parsing_timer.Elapsed().count();
CypherQuery const *const cypher_query = [&]() -> CypherQuery * {
if (auto *cypher_query = utils::Downcast<CypherQuery>(parsed_query.query)) {
return cypher_query;
}
if (auto *profile_query = utils::Downcast<ProfileQuery>(parsed_query.query)) {
return profile_query->cypher_query_;
}
return nullptr;
}(); // IILE
auto const [usePool, hasAllShortestPaths] = [&]() -> std::pair<bool, bool> {
if (!cypher_query) {
return {false, false};
}
auto const &clauses = cypher_query->single_query_->clauses_;
bool hasAllShortestPaths = IsAllShortestPathsQuery(clauses);
// Using PoolResource without MonotonicMemoryResouce for LOAD CSV reduces memory usage.
bool usePool = hasAllShortestPaths || IsCallBatchedProcedureQuery(clauses) || IsLoadCsvQuery(clauses);
return {usePool, hasAllShortestPaths};
}(); // IILE
// Setup QueryExecution // Setup QueryExecution
// its MemoryResource is mostly used for allocations done on Frame and storing `row`s query_executions_.emplace_back(QueryExecution::Create());
if (usePool) {
query_executions_.emplace_back(QueryExecution::Create(utils::PoolResource(128, kExecutionPoolMaxBlockSize)));
} else {
query_executions_.emplace_back(QueryExecution::Create(utils::MonotonicBufferResource(kExecutionMemoryBlockSize)));
}
auto &query_execution = query_executions_.back(); auto &query_execution = query_executions_.back();
query_execution_ptr = &query_execution; query_execution_ptr = &query_execution;
@ -4379,9 +4256,7 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
utils::Timer planning_timer; utils::Timer planning_timer;
PreparedQuery prepared_query; PreparedQuery prepared_query;
utils::MemoryResource *memory_resource = utils::MemoryResource *memory_resource = query_execution->execution_memory.resource();
std::visit([](auto &execution_memory) -> utils::MemoryResource * { return &execution_memory; },
query_execution->execution_memory);
frame_change_collector_.reset(); frame_change_collector_.reset();
frame_change_collector_.emplace(); frame_change_collector_.emplace();
if (utils::Downcast<CypherQuery>(parsed_query.query)) { if (utils::Downcast<CypherQuery>(parsed_query.query)) {
@ -4392,10 +4267,10 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
prepared_query = PrepareExplainQuery(std::move(parsed_query), &query_execution->summary, prepared_query = PrepareExplainQuery(std::move(parsed_query), &query_execution->summary,
&query_execution->notifications, interpreter_context_, current_db_); &query_execution->notifications, interpreter_context_, current_db_);
} else if (utils::Downcast<ProfileQuery>(parsed_query.query)) { } else if (utils::Downcast<ProfileQuery>(parsed_query.query)) {
prepared_query = PrepareProfileQuery(std::move(parsed_query), in_explicit_transaction_, &query_execution->summary, prepared_query =
&query_execution->notifications, interpreter_context_, current_db_, PrepareProfileQuery(std::move(parsed_query), in_explicit_transaction_, &query_execution->summary,
&query_execution->execution_memory_with_exception, user_or_role_, &query_execution->notifications, interpreter_context_, current_db_, memory_resource,
&transaction_status_, current_timeout_timer_, &*frame_change_collector_); user_or_role_, &transaction_status_, current_timeout_timer_, &*frame_change_collector_);
} else if (utils::Downcast<DumpQuery>(parsed_query.query)) { } else if (utils::Downcast<DumpQuery>(parsed_query.query)) {
prepared_query = PrepareDumpQuery(std::move(parsed_query), current_db_); prepared_query = PrepareDumpQuery(std::move(parsed_query), current_db_);
} else if (utils::Downcast<IndexQuery>(parsed_query.query)) { } else if (utils::Downcast<IndexQuery>(parsed_query.query)) {
@ -4597,7 +4472,7 @@ void RunTriggersAfterCommit(dbms::DatabaseAccess db_acc, InterpreterContext *int
std::atomic<TransactionStatus> *transaction_status) { std::atomic<TransactionStatus> *transaction_status) {
// Run the triggers // Run the triggers
for (const auto &trigger : db_acc->trigger_store()->AfterCommitTriggers().access()) { for (const auto &trigger : db_acc->trigger_store()->AfterCommitTriggers().access()) {
utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize}; QueryAllocator execution_memory{};
// create a new transaction for each trigger // create a new transaction for each trigger
auto tx_acc = db_acc->Access(); auto tx_acc = db_acc->Access();
@ -4608,7 +4483,7 @@ void RunTriggersAfterCommit(dbms::DatabaseAccess db_acc, InterpreterContext *int
auto trigger_context = original_trigger_context; auto trigger_context = original_trigger_context;
trigger_context.AdaptForAccessor(&db_accessor); trigger_context.AdaptForAccessor(&db_accessor);
try { try {
trigger.Execute(&db_accessor, &execution_memory, flags::run_time::GetExecutionTimeout(), trigger.Execute(&db_accessor, execution_memory.resource(), flags::run_time::GetExecutionTimeout(),
&interpreter_context->is_shutting_down, transaction_status, trigger_context); &interpreter_context->is_shutting_down, transaction_status, trigger_context);
} catch (const utils::BasicException &exception) { } catch (const utils::BasicException &exception) {
spdlog::warn("Trigger '{}' failed with exception:\n{}", trigger.Name(), exception.what()); spdlog::warn("Trigger '{}' failed with exception:\n{}", trigger.Name(), exception.what());
@ -4762,11 +4637,12 @@ void Interpreter::Commit() {
if (trigger_context) { if (trigger_context) {
// Run the triggers // Run the triggers
for (const auto &trigger : db->trigger_store()->BeforeCommitTriggers().access()) { for (const auto &trigger : db->trigger_store()->BeforeCommitTriggers().access()) {
utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize}; QueryAllocator execution_memory{};
AdvanceCommand(); AdvanceCommand();
try { try {
trigger.Execute(&*current_db_.execution_db_accessor_, &execution_memory, flags::run_time::GetExecutionTimeout(), trigger.Execute(&*current_db_.execution_db_accessor_, execution_memory.resource(),
&interpreter_context_->is_shutting_down, &transaction_status_, *trigger_context); flags::run_time::GetExecutionTimeout(), &interpreter_context_->is_shutting_down,
&transaction_status_, *trigger_context);
} catch (const utils::BasicException &e) { } catch (const utils::BasicException &e) {
throw utils::BasicException( throw utils::BasicException(
fmt::format("Trigger '{}' caused the transaction to fail.\nException: {}", trigger.Name(), e.what())); fmt::format("Trigger '{}' caused the transaction to fail.\nException: {}", trigger.Name(), e.what()));

92
src/query/interpreter.hpp
View File

@ -65,6 +65,54 @@ extern const Event SuccessfulQuery;
namespace memgraph::query { namespace memgraph::query {
struct QueryAllocator {
QueryAllocator() = default;
QueryAllocator(QueryAllocator const &) = delete;
QueryAllocator &operator=(QueryAllocator const &) = delete;
// No move addresses to pool & monotonic fields must be stable
QueryAllocator(QueryAllocator &&) = delete;
QueryAllocator &operator=(QueryAllocator &&) = delete;
auto resource() -> utils::MemoryResource * {
#ifndef MG_MEMORY_PROFILE
return &pool;
#else
return upstream_resource();
#endif
}
auto resource_without_pool() -> utils::MemoryResource * {
#ifndef MG_MEMORY_PROFILE
return &monotonic;
#else
return upstream_resource();
#endif
}
auto resource_without_pool_or_mono() -> utils::MemoryResource * { return upstream_resource(); }
private:
// At least one page to ensure not sharing page with other subsystems
static constexpr auto kMonotonicInitialSize = 4UL * 1024UL;
// TODO: need to profile to check for good defaults, also maybe PoolResource
// needs to be smarter. We expect more reuse of smaller objects than larger
// objects. 64*1024B is maybe wasteful, whereas 256*32B maybe sensible.
// Depends on number of small objects expected.
static constexpr auto kPoolBlockPerChunk = 64UL;
static constexpr auto kPoolMaxBlockSize = 1024UL;
static auto upstream_resource() -> utils::MemoryResource * {
// singleton ResourceWithOutOfMemoryException
// explicitly backed by NewDeleteResource
static auto upstream = utils::ResourceWithOutOfMemoryException{utils::NewDeleteResource()};
return &upstream;
}
#ifndef MG_MEMORY_PROFILE
memgraph::utils::MonotonicBufferResource monotonic{kMonotonicInitialSize, upstream_resource()};
memgraph::utils::PoolResource pool{kPoolBlockPerChunk, kPoolMaxBlockSize, &monotonic, upstream_resource()};
#endif
};
struct InterpreterContext; struct InterpreterContext;
inline constexpr size_t kExecutionMemoryBlockSize = 1UL * 1024UL * 1024UL; inline constexpr size_t kExecutionMemoryBlockSize = 1UL * 1024UL * 1024UL;
@ -304,46 +352,30 @@ class Interpreter final {
} }
struct QueryExecution { struct QueryExecution {
std::variant<utils::MonotonicBufferResource, utils::PoolResource> execution_memory; QueryAllocator execution_memory; // NOTE: before all other fields which uses this memory
utils::ResourceWithOutOfMemoryException execution_memory_with_exception;
std::optional<PreparedQuery> prepared_query;
std::optional<PreparedQuery> prepared_query;
std::map<std::string, TypedValue> summary; std::map<std::string, TypedValue> summary;
std::vector<Notification> notifications; std::vector<Notification> notifications;
static auto Create(std::variant<utils::MonotonicBufferResource, utils::PoolResource> memory_resource, static auto Create() -> std::unique_ptr<QueryExecution> { return std::make_unique<QueryExecution>(); }
std::optional<PreparedQuery> prepared_query = std::nullopt) -> std::unique_ptr<QueryExecution> {
return std::make_unique<QueryExecution>(std::move(memory_resource), std::move(prepared_query));
}
explicit QueryExecution(std::variant<utils::MonotonicBufferResource, utils::PoolResource> memory_resource, explicit QueryExecution() = default;
std::optional<PreparedQuery> prepared_query)
: execution_memory(std::move(memory_resource)), prepared_query{std::move(prepared_query)} {
std::visit(
[&](auto &memory_resource) {
execution_memory_with_exception = utils::ResourceWithOutOfMemoryException(&memory_resource);
},
execution_memory);
};
QueryExecution(const QueryExecution &) = delete; QueryExecution(const QueryExecution &) = delete;
QueryExecution(QueryExecution &&) = default; QueryExecution(QueryExecution &&) = delete;
QueryExecution &operator=(const QueryExecution &) = delete; QueryExecution &operator=(const QueryExecution &) = delete;
QueryExecution &operator=(QueryExecution &&) = default; QueryExecution &operator=(QueryExecution &&) = delete;
~QueryExecution() { ~QueryExecution() = default;
// We should always release the execution memory AFTER we
// destroy the prepared query which is using that instance
// of execution memory.
prepared_query.reset();
std::visit([](auto &memory_resource) { memory_resource.Release(); }, execution_memory);
}
void CleanRuntimeData() { void CleanRuntimeData() {
if (prepared_query.has_value()) { // Called from Commit/Abort once query has been fully used
prepared_query.reset();
} prepared_query.reset();
notifications.clear(); notifications.clear();
// TODO: double check is summary still needed here
// can we dispose of it and also execution_memory at this point?
} }
}; };
@ -413,9 +445,7 @@ std::map<std::string, TypedValue> Interpreter::Pull(TStream *result_stream, std:
try { try {
// Wrap the (statically polymorphic) stream type into a common type which // Wrap the (statically polymorphic) stream type into a common type which
// the handler knows. // the handler knows.
AnyStream stream{result_stream, AnyStream stream{result_stream, query_execution->execution_memory.resource()};
std::visit([](auto &execution_memory) -> utils::MemoryResource * { return &execution_memory; },
query_execution->execution_memory)};
const auto maybe_res = query_execution->prepared_query->query_handler(&stream, n); const auto maybe_res = query_execution->prepared_query->query_handler(&stream, n);
// Stream is using execution memory of the query_execution which // Stream is using execution memory of the query_execution which
// can be deleted after its execution so the stream should be cleared // can be deleted after its execution so the stream should be cleared

19
src/query/plan/operator.hpp
View File

@ -76,18 +76,13 @@ using UniqueCursorPtr = std::unique_ptr<Cursor, std::function<void(Cursor *)>>;
template <class TCursor, class... TArgs> template <class TCursor, class... TArgs>
std::unique_ptr<Cursor, std::function<void(Cursor *)>> MakeUniqueCursorPtr(utils::Allocator<TCursor> allocator, std::unique_ptr<Cursor, std::function<void(Cursor *)>> MakeUniqueCursorPtr(utils::Allocator<TCursor> allocator,
TArgs &&...args) { TArgs &&...args) {
auto *ptr = allocator.allocate(1); auto *cursor = allocator.template new_object<TCursor>(std::forward<TArgs>(args)...);
try { auto dtr = [allocator](Cursor *base_ptr) mutable {
auto *cursor = new (ptr) TCursor(std::forward<TArgs>(args)...); auto *p = static_cast<TCursor *>(base_ptr);
return std::unique_ptr<Cursor, std::function<void(Cursor *)>>(cursor, [allocator](Cursor *base_ptr) mutable { allocator.delete_object(p);
auto *p = static_cast<TCursor *>(base_ptr); };
p->~TCursor(); // TODO: not std::function
allocator.deallocate(p, 1); return std::unique_ptr<Cursor, std::function<void(Cursor *)>>(cursor, std::move(dtr));
});
} catch (...) {
allocator.deallocate(ptr, 1);
throw;
}
} }
class Once; class Once;

23
src/query/trigger.cpp
View File

@ -191,9 +191,9 @@ std::shared_ptr<Trigger::TriggerPlan> Trigger::GetPlan(DbAccessor *db_accessor)
return trigger_plan_; return trigger_plan_;
} }
void Trigger::Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution_memory, void Trigger::Execute(DbAccessor *dba, utils::MemoryResource *execution_memory, const double max_execution_time_sec,
const double max_execution_time_sec, std::atomic<bool> *is_shutting_down, std::atomic<bool> *is_shutting_down, std::atomic<TransactionStatus> *transaction_status,
std::atomic<TransactionStatus> *transaction_status, const TriggerContext &context) const { const TriggerContext &context) const {
if (!context.ShouldEventTrigger(event_type_)) { if (!context.ShouldEventTrigger(event_type_)) {
return; return;
} }
@ -214,22 +214,7 @@ void Trigger::Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution
ctx.is_shutting_down = is_shutting_down; ctx.is_shutting_down = is_shutting_down;
ctx.transaction_status = transaction_status; ctx.transaction_status = transaction_status;
ctx.is_profile_query = false; ctx.is_profile_query = false;
ctx.evaluation_context.memory = execution_memory;
// 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`.
static constexpr size_t stack_size = 256UL * 1024UL;
char stack_data[stack_size];
// 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.
utils::ResourceWithOutOfMemoryException resource_with_exception;
utils::MonotonicBufferResource monotonic_memory(&stack_data[0], stack_size, &resource_with_exception);
// TODO (mferencevic): Tune the parameters accordingly.
utils::PoolResource pool_memory(128, 1024, &monotonic_memory);
ctx.evaluation_context.memory = &pool_memory;
auto cursor = plan.plan().MakeCursor(execution_memory); auto cursor = plan.plan().MakeCursor(execution_memory);
Frame frame{plan.symbol_table().max_position(), execution_memory}; Frame frame{plan.symbol_table().max_position(), execution_memory};

2
src/query/trigger.hpp
View File

@ -39,7 +39,7 @@ struct Trigger {
utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor, utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor,
const InterpreterConfig::Query &query_config, std::shared_ptr<QueryUserOrRole> owner); const InterpreterConfig::Query &query_config, std::shared_ptr<QueryUserOrRole> owner);
void Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution_memory, double max_execution_time_sec, void Execute(DbAccessor *dba, utils::MemoryResource *execution_memory, double max_execution_time_sec,
std::atomic<bool> *is_shutting_down, std::atomic<TransactionStatus> *transaction_status, std::atomic<bool> *is_shutting_down, std::atomic<TransactionStatus> *transaction_status,
const TriggerContext &context) const; const TriggerContext &context) const;

21
src/utils/memory.cpp
View File

@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd. // Copyright 2024 Memgraph Ltd.
// //
// Use of this software is governed by the Business Source License // 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 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -355,4 +355,23 @@ void PoolResource::Release() {
// PoolResource END // PoolResource END
struct NullMemoryResourceImpl final : public MemoryResource {
NullMemoryResourceImpl() = default;
~NullMemoryResourceImpl() override = default;
private:
void *DoAllocate(size_t bytes, size_t alignment) override { throw BadAlloc{"NullMemoryResource doesn't allocate"}; }
void DoDeallocate(void *p, size_t bytes, size_t alignment) override {
throw BadAlloc{"NullMemoryResource doesn't deallocate"};
}
bool DoIsEqual(MemoryResource const &other) const noexcept override {
return dynamic_cast<NullMemoryResourceImpl const *>(&other) != nullptr;
}
};
MemoryResource *NullMemoryResource() noexcept {
static auto res = NullMemoryResourceImpl{};
return &res;
}
} // namespace memgraph::utils } // namespace memgraph::utils

4
src/utils/memory.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd. // Copyright 2024 Memgraph Ltd.
// //
// Use of this software is governed by the Business Source License // 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 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -248,6 +248,8 @@ bool operator!=(const Allocator<T> &a, const Allocator<U> &b) {
return !(a == b); return !(a == b);
} }
auto NullMemoryResource() noexcept -> MemoryResource *;
/// Wraps std::pmr::memory_resource for use with out MemoryResource /// Wraps std::pmr::memory_resource for use with out MemoryResource
class StdMemoryResource final : public MemoryResource { class StdMemoryResource final : public MemoryResource {
public: public: