Compare commits
39 Commits
upgrade-di
...
MG-test-an
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
47c82c7d65 | ||
|
|
686c1bbc1a | ||
|
|
ccb609be5b | ||
|
|
860a1a0ea6 | ||
|
|
c79e843de6 | ||
|
|
805bab80d3 | ||
|
|
33216920f7 | ||
|
|
42cfd2be28 | ||
|
|
012a02dc93 | ||
|
|
b0e7b08c85 | ||
|
|
15318ef4f4 | ||
|
|
6cff70b8e8 | ||
|
|
c8ae33de89 | ||
|
|
caac055f80 | ||
|
|
846403f727 | ||
|
|
96acf2e9d9 | ||
|
|
912e20dd07 | ||
|
|
a812a256cf | ||
|
|
ca7d3bfde0 | ||
|
|
2e2d2c1a8f | ||
|
|
cf3ebf1b47 | ||
|
|
24b419db8b | ||
|
|
4397b2db49 | ||
|
|
4d96a234d7 | ||
|
|
609c9c2d50 | ||
|
|
3d6112b4fd | ||
|
|
7812af37cb | ||
|
|
575231e15f | ||
|
|
565e2b3dd4 | ||
|
|
0f9f3523f5 | ||
|
|
73b7ffda10 | ||
|
|
c9b4361f7e | ||
|
|
bd8e96ed75 | ||
|
|
ee9badf828 | ||
|
|
8cca35daeb | ||
|
|
e6566b3655 | ||
|
|
82e4e485f8 | ||
|
|
e760a89eb7 | ||
|
|
613f0b7ce7 |
307
.github/workflows/diff.yaml
vendored
307
.github/workflows/diff.yaml
vendored
@@ -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
|
||||
|
||||
@@ -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)`.
|
||||
|
||||
@@ -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})
|
||||
|
||||
@@ -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) {
|
||||
|
||||
140
src/query/cypher_query_interpreter.cpp
Normal file
140
src/query/cypher_query_interpreter.cpp
Normal 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> ¶ms,
|
||||
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 ¶m_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 ¶meters,
|
||||
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 ¶meters, 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
|
||||
141
src/query/cypher_query_interpreter.hpp
Normal file
141
src/query/cypher_query_interpreter.hpp
Normal 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> ¶ms,
|
||||
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 ¶meters,
|
||||
utils::SkipList<PlanCacheEntry> *plan_cache, DbAccessor *db_accessor, bool is_cacheable = true,
|
||||
std::unordered_map<std::string, Identifier *> predefined_identifiers = {});
|
||||
|
||||
} // namespace query
|
||||
@@ -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) {
|
||||
|
||||
@@ -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) {}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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> ¶ms,
|
||||
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 ¶m_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 ¶meters, 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 ¶meters, 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 ¶meters,
|
||||
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 ¶meters, 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() {
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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
122
src/query/trigger.cpp
Normal 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
52
src/query/trigger.hpp
Normal 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
|
||||
@@ -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 {
|
||||
|
||||
@@ -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_;
|
||||
};
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user