Compare commits

...

39 Commits

Author SHA1 Message Date
Antonio Andelic
47c82c7d65 Ignore header files for clang tidy diff 2021-04-21 15:07:54 +02:00
Antonio Andelic
686c1bbc1a Print absolute path for build folder 2021-04-21 11:29:24 +02:00
Antonio Andelic
ccb609be5b save compile commands 2021-04-21 11:29:13 +02:00
Antonio Andelic
860a1a0ea6 Print out build folder for clang-tidy 2021-04-21 11:12:01 +02:00
Antonio Andelic
c79e843de6 Remove unwanted code 2021-04-21 11:04:01 +02:00
Antonio Andelic
805bab80d3 Change logic for getting TypedValue for trigger context 2021-04-21 10:57:11 +02:00
Antonio Andelic
33216920f7 Allow predefined identifiers in any scope 2021-04-21 10:20:20 +02:00
Antonio Andelic
42cfd2be28 Allow predefined variables in any scope but only in one 2021-04-20 15:19:47 +02:00
Antonio Andelic
012a02dc93 Check after commit trigger size 2021-04-20 13:22:03 +02:00
Antonio Andelic
b0e7b08c85 Add test for predefined variables in symbol table 2021-04-20 13:17:23 +02:00
Antonio Andelic
15318ef4f4 Remove predefined identifiers in union 2021-04-20 13:16:46 +02:00
Antonio Andelic
6cff70b8e8 Use pointer instead of nonconst ref 2021-04-20 11:51:43 +02:00
Antonio Andelic
c8ae33de89 Merge E025 2021-04-20 11:19:49 +02:00
antonio2368
caac055f80 Add support for after commit triggers (#136) 2021-04-20 11:10:00 +02:00
antonio2368
846403f727 Initial trigger definition (#133)
* Pull out cypher query parsing logic

* Define trigger structure

* Run triggers before commit

* Use skip list for saving triggers
2021-04-20 11:10:00 +02:00
Antonio Andelic
96acf2e9d9 Update CHANGELOG version 2021-04-20 11:10:00 +02:00
Antonio Andelic
912e20dd07 Adapt TriggerContext inplace 2021-04-20 11:08:12 +02:00
Antonio Andelic
a812a256cf Refactor the code 2021-04-20 10:41:21 +02:00
Antonio Andelic
ca7d3bfde0 Move shared lock with the accessor 2021-04-20 10:36:30 +02:00
Antonio Andelic
2e2d2c1a8f Extend TriggerContext 2021-04-20 10:08:21 +02:00
Antonio Andelic
cf3ebf1b47 Add trigger context 2021-04-19 16:15:54 +02:00
Antonio Andelic
24b419db8b Initial version of predefined variable support 2021-04-19 13:19:34 +02:00
Antonio Andelic
4397b2db49 Rename function for running queries 2021-04-16 14:54:47 +02:00
Antonio Andelic
4d96a234d7 Initial predefined variable implementation 2021-04-16 12:32:32 +02:00
Antonio Andelic
609c9c2d50 Run before commit triggers in same transaction 2021-04-15 13:05:56 +02:00
Antonio Andelic
3d6112b4fd Merge E025 2021-04-15 10:53:39 +02:00
antonio2368
7812af37cb Initial trigger definition (#133)
* Pull out cypher query parsing logic

* Define trigger structure

* Run triggers before commit

* Use skip list for saving triggers
2021-04-15 10:49:40 +02:00
Antonio Andelic
575231e15f Ignore clang-tidy warning 2021-04-15 08:48:04 +02:00
Antonio Andelic
565e2b3dd4 Add suport for after commit triggers 2021-04-14 16:21:33 +02:00
Antonio Andelic
0f9f3523f5 Merge branch 'E025-MG-triggers' into T0544-MG-before-commit-triggers 2021-04-13 14:20:41 +02:00
Antonio Andelic
73b7ffda10 Update CHANGELOG version 2021-04-13 14:20:21 +02:00
Antonio Andelic
c9b4361f7e Add timeout for trigger queries 2021-04-13 11:15:03 +02:00
Antonio Andelic
bd8e96ed75 Disable only specific clang-tidy checks 2021-04-13 10:54:02 +02:00
Antonio Andelic
ee9badf828 Merge branch 'E025-MG-triggers' into T0544-MG-before-commit-triggers 2021-04-13 10:43:00 +02:00
Antonio Andelic
8cca35daeb Fix clang-tidy warnings 2021-04-08 12:55:51 +02:00
Antonio Andelic
e6566b3655 Use skip list for saving triggers 2021-04-08 11:27:12 +02:00
Antonio Andelic
82e4e485f8 Run triggers before commit 2021-04-08 11:01:55 +02:00
Antonio Andelic
e760a89eb7 Define trigger structure 2021-04-08 10:27:42 +02:00
Antonio Andelic
613f0b7ce7 Pull out cypher query parsing logic 2021-04-08 09:09:04 +02:00
19 changed files with 725 additions and 584 deletions

View File

@@ -1,5 +1,4 @@
name: Diff
on:
push:
paths-ignore:
@@ -8,63 +7,6 @@ on:
- '.clang-*'
jobs:
community_build:
name: "Community build"
runs-on: [self-hosted, General, Linux, X64, Debian10]
env:
THREADS: 24
steps:
- name: Set up repository
uses: actions/checkout@v2
with:
# Number of commits to fetch. `0` indicates all history for all
# branches and tags. (default: 1)
fetch-depth: 0
- name: Build community binaries
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Initialize dependencies.
./init
# Build community binaries.
cd build
cmake -DCMAKE_BUILD_TYPE=release -DMG_ENTERPRISE=OFF ..
make -j$THREADS
- name: Run unit tests
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Run unit tests.
cd build
ctest -R memgraph__unit --output-on-failure -j$THREADS
- name: Run stress test
run: |
cd tests/stress
./continuous_integration
- name: Create community DEB package
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Create community DEB package.
cd build
mkdir output && cd output
cpack -G DEB --config ../CPackConfig.cmake
- name: Save community DEB package
uses: actions/upload-artifact@v2
with:
name: "Community DEB package"
path: build/output/memgraph*.deb
coverage_build:
name: "Code analysis"
runs-on: [self-hosted, General, Linux, X64, Debian10]
@@ -91,259 +33,18 @@ jobs:
cmake -DTEST_COVERAGE=ON -DASAN=ON -DUBSAN=ON ..
make -j$THREADS memgraph__unit
- name: Run unit tests
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Run unit tests. It is restricted to 2 threads intentionally, because higher concurrency makes the timing related tests unstable.
cd build
LSAN_OPTIONS=suppressions=$PWD/../tools/lsan.supp UBSAN_OPTIONS=halt_on_error=1 ctest -R memgraph__unit --output-on-failure -j2
- name: Compute code coverage
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Compute code coverage.
cd tools/github
./coverage_convert
# Package code coverage.
cd generated
tar -czf code_coverage.tar.gz coverage.json html report.json summary.rmu
- name: Save code coverage
uses: actions/upload-artifact@v2
with:
name: "Code coverage"
path: tools/github/generated/code_coverage.tar.gz
- name: Run clang-tidy
run: |
source /opt/toolchain-v2/activate
# Restrict clang-tidy results only to the modified parts
git diff -U0 master... -- src | ./tools/github/clang-tidy/clang-tidy-diff.py -p 1 -j $THREADS -path build | tee ./build/clang_tidy_output.txt
git diff -U0 master... -- src ':!*.hpp' | ./tools/github/clang-tidy/clang-tidy-diff.py -p 1 -j $THREADS -path build | tee ./build/clang_tidy_output.txt
# Fail if any warning is reported
! cat ./build/clang_tidy_output.txt | ./tools/github/clang-tidy/grep_error_lines.sh > /dev/null
debug_build:
name: "Debug build"
runs-on: [self-hosted, General, Linux, X64, Debian10]
env:
THREADS: 24
steps:
- name: Set up repository
uses: actions/checkout@v2
with:
# Number of commits to fetch. `0` indicates all history for all
# branches and tags. (default: 1)
fetch-depth: 0
- name: Build debug binaries
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Initialize dependencies.
./init
# Build debug binaries.
cd build
cmake ..
make -j$THREADS
- name: Run leftover CTest tests
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Run leftover CTest tests (all except unit and benchmark tests).
cd build
ctest -E "(memgraph__unit|memgraph__benchmark)" --output-on-failure
- name: Run drivers tests
run: |
./tests/drivers/run.sh
- name: Run integration tests
run: |
cd tests/integration
for name in *; do
if [ ! -d $name ]; then continue; fi
pushd $name >/dev/null
echo "Running: $name"
if [ -x prepare.sh ]; then
./prepare.sh
fi
if [ -x runner.py ]; then
./runner.py
elif [ -x runner.sh ]; then
./runner.sh
fi
echo
popd >/dev/null
done
- name: Run cppcheck and clang-format
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Run cppcheck and clang-format.
cd tools/github
./cppcheck_and_clang_format diff
- name: Save cppcheck and clang-format errors
uses: actions/upload-artifact@v2
with:
name: "Code coverage"
path: tools/github/cppcheck_and_clang_format.txt
release_build:
name: "Release build"
runs-on: [self-hosted, General, Linux, X64, Debian10]
env:
THREADS: 24
steps:
- name: Set up repository
uses: actions/checkout@v2
with:
# Number of commits to fetch. `0` indicates all history for all
# branches and tags. (default: 1)
fetch-depth: 0
- name: Build release binaries
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Initialize dependencies.
./init
# Build release binaries.
cd build
cmake -DCMAKE_BUILD_TYPE=release ..
make -j$THREADS
- name: Run macro benchmark tests
run: |
cd tests/macro_benchmark
./harness QuerySuite MemgraphRunner \
--groups aggregation 1000_create unwind_create dense_expand match \
--no-strict
- name: Run GQL Behave tests
run: |
cd tests/gql_behave
./continuous_integration
- name: Save quality assurance status
uses: actions/upload-artifact@v2
with:
name: "GQL Behave Status"
path: |
tests/gql_behave/gql_behave_status.csv
tests/gql_behave/gql_behave_status.html
- name: Run e2e replication tests
run: |
# TODO(gitbuda): Setup mgclient and pymgclient properly.
cd tests
./setup.sh
source ve3/bin/activate
cd e2e
LD_LIBRARY_PATH=$LD_LIBRARY_PATH:../../libs/mgclient/lib python runner.py --workloads-path replication/workloads.yaml
- name: Run e2e memory control tests
run: |
cd tests
./setup.sh
source ve3/bin/activate
cd e2e
LD_LIBRARY_PATH=$LD_LIBRARY_PATH:../../libs/mgclient/lib python runner.py --workloads-path memory/workloads.yaml
- name: Run stress test (plain)
run: |
cd tests/stress
./continuous_integration
- name: Run stress test (SSL)
run: |
cd tests/stress
./continuous_integration --use-ssl
- name: Run durability test
run: |
cd tests/stress
source ve3/bin/activate
python3 durability --num-steps 5
- name: Create enterprise DEB package
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Create enterprise DEB package.
cd build
mkdir output && cd output
cpack -G DEB --config ../CPackConfig.cmake
- name: Save enterprise DEB package
uses: actions/upload-artifact@v2
with:
name: "Enterprise DEB package"
path: build/output/memgraph*.deb
- name: Save test data
- name: Save compile commands
uses: actions/upload-artifact@v2
if: always()
with:
name: "Test data"
path: |
# multiple paths could be defined
build/logs
release_jepsen_test:
name: "Release Jepsen Test"
runs-on: [self-hosted, Linux, X64, Debian10, JepsenControl]
#continue-on-error: true
env:
THREADS: 24
steps:
- name: Set up repository
uses: actions/checkout@v2
with:
# Number of commits to fetch. `0` indicates all history for all
# branches and tags. (default: 1)
fetch-depth: 0
- name: Build release binaries
run: |
# Activate toolchain.
source /opt/toolchain-v2/activate
# Initialize dependencies.
./init
# Build only memgraph release binarie.
cd build
cmake -DCMAKE_BUILD_TYPE=release ..
make -j$THREADS memgraph
- name: Run Jepsen tests
run: |
cd tests/jepsen
./run.sh test --binary ../../build/memgraph --run-args "test-all --node-configs resources/node-config.edn" --ignore-run-stdout-logs --ignore-run-stderr-logs
- name: Save Jepsen report
uses: actions/upload-artifact@v2
if: ${{ always() }}
with:
name: "Jepsen Report"
path: tests/jepsen/Jepsen.tar.gz
name: "compile commands"
path: build/compile_commands.json

View File

@@ -2,6 +2,10 @@
## Future
### Major Feature and Improvements
## v1.4.0
### Breaking Changes
* Changed `MEMORY LIMIT num (KB|MB)` clause in the procedure calls to `PROCEDURE MEMORY LIMIT num (KB|MB)`.

View File

@@ -9,6 +9,7 @@ add_custom_target(generate_lcp_query DEPENDS ${generated_lcp_query_files})
set(mg_query_sources
${lcp_query_cpp_files}
common.cpp
cypher_query_interpreter.cpp
dump.cpp
frontend/ast/cypher_main_visitor.cpp
frontend/ast/pretty_print.cpp
@@ -30,6 +31,7 @@ set(mg_query_sources
procedure/mg_procedure_impl.cpp
procedure/module.cpp
procedure/py_module.cpp
trigger.cpp
typed_value.cpp)
add_library(mg-query STATIC ${mg_query_sources})

View File

@@ -4,6 +4,7 @@
#include "query/frontend/semantic/symbol_table.hpp"
#include "query/parameters.hpp"
#include "query/plan/profile.hpp"
#include "query/trigger.hpp"
#include "utils/tsc.hpp"
namespace query {
@@ -56,6 +57,9 @@ struct ExecutionContext {
std::chrono::duration<double> profile_execution_time;
plan::ProfilingStats stats;
plan::ProfilingStats *stats_root{nullptr};
// trigger context
TriggerContext *trigger_context{nullptr};
};
inline bool MustAbort(const ExecutionContext &context) {

View File

@@ -0,0 +1,140 @@
#include "query/cypher_query_interpreter.hpp"
// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_HIDDEN_bool(query_cost_planner, true, "Use the cost-estimating query planner.");
// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_VALIDATED_int32(query_plan_cache_ttl, 60, "Time to live for cached query plans, in seconds.",
FLAG_IN_RANGE(0, std::numeric_limits<int32_t>::max()));
namespace query {
CachedPlan::CachedPlan(std::unique_ptr<LogicalPlan> plan) : plan_(std::move(plan)) {}
ParsedQuery ParseQuery(const std::string &query_string, const std::map<std::string, storage::PropertyValue> &params,
utils::SkipList<QueryCacheEntry> *cache, utils::SpinLock *antlr_lock) {
// Strip the query for caching purposes. The process of stripping a query
// "normalizes" it by replacing any literals with new parameters. This
// results in just the *structure* of the query being taken into account for
// caching.
frontend::StrippedQuery stripped_query{query_string};
// Copy over the parameters that were introduced during stripping.
Parameters parameters{stripped_query.literals()};
// Check that all user-specified parameters are provided.
for (const auto &param_pair : stripped_query.parameters()) {
auto it = params.find(param_pair.second);
if (it == params.end()) {
throw query::UnprovidedParameterError("Parameter ${} not provided.", param_pair.second);
}
parameters.Add(param_pair.first, it->second);
}
// Cache the query's AST if it isn't already.
auto hash = stripped_query.hash();
auto accessor = cache->access();
auto it = accessor.find(hash);
std::unique_ptr<frontend::opencypher::Parser> parser;
// Return a copy of both the AST storage and the query.
CachedQuery result;
bool is_cacheable = true;
auto get_information_from_cache = [&](const auto &cached_query) {
result.ast_storage.properties_ = cached_query.ast_storage.properties_;
result.ast_storage.labels_ = cached_query.ast_storage.labels_;
result.ast_storage.edge_types_ = cached_query.ast_storage.edge_types_;
result.query = cached_query.query->Clone(&result.ast_storage);
result.required_privileges = cached_query.required_privileges;
};
if (it == accessor.end()) {
{
std::unique_lock<utils::SpinLock> guard(*antlr_lock);
try {
parser = std::make_unique<frontend::opencypher::Parser>(stripped_query.query());
} catch (const SyntaxException &e) {
// There is a syntax exception in the stripped query. Re-run the parser
// on the original query to get an appropriate error messsage.
parser = std::make_unique<frontend::opencypher::Parser>(query_string);
// If an exception was not thrown here, the stripper messed something
// up.
LOG_FATAL("The stripped query can't be parsed, but the original can.");
}
}
// Convert the ANTLR4 parse tree into an AST.
AstStorage ast_storage;
frontend::ParsingContext context{true};
frontend::CypherMainVisitor visitor(context, &ast_storage);
visitor.visit(parser->tree());
if (visitor.IsCacheable()) {
CachedQuery cached_query{std::move(ast_storage), visitor.query(), query::GetRequiredPrivileges(visitor.query())};
it = accessor.insert({hash, std::move(cached_query)}).first;
get_information_from_cache(it->second);
} else {
result.ast_storage.properties_ = ast_storage.properties_;
result.ast_storage.labels_ = ast_storage.labels_;
result.ast_storage.edge_types_ = ast_storage.edge_types_;
result.query = visitor.query()->Clone(&result.ast_storage);
result.required_privileges = query::GetRequiredPrivileges(visitor.query());
is_cacheable = false;
}
} else {
get_information_from_cache(it->second);
}
return ParsedQuery{query_string,
params,
std::move(parameters),
std::move(stripped_query),
std::move(result.ast_storage),
result.query,
std::move(result.required_privileges),
is_cacheable};
}
namespace {
std::unique_ptr<LogicalPlan> MakeLogicalPlan(AstStorage ast_storage, CypherQuery *query, const Parameters &parameters,
DbAccessor *db_accessor,
std::unordered_map<std::string, Identifier *> predefined_identifiers) {
auto vertex_counts = plan::MakeVertexCountCache(db_accessor);
auto symbol_table = MakeSymbolTable(query, std::move(predefined_identifiers));
auto planning_context = plan::MakePlanningContext(&ast_storage, &symbol_table, query, &vertex_counts);
auto [root, cost] = plan::MakeLogicalPlan(&planning_context, parameters, FLAGS_query_cost_planner);
return std::make_unique<SingleNodeLogicalPlan>(std::move(root), cost, std::move(ast_storage),
std::move(symbol_table));
}
} // namespace
std::shared_ptr<CachedPlan> CypherQueryToPlan(uint64_t hash, AstStorage ast_storage, CypherQuery *query,
const Parameters &parameters, utils::SkipList<PlanCacheEntry> *plan_cache,
DbAccessor *db_accessor, const bool is_cacheable,
std::unordered_map<std::string, Identifier *> predefined_identifiers) {
auto plan_cache_access = plan_cache->access();
auto it = plan_cache_access.find(hash);
if (it != plan_cache_access.end()) {
if (it->second->IsExpired()) {
plan_cache_access.remove(hash);
} else {
return it->second;
}
}
auto plan = std::make_shared<CachedPlan>(
MakeLogicalPlan(std::move(ast_storage), query, parameters, db_accessor, std::move(predefined_identifiers)));
if (is_cacheable) {
plan_cache_access.insert({hash, plan});
}
return plan;
}
} // namespace query

View File

@@ -0,0 +1,141 @@
#pragma once
//////////////////////////////////////////////////////
// THIS INCLUDE SHOULD ALWAYS COME BEFORE THE
// "cypher_main_visitor.hpp"
// "planner.hpp" includes json.hpp which uses libc's
// EOF macro while "cypher_main_visitor.hpp" includes
// "antlr4-runtime.h" which contains a static variable
// of the same name, EOF.
// This hides the definition of the macro which causes
// the compilation to fail.
#include "query/plan/planner.hpp"
//////////////////////////////////////////////////////
#include "query/frontend/ast/cypher_main_visitor.hpp"
#include "query/frontend/opencypher/parser.hpp"
#include "query/frontend/semantic/required_privileges.hpp"
#include "query/frontend/semantic/symbol_generator.hpp"
#include "query/frontend/stripped.hpp"
#include "utils/flag_validation.hpp"
#include "utils/timer.hpp"
// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
DECLARE_bool(query_cost_planner);
// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
DECLARE_int32(query_plan_cache_ttl);
namespace query {
// TODO: Maybe this should move to query/plan/planner.
/// Interface for accessing the root operator of a logical plan.
class LogicalPlan {
public:
explicit LogicalPlan() = default;
virtual ~LogicalPlan() = default;
LogicalPlan(const LogicalPlan &) = default;
LogicalPlan &operator=(const LogicalPlan &) = default;
LogicalPlan(LogicalPlan &&) = default;
LogicalPlan &operator=(LogicalPlan &&) = default;
virtual const plan::LogicalOperator &GetRoot() const = 0;
virtual double GetCost() const = 0;
virtual const SymbolTable &GetSymbolTable() const = 0;
virtual const AstStorage &GetAstStorage() const = 0;
};
class CachedPlan {
public:
explicit CachedPlan(std::unique_ptr<LogicalPlan> plan);
const auto &plan() const { return plan_->GetRoot(); }
double cost() const { return plan_->GetCost(); }
const auto &symbol_table() const { return plan_->GetSymbolTable(); }
const auto &ast_storage() const { return plan_->GetAstStorage(); }
bool IsExpired() const {
// NOLINTNEXTLINE (modernize-use-nullptr)
return cache_timer_.Elapsed() > std::chrono::seconds(FLAGS_query_plan_cache_ttl);
};
private:
std::unique_ptr<LogicalPlan> plan_;
utils::Timer cache_timer_;
};
struct CachedQuery {
AstStorage ast_storage;
Query *query;
std::vector<AuthQuery::Privilege> required_privileges;
};
struct QueryCacheEntry {
bool operator==(const QueryCacheEntry &other) const { return first == other.first; }
bool operator<(const QueryCacheEntry &other) const { return first < other.first; }
bool operator==(const uint64_t &other) const { return first == other; }
bool operator<(const uint64_t &other) const { return first < other; }
uint64_t first;
// TODO: Maybe store the query string here and use it as a key with the hash
// so that we eliminate the risk of hash collisions.
CachedQuery second;
};
struct PlanCacheEntry {
bool operator==(const PlanCacheEntry &other) const { return first == other.first; }
bool operator<(const PlanCacheEntry &other) const { return first < other.first; }
bool operator==(const uint64_t &other) const { return first == other; }
bool operator<(const uint64_t &other) const { return first < other; }
uint64_t first;
// TODO: Maybe store the query string here and use it as a key with the hash
// so that we eliminate the risk of hash collisions.
std::shared_ptr<CachedPlan> second;
};
/**
* A container for data related to the parsing of a query.
*/
struct ParsedQuery {
std::string query_string;
std::map<std::string, storage::PropertyValue> user_parameters;
Parameters parameters;
frontend::StrippedQuery stripped_query;
AstStorage ast_storage;
Query *query;
std::vector<AuthQuery::Privilege> required_privileges;
bool is_cacheable{true};
};
ParsedQuery ParseQuery(const std::string &query_string, const std::map<std::string, storage::PropertyValue> &params,
utils::SkipList<QueryCacheEntry> *cache, utils::SpinLock *antlr_lock);
class SingleNodeLogicalPlan final : public LogicalPlan {
public:
SingleNodeLogicalPlan(std::unique_ptr<plan::LogicalOperator> root, double cost, AstStorage storage,
const SymbolTable &symbol_table)
: root_(std::move(root)), cost_(cost), storage_(std::move(storage)), symbol_table_(symbol_table) {}
const plan::LogicalOperator &GetRoot() const override { return *root_; }
double GetCost() const override { return cost_; }
const SymbolTable &GetSymbolTable() const override { return symbol_table_; }
const AstStorage &GetAstStorage() const override { return storage_; }
private:
std::unique_ptr<plan::LogicalOperator> root_;
double cost_;
AstStorage storage_;
SymbolTable symbol_table_;
};
/**
* Return the parsed *Cypher* query's AST cached logical plan, or create and
* cache a fresh one if it doesn't yet exist.
*/
std::shared_ptr<CachedPlan> CypherQueryToPlan(
uint64_t hash, AstStorage ast_storage, CypherQuery *query, const Parameters &parameters,
utils::SkipList<PlanCacheEntry> *plan_cache, DbAccessor *db_accessor, bool is_cacheable = true,
std::unordered_map<std::string, Identifier *> predefined_identifiers = {});
} // namespace query

View File

@@ -208,6 +208,8 @@ class DbAccessor final {
return std::nullopt;
}
void FinalizeTransaction() { accessor_->FinalizeTransaction(); }
VerticesIterable Vertices(storage::View view) { return VerticesIterable(accessor_->Vertices(view)); }
VerticesIterable Vertices(storage::View view, storage::LabelId label) {

View File

@@ -686,9 +686,7 @@ cpp<#
symbol_pos_ = symbol.position();
return this;
}
cpp<#)
(:protected
#>cpp
explicit Identifier(const std::string &name) : name_(name) {}
Identifier(const std::string &name, bool user_declared)
: name_(name), user_declared_(user_declared) {}

View File

@@ -13,7 +13,7 @@
namespace query {
auto SymbolGenerator::CreateSymbol(const std::string &name, bool user_declared, Symbol::Type type, int token_position) {
auto symbol = symbol_table_.CreateSymbol(name, user_declared, type, token_position);
auto symbol = symbol_table_->CreateSymbol(name, user_declared, type, token_position);
scope_.symbols[name] = symbol;
return symbol;
}
@@ -227,7 +227,7 @@ bool SymbolGenerator::PostVisit(Match &) {
// Check variables in property maps after visiting Match, so that they can
// reference symbols out of bind order.
for (auto &ident : scope_.identifiers_in_match) {
if (!HasSymbol(ident->name_)) throw UnboundVariableError(ident->name_);
if (!HasSymbol(ident->name_) && !PredefinedIdentifier(ident->name_)) throw UnboundVariableError(ident->name_);
ident->MapTo(scope_.symbols[ident->name_]);
}
scope_.identifiers_in_match.clear();
@@ -277,7 +277,7 @@ SymbolGenerator::ReturnType SymbolGenerator::Visit(Identifier &ident) {
scope_.identifiers_in_match.emplace_back(&ident);
} else {
// Everything else references a bound symbol.
if (!HasSymbol(ident.name_)) throw UnboundVariableError(ident.name_);
if (!HasSymbol(ident.name_) && !PredefinedIdentifier(ident.name_)) throw UnboundVariableError(ident.name_);
symbol = scope_.symbols[ident.name_];
}
ident.MapTo(symbol);
@@ -448,10 +448,10 @@ bool SymbolGenerator::PreVisit(EdgeAtom &edge_atom) {
// Create inner symbols, but don't bind them in scope, since they are to
// be used in the missing filter expression.
auto *inner_edge = edge_atom.filter_lambda_.inner_edge;
inner_edge->MapTo(symbol_table_.CreateSymbol(inner_edge->name_, inner_edge->user_declared_, Symbol::Type::EDGE));
inner_edge->MapTo(symbol_table_->CreateSymbol(inner_edge->name_, inner_edge->user_declared_, Symbol::Type::EDGE));
auto *inner_node = edge_atom.filter_lambda_.inner_node;
inner_node->MapTo(
symbol_table_.CreateSymbol(inner_node->name_, inner_node->user_declared_, Symbol::Type::VERTEX));
symbol_table_->CreateSymbol(inner_node->name_, inner_node->user_declared_, Symbol::Type::VERTEX));
}
if (edge_atom.weight_lambda_.expression) {
VisitWithIdentifiers(edge_atom.weight_lambda_.expression,
@@ -506,4 +506,18 @@ void SymbolGenerator::VisitWithIdentifiers(Expression *expr, const std::vector<I
bool SymbolGenerator::HasSymbol(const std::string &name) { return scope_.symbols.find(name) != scope_.symbols.end(); }
bool SymbolGenerator::PredefinedIdentifier(const std::string &name) {
auto it = predefined_identifiers_.find(name);
if (it == predefined_identifiers_.end()) {
return false;
}
// we can only use the predefined identifier in a single scope
auto &identifier = it->second;
identifier->MapTo(CreateSymbol(identifier->name_, identifier->user_declared_));
predefined_identifiers_.erase(it);
return true;
}
} // namespace query

View File

@@ -17,7 +17,9 @@ namespace query {
/// variable types.
class SymbolGenerator : public HierarchicalTreeVisitor {
public:
explicit SymbolGenerator(SymbolTable &symbol_table) : symbol_table_(symbol_table) {}
explicit SymbolGenerator(SymbolTable *symbol_table,
std::unordered_map<std::string, Identifier *> predefined_identifiers)
: symbol_table_(symbol_table), predefined_identifiers_{std::move(predefined_identifiers)} {}
using HierarchicalTreeVisitor::PostVisit;
using HierarchicalTreeVisitor::PreVisit;
@@ -116,6 +118,8 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
bool HasSymbol(const std::string &name);
bool PredefinedIdentifier(const std::string &name);
// Returns a freshly generated symbol. Previous mapping of the same name to a
// different symbol is replaced with the new one.
auto CreateSymbol(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY,
@@ -129,15 +133,18 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
void VisitWithIdentifiers(Expression *, const std::vector<Identifier *> &);
SymbolTable &symbol_table_;
SymbolTable *symbol_table_;
AstStorage storage_;
std::unordered_map<std::string, Identifier *> predefined_identifiers_;
Scope scope_;
std::unordered_set<std::string> prev_return_names_;
std::unordered_set<std::string> curr_return_names_;
};
inline SymbolTable MakeSymbolTable(CypherQuery *query) {
inline SymbolTable MakeSymbolTable(CypherQuery *query,
std::unordered_map<std::string, Identifier *> predefined_identifiers = {}) {
SymbolTable symbol_table;
SymbolGenerator symbol_generator(symbol_table);
SymbolGenerator symbol_generator(&symbol_table, std::move(predefined_identifiers));
query->single_query_->Accept(symbol_generator);
for (auto *cypher_union : query->cypher_unions_) {
cypher_union->Accept(symbol_generator);

View File

@@ -31,10 +31,6 @@
#include "utils/string.hpp"
#include "utils/tsc.hpp"
DEFINE_HIDDEN_bool(query_cost_planner, true, "Use the cost-estimating query planner.");
DEFINE_VALIDATED_int32(query_plan_cache_ttl, 60, "Time to live for cached query plans, in seconds.",
FLAG_IN_RANGE(0, std::numeric_limits<int32_t>::max()));
namespace EventCounter {
extern Event ReadQuery;
extern Event WriteQuery;
@@ -62,135 +58,6 @@ void UpdateTypeCount(const plan::ReadWriteTypeChecker::RWType type) {
break;
}
}
} // namespace
/**
* A container for data related to the parsing of a query.
*/
struct ParsedQuery {
std::string query_string;
std::map<std::string, storage::PropertyValue> user_parameters;
Parameters parameters;
frontend::StrippedQuery stripped_query;
AstStorage ast_storage;
Query *query;
std::vector<AuthQuery::Privilege> required_privileges;
bool is_cacheable{true};
};
ParsedQuery ParseQuery(const std::string &query_string, const std::map<std::string, storage::PropertyValue> &params,
utils::SkipList<QueryCacheEntry> *cache, utils::SpinLock *antlr_lock) {
// Strip the query for caching purposes. The process of stripping a query
// "normalizes" it by replacing any literals with new parameters. This
// results in just the *structure* of the query being taken into account for
// caching.
frontend::StrippedQuery stripped_query{query_string};
// Copy over the parameters that were introduced during stripping.
Parameters parameters{stripped_query.literals()};
// Check that all user-specified parameters are provided.
for (const auto &param_pair : stripped_query.parameters()) {
auto it = params.find(param_pair.second);
if (it == params.end()) {
throw query::UnprovidedParameterError("Parameter ${} not provided.", param_pair.second);
}
parameters.Add(param_pair.first, it->second);
}
// Cache the query's AST if it isn't already.
auto hash = stripped_query.hash();
auto accessor = cache->access();
auto it = accessor.find(hash);
std::unique_ptr<frontend::opencypher::Parser> parser;
// Return a copy of both the AST storage and the query.
CachedQuery result;
bool is_cacheable = true;
auto get_information_from_cache = [&](const auto &cached_query) {
result.ast_storage.properties_ = cached_query.ast_storage.properties_;
result.ast_storage.labels_ = cached_query.ast_storage.labels_;
result.ast_storage.edge_types_ = cached_query.ast_storage.edge_types_;
result.query = cached_query.query->Clone(&result.ast_storage);
result.required_privileges = cached_query.required_privileges;
};
if (it == accessor.end()) {
{
std::unique_lock<utils::SpinLock> guard(*antlr_lock);
try {
parser = std::make_unique<frontend::opencypher::Parser>(stripped_query.query());
} catch (const SyntaxException &e) {
// There is a syntax exception in the stripped query. Re-run the parser
// on the original query to get an appropriate error messsage.
parser = std::make_unique<frontend::opencypher::Parser>(query_string);
// If an exception was not thrown here, the stripper messed something
// up.
LOG_FATAL("The stripped query can't be parsed, but the original can.");
}
}
// Convert the ANTLR4 parse tree into an AST.
AstStorage ast_storage;
frontend::ParsingContext context{true};
frontend::CypherMainVisitor visitor(context, &ast_storage);
visitor.visit(parser->tree());
if (visitor.IsCacheable()) {
CachedQuery cached_query{std::move(ast_storage), visitor.query(), query::GetRequiredPrivileges(visitor.query())};
it = accessor.insert({hash, std::move(cached_query)}).first;
get_information_from_cache(it->second);
} else {
result.ast_storage.properties_ = ast_storage.properties_;
result.ast_storage.labels_ = ast_storage.labels_;
result.ast_storage.edge_types_ = ast_storage.edge_types_;
result.query = visitor.query()->Clone(&result.ast_storage);
result.required_privileges = query::GetRequiredPrivileges(visitor.query());
is_cacheable = false;
}
} else {
get_information_from_cache(it->second);
}
return ParsedQuery{query_string,
params,
std::move(parameters),
std::move(stripped_query),
std::move(result.ast_storage),
result.query,
std::move(result.required_privileges),
is_cacheable};
}
class SingleNodeLogicalPlan final : public LogicalPlan {
public:
SingleNodeLogicalPlan(std::unique_ptr<plan::LogicalOperator> root, double cost, AstStorage storage,
const SymbolTable &symbol_table)
: root_(std::move(root)), cost_(cost), storage_(std::move(storage)), symbol_table_(symbol_table) {}
const plan::LogicalOperator &GetRoot() const override { return *root_; }
double GetCost() const override { return cost_; }
const SymbolTable &GetSymbolTable() const override { return symbol_table_; }
const AstStorage &GetAstStorage() const override { return storage_; }
private:
std::unique_ptr<plan::LogicalOperator> root_;
double cost_;
AstStorage storage_;
SymbolTable symbol_table_;
};
CachedPlan::CachedPlan(std::unique_ptr<LogicalPlan> plan) : plan_(std::move(plan)) {}
struct Callback {
std::vector<std::string> header;
@@ -575,11 +442,6 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, ReplQueryHandler *
}
}
Interpreter::Interpreter(InterpreterContext *interpreter_context) : interpreter_context_(interpreter_context) {
MG_ASSERT(interpreter_context_, "Interpreter context must not be NULL");
}
namespace {
// Struct for lazy pulling from a vector
struct PullPlanVector {
explicit PullPlanVector(std::vector<std::vector<TypedValue>> values) : values_(std::move(values)) {}
@@ -605,7 +467,8 @@ struct PullPlanVector {
struct PullPlan {
explicit PullPlan(std::shared_ptr<CachedPlan> plan, const Parameters &parameters, bool is_profile_query,
DbAccessor *dba, InterpreterContext *interpreter_context,
utils::MonotonicBufferResource *execution_memory, std::optional<size_t> memory_limit = {});
utils::MonotonicBufferResource *execution_memory, TriggerContext *trigger_context = nullptr,
std::optional<size_t> memory_limit = {});
std::optional<ExecutionContext> Pull(AnyStream *stream, std::optional<int> n,
const std::vector<Symbol> &output_symbols,
std::map<std::string, TypedValue> *summary);
@@ -633,7 +496,8 @@ struct PullPlan {
PullPlan::PullPlan(const std::shared_ptr<CachedPlan> plan, const Parameters &parameters, const bool is_profile_query,
DbAccessor *dba, InterpreterContext *interpreter_context,
utils::MonotonicBufferResource *execution_memory, const std::optional<size_t> memory_limit)
utils::MonotonicBufferResource *execution_memory, TriggerContext *trigger_context,
const std::optional<size_t> memory_limit)
: plan_(plan),
cursor_(plan->plan().MakeCursor(execution_memory)),
frame_(plan->symbol_table().max_position(), execution_memory),
@@ -650,6 +514,7 @@ PullPlan::PullPlan(const std::shared_ptr<CachedPlan> plan, const Parameters &par
ctx_.max_execution_time_sec = interpreter_context->execution_timeout_sec;
ctx_.is_shutting_down = &interpreter_context->is_shutting_down;
ctx_.is_profile_query = is_profile_query;
ctx_.trigger_context = trigger_context;
}
std::optional<ExecutionContext> PullPlan::Pull(AnyStream *stream, std::optional<int> n,
@@ -727,54 +592,15 @@ std::optional<ExecutionContext> PullPlan::Pull(AnyStream *stream, std::optional<
summary->insert_or_assign("plan_execution_time", execution_time_.count());
cursor_->Shutdown();
ctx_.profile_execution_time = execution_time_;
return ctx_;
}
} // namespace
/**
* Convert a parsed *Cypher* query's AST into a logical plan.
*
* The created logical plan will take ownership of the `AstStorage` within
* `ParsedQuery` and might modify it during planning.
*/
std::unique_ptr<LogicalPlan> MakeLogicalPlan(AstStorage ast_storage, CypherQuery *query, const Parameters &parameters,
DbAccessor *db_accessor) {
auto vertex_counts = plan::MakeVertexCountCache(db_accessor);
auto symbol_table = MakeSymbolTable(query);
auto planning_context = plan::MakePlanningContext(&ast_storage, &symbol_table, query, &vertex_counts);
std::unique_ptr<plan::LogicalOperator> root;
double cost;
std::tie(root, cost) = plan::MakeLogicalPlan(&planning_context, parameters, FLAGS_query_cost_planner);
return std::make_unique<SingleNodeLogicalPlan>(std::move(root), cost, std::move(ast_storage),
std::move(symbol_table));
}
/**
* Return the parsed *Cypher* query's AST cached logical plan, or create and
* cache a fresh one if it doesn't yet exist.
*/
std::shared_ptr<CachedPlan> CypherQueryToPlan(uint64_t hash, AstStorage ast_storage, CypherQuery *query,
const Parameters &parameters, utils::SkipList<PlanCacheEntry> *plan_cache,
DbAccessor *db_accessor, const bool is_cacheable = true) {
auto plan_cache_access = plan_cache->access();
auto it = plan_cache_access.find(hash);
if (it != plan_cache_access.end()) {
if (it->second->IsExpired()) {
plan_cache_access.remove(hash);
} else {
return it->second;
}
}
auto plan = std::make_shared<CachedPlan>(MakeLogicalPlan(std::move(ast_storage), (query), parameters, db_accessor));
if (is_cacheable) {
plan_cache_access.insert({hash, plan});
}
return plan;
return std::move(ctx_);
}
using RWType = plan::ReadWriteTypeChecker::RWType;
} // namespace
Interpreter::Interpreter(InterpreterContext *interpreter_context) : interpreter_context_(interpreter_context) {
MG_ASSERT(interpreter_context_, "Interpreter context must not be NULL");
}
PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper) {
std::function<void()> handler;
@@ -787,8 +613,8 @@ PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper)
in_explicit_transaction_ = true;
expect_rollback_ = false;
db_accessor_.emplace(interpreter_context_->db->Access());
execution_db_accessor_.emplace(&*db_accessor_);
db_accessor_ = std::make_unique<storage::Storage::Accessor>(interpreter_context_->db->Access());
execution_db_accessor_.emplace(db_accessor_.get());
};
} else if (query_upper == "COMMIT") {
handler = [this] {
@@ -835,7 +661,8 @@ PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper)
PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string, TypedValue> *summary,
InterpreterContext *interpreter_context, DbAccessor *dba,
utils::MonotonicBufferResource *execution_memory) {
utils::MonotonicBufferResource *execution_memory,
TriggerContext *trigger_context = nullptr) {
auto *cypher_query = utils::Downcast<CypherQuery>(parsed_query.query);
Frame frame(0);
@@ -872,7 +699,7 @@ PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string,
}
auto pull_plan = std::make_shared<PullPlan>(plan, parsed_query.parameters, false, dba, interpreter_context,
execution_memory, memory_limit);
execution_memory, trigger_context, memory_limit);
return PreparedQuery{std::move(header), std::move(parsed_query.required_privileges),
[pull_plan = std::move(pull_plan), output_symbols = std::move(output_symbols), summary](
AnyStream *stream, std::optional<int> n) -> std::optional<QueryHandlerResult> {
@@ -997,7 +824,7 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
AnyStream *stream, std::optional<int> n) mutable -> std::optional<QueryHandlerResult> {
// No output symbols are given so that nothing is streamed.
if (!ctx) {
ctx = PullPlan(plan, parameters, true, dba, interpreter_context, execution_memory, memory_limit)
ctx = PullPlan(plan, parameters, true, dba, interpreter_context, execution_memory, nullptr, memory_limit)
.Pull(stream, {}, {}, summary);
pull_plan = std::make_shared<PullPlanVector>(ProfilingStatsToTable(ctx->stats, ctx->profile_execution_time));
}
@@ -1500,16 +1327,22 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
if (!in_explicit_transaction_ &&
(utils::Downcast<CypherQuery>(parsed_query.query) || utils::Downcast<ExplainQuery>(parsed_query.query) ||
utils::Downcast<ProfileQuery>(parsed_query.query) || utils::Downcast<DumpQuery>(parsed_query.query))) {
db_accessor_.emplace(interpreter_context_->db->Access());
execution_db_accessor_.emplace(&*db_accessor_);
db_accessor_ = std::make_unique<storage::Storage::Accessor>(interpreter_context_->db->Access());
execution_db_accessor_.emplace(db_accessor_.get());
}
utils::Timer planning_timer;
PreparedQuery prepared_query;
if (utils::Downcast<CypherQuery>(parsed_query.query)) {
if (interpreter_context_->before_commit_triggers.size() > 0 ||
interpreter_context_->after_commit_triggers.size() > 0) {
trigger_context_.emplace();
}
prepared_query = PrepareCypherQuery(std::move(parsed_query), &query_execution->summary, interpreter_context_,
&*execution_db_accessor_, &query_execution->execution_memory);
&*execution_db_accessor_, &query_execution->execution_memory,
trigger_context_ ? &*trigger_context_ : nullptr);
} else if (utils::Downcast<ExplainQuery>(parsed_query.query)) {
prepared_query = PrepareExplainQuery(std::move(parsed_query), &query_execution->summary, interpreter_context_,
&*execution_db_accessor_, &query_execution->execution_memory);
@@ -1576,9 +1409,60 @@ void Interpreter::Abort() {
if (!db_accessor_) return;
db_accessor_->Abort();
execution_db_accessor_ = std::nullopt;
db_accessor_ = std::nullopt;
db_accessor_.reset();
trigger_context_.reset();
}
namespace {
void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, InterpreterContext *interpreter_context,
TriggerContext trigger_context) {
// Run the triggers
for (const auto &trigger : triggers.access()) {
spdlog::debug("Executing trigger '{}'", trigger.name());
utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize};
// create a new transaction for each trigger
auto storage_acc = interpreter_context->db->Access();
DbAccessor db_accessor{&storage_acc};
trigger_context.AdaptForAccessor(&db_accessor);
try {
trigger.Execute(&interpreter_context->plan_cache, &db_accessor, &execution_memory,
*interpreter_context->tsc_frequency, interpreter_context->execution_timeout_sec,
&interpreter_context->is_shutting_down, trigger_context);
} catch (const utils::BasicException &exception) {
spdlog::warn("Trigger '{}' failed with exception:\n{}", trigger.name(), exception.what());
db_accessor.Abort();
continue;
}
auto maybe_constraint_violation = db_accessor.Commit();
if (maybe_constraint_violation.HasError()) {
const auto &constraint_violation = maybe_constraint_violation.GetError();
switch (constraint_violation.type) {
case storage::ConstraintViolation::Type::EXISTENCE: {
const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
MG_ASSERT(constraint_violation.properties.size() == 1U);
const auto &property_name = db_accessor.PropertyToName(*constraint_violation.properties.begin());
spdlog::warn("Trigger '{}' failed to commit due to existence constraint violation on :{}({})", trigger.name(),
label_name, property_name);
break;
}
case storage::ConstraintViolation::Type::UNIQUE: {
const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
std::stringstream property_names_stream;
utils::PrintIterable(property_names_stream, constraint_violation.properties, ", ",
[&](auto &stream, const auto &prop) { stream << db_accessor.PropertyToName(prop); });
spdlog::warn("Trigger '{}' failed to commit due to unique constraint violation on :{}({})", trigger.name(),
label_name, property_names_stream.str());
break;
}
}
}
}
}
} // namespace
void Interpreter::Commit() {
// It's possible that some queries did not finish because the user did
// not pull all of the results from the query.
@@ -1586,6 +1470,19 @@ void Interpreter::Commit() {
// We should document clearly that all results should be pulled to complete
// a query.
if (!db_accessor_) return;
if (trigger_context_) {
// Run the triggers
for (const auto &trigger : interpreter_context_->before_commit_triggers.access()) {
spdlog::debug("Executing trigger '{}'", trigger.name());
utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize};
trigger.Execute(&interpreter_context_->plan_cache, &*execution_db_accessor_, &execution_memory,
*interpreter_context_->tsc_frequency, interpreter_context_->execution_timeout_sec,
&interpreter_context_->is_shutting_down, *trigger_context_);
}
SPDLOG_DEBUG("Finished executing before commit triggers");
}
auto maybe_constraint_violation = db_accessor_->Commit();
if (maybe_constraint_violation.HasError()) {
const auto &constraint_violation = maybe_constraint_violation.GetError();
@@ -1595,7 +1492,8 @@ void Interpreter::Commit() {
MG_ASSERT(constraint_violation.properties.size() == 1U);
auto property_name = execution_db_accessor_->PropertyToName(*constraint_violation.properties.begin());
execution_db_accessor_ = std::nullopt;
db_accessor_ = std::nullopt;
db_accessor_.reset();
trigger_context_.reset();
throw QueryException("Unable to commit due to existence constraint violation on :{}({})", label_name,
property_name);
break;
@@ -1607,15 +1505,31 @@ void Interpreter::Commit() {
property_names_stream, constraint_violation.properties, ", ",
[this](auto &stream, const auto &prop) { stream << execution_db_accessor_->PropertyToName(prop); });
execution_db_accessor_ = std::nullopt;
db_accessor_ = std::nullopt;
db_accessor_.reset();
trigger_context_.reset();
throw QueryException("Unable to commit due to unique constraint violation on :{}({})", label_name,
property_names_stream.str());
break;
}
}
}
if (trigger_context_) {
background_thread_.AddTask([trigger_context = std::move(*trigger_context_),
interpreter_context = this->interpreter_context_,
user_transaction = std::shared_ptr(std::move(db_accessor_))]() mutable {
RunTriggersIndividually(interpreter_context->after_commit_triggers, interpreter_context,
std::move(trigger_context));
user_transaction->FinalizeTransaction();
SPDLOG_DEBUG("Finished executing after commit triggers"); // NOLINT(bugprone-lambda-function-name)
});
}
execution_db_accessor_ = std::nullopt;
db_accessor_ = std::nullopt;
db_accessor_.reset();
trigger_context_.reset();
SPDLOG_DEBUG("Finished comitting the transaction");
}
void Interpreter::AdvanceCommand() {

View File

@@ -3,6 +3,7 @@
#include <gflags/gflags.h>
#include "query/context.hpp"
#include "query/cypher_query_interpreter.hpp"
#include "query/db_accessor.hpp"
#include "query/exceptions.hpp"
#include "query/frontend/ast/ast.hpp"
@@ -12,18 +13,17 @@
#include "query/plan/operator.hpp"
#include "query/plan/read_write_type_checker.hpp"
#include "query/stream.hpp"
#include "query/trigger.hpp"
#include "query/typed_value.hpp"
#include "utils/event_counter.hpp"
#include "utils/logging.hpp"
#include "utils/memory.hpp"
#include "utils/skip_list.hpp"
#include "utils/spin_lock.hpp"
#include "utils/thread_pool.hpp"
#include "utils/timer.hpp"
#include "utils/tsc.hpp"
DECLARE_bool(query_cost_planner);
DECLARE_int32(query_plan_cache_ttl);
namespace EventCounter {
extern const Event FailedQuery;
} // namespace EventCounter
@@ -139,64 +139,6 @@ struct PreparedQuery {
plan::ReadWriteTypeChecker::RWType rw_type;
};
// TODO: Maybe this should move to query/plan/planner.
/// Interface for accessing the root operator of a logical plan.
class LogicalPlan {
public:
virtual ~LogicalPlan() {}
virtual const plan::LogicalOperator &GetRoot() const = 0;
virtual double GetCost() const = 0;
virtual const SymbolTable &GetSymbolTable() const = 0;
virtual const AstStorage &GetAstStorage() const = 0;
};
class CachedPlan {
public:
explicit CachedPlan(std::unique_ptr<LogicalPlan> plan);
const auto &plan() const { return plan_->GetRoot(); }
double cost() const { return plan_->GetCost(); }
const auto &symbol_table() const { return plan_->GetSymbolTable(); }
const auto &ast_storage() const { return plan_->GetAstStorage(); }
bool IsExpired() const { return cache_timer_.Elapsed() > std::chrono::seconds(FLAGS_query_plan_cache_ttl); };
private:
std::unique_ptr<LogicalPlan> plan_;
utils::Timer cache_timer_;
};
struct CachedQuery {
AstStorage ast_storage;
Query *query;
std::vector<AuthQuery::Privilege> required_privileges;
};
struct QueryCacheEntry {
bool operator==(const QueryCacheEntry &other) const { return first == other.first; }
bool operator<(const QueryCacheEntry &other) const { return first < other.first; }
bool operator==(const uint64_t &other) const { return first == other; }
bool operator<(const uint64_t &other) const { return first < other; }
uint64_t first;
// TODO: Maybe store the query string here and use it as a key with the hash
// so that we eliminate the risk of hash collisions.
CachedQuery second;
};
struct PlanCacheEntry {
bool operator==(const PlanCacheEntry &other) const { return first == other.first; }
bool operator<(const PlanCacheEntry &other) const { return first < other.first; }
bool operator==(const uint64_t &other) const { return first == other; }
bool operator<(const uint64_t &other) const { return first < other; }
uint64_t first;
// TODO: Maybe store the query string here and use it as a key with the hash
// so that we eliminate the risk of hash collisions.
std::shared_ptr<CachedPlan> second;
};
/**
* Holds data shared between multiple `Interpreter` instances (which might be
* running concurrently).
@@ -205,7 +147,28 @@ struct PlanCacheEntry {
* been passed to an `Interpreter` instance.
*/
struct InterpreterContext {
explicit InterpreterContext(storage::Storage *db) : db(db) {}
explicit InterpreterContext(storage::Storage *db) : db(db) {
// try {
// {
// auto storage_acc = db->Access();
// DbAccessor dba(&storage_acc);
// auto triggers_acc = before_commit_triggers.access();
// triggers_acc.insert(Trigger{"BeforeCreator", "UNWIND createdVertices as u SET u.before = u.id + 1",
// &ast_cache,
// &plan_cache, &dba, &antlr_lock});
// }
// {
// auto storage_acc = db->Access();
// DbAccessor dba(&storage_acc);
// auto triggers_acc = after_commit_triggers.access();
// triggers_acc.insert(Trigger{"AfterCreator", "UNWIND createdVertices as u SET u.after = u.id - 1",
// &ast_cache,
// &plan_cache, &dba, &antlr_lock});
// }
// } catch (const utils::BasicException &e) {
// spdlog::critical("Failed to create a trigger because: {}", e.what());
// }
}
storage::Storage *db;
@@ -225,6 +188,10 @@ struct InterpreterContext {
utils::SkipList<QueryCacheEntry> ast_cache;
utils::SkipList<PlanCacheEntry> plan_cache;
// use a thread safe container
utils::SkipList<Trigger> before_commit_triggers;
utils::SkipList<Trigger> after_commit_triggers;
};
/// Function that is used to tell all active interpreters that they should stop
@@ -350,11 +317,14 @@ class Interpreter final {
InterpreterContext *interpreter_context_;
std::optional<storage::Storage::Accessor> db_accessor_;
std::unique_ptr<storage::Storage::Accessor> db_accessor_;
std::optional<DbAccessor> execution_db_accessor_;
std::optional<TriggerContext> trigger_context_;
bool in_explicit_transaction_{false};
bool expect_rollback_{false};
utils::ThreadPool background_thread_{1};
PreparedQuery PrepareTransactionQuery(std::string_view query_upper);
void Commit();
void AdvanceCommand();

View File

@@ -206,7 +206,10 @@ bool CreateNode::CreateNodeCursor::Pull(Frame &frame, ExecutionContext &context)
SCOPED_PROFILE_OP("CreateNode");
if (input_cursor_->Pull(frame, context)) {
CreateLocalVertex(self_.node_info_, &frame, context);
auto created_vertex = CreateLocalVertex(self_.node_info_, &frame, context);
if (context.trigger_context) {
context.trigger_context->RegisterCreatedVertex(created_vertex);
}
return true;
}

122
src/query/trigger.cpp Normal file
View File

@@ -0,0 +1,122 @@
#include "query/trigger.hpp"
#include "query/context.hpp"
#include "query/cypher_query_interpreter.hpp"
#include "query/db_accessor.hpp"
#include "query/frontend/ast/ast.hpp"
#include "query/interpret/frame.hpp"
#include "utils/memory.hpp"
namespace query {
namespace {
std::vector<std::pair<Identifier, trigger::IdentifierTag>> GetPredefinedIdentifiers() {
return {{{"createdVertices", false}, trigger::IdentifierTag::CREATED_VERTICES}};
}
} // namespace
void TriggerContext::RegisterCreatedVertex(const VertexAccessor created_vertex) {
created_vertices_.push_back(created_vertex);
}
TypedValue TriggerContext::GetTypedValue(const trigger::IdentifierTag tag) const {
switch (tag) {
case trigger::IdentifierTag::CREATED_VERTICES: {
std::vector<TypedValue> typed_created_vertices;
typed_created_vertices.reserve(created_vertices_.size());
std::transform(std::begin(created_vertices_), std::end(created_vertices_),
std::back_inserter(typed_created_vertices),
[](const auto &accessor) { return TypedValue(accessor); });
return TypedValue(typed_created_vertices);
}
}
}
void TriggerContext::AdaptForAccessor(DbAccessor *accessor) {
// adapt created_vertices_
auto it = created_vertices_.begin();
for (const auto &created_vertex : created_vertices_) {
if (auto maybe_vertex = accessor->FindVertex(created_vertex.Gid(), storage::View::OLD); maybe_vertex) {
*it = *maybe_vertex;
++it;
}
}
created_vertices_.erase(it, created_vertices_.end());
}
Trigger::Trigger(std::string name, std::string query, utils::SkipList<QueryCacheEntry> *query_cache,
utils::SkipList<PlanCacheEntry> *plan_cache, DbAccessor *db_accessor, utils::SpinLock *antlr_lock)
: name_(std::move(name)),
parsed_statements_{ParseQuery(query, {}, query_cache, antlr_lock)},
identifiers_{GetPredefinedIdentifiers()} {
GetPlan(plan_cache, db_accessor);
}
std::shared_ptr<CachedPlan> Trigger::GetPlan(utils::SkipList<PlanCacheEntry> *plan_cache,
DbAccessor *db_accessor) const {
AstStorage ast_storage;
ast_storage.properties_ = parsed_statements_.ast_storage.properties_;
ast_storage.labels_ = parsed_statements_.ast_storage.labels_;
ast_storage.edge_types_ = parsed_statements_.ast_storage.edge_types_;
std::unordered_map<std::string, Identifier *> predefined_identifiers;
for (auto &[identifier, tag] : identifiers_) {
predefined_identifiers.emplace(identifier.name_, &identifier);
}
return CypherQueryToPlan(parsed_statements_.stripped_query.hash(), std::move(ast_storage),
utils::Downcast<CypherQuery>(parsed_statements_.query), parsed_statements_.parameters,
plan_cache, db_accessor, parsed_statements_.is_cacheable, std::move(predefined_identifiers));
}
void Trigger::Execute(utils::SkipList<PlanCacheEntry> *plan_cache, DbAccessor *dba,
utils::MonotonicBufferResource *execution_memory, const double tsc_frequency,
const double max_execution_time_sec, std::atomic<bool> *is_shutting_down,
const TriggerContext &context) const {
auto plan = GetPlan(plan_cache, dba);
ExecutionContext ctx;
ctx.db_accessor = dba;
ctx.symbol_table = plan->symbol_table();
ctx.evaluation_context.timestamp =
std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch())
.count();
ctx.evaluation_context.parameters = parsed_statements_.parameters;
ctx.evaluation_context.properties = NamesToProperties(plan->ast_storage().properties_, dba);
ctx.evaluation_context.labels = NamesToLabels(plan->ast_storage().labels_, dba);
ctx.execution_tsc_timer = utils::TSCTimer(tsc_frequency);
ctx.max_execution_time_sec = max_execution_time_sec;
ctx.is_shutting_down = is_shutting_down;
ctx.is_profile_query = false;
// 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`.
constexpr size_t stack_size = 256 * 1024;
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);
Frame frame{plan->symbol_table().max_position(), execution_memory};
for (const auto &[identifier, tag] : identifiers_) {
if (identifier.symbol_pos_ == -1) {
continue;
}
frame[plan->symbol_table().at(identifier)] = context.GetTypedValue(tag);
}
while (cursor->Pull(frame, ctx))
;
cursor->Shutdown();
}
} // namespace query

52
src/query/trigger.hpp Normal file
View File

@@ -0,0 +1,52 @@
#pragma once
#include "query/cypher_query_interpreter.hpp"
#include "query/db_accessor.hpp"
#include "query/frontend/ast/ast.hpp"
namespace query {
namespace trigger {
enum class IdentifierTag : uint8_t { CREATED_VERTICES };
} // namespace trigger
struct TriggerContext {
void RegisterCreatedVertex(VertexAccessor created_vertex);
// Adapt the TriggerContext object inplace for a different DbAccessor
// (each dirived accessor, e.g. VertexAccessor, gets adapted
// to the sent DbAccessor so they can be used safely)
void AdaptForAccessor(DbAccessor *accessor);
TypedValue GetTypedValue(trigger::IdentifierTag tag) const;
private:
std::vector<VertexAccessor> created_vertices_;
};
struct Trigger {
explicit Trigger(std::string name, std::string query, utils::SkipList<QueryCacheEntry> *query_cache,
utils::SkipList<PlanCacheEntry> *plan_cache, DbAccessor *db_accessor, utils::SpinLock *antlr_lock);
void Execute(utils::SkipList<PlanCacheEntry> *plan_cache, DbAccessor *dba,
utils::MonotonicBufferResource *execution_memory, double tsc_frequency, double max_execution_time_sec,
std::atomic<bool> *is_shutting_down, const TriggerContext &context) const;
bool operator==(const Trigger &other) const { return name_ == other.name_; }
// NOLINTNEXTLINE (modernize-use-nullptr)
bool operator<(const Trigger &other) const { return name_ < other.name_; }
bool operator==(const std::string &other) const { return name_ == other; }
// NOLINTNEXTLINE (modernize-use-nullptr)
bool operator<(const std::string &other) const { return name_ < other; }
const auto &name() const noexcept { return name_; }
private:
std::shared_ptr<CachedPlan> GetPlan(utils::SkipList<PlanCacheEntry> *plan_cache, DbAccessor *db_accessor) const;
std::string name_;
ParsedQuery parsed_statements_;
mutable std::vector<std::pair<Identifier, trigger::IdentifierTag>> identifiers_;
};
} // namespace query

View File

@@ -404,17 +404,22 @@ Storage::Accessor::Accessor(Storage *storage)
Storage::Accessor::Accessor(Accessor &&other) noexcept
: storage_(other.storage_),
storage_guard_(std::move(other.storage_guard_)),
transaction_(std::move(other.transaction_)),
commit_timestamp_(other.commit_timestamp_),
is_transaction_active_(other.is_transaction_active_),
config_(other.config_) {
// Don't allow the other accessor to abort our transaction in destructor.
other.is_transaction_active_ = false;
other.commit_timestamp_.reset();
}
Storage::Accessor::~Accessor() {
if (is_transaction_active_) {
Abort();
}
FinalizeTransaction();
}
VertexAccessor Storage::Accessor::CreateVertex() {
@@ -793,11 +798,10 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
// Save these so we can mark them used in the commit log.
uint64_t start_timestamp = transaction_.start_timestamp;
uint64_t commit_timestamp;
{
std::unique_lock<utils::SpinLock> engine_guard(storage_->engine_lock_);
commit_timestamp = storage_->CommitTimestamp(desired_commit_timestamp);
commit_timestamp_.emplace(storage_->CommitTimestamp(desired_commit_timestamp));
// Before committing and validating vertices against unique constraints,
// we have to update unique constraints with the vertices that are going
@@ -821,7 +825,7 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
// No need to take any locks here because we modified this vertex and no
// one else can touch it until we commit.
unique_constraint_violation =
storage_->constraints_.unique_constraints.Validate(*prev.vertex, transaction_, commit_timestamp);
storage_->constraints_.unique_constraints.Validate(*prev.vertex, transaction_, *commit_timestamp_);
if (unique_constraint_violation) {
break;
}
@@ -838,7 +842,7 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
// Replica can log only the write transaction received from Main
// so the Wal files are consistent
if (storage_->replication_role_ == ReplicationRole::MAIN || desired_commit_timestamp.has_value()) {
storage_->AppendToWal(transaction_, commit_timestamp);
storage_->AppendToWal(transaction_, *commit_timestamp_);
}
// Take committed_transactions lock while holding the engine lock to
@@ -848,12 +852,12 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
// TODO: release lock, and update all deltas to have a local copy
// of the commit timestamp
MG_ASSERT(transaction_.commit_timestamp != nullptr, "Invalid database state!");
transaction_.commit_timestamp->store(commit_timestamp, std::memory_order_release);
transaction_.commit_timestamp->store(*commit_timestamp_, std::memory_order_release);
// Replica can only update the last commit timestamp with
// the commits received from main.
if (storage_->replication_role_ == ReplicationRole::MAIN || desired_commit_timestamp.has_value()) {
// Update the last commit timestamp
storage_->last_commit_timestamp_.store(commit_timestamp);
storage_->last_commit_timestamp_.store(*commit_timestamp_);
}
// Release engine lock because we don't have to hold it anymore
// and emplace back could take a long time.
@@ -862,13 +866,11 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
});
storage_->commit_log_->MarkFinished(start_timestamp);
storage_->commit_log_->MarkFinished(commit_timestamp);
}
}
if (unique_constraint_violation) {
Abort();
storage_->commit_log_->MarkFinished(commit_timestamp);
return *unique_constraint_violation;
}
}
@@ -1041,6 +1043,13 @@ void Storage::Accessor::Abort() {
is_transaction_active_ = false;
}
void Storage::Accessor::FinalizeTransaction() {
if (commit_timestamp_) {
storage_->commit_log_->MarkFinished(*commit_timestamp_);
commit_timestamp_.reset();
}
}
const std::string &Storage::LabelToName(LabelId label) const { return name_id_mapper_.IdToName(label.AsUint()); }
const std::string &Storage::PropertyToName(PropertyId property) const {

View File

@@ -300,6 +300,8 @@ class Storage final {
/// @throw std::bad_alloc
void Abort();
void FinalizeTransaction();
private:
/// @throw std::bad_alloc
VertexAccessor CreateVertex(storage::Gid gid);
@@ -310,6 +312,7 @@ class Storage final {
Storage *storage_;
std::shared_lock<utils::RWLock> storage_guard_;
Transaction transaction_;
std::optional<uint64_t> commit_timestamp_;
bool is_transaction_active_;
Config::Items config_;
};

View File

@@ -3,6 +3,7 @@
#include "gtest/gtest.h"
#include "query/exceptions.hpp"
#include "query/frontend/ast/ast.hpp"
#include "query/frontend/semantic/symbol_generator.hpp"
#include "query/frontend/semantic/symbol_table.hpp"
@@ -1093,3 +1094,54 @@ TEST(TestSymbolTable, CreateAnonymousSymbolWithExistingUserSymbolCalledAnon) {
auto anon2 = symbol_table.CreateAnonymousSymbol();
ASSERT_EQ(anon2.name_, "anon2");
}
TEST_F(TestSymbolGenerator, PredefinedIdentifiers) {
auto *first_op = IDENT("first_op");
auto *second_op = IDENT("second_op");
// RETURN first_op + second_op AS result
auto query = QUERY(SINGLE_QUERY(RETURN(ADD(first_op, second_op), AS("result"))));
EXPECT_THROW(query::MakeSymbolTable(query), SemanticException);
EXPECT_THROW(query::MakeSymbolTable(query, {{first_op->name_, first_op}}), SemanticException);
EXPECT_THROW(query::MakeSymbolTable(query, {{second_op->name_, second_op}}), SemanticException);
auto symbol_table = query::MakeSymbolTable(query, {{first_op->name_, first_op}, {second_op->name_, second_op}});
ASSERT_EQ(symbol_table.max_position(), 3);
// predefined identifier can only be used in one scope
// RETURN first_op + second_op AS result UNION RETURN second_op + first_op AS result
query = QUERY(SINGLE_QUERY(RETURN(ADD(first_op, second_op), AS("result"))),
UNION(SINGLE_QUERY(RETURN(ADD(second_op, first_op), AS("result")))));
ASSERT_THROW(query::MakeSymbolTable(query, {{first_op->name_, first_op}, {second_op->name_, second_op}}),
SemanticException);
// predefined identifier can be introduced in any of the scope
// different predefined identifiers can be introduced in different scopes
// RETURN first_op AS result UNION RETURN second_op AS result
query = QUERY(SINGLE_QUERY(RETURN(first_op, AS("result"))), UNION(SINGLE_QUERY(RETURN(second_op, AS("result")))));
ASSERT_THROW(query::MakeSymbolTable(query), SemanticException);
symbol_table = query::MakeSymbolTable(query, {{first_op->name_, first_op}, {second_op->name_, second_op}});
ASSERT_EQ(symbol_table.max_position(), 5);
// WITH statement resets the scope, but the predefined identifier is okay
// because it's the first introduction of it in the query
// WITH 1 as one RETURN first_op AS first
query = QUERY(SINGLE_QUERY(WITH(LITERAL(1), AS("one")), RETURN(first_op, AS("first"))));
ASSERT_THROW(query::MakeSymbolTable(query), SemanticException);
symbol_table = query::MakeSymbolTable(query, {{first_op->name_, first_op}});
ASSERT_EQ(symbol_table.max_position(), 3);
// In the first scope, first_op represents identifier created by match,
// in the second it represent the predefined identifier
// MATCH(first_op) WITH first_op as n RETURN first_op, n
query = QUERY(SINGLE_QUERY(MATCH(PATTERN(NODE("first_op"))), WITH("first_op", AS("n")), RETURN("first_op", "n")));
ASSERT_THROW(query::MakeSymbolTable(query), SemanticException);
symbol_table = query::MakeSymbolTable(query, {{first_op->name_, first_op}});
ASSERT_EQ(symbol_table.max_position(), 6);
// You cannot redaclare the predefined identifier in the same scope
// UNWIND first_op as u CREATE(first_op {prop: u})
auto unwind = UNWIND(first_op, AS("u"));
auto node = NODE("first_op");
node->properties_[storage.GetPropertyIx("prop")] = dynamic_cast<Identifier *>(unwind->named_expression_->expression_);
query = QUERY(SINGLE_QUERY(unwind, CREATE(PATTERN(node))));
ASSERT_THROW(query::MakeSymbolTable(query, {{first_op->name_, first_op}}), SemanticException);
}

View File

@@ -216,6 +216,7 @@ def main():
start_workers(max_task_count, run_tidy, task_queue, lock, args.timeout)
# Form the common args list.
print(f"BUILD PATH {os.path.abspath(args.build_path)}")
common_clang_tidy_args = []
if args.fix:
common_clang_tidy_args.append('-fix')
@@ -224,6 +225,8 @@ def main():
if args.quiet:
common_clang_tidy_args.append('-quiet')
if args.build_path is not None:
for f in os.listdir(args.build_path):
print(f"FILE IN BUILD {f}")
common_clang_tidy_args.append('-p=%s' % args.build_path)
for arg in args.extra_arg:
common_clang_tidy_args.append('-extra-arg=%s' % arg)