Compare commits

...

32 Commits

Author SHA1 Message Date
antejavor
c8719c20df Run on docker release version. 2023-06-21 13:15:39 +02:00
antejavor
de9aa1fd56 Decrease query quantity. 2023-06-19 14:47:53 +02:00
antejavor
1c55f9c5bd Add query module writes and reads. 2023-06-19 14:45:57 +02:00
antejavor
c7fb5016d3 Update running script. 2023-06-16 14:22:45 +02:00
antejavor
f9ce2a4be6 Init replication script. 2023-06-16 14:19:23 +02:00
Marko Budiselić
cf1a86ed13 Refactor tests/integration/run.sh (#1016) 2023-06-15 23:10:52 +02:00
Marko Budiselić
7fb3f62703 Upgrade to RocksDB 8.1.1 (#1013) 2023-06-15 11:54:24 +02:00
Marko Budiselić
cb4b71bdbd Update pull_request_template.md 2023-06-14 16:04:35 +02:00
andrejtonev
30ec570bb9 Add Bolt v5 support (#938) 2023-06-12 18:55:15 +02:00
Antonio Filipovic
d917c3f0fd Fix slow IN LIST evaluation (#901) 2023-05-29 17:52:20 +02:00
andrejtonev
d842adbed3 Handle user-defined metadata and expose it with SHOW TRANSACTIONS(#945) 2023-05-29 11:40:14 +02:00
Bruno Sačarić
cdfcbc106c Update license date (#941) 2023-05-18 11:42:12 +02:00
Josipmrden
651b6f3a5a Expose system metrics over HTTP Endpoint (#940) 2023-05-18 05:10:57 +00:00
Ante Pušić
0d9bd74a8a Add support for map projection (#892) 2023-05-16 20:05:35 +02:00
andrejtonev
802f8aceda Add data directory status and (un)lock query (#933) 2023-05-16 18:36:04 +02:00
gvolfing
7ddce539fa Add return build type command (#894) 2023-05-16 16:02:03 +02:00
gvolfing
c3e4f81026 Include additional info inside storage mode info query (#883) 2023-05-16 14:25:41 +02:00
Antonio Filipovic
208705f296 Reduce memory consumption on return from python procedures (#932) 2023-05-16 10:33:09 +02:00
Ante Javor
69634a5354 Fix typo in mgbench 2023-05-10 14:02:46 +02:00
Aidar Samerkhanov
b8f282468d Update pulsar client for e2e tests 2023-05-09 12:23:28 +02:00
Ante Javor
ab38161cd2 FIix methodology links (#903)
Co-authored-by: Josip Mrden <josip.mrden@memgraph.io>
2023-05-03 16:37:36 +02:00
János Benjamin Antal
3a5f140c2b Order chunks in utils::Pool to speed up deallocation (#898) 2023-05-02 13:08:20 +02:00
Ante Javor
eead0f79fc Fix missing argument in daily benchmark (#907) 2023-05-02 11:00:44 +02:00
Antonio Filipovic
91017b7f36 Update profile query to use PoolResource for LOAD CSV (#885) 2023-04-26 18:04:13 +02:00
gvolfing
00f8d54249 Parallelize index creation (#882) 2023-04-26 16:28:02 +02:00
János Benjamin Antal
4fcdd52f88 Use correct memory resource (#900) 2023-04-26 10:02:55 +02:00
János Benjamin Antal
6c947947eb Parallelize recovery (#868)
* Parallelize edge recovery

* Load vertex labels and properties parallel

* Add parallel connectivity loading

* Add batches information to snapshot

* Introduce `items_per_batch` and `recovery_thread_count` flags

* Make possible to load snapshots with old version

* Add vertex batches to `RecoveryInfo`

* Extend durability integration tests with v15 test cases

* Add `std::vector` based `InitProperties`

* Use `InitProperties` in snapshot loading
2023-04-25 16:25:25 +02:00
Ante Javor
64fd281b2e Update benchgraph methodology (#899) 2023-04-25 09:45:25 +02:00
János Benjamin Antal
97e250129e Change AccumulateCursor to use utils::pmr::deque (#888)
* Increase performance by eliminating unnecessary `TypedValue` copies
2023-04-24 16:22:22 +02:00
Marko Budiselić
b02b201129 Improve Jepsen setup (#893) 2023-04-23 16:16:49 +02:00
Antonio Filipovic
2c6a55775d Fix max block size bug on LOAD CSV(#877) 2023-04-19 16:10:20 +02:00
Ante Javor
940bf6722c Add mgbench tutorial (#836)
* Add Docker runner
* Add Docker client
* Add benchgraph.sh script
* Add package script
2023-04-19 08:21:55 +02:00
208 changed files with 8554 additions and 1117 deletions

View File

@@ -1,6 +1,7 @@
---
BasedOnStyle: Google
---
Language: Cpp
BasedOnStyle: Google
Standard: "c++20"
UseTab: Never
DerivePointerAlignment: false

View File

@@ -10,5 +10,5 @@
To keep docs changelog up to date, one more thing to do:
- [ ] Write a release note here
- [ ] Write a release note here, including added/changed clauses
- [ ] Tag someone from docs team in the comments

View File

@@ -67,7 +67,7 @@ jobs:
- name: Run mgbench
run: |
cd tests/mgbench
./benchmark.py --num-workers-for-benchmark 12 --export-results benchmark_result.json pokec/medium/*/*
./benchmark.py vendor-native --num-workers-for-benchmark 12 --export-results benchmark_result.json pokec/medium/*/*
- name: Upload mgbench results
run: |

View File

@@ -196,22 +196,7 @@ jobs:
- 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
tests/integration/run.sh
- name: Run cppcheck and clang-format
run: |
@@ -437,7 +422,7 @@ jobs:
- name: Run mgbench
run: |
cd tests/mgbench
./benchmark.py --num-workers-for-benchmark 12 --export-results benchmark_result.json pokec/medium/*/*
./benchmark.py vendor-native --num-workers-for-benchmark 12 --export-results benchmark_result.json pokec/medium/*/*
- name: Upload mgbench results
run: |

View File

@@ -146,22 +146,7 @@ jobs:
- 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
tests/integration/run.sh
- name: Run cppcheck and clang-format
run: |

View File

@@ -146,22 +146,7 @@ jobs:
- 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
tests/integration/run.sh
- name: Run cppcheck and clang-format
run: |

View File

@@ -0,0 +1,63 @@
name: "Mgbench Bolt Client Publish Docker Image"
on:
workflow_dispatch:
inputs:
version:
description: "Mgbench bolt client version to publish on Dockerhub."
required: true
force_release:
type: boolean
required: false
default: false
jobs:
mgbench_docker_publish:
runs-on: ubuntu-latest
env:
DOCKER_ORGANIZATION_NAME: memgraph
DOCKER_REPOSITORY_NAME: mgbench-client
steps:
- name: Checkout
uses: actions/checkout@v3
- name: Set up QEMU
uses: docker/setup-qemu-action@v2
- name: Set up Docker Buildx
id: buildx
uses: docker/setup-buildx-action@v2
- name: Log in to Docker Hub
uses: docker/login-action@v2
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Check if specified version is already pushed
run: |
EXISTS=$(docker manifest inspect $DOCKER_ORGANIZATION_NAME/$DOCKER_REPOSITORY_NAME:${{ github.event.inputs.version }} > /dev/null; echo $?)
echo $EXISTS
if [[ ${EXISTS} -eq 0 ]]; then
echo 'The specified version has been already released to DockerHub.'
if [[ ${{ github.event.inputs.force_release }} = true ]]; then
echo 'Forcing the release!'
else
echo 'Stopping the release!'
exit 1
fi
else
echo 'All good the specified version has not been release to DockerHub.'
fi
- name: Build & push docker images
run: |
cd tests/mgbench
docker buildx build \
--build-arg TOOLCHAIN_VERSION=toolchain-v4 \
--platform linux/amd64,linux/arm64 \
--tag $DOCKER_ORGANIZATION_NAME/$DOCKER_REPOSITORY_NAME:${{ github.event.inputs.version }} \
--tag $DOCKER_ORGANIZATION_NAME/$DOCKER_REPOSITORY_NAME:latest \
--file Dockerfile.mgbench_client \
--push .

View File

@@ -146,22 +146,7 @@ jobs:
- 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
tests/integration/run.sh
- name: Run cppcheck and clang-format
run: |

View File

@@ -3,6 +3,7 @@ repos:
rev: v4.4.0
hooks:
- id: check-yaml
args: [--allow-multiple-documents]
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: https://github.com/psf/black

View File

@@ -231,6 +231,8 @@ endif()
message(STATUS "CMake build type: ${CMAKE_BUILD_TYPE}")
# -----------------------------------------------------------------------------
add_definitions( -DCMAKE_BUILD_TYPE_NAME="${CMAKE_BUILD_TYPE}")
if (NOT MG_ARCH)
set(MG_ARCH_DESCR "Host architecture to build Memgraph on. Supported values are x86_64, ARM64.")
if (${CMAKE_HOST_SYSTEM_PROCESSOR} MATCHES "aarch64")

View File

@@ -103,6 +103,10 @@ modifications:
value: "true"
override: false
- name: "storage_parallel_index_recovery"
value: "false"
override: true
undocumented:
- "flag_file"
- "also_log_to_stderr"

1
libs/.gitignore vendored
View File

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

13
libs/rocksdb8.1.1.patch Normal file
View File

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

View File

@@ -117,7 +117,7 @@ declare -A primary_urls=(
["mgconsole"]="http://$local_cache_host/git/mgconsole.git"
["spdlog"]="http://$local_cache_host/git/spdlog"
["nlohmann"]="http://$local_cache_host/file/nlohmann/json/4f8fba14066156b73f1189a2b8bd568bde5284c5/single_include/nlohmann/json.hpp"
["neo4j"]="http://$local_cache_host/file/neo4j-community-3.2.3-unix.tar.gz"
["neo4j"]="http://$local_cache_host/file/neo4j-community-5.6.0-unix.tar.gz"
["librdkafka"]="http://$local_cache_host/git/librdkafka.git"
["protobuf"]="http://$local_cache_host/git/protobuf.git"
["pulsar"]="http://$local_cache_host/git/pulsar.git"
@@ -142,7 +142,7 @@ declare -A secondary_urls=(
["mgconsole"]="http://github.com/memgraph/mgconsole.git"
["spdlog"]="https://github.com/gabime/spdlog"
["nlohmann"]="https://raw.githubusercontent.com/nlohmann/json/4f8fba14066156b73f1189a2b8bd568bde5284c5/single_include/nlohmann/json.hpp"
["neo4j"]="https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/neo4j-community-3.2.3-unix.tar.gz"
["neo4j"]="https://dist.neo4j.org/neo4j-community-5.6.0-unix.tar.gz"
["librdkafka"]="https://github.com/edenhill/librdkafka.git"
["protobuf"]="https://github.com/protocolbuffers/protobuf.git"
["pulsar"]="https://github.com/apache/pulsar.git"
@@ -180,9 +180,9 @@ repo_clone_try_double "${primary_urls[libbcrypt]}" "${secondary_urls[libbcrypt]}
# neo4j
file_get_try_double "${primary_urls[neo4j]}" "${secondary_urls[neo4j]}"
tar -xzf neo4j-community-3.2.3-unix.tar.gz
mv neo4j-community-3.2.3 neo4j
rm neo4j-community-3.2.3-unix.tar.gz
tar -xzf neo4j-community-5.6.0-unix.tar.gz
mv neo4j-community-5.6.0 neo4j
rm neo4j-community-5.6.0-unix.tar.gz
# nlohmann json
# We wget header instead of cloning repo since repo is huge (lots of test data).
@@ -192,10 +192,10 @@ cd json
file_get_try_double "${primary_urls[nlohmann]}" "${secondary_urls[nlohmann]}"
cd ..
rocksdb_tag="v6.14.6" # (2020-10-14)
rocksdb_tag="v8.1.1" # (2023-04-21)
repo_clone_try_double "${primary_urls[rocksdb]}" "${secondary_urls[rocksdb]}" "rocksdb" "$rocksdb_tag" true
pushd rocksdb
git apply ../rocksdb.patch
git apply ../rocksdb8.1.1.patch
popd
# mgclient

View File

@@ -36,7 +36,7 @@ ADDITIONAL USE GRANT: You may use the Licensed Work in accordance with the
3. using the Licensed Work to create a work or solution
which competes (or might reasonably be expected to
compete) with the Licensed Work.
CHANGE DATE: 2027-05-04
CHANGE DATE: 2027-18-05
CHANGE LICENSE: Apache License, Version 2.0
For information about alternative licensing arrangements, please visit: https://memgraph.com/legal.

202
licenses/third-party/ldbc/LICENSE vendored Normal file
View File

@@ -0,0 +1,202 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -20,6 +20,8 @@ inline constexpr uint8_t kPreamble[4] = {0x60, 0x60, 0xB0, 0x17};
enum class Signature : uint8_t {
Noop = 0x00,
Init = 0x01,
LogOn = 0x6A,
LogOff = 0x6B,
AckFailure = 0x0E, // only v1
Reset = 0x0F,
Goodbye = 0x02,

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -28,7 +28,7 @@ inline constexpr size_t kChunkWholeSize = kChunkHeaderSize + kChunkMaxDataSize;
*/
inline constexpr size_t kHandshakeSize = 20;
inline constexpr uint16_t kSupportedVersions[] = {0x0100, 0x0400, 0x0401, 0x0403};
inline constexpr uint16_t kSupportedVersions[] = {0x0100, 0x0400, 0x0401, 0x0403, 0x0502};
inline constexpr int kPullAll = -1;
inline constexpr int kPullLast = -1;

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -33,7 +33,14 @@ namespace memgraph::communication::bolt {
template <typename Buffer>
class Decoder {
public:
explicit Decoder(Buffer &buffer) : buffer_(buffer) {}
explicit Decoder(Buffer &buffer) : buffer_(buffer), major_v_(0) {}
/**
* Lets the user update the version.
* This is all single thread for now. TODO: Update if ever multithreaded.
* @param major_v the major version of the Bolt protocol used.
*/
void UpdateVersion(int major_v) { major_v_ = major_v; }
/**
* Reads a Value from the available data in the buffer.
@@ -208,6 +215,10 @@ class Decoder {
protected:
Buffer &buffer_;
int major_v_; //!< Major version of the underlying Bolt protocol
// TODO: when refactoring
// Ideally the major_v would be a compile time constant. If the higher level (Bolt driver) ends up being separate
// classes, this could be just a template and each version of the driver would use the appropriate decoder.
private:
bool ReadNull(const Marker &marker, Value *data) {
@@ -370,11 +381,7 @@ class Decoder {
}
ret.emplace(std::move(dv_key.ValueString()), std::move(dv_val));
}
if (ret.size() != size) {
return false;
}
return true;
return ret.size() == size;
}
bool ReadVertex(Value *data) {
@@ -407,6 +414,14 @@ class Decoder {
}
vertex.properties = std::move(dv.ValueMap());
if (major_v_ > 4) {
// element_id introduced in v5.0
if (!ReadValue(&dv, Value::Type::String)) {
return false;
}
vertex.element_id = std::move(dv.ValueString());
}
return true;
}
@@ -445,6 +460,23 @@ class Decoder {
}
edge.properties = std::move(dv.ValueMap());
if (major_v_ > 4) {
// element_id introduced in v5.0
if (!ReadValue(&dv, Value::Type::String)) {
return false;
}
edge.element_id = std::move(dv.ValueString());
// from_element_id introduced in v5.0
if (!ReadValue(&dv, Value::Type::String)) {
return false;
}
edge.from_element_id = std::move(dv.ValueString());
// to_element_id introduced in v5.0
if (!ReadValue(&dv, Value::Type::String)) {
return false;
}
edge.to_element_id = std::move(dv.ValueString());
}
return true;
}
@@ -471,6 +503,14 @@ class Decoder {
}
edge.properties = std::move(dv.ValueMap());
if (major_v_ > 4) {
// element_id introduced in v5.0
if (!ReadValue(&dv, Value::Type::String)) {
return false;
}
edge.element_id = std::move(dv.ValueString());
}
return true;
}

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -36,7 +36,14 @@ namespace memgraph::communication::bolt {
template <typename Buffer>
class BaseEncoder {
public:
explicit BaseEncoder(Buffer &buffer) : buffer_(buffer) {}
explicit BaseEncoder(Buffer &buffer) : buffer_(buffer), major_v_(0) {}
/**
* Lets the user update the version.
* This is all single thread for now. TODO: Update if ever multithreaded.
* @param major_v the major version of the Bolt protocol used.
*/
void UpdateVersion(int major_v) { major_v_ = major_v; }
void WriteRAW(const uint8_t *data, uint64_t len) { buffer_.Write(data, len); }
@@ -116,7 +123,8 @@ class BaseEncoder {
}
void WriteVertex(const Vertex &vertex) {
WriteRAW(utils::UnderlyingCast(Marker::TinyStruct) + 3);
int struct_n = 3 + 1 * int(major_v_ > 4); // element_id introduced from v5
WriteRAW(utils::UnderlyingCast(Marker::TinyStruct) + struct_n);
WriteRAW(utils::UnderlyingCast(Signature::Node));
WriteInt(vertex.id.AsInt());
@@ -132,10 +140,16 @@ class BaseEncoder {
WriteString(prop.first);
WriteValue(prop.second);
}
if (major_v_ > 4) {
// element_id introduced in v5.0
WriteString(vertex.element_id);
}
}
void WriteEdge(const Edge &edge, bool unbound = false) {
WriteRAW(utils::UnderlyingCast(Marker::TinyStruct) + (unbound ? 3 : 5));
int struct_n = (unbound ? 3 + 1 * int(major_v_ > 4) : 5 + 3 * int(major_v_ > 4)); // element_id introduced from v5
WriteRAW(utils::UnderlyingCast(Marker::TinyStruct) + struct_n);
WriteRAW(utils::UnderlyingCast(unbound ? Signature::UnboundRelationship : Signature::Relationship));
WriteInt(edge.id.AsInt());
@@ -152,10 +166,22 @@ class BaseEncoder {
WriteString(prop.first);
WriteValue(prop.second);
}
if (major_v_ > 4) {
// element_id introduced in v5.0
WriteString(edge.element_id);
if (!unbound) {
// from_element_id introduced in v5.0
WriteString(edge.from_element_id);
// to_element_id introduced in v5.0
WriteString(edge.to_element_id);
}
}
}
void WriteEdge(const UnboundedEdge &edge) {
WriteRAW(utils::UnderlyingCast(Marker::TinyStruct) + 3);
const int struct_n = 3 + 1 * int(major_v_ > 4); // element_id introduced from v5
WriteRAW(utils::UnderlyingCast(Marker::TinyStruct) + struct_n);
WriteRAW(utils::UnderlyingCast(Signature::UnboundRelationship));
WriteInt(edge.id.AsInt());
@@ -168,6 +194,11 @@ class BaseEncoder {
WriteString(prop.first);
WriteValue(prop.second);
}
if (major_v_ > 4) {
// element_id introduced in v5.0
WriteString(edge.element_id);
}
}
void WritePath(const Path &path) {
@@ -264,6 +295,7 @@ class BaseEncoder {
protected:
Buffer &buffer_;
int major_v_; //!< Major version of the underlying Bolt protocol (TODO: Think about reimplementing the versioning)
private:
template <class T>

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -41,6 +41,8 @@ class ClientEncoder : private BaseEncoder<Buffer> {
public:
ClientEncoder(Buffer &buffer) : BaseEncoder<Buffer>(buffer) {}
using BaseEncoder<Buffer>::UpdateVersion;
/**
* Writes a Init message.
*

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -34,6 +34,8 @@ class Encoder : private BaseEncoder<Buffer> {
public:
Encoder(Buffer &buffer) : BaseEncoder<Buffer>(buffer) {}
using BaseEncoder<Buffer>::UpdateVersion;
/**
* Sends a Record message.
*

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -63,7 +63,8 @@ class Session {
* if an explicit transaction was started.
*/
virtual std::pair<std::vector<std::string>, std::optional<int>> Interpret(
const std::string &query, const std::map<std::string, Value> &params) = 0;
const std::string &query, const std::map<std::string, Value> &params,
const std::map<std::string, memgraph::communication::bolt::Value> &metadata) = 0;
/**
* Put results of the processed query in the `encoder`.
@@ -85,7 +86,7 @@ class Session {
*/
virtual std::map<std::string, Value> Discard(std::optional<int> n, std::optional<int> qid) = 0;
virtual void BeginTransaction() = 0;
virtual void BeginTransaction(const std::map<std::string, memgraph::communication::bolt::Value> &) = 0;
virtual void CommitTransaction() = 0;
virtual void RollbackTransaction() = 0;
@@ -120,6 +121,9 @@ class Session {
return;
}
handshake_done_ = true;
// Update the decoder's Bolt version (v5 has changed the undelying structure)
decoder_.UpdateVersion(version_.major);
encoder_.UpdateVersion(version_.major);
}
ChunkState chunk_state;

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -22,10 +22,15 @@
#include "communication/bolt/v1/state.hpp"
#include "communication/bolt/v1/states/handlers.hpp"
#include "communication/bolt/v1/value.hpp"
#include "utils/event_counter.hpp"
#include "utils/likely.hpp"
#include "utils/logging.hpp"
#include "utils/message.hpp"
namespace memgraph::metrics {
extern const Event BoltMessages;
} // namespace memgraph::metrics
namespace memgraph::communication::bolt {
template <typename TSession>
@@ -86,6 +91,37 @@ State RunHandlerV4(Signature signature, TSession &session, State state, Marker m
}
}
template <typename TSession>
State RunHandlerV5(Signature signature, TSession &session, State state, Marker marker) {
switch (signature) {
case Signature::Run:
return HandleRunV5<TSession>(session, state, marker);
case Signature::Pull:
return HandlePullV5<TSession>(session, state, marker);
case Signature::Discard:
return HandleDiscardV5<TSession>(session, state, marker);
case Signature::Reset:
return HandleReset<TSession>(session, marker);
case Signature::Begin:
return HandleBegin<TSession>(session, state, marker);
case Signature::Commit:
return HandleCommit<TSession>(session, state, marker);
case Signature::Goodbye:
return HandleGoodbye<TSession>();
case Signature::Rollback:
return HandleRollback<TSession>(session, state, marker);
case Signature::Noop:
return HandleNoop<TSession>(state);
case Signature::Route:
return HandleRoute<TSession>(session, marker);
case Signature::LogOff:
return HandleLogOff<TSession>();
default:
spdlog::trace("Unrecognized signature received (0x{:02X})!", utils::UnderlyingCast(signature));
return State::Close;
}
}
/**
* Executor state run function
* This function executes an initialized Bolt session.
@@ -103,8 +139,10 @@ State StateExecutingRun(TSession &session, State state) {
switch (session.version_.major) {
case 1:
memgraph::metrics::IncrementCounter(memgraph::metrics::BoltMessages);
return RunHandlerV1(signature, session, state, marker);
case 4: {
memgraph::metrics::IncrementCounter(memgraph::metrics::BoltMessages);
if (session.version_.minor >= 3) {
return RunHandlerV4<TSession, 3>(signature, session, state, marker);
}
@@ -113,6 +151,8 @@ State StateExecutingRun(TSession &session, State state) {
}
return RunHandlerV4<TSession>(signature, session, state, marker);
}
case 5:
return RunHandlerV5<TSession>(signature, session, state, marker);
default:
spdlog::trace("Unsupported bolt version:{}.{})!", session.version_.major, session.version_.minor);
return State::Close;

View File

@@ -12,6 +12,7 @@
#pragma once
#include <map>
#include <optional>
#include <string>
#include <string_view>
#include <vector>
@@ -22,6 +23,7 @@
#include "communication/bolt/v1/state.hpp"
#include "communication/bolt/v1/value.hpp"
#include "communication/exceptions.hpp"
#include "storage/v2/property_value.hpp"
#include "utils/logging.hpp"
#include "utils/message.hpp"
@@ -71,6 +73,23 @@ inline std::pair<std::string, std::string> ExceptionToErrorMessage(const std::ex
"should be in database logs."};
}
namespace helpers {
/** Extracts metadata from the extras field.
* NOTE: In order to avoid a copy, the metadata in moved.
* TODO: Update if extra field is used for anything else.
*/
inline std::map<std::string, Value> ConsumeMetadata(Value &extra) {
std::map<std::string, Value> md;
auto &md_tv = extra.ValueMap()["tx_metadata"];
if (md_tv.IsMap()) {
md = std::move(md_tv.ValueMap());
}
return md;
}
} // namespace helpers
namespace details {
template <bool is_pull, typename TSession>
@@ -209,7 +228,7 @@ State HandleRunV1(TSession &session, const State state, const Marker marker) {
try {
// Interpret can throw.
const auto [header, qid] = session.Interpret(query.ValueString(), params.ValueMap());
const auto [header, qid] = session.Interpret(query.ValueString(), params.ValueMap(), {});
// Convert std::string to Value
std::vector<Value> vec;
std::map<std::string, Value> data;
@@ -250,6 +269,7 @@ State HandleRunV4(TSession &session, const State state, const Marker marker) {
// Even though this part seems unnecessary it is needed to move the buffer
if (!session.decoder_.ReadValue(&extra, Value::Type::Map)) {
spdlog::trace("Couldn't read extra field!");
return State::Close;
}
if (state != State::Idle) {
@@ -266,7 +286,8 @@ State HandleRunV4(TSession &session, const State state, const Marker marker) {
try {
// Interpret can throw.
const auto [header, qid] = session.Interpret(query.ValueString(), params.ValueMap());
const auto [header, qid] =
session.Interpret(query.ValueString(), params.ValueMap(), helpers::ConsumeMetadata(extra));
// Convert std::string to Value
std::vector<Value> vec;
std::map<std::string, Value> data;
@@ -288,6 +309,12 @@ State HandleRunV4(TSession &session, const State state, const Marker marker) {
}
}
template <typename TSession>
State HandleRunV5(TSession &session, const State state, const Marker marker) {
// Using V4 on purpose
return HandleRunV4<TSession>(session, state, marker);
}
template <typename TSession>
State HandlePullV1(TSession &session, const State state, const Marker marker) {
return details::HandlePullDiscardV1<true>(session, state, marker);
@@ -298,6 +325,12 @@ State HandlePullV4(TSession &session, const State state, const Marker marker) {
return details::HandlePullDiscardV4<true>(session, state, marker);
}
template <typename TSession>
State HandlePullV5(TSession &session, const State state, const Marker marker) {
// Using V4 on purpose
return HandlePullV4<TSession>(session, state, marker);
}
template <typename TSession>
State HandleDiscardV1(TSession &session, const State state, const Marker marker) {
return details::HandlePullDiscardV1<false>(session, state, marker);
@@ -308,6 +341,12 @@ State HandleDiscardV4(TSession &session, const State state, const Marker marker)
return details::HandlePullDiscardV4<false>(session, state, marker);
}
template <typename TSession>
State HandleDiscardV5(TSession &session, const State state, const Marker marker) {
// Using V4 on purpose
return HandleDiscardV4<TSession>(session, state, marker);
}
template <typename TSession>
State HandleReset(TSession &session, const Marker marker) {
// IMPORTANT: This implementation of the Bolt RESET command isn't fully
@@ -360,7 +399,7 @@ State HandleBegin(TSession &session, const State state, const Marker marker) {
}
try {
session.BeginTransaction();
session.BeginTransaction(helpers::ConsumeMetadata(extra));
} catch (const std::exception &e) {
return HandleFailure(session, e);
}
@@ -465,4 +504,10 @@ State HandleRoute(TSession &session, const Marker marker) {
}
return State::Error;
}
template <typename TSession>
State HandleLogOff() {
// Not arguments sent, the user just needs to reauthenticate
return State::Init;
}
} // namespace memgraph::communication::bolt

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -27,17 +27,25 @@ namespace details {
template <typename TSession>
std::optional<State> AuthenticateUser(TSession &session, Value &metadata) {
// Get authentication data.
// From neo4j driver v4.4, fields that have a default value are not sent.
// In order to have back-compatibility, the missing fields will be added.
auto &data = metadata.ValueMap();
if (!data.count("scheme")) {
spdlog::warn("The client didn't supply authentication information!");
return State::Close;
if (data.empty()) { // Special case auth=None
spdlog::warn("The client didn't supply the authentication scheme! Trying with \"none\"...");
data["scheme"] = "none";
}
std::string username;
std::string password;
if (data["scheme"].ValueString() == "basic") {
if (!data.count("principal") || !data.count("credentials")) {
spdlog::warn("The client didn't supply authentication information!");
return State::Close;
if (!data.count("principal")) { // Special case principal = ""
spdlog::warn("The client didn't supply the principal field! Trying with \"\"...");
data["principal"] = "";
}
if (!data.count("credentials")) { // Special case credentials = ""
spdlog::warn("The client didn't supply the credentials field! Trying with \"\"...");
data["credentials"] = "";
}
username = data["principal"].ValueString();
password = data["credentials"].ValueString();
@@ -106,6 +114,30 @@ std::optional<Value> GetMetadataV4(TSession &session, const Marker marker) {
return std::nullopt;
}
auto &data = metadata.ValueMap();
if (!data.count("user_agent")) {
spdlog::warn("The client didn't supply the user agent!");
return std::nullopt;
}
spdlog::info("Client connected '{}'", data.at("user_agent").ValueString());
return metadata;
}
template <typename TSession>
std::optional<Value> GetInitDataV5(TSession &session, const Marker marker) {
if (marker != Marker::TinyStruct1) [[unlikely]] {
spdlog::trace("Expected TinyStruct1 marker, but received 0x{:02X}!", utils::UnderlyingCast(marker));
return std::nullopt;
}
Value metadata;
if (!session.decoder_.ReadValue(&metadata, Value::Type::Map)) {
spdlog::trace("Couldn't read metadata!");
return std::nullopt;
}
const auto &data = metadata.ValueMap();
if (!data.count("user_agent")) {
spdlog::warn("The client didn't supply the user agent!");
@@ -117,6 +149,22 @@ std::optional<Value> GetMetadataV4(TSession &session, const Marker marker) {
return metadata;
}
template <typename TSession>
std::optional<Value> GetAuthDataV5(TSession &session, const Marker marker) {
if (marker != Marker::TinyStruct1) [[unlikely]] {
spdlog::trace("Expected TinyStruct1 marker, but received 0x{:02X}!", utils::UnderlyingCast(marker));
return std::nullopt;
}
Value metadata;
if (!session.decoder_.ReadValue(&metadata, Value::Type::Map)) {
spdlog::trace("Couldn't read metadata!");
return std::nullopt;
}
return metadata;
}
template <typename TSession>
State SendSuccessMessage(TSession &session) {
// Neo4j's Java driver 4.1.1+ requires connection_id.
@@ -180,6 +228,57 @@ State StateInitRunV4(TSession &session, Marker marker, Signature signature) {
return SendSuccessMessage(session);
}
template <typename TSession>
State StateInitRunV5(TSession &session, Marker marker, Signature signature) {
if (signature == Signature::Noop) [[unlikely]] {
SPDLOG_DEBUG("Received NOOP message");
return State::Init;
}
if (signature == Signature::Init) {
auto maybeMetadata = GetInitDataV5(session, marker);
if (!maybeMetadata) {
return State::Close;
}
if (SendSuccessMessage(session) == State::Close) {
return State::Close;
}
// Stay in Init
return State::Init;
} else if (signature == Signature::LogOn) {
if (marker != Marker::TinyStruct1) [[unlikely]] {
spdlog::trace("Expected TinyStruct1 marker, but received 0x{:02X}!", utils::UnderlyingCast(marker));
spdlog::trace(
"The client sent malformed data, but we are continuing "
"because the official Neo4j Java driver sends malformed "
"data. D'oh!");
return State::Close;
}
auto maybeMetadata = GetAuthDataV5(session, marker);
if (!maybeMetadata) {
return State::Close;
}
auto result = AuthenticateUser(session, *maybeMetadata);
if (result) {
spdlog::trace("Failed to authenticate, closing connection...");
return State::Close;
}
if (SendSuccessMessage(session) == State::Close) {
return State::Close;
}
return State::Idle;
} else [[unlikely]] {
spdlog::trace("Expected Init signature, but received 0x{:02X}!", utils::UnderlyingCast(signature));
return State::Close;
}
}
} // namespace details
/**
@@ -208,6 +307,9 @@ State StateInitRun(TSession &session) {
}
return details::StateInitRunV4<TSession>(session, marker, signature);
}
case 5: {
return details::StateInitRunV5<TSession>(session, marker, signature);
}
}
spdlog::trace("Unsupported bolt version:{}.{})!", session.version_.major, session.version_.minor);
return State::Close;

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -342,6 +342,9 @@ std::ostream &operator<<(std::ostream &os, const Vertex &vertex) {
[&](auto &stream, const auto &pair) { stream << pair.first << ": " << pair.second; });
os << "}";
}
if (!vertex.element_id.empty()) {
os << " element_id: " << vertex.element_id;
}
return os << ")";
}

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -57,6 +57,7 @@ struct Vertex {
Id id;
std::vector<std::string> labels;
std::map<std::string, Value> properties;
std::string element_id;
};
/**
@@ -69,6 +70,9 @@ struct Edge {
Id to;
std::string type;
std::map<std::string, Value> properties;
std::string element_id;
std::string from_element_id;
std::string to_element_id;
};
/**
@@ -79,6 +83,7 @@ struct UnboundedEdge {
Id id;
std::string type;
std::map<std::string, Value> properties;
std::string element_id;
};
/**

View File

@@ -0,0 +1,108 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <list>
#include <memory>
#include <spdlog/spdlog.h>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/strand.hpp>
#include <boost/beast/core.hpp>
#include "communication/context.hpp"
#include "communication/http/session.hpp"
#include "utils/spin_lock.hpp"
#include "utils/synchronized.hpp"
namespace memgraph::communication::http {
template <class TRequestHandler, typename TSessionData>
class Listener final : public std::enable_shared_from_this<Listener<TRequestHandler, TSessionData>> {
using tcp = boost::asio::ip::tcp;
using SessionHandler = Session<TRequestHandler, TSessionData>;
using std::enable_shared_from_this<Listener<TRequestHandler, TSessionData>>::shared_from_this;
public:
Listener(const Listener &) = delete;
Listener(Listener &&) = delete;
Listener &operator=(const Listener &) = delete;
Listener &operator=(Listener &&) = delete;
~Listener() {}
template <typename... Args>
static std::shared_ptr<Listener> Create(Args &&...args) {
return std::shared_ptr<Listener>{new Listener(std::forward<Args>(args)...)};
}
// Start accepting incoming connections
void Run() { DoAccept(); }
tcp::endpoint GetEndpoint() const { return acceptor_.local_endpoint(); }
private:
Listener(boost::asio::io_context &ioc, TSessionData *data, ServerContext *context, tcp::endpoint endpoint)
: ioc_(ioc), data_(data), context_(context), acceptor_(ioc) {
boost::beast::error_code ec;
// Open the acceptor
acceptor_.open(endpoint.protocol(), ec);
if (ec) {
LogError(ec, "open");
return;
}
// Allow address reuse
acceptor_.set_option(boost::asio::socket_base::reuse_address(true), ec);
if (ec) {
LogError(ec, "set_option");
return;
}
// Bind to the server address
acceptor_.bind(endpoint, ec);
if (ec) {
LogError(ec, "bind");
return;
}
acceptor_.listen(boost::asio::socket_base::max_listen_connections, ec);
if (ec) {
LogError(ec, "listen");
return;
}
spdlog::info("HTTP server is listening on {}:{}", endpoint.address(), endpoint.port());
}
void DoAccept() {
acceptor_.async_accept(ioc_, [shared_this = shared_from_this()](auto ec, auto socket) {
shared_this->OnAccept(ec, std::move(socket));
});
}
void OnAccept(boost::beast::error_code ec, tcp::socket socket) {
if (ec) {
return LogError(ec, "accept");
}
SessionHandler::Create(std::move(socket), data_, *context_)->Run();
DoAccept();
}
boost::asio::io_context &ioc_;
TSessionData *data_;
ServerContext *context_;
tcp::acceptor acceptor_;
};
} // namespace memgraph::communication::http

View File

@@ -0,0 +1,65 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <thread>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include "communication/http/listener.hpp"
#include "io/network/endpoint.hpp"
namespace memgraph::communication::http {
template <class TRequestHandler, typename TSessionData>
class Server final {
using tcp = boost::asio::ip::tcp;
public:
explicit Server(io::network::Endpoint endpoint, TSessionData *data, ServerContext *context)
: listener_{Listener<TRequestHandler, TSessionData>::Create(
ioc_, data, context, tcp::endpoint{boost::asio::ip::make_address(endpoint.address), endpoint.port})} {}
Server(const Server &) = delete;
Server(Server &&) = delete;
Server &operator=(const Server &) = delete;
Server &operator=(Server &&) = delete;
~Server() {
MG_ASSERT(!background_thread_ || (ioc_.stopped() && !background_thread_->joinable()),
"Server wasn't shutdown properly");
}
void Start() {
MG_ASSERT(!background_thread_, "The server was already started!");
listener_->Run();
background_thread_.emplace([this] { ioc_.run(); });
}
void Shutdown() { ioc_.stop(); }
void AwaitShutdown() {
if (background_thread_ && background_thread_->joinable()) {
background_thread_->join();
}
}
bool IsRunning() const { return background_thread_ && !ioc_.stopped(); }
tcp::endpoint GetEndpoint() const { return listener_->GetEndpoint(); }
private:
boost::asio::io_context ioc_;
std::shared_ptr<Listener<TRequestHandler, TSessionData>> listener_;
std::optional<std::thread> background_thread_;
};
} // namespace memgraph::communication::http

View File

@@ -0,0 +1,193 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <deque>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <variant>
#include <spdlog/spdlog.h>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/dispatch.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/strand.hpp>
#include <boost/beast/core/buffers_to_string.hpp>
#include <boost/beast/core/stream_traits.hpp>
#include <boost/beast/core/tcp_stream.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/ssl.hpp>
#include <boost/beast/version.hpp>
#include <json/json.hpp>
#include "communication/context.hpp"
#include "utils/logging.hpp"
#include "utils/variant_helpers.hpp"
namespace memgraph::communication::http {
inline constexpr uint16_t kSSLExpirySeconds = 30;
inline void LogError(boost::beast::error_code ec, const std::string_view what) {
spdlog::warn("HTTP session failed on {}: {}", what, ec.message());
}
template <class TRequestHandler, typename TSessionData>
class Session : public std::enable_shared_from_this<Session<TRequestHandler, TSessionData>> {
using tcp = boost::asio::ip::tcp;
using std::enable_shared_from_this<Session<TRequestHandler, TSessionData>>::shared_from_this;
public:
template <typename... Args>
static std::shared_ptr<Session> Create(Args &&...args) {
return std::shared_ptr<Session>{new Session{std::forward<Args>(args)...}};
}
void Run() {
if (auto *ssl = std::get_if<SSLSocket>(&stream_); ssl != nullptr) {
try {
boost::beast::get_lowest_layer(*ssl).expires_after(std::chrono::seconds(kSSLExpirySeconds));
ssl->handshake(boost::asio::ssl::stream_base::server);
} catch (const boost::system::system_error &e) {
spdlog::warn("Failed on SSL handshake: {}", e.what());
return;
}
}
// run on the strand
boost::asio::dispatch(strand_, [shared_this = shared_from_this()] { shared_this->DoRead(); });
}
private:
using PlainSocket = boost::beast::tcp_stream;
using SSLSocket = boost::beast::ssl_stream<boost::beast::tcp_stream>;
explicit Session(tcp::socket &&socket, TSessionData *data, ServerContext &context)
: stream_(CreateSocket(std::move(socket), context)),
handler_(data),
strand_{boost::asio::make_strand(GetExecutor())} {}
std::variant<PlainSocket, SSLSocket> CreateSocket(tcp::socket &&socket, ServerContext &context) {
if (context.use_ssl()) {
ssl_context_.emplace(context.context_clone());
return Session::SSLSocket{std::move(socket), *ssl_context_};
}
return Session::PlainSocket{std::move(socket)};
}
void OnWrite(boost::beast::error_code ec, size_t bytes_transferred) {
boost::ignore_unused(bytes_transferred);
if (ec) {
close_ = true;
return LogError(ec, "write");
}
if (close_) {
DoClose();
return;
}
res_ = nullptr;
DoRead();
}
void DoRead() {
req_ = {};
ExecuteForStream([this](auto &&stream) {
boost::beast::get_lowest_layer(stream).expires_after(std::chrono::seconds(kSSLExpirySeconds));
boost::beast::http::async_read(
stream, buffer_, req_,
boost::asio::bind_executor(strand_, std::bind_front(&Session::OnRead, shared_from_this())));
});
}
void OnRead(boost::beast::error_code ec, size_t bytes_transferred) {
boost::ignore_unused(bytes_transferred);
if (ec == boost::beast::http::error::end_of_stream) {
DoClose();
return;
}
if (ec) {
return LogError(ec, "read");
}
auto async_write = [this](boost::beast::http::response<boost::beast::http::string_body> msg) {
ExecuteForStream([this, &msg](auto &&stream) {
// The lifetime of the message has to extend
// for the duration of the async operation so
// we use a shared_ptr to manage it.
auto sp = std::make_shared<boost::beast::http::response<boost::beast::http::string_body>>(std::move(msg));
// Store a type-erased version of the shared
// pointer in the class to keep it alive.
res_ = sp;
// Write the response
boost::beast::http::async_write(
stream, *sp, boost::asio::bind_executor(strand_, std::bind_front(&Session::OnWrite, shared_from_this())));
});
};
// handle request
handler_.HandleRequest(std::move(req_), async_write);
}
void DoClose() {
std::visit(utils::Overloaded{[this](SSLSocket &stream) {
boost::beast::get_lowest_layer(stream).expires_after(std::chrono::seconds(30));
// Perform the SSL shutdown
stream.async_shutdown(
boost::beast::bind_front_handler(&Session::OnClose, shared_from_this()));
},
[](PlainSocket &stream) {
boost::beast::error_code ec;
stream.socket().shutdown(tcp::socket::shutdown_send, ec);
}},
stream_);
}
void OnClose(boost::beast::error_code ec) {
if (ec) {
LogError(ec, "close");
}
// At this point the connection is closed gracefully
}
auto GetExecutor() {
return std::visit(utils::Overloaded{[](auto &&stream) { return stream.get_executor(); }}, stream_);
}
template <typename F>
decltype(auto) ExecuteForStream(F &&fn) {
return std::visit(utils::Overloaded{std::forward<F>(fn)}, stream_);
}
std::optional<std::reference_wrapper<boost::asio::ssl::context>> ssl_context_;
std::variant<PlainSocket, SSLSocket> stream_;
boost::beast::flat_buffer buffer_;
TRequestHandler handler_;
boost::beast::http::request<boost::beast::http::string_body> req_;
std::shared_ptr<void> res_;
boost::asio::strand<boost::beast::tcp_stream::executor_type> strand_;
bool close_{false};
};
} // namespace memgraph::communication::http

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -30,6 +30,7 @@
#include "communication/context.hpp"
#include "communication/v2/pool.hpp"
#include "communication/v2/session.hpp"
#include "utils/message.hpp"
#include "utils/spin_lock.hpp"
#include "utils/synchronized.hpp"

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -41,11 +41,21 @@
#include <boost/beast/websocket/rfc6455.hpp>
#include <boost/system/detail/error_code.hpp>
#include "communication/buffer.hpp"
#include "communication/context.hpp"
#include "communication/exceptions.hpp"
#include "utils/event_counter.hpp"
#include "utils/logging.hpp"
#include "utils/on_scope_exit.hpp"
#include "utils/variant_helpers.hpp"
namespace memgraph::metrics {
extern const Event ActiveSessions;
extern const Event ActiveTCPSessions;
extern const Event ActiveSSLSessions;
extern const Event ActiveWebSocketSessions;
} // namespace memgraph::metrics
namespace memgraph::communication::v2 {
/**
@@ -99,6 +109,8 @@ class WebsocketSession : public std::enable_shared_from_this<WebsocketSession<TS
// Start the asynchronous accept operation
template <class Body, class Allocator>
void DoAccept(boost::beast::http::request<Body, boost::beast::http::basic_fields<Allocator>> req) {
memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveWebSocketSessions);
execution_active_ = true;
// Set suggested timeout settings for the websocket
ws_.set_option(boost::beast::websocket::stream_base::timeout::suggested(boost::beast::role_type::server));
@@ -213,6 +225,10 @@ class WebsocketSession : public std::enable_shared_from_this<WebsocketSession<TS
if (!IsConnected()) {
return;
}
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveSessions);
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveWebSocketSessions);
if (ec) {
return OnError(ec, "close");
}
@@ -259,12 +275,19 @@ class Session final : public std::enable_shared_from_this<Session<TSession, TSes
if (execution_active_) {
return false;
}
memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveSessions);
execution_active_ = true;
timeout_timer_.async_wait(boost::asio::bind_executor(strand_, std::bind(&Session::OnTimeout, shared_from_this())));
if (std::holds_alternative<SSLSocket>(socket_)) {
utils::OnScopeExit increment_counter(
[] { memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveSSLSessions); });
boost::asio::dispatch(strand_, [shared_this = shared_from_this()] { shared_this->DoHandshake(); });
} else {
utils::OnScopeExit increment_counter(
[] { memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveTCPSessions); });
boost::asio::dispatch(strand_, [shared_this = shared_from_this()] { shared_this->DoRead(); });
}
return true;
@@ -450,6 +473,14 @@ class Session final : public std::enable_shared_from_this<Session<TSession, TSes
}
void OnClose(const boost::system::error_code &ec) {
if (ssl_context_.has_value()) {
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveSSLSessions);
} else {
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveTCPSessions);
}
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveSessions);
if (ec) {
return OnError(ec);
}
@@ -465,7 +496,7 @@ class Session final : public std::enable_shared_from_this<Session<TSession, TSes
if (timeout_timer_.expiry() <= boost::asio::steady_timer::clock_type::now()) {
// The deadline has passed. Stop the session. The other actors will
// terminate as soon as possible.
spdlog::info("Shutting down session after {} of inactivity", timeout_seconds_);
spdlog::info("Shutting down session after {} seconds of inactivity", timeout_seconds_.count());
DoShutdown();
} else {
// Put the actor back to sleep.

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -150,7 +150,9 @@ storage::Result<communication::bolt::Vertex> ToBoltVertex(const storage::VertexA
for (const auto &prop : *maybe_properties) {
properties[db.PropertyToName(prop.first)] = ToBoltValue(prop.second);
}
return communication::bolt::Vertex{id, labels, properties};
// Introduced in Bolt v5 (for now just send the ID)
const auto element_id = std::to_string(id.AsInt());
return communication::bolt::Vertex{id, labels, properties, element_id};
}
storage::Result<communication::bolt::Edge> ToBoltEdge(const storage::EdgeAccessor &edge, const storage::Storage &db,
@@ -165,7 +167,11 @@ storage::Result<communication::bolt::Edge> ToBoltEdge(const storage::EdgeAccesso
for (const auto &prop : *maybe_properties) {
properties[db.PropertyToName(prop.first)] = ToBoltValue(prop.second);
}
return communication::bolt::Edge{id, from, to, type, properties};
// Introduced in Bolt v5 (for now just send the ID)
const auto element_id = std::to_string(id.AsInt());
const auto from_element_id = std::to_string(from.AsInt());
const auto to_element_id = std::to_string(to.AsInt());
return communication::bolt::Edge{id, from, to, type, properties, element_id, from_element_id, to_element_id};
}
storage::Result<communication::bolt::Path> ToBoltPath(const query::Path &path, const storage::Storage &db,

View File

@@ -0,0 +1,4 @@
set(mg_http_handlers_sources)
add_library(mg-http-handlers STATIC ${mg_http_handlers_sources})
target_link_libraries(mg-http-handlers mg-query mg-storage-v2)

View File

@@ -0,0 +1,211 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <atomic>
#include <tuple>
#include <vector>
#include <spdlog/spdlog.h>
#include <boost/beast/http.hpp>
#include <boost/beast/version.hpp>
#include <json/json.hpp>
#include <utils/event_counter.hpp>
#include <utils/event_gauge.hpp>
#include "storage/v2/storage.hpp"
#include "utils/event_gauge.hpp"
#include "utils/event_histogram.hpp"
namespace memgraph::http {
struct MetricsResponse {
uint64_t vertex_count;
uint64_t edge_count;
double average_degree;
uint64_t memory_usage;
uint64_t disk_usage;
// Storage of all the counter values throughout the system
// e.g. number of active transactions
std::vector<std::tuple<std::string, std::string, uint64_t>> event_counters{};
// Storage of all the current values throughout the system
std::vector<std::tuple<std::string, std::string, uint64_t>> event_gauges{};
// Storage of all the percentile values across the histograms in the system
// e.g. query latency percentiles, snapshot recovery duration percentiles, etc.
std::vector<std::tuple<std::string, std::string, uint64_t>> event_histograms{};
};
template <typename TSessionData>
class MetricsService {
public:
explicit MetricsService(TSessionData *data) : db_(data->db) {}
nlohmann::json GetMetricsJSON() {
auto response = GetMetrics();
return AsJson(response);
}
private:
const storage::Storage *db_;
MetricsResponse GetMetrics() {
auto info = db_->GetInfo();
return MetricsResponse{.vertex_count = info.vertex_count,
.edge_count = info.edge_count,
.average_degree = info.average_degree,
.memory_usage = info.memory_usage,
.disk_usage = info.disk_usage,
.event_counters = GetEventCounters(),
.event_gauges = GetEventGauges(),
.event_histograms = GetEventHistograms()};
}
nlohmann::json AsJson(MetricsResponse response) {
auto metrics_response = nlohmann::json();
const auto *general_type = "General";
metrics_response[general_type]["vertex_count"] = response.vertex_count;
metrics_response[general_type]["edge_count"] = response.edge_count;
metrics_response[general_type]["average_degree"] = response.average_degree;
metrics_response[general_type]["memory_usage"] = response.memory_usage;
metrics_response[general_type]["disk_usage"] = response.disk_usage;
for (const auto &[name, type, value] : response.event_counters) {
metrics_response[type][name] = value;
}
for (const auto &[name, type, value] : response.event_gauges) {
metrics_response[type][name] = value;
}
for (const auto &[name, type, value] : response.event_histograms) {
metrics_response[type][name] = value;
}
return metrics_response;
}
auto GetEventCounters() {
// NOLINTNEXTLINE(cppcoreguidelines-init-variables)
std::vector<std::tuple<std::string, std::string, uint64_t>> event_counters{};
for (auto i = 0; i < memgraph::metrics::CounterEnd(); i++) {
event_counters.emplace_back(memgraph::metrics::GetCounterName(i), memgraph::metrics::GetCounterType(i),
memgraph::metrics::global_counters[i].load(std::memory_order_acquire));
}
return event_counters;
}
auto GetEventGauges() {
// NOLINTNEXTLINE(cppcoreguidelines-init-variables)
std::vector<std::tuple<std::string, std::string, uint64_t>> event_gauges{};
for (auto i = 0; i < memgraph::metrics::GaugeEnd(); i++) {
event_gauges.emplace_back(memgraph::metrics::GetGaugeName(i), memgraph::metrics::GetGaugeType(i),
memgraph::metrics::global_gauges[i].load(std::memory_order_acquire));
}
return event_gauges;
}
auto GetEventHistograms() {
// NOLINTNEXTLINE(cppcoreguidelines-init-variables)
std::vector<std::tuple<std::string, std::string, uint64_t>> event_histograms{};
for (auto i = 0; i < memgraph::metrics::HistogramEnd(); i++) {
const auto *name = memgraph::metrics::GetHistogramName(i);
auto &histogram = memgraph::metrics::global_histograms[i];
for (auto &[percentile, value] : histogram.YieldPercentiles()) {
auto metric_name = std::string(name) + "_" + std::to_string(percentile) + "p";
event_histograms.emplace_back(metric_name, memgraph::metrics::GetHistogramType(i), value);
}
}
return event_histograms;
}
};
template <typename TSessionData>
class MetricsRequestHandler final {
public:
explicit MetricsRequestHandler(TSessionData *data) : service_(data) {
spdlog::info("Basic request handler started!");
}
MetricsRequestHandler(const MetricsRequestHandler &) = delete;
MetricsRequestHandler(MetricsRequestHandler &&) = delete;
MetricsRequestHandler &operator=(const MetricsRequestHandler &) = delete;
MetricsRequestHandler &operator=(MetricsRequestHandler &&) = delete;
~MetricsRequestHandler() = default;
template <class Body, class Allocator>
// NOLINTNEXTLINE(misc-unused-parameters)
void HandleRequest(boost::beast::http::request<Body, boost::beast::http::basic_fields<Allocator>> &&req,
std::function<void(boost::beast::http::response<boost::beast::http::string_body>)> &&send) {
auto response_json = nlohmann::json();
// Returns a bad request response
auto const bad_request = [&req, &response_json](const auto why) {
response_json["error"] = std::string(why);
// NOLINTNEXTLINE(cppcoreguidelines-init-variables)
boost::beast::http::response<boost::beast::http::string_body> res{boost::beast::http::status::bad_request,
req.version()};
res.set(boost::beast::http::field::server, BOOST_BEAST_VERSION_STRING);
res.set(boost::beast::http::field::content_type, "application/json");
res.keep_alive(req.keep_alive());
res.body() = response_json.dump();
res.prepare_payload();
return res;
};
// Make sure we can handle the method
if (req.method() != boost::beast::http::verb::get) {
return send(bad_request("Unknown HTTP-method"));
}
// Request path must be absolute and not contain "..".
if (req.target().empty() || req.target()[0] != '/' || req.target().find("..") != boost::beast::string_view::npos) {
return send(bad_request("Illegal request-target"));
}
// NOLINTNEXTLINE(cppcoreguidelines-init-variables)
boost::beast::http::string_body::value_type body;
auto service_response = service_.GetMetricsJSON();
body.append(service_response.dump());
// Cache the size since we need it after the move
const auto size = body.size();
// Respond to GET request
// NOLINTNEXTLINE(cppcoreguidelines-init-variables)
boost::beast::http::response<boost::beast::http::string_body> res{
std::piecewise_construct, std::make_tuple(std::move(body)),
std::make_tuple(boost::beast::http::status::ok, req.version())};
res.set(boost::beast::http::field::server, BOOST_BEAST_VERSION_STRING);
res.set(boost::beast::http::field::content_type, "application/json");
res.content_length(size);
res.keep_alive(req.keep_alive());
return send(std::move(res));
}
private:
MetricsService<TSessionData> service_;
};
} // namespace memgraph::http

View File

@@ -36,11 +36,13 @@
#include "auth/models.hpp"
#include "communication/bolt/v1/constants.hpp"
#include "communication/http/server.hpp"
#include "communication/websocket/auth.hpp"
#include "communication/websocket/server.hpp"
#include "glue/auth_checker.hpp"
#include "glue/auth_handler.hpp"
#include "helpers.hpp"
#include "http_handlers/metrics.hpp"
#include "license/license.hpp"
#include "license/license_sender.hpp"
#include "py/py.hpp"
@@ -113,6 +115,9 @@ DEFINE_string(bolt_address, "0.0.0.0", "IP address on which the Bolt server shou
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_string(monitoring_address, "0.0.0.0",
"IP address on which the websocket server for Memgraph monitoring should listen.");
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_string(metrics_address, "0.0.0.0",
"IP address on which the Memgraph server for exposing metrics should listen.");
DEFINE_VALIDATED_int32(bolt_port, 7687, "Port on which the Bolt server should listen.",
FLAG_IN_RANGE(0, std::numeric_limits<uint16_t>::max()));
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
@@ -120,6 +125,9 @@ DEFINE_VALIDATED_int32(monitoring_port, 7444,
"Port on which the websocket server for Memgraph monitoring should listen.",
FLAG_IN_RANGE(0, std::numeric_limits<uint16_t>::max()));
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_VALIDATED_int32(metrics_port, 9091, "Port on which the Memgraph server for exposing metrics should listen.",
FLAG_IN_RANGE(0, std::numeric_limits<uint16_t>::max()));
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_VALIDATED_int32(bolt_num_workers, std::max(std::thread::hardware_concurrency(), 1U),
"Number of workers used by the Bolt server. By default, this will be the "
"number of processing units available on the machine.",
@@ -192,6 +200,20 @@ DEFINE_VALIDATED_uint64(storage_wal_file_flush_every_n_tx,
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_bool(storage_snapshot_on_exit, false, "Controls whether the storage creates another snapshot on exit.");
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_uint64(storage_items_per_batch, memgraph::storage::Config::Durability().items_per_batch,
"The number of edges and vertices stored in a batch in a snapshot file.");
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_bool(storage_parallel_index_recovery, false,
"Controls whether the index creation can be done in a multithreaded fashion.");
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_uint64(storage_recovery_thread_count,
std::max(static_cast<uint64_t>(std::thread::hardware_concurrency()),
memgraph::storage::Config::Durability().recovery_thread_count),
"The number of threads used to recover persisted data from disk.");
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_bool(telemetry_enabled, false,
"Set to true to enable telemetry. We collect information about the "
@@ -478,6 +500,10 @@ void InitFromCypherlFile(memgraph::query::InterpreterContext &ctx, std::string c
}
}
namespace memgraph::metrics {
extern const Event ActiveBoltSessions;
} // namespace memgraph::metrics
class BoltSession final : public memgraph::communication::bolt::Session<memgraph::communication::v2::InputStream,
memgraph::communication::v2::OutputStream> {
public:
@@ -495,26 +521,41 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
#endif
endpoint_(endpoint),
run_id_(data->run_id) {
memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveBoltSessions);
interpreter_context_->interpreters.WithLock([this](auto &interpreters) { interpreters.insert(&interpreter_); });
}
~BoltSession() override {
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveBoltSessions);
interpreter_context_->interpreters.WithLock([this](auto &interpreters) { interpreters.erase(&interpreter_); });
}
using memgraph::communication::bolt::Session<memgraph::communication::v2::InputStream,
memgraph::communication::v2::OutputStream>::TEncoder;
void BeginTransaction() override { interpreter_.BeginTransaction(); }
void BeginTransaction(const std::map<std::string, memgraph::communication::bolt::Value> &metadata) override {
std::map<std::string, memgraph::storage::PropertyValue> metadata_pv;
for (const auto &[key, bolt_value] : metadata) {
metadata_pv.emplace(key, memgraph::glue::ToPropertyValue(bolt_value));
}
interpreter_.BeginTransaction(metadata_pv);
}
void CommitTransaction() override { interpreter_.CommitTransaction(); }
void RollbackTransaction() override { interpreter_.RollbackTransaction(); }
std::pair<std::vector<std::string>, std::optional<int>> Interpret(
const std::string &query, const std::map<std::string, memgraph::communication::bolt::Value> &params) override {
const std::string &query, const std::map<std::string, memgraph::communication::bolt::Value> &params,
const std::map<std::string, memgraph::communication::bolt::Value> &metadata) override {
std::map<std::string, memgraph::storage::PropertyValue> params_pv;
for (const auto &kv : params) params_pv.emplace(kv.first, memgraph::glue::ToPropertyValue(kv.second));
std::map<std::string, memgraph::storage::PropertyValue> metadata_pv;
for (const auto &[key, bolt_param] : params) {
params_pv.emplace(key, memgraph::glue::ToPropertyValue(bolt_param));
}
for (const auto &[key, bolt_md] : metadata) {
metadata_pv.emplace(key, memgraph::glue::ToPropertyValue(bolt_md));
}
const std::string *username{nullptr};
if (user_) {
username = &user_->username();
@@ -526,7 +567,7 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
}
#endif
try {
auto result = interpreter_.Prepare(query, params_pv, username);
auto result = interpreter_.Prepare(query, params_pv, username, metadata_pv);
if (user_ && !memgraph::glue::AuthChecker::IsUserAuthorized(*user_, result.privileges)) {
interpreter_.Abort();
throw memgraph::communication::bolt::ClientError(
@@ -658,6 +699,8 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
};
using ServerT = memgraph::communication::v2::Server<BoltSession, SessionData>;
using MonitoringServerT =
memgraph::communication::http::Server<memgraph::http::MetricsRequestHandler<SessionData>, SessionData>;
using memgraph::communication::ServerContext;
// Needed to correctly handle memgraph destruction from a signal handler.
@@ -852,7 +895,10 @@ int main(int argc, char **argv) {
.wal_file_size_kibibytes = FLAGS_storage_wal_file_size_kib,
.wal_file_flush_every_n_tx = FLAGS_storage_wal_file_flush_every_n_tx,
.snapshot_on_exit = FLAGS_storage_snapshot_on_exit,
.restore_replicas_on_startup = true},
.restore_replicas_on_startup = true,
.items_per_batch = FLAGS_storage_items_per_batch,
.recovery_thread_count = FLAGS_storage_recovery_thread_count,
.allow_parallel_index_creation = FLAGS_storage_parallel_index_recovery},
.transaction = {.isolation_level = ParseIsolationLevel()}};
if (FLAGS_storage_snapshot_interval_sec == 0) {
if (FLAGS_storage_wal_enabled) {
@@ -964,8 +1010,9 @@ int main(int argc, char **argv) {
});
telemetry->AddCollector("event_counters", []() -> nlohmann::json {
nlohmann::json ret;
for (size_t i = 0; i < EventCounter::End(); ++i) {
ret[EventCounter::GetName(i)] = EventCounter::global_counters[i].load(std::memory_order_relaxed);
for (size_t i = 0; i < memgraph::metrics::CounterEnd(); ++i) {
ret[memgraph::metrics::GetCounterName(i)] =
memgraph::metrics::global_counters[i].load(std::memory_order_relaxed);
}
return ret;
});
@@ -981,6 +1028,43 @@ int main(int argc, char **argv) {
{FLAGS_monitoring_address, static_cast<uint16_t>(FLAGS_monitoring_port)}, &context, websocket_auth};
AddLoggerSink(websocket_server.GetLoggingSink());
MonitoringServerT metrics_server{
{FLAGS_metrics_address, static_cast<uint16_t>(FLAGS_metrics_port)}, &session_data, &context};
#ifdef MG_ENTERPRISE
if (memgraph::license::global_license_checker.IsEnterpriseValidFast()) {
// Handler for regular termination signals
auto shutdown = [&metrics_server, &websocket_server, &server, &interpreter_context] {
// Server needs to be shutdown first and then the database. This prevents
// a race condition when a transaction is accepted during server shutdown.
server.Shutdown();
// After the server is notified to stop accepting and processing
// connections we tell the execution engine to stop processing all pending
// queries.
memgraph::query::Shutdown(&interpreter_context);
websocket_server.Shutdown();
metrics_server.Shutdown();
};
InitSignalHandlers(shutdown);
} else {
// Handler for regular termination signals
auto shutdown = [&websocket_server, &server, &interpreter_context] {
// Server needs to be shutdown first and then the database. This prevents
// a race condition when a transaction is accepted during server shutdown.
server.Shutdown();
// After the server is notified to stop accepting and processing
// connections we tell the execution engine to stop processing all pending
// queries.
memgraph::query::Shutdown(&interpreter_context);
websocket_server.Shutdown();
};
InitSignalHandlers(shutdown);
}
#else
// Handler for regular termination signals
auto shutdown = [&websocket_server, &server, &interpreter_context] {
// Server needs to be shutdown first and then the database. This prevents
@@ -990,14 +1074,22 @@ int main(int argc, char **argv) {
// connections we tell the execution engine to stop processing all pending
// queries.
memgraph::query::Shutdown(&interpreter_context);
websocket_server.Shutdown();
};
InitSignalHandlers(shutdown);
#endif
MG_ASSERT(server.Start(), "Couldn't start the Bolt server!");
websocket_server.Start();
#ifdef MG_ENTERPRISE
if (memgraph::license::global_license_checker.IsEnterpriseValidFast()) {
metrics_server.Start();
}
#endif
if (!FLAGS_init_data_file.empty()) {
spdlog::info("Running init data file.");
#ifdef MG_ENTERPRISE
@@ -1011,6 +1103,11 @@ int main(int argc, char **argv) {
server.AwaitShutdown();
websocket_server.AwaitShutdown();
#ifdef MG_ENTERPRISE
if (memgraph::license::global_license_checker.IsEnterpriseValidFast()) {
metrics_server.AwaitShutdown();
}
#endif
memgraph::query::procedure::gModuleRegistry.UnloadAllModules();

View File

@@ -22,6 +22,8 @@
#include "query/trigger.hpp"
#include "utils/async_timer.hpp"
#include "query/frame_change.hpp"
namespace memgraph::query {
enum class TransactionStatus {
@@ -82,6 +84,7 @@ struct ExecutionContext {
plan::ProfilingStats *stats_root{nullptr};
ExecutionStats execution_stats;
TriggerContextCollector *trigger_context_collector{nullptr};
FrameChangeCollector *frame_change_collector{nullptr};
utils::AsyncTimer timer;
#ifdef MG_ENTERPRISE
std::unique_ptr<FineGrainedAuthChecker> auth_checker{nullptr};

122
src/query/frame_change.hpp Normal file
View File

@@ -0,0 +1,122 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "query/typed_value.hpp"
#include "utils/memory.hpp"
#include "utils/pmr/unordered_map.hpp"
#include "utils/pmr/vector.hpp"
namespace memgraph::query {
// Key is hash output, value is vector of unique elements
using CachedType = utils::pmr::unordered_map<size_t, std::vector<TypedValue>>;
struct CachedValue {
// Cached value, this can be probably templateized
CachedType cache_;
explicit CachedValue(utils::MemoryResource *mem) : cache_(mem) {}
CachedValue(CachedType &&cache, memgraph::utils::MemoryResource *memory) : cache_(std::move(cache), memory) {}
CachedValue(const CachedValue &other, memgraph::utils::MemoryResource *memory) : cache_(other.cache_, memory) {}
CachedValue(CachedValue &&other, memgraph::utils::MemoryResource *memory) : cache_(std::move(other.cache_), memory) {}
CachedValue(CachedValue &&other) noexcept = delete;
/// Copy construction without memgraph::utils::MemoryResource is not allowed.
CachedValue(const CachedValue &) = delete;
CachedValue &operator=(const CachedValue &) = delete;
CachedValue &operator=(CachedValue &&) = delete;
~CachedValue() = default;
memgraph::utils::MemoryResource *GetMemoryResource() const noexcept {
return cache_.get_allocator().GetMemoryResource();
}
// Func to check if cache_ contains value
bool CacheValue(const TypedValue &value) {
if (!value.IsList()) {
return false;
}
const auto &list = value.ValueList();
TypedValue::Hash hash{};
for (const TypedValue &element : list) {
const auto key = hash(element);
auto &vector_values = cache_[key];
if (!IsValueInVec(vector_values, element)) {
vector_values.push_back(element);
}
}
return true;
}
// Func to cache_value inside cache_
bool ContainsValue(const TypedValue &value) const {
TypedValue::Hash hash{};
const auto key = hash(value);
if (cache_.contains(key)) {
return IsValueInVec(cache_.at(key), value);
}
return false;
}
private:
bool IsValueInVec(const std::vector<TypedValue> &vec_values, const TypedValue &value) const {
return std::any_of(vec_values.begin(), vec_values.end(), [&value](auto &vec_value) {
const auto is_value_equal = vec_value == value;
if (is_value_equal.IsNull()) return false;
return is_value_equal.ValueBool();
});
}
};
// Class tracks keys for which user can cache values which help with faster search or faster retrieval
// in the future.
class FrameChangeCollector {
public:
explicit FrameChangeCollector(utils::MemoryResource *mem) : tracked_values_(mem){};
// Add tracking key to cache later value
CachedValue &AddTrackingKey(const std::string &key) {
const auto &[it, _] = tracked_values_.emplace(key, tracked_values_.get_allocator().GetMemoryResource());
return it->second;
}
// Is key tracked
bool IsKeyTracked(const std::string &key) const { return tracked_values_.contains(key); }
// Is value for given key cached
bool IsKeyValueCached(const std::string &key) const {
return tracked_values_.contains(key) && !tracked_values_.at(key).cache_.empty();
}
// Reset value for tracking key
bool ResetTrackingValue(const std::string &key) {
if (tracked_values_.contains(key)) {
tracked_values_.erase(key);
AddTrackingKey(key);
}
return true;
}
// Get value cached for tracking key, throws if key is not in tracked
CachedValue &GetCachedValue(const std::string &key) { return tracked_values_.at(key); }
// Checks for keys tracked
bool IsTrackingValues() const { return !tracked_values_.empty(); }
private:
// Key is output of utils::GetFrameChangeId, value is utils::pmr::unordered_map
memgraph::utils::pmr::unordered_map<std::string, CachedValue> tracked_values_;
};
} // namespace memgraph::query

View File

@@ -114,12 +114,18 @@ constexpr utils::TypeInfo query::ListLiteral::kType{utils::TypeId::AST_LIST_LITE
constexpr utils::TypeInfo query::MapLiteral::kType{utils::TypeId::AST_MAP_LITERAL, "MapLiteral",
&query::BaseLiteral::kType};
constexpr utils::TypeInfo query::MapProjectionLiteral::kType{utils::TypeId::AST_MAP_PROJECTION_LITERAL,
"MapProjectionLiteral", &query::BaseLiteral::kType};
constexpr utils::TypeInfo query::Identifier::kType{utils::TypeId::AST_IDENTIFIER, "Identifier",
&query::Expression::kType};
constexpr utils::TypeInfo query::PropertyLookup::kType{utils::TypeId::AST_PROPERTY_LOOKUP, "PropertyLookup",
&query::Expression::kType};
constexpr utils::TypeInfo query::AllPropertiesLookup::kType{utils::TypeId::AST_ALL_PROPERTIES_LOOKUP,
"AllPropertiesLookup", &query::Expression::kType};
constexpr utils::TypeInfo query::LabelsTest::kType{utils::TypeId::AST_LABELS_TEST, "LabelsTest",
&query::Expression::kType};

View File

@@ -1063,8 +1063,9 @@ class MapLiteral : public memgraph::query::BaseLiteral {
DEFVISITABLE(ExpressionVisitor<void>);
bool Accept(HierarchicalTreeVisitor &visitor) override {
if (visitor.PreVisit(*this)) {
for (auto pair : elements_)
for (auto pair : elements_) {
if (!pair.second->Accept(visitor)) break;
}
}
return visitor.PostVisit(*this);
}
@@ -1087,6 +1088,60 @@ class MapLiteral : public memgraph::query::BaseLiteral {
friend class AstStorage;
};
struct MapProjectionData {
Expression *map_variable;
std::unordered_map<PropertyIx, Expression *> elements;
};
class MapProjectionLiteral : public memgraph::query::BaseLiteral {
public:
static const utils::TypeInfo kType;
const utils::TypeInfo &GetTypeInfo() const override { return kType; }
MapProjectionLiteral() = default;
DEFVISITABLE(ExpressionVisitor<TypedValue>);
DEFVISITABLE(ExpressionVisitor<TypedValue *>);
DEFVISITABLE(ExpressionVisitor<void>);
bool Accept(HierarchicalTreeVisitor &visitor) override {
if (visitor.PreVisit(*this)) {
for (auto pair : elements_) {
if (!pair.second) continue;
if (!pair.second->Accept(visitor)) break;
}
}
return visitor.PostVisit(*this);
}
Expression *map_variable_;
std::unordered_map<PropertyIx, Expression *> elements_;
MapProjectionLiteral *Clone(AstStorage *storage) const override {
MapProjectionLiteral *object = storage->Create<MapProjectionLiteral>();
object->map_variable_ = map_variable_;
for (const auto &entry : elements_) {
auto key = storage->GetPropertyIx(entry.first.name);
if (!entry.second) {
object->elements_[key] = nullptr;
continue;
}
object->elements_[key] = entry.second->Clone(storage);
}
return object;
}
protected:
explicit MapProjectionLiteral(Expression *map_variable, std::unordered_map<PropertyIx, Expression *> &&elements)
: map_variable_(map_variable), elements_(std::move(elements)) {}
private:
friend class AstStorage;
};
class Identifier : public memgraph::query::Expression {
public:
static const utils::TypeInfo kType;
@@ -1158,6 +1213,38 @@ class PropertyLookup : public memgraph::query::Expression {
friend class AstStorage;
};
class AllPropertiesLookup : public memgraph::query::Expression {
public:
static const utils::TypeInfo kType;
const utils::TypeInfo &GetTypeInfo() const override { return kType; }
AllPropertiesLookup() = default;
DEFVISITABLE(ExpressionVisitor<TypedValue>);
DEFVISITABLE(ExpressionVisitor<TypedValue *>);
DEFVISITABLE(ExpressionVisitor<void>);
bool Accept(HierarchicalTreeVisitor &visitor) override {
if (visitor.PreVisit(*this)) {
expression_->Accept(visitor);
}
return visitor.PostVisit(*this);
}
memgraph::query::Expression *expression_{nullptr};
AllPropertiesLookup *Clone(AstStorage *storage) const override {
AllPropertiesLookup *object = storage->Create<AllPropertiesLookup>();
object->expression_ = expression_ ? expression_->Clone(storage) : nullptr;
return object;
}
protected:
explicit AllPropertiesLookup(Expression *expression) : expression_(expression) {}
private:
friend class AstStorage;
};
class LabelsTest : public memgraph::query::Expression {
public:
static const utils::TypeInfo kType;
@@ -2786,7 +2873,7 @@ class InfoQuery : public memgraph::query::Query {
static const utils::TypeInfo kType;
const utils::TypeInfo &GetTypeInfo() const override { return kType; }
enum class InfoType { STORAGE, INDEX, CONSTRAINT };
enum class InfoType { STORAGE, INDEX, CONSTRAINT, BUILD };
DEFVISITABLE(QueryVisitor<void>);
@@ -2898,7 +2985,7 @@ class LockPathQuery : public memgraph::query::Query {
static const utils::TypeInfo kType;
const utils::TypeInfo &GetTypeInfo() const override { return kType; }
enum class Action { LOCK_PATH, UNLOCK_PATH };
enum class Action { LOCK_PATH, UNLOCK_PATH, STATUS };
LockPathQuery() = default;

View File

@@ -22,6 +22,7 @@ class CypherUnion;
class NamedExpression;
class Identifier;
class PropertyLookup;
class AllPropertiesLookup;
class LabelsTest;
class Aggregation;
class Function;
@@ -44,6 +45,7 @@ class EdgeAtom;
class PrimitiveLiteral;
class ListLiteral;
class MapLiteral;
class MapProjectionLiteral;
class OrOperator;
class XorOperator;
class AndOperator;
@@ -106,9 +108,10 @@ using TreeCompositeVisitor = utils::CompositeVisitor<
SubtractionOperator, MultiplicationOperator, DivisionOperator, ModOperator, NotEqualOperator, EqualOperator,
LessOperator, GreaterOperator, LessEqualOperator, GreaterEqualOperator, InListOperator, SubscriptOperator,
ListSlicingOperator, IfOperator, UnaryPlusOperator, UnaryMinusOperator, IsNullOperator, ListLiteral, MapLiteral,
PropertyLookup, LabelsTest, Aggregation, Function, Reduce, Coalesce, Extract, All, Single, Any, None, CallProcedure,
Create, Match, Return, With, Pattern, NodeAtom, EdgeAtom, Delete, Where, SetProperty, SetProperties, SetLabels,
RemoveProperty, RemoveLabels, Merge, Unwind, RegexMatch, LoadCsv, Foreach, Exists, CallSubquery, CypherQuery>;
MapProjectionLiteral, PropertyLookup, AllPropertiesLookup, LabelsTest, Aggregation, Function, Reduce, Coalesce,
Extract, All, Single, Any, None, CallProcedure, Create, Match, Return, With, Pattern, NodeAtom, EdgeAtom, Delete,
Where, SetProperty, SetProperties, SetLabels, RemoveProperty, RemoveLabels, Merge, Unwind, RegexMatch, LoadCsv,
Foreach, Exists, CallSubquery, CypherQuery>;
using TreeLeafVisitor = utils::LeafVisitor<Identifier, PrimitiveLiteral, ParameterLookup>;
@@ -122,13 +125,14 @@ class HierarchicalTreeVisitor : public TreeCompositeVisitor, public TreeLeafVisi
template <class TResult>
class ExpressionVisitor
: public utils::Visitor<
TResult, NamedExpression, OrOperator, XorOperator, AndOperator, NotOperator, AdditionOperator,
SubtractionOperator, MultiplicationOperator, DivisionOperator, ModOperator, NotEqualOperator, EqualOperator,
LessOperator, GreaterOperator, LessEqualOperator, GreaterEqualOperator, InListOperator, SubscriptOperator,
ListSlicingOperator, IfOperator, UnaryPlusOperator, UnaryMinusOperator, IsNullOperator, ListLiteral,
MapLiteral, PropertyLookup, LabelsTest, Aggregation, Function, Reduce, Coalesce, Extract, All, Single, Any,
None, ParameterLookup, Identifier, PrimitiveLiteral, RegexMatch, Exists> {};
: public utils::Visitor<TResult, NamedExpression, OrOperator, XorOperator, AndOperator, NotOperator,
AdditionOperator, SubtractionOperator, MultiplicationOperator, DivisionOperator,
ModOperator, NotEqualOperator, EqualOperator, LessOperator, GreaterOperator,
LessEqualOperator, GreaterEqualOperator, InListOperator, SubscriptOperator,
ListSlicingOperator, IfOperator, UnaryPlusOperator, UnaryMinusOperator, IsNullOperator,
ListLiteral, MapLiteral, MapProjectionLiteral, PropertyLookup, AllPropertiesLookup,
LabelsTest, Aggregation, Function, Reduce, Coalesce, Extract, All, Single, Any, None,
ParameterLookup, Identifier, PrimitiveLiteral, RegexMatch, Exists> {};
template <class TResult>
class QueryVisitor

View File

@@ -124,6 +124,9 @@ antlrcpp::Any CypherMainVisitor::visitInfoQuery(MemgraphCypher::InfoQueryContext
} else if (ctx->constraintInfo()) {
info_query->info_type_ = InfoQuery::InfoType::CONSTRAINT;
return info_query;
} else if (ctx->buildInfo()) {
info_query->info_type_ = InfoQuery::InfoType::BUILD;
return info_query;
} else {
throw utils::NotYetImplemented("Info query: '{}'", ctx->getText());
}
@@ -325,7 +328,9 @@ antlrcpp::Any CypherMainVisitor::visitShowReplicas(MemgraphCypher::ShowReplicasC
antlrcpp::Any CypherMainVisitor::visitLockPathQuery(MemgraphCypher::LockPathQueryContext *ctx) {
auto *lock_query = storage_->Create<LockPathQuery>();
if (ctx->LOCK()) {
if (ctx->STATUS()) {
lock_query->action_ = LockPathQuery::Action::STATUS;
} else if (ctx->LOCK()) {
lock_query->action_ = LockPathQuery::Action::LOCK_PATH;
} else if (ctx->UNLOCK()) {
lock_query->action_ = LockPathQuery::Action::UNLOCK_PATH;
@@ -1696,6 +1701,38 @@ antlrcpp::Any CypherMainVisitor::visitMapLiteral(MemgraphCypher::MapLiteralConte
return map;
}
antlrcpp::Any CypherMainVisitor::visitMapProjectionLiteral(MemgraphCypher::MapProjectionLiteralContext *ctx) {
MapProjectionData map_projection_data;
map_projection_data.map_variable =
storage_->Create<Identifier>(std::any_cast<std::string>(ctx->variable()->accept(this)));
for (auto *map_el : ctx->mapElement()) {
if (map_el->propertyLookup()) {
auto key = std::any_cast<PropertyIx>(map_el->propertyLookup()->propertyKeyName()->accept(this));
auto property = std::any_cast<PropertyIx>(map_el->propertyLookup()->accept(this));
auto *property_lookup = storage_->Create<PropertyLookup>(map_projection_data.map_variable, property);
map_projection_data.elements.insert_or_assign(key, property_lookup);
}
if (map_el->allPropertiesLookup()) {
auto key = AddProperty("*");
auto *all_properties_lookup = storage_->Create<AllPropertiesLookup>(map_projection_data.map_variable);
map_projection_data.elements.insert_or_assign(key, all_properties_lookup);
}
if (map_el->variable()) {
auto key = AddProperty(std::any_cast<std::string>(map_el->variable()->accept(this)));
auto *variable = storage_->Create<Identifier>(std::any_cast<std::string>(map_el->variable()->accept(this)));
map_projection_data.elements.insert_or_assign(key, variable);
}
if (map_el->propertyKeyValuePair()) {
auto key = std::any_cast<PropertyIx>(map_el->propertyKeyValuePair()->propertyKeyName()->accept(this));
auto *value = std::any_cast<Expression *>(map_el->propertyKeyValuePair()->expression()->accept(this));
map_projection_data.elements.insert_or_assign(key, value);
}
}
return map_projection_data;
}
antlrcpp::Any CypherMainVisitor::visitListLiteral(MemgraphCypher::ListLiteralContext *ctx) {
std::vector<Expression *> expressions;
for (auto *expr_ctx : ctx->expression()) {
@@ -2276,6 +2313,10 @@ antlrcpp::Any CypherMainVisitor::visitLiteral(MemgraphCypher::LiteralContext *ct
} else if (ctx->listLiteral()) {
return static_cast<Expression *>(
storage_->Create<ListLiteral>(std::any_cast<std::vector<Expression *>>(ctx->listLiteral()->accept(this))));
} else if (ctx->mapProjectionLiteral()) {
auto map_projection_data = std::any_cast<MapProjectionData>(ctx->mapProjectionLiteral()->accept(this));
return static_cast<Expression *>(storage_->Create<MapProjectionLiteral>(map_projection_data.map_variable,
std::move(map_projection_data.elements)));
} else {
return static_cast<Expression *>(storage_->Create<MapLiteral>(
std::any_cast<std::unordered_map<PropertyIx, Expression *>>(ctx->mapLiteral()->accept(this))));

View File

@@ -15,8 +15,6 @@
#include <unordered_set>
#include <utility>
#include <antlr4-runtime.h>
#include "query/frontend/ast/ast.hpp"
#include "query/frontend/opencypher/generated/MemgraphCypherBaseVisitor.h"
#include "utils/exceptions.hpp"
@@ -608,6 +606,11 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
*/
antlrcpp::Any visitMapLiteral(MemgraphCypher::MapLiteralContext *ctx) override;
/**
* @return MapProjectionData
*/
antlrcpp::Any visitMapProjectionLiteral(MemgraphCypher::MapProjectionLiteralContext *ctx) override;
/**
* @return vector<Expression*>
*/

View File

@@ -54,6 +54,7 @@ class ExpressionPrettyPrinter : public ExpressionVisitor<void> {
void Visit(IfOperator &op) override;
void Visit(ListLiteral &op) override;
void Visit(MapLiteral &op) override;
void Visit(MapProjectionLiteral &op) override;
void Visit(LabelsTest &op) override;
void Visit(Aggregation &op) override;
void Visit(Function &op) override;
@@ -68,6 +69,7 @@ class ExpressionPrettyPrinter : public ExpressionVisitor<void> {
void Visit(Identifier &op) override;
void Visit(PrimitiveLiteral &op) override;
void Visit(PropertyLookup &op) override;
void Visit(AllPropertiesLookup &op) override;
void Visit(ParameterLookup &op) override;
void Visit(NamedExpression &op) override;
void Visit(RegexMatch &op) override;
@@ -89,6 +91,8 @@ void PrintObject(std::ostream *out, Aggregation::Op op);
void PrintObject(std::ostream *out, Expression *expr);
void PrintObject(std::ostream *out, AllPropertiesLookup *apl);
void PrintObject(std::ostream *out, Identifier *expr);
void PrintObject(std::ostream *out, const storage::PropertyValue &value);
@@ -122,6 +126,15 @@ void PrintObject(std::ostream *out, Expression *expr) {
}
}
void PrintObject(std::ostream *out, AllPropertiesLookup *apl) {
if (apl) {
ExpressionPrettyPrinter printer{out};
*out << ".*";
} else {
*out << "<null>";
}
}
void PrintObject(std::ostream *out, Identifier *expr) { PrintObject(out, static_cast<Expression *>(expr)); }
void PrintObject(std::ostream *out, const storage::PropertyValue &value) {
@@ -249,6 +262,17 @@ void ExpressionPrettyPrinter::Visit(MapLiteral &op) {
PrintObject(out_, map);
}
void ExpressionPrettyPrinter::Visit(MapProjectionLiteral &op) {
std::map<std::string, Expression *> map_projection_elements;
for (const auto &kv : op.elements_) {
map_projection_elements[kv.first.name] = kv.second;
}
PrintObject(out_, op.map_variable_);
PrintObject(out_, map_projection_elements);
}
void ExpressionPrettyPrinter::Visit(AllPropertiesLookup &op) { PrintObject(out_, &op); }
void ExpressionPrettyPrinter::Visit(LabelsTest &op) { PrintOperator(out_, "LabelsTest", op.expression_); }
void ExpressionPrettyPrinter::Visit(Aggregation &op) { PrintOperator(out_, "Aggregation", op.op_); }

View File

@@ -46,7 +46,9 @@ indexInfo : INDEX INFO ;
constraintInfo : CONSTRAINT INFO ;
infoQuery : SHOW ( storageInfo | indexInfo | constraintInfo ) ;
buildInfo : BUILD INFO ;
infoQuery : SHOW ( storageInfo | indexInfo | constraintInfo | buildInfo) ;
explainQuery : EXPLAIN cypherQuery ;
@@ -248,6 +250,7 @@ literal : numberLiteral
| booleanLiteral
| CYPHERNULL
| mapLiteral
| mapProjectionLiteral
| listLiteral
;
@@ -290,6 +293,8 @@ patternComprehension : '[' ( variable '=' )? relationshipsPattern ( WHERE expres
propertyLookup : '.' ( propertyKeyName ) ;
allPropertiesLookup : '.' '*' ;
caseExpression : ( ( CASE ( caseAlternatives )+ ) | ( CASE test=expression ( caseAlternatives )+ ) ) ( ELSE else_expression=expression )? END ;
caseAlternatives : WHEN when_expression=expression THEN then_expression=expression ;
@@ -302,12 +307,22 @@ numberLiteral : doubleLiteral
mapLiteral : '{' ( propertyKeyName ':' expression ( ',' propertyKeyName ':' expression )* )? '}' ;
mapProjectionLiteral : variable '{' ( mapElement ( ',' mapElement )* )? '}' ;
mapElement : propertyLookup
| allPropertiesLookup
| variable
| propertyKeyValuePair
;
parameter : '$' ( symbolicName | DecimalLiteral ) ;
propertyExpression : atom ( propertyLookup )+ ;
propertyKeyName : symbolicName ;
propertyKeyValuePair : propertyKeyName ':' expression ;
integerLiteral : DecimalLiteral
| OctalLiteral
| HexadecimalLiteral

View File

@@ -31,6 +31,7 @@ memgraphCypherKeyword : cypherKeyword
| BATCH_SIZE
| BEFORE
| BOOTSTRAP_SERVERS
| BUILD
| CHECK
| CLEAR
| COMMIT
@@ -90,6 +91,7 @@ memgraphCypherKeyword : cypherKeyword
| SNAPSHOT
| START
| STATS
| STATUS
| STORAGE
| STREAM
| STREAMS
@@ -334,7 +336,7 @@ dropReplica : DROP REPLICA replicaName ;
showReplicas : SHOW REPLICAS ;
lockPathQuery : ( LOCK | UNLOCK ) DATA DIRECTORY ;
lockPathQuery : ( LOCK | UNLOCK ) DATA DIRECTORY | DATA DIRECTORY LOCK STATUS;
freeMemoryQuery : FREE MEMORY ;

View File

@@ -35,6 +35,7 @@ BATCH_INTERVAL : B A T C H UNDERSCORE I N T E R V A L ;
BATCH_LIMIT : B A T C H UNDERSCORE L I M I T ;
BATCH_SIZE : B A T C H UNDERSCORE S I Z E ;
BEFORE : B E F O R E ;
BUILD : B U I L D ;
BOOTSTRAP_SERVERS : B O O T S T R A P UNDERSCORE S E R V E R S ;
CALL : C A L L ;
CHECK : C H E C K ;
@@ -106,6 +107,7 @@ SNAPSHOT : S N A P S H O T ;
START : S T A R T ;
STATISTICS : S T A T I S T I C S ;
STATS : S T A T S ;
STATUS : S T A T U S ;
STOP : S T O P ;
STORAGE : S T O R A G E;
STORAGE_MODE : S T O R A G E UNDERSCORE MODE;

View File

@@ -43,6 +43,7 @@ class PrivilegeExtractor : public QueryVisitor<void>, public HierarchicalTreeVis
AddPrivilege(AuthQuery::Privilege::INDEX);
break;
case InfoQuery::InfoType::STORAGE:
case InfoQuery::InfoType::BUILD:
AddPrivilege(AuthQuery::Privilege::STATS);
break;
case InfoQuery::InfoType::CONSTRAINT:

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -145,6 +145,7 @@ const trie::Trie kKeywords = {"union",
"drop",
"show",
"stats",
"status",
"unique",
"explain",
"profile",
@@ -211,7 +212,12 @@ const trie::Trie kKeywords = {"union",
"edge_types",
"off",
"in_memory_transactional",
"in_memory_analytical"};
"in_memory_analytical",
"data",
"directory",
"lock",
"unlock"
"build"};
// Unicode codepoints that are allowed at the start of the unescaped name.
const std::bitset<kBitsetSize> kUnescapedNameAllowedStarts(

View File

@@ -13,10 +13,12 @@
#pragma once
#include <algorithm>
#include <cstddef>
#include <limits>
#include <map>
#include <optional>
#include <regex>
#include <string>
#include <vector>
#include "query/common.hpp"
@@ -27,7 +29,11 @@
#include "query/frontend/semantic/symbol_table.hpp"
#include "query/interpret/frame.hpp"
#include "query/typed_value.hpp"
#include "spdlog/spdlog.h"
#include "utils/exceptions.hpp"
#include "utils/frame_change_id.hpp"
#include "utils/logging.hpp"
#include "utils/pmr/unordered_map.hpp"
namespace memgraph::query {
@@ -73,11 +79,13 @@ class ReferenceExpressionEvaluator : public ExpressionVisitor<TypedValue *> {
UNSUCCESSFUL_VISIT(ListSlicingOperator);
UNSUCCESSFUL_VISIT(IsNullOperator);
UNSUCCESSFUL_VISIT(PropertyLookup);
UNSUCCESSFUL_VISIT(AllPropertiesLookup);
UNSUCCESSFUL_VISIT(LabelsTest);
UNSUCCESSFUL_VISIT(PrimitiveLiteral);
UNSUCCESSFUL_VISIT(ListLiteral);
UNSUCCESSFUL_VISIT(MapLiteral);
UNSUCCESSFUL_VISIT(MapProjectionLiteral);
UNSUCCESSFUL_VISIT(Aggregation);
UNSUCCESSFUL_VISIT(Coalesce);
UNSUCCESSFUL_VISIT(Function);
@@ -100,8 +108,13 @@ class ReferenceExpressionEvaluator : public ExpressionVisitor<TypedValue *> {
class ExpressionEvaluator : public ExpressionVisitor<TypedValue> {
public:
ExpressionEvaluator(Frame *frame, const SymbolTable &symbol_table, const EvaluationContext &ctx, DbAccessor *dba,
storage::View view)
: frame_(frame), symbol_table_(&symbol_table), ctx_(&ctx), dba_(dba), view_(view) {}
storage::View view, FrameChangeCollector *frame_change_collector = nullptr)
: frame_(frame),
symbol_table_(&symbol_table),
ctx_(&ctx),
dba_(dba),
view_(view),
frame_change_collector_(frame_change_collector) {}
using ExpressionVisitor<TypedValue>::Visit;
@@ -190,25 +203,78 @@ class ExpressionEvaluator : public ExpressionVisitor<TypedValue> {
}
TypedValue Visit(InListOperator &in_list) override {
TypedValue *_list_ptr = nullptr;
TypedValue _list;
auto literal = in_list.expression1_->Accept(*this);
auto _list = in_list.expression2_->Accept(*this);
if (_list.IsNull()) {
return TypedValue(ctx_->memory);
auto get_list_literal = [this, &in_list, &_list, &_list_ptr]() -> void {
ReferenceExpressionEvaluator reference_expression_evaluator{frame_, symbol_table_, ctx_};
_list_ptr = in_list.expression2_->Accept(reference_expression_evaluator);
if (nullptr == _list_ptr) {
_list = in_list.expression2_->Accept(*this);
_list_ptr = &_list;
}
};
auto do_list_literal_checks = [this, &literal, &_list_ptr]() -> std::optional<TypedValue> {
MG_ASSERT(_list_ptr, "List literal should have been defined");
if (_list_ptr->IsNull()) {
return TypedValue(ctx_->memory);
}
// Exceptions have higher priority than returning nulls when list expression
// is not null.
if (_list_ptr->type() != TypedValue::Type::List) {
throw QueryRuntimeException("IN expected a list, got {}.", _list_ptr->type());
}
const auto &list = _list_ptr->ValueList();
// If literal is NULL there is no need to try to compare it with every
// element in the list since result of every comparison will be NULL. There
// is one special case that we must test explicitly: if list is empty then
// result is false since no comparison will be performed.
if (list.empty()) return TypedValue(false, ctx_->memory);
if (literal.IsNull()) return TypedValue(ctx_->memory);
return {};
};
const auto cached_id = memgraph::utils::GetFrameChangeId(in_list);
const auto do_cache{frame_change_collector_ != nullptr && cached_id &&
frame_change_collector_->IsKeyTracked(*cached_id)};
if (do_cache) {
if (!frame_change_collector_->IsKeyValueCached(*cached_id)) {
// Check only first time if everything is okay, later when we use
// cache there is no need to check again as we did check first time
get_list_literal();
auto preoperational_checks = do_list_literal_checks();
if (preoperational_checks) {
return std::move(*preoperational_checks);
}
auto &cached_value = frame_change_collector_->GetCachedValue(*cached_id);
cached_value.CacheValue(*_list_ptr);
spdlog::trace("Value cached {}", *cached_id);
}
const auto &cached_value = frame_change_collector_->GetCachedValue(*cached_id);
if (cached_value.ContainsValue(literal)) {
return TypedValue(true, ctx_->memory);
}
// has null
if (cached_value.ContainsValue(TypedValue(ctx_->memory))) {
return TypedValue(ctx_->memory);
}
return TypedValue(false, ctx_->memory);
}
// Exceptions have higher priority than returning nulls when list expression
// is not null.
if (_list.type() != TypedValue::Type::List) {
throw QueryRuntimeException("IN expected a list, got {}.", _list.type());
// When caching is not an option, we need to evaluate list literal every time
// and do the checks
get_list_literal();
auto preoperational_checks = do_list_literal_checks();
if (preoperational_checks) {
return std::move(*preoperational_checks);
}
const auto &list = _list.ValueList();
// If literal is NULL there is no need to try to compare it with every
// element in the list since result of every comparison will be NULL. There
// is one special case that we must test explicitly: if list is empty then
// result is false since no comparison will be performed.
if (list.empty()) return TypedValue(false, ctx_->memory);
if (literal.IsNull()) return TypedValue(ctx_->memory);
spdlog::trace("Not using cache on IN LIST operator");
auto has_null = false;
for (const auto &element : list) {
auto result = literal == element;
@@ -466,7 +532,101 @@ class ExpressionEvaluator : public ExpressionVisitor<TypedValue> {
throw QueryRuntimeException("Invalid property name {} for Graph", prop_name);
}
default:
throw QueryRuntimeException("Only nodes, edges, maps and temporal types have properties to be looked-up.");
throw QueryRuntimeException(
"Only nodes, edges, maps, temporal types and graphs have properties to be looked up.");
}
}
TypedValue Visit(AllPropertiesLookup &all_properties_lookup) override {
TypedValue::TMap result(ctx_->memory);
auto expression_result = all_properties_lookup.expression_->Accept(*this);
switch (expression_result.type()) {
case TypedValue::Type::Null:
return TypedValue(ctx_->memory);
case TypedValue::Type::Vertex: {
for (const auto properties = *expression_result.ValueVertex().Properties(view_);
const auto &[property_id, value] : properties) {
result.emplace(dba_->PropertyToName(property_id), value);
}
return TypedValue(result, ctx_->memory);
}
case TypedValue::Type::Edge: {
for (const auto properties = *expression_result.ValueEdge().Properties(view_);
const auto &[property_id, value] : properties) {
result.emplace(dba_->PropertyToName(property_id), value);
}
return TypedValue(result, ctx_->memory);
}
case TypedValue::Type::Map: {
for (auto &[name, value] : expression_result.ValueMap()) {
result.emplace(name, value);
}
return TypedValue(result, ctx_->memory);
}
case TypedValue::Type::Duration: {
const auto &dur = expression_result.ValueDuration();
result.emplace("day", TypedValue(dur.Days(), ctx_->memory));
result.emplace("hour", TypedValue(dur.SubDaysAsHours(), ctx_->memory));
result.emplace("minute", TypedValue(dur.SubDaysAsMinutes(), ctx_->memory));
result.emplace("second", TypedValue(dur.SubDaysAsSeconds(), ctx_->memory));
result.emplace("millisecond", TypedValue(dur.SubDaysAsMilliseconds(), ctx_->memory));
result.emplace("microseconds", TypedValue(dur.SubDaysAsMicroseconds(), ctx_->memory));
result.emplace("nanoseconds", TypedValue(dur.SubDaysAsNanoseconds(), ctx_->memory));
return TypedValue(result, ctx_->memory);
}
case TypedValue::Type::Date: {
const auto &date = expression_result.ValueDate();
result.emplace("year", TypedValue(date.year, ctx_->memory));
result.emplace("month", TypedValue(date.month, ctx_->memory));
result.emplace("day", TypedValue(date.day, ctx_->memory));
return TypedValue(result, ctx_->memory);
}
case TypedValue::Type::LocalTime: {
const auto &lt = expression_result.ValueLocalTime();
result.emplace("hour", TypedValue(lt.hour, ctx_->memory));
result.emplace("minute", TypedValue(lt.minute, ctx_->memory));
result.emplace("second", TypedValue(lt.second, ctx_->memory));
result.emplace("millisecond", TypedValue(lt.millisecond, ctx_->memory));
result.emplace("microsecond", TypedValue(lt.microsecond, ctx_->memory));
return TypedValue(result, ctx_->memory);
}
case TypedValue::Type::LocalDateTime: {
const auto &ldt = expression_result.ValueLocalDateTime();
const auto &date = ldt.date;
const auto &lt = ldt.local_time;
result.emplace("year", TypedValue(date.year, ctx_->memory));
result.emplace("month", TypedValue(date.month, ctx_->memory));
result.emplace("day", TypedValue(date.day, ctx_->memory));
result.emplace("hour", TypedValue(lt.hour, ctx_->memory));
result.emplace("minute", TypedValue(lt.minute, ctx_->memory));
result.emplace("second", TypedValue(lt.second, ctx_->memory));
result.emplace("millisecond", TypedValue(lt.millisecond, ctx_->memory));
result.emplace("microsecond", TypedValue(lt.microsecond, ctx_->memory));
return TypedValue(result, ctx_->memory);
}
case TypedValue::Type::Graph: {
const auto &graph = expression_result.ValueGraph();
utils::pmr::vector<TypedValue> vertices(ctx_->memory);
vertices.reserve(graph.vertices().size());
for (const auto &v : graph.vertices()) {
vertices.emplace_back(TypedValue(v, ctx_->memory));
}
result.emplace("nodes", TypedValue(std::move(vertices), ctx_->memory));
utils::pmr::vector<TypedValue> edges(ctx_->memory);
edges.reserve(graph.edges().size());
for (const auto &e : graph.edges()) {
edges.emplace_back(TypedValue(e, ctx_->memory));
}
result.emplace("edges", TypedValue(std::move(edges), ctx_->memory));
return TypedValue(result, ctx_->memory);
}
default:
throw QueryRuntimeException(
"Only nodes, edges, maps, temporal types and graphs have properties to be looked up.");
}
}
@@ -531,6 +691,30 @@ class ExpressionEvaluator : public ExpressionVisitor<TypedValue> {
return TypedValue(result, ctx_->memory);
}
TypedValue Visit(MapProjectionLiteral &literal) override {
constexpr std::string_view kAllPropertiesSelector{"*"};
TypedValue::TMap result(ctx_->memory);
TypedValue::TMap all_properties_lookup(ctx_->memory);
for (const auto &[property_key, property_value] : literal.elements_) {
if (property_key.name == kAllPropertiesSelector.data()) {
auto maybe_all_properties_lookup = property_value->Accept(*this);
if (maybe_all_properties_lookup.type() != TypedValue::Type::Map) {
throw QueryRuntimeException("Expected a map from AllPropertiesLookup, got {}.",
maybe_all_properties_lookup.type());
}
all_properties_lookup = std::move(maybe_all_properties_lookup.ValueMap());
continue;
}
result.emplace(property_key.name, property_value->Accept(*this));
}
if (!all_properties_lookup.empty()) result.merge(all_properties_lookup);
return TypedValue(result, ctx_->memory);
}
TypedValue Visit(Aggregation &aggregation) override {
return TypedValue(frame_->at(symbol_table_->at(aggregation)), ctx_->memory);
}
@@ -852,7 +1036,8 @@ class ExpressionEvaluator : public ExpressionVisitor<TypedValue> {
DbAccessor *dba_;
// which switching approach should be used when evaluating
storage::View view_;
};
FrameChangeCollector *frame_change_collector_;
}; // namespace memgraph::query
/// A helper function for evaluating an expression that's an int.
///

View File

@@ -15,6 +15,7 @@
#include <algorithm>
#include <atomic>
#include <chrono>
#include <concepts>
#include <cstddef>
#include <cstdint>
#include <functional>
@@ -44,19 +45,26 @@
#include "query/frontend/semantic/required_privileges.hpp"
#include "query/frontend/semantic/symbol_generator.hpp"
#include "query/interpret/eval.hpp"
#include "query/interpret/frame.hpp"
#include "query/metadata.hpp"
#include "query/plan/planner.hpp"
#include "query/plan/profile.hpp"
#include "query/plan/vertex_count_cache.hpp"
#include "query/stream.hpp"
#include "query/stream/common.hpp"
#include "query/trigger.hpp"
#include "query/typed_value.hpp"
#include "spdlog/spdlog.h"
#include "storage/v2/edge.hpp"
#include "storage/v2/id_types.hpp"
#include "storage/v2/isolation_level.hpp"
#include "storage/v2/property_value.hpp"
#include "storage/v2/storage_mode.hpp"
#include "utils/algorithm.hpp"
#include "utils/build_info.hpp"
#include "utils/csv_parsing.hpp"
#include "utils/event_counter.hpp"
#include "utils/event_histogram.hpp"
#include "utils/exceptions.hpp"
#include "utils/flag_validation.hpp"
#include "utils/likely.hpp"
@@ -71,17 +79,20 @@
#include "utils/typeinfo.hpp"
#include "utils/variant_helpers.hpp"
namespace EventCounter {
namespace memgraph::metrics {
extern Event ReadQuery;
extern Event WriteQuery;
extern Event ReadWriteQuery;
extern const Event LabelIndexCreated;
extern const Event LabelPropertyIndexCreated;
extern const Event StreamsCreated;
extern const Event TriggersCreated;
} // namespace EventCounter
extern const Event QueryExecutionLatency_us;
extern const Event CommitedTransactions;
extern const Event RollbackedTransactions;
extern const Event ActiveTransactions;
} // namespace memgraph::metrics
namespace memgraph::query {
@@ -92,19 +103,29 @@ namespace {
void UpdateTypeCount(const plan::ReadWriteTypeChecker::RWType type) {
switch (type) {
case plan::ReadWriteTypeChecker::RWType::R:
EventCounter::IncrementCounter(EventCounter::ReadQuery);
memgraph::metrics::IncrementCounter(memgraph::metrics::ReadQuery);
break;
case plan::ReadWriteTypeChecker::RWType::W:
EventCounter::IncrementCounter(EventCounter::WriteQuery);
memgraph::metrics::IncrementCounter(memgraph::metrics::WriteQuery);
break;
case plan::ReadWriteTypeChecker::RWType::RW:
EventCounter::IncrementCounter(EventCounter::ReadWriteQuery);
memgraph::metrics::IncrementCounter(memgraph::metrics::ReadWriteQuery);
break;
default:
break;
}
}
template <typename T>
concept HasEmpty = requires(T t) {
{ t.empty() } -> std::convertible_to<bool>;
};
template <typename T>
inline std::optional<T> GenOptional(const T &in) {
return in.empty() ? std::nullopt : std::make_optional<T>(in);
}
struct Callback {
std::vector<std::string> header;
using CallbackFunction = std::function<std::vector<std::vector<TypedValue>>()>;
@@ -663,6 +684,8 @@ Callback::CallbackFunction GetKafkaCreateCallback(StreamQuery *stream_query, Exp
return config_map;
};
memgraph::metrics::IncrementCounter(memgraph::metrics::StreamsCreated);
return [interpreter_context, stream_name = stream_query->stream_name_,
topic_names = EvaluateTopicNames(evaluator, stream_query->topic_names_),
consumer_group = std::move(consumer_group), common_stream_info = std::move(common_stream_info),
@@ -693,6 +716,8 @@ Callback::CallbackFunction GetPulsarCreateCallback(StreamQuery *stream_query, Ex
throw SemanticException("Service URL must not be an empty string!");
}
auto common_stream_info = GetCommonStreamInfo(stream_query, evaluator);
memgraph::metrics::IncrementCounter(memgraph::metrics::StreamsCreated);
return [interpreter_context, stream_name = stream_query->stream_name_,
topic_names = EvaluateTopicNames(evaluator, stream_query->topic_names_),
common_stream_info = std::move(common_stream_info), service_url = std::move(service_url),
@@ -723,7 +748,6 @@ Callback HandleStreamQuery(StreamQuery *stream_query, const Parameters &paramete
Callback callback;
switch (stream_query->action_) {
case StreamQuery::Action::CREATE_STREAM: {
EventCounter::IncrementCounter(EventCounter::StreamsCreated);
switch (stream_query->type_) {
case StreamQuery::Type::KAFKA:
callback.fn = GetKafkaCreateCallback(stream_query, evaluator, interpreter_context, username);
@@ -983,7 +1007,8 @@ struct PullPlan {
DbAccessor *dba, InterpreterContext *interpreter_context, utils::MemoryResource *execution_memory,
std::optional<std::string> username, std::atomic<TransactionStatus> *transaction_status,
TriggerContextCollector *trigger_context_collector = nullptr,
std::optional<size_t> memory_limit = {}, bool use_monotonic_memory = true);
std::optional<size_t> memory_limit = {}, bool use_monotonic_memory = true,
FrameChangeCollector *frame_change_collector_ = nullptr);
std::optional<plan::ProfilingStatsWithTotalTime> Pull(AnyStream *stream, std::optional<int> n,
const std::vector<Symbol> &output_symbols,
@@ -1021,7 +1046,7 @@ PullPlan::PullPlan(const std::shared_ptr<CachedPlan> plan, const Parameters &par
DbAccessor *dba, InterpreterContext *interpreter_context, utils::MemoryResource *execution_memory,
std::optional<std::string> username, std::atomic<TransactionStatus> *transaction_status,
TriggerContextCollector *trigger_context_collector, const std::optional<size_t> memory_limit,
bool use_monotonic_memory)
bool use_monotonic_memory, FrameChangeCollector *frame_change_collector)
: plan_(plan),
cursor_(plan->plan().MakeCursor(execution_memory)),
frame_(plan->symbol_table().max_position(), execution_memory),
@@ -1052,6 +1077,7 @@ PullPlan::PullPlan(const std::shared_ptr<CachedPlan> plan, const Parameters &par
ctx_.transaction_status = transaction_status;
ctx_.is_profile_query = is_profile_query;
ctx_.trigger_context_collector = trigger_context_collector;
ctx_.frame_change_collector = frame_change_collector;
}
std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *stream, std::optional<int> n,
@@ -1066,16 +1092,18 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *strea
utils::ResourceWithOutOfMemoryException resource_with_exception;
utils::MonotonicBufferResource monotonic_memory{&stack_data[0], stack_size, &resource_with_exception};
std::optional<utils::PoolResource> pool_memory;
static constexpr auto kMaxBlockPerChunks = 128;
if (!use_monotonic_memory_) {
pool_memory.emplace(8, kExecutionPoolMaxBlockSize, utils::NewDeleteResource(), utils::NewDeleteResource());
pool_memory.emplace(kMaxBlockPerChunks, kExecutionPoolMaxBlockSize, &resource_with_exception,
&resource_with_exception);
} else {
// We can throw on every query because a simple queries for deleting will use only
// the stack allocated buffer.
// Also, we want to throw only when the query engine requests more memory and not the storage
// so we add the exception to the allocator.
// TODO (mferencevic): Tune the parameters accordingly.
pool_memory.emplace(128, 1024, &monotonic_memory, utils::NewDeleteResource());
pool_memory.emplace(kMaxBlockPerChunks, 1024, &monotonic_memory, &resource_with_exception);
}
std::optional<utils::LimitedMemoryResource> maybe_limited_resource;
@@ -1132,7 +1160,11 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *strea
if (has_unsent_results_) {
return std::nullopt;
}
summary->insert_or_assign("plan_execution_time", execution_time_.count());
memgraph::metrics::Measure(memgraph::metrics::QueryExecutionLatency_us,
std::chrono::duration_cast<std::chrono::microseconds>(execution_time_).count());
// We are finished with pulling all the data, therefore we can send any
// metadata about the results i.e. notifications and statistics
const bool is_any_counter_set =
@@ -1161,16 +1193,23 @@ Interpreter::Interpreter(InterpreterContext *interpreter_context) : interpreter_
MG_ASSERT(interpreter_context_, "Interpreter context must not be NULL");
}
PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper) {
PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper,
const std::map<std::string, storage::PropertyValue> &metadata) {
std::function<void()> handler;
if (query_upper == "BEGIN") {
handler = [this] {
// TODO: Evaluate doing move(metadata). Currently the metadata is very small, but this will be important if it ever
// becomes large.
handler = [this, metadata] {
if (in_explicit_transaction_) {
throw ExplicitTransactionUsageException("Nested transactions are not supported.");
}
memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveTransactions);
in_explicit_transaction_ = true;
expect_rollback_ = false;
metadata_ = GenOptional(metadata);
db_accessor_ =
std::make_unique<storage::Storage::Accessor>(interpreter_context_->db->Access(GetIsolationLevelOverride()));
@@ -1201,15 +1240,20 @@ PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper)
expect_rollback_ = false;
in_explicit_transaction_ = false;
metadata_ = std::nullopt;
};
} else if (query_upper == "ROLLBACK") {
handler = [this] {
if (!in_explicit_transaction_) {
throw ExplicitTransactionUsageException("No current transaction to rollback.");
}
memgraph::metrics::IncrementCounter(memgraph::metrics::RollbackedTransactions);
Abort();
expect_rollback_ = false;
in_explicit_transaction_ = false;
metadata_ = std::nullopt;
};
} else {
LOG_FATAL("Should not get here -- unknown transaction query!");
@@ -1224,11 +1268,28 @@ PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper)
RWType::NONE};
}
inline static void TryCaching(const AstStorage &ast_storage, FrameChangeCollector *frame_change_collector) {
if (!frame_change_collector) return;
for (const auto &tree : ast_storage.storage_) {
if (tree->GetTypeInfo() != memgraph::query::InListOperator::kType) {
continue;
}
auto *in_list_operator = utils::Downcast<InListOperator>(tree.get());
const auto cached_id = memgraph::utils::GetFrameChangeId(*in_list_operator);
if (!cached_id || cached_id->empty()) {
continue;
}
frame_change_collector->AddTrackingKey(*cached_id);
spdlog::trace("Tracking {} operator, by id: {}", InListOperator::kType.name, *cached_id);
}
}
PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string, TypedValue> *summary,
InterpreterContext *interpreter_context, DbAccessor *dba,
utils::MemoryResource *execution_memory, std::vector<Notification> *notifications,
const std::string *username, std::atomic<TransactionStatus> *transaction_status,
TriggerContextCollector *trigger_context_collector = nullptr) {
TriggerContextCollector *trigger_context_collector = nullptr,
FrameChangeCollector *frame_change_collector = nullptr) {
auto *cypher_query = utils::Downcast<CypherQuery>(parsed_query.query);
Frame frame(0);
@@ -1259,6 +1320,7 @@ PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string,
parsed_query.parameters,
parsed_query.is_cacheable ? &interpreter_context->plan_cache : nullptr, dba);
TryCaching(plan->ast_storage(), frame_change_collector);
summary->insert_or_assign("cost_estimate", plan->cost());
auto rw_type_checker = plan::ReadWriteTypeChecker();
rw_type_checker.InferRWType(const_cast<plan::LogicalOperator &>(plan->plan()));
@@ -1275,9 +1337,10 @@ PreparedQuery PrepareCypherQuery(ParsedQuery parsed_query, std::map<std::string,
header.push_back(
utils::FindOr(parsed_query.stripped_query.named_expressions(), symbol.token_position(), symbol.name()).first);
}
auto pull_plan = std::make_shared<PullPlan>(plan, parsed_query.parameters, false, dba, interpreter_context,
execution_memory, StringPointerToOptional(username), transaction_status,
trigger_context_collector, memory_limit, use_monotonic_memory);
auto pull_plan = std::make_shared<PullPlan>(
plan, parsed_query.parameters, false, dba, interpreter_context, execution_memory,
StringPointerToOptional(username), transaction_status, trigger_context_collector, memory_limit,
use_monotonic_memory, frame_change_collector->IsTrackingValues() ? frame_change_collector : nullptr);
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> {
@@ -1338,7 +1401,8 @@ PreparedQuery PrepareExplainQuery(ParsedQuery parsed_query, std::map<std::string
PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_transaction,
std::map<std::string, TypedValue> *summary, InterpreterContext *interpreter_context,
DbAccessor *dba, utils::MemoryResource *execution_memory, const std::string *username,
std::atomic<TransactionStatus> *transaction_status) {
std::atomic<TransactionStatus> *transaction_status,
FrameChangeCollector *frame_change_collector) {
const std::string kProfileQueryStart = "profile ";
MG_ASSERT(utils::StartsWith(utils::ToLowerCase(parsed_query.stripped_query.query()), kProfileQueryStart),
@@ -1375,6 +1439,16 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
&interpreter_context->ast_cache, interpreter_context->config.query);
auto *cypher_query = utils::Downcast<CypherQuery>(parsed_inner_query.query);
bool contains_csv = false;
auto clauses = cypher_query->single_query_->clauses_;
if (std::any_of(clauses.begin(), clauses.end(),
[](const auto *clause) { return clause->GetTypeInfo() == LoadCsv::kType; })) {
contains_csv = true;
}
// If this is LOAD CSV query, use PoolResource without MonotonicMemoryResource as we want to reuse allocated memory
auto use_monotonic_memory = !contains_csv;
MG_ASSERT(cypher_query, "Cypher grammar should not allow other queries in PROFILE");
Frame frame(0);
SymbolTable symbol_table;
@@ -1387,39 +1461,42 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
auto cypher_query_plan = CypherQueryToPlan(
parsed_inner_query.stripped_query.hash(), std::move(parsed_inner_query.ast_storage), cypher_query,
parsed_inner_query.parameters, parsed_inner_query.is_cacheable ? &interpreter_context->plan_cache : nullptr, dba);
TryCaching(cypher_query_plan->ast_storage(), frame_change_collector);
auto rw_type_checker = plan::ReadWriteTypeChecker();
auto optional_username = StringPointerToOptional(username);
rw_type_checker.InferRWType(const_cast<plan::LogicalOperator &>(cypher_query_plan->plan()));
return PreparedQuery{{"OPERATOR", "ACTUAL HITS", "RELATIVE TIME", "ABSOLUTE TIME"},
std::move(parsed_query.required_privileges),
[plan = std::move(cypher_query_plan), parameters = std::move(parsed_inner_query.parameters),
summary, dba, interpreter_context, execution_memory, memory_limit, optional_username,
// We want to execute the query we are profiling lazily, so we delay
// the construction of the corresponding context.
stats_and_total_time = std::optional<plan::ProfilingStatsWithTotalTime>{},
pull_plan = std::shared_ptr<PullPlanVector>(nullptr), transaction_status](
AnyStream *stream, std::optional<int> n) mutable -> std::optional<QueryHandlerResult> {
// No output symbols are given so that nothing is streamed.
if (!stats_and_total_time) {
stats_and_total_time =
PullPlan(plan, parameters, true, dba, interpreter_context, execution_memory,
optional_username, transaction_status, nullptr, memory_limit)
.Pull(stream, {}, {}, summary);
pull_plan = std::make_shared<PullPlanVector>(ProfilingStatsToTable(*stats_and_total_time));
}
return PreparedQuery{
{"OPERATOR", "ACTUAL HITS", "RELATIVE TIME", "ABSOLUTE TIME"},
std::move(parsed_query.required_privileges),
[plan = std::move(cypher_query_plan), parameters = std::move(parsed_inner_query.parameters), summary, dba,
interpreter_context, execution_memory, memory_limit, optional_username,
// We want to execute the query we are profiling lazily, so we delay
// the construction of the corresponding context.
stats_and_total_time = std::optional<plan::ProfilingStatsWithTotalTime>{},
pull_plan = std::shared_ptr<PullPlanVector>(nullptr), transaction_status, use_monotonic_memory,
frame_change_collector](AnyStream *stream, std::optional<int> n) mutable -> std::optional<QueryHandlerResult> {
// No output symbols are given so that nothing is streamed.
if (!stats_and_total_time) {
stats_and_total_time =
PullPlan(plan, parameters, true, dba, interpreter_context, execution_memory, optional_username,
transaction_status, nullptr, memory_limit, use_monotonic_memory,
frame_change_collector->IsTrackingValues() ? frame_change_collector : nullptr)
.Pull(stream, {}, {}, summary);
pull_plan = std::make_shared<PullPlanVector>(ProfilingStatsToTable(*stats_and_total_time));
}
MG_ASSERT(stats_and_total_time, "Failed to execute the query!");
MG_ASSERT(stats_and_total_time, "Failed to execute the query!");
if (pull_plan->Pull(stream, n)) {
summary->insert_or_assign("profile", ProfilingStatsToJson(*stats_and_total_time).dump());
return QueryHandlerResult::ABORT;
}
if (pull_plan->Pull(stream, n)) {
summary->insert_or_assign("profile", ProfilingStatsToJson(*stats_and_total_time).dump());
return QueryHandlerResult::ABORT;
}
return std::nullopt;
},
rw_type_checker.type};
return std::nullopt;
},
rw_type_checker.type};
}
PreparedQuery PrepareDumpQuery(ParsedQuery parsed_query, std::map<std::string, TypedValue> *summary, DbAccessor *dba,
@@ -1623,7 +1700,6 @@ PreparedQuery PrepareIndexQuery(ParsedQuery parsed_query, bool in_explicit_trans
[&index_notification, &label_name, &properties_stringified]<typename T>(T &&) {
using ErrorType = std::remove_cvref_t<T>;
if constexpr (std::is_same_v<ErrorType, storage::ReplicationError>) {
EventCounter::IncrementCounter(EventCounter::LabelIndexCreated);
throw ReplicationException(
fmt::format("At least one SYNC replica has not confirmed the creation of the index on label {} "
"on properties {}.",
@@ -1637,8 +1713,6 @@ PreparedQuery PrepareIndexQuery(ParsedQuery parsed_query, bool in_explicit_trans
}
},
error);
} else {
EventCounter::IncrementCounter(EventCounter::LabelIndexCreated);
}
};
break;
@@ -1765,25 +1839,49 @@ PreparedQuery PrepareLockPathQuery(ParsedQuery parsed_query, bool in_explicit_tr
auto *lock_path_query = utils::Downcast<LockPathQuery>(parsed_query.query);
return PreparedQuery{{},
std::move(parsed_query.required_privileges),
[interpreter_context, action = lock_path_query->action_](
AnyStream *stream, std::optional<int> n) -> std::optional<QueryHandlerResult> {
switch (action) {
case LockPathQuery::Action::LOCK_PATH:
if (!interpreter_context->db->LockPath()) {
throw QueryRuntimeException("Failed to lock the data directory");
}
break;
case LockPathQuery::Action::UNLOCK_PATH:
if (!interpreter_context->db->UnlockPath()) {
throw QueryRuntimeException("Failed to unlock the data directory");
}
break;
}
return QueryHandlerResult::COMMIT;
},
RWType::NONE};
return PreparedQuery{
{"STATUS"},
std::move(parsed_query.required_privileges),
[interpreter_context, action = lock_path_query->action_](
AnyStream *stream, std::optional<int> n) -> std::optional<QueryHandlerResult> {
std::vector<std::vector<TypedValue>> status;
std::string res;
switch (action) {
case LockPathQuery::Action::LOCK_PATH: {
const auto lock_success = interpreter_context->db->LockPath();
if (lock_success.HasError()) [[unlikely]] {
throw QueryRuntimeException("Failed to lock the data directory");
}
res = lock_success.GetValue() ? "Data directory is now locked." : "Data directory is already locked.";
break;
}
case LockPathQuery::Action::UNLOCK_PATH: {
const auto unlock_success = interpreter_context->db->UnlockPath();
if (unlock_success.HasError()) [[unlikely]] {
throw QueryRuntimeException("Failed to unlock the data directory");
}
res = unlock_success.GetValue() ? "Data directory is now unlocked." : "Data directory is already unlocked.";
break;
}
case LockPathQuery::Action::STATUS: {
const auto locked_status = interpreter_context->db->IsPathLocked();
if (locked_status.HasError()) [[unlikely]] {
throw QueryRuntimeException("Failed to access the data directory");
}
res = locked_status.GetValue() ? "Data directory is locked." : "Data directory is unlocked.";
break;
}
}
status.emplace_back(std::vector<TypedValue>{TypedValue(res)});
auto pull_plan = std::make_shared<PullPlanVector>(std::move(status));
if (pull_plan->Pull(stream, n)) {
return QueryHandlerResult::COMMIT;
}
return std::nullopt;
},
RWType::NONE};
}
PreparedQuery PrepareFreeMemoryQuery(ParsedQuery parsed_query, bool in_explicit_transaction,
@@ -1871,6 +1969,7 @@ Callback CreateTrigger(TriggerQuery *trigger_query,
std::move(trigger_name), trigger_statement, user_parameters, ToTriggerEventType(event_type),
before_commit ? TriggerPhase::BEFORE_COMMIT : TriggerPhase::AFTER_COMMIT, &interpreter_context->ast_cache,
dba, interpreter_context->config.query, std::move(owner), interpreter_context->auth_checker);
memgraph::metrics::IncrementCounter(memgraph::metrics::TriggersCreated);
return {};
}};
}
@@ -1925,7 +2024,6 @@ PreparedQuery PrepareTriggerQuery(ParsedQuery parsed_query, bool in_explicit_tra
case TriggerQuery::Action::CREATE_TRIGGER:
trigger_notification.emplace(SeverityLevel::INFO, NotificationCode::CREATE_TRIGGER,
fmt::format("Created trigger {}.", trigger_query->trigger_name_));
EventCounter::IncrementCounter(EventCounter::TriggersCreated);
return CreateTrigger(trigger_query, user_parameters, interpreter_context, dba, std::move(owner));
case TriggerQuery::Action::DROP_TRIGGER:
trigger_notification.emplace(SeverityLevel::INFO, NotificationCode::DROP_TRIGGER,
@@ -2159,6 +2257,14 @@ std::vector<std::vector<TypedValue>> TransactionQueueQueryHandler::ShowTransacti
const auto &typed_queries = interpreter->GetQueries();
results.push_back({TypedValue(interpreter->username_.value_or("")),
TypedValue(std::to_string(transaction_id.value())), TypedValue(typed_queries)});
// Handle user-defined metadata
std::map<std::string, TypedValue> metadata_tv;
if (interpreter->metadata_) {
for (const auto &md : *(interpreter->metadata_)) {
metadata_tv.emplace(md.first, TypedValue(md.second));
}
}
results.back().push_back(TypedValue(metadata_tv));
}
}
return results;
@@ -2228,7 +2334,7 @@ Callback HandleTransactionQueueQuery(TransactionQueueQuery *transaction_query,
Callback callback;
switch (transaction_query->action_) {
case TransactionQueueQuery::Action::SHOW_TRANSACTIONS: {
callback.header = {"username", "transaction_id", "query"};
callback.header = {"username", "transaction_id", "query", "metadata"};
callback.fn = [handler = TransactionQueueQueryHandler(), interpreter_context, username,
hasTransactionManagementPrivilege]() mutable {
std::vector<std::vector<TypedValue>> results;
@@ -2306,8 +2412,10 @@ PreparedQuery PrepareVersionQuery(ParsedQuery parsed_query, bool in_explicit_tra
}
PreparedQuery PrepareInfoQuery(ParsedQuery parsed_query, bool in_explicit_transaction,
std::map<std::string, TypedValue> *summary, InterpreterContext *interpreter_context,
storage::Storage *db, utils::MemoryResource *execution_memory) {
std::map<std::string, TypedValue> * /*summary*/, InterpreterContext *interpreter_context,
storage::Storage *db, utils::MemoryResource * /*execution_memory*/,
std::optional<storage::IsolationLevel> interpreter_isolation_level,
std::optional<storage::IsolationLevel> next_transaction_isolation_level) {
if (in_explicit_transaction) {
throw InfoInMulticommandTxException();
}
@@ -2319,7 +2427,8 @@ PreparedQuery PrepareInfoQuery(ParsedQuery parsed_query, bool in_explicit_transa
switch (info_query->info_type_) {
case InfoQuery::InfoType::STORAGE:
header = {"storage info", "value"};
handler = [db] {
handler = [db, interpreter_isolation_level, next_transaction_isolation_level] {
auto info = db->GetInfo();
std::vector<std::vector<TypedValue>> results{
{TypedValue("vertex_count"), TypedValue(static_cast<int64_t>(info.vertex_count))},
@@ -2328,8 +2437,12 @@ PreparedQuery PrepareInfoQuery(ParsedQuery parsed_query, bool in_explicit_transa
{TypedValue("memory_usage"), TypedValue(static_cast<int64_t>(info.memory_usage))},
{TypedValue("disk_usage"), TypedValue(static_cast<int64_t>(info.disk_usage))},
{TypedValue("memory_allocated"), TypedValue(static_cast<int64_t>(utils::total_memory_tracker.Amount()))},
{TypedValue("allocation_limit"),
TypedValue(static_cast<int64_t>(utils::total_memory_tracker.HardLimit()))}};
{TypedValue("allocation_limit"), TypedValue(static_cast<int64_t>(utils::total_memory_tracker.HardLimit()))},
{TypedValue("global_isolation_level"), TypedValue(IsolationLevelToString(db->GetIsolationLevel()))},
{TypedValue("session_isolation_level"), TypedValue(IsolationLevelToString(interpreter_isolation_level))},
{TypedValue("next_session_isolation_level"),
TypedValue(IsolationLevelToString(next_transaction_isolation_level))},
{TypedValue("storage_mode"), TypedValue(StorageModeToString(db->GetStorageMode()))}};
return std::pair{results, QueryHandlerResult::COMMIT};
};
break;
@@ -2373,6 +2486,15 @@ PreparedQuery PrepareInfoQuery(ParsedQuery parsed_query, bool in_explicit_transa
return std::pair{results, QueryHandlerResult::NOTHING};
};
break;
case InfoQuery::InfoType::BUILD:
header = {"build info", "value"};
handler = [] {
std::vector<std::vector<TypedValue>> results{
{TypedValue("build_type"), TypedValue(utils::GetBuildInfo().build_name)}};
return std::pair{results, QueryHandlerResult::NOTHING};
};
break;
}
return PreparedQuery{std::move(header), std::move(parsed_query.required_privileges),
@@ -2648,8 +2770,8 @@ std::optional<uint64_t> Interpreter::GetTransactionId() const {
return {};
}
void Interpreter::BeginTransaction() {
const auto prepared_query = PrepareTransactionQuery("BEGIN");
void Interpreter::BeginTransaction(const std::map<std::string, storage::PropertyValue> &metadata) {
const auto prepared_query = PrepareTransactionQuery("BEGIN", metadata);
prepared_query.query_handler(nullptr, {});
}
@@ -2669,10 +2791,13 @@ void Interpreter::RollbackTransaction() {
Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
const std::map<std::string, storage::PropertyValue> &params,
const std::string *username) {
const std::string *username,
const std::map<std::string, storage::PropertyValue> &metadata) {
if (!in_explicit_transaction_) {
query_executions_.clear();
transaction_queries_->clear();
// Handle user-defined metadata in auto-transactions
metadata_ = GenOptional(metadata);
}
// This will be done in the handle transaction query. Our handler can save username and then send it to the kill and
@@ -2692,7 +2817,7 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
std::optional<int> qid =
in_explicit_transaction_ ? static_cast<int>(query_executions_.size() - 1) : std::optional<int>{};
query_execution->prepared_query.emplace(PrepareTransactionQuery(trimmed_query));
query_execution->prepared_query.emplace(PrepareTransactionQuery(trimmed_query, metadata));
return {query_execution->prepared_query->header, query_execution->prepared_query->privileges, qid};
}
@@ -2703,14 +2828,14 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
// an explicit transaction block.
if (in_explicit_transaction_) {
AdvanceCommand();
}
// If we're not in an explicit transaction block and we have an open
// transaction, abort it since we're about to prepare a new query.
else if (db_accessor_) {
} else if (db_accessor_) {
// If we're not in an explicit transaction block and we have an open
// transaction, abort it since we're about to prepare a new query.
query_executions_.emplace_back(
std::make_unique<QueryExecution>(utils::MonotonicBufferResource(kExecutionMemoryBlockSize)));
AbortCommand(&query_executions_.back());
}
std::unique_ptr<QueryExecution> *query_execution_ptr = nullptr;
try {
query_executions_.emplace_back(
@@ -2721,15 +2846,22 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
ParseQuery(query_string, params, &interpreter_context_->ast_cache, interpreter_context_->config.query);
TypedValue parsing_time{parsing_timer.Elapsed().count()};
if (utils::Downcast<CypherQuery>(parsed_query.query)) {
auto *cypher_query = utils::Downcast<CypherQuery>(parsed_query.query);
if ((utils::Downcast<CypherQuery>(parsed_query.query) || utils::Downcast<ProfileQuery>(parsed_query.query))) {
CypherQuery *cypher_query = nullptr;
if (utils::Downcast<CypherQuery>(parsed_query.query)) {
cypher_query = utils::Downcast<CypherQuery>(parsed_query.query);
} else {
auto *profile_query = utils::Downcast<ProfileQuery>(parsed_query.query);
cypher_query = profile_query->cypher_query_;
}
if (const auto &clauses = cypher_query->single_query_->clauses_;
std::any_of(clauses.begin(), clauses.end(),
[](const auto *clause) { return clause->GetTypeInfo() == LoadCsv::kType; })) {
// Using PoolResource without MonotonicMemoryResouce for LOAD CSV reduces memory usage.
// QueryExecution MemoryResource is mostly used for allocations done on Frame and storing `row`s
query_executions_[query_executions_.size() - 1] = std::make_unique<QueryExecution>(
utils::PoolResource(1, kExecutionPoolMaxBlockSize, utils::NewDeleteResource(), utils::NewDeleteResource()));
utils::PoolResource(8, kExecutionPoolMaxBlockSize, utils::NewDeleteResource(), utils::NewDeleteResource()));
query_execution_ptr = &query_executions_.back();
}
}
@@ -2751,6 +2883,7 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
utils::Downcast<ProfileQuery>(parsed_query.query) || utils::Downcast<DumpQuery>(parsed_query.query) ||
utils::Downcast<TriggerQuery>(parsed_query.query) || utils::Downcast<AnalyzeGraphQuery>(parsed_query.query) ||
utils::Downcast<TransactionQueueQuery>(parsed_query.query))) {
memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveTransactions);
db_accessor_ =
std::make_unique<storage::Storage::Accessor>(interpreter_context_->db->Access(GetIsolationLevelOverride()));
execution_db_accessor_.emplace(db_accessor_.get());
@@ -2766,18 +2899,21 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
utils::MemoryResource *memory_resource =
std::visit([](auto &execution_memory) -> utils::MemoryResource * { return &execution_memory; },
query_execution->execution_memory);
frame_change_collector_.reset();
frame_change_collector_.emplace(memory_resource);
if (utils::Downcast<CypherQuery>(parsed_query.query)) {
prepared_query =
PrepareCypherQuery(std::move(parsed_query), &query_execution->summary, interpreter_context_,
&*execution_db_accessor_, memory_resource, &query_execution->notifications, username,
&transaction_status_, trigger_context_collector_ ? &*trigger_context_collector_ : nullptr);
prepared_query = PrepareCypherQuery(
std::move(parsed_query), &query_execution->summary, interpreter_context_, &*execution_db_accessor_,
memory_resource, &query_execution->notifications, username, &transaction_status_,
trigger_context_collector_ ? &*trigger_context_collector_ : nullptr, &*frame_change_collector_);
} 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_with_exception);
} else if (utils::Downcast<ProfileQuery>(parsed_query.query)) {
prepared_query = PrepareProfileQuery(
std::move(parsed_query), in_explicit_transaction_, &query_execution->summary, interpreter_context_,
&*execution_db_accessor_, &query_execution->execution_memory_with_exception, username, &transaction_status_);
prepared_query = PrepareProfileQuery(std::move(parsed_query), in_explicit_transaction_, &query_execution->summary,
interpreter_context_, &*execution_db_accessor_,
&query_execution->execution_memory_with_exception, username,
&transaction_status_, &*frame_change_collector_);
} else if (utils::Downcast<DumpQuery>(parsed_query.query)) {
prepared_query = PrepareDumpQuery(std::move(parsed_query), &query_execution->summary, &*execution_db_accessor_,
memory_resource);
@@ -2794,7 +2930,8 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
} else if (utils::Downcast<InfoQuery>(parsed_query.query)) {
prepared_query = PrepareInfoQuery(std::move(parsed_query), in_explicit_transaction_, &query_execution->summary,
interpreter_context_, interpreter_context_->db,
&query_execution->execution_memory_with_exception);
&query_execution->execution_memory_with_exception, interpreter_isolation_level,
next_transaction_isolation_level);
} else if (utils::Downcast<ConstraintQuery>(parsed_query.query)) {
prepared_query = PrepareConstraintQuery(std::move(parsed_query), in_explicit_transaction_,
&query_execution->notifications, interpreter_context_);
@@ -2853,7 +2990,7 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
return {query_execution->prepared_query->header, query_execution->prepared_query->privileges, qid};
} catch (const utils::BasicException &) {
EventCounter::IncrementCounter(EventCounter::FailedQuery);
memgraph::metrics::IncrementCounter(memgraph::metrics::FailedQuery);
AbortCommand(query_execution_ptr);
throw;
}
@@ -2884,11 +3021,17 @@ void Interpreter::Abort() {
expect_rollback_ = false;
in_explicit_transaction_ = false;
metadata_ = std::nullopt;
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveTransactions);
if (!db_accessor_) return;
db_accessor_->Abort();
execution_db_accessor_.reset();
db_accessor_.reset();
trigger_context_collector_.reset();
frame_change_collector_.reset();
}
namespace {
@@ -2980,12 +3123,21 @@ void Interpreter::Commit() {
utils::OnScopeExit clean_status(
[this]() { transaction_status_.store(TransactionStatus::IDLE, std::memory_order_release); });
utils::OnScopeExit update_metrics([]() {
memgraph::metrics::IncrementCounter(memgraph::metrics::CommitedTransactions);
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveTransactions);
});
std::optional<TriggerContext> trigger_context = std::nullopt;
if (trigger_context_collector_) {
trigger_context.emplace(std::move(*trigger_context_collector_).TransformToTriggerContext());
trigger_context_collector_.reset();
}
if (frame_change_collector_) {
frame_change_collector_.reset();
}
if (trigger_context) {
// Run the triggers
for (const auto &trigger : interpreter_context_->trigger_store.BeforeCommitTriggers().access()) {

View File

@@ -44,14 +44,14 @@
#include "utils/timer.hpp"
#include "utils/tsc.hpp"
namespace EventCounter {
namespace memgraph::metrics {
extern const Event FailedQuery;
} // namespace EventCounter
} // namespace memgraph::metrics
namespace memgraph::query {
inline constexpr size_t kExecutionMemoryBlockSize = 1UL * 1024UL * 1024UL;
inline constexpr size_t kExecutionPoolMaxBlockSize = 32768UL; // 2 ^ 15
inline constexpr size_t kExecutionPoolMaxBlockSize = 1024UL; // 2 ^ 10
class AuthQueryHandler {
public:
@@ -261,6 +261,7 @@ class Interpreter final {
std::optional<std::string> username_;
bool in_explicit_transaction_{false};
bool expect_rollback_{false};
std::optional<std::map<std::string, storage::PropertyValue>> metadata_{}; //!< User defined transaction metadata
/**
* Prepare a query for execution.
@@ -271,7 +272,8 @@ class Interpreter final {
* @throw query::QueryException
*/
PrepareResult Prepare(const std::string &query, const std::map<std::string, storage::PropertyValue> &params,
const std::string *username);
const std::string *username,
const std::map<std::string, storage::PropertyValue> &metadata = {});
/**
* Execute the last prepared query and stream *all* of the results into the
@@ -315,7 +317,7 @@ class Interpreter final {
std::map<std::string, TypedValue> Pull(TStream *result_stream, std::optional<int> n = {},
std::optional<int> qid = {});
void BeginTransaction();
void BeginTransaction(const std::map<std::string, storage::PropertyValue> &metadata = {});
/*
Returns transaction id or empty if the db_accessor is not initialized.
@@ -401,11 +403,13 @@ class Interpreter final {
std::unique_ptr<storage::Storage::Accessor> db_accessor_;
std::optional<DbAccessor> execution_db_accessor_;
std::optional<TriggerContextCollector> trigger_context_collector_;
std::optional<FrameChangeCollector> frame_change_collector_;
std::optional<storage::IsolationLevel> interpreter_isolation_level;
std::optional<storage::IsolationLevel> next_transaction_isolation_level;
PreparedQuery PrepareTransactionQuery(std::string_view query_upper);
PreparedQuery PrepareTransactionQuery(std::string_view query_upper,
const std::map<std::string, storage::PropertyValue> &metadata = {});
void Commit();
void AdvanceCommand();
void AbortCommand(std::unique_ptr<QueryExecution> *query_execution);
@@ -515,7 +519,7 @@ std::map<std::string, TypedValue> Interpreter::Pull(TStream *result_stream, std:
query_execution.reset(nullptr);
throw;
} catch (const utils::BasicException &) {
EventCounter::IncrementCounter(EventCounter::FailedQuery);
memgraph::metrics::IncrementCounter(memgraph::metrics::FailedQuery);
AbortCommand(&query_execution);
throw;
}

View File

@@ -53,6 +53,7 @@
#include "utils/likely.hpp"
#include "utils/logging.hpp"
#include "utils/memory.hpp"
#include "utils/pmr/deque.hpp"
#include "utils/pmr/list.hpp"
#include "utils/pmr/unordered_map.hpp"
#include "utils/pmr/unordered_set.hpp"
@@ -60,6 +61,7 @@
#include "utils/readable_size.hpp"
#include "utils/string.hpp"
#include "utils/temporal.hpp"
#include "utils/typeinfo.hpp"
// macro for the default implementation of LogicalOperator::Accept
// that accepts the visitor and visits it's input_ operator
@@ -80,7 +82,7 @@
LOG_FATAL("Operator " #class_name " has no single input!"); \
}
namespace EventCounter {
namespace memgraph::metrics {
extern const Event OnceOperator;
extern const Event CreateNodeOperator;
extern const Event CreateExpandOperator;
@@ -118,7 +120,7 @@ extern const Event ForeachOperator;
extern const Event EmptyResultOperator;
extern const Event EvaluatePatternFilterOperator;
extern const Event ApplyOperator;
} // namespace EventCounter
} // namespace memgraph::metrics
namespace memgraph::query::plan {
@@ -169,7 +171,7 @@ bool Once::OnceCursor::Pull(Frame &, ExecutionContext &context) {
}
UniqueCursorPtr Once::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::OnceOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::OnceOperator);
return MakeUniqueCursorPtr<OnceCursor>(mem);
}
@@ -231,7 +233,7 @@ VertexAccessor &CreateLocalVertex(const NodeCreationInfo &node_info, Frame *fram
ACCEPT_WITH_INPUT(CreateNode)
UniqueCursorPtr CreateNode::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::CreateNodeOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::CreateNodeOperator);
return MakeUniqueCursorPtr<CreateNodeCursor>(mem, *this, mem);
}
@@ -281,7 +283,7 @@ CreateExpand::CreateExpand(const NodeCreationInfo &node_info, const EdgeCreation
ACCEPT_WITH_INPUT(CreateExpand)
UniqueCursorPtr CreateExpand::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::CreateNodeOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::CreateNodeOperator);
return MakeUniqueCursorPtr<CreateExpandCursor>(mem, *this, mem);
}
@@ -488,7 +490,7 @@ ScanAll::ScanAll(const std::shared_ptr<LogicalOperator> &input, Symbol output_sy
ACCEPT_WITH_INPUT(ScanAll)
UniqueCursorPtr ScanAll::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ScanAllOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ScanAllOperator);
auto vertices = [this](Frame &, ExecutionContext &context) {
auto *db = context.db_accessor;
@@ -511,7 +513,7 @@ ScanAllByLabel::ScanAllByLabel(const std::shared_ptr<LogicalOperator> &input, Sy
ACCEPT_WITH_INPUT(ScanAllByLabel)
UniqueCursorPtr ScanAllByLabel::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ScanAllByLabelOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ScanAllByLabelOperator);
auto vertices = [this](Frame &, ExecutionContext &context) {
auto *db = context.db_accessor;
@@ -541,7 +543,7 @@ ScanAllByLabelPropertyRange::ScanAllByLabelPropertyRange(const std::shared_ptr<L
ACCEPT_WITH_INPUT(ScanAllByLabelPropertyRange)
UniqueCursorPtr ScanAllByLabelPropertyRange::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ScanAllByLabelPropertyRangeOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ScanAllByLabelPropertyRangeOperator);
auto vertices = [this](Frame &frame, ExecutionContext &context)
-> std::optional<decltype(context.db_accessor->Vertices(view_, label_, property_, std::nullopt, std::nullopt))> {
@@ -601,7 +603,7 @@ ScanAllByLabelPropertyValue::ScanAllByLabelPropertyValue(const std::shared_ptr<L
ACCEPT_WITH_INPUT(ScanAllByLabelPropertyValue)
UniqueCursorPtr ScanAllByLabelPropertyValue::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ScanAllByLabelPropertyValueOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ScanAllByLabelPropertyValueOperator);
auto vertices = [this](Frame &frame, ExecutionContext &context)
-> std::optional<decltype(context.db_accessor->Vertices(view_, label_, property_, storage::PropertyValue()))> {
@@ -626,7 +628,7 @@ ScanAllByLabelProperty::ScanAllByLabelProperty(const std::shared_ptr<LogicalOper
ACCEPT_WITH_INPUT(ScanAllByLabelProperty)
UniqueCursorPtr ScanAllByLabelProperty::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ScanAllByLabelPropertyOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ScanAllByLabelPropertyOperator);
auto vertices = [this](Frame &frame, ExecutionContext &context) {
auto *db = context.db_accessor;
@@ -645,7 +647,7 @@ ScanAllById::ScanAllById(const std::shared_ptr<LogicalOperator> &input, Symbol o
ACCEPT_WITH_INPUT(ScanAllById)
UniqueCursorPtr ScanAllById::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ScanAllByIdOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ScanAllByIdOperator);
auto vertices = [this](Frame &frame, ExecutionContext &context) -> std::optional<std::vector<VertexAccessor>> {
auto *db = context.db_accessor;
@@ -700,7 +702,7 @@ Expand::Expand(const std::shared_ptr<LogicalOperator> &input, Symbol input_symbo
ACCEPT_WITH_INPUT(Expand)
UniqueCursorPtr Expand::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ExpandOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ExpandOperator);
return MakeUniqueCursorPtr<ExpandCursor>(mem, *this, mem);
}
@@ -2170,7 +2172,7 @@ class ExpandAllShortestPathsCursor : public query::plan::Cursor {
};
UniqueCursorPtr ExpandVariable::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ExpandVariableOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ExpandVariableOperator);
switch (type_) {
case EdgeAtom::Type::BREADTH_FIRST:
@@ -2273,7 +2275,7 @@ class ConstructNamedPathCursor : public Cursor {
ACCEPT_WITH_INPUT(ConstructNamedPath)
UniqueCursorPtr ConstructNamedPath::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ConstructNamedPathOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ConstructNamedPathOperator);
return MakeUniqueCursorPtr<ConstructNamedPathCursor>(mem, *this, mem);
}
@@ -2299,7 +2301,7 @@ bool Filter::Accept(HierarchicalLogicalOperatorVisitor &visitor) {
}
UniqueCursorPtr Filter::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::FilterOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::FilterOperator);
return MakeUniqueCursorPtr<FilterCursor>(mem, *this, mem);
}
@@ -2331,12 +2333,11 @@ bool Filter::FilterCursor::Pull(Frame &frame, ExecutionContext &context) {
// Like all filters, newly set values should not affect filtering of old
// nodes and edges.
ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, context.db_accessor,
storage::View::OLD);
storage::View::OLD, context.frame_change_collector);
while (input_cursor_->Pull(frame, context)) {
for (const auto &pattern_filter_cursor : pattern_filter_cursors_) {
pattern_filter_cursor->Pull(frame, context);
}
if (EvaluateFilter(evaluator, self_.expression_)) return true;
}
return false;
@@ -2352,7 +2353,7 @@ EvaluatePatternFilter::EvaluatePatternFilter(const std::shared_ptr<LogicalOperat
ACCEPT_WITH_INPUT(EvaluatePatternFilter);
UniqueCursorPtr EvaluatePatternFilter::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::EvaluatePatternFilterOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::EvaluatePatternFilterOperator);
return MakeUniqueCursorPtr<EvaluatePatternFilterCursor>(mem, *this, mem);
}
@@ -2385,7 +2386,7 @@ Produce::Produce(const std::shared_ptr<LogicalOperator> &input, const std::vecto
ACCEPT_WITH_INPUT(Produce)
UniqueCursorPtr Produce::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ProduceOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ProduceOperator);
return MakeUniqueCursorPtr<ProduceCursor>(mem, *this, mem);
}
@@ -2409,9 +2410,13 @@ bool Produce::ProduceCursor::Pull(Frame &frame, ExecutionContext &context) {
if (input_cursor_->Pull(frame, context)) {
// Produce should always yield the latest results.
ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, context.db_accessor,
storage::View::NEW);
for (auto named_expr : self_.named_expressions_) named_expr->Accept(evaluator);
storage::View::NEW, context.frame_change_collector);
for (auto *named_expr : self_.named_expressions_) {
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(named_expr->name_)) {
context.frame_change_collector->ResetTrackingValue(named_expr->name_);
}
named_expr->Accept(evaluator);
}
return true;
}
return false;
@@ -2428,7 +2433,7 @@ Delete::Delete(const std::shared_ptr<LogicalOperator> &input_, const std::vector
ACCEPT_WITH_INPUT(Delete)
UniqueCursorPtr Delete::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::DeleteOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::DeleteOperator);
return MakeUniqueCursorPtr<DeleteCursor>(mem, *this, mem);
}
@@ -2580,7 +2585,7 @@ SetProperty::SetProperty(const std::shared_ptr<LogicalOperator> &input, storage:
ACCEPT_WITH_INPUT(SetProperty)
UniqueCursorPtr SetProperty::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::SetPropertyOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::SetPropertyOperator);
return MakeUniqueCursorPtr<SetPropertyCursor>(mem, *this, mem);
}
@@ -2663,7 +2668,7 @@ SetProperties::SetProperties(const std::shared_ptr<LogicalOperator> &input, Symb
ACCEPT_WITH_INPUT(SetProperties)
UniqueCursorPtr SetProperties::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::SetPropertiesOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::SetPropertiesOperator);
return MakeUniqueCursorPtr<SetPropertiesCursor>(mem, *this, mem);
}
@@ -2860,7 +2865,7 @@ SetLabels::SetLabels(const std::shared_ptr<LogicalOperator> &input, Symbol input
ACCEPT_WITH_INPUT(SetLabels)
UniqueCursorPtr SetLabels::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::SetLabelsOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::SetLabelsOperator);
return MakeUniqueCursorPtr<SetLabelsCursor>(mem, *this, mem);
}
@@ -2932,7 +2937,7 @@ RemoveProperty::RemoveProperty(const std::shared_ptr<LogicalOperator> &input, st
ACCEPT_WITH_INPUT(RemoveProperty)
UniqueCursorPtr RemoveProperty::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::RemovePropertyOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::RemovePropertyOperator);
return MakeUniqueCursorPtr<RemovePropertyCursor>(mem, *this, mem);
}
@@ -3018,7 +3023,7 @@ RemoveLabels::RemoveLabels(const std::shared_ptr<LogicalOperator> &input, Symbol
ACCEPT_WITH_INPUT(RemoveLabels)
UniqueCursorPtr RemoveLabels::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::RemoveLabelsOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::RemoveLabelsOperator);
return MakeUniqueCursorPtr<RemoveLabelsCursor>(mem, *this, mem);
}
@@ -3091,7 +3096,7 @@ EdgeUniquenessFilter::EdgeUniquenessFilter(const std::shared_ptr<LogicalOperator
ACCEPT_WITH_INPUT(EdgeUniquenessFilter)
UniqueCursorPtr EdgeUniquenessFilter::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::EdgeUniquenessFilterOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::EdgeUniquenessFilterOperator);
return MakeUniqueCursorPtr<EdgeUniquenessFilterCursor>(mem, *this, mem);
}
@@ -3193,7 +3198,7 @@ class EmptyResultCursor : public Cursor {
};
UniqueCursorPtr EmptyResult::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::EmptyResultOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::EmptyResultOperator);
return MakeUniqueCursorPtr<EmptyResultCursor>(mem, *this, mem);
}
@@ -3232,7 +3237,12 @@ class AccumulateCursor : public Cursor {
if (MustAbort(context)) throw HintedAbortError();
if (cache_it_ == cache_.end()) return false;
auto row_it = (cache_it_++)->begin();
for (const Symbol &symbol : self_.symbols_) frame[symbol] = *row_it++;
for (const Symbol &symbol : self_.symbols_) {
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(symbol.name())) {
context.frame_change_collector->ResetTrackingValue(symbol.name());
}
frame[symbol] = *row_it++;
}
return true;
}
@@ -3248,13 +3258,13 @@ class AccumulateCursor : public Cursor {
private:
const Accumulate &self_;
const UniqueCursorPtr input_cursor_;
utils::pmr::vector<utils::pmr::vector<TypedValue>> cache_;
utils::pmr::deque<utils::pmr::vector<TypedValue>> cache_;
decltype(cache_.begin()) cache_it_ = cache_.begin();
bool pulled_all_input_{false};
};
UniqueCursorPtr Accumulate::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::AccumulateOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::AccumulateOperator);
return MakeUniqueCursorPtr<AccumulateCursor>(mem, *this, mem);
}
@@ -3314,10 +3324,20 @@ class AggregateCursor : public Cursor {
if (aggregation_.empty()) {
auto *pull_memory = context.evaluation_context.memory;
// place default aggregation values on the frame
for (const auto &elem : self_.aggregations_)
for (const auto &elem : self_.aggregations_) {
frame[elem.output_sym] = DefaultAggregationOpValue(elem, pull_memory);
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(elem.output_sym.name())) {
context.frame_change_collector->ResetTrackingValue(elem.output_sym.name());
}
}
// place null as remember values on the frame
for (const Symbol &remember_sym : self_.remember_) frame[remember_sym] = TypedValue(pull_memory);
for (const Symbol &remember_sym : self_.remember_) {
frame[remember_sym] = TypedValue(pull_memory);
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(remember_sym.name())) {
context.frame_change_collector->ResetTrackingValue(remember_sym.name());
}
}
return true;
}
}
@@ -3616,7 +3636,7 @@ class AggregateCursor : public Cursor {
};
UniqueCursorPtr Aggregate::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::AggregateOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::AggregateOperator);
return MakeUniqueCursorPtr<AggregateCursor>(mem, *this, mem);
}
@@ -3627,7 +3647,7 @@ Skip::Skip(const std::shared_ptr<LogicalOperator> &input, Expression *expression
ACCEPT_WITH_INPUT(Skip)
UniqueCursorPtr Skip::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::SkipOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::SkipOperator);
return MakeUniqueCursorPtr<SkipCursor>(mem, *this, mem);
}
@@ -3680,7 +3700,7 @@ Limit::Limit(const std::shared_ptr<LogicalOperator> &input, Expression *expressi
ACCEPT_WITH_INPUT(Limit)
UniqueCursorPtr Limit::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::LimitOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::LimitOperator);
return MakeUniqueCursorPtr<LimitCursor>(mem, *this, mem);
}
@@ -3797,8 +3817,12 @@ class OrderByCursor : public Cursor {
"Number of values does not match the number of output symbols "
"in OrderBy");
auto output_sym_it = self_.output_symbols_.begin();
for (const TypedValue &output : cache_it_->remember) frame[*output_sym_it++] = output;
for (const TypedValue &output : cache_it_->remember) {
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(output_sym_it->name())) {
context.frame_change_collector->ResetTrackingValue(output_sym_it->name());
}
frame[*output_sym_it++] = output;
}
cache_it_++;
return true;
}
@@ -3828,7 +3852,7 @@ class OrderByCursor : public Cursor {
};
UniqueCursorPtr OrderBy::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::OrderByOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::OrderByOperator);
return MakeUniqueCursorPtr<OrderByCursor>(mem, *this, mem);
}
@@ -3845,7 +3869,7 @@ bool Merge::Accept(HierarchicalLogicalOperatorVisitor &visitor) {
}
UniqueCursorPtr Merge::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::MergeOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::MergeOperator);
return MakeUniqueCursorPtr<MergeCursor>(mem, *this, mem);
}
@@ -3925,7 +3949,7 @@ bool Optional::Accept(HierarchicalLogicalOperatorVisitor &visitor) {
}
UniqueCursorPtr Optional::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::OptionalOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::OptionalOperator);
return MakeUniqueCursorPtr<OptionalCursor>(mem, *this, mem);
}
@@ -4031,6 +4055,9 @@ class UnwindCursor : public Cursor {
if (input_value_it_ == input_value_.end()) continue;
frame[self_.output_symbol_] = *input_value_it_++;
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(self_.output_symbol_.name_)) {
context.frame_change_collector->ResetTrackingValue(self_.output_symbol_.name_);
}
return true;
}
}
@@ -4053,7 +4080,7 @@ class UnwindCursor : public Cursor {
};
UniqueCursorPtr Unwind::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::UnwindOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::UnwindOperator);
return MakeUniqueCursorPtr<UnwindCursor>(mem, *this, mem);
}
@@ -4107,7 +4134,7 @@ Distinct::Distinct(const std::shared_ptr<LogicalOperator> &input, const std::vec
ACCEPT_WITH_INPUT(Distinct)
UniqueCursorPtr Distinct::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::DistinctOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::DistinctOperator);
return MakeUniqueCursorPtr<DistinctCursor>(mem, *this, mem);
}
@@ -4129,7 +4156,7 @@ Union::Union(const std::shared_ptr<LogicalOperator> &left_op, const std::shared_
right_symbols_(right_symbols) {}
UniqueCursorPtr Union::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::UnionOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::UnionOperator);
return MakeUniqueCursorPtr<Union::UnionCursor>(mem, *this, mem);
}
@@ -4160,11 +4187,17 @@ bool Union::UnionCursor::Pull(Frame &frame, ExecutionContext &context) {
// collect values from the left child
for (const auto &output_symbol : self_.left_symbols_) {
results[output_symbol.name()] = frame[output_symbol];
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(output_symbol.name())) {
context.frame_change_collector->ResetTrackingValue(output_symbol.name());
}
}
} else if (right_cursor_->Pull(frame, context)) {
// collect values from the right child
for (const auto &output_symbol : self_.right_symbols_) {
results[output_symbol.name()] = frame[output_symbol];
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(output_symbol.name())) {
context.frame_change_collector->ResetTrackingValue(output_symbol.name());
}
}
} else {
return false;
@@ -4173,6 +4206,9 @@ bool Union::UnionCursor::Pull(Frame &frame, ExecutionContext &context) {
// put collected values on frame under union symbols
for (const auto &symbol : self_.union_symbols_) {
frame[symbol] = results[symbol.name()];
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(symbol.name())) {
context.frame_change_collector->ResetTrackingValue(symbol.name());
}
}
return true;
}
@@ -4237,9 +4273,12 @@ class CartesianCursor : public Cursor {
return false;
}
auto restore_frame = [&frame](const auto &symbols, const auto &restore_from) {
auto restore_frame = [&frame, &context](const auto &symbols, const auto &restore_from) {
for (const auto &symbol : symbols) {
frame[symbol] = restore_from[symbol.position()];
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(symbol.name())) {
context.frame_change_collector->ResetTrackingValue(symbol.name());
}
}
};
@@ -4288,7 +4327,7 @@ class CartesianCursor : public Cursor {
} // namespace
UniqueCursorPtr Cartesian::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::CartesianOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::CartesianOperator);
return MakeUniqueCursorPtr<CartesianCursor>(mem, *this, mem);
}
@@ -4317,6 +4356,10 @@ class OutputTableCursor : public Cursor {
if (current_row_ < rows_.size()) {
for (size_t i = 0; i < self_.output_symbols_.size(); ++i) {
frame[self_.output_symbols_[i]] = rows_[current_row_][i];
if (context.frame_change_collector &&
context.frame_change_collector->IsKeyTracked(self_.output_symbols_[i].name())) {
context.frame_change_collector->ResetTrackingValue(self_.output_symbols_[i].name());
}
}
current_row_++;
return true;
@@ -4360,6 +4403,10 @@ class OutputTableStreamCursor : public Cursor {
MG_ASSERT(row->size() == self_->output_symbols_.size(), "Wrong number of columns in row!");
for (size_t i = 0; i < self_->output_symbols_.size(); ++i) {
frame[self_->output_symbols_[i]] = row->at(i);
if (context.frame_change_collector &&
context.frame_change_collector->IsKeyTracked(self_->output_symbols_[i].name())) {
context.frame_change_collector->ResetTrackingValue(self_->output_symbols_[i].name());
}
}
return true;
}
@@ -4544,7 +4591,7 @@ class CallProcedureCursor : public Cursor {
result_row_it_ = result_.rows.begin();
}
const auto &values = result_row_it_->values;
auto &values = result_row_it_->values;
// Check that the row has all fields as required by the result signature.
// C API guarantees that it's impossible to set fields which are not part of
// the result record, but it does not gurantee that some may be missing. See
@@ -4562,7 +4609,11 @@ class CallProcedureCursor : public Cursor {
throw QueryRuntimeException("Procedure '{}' did not yield a record with '{}' field.", self_->procedure_name_,
field_name);
}
frame[self_->result_symbols_[i]] = result_it->second;
frame[self_->result_symbols_[i]] = std::move(result_it->second);
if (context.frame_change_collector &&
context.frame_change_collector->IsKeyTracked(self_->result_symbols_[i].name())) {
context.frame_change_collector->ResetTrackingValue(self_->result_symbols_[i].name());
}
}
++result_row_it_;
@@ -4579,7 +4630,7 @@ class CallProcedureCursor : public Cursor {
};
UniqueCursorPtr CallProcedure::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::CallProcedureOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::CallProcedureOperator);
CallProcedure::IncrementCounter(procedure_name_);
return MakeUniqueCursorPtr<CallProcedureCursor>(mem, this, mem);
@@ -4689,6 +4740,9 @@ class LoadCsvCursor : public Cursor {
frame[self_->row_var_] =
CsvRowToTypedMap(*row, csv::Reader::Header(reader_->GetHeader(), context.evaluation_context.memory));
}
if (context.frame_change_collector && context.frame_change_collector->IsKeyTracked(self_->row_var_.name())) {
context.frame_change_collector->ResetTrackingValue(self_->row_var_.name());
}
return true;
}
@@ -4785,7 +4839,7 @@ Foreach::Foreach(std::shared_ptr<LogicalOperator> input, std::shared_ptr<Logical
loop_variable_symbol_(loop_variable_symbol) {}
UniqueCursorPtr Foreach::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ForeachOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ForeachOperator);
return MakeUniqueCursorPtr<ForeachCursor>(mem, *this, mem);
}
@@ -4817,7 +4871,7 @@ bool Apply::Accept(HierarchicalLogicalOperatorVisitor &visitor) {
}
UniqueCursorPtr Apply::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ApplyOperator);
memgraph::metrics::IncrementCounter(memgraph::metrics::ApplyOperator);
return MakeUniqueCursorPtr<ApplyCursor>(mem, *this, mem);
}

View File

@@ -180,6 +180,7 @@ class PatternFilterVisitor : public ExpressionVisitor<void> {
void Visit(IfOperator &op) override{};
void Visit(ListLiteral &op) override{};
void Visit(MapLiteral &op) override{};
void Visit(MapProjectionLiteral &op) override{};
void Visit(LabelsTest &op) override{};
void Visit(Aggregation &op) override{};
void Visit(Function &op) override{};
@@ -194,6 +195,7 @@ class PatternFilterVisitor : public ExpressionVisitor<void> {
void Visit(Identifier &op) override{};
void Visit(PrimitiveLiteral &op) override{};
void Visit(PropertyLookup &op) override{};
void Visit(AllPropertiesLookup &op) override{};
void Visit(ParameterLookup &op) override{};
void Visit(NamedExpression &op) override{};
void Visit(RegexMatch &op) override{};

View File

@@ -124,11 +124,18 @@ class ReturnBodyContext : public HierarchicalTreeVisitor {
bool PostVisit(MapLiteral &map_literal) override {
MG_ASSERT(map_literal.elements_.size() <= has_aggregation_.size(),
"Expected has_aggregation_ flags as much as there are map elements.");
"Expected as many has_aggregation_ flags as there are map elements.");
PostVisitCollectionLiteral(map_literal, [](auto it) { return it->second; });
return true;
}
bool PostVisit(MapProjectionLiteral &map_projection_literal) override {
MG_ASSERT(map_projection_literal.elements_.size() <= has_aggregation_.size(),
"Expected as many has_aggregation_ flags as there are map elements.");
PostVisitCollectionLiteral(map_projection_literal, [](auto it) { return it->second; });
return true;
}
bool PostVisit(All &all) override {
// Remove the symbol which is bound by all, because we are only interested
// in free (unbound) symbols.

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -14,6 +14,7 @@
#include <datetime.h>
#include <pyerrors.h>
#include <array>
#include <optional>
#include <sstream>
#include <stdexcept>
#include <string>
@@ -860,7 +861,7 @@ py::Object MgpListToPyTuple(mgp_list *list, PyObject *py_graph) {
}
namespace {
std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Object py_record) {
std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Object py_record, mgp_memory *memory) {
py::Object py_mgp(PyImport_ImportModule("mgp"));
if (!py_mgp) return py::FetchError();
auto record_cls = py_mgp.GetAttr("Record");
@@ -902,8 +903,8 @@ std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Obj
if (!field_name) return py::FetchError();
auto *val = PyTuple_GetItem(item, 1);
if (!val) return py::FetchError();
mgp_memory memory{result->rows.get_allocator().GetMemoryResource()};
mgp_value *field_val = PyObjectToMgpValueWithPythonExceptions(val, &memory);
// This memory is one dedicated for mg_procedure.
mgp_value *field_val = PyObjectToMgpValueWithPythonExceptions(val, memory);
if (field_val == nullptr) {
return py::FetchError();
}
@@ -921,15 +922,26 @@ std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Obj
return std::nullopt;
}
std::optional<py::ExceptionInfo> AddMultipleRecordsFromPython(mgp_result *result, py::Object py_seq) {
std::optional<py::ExceptionInfo> AddMultipleRecordsFromPython(mgp_result *result, py::Object py_seq,
mgp_memory *memory) {
Py_ssize_t len = PySequence_Size(py_seq.Ptr());
if (len == -1) return py::FetchError();
for (Py_ssize_t i = 0; i < len; ++i) {
py::Object py_record(PySequence_GetItem(py_seq.Ptr(), i));
result->rows.reserve(len);
// This proved to be good enough constant not to lose performance on transformation
static constexpr auto del_cnt{100000};
for (Py_ssize_t i = 0, curr_item = 0; i < len; ++i, ++curr_item) {
py::Object py_record(PySequence_GetItem(py_seq.Ptr(), curr_item));
if (!py_record) return py::FetchError();
auto maybe_exc = AddRecordFromPython(result, py_record);
auto maybe_exc = AddRecordFromPython(result, py_record, memory);
if (maybe_exc) return maybe_exc;
// Once PySequence_DelSlice deletes "transformed" objects, starting index is 0 again.
if (i && i % del_cnt == 0) {
PySequence_DelSlice(py_seq.Ptr(), 0, del_cnt);
curr_item = -1;
}
}
// Clear at the end what left
PySequence_DelSlice(py_seq.Ptr(), 0, PySequence_Size(py_seq.Ptr()));
return std::nullopt;
}
@@ -962,6 +974,7 @@ std::function<void()> PyObjectCleanup(py::Object &py_object) {
void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *graph, mgp_result *result,
mgp_memory *memory) {
// *memory here is memory from `EvalContext`
auto gil = py::EnsureGIL();
auto error_to_msg = [](const std::optional<py::ExceptionInfo> &exc_info) -> std::optional<std::string> {
@@ -979,9 +992,9 @@ void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *gra
auto py_res = py_cb.Call(py_graph, py_args);
if (!py_res) return py::FetchError();
if (PySequence_Check(py_res.Ptr())) {
return AddMultipleRecordsFromPython(result, py_res);
return AddMultipleRecordsFromPython(result, py_res, memory);
} else {
return AddRecordFromPython(result, py_res);
return AddRecordFromPython(result, py_res, memory);
}
};
@@ -1027,9 +1040,9 @@ void CallPythonTransformation(const py::Object &py_cb, mgp_messages *msgs, mgp_g
auto py_res = py_cb.Call(py_graph, py_messages);
if (!py_res) return py::FetchError();
if (PySequence_Check(py_res.Ptr())) {
return AddMultipleRecordsFromPython(result, py_res);
return AddMultipleRecordsFromPython(result, py_res, memory);
}
return AddRecordFromPython(result, py_res);
return AddRecordFromPython(result, py_res, memory);
};
// It is *VERY IMPORTANT* to note that this code takes great care not to keep

View File

@@ -36,9 +36,9 @@
#include "utils/pmr/string.hpp"
#include "utils/variant_helpers.hpp"
namespace EventCounter {
namespace memgraph::metrics {
extern const Event MessagesConsumed;
} // namespace EventCounter
} // namespace memgraph::metrics
namespace memgraph::query::stream {
namespace {
@@ -495,7 +495,7 @@ Streams::StreamsMap::iterator Streams::CreateConsumer(StreamsMap &map, const std
utils::OnScopeExit interpreter_cleanup{
[interpreter_context, interpreter]() { interpreter_context->interpreters->erase(interpreter.get()); }};
EventCounter::IncrementCounter(EventCounter::MessagesConsumed, messages.size());
memgraph::metrics::IncrementCounter(memgraph::metrics::MessagesConsumed, messages.size());
CallCustomTransformation(transformation_name, messages, result, accessor, *memory_resource, stream_name);
DiscardValueResultStream stream;

View File

@@ -25,9 +25,9 @@
#include "utils/event_counter.hpp"
#include "utils/memory.hpp"
namespace EventCounter {
namespace memgraph::metrics {
extern const Event TriggersExecuted;
} // namespace EventCounter
} // namespace memgraph::metrics
namespace memgraph::query {
namespace {
@@ -248,7 +248,7 @@ void Trigger::Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution
;
cursor->Shutdown();
EventCounter::IncrementCounter(EventCounter::TriggersExecuted);
memgraph::metrics::IncrementCounter(memgraph::metrics::TriggersExecuted);
}
namespace {

View File

@@ -10,7 +10,9 @@ set(storage_v2_src_files
indices.cpp
property_store.cpp
vertex_accessor.cpp
storage.cpp)
storage.cpp
storage_mode.cpp
isolation_level.cpp)
set(storage_v2_src_files

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -50,6 +50,11 @@ struct Config {
bool snapshot_on_exit{false};
bool restore_replicas_on_startup{false};
uint64_t items_per_batch{1'000'000};
uint64_t recovery_thread_count{8};
bool allow_parallel_index_creation{false};
} durability;
struct Transaction {

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -27,9 +27,15 @@
#include "storage/v2/durability/paths.hpp"
#include "storage/v2/durability/snapshot.hpp"
#include "storage/v2/durability/wal.hpp"
#include "utils/event_histogram.hpp"
#include "utils/logging.hpp"
#include "utils/memory_tracker.hpp"
#include "utils/message.hpp"
#include "utils/timer.hpp"
namespace memgraph::metrics {
extern const Event SnapshotRecoveryLatency_us;
} // namespace memgraph::metrics
namespace memgraph::storage::durability {
@@ -113,13 +119,15 @@ std::optional<std::vector<WalDurabilityInfo>> GetWalFiles(const std::filesystem:
// to ensure that the indices and constraints are consistent at the end of the
// recovery process.
void RecoverIndicesAndConstraints(const RecoveredIndicesAndConstraints &indices_constraints, Indices *indices,
Constraints *constraints, utils::SkipList<Vertex> *vertices) {
Constraints *constraints, utils::SkipList<Vertex> *vertices,
const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info) {
spdlog::info("Recreating indices from metadata.");
// Recover label indices.
spdlog::info("Recreating {} label indices from metadata.", indices_constraints.indices.label.size());
for (const auto &item : indices_constraints.indices.label) {
if (!indices->label_index.CreateIndex(item, vertices->access()))
if (!indices->label_index.CreateIndex(item, vertices->access(), paralell_exec_info))
throw RecoveryFailure("The label index must be created here!");
spdlog::info("A label index is recreated from metadata.");
}
spdlog::info("Label indices are recreated.");
@@ -163,7 +171,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
std::deque<std::pair<std::string, uint64_t>> *epoch_history,
utils::SkipList<Vertex> *vertices, utils::SkipList<Edge> *edges,
std::atomic<uint64_t> *edge_count, NameIdMapper *name_id_mapper,
Indices *indices, Constraints *constraints, Config::Items items,
Indices *indices, Constraints *constraints, const Config &config,
uint64_t *wal_seq_num) {
utils::MemoryTracker::OutOfMemoryExceptionEnabler oom_exception;
spdlog::info("Recovering persisted data using snapshot ({}) and WAL directory ({}).", snapshot_directory,
@@ -174,6 +182,8 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
return std::nullopt;
}
utils::Timer timer;
auto snapshot_files = GetSnapshotFiles(snapshot_directory);
RecoveryInfo recovery_info;
@@ -195,7 +205,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
}
spdlog::info("Starting snapshot recovery from {}.", path);
try {
recovered_snapshot = LoadSnapshot(path, vertices, edges, epoch_history, name_id_mapper, edge_count, items);
recovered_snapshot = LoadSnapshot(path, vertices, edges, epoch_history, name_id_mapper, edge_count, config);
spdlog::info("Snapshot recovery successful!");
break;
} catch (const RecoveryFailure &e) {
@@ -213,7 +223,11 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
*epoch_id = std::move(recovered_snapshot->snapshot_info.epoch_id);
if (!utils::DirExists(wal_directory)) {
RecoverIndicesAndConstraints(indices_constraints, indices, constraints, vertices);
const auto par_exec_info = config.durability.allow_parallel_index_creation
? std::make_optional(std::make_pair(recovery_info.vertex_batches,
config.durability.recovery_thread_count))
: std::nullopt;
RecoverIndicesAndConstraints(indices_constraints, indices, constraints, vertices, par_exec_info);
return recovered_snapshot->recovery_info;
}
} else {
@@ -319,7 +333,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
}
try {
auto info = LoadWal(wal_file.path, &indices_constraints, last_loaded_timestamp, vertices, edges, name_id_mapper,
edge_count, items);
edge_count, config.items);
recovery_info.next_vertex_id = std::max(recovery_info.next_vertex_id, info.next_vertex_id);
recovery_info.next_edge_id = std::max(recovery_info.next_edge_id, info.next_edge_id);
recovery_info.next_timestamp = std::max(recovery_info.next_timestamp, info.next_timestamp);
@@ -341,6 +355,10 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
}
RecoverIndicesAndConstraints(indices_constraints, indices, constraints, vertices);
memgraph::metrics::Measure(memgraph::metrics::SnapshotRecoveryLatency_us,
std::chrono::duration_cast<std::chrono::microseconds>(timer.Elapsed()).count());
return recovery_info;
}

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -91,13 +91,18 @@ std::optional<std::vector<WalDurabilityInfo>> GetWalFiles(const std::filesystem:
std::string_view uuid = "",
std::optional<size_t> current_seq_num = {});
using ParalellizedIndexCreationInfo =
std::pair<std::vector<std::pair<Gid, uint64_t>> /*vertex_recovery_info*/, uint64_t /*thread_count*/>;
// Helper function used to recover all discovered indices and constraints. The
// indices and constraints must be recovered after the data recovery is done
// to ensure that the indices and constraints are consistent at the end of the
// recovery process.
/// @throw RecoveryFailure
void RecoverIndicesAndConstraints(const RecoveredIndicesAndConstraints &indices_constraints, Indices *indices,
Constraints *constraints, utils::SkipList<Vertex> *vertices);
void RecoverIndicesAndConstraints(
const RecoveredIndicesAndConstraints &indices_constraints, Indices *indices, Constraints *constraints,
utils::SkipList<Vertex> *vertices,
const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info = std::nullopt);
/// Recovers data either from a snapshot and/or WAL files.
/// @throw RecoveryFailure
@@ -108,7 +113,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
std::deque<std::pair<std::string, uint64_t>> *epoch_history,
utils::SkipList<Vertex> *vertices, utils::SkipList<Edge> *edges,
std::atomic<uint64_t> *edge_count, NameIdMapper *name_id_mapper,
Indices *indices, Constraints *constraints, Config::Items items,
Indices *indices, Constraints *constraints, const Config &config,
uint64_t *wal_seq_num);
} // namespace memgraph::storage::durability

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -12,6 +12,7 @@
#pragma once
#include <algorithm>
#include <optional>
#include <set>
#include <utility>
#include <vector>
@@ -29,6 +30,8 @@ struct RecoveryInfo {
// last timestamp read from a WAL file
std::optional<uint64_t> last_commit_timestamp;
std::vector<std::pair<Gid /*first vertex gid*/, uint64_t /*batch size*/>> vertex_batches;
};
/// Structure used to track indices and constraints during recovery.

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -11,18 +11,26 @@
#include "storage/v2/durability/snapshot.hpp"
#include <thread>
#include "storage/v2/durability/exceptions.hpp"
#include "storage/v2/durability/paths.hpp"
#include "storage/v2/durability/serialization.hpp"
#include "storage/v2/durability/version.hpp"
#include "storage/v2/durability/wal.hpp"
#include "storage/v2/edge.hpp"
#include "storage/v2/edge_accessor.hpp"
#include "storage/v2/edge_ref.hpp"
#include "storage/v2/id_types.hpp"
#include "storage/v2/mvcc.hpp"
#include "storage/v2/vertex.hpp"
#include "storage/v2/vertex_accessor.hpp"
#include "utils/concepts.hpp"
#include "utils/file_locker.hpp"
#include "utils/logging.hpp"
#include "utils/message.hpp"
#include "utils/spin_lock.hpp"
#include "utils/synchronized.hpp"
namespace memgraph::storage::durability {
@@ -40,6 +48,8 @@ namespace memgraph::storage::durability {
// * offset to the constraints section
// * offset to the mapper section
// * offset to the metadata section
// * offset to the offset-count pair of the first edge batch (`0` if properties on edges are disabled)
// * offset to the offset-count pair of the first vertex batch
//
// 4) Encoded edges (if properties on edges are enabled); each edge is written
// in the following format:
@@ -87,9 +97,23 @@ namespace memgraph::storage::durability {
// * number of edges
// * number of vertices
//
// 10) Batch infos
// * number of edge batch infos
// * edge batch infos
// * starting offset of the batch
// * number of edges in the batch
// * vertex batch infos
// * starting offset of the batch
// * number of vertices in the batch
//
// IMPORTANT: When changing snapshot encoding/decoding bump the snapshot/WAL
// version in `version.hpp`.
struct BatchInfo {
uint64_t offset;
uint64_t count;
};
// Function used to read information about the snapshot file.
SnapshotInfo ReadSnapshotInfo(const std::filesystem::path &path) {
// Check magic and version.
@@ -124,6 +148,13 @@ SnapshotInfo ReadSnapshotInfo(const std::filesystem::path &path) {
info.offset_mapper = read_offset();
info.offset_epoch_history = read_offset();
info.offset_metadata = read_offset();
if (*version >= 15U) {
info.offset_edge_batches = read_offset();
info.offset_vertex_batches = read_offset();
} else {
info.offset_edge_batches = 0U;
info.offset_vertex_batches = 0U;
}
}
// Read metadata.
@@ -157,17 +188,385 @@ SnapshotInfo ReadSnapshotInfo(const std::filesystem::path &path) {
return info;
}
RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipList<Vertex> *vertices,
utils::SkipList<Edge> *edges,
std::deque<std::pair<std::string, uint64_t>> *epoch_history,
NameIdMapper *name_id_mapper, std::atomic<uint64_t> *edge_count, Config::Items items) {
std::vector<BatchInfo> ReadBatchInfos(Decoder &snapshot) {
std::vector<BatchInfo> infos;
const auto infos_size = snapshot.ReadUint();
if (!infos_size.has_value()) {
throw RecoveryFailure("Invalid snapshot data!");
}
infos.reserve(*infos_size);
for (auto i{0U}; i < *infos_size; ++i) {
const auto offset = snapshot.ReadUint();
if (!offset.has_value()) {
throw RecoveryFailure("Invalid snapshot data!");
}
const auto count = snapshot.ReadUint();
if (!count.has_value()) {
throw RecoveryFailure("Invalid snapshot data!");
}
infos.push_back(BatchInfo{*offset, *count});
}
return infos;
}
template <typename TFunc>
void LoadPartialEdges(const std::filesystem::path &path, utils::SkipList<Edge> &edges, const uint64_t from_offset,
const uint64_t edges_count, const Config::Items items, TFunc get_property_from_id) {
Decoder snapshot;
snapshot.Initialize(path, kSnapshotMagic);
// Recover edges.
auto edge_acc = edges.access();
uint64_t last_edge_gid = 0;
spdlog::info("Recovering {} edges.", edges_count);
if (!snapshot.SetPosition(from_offset)) throw RecoveryFailure("Couldn't read data from snapshot!");
std::vector<std::pair<PropertyId, PropertyValue>> read_properties;
for (uint64_t i = 0; i < edges_count; ++i) {
{
const auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_EDGE) throw RecoveryFailure("Invalid snapshot data!");
}
// Read edge GID.
auto gid = snapshot.ReadUint();
if (!gid) throw RecoveryFailure("Invalid snapshot data!");
if (i > 0 && *gid <= last_edge_gid) throw RecoveryFailure("Invalid snapshot data!");
last_edge_gid = *gid;
if (items.properties_on_edges) {
spdlog::debug("Recovering edge {} with properties.", *gid);
auto [it, inserted] = edge_acc.insert(Edge{Gid::FromUint(*gid), nullptr});
if (!inserted) throw RecoveryFailure("The edge must be inserted here!");
// Recover properties.
{
auto props_size = snapshot.ReadUint();
if (!props_size) throw RecoveryFailure("Invalid snapshot data!");
auto &props = it->properties;
read_properties.clear();
read_properties.reserve(*props_size);
for (uint64_t j = 0; j < *props_size; ++j) {
auto key = snapshot.ReadUint();
if (!key) throw RecoveryFailure("Invalid snapshot data!");
auto value = snapshot.ReadPropertyValue();
if (!value) throw RecoveryFailure("Invalid snapshot data!");
read_properties.emplace_back(get_property_from_id(*key), std::move(*value));
}
props.InitProperties(std::move(read_properties));
}
} else {
spdlog::debug("Ensuring edge {} doesn't have any properties.", *gid);
// Read properties.
{
auto props_size = snapshot.ReadUint();
if (!props_size) throw RecoveryFailure("Invalid snapshot data!");
if (*props_size != 0)
throw RecoveryFailure(
"The snapshot has properties on edges, but the storage is "
"configured without properties on edges!");
}
}
}
spdlog::info("Partial edges are recovered.");
}
// Returns the gid of the last recovered vertex
template <typename TLabelFromIdFunc, typename TPropertyFromIdFunc>
uint64_t LoadPartialVertices(const std::filesystem::path &path, utils::SkipList<Vertex> &vertices,
const uint64_t from_offset, const uint64_t vertices_count,
TLabelFromIdFunc get_label_from_id, TPropertyFromIdFunc get_property_from_id) {
Decoder snapshot;
snapshot.Initialize(path, kSnapshotMagic);
if (!snapshot.SetPosition(from_offset)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto vertex_acc = vertices.access();
uint64_t last_vertex_gid = 0;
spdlog::info("Recovering {} vertices.", vertices_count);
std::vector<std::pair<PropertyId, PropertyValue>> read_properties;
for (uint64_t i = 0; i < vertices_count; ++i) {
{
auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_VERTEX) throw RecoveryFailure("Invalid snapshot data!");
}
// Insert vertex.
auto gid = snapshot.ReadUint();
if (!gid) throw RecoveryFailure("Invalid snapshot data!");
if (i > 0 && *gid <= last_vertex_gid) {
throw RecoveryFailure("Invalid snapshot data!");
}
last_vertex_gid = *gid;
spdlog::debug("Recovering vertex {}.", *gid);
auto [it, inserted] = vertex_acc.insert(Vertex{Gid::FromUint(*gid), nullptr});
if (!inserted) throw RecoveryFailure("The vertex must be inserted here!");
// Recover labels.
spdlog::trace("Recovering labels for vertex {}.", *gid);
{
auto labels_size = snapshot.ReadUint();
if (!labels_size) throw RecoveryFailure("Invalid snapshot data!");
auto &labels = it->labels;
labels.reserve(*labels_size);
for (uint64_t j = 0; j < *labels_size; ++j) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
labels.emplace_back(get_label_from_id(*label));
}
}
// Recover properties.
spdlog::trace("Recovering properties for vertex {}.", *gid);
{
auto props_size = snapshot.ReadUint();
if (!props_size) throw RecoveryFailure("Invalid snapshot data!");
auto &props = it->properties;
read_properties.clear();
read_properties.reserve(*props_size);
for (uint64_t j = 0; j < *props_size; ++j) {
auto key = snapshot.ReadUint();
if (!key) throw RecoveryFailure("Invalid snapshot data!");
auto value = snapshot.ReadPropertyValue();
if (!value) throw RecoveryFailure("Invalid snapshot data!");
read_properties.emplace_back(get_property_from_id(*key), std::move(*value));
}
props.InitProperties(std::move(read_properties));
}
// Skip in edges.
{
auto in_size = snapshot.ReadUint();
if (!in_size) throw RecoveryFailure("Invalid snapshot data!");
for (uint64_t j = 0; j < *in_size; ++j) {
auto edge_gid = snapshot.ReadUint();
if (!edge_gid) throw RecoveryFailure("Invalid snapshot data!");
auto from_gid = snapshot.ReadUint();
if (!from_gid) throw RecoveryFailure("Invalid snapshot data!");
auto edge_type = snapshot.ReadUint();
if (!edge_type) throw RecoveryFailure("Invalid snapshot data!");
}
}
// Skip out edges.
auto out_size = snapshot.ReadUint();
if (!out_size) throw RecoveryFailure("Invalid snapshot data!");
for (uint64_t j = 0; j < *out_size; ++j) {
auto edge_gid = snapshot.ReadUint();
if (!edge_gid) throw RecoveryFailure("Invalid snapshot data!");
auto to_gid = snapshot.ReadUint();
if (!to_gid) throw RecoveryFailure("Invalid snapshot data!");
auto edge_type = snapshot.ReadUint();
if (!edge_type) throw RecoveryFailure("Invalid snapshot data!");
}
}
spdlog::info("Partial vertices are recovered.");
return last_vertex_gid;
}
// Returns the number of edges recovered
struct LoadPartialConnectivityResult {
uint64_t edge_count;
uint64_t highest_edge_id;
Gid first_vertex_gid;
};
template <typename TEdgeTypeFromIdFunc>
LoadPartialConnectivityResult LoadPartialConnectivity(const std::filesystem::path &path,
utils::SkipList<Vertex> &vertices, utils::SkipList<Edge> &edges,
const uint64_t from_offset, const uint64_t vertices_count,
const Config::Items items, const bool snapshot_has_edges,
TEdgeTypeFromIdFunc get_edge_type_from_id) {
Decoder snapshot;
snapshot.Initialize(path, kSnapshotMagic);
if (!snapshot.SetPosition(from_offset)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto vertex_acc = vertices.access();
auto edge_acc = edges.access();
// Read the first gid to find the necessary iterator in vertices
const auto first_vertex_gid = std::invoke([&]() mutable {
{
auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_VERTEX) throw RecoveryFailure("Invalid snapshot data!");
}
auto gid = snapshot.ReadUint();
if (!gid) throw RecoveryFailure("Invalid snapshot data!");
return Gid::FromUint(*gid);
});
uint64_t edge_count{0};
uint64_t highest_edge_gid{0};
auto vertex_it = vertex_acc.find(first_vertex_gid);
if (vertex_it == vertex_acc.end()) {
throw RecoveryFailure("Invalid snapshot data!");
}
spdlog::info("Recovering connectivity for {} vertices.", vertices_count);
if (!snapshot.SetPosition(from_offset)) throw RecoveryFailure("Couldn't read data from snapshot!");
for (uint64_t i = 0; i < vertices_count; ++i) {
auto &vertex = *vertex_it;
{
auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_VERTEX) throw RecoveryFailure("Invalid snapshot data!");
}
auto gid = snapshot.ReadUint();
if (!gid) throw RecoveryFailure("Invalid snapshot data!");
if (gid != vertex.gid.AsUint()) throw RecoveryFailure("Invalid snapshot data!");
// Skip labels.
{
auto labels_size = snapshot.ReadUint();
if (!labels_size) throw RecoveryFailure("Invalid snapshot data!");
for (uint64_t j = 0; j < *labels_size; ++j) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
}
}
// Skip properties.
{
auto props_size = snapshot.ReadUint();
if (!props_size) throw RecoveryFailure("Invalid snapshot data!");
for (uint64_t j = 0; j < *props_size; ++j) {
auto key = snapshot.ReadUint();
if (!key) throw RecoveryFailure("Invalid snapshot data!");
auto value = snapshot.SkipPropertyValue();
if (!value) throw RecoveryFailure("Invalid snapshot data!");
}
}
// Recover in edges.
{
spdlog::trace("Recovering inbound edges for vertex {}.", vertex.gid.AsUint());
auto in_size = snapshot.ReadUint();
if (!in_size) throw RecoveryFailure("Invalid snapshot data!");
vertex.in_edges.reserve(*in_size);
for (uint64_t j = 0; j < *in_size; ++j) {
auto edge_gid = snapshot.ReadUint();
if (!edge_gid) throw RecoveryFailure("Invalid snapshot data!");
highest_edge_gid = std::max(highest_edge_gid, *edge_gid);
auto from_gid = snapshot.ReadUint();
if (!from_gid) throw RecoveryFailure("Invalid snapshot data!");
auto edge_type = snapshot.ReadUint();
if (!edge_type) throw RecoveryFailure("Invalid snapshot data!");
auto from_vertex = vertex_acc.find(Gid::FromUint(*from_gid));
if (from_vertex == vertex_acc.end()) throw RecoveryFailure("Invalid from vertex!");
EdgeRef edge_ref(Gid::FromUint(*edge_gid));
if (items.properties_on_edges) {
// The snapshot contains the individiual edges only if it was created with a config where properties are
// allowed on edges. That means the snapshots that were created without edge properties will only contain the
// edges in the in/out edges list of vertices, therefore the edges has to be created here.
if (snapshot_has_edges) {
auto edge = edge_acc.find(Gid::FromUint(*edge_gid));
if (edge == edge_acc.end()) throw RecoveryFailure("Invalid edge!");
edge_ref = EdgeRef(&*edge);
} else {
auto [edge, inserted] = edge_acc.insert(Edge{Gid::FromUint(*edge_gid), nullptr});
edge_ref = EdgeRef(&*edge);
}
}
vertex.in_edges.emplace_back(get_edge_type_from_id(*edge_type), &*from_vertex, edge_ref);
}
}
// Recover out edges.
{
spdlog::trace("Recovering outbound edges for vertex {}.", vertex.gid.AsUint());
auto out_size = snapshot.ReadUint();
if (!out_size) throw RecoveryFailure("Invalid snapshot data!");
vertex.out_edges.reserve(*out_size);
for (uint64_t j = 0; j < *out_size; ++j) {
auto edge_gid = snapshot.ReadUint();
if (!edge_gid) throw RecoveryFailure("Invalid snapshot data!");
auto to_gid = snapshot.ReadUint();
if (!to_gid) throw RecoveryFailure("Invalid snapshot data!");
auto edge_type = snapshot.ReadUint();
if (!edge_type) throw RecoveryFailure("Invalid snapshot data!");
auto to_vertex = vertex_acc.find(Gid::FromUint(*to_gid));
if (to_vertex == vertex_acc.end()) throw RecoveryFailure("Invalid to vertex!");
EdgeRef edge_ref(Gid::FromUint(*edge_gid));
if (items.properties_on_edges) {
// The snapshot contains the individiual edges only if it was created with a config where properties are
// allowed on edges. That means the snapshots that were created without edge properties will only contain the
// edges in the in/out edges list of vertices, therefore the edges has to be created here.
if (snapshot_has_edges) {
auto edge = edge_acc.find(Gid::FromUint(*edge_gid));
if (edge == edge_acc.end()) throw RecoveryFailure("Invalid edge!");
edge_ref = EdgeRef(&*edge);
} else {
auto [edge, inserted] = edge_acc.insert(Edge{Gid::FromUint(*edge_gid), nullptr});
edge_ref = EdgeRef(&*edge);
}
}
vertex.out_edges.emplace_back(get_edge_type_from_id(*edge_type), &*to_vertex, edge_ref);
// Increment edge count. We only increment the count here because the
// information is duplicated in in_edges.
edge_count++;
}
}
++vertex_it;
}
spdlog::info("Partial connectivities are recovered.");
return {edge_count, highest_edge_gid, first_vertex_gid};
}
template <typename TFunc>
void RecoverOnMultipleThreads(size_t thread_count, const TFunc &func, const std::vector<BatchInfo> &batches) {
utils::Synchronized<std::optional<RecoveryFailure>, utils::SpinLock> maybe_error{};
{
std::atomic<uint64_t> batch_counter = 0;
thread_count = std::min(thread_count, batches.size());
std::vector<std::jthread> threads;
threads.reserve(thread_count);
for (auto i{0U}; i < thread_count; ++i) {
threads.emplace_back([&func, &batches, &maybe_error, &batch_counter]() {
while (!maybe_error.Lock()->has_value()) {
const auto batch_index = batch_counter++;
if (batch_index >= batches.size()) {
return;
}
const auto &batch = batches[batch_index];
try {
func(batch_index, batch);
} catch (RecoveryFailure &failure) {
*maybe_error.Lock() = std::move(failure);
}
}
});
}
}
if (maybe_error.Lock()->has_value()) {
throw RecoveryFailure((*maybe_error.Lock())->what());
}
}
RecoveredSnapshot LoadSnapshotVersion14(const std::filesystem::path &path, utils::SkipList<Vertex> *vertices,
utils::SkipList<Edge> *edges,
std::deque<std::pair<std::string, uint64_t>> *epoch_history,
NameIdMapper *name_id_mapper, std::atomic<uint64_t> *edge_count,
Config::Items items) {
RecoveryInfo ret;
RecoveredIndicesAndConstraints indices_constraints;
Decoder snapshot;
auto version = snapshot.Initialize(path, kSnapshotMagic);
if (!version) throw RecoveryFailure("Couldn't read snapshot magic and/or version!");
if (!IsVersionSupported(*version)) throw RecoveryFailure(fmt::format("Invalid snapshot version {}", *version));
if (*version != 14U) throw RecoveryFailure(fmt::format("Expected snapshot version is 14, but got {}", *version));
// Cleanup of loaded data in case of failure.
bool success = false;
@@ -625,10 +1024,297 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
return {info, ret, std::move(indices_constraints)};
}
RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipList<Vertex> *vertices,
utils::SkipList<Edge> *edges,
std::deque<std::pair<std::string, uint64_t>> *epoch_history,
NameIdMapper *name_id_mapper, std::atomic<uint64_t> *edge_count, const Config &config) {
RecoveryInfo recovery_info;
RecoveredIndicesAndConstraints indices_constraints;
Decoder snapshot;
const auto version = snapshot.Initialize(path, kSnapshotMagic);
if (!version) throw RecoveryFailure("Couldn't read snapshot magic and/or version!");
if (!IsVersionSupported(*version)) throw RecoveryFailure(fmt::format("Invalid snapshot version {}", *version));
if (*version == 14U) {
return LoadSnapshotVersion14(path, vertices, edges, epoch_history, name_id_mapper, edge_count, config.items);
}
// Cleanup of loaded data in case of failure.
bool success = false;
utils::OnScopeExit cleanup([&] {
if (!success) {
edges->clear();
vertices->clear();
epoch_history->clear();
}
});
// Read snapshot info.
const auto info = ReadSnapshotInfo(path);
spdlog::info("Recovering {} vertices and {} edges.", info.vertices_count, info.edges_count);
// Check for edges.
bool snapshot_has_edges = info.offset_edges != 0;
// Recover mapper.
std::unordered_map<uint64_t, uint64_t> snapshot_id_map;
{
spdlog::info("Recovering mapper metadata.");
if (!snapshot.SetPosition(info.offset_mapper)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_MAPPER) throw RecoveryFailure("Invalid snapshot data!");
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
for (uint64_t i = 0; i < *size; ++i) {
auto id = snapshot.ReadUint();
if (!id) throw RecoveryFailure("Invalid snapshot data!");
auto name = snapshot.ReadString();
if (!name) throw RecoveryFailure("Invalid snapshot data!");
auto my_id = name_id_mapper->NameToId(*name);
snapshot_id_map.emplace(*id, my_id);
SPDLOG_TRACE("Mapping \"{}\"from snapshot id {} to actual id {}.", *name, *id, my_id);
}
}
auto get_label_from_id = [&snapshot_id_map](uint64_t snapshot_id) {
auto it = snapshot_id_map.find(snapshot_id);
if (it == snapshot_id_map.end()) throw RecoveryFailure("Invalid snapshot data!");
return LabelId::FromUint(it->second);
};
auto get_property_from_id = [&snapshot_id_map](uint64_t snapshot_id) {
auto it = snapshot_id_map.find(snapshot_id);
if (it == snapshot_id_map.end()) throw RecoveryFailure("Invalid snapshot data!");
return PropertyId::FromUint(it->second);
};
auto get_edge_type_from_id = [&snapshot_id_map](uint64_t snapshot_id) {
auto it = snapshot_id_map.find(snapshot_id);
if (it == snapshot_id_map.end()) throw RecoveryFailure("Invalid snapshot data!");
return EdgeTypeId::FromUint(it->second);
};
// Reset current edge count.
edge_count->store(0, std::memory_order_release);
{
spdlog::info("Recovering edges.");
// Recover edges.
if (snapshot_has_edges) {
// We don't need to check whether we store properties on edge or not, because `LoadPartialEdges` will always
// iterate over the edges in the snapshot (if they exist) and the current configuration of properties on edge only
// affect what it does:
// 1. If properties are allowed on edges, then it loads the edges.
// 2. If properties are not allowed on edges, then it checks that none of the edges have any properties.
if (!snapshot.SetPosition(info.offset_edge_batches)) {
throw RecoveryFailure("Couldn't read data from snapshot!");
}
const auto edge_batches = ReadBatchInfos(snapshot);
RecoverOnMultipleThreads(
config.durability.recovery_thread_count,
[path, edges, items = config.items, &get_property_from_id](const size_t /*batch_index*/,
const BatchInfo &batch) {
LoadPartialEdges(path, *edges, batch.offset, batch.count, items, get_property_from_id);
},
edge_batches);
}
spdlog::info("Edges are recovered.");
// Recover vertices (labels and properties).
spdlog::info("Recovering vertices.", info.vertices_count);
uint64_t last_vertex_gid{0};
if (!snapshot.SetPosition(info.offset_vertex_batches)) {
throw RecoveryFailure("Couldn't read data from snapshot!");
}
const auto vertex_batches = ReadBatchInfos(snapshot);
RecoverOnMultipleThreads(
config.durability.recovery_thread_count,
[path, vertices, &vertex_batches, &get_label_from_id, &get_property_from_id, &last_vertex_gid](
const size_t batch_index, const BatchInfo &batch) {
const auto last_vertex_gid_in_batch =
LoadPartialVertices(path, *vertices, batch.offset, batch.count, get_label_from_id, get_property_from_id);
if (batch_index == vertex_batches.size() - 1) {
last_vertex_gid = last_vertex_gid_in_batch;
}
},
vertex_batches);
spdlog::info("Vertices are recovered.");
// Recover vertices (in/out edges).
spdlog::info("Recover connectivity.");
recovery_info.vertex_batches.reserve(vertex_batches.size());
for (const auto batch : vertex_batches) {
recovery_info.vertex_batches.emplace_back(std::make_pair(Gid::FromUint(0), batch.count));
}
std::atomic<uint64_t> highest_edge_gid{0};
RecoverOnMultipleThreads(
config.durability.recovery_thread_count,
[path, vertices, edges, edge_count, items = config.items, snapshot_has_edges, &get_edge_type_from_id,
&highest_edge_gid, &recovery_info](const size_t batch_index, const BatchInfo &batch) {
const auto result = LoadPartialConnectivity(path, *vertices, *edges, batch.offset, batch.count, items,
snapshot_has_edges, get_edge_type_from_id);
edge_count->fetch_add(result.edge_count);
auto known_highest_edge_gid = highest_edge_gid.load();
while (known_highest_edge_gid < result.highest_edge_id) {
highest_edge_gid.compare_exchange_weak(known_highest_edge_gid, result.highest_edge_id);
}
recovery_info.vertex_batches[batch_index].first = result.first_vertex_gid;
},
vertex_batches);
spdlog::info("Connectivity is recovered.");
// Set initial values for edge/vertex ID generators.
recovery_info.next_edge_id = highest_edge_gid + 1;
recovery_info.next_vertex_id = last_vertex_gid + 1;
}
// Recover indices.
{
spdlog::info("Recovering metadata of indices.");
if (!snapshot.SetPosition(info.offset_indices)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_INDICES) throw RecoveryFailure("Invalid snapshot data!");
// Recover label indices.
{
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
spdlog::info("Recovering metadata of {} label indices.", *size);
for (uint64_t i = 0; i < *size; ++i) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
AddRecoveredIndexConstraint(&indices_constraints.indices.label, get_label_from_id(*label),
"The label index already exists!");
SPDLOG_TRACE("Recovered metadata of label index for :{}", name_id_mapper->IdToName(snapshot_id_map.at(*label)));
}
spdlog::info("Metadata of label indices are recovered.");
}
// Recover label+property indices.
{
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
spdlog::info("Recovering metadata of {} label+property indices.", *size);
for (uint64_t i = 0; i < *size; ++i) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
auto property = snapshot.ReadUint();
if (!property) throw RecoveryFailure("Invalid snapshot data!");
AddRecoveredIndexConstraint(&indices_constraints.indices.label_property,
{get_label_from_id(*label), get_property_from_id(*property)},
"The label+property index already exists!");
SPDLOG_TRACE("Recovered metadata of label+property index for :{}({})",
name_id_mapper->IdToName(snapshot_id_map.at(*label)),
name_id_mapper->IdToName(snapshot_id_map.at(*property)));
}
spdlog::info("Metadata of label+property indices are recovered.");
}
spdlog::info("Metadata of indices are recovered.");
}
// Recover constraints.
{
spdlog::info("Recovering metadata of constraints.");
if (!snapshot.SetPosition(info.offset_constraints)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_CONSTRAINTS) throw RecoveryFailure("Invalid snapshot data!");
// Recover existence constraints.
{
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
spdlog::info("Recovering metadata of {} existence constraints.", *size);
for (uint64_t i = 0; i < *size; ++i) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
auto property = snapshot.ReadUint();
if (!property) throw RecoveryFailure("Invalid snapshot data!");
AddRecoveredIndexConstraint(&indices_constraints.constraints.existence,
{get_label_from_id(*label), get_property_from_id(*property)},
"The existence constraint already exists!");
SPDLOG_TRACE("Recovered metadata of existence constraint for :{}({})",
name_id_mapper->IdToName(snapshot_id_map.at(*label)),
name_id_mapper->IdToName(snapshot_id_map.at(*property)));
}
spdlog::info("Metadata of existence constraints are recovered.");
}
// Recover unique constraints.
// Snapshot version should be checked since unique constraints were
// implemented in later versions of snapshot.
if (*version >= kUniqueConstraintVersion) {
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
spdlog::info("Recovering metadata of {} unique constraints.", *size);
for (uint64_t i = 0; i < *size; ++i) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
auto properties_count = snapshot.ReadUint();
if (!properties_count) throw RecoveryFailure("Invalid snapshot data!");
std::set<PropertyId> properties;
for (uint64_t j = 0; j < *properties_count; ++j) {
auto property = snapshot.ReadUint();
if (!property) throw RecoveryFailure("Invalid snapshot data!");
properties.insert(get_property_from_id(*property));
}
AddRecoveredIndexConstraint(&indices_constraints.constraints.unique, {get_label_from_id(*label), properties},
"The unique constraint already exists!");
SPDLOG_TRACE("Recovered metadata of unique constraints for :{}",
name_id_mapper->IdToName(snapshot_id_map.at(*label)));
}
spdlog::info("Metadata of unique constraints are recovered.");
}
spdlog::info("Metadata of constraints are recovered.");
}
spdlog::info("Recovering metadata.");
// Recover epoch history
{
if (!snapshot.SetPosition(info.offset_epoch_history)) throw RecoveryFailure("Couldn't read data from snapshot!");
const auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_EPOCH_HISTORY) throw RecoveryFailure("Invalid snapshot data!");
const auto history_size = snapshot.ReadUint();
if (!history_size) {
throw RecoveryFailure("Invalid snapshot data!");
}
for (int i = 0; i < *history_size; ++i) {
auto maybe_epoch_id = snapshot.ReadString();
if (!maybe_epoch_id) {
throw RecoveryFailure("Invalid snapshot data!");
}
const auto maybe_last_commit_timestamp = snapshot.ReadUint();
if (!maybe_last_commit_timestamp) {
throw RecoveryFailure("Invalid snapshot data!");
}
epoch_history->emplace_back(std::move(*maybe_epoch_id), *maybe_last_commit_timestamp);
}
}
spdlog::info("Metadata recovered.");
// Recover timestamp.
recovery_info.next_timestamp = info.start_timestamp + 1;
// Set success flag (to disable cleanup).
success = true;
return {info, recovery_info, std::move(indices_constraints)};
}
void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snapshot_directory,
const std::filesystem::path &wal_directory, uint64_t snapshot_retention_count,
utils::SkipList<Vertex> *vertices, utils::SkipList<Edge> *edges, NameIdMapper *name_id_mapper,
Indices *indices, Constraints *constraints, Config::Items items, const std::string &uuid,
Indices *indices, Constraints *constraints, const Config &config, const std::string &uuid,
const std::string_view epoch_id, const std::deque<std::pair<std::string, uint64_t>> &epoch_history,
utils::FileRetainer *file_retainer) {
// Ensure that the storage directory exists.
@@ -649,6 +1335,8 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
uint64_t offset_mapper = 0;
uint64_t offset_metadata = 0;
uint64_t offset_epoch_history = 0;
uint64_t offset_edge_batches = 0;
uint64_t offset_vertex_batches = 0;
{
snapshot.WriteMarker(Marker::SECTION_OFFSETS);
offset_offsets = snapshot.GetPosition();
@@ -659,6 +1347,8 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
snapshot.WriteUint(offset_mapper);
snapshot.WriteUint(offset_epoch_history);
snapshot.WriteUint(offset_metadata);
snapshot.WriteUint(offset_edge_batches);
snapshot.WriteUint(offset_vertex_batches);
}
// Object counters.
@@ -672,9 +1362,13 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
snapshot.WriteUint(mapping.AsUint());
};
std::vector<BatchInfo> edge_batch_infos;
auto items_in_current_batch{0UL};
auto batch_start_offset{0UL};
// Store all edges.
if (items.properties_on_edges) {
if (config.items.properties_on_edges) {
offset_edges = snapshot.GetPosition();
batch_start_offset = offset_edges;
auto acc = edges->access();
for (auto &edge : acc) {
// The edge visibility check must be done here manually because we don't
@@ -713,8 +1407,8 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
// type and invalid from/to pointers because we don't know them here,
// but that isn't an issue because we won't use that part of the API
// here.
auto ea =
EdgeAccessor{edge_ref, EdgeTypeId::FromUint(0UL), nullptr, nullptr, transaction, indices, constraints, items};
auto ea = EdgeAccessor{
edge_ref, EdgeTypeId::FromUint(0UL), nullptr, nullptr, transaction, indices, constraints, config.items};
// Get edge data.
auto maybe_props = ea.Properties(View::OLD);
@@ -733,16 +1427,29 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
}
++edges_count;
++items_in_current_batch;
if (items_in_current_batch == config.durability.items_per_batch) {
edge_batch_infos.push_back(BatchInfo{batch_start_offset, items_in_current_batch});
batch_start_offset = snapshot.GetPosition();
items_in_current_batch = 0;
}
}
}
if (items_in_current_batch > 0) {
edge_batch_infos.push_back(BatchInfo{batch_start_offset, items_in_current_batch});
}
std::vector<BatchInfo> vertex_batch_infos;
// Store all vertices.
{
items_in_current_batch = 0;
offset_vertices = snapshot.GetPosition();
batch_start_offset = offset_vertices;
auto acc = vertices->access();
for (auto &vertex : acc) {
// The visibility check is implemented for vertices so we use it here.
auto va = VertexAccessor::Create(&vertex, transaction, indices, constraints, items, View::OLD);
auto va = VertexAccessor::Create(&vertex, transaction, indices, constraints, config.items, View::OLD);
if (!va) continue;
// Get vertex data.
@@ -789,6 +1496,16 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
}
++vertices_count;
++items_in_current_batch;
if (items_in_current_batch == config.durability.items_per_batch) {
vertex_batch_infos.push_back(BatchInfo{batch_start_offset, items_in_current_batch});
batch_start_offset = snapshot.GetPosition();
items_in_current_batch = 0;
}
}
if (items_in_current_batch > 0) {
vertex_batch_infos.push_back(BatchInfo{batch_start_offset, items_in_current_batch});
}
}
@@ -879,6 +1596,26 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
snapshot.WriteUint(vertices_count);
}
auto write_batch_infos = [&snapshot](const std::vector<BatchInfo> &batch_infos) {
snapshot.WriteUint(batch_infos.size());
for (const auto &batch_info : batch_infos) {
snapshot.WriteUint(batch_info.offset);
snapshot.WriteUint(batch_info.count);
}
};
// Write edge batches
{
offset_edge_batches = snapshot.GetPosition();
write_batch_infos(edge_batch_infos);
}
// Write vertex batches
{
offset_vertex_batches = snapshot.GetPosition();
write_batch_infos(vertex_batch_infos);
}
// Write true offsets.
{
snapshot.SetPosition(offset_offsets);
@@ -889,6 +1626,8 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
snapshot.WriteUint(offset_mapper);
snapshot.WriteUint(offset_epoch_history);
snapshot.WriteUint(offset_metadata);
snapshot.WriteUint(offset_edge_batches);
snapshot.WriteUint(offset_vertex_batches);
}
// Finalize snapshot file.

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -37,6 +37,8 @@ struct SnapshotInfo {
uint64_t offset_mapper;
uint64_t offset_epoch_history;
uint64_t offset_metadata;
uint64_t offset_edge_batches;
uint64_t offset_vertex_batches;
std::string uuid;
std::string epoch_id;
@@ -62,13 +64,13 @@ SnapshotInfo ReadSnapshotInfo(const std::filesystem::path &path);
RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipList<Vertex> *vertices,
utils::SkipList<Edge> *edges,
std::deque<std::pair<std::string, uint64_t>> *epoch_history,
NameIdMapper *name_id_mapper, std::atomic<uint64_t> *edge_count, Config::Items items);
NameIdMapper *name_id_mapper, std::atomic<uint64_t> *edge_count, const Config &config);
/// Function used to create a snapshot using the given transaction.
void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snapshot_directory,
const std::filesystem::path &wal_directory, uint64_t snapshot_retention_count,
utils::SkipList<Vertex> *vertices, utils::SkipList<Edge> *edges, NameIdMapper *name_id_mapper,
Indices *indices, Constraints *constraints, Config::Items items, const std::string &uuid,
Indices *indices, Constraints *constraints, const Config &config, const std::string &uuid,
std::string_view epoch_id, const std::deque<std::pair<std::string, uint64_t>> &epoch_history,
utils::FileRetainer *file_retainer);

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -20,7 +20,7 @@ namespace memgraph::storage::durability {
// The current version of snapshot and WAL encoding / decoding.
// IMPORTANT: Please bump this version for every snapshot and/or WAL format
// change!!!
const uint64_t kVersion{14};
const uint64_t kVersion{15};
const uint64_t kOldestSupportedVersion{14};
const uint64_t kUniqueConstraintVersion{13};

View File

@@ -13,12 +13,14 @@
#include <algorithm>
#include <iterator>
#include <limits>
#include <thread>
#include "storage/v2/mvcc.hpp"
#include "storage/v2/property_value.hpp"
#include "utils/bound.hpp"
#include "utils/logging.hpp"
#include "utils/memory_tracker.hpp"
#include "utils/synchronized.hpp"
namespace memgraph::storage {
@@ -263,6 +265,95 @@ bool CurrentVersionHasLabelProperty(const Vertex &vertex, LabelId label, Propert
return !deleted && has_label && current_value_equal_to_value;
}
template <typename TIndexAccessor>
void TryInsertLabelIndex(Vertex &vertex, LabelId label, TIndexAccessor &index_accessor) {
if (vertex.deleted || !utils::Contains(vertex.labels, label)) {
return;
}
index_accessor.insert({&vertex, 0});
}
template <typename TIndexAccessor>
void TryInsertLabelPropertyIndex(Vertex &vertex, std::pair<LabelId, PropertyId> label_property_pair,
TIndexAccessor &index_accessor) {
if (vertex.deleted || !utils::Contains(vertex.labels, label_property_pair.first)) {
return;
}
auto value = vertex.properties.GetProperty(label_property_pair.second);
if (value.IsNull()) {
return;
}
index_accessor.insert({std::move(value), &vertex, 0});
}
template <typename TSkiplistIter, typename TIndex, typename TIndexKey, typename TFunc>
void CreateIndexOnSingleThread(utils::SkipList<Vertex>::Accessor &vertices, TSkiplistIter it, TIndex &index,
TIndexKey key, const TFunc &func) {
utils::MemoryTracker::OutOfMemoryExceptionEnabler oom_exception;
try {
auto acc = it->second.access();
for (Vertex &vertex : vertices) {
func(vertex, key, acc);
}
} catch (const utils::OutOfMemoryException &) {
utils::MemoryTracker::OutOfMemoryExceptionBlocker oom_exception_blocker;
index.erase(it);
throw;
}
}
template <typename TIndex, typename TIndexKey, typename TSKiplistIter, typename TFunc>
void CreateIndexOnMultipleThreads(utils::SkipList<Vertex>::Accessor &vertices, TSKiplistIter skiplist_iter,
TIndex &index, TIndexKey key, const ParalellizedIndexCreationInfo &paralell_exec_info,
const TFunc &func) {
utils::MemoryTracker::OutOfMemoryExceptionEnabler oom_exception;
const auto &vertex_batches = paralell_exec_info.first;
const auto thread_count = std::min(paralell_exec_info.second, vertex_batches.size());
MG_ASSERT(!vertex_batches.empty(),
"The size of batches should always be greater than zero if you want to use the parallel version of index "
"creation!");
std::atomic<uint64_t> batch_counter = 0;
utils::Synchronized<std::optional<utils::OutOfMemoryException>, utils::SpinLock> maybe_error{};
{
std::vector<std::jthread> threads;
threads.reserve(thread_count);
for (auto i{0U}; i < thread_count; ++i) {
threads.emplace_back(
[&skiplist_iter, &func, &index, &vertex_batches, &maybe_error, &batch_counter, &key, &vertices]() {
while (!maybe_error.Lock()->has_value()) {
const auto batch_index = batch_counter++;
if (batch_index >= vertex_batches.size()) {
return;
}
const auto &batch = vertex_batches[batch_index];
auto index_accessor = index.at(key).access();
auto it = vertices.find(batch.first);
try {
for (auto i{0U}; i < batch.second; ++i, ++it) {
func(*it, key, index_accessor);
}
} catch (utils::OutOfMemoryException &failure) {
utils::MemoryTracker::OutOfMemoryExceptionBlocker oom_exception_blocker;
index.erase(skiplist_iter);
*maybe_error.Lock() = std::move(failure);
}
}
});
}
}
if (maybe_error.Lock()->has_value()) {
throw utils::OutOfMemoryException((*maybe_error.Lock())->what());
}
}
} // namespace
void LabelIndex::UpdateOnAddLabel(LabelId label, Vertex *vertex, const Transaction &tx) {
@@ -272,27 +363,43 @@ void LabelIndex::UpdateOnAddLabel(LabelId label, Vertex *vertex, const Transacti
acc.insert(Entry{vertex, tx.start_timestamp});
}
bool LabelIndex::CreateIndex(LabelId label, utils::SkipList<Vertex>::Accessor vertices) {
utils::MemoryTracker::OutOfMemoryExceptionEnabler oom_exception;
bool LabelIndex::CreateIndex(LabelId label, utils::SkipList<Vertex>::Accessor vertices,
const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info) {
auto create_index_seq = [this](LabelId label, utils::SkipList<Vertex>::Accessor &vertices,
std::map<LabelId, utils::SkipList<Entry>>::iterator it) {
using IndexAccessor = decltype(it->second.access());
CreateIndexOnSingleThread(vertices, it, index_, label,
[](Vertex &vertex, LabelId label, IndexAccessor &index_accessor) {
TryInsertLabelIndex(vertex, label, index_accessor);
});
return true;
};
auto create_index_par = [this](LabelId label, utils::SkipList<Vertex>::Accessor &vertices,
std::map<LabelId, utils::SkipList<Entry>>::iterator label_it,
const ParalellizedIndexCreationInfo &paralell_exec_info) {
using IndexAccessor = decltype(label_it->second.access());
CreateIndexOnMultipleThreads(vertices, label_it, index_, label, paralell_exec_info,
[](Vertex &vertex, LabelId label, IndexAccessor &index_accessor) {
TryInsertLabelIndex(vertex, label, index_accessor);
});
return true;
};
auto [it, emplaced] = index_.emplace(std::piecewise_construct, std::forward_as_tuple(label), std::forward_as_tuple());
if (!emplaced) {
// Index already exists.
return false;
}
try {
auto acc = it->second.access();
for (Vertex &vertex : vertices) {
if (vertex.deleted || !utils::Contains(vertex.labels, label)) {
continue;
}
acc.insert(Entry{&vertex, 0});
}
} catch (const utils::OutOfMemoryException &) {
utils::MemoryTracker::OutOfMemoryExceptionBlocker oom_exception_blocker;
index_.erase(it);
throw;
if (paralell_exec_info) {
return create_index_par(label, vertices, it, *paralell_exec_info);
}
return true;
return create_index_seq(label, vertices, it);
}
std::vector<LabelId> LabelIndex::ListIndices() const {
@@ -418,32 +525,46 @@ void LabelPropertyIndex::UpdateOnSetProperty(PropertyId property, const Property
}
}
bool LabelPropertyIndex::CreateIndex(LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor vertices) {
utils::MemoryTracker::OutOfMemoryExceptionEnabler oom_exception;
bool LabelPropertyIndex::CreateIndex(LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor vertices,
const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info) {
auto create_index_seq = [this](LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor &vertices,
std::map<std::pair<LabelId, PropertyId>, utils::SkipList<Entry>>::iterator it) {
using IndexAccessor = decltype(it->second.access());
CreateIndexOnSingleThread(vertices, it, index_, std::make_pair(label, property),
[](Vertex &vertex, std::pair<LabelId, PropertyId> key, IndexAccessor &index_accessor) {
TryInsertLabelPropertyIndex(vertex, key, index_accessor);
});
return true;
};
auto create_index_par =
[this](LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor &vertices,
std::map<std::pair<LabelId, PropertyId>, utils::SkipList<Entry>>::iterator label_property_it,
const ParalellizedIndexCreationInfo &paralell_exec_info) {
using IndexAccessor = decltype(label_property_it->second.access());
CreateIndexOnMultipleThreads(
vertices, label_property_it, index_, std::make_pair(label, property), paralell_exec_info,
[](Vertex &vertex, std::pair<LabelId, PropertyId> key, IndexAccessor &index_accessor) {
TryInsertLabelPropertyIndex(vertex, key, index_accessor);
});
return true;
};
auto [it, emplaced] =
index_.emplace(std::piecewise_construct, std::forward_as_tuple(label, property), std::forward_as_tuple());
if (!emplaced) {
// Index already exists.
return false;
}
try {
auto acc = it->second.access();
for (Vertex &vertex : vertices) {
if (vertex.deleted || !utils::Contains(vertex.labels, label)) {
continue;
}
auto value = vertex.properties.GetProperty(property);
if (value.IsNull()) {
continue;
}
acc.insert(Entry{std::move(value), &vertex, 0});
}
} catch (const utils::OutOfMemoryException &) {
utils::MemoryTracker::OutOfMemoryExceptionBlocker oom_exception_blocker;
index_.erase(it);
throw;
if (paralell_exec_info) {
return create_index_par(label, property, vertices, it, *paralell_exec_info);
}
return true;
return create_index_seq(label, property, vertices, it);
}
std::vector<std::pair<LabelId, PropertyId>> LabelPropertyIndex::ListIndices() const {

View File

@@ -28,6 +28,9 @@ namespace memgraph::storage {
struct Indices;
struct Constraints;
using ParalellizedIndexCreationInfo =
std::pair<std::vector<std::pair<Gid, uint64_t>> /*vertex_recovery_info*/, uint64_t /*thread_count*/>;
class LabelIndex {
private:
struct Entry {
@@ -58,7 +61,8 @@ class LabelIndex {
void UpdateOnAddLabel(LabelId label, Vertex *vertex, const Transaction &tx);
/// @throw std::bad_alloc
bool CreateIndex(LabelId label, utils::SkipList<Vertex>::Accessor vertices);
bool CreateIndex(LabelId label, utils::SkipList<Vertex>::Accessor vertices,
const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info = std::nullopt);
/// Returns false if there was no index to drop
bool DropIndex(LabelId label) { return index_.erase(label) > 0; }
@@ -160,7 +164,8 @@ class LabelPropertyIndex {
void UpdateOnSetProperty(PropertyId property, const PropertyValue &value, Vertex *vertex, const Transaction &tx);
/// @throw std::bad_alloc
bool CreateIndex(LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor vertices);
bool CreateIndex(LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor vertices,
const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info = std::nullopt);
bool DropIndex(LabelId label, PropertyId property) { return index_.erase({label, property}) > 0; }

View File

@@ -0,0 +1,34 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "isolation_level.hpp"
namespace memgraph::storage {
std::string_view IsolationLevelToString(IsolationLevel isolation_level) {
switch (isolation_level) {
case IsolationLevel::READ_COMMITTED:
return "READ_COMMITTED";
case IsolationLevel::READ_UNCOMMITTED:
return "READ_UNCOMMITTED";
case IsolationLevel::SNAPSHOT_ISOLATION:
return "SNAPSHOT_ISOLATION";
}
}
std::string_view IsolationLevelToString(std::optional<IsolationLevel> isolation_level) {
if (isolation_level) {
return IsolationLevelToString(*isolation_level);
}
return "";
}
} // namespace memgraph::storage

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -12,9 +12,14 @@
#pragma once
#include <cstdint>
#include <optional>
#include <string_view>
namespace memgraph::storage {
enum class IsolationLevel : std::uint8_t { SNAPSHOT_ISOLATION, READ_COMMITTED, READ_UNCOMMITTED };
std::string_view IsolationLevelToString(IsolationLevel isolation_level);
std::string_view IsolationLevelToString(std::optional<IsolationLevel> isolation_level);
} // namespace memgraph::storage

View File

@@ -1144,7 +1144,8 @@ bool PropertyStore::SetProperty(PropertyId property, const PropertyValue &value)
return !existed;
}
bool PropertyStore::InitProperties(const std::map<storage::PropertyId, storage::PropertyValue> &properties) {
template <typename TContainer>
bool PropertyStore::DoInitProperties(const TContainer &properties) {
uint64_t size = 0;
uint8_t *data = nullptr;
std::tie(size, data) = GetSizeData(buffer_);
@@ -1201,6 +1202,20 @@ bool PropertyStore::InitProperties(const std::map<storage::PropertyId, storage::
return true;
}
template bool PropertyStore::DoInitProperties<std::map<PropertyId, PropertyValue>>(
const std::map<PropertyId, PropertyValue> &);
template bool PropertyStore::DoInitProperties<std::vector<std::pair<PropertyId, PropertyValue>>>(
const std::vector<std::pair<PropertyId, PropertyValue>> &);
bool PropertyStore::InitProperties(const std::map<storage::PropertyId, storage::PropertyValue> &properties) {
return DoInitProperties(properties);
}
bool PropertyStore::InitProperties(std::vector<std::pair<storage::PropertyId, storage::PropertyValue>> properties) {
std::sort(properties.begin(), properties.end());
return DoInitProperties(properties);
}
bool PropertyStore::ClearProperties() {
bool in_local_buffer = false;

View File

@@ -60,11 +60,17 @@ class PropertyStore {
bool SetProperty(PropertyId property, const PropertyValue &value);
/// Init property values and return `true` if insertion took place. `false` is
/// returned if there exists property in property store and insertion couldn't take place. The time complexity of this
/// function is O(n).
/// returned if there is any existing property in property store and insertion couldn't take place. The time
/// complexity of this function is O(n).
/// @throw std::bad_alloc
bool InitProperties(const std::map<storage::PropertyId, storage::PropertyValue> &properties);
/// Init property values and return `true` if insertion took place. `false` is
/// returned if there is any existing property in property store and insertion couldn't take place. The time
/// complexity of this function is O(n*log(n)):
/// @throw std::bad_alloc
bool InitProperties(std::vector<std::pair<storage::PropertyId, storage::PropertyValue>> properties);
/// Remove all properties and return `true` if any removal took place.
/// `false` is returned if there were no properties to remove. The time
/// complexity of this function is O(1).
@@ -72,6 +78,9 @@ class PropertyStore {
bool ClearProperties();
private:
template <typename TContainer>
bool DoInitProperties(const TContainer &properties);
uint8_t buffer_[sizeof(uint64_t) + sizeof(uint8_t *)];
};

View File

@@ -399,7 +399,8 @@ std::vector<Storage::ReplicationClient::RecoveryStep> Storage::ReplicationClient
// we cannot know if the difference is only in the current WAL or we need
// to send the snapshot.
if (latest_snapshot) {
locker_acc.AddPath(latest_snapshot->path);
const auto lock_success = locker_acc.AddPath(latest_snapshot->path);
MG_ASSERT(!lock_success.HasError(), "Tried to lock a nonexistant path.");
recovery_steps.emplace_back(std::in_place_type_t<RecoverySnapshot>{}, std::move(latest_snapshot->path));
}
// if there are no finalized WAL files, snapshot left the current WAL
@@ -446,7 +447,8 @@ std::vector<Storage::ReplicationClient::RecoveryStep> Storage::ReplicationClient
// We need to lock these files and add them to the chain
for (auto result_wal_it = wal_files->begin() + distance_from_first; result_wal_it != wal_files->end();
++result_wal_it) {
locker_acc.AddPath(result_wal_it->path);
const auto lock_success = locker_acc.AddPath(result_wal_it->path);
MG_ASSERT(!lock_success.HasError(), "Tried to lock a nonexistant path.");
wal_chain.push_back(std::move(result_wal_it->path));
}
@@ -464,7 +466,8 @@ std::vector<Storage::ReplicationClient::RecoveryStep> Storage::ReplicationClient
MG_ASSERT(latest_snapshot, "Invalid durability state, missing snapshot");
// We didn't manage to find a WAL chain, we need to send the latest snapshot
// with its WALs
locker_acc.AddPath(latest_snapshot->path);
const auto lock_success = locker_acc.AddPath(latest_snapshot->path);
MG_ASSERT(!lock_success.HasError(), "Tried to lock a nonexistant path.");
recovery_steps.emplace_back(std::in_place_type_t<RecoverySnapshot>{}, std::move(latest_snapshot->path));
std::vector<std::filesystem::path> recovery_wal_files;
@@ -483,13 +486,15 @@ std::vector<Storage::ReplicationClient::RecoveryStep> Storage::ReplicationClient
}
for (; wal_it != wal_files->end(); ++wal_it) {
locker_acc.AddPath(wal_it->path);
const auto lock_success = locker_acc.AddPath(wal_it->path);
MG_ASSERT(!lock_success.HasError(), "Tried to lock a nonexistant path.");
recovery_wal_files.push_back(std::move(wal_it->path));
}
// We only have a WAL before the snapshot
if (recovery_wal_files.empty()) {
locker_acc.AddPath(wal_files->back().path);
const auto lock_success = locker_acc.AddPath(wal_files->back().path);
MG_ASSERT(!lock_success.HasError(), "Tried to lock a nonexistant path.");
recovery_wal_files.push_back(std::move(wal_files->back().path));
}

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -173,7 +173,7 @@ void Storage::ReplicationServer::SnapshotHandler(slk::Reader *req_reader, slk::B
spdlog::debug("Loading snapshot");
auto recovered_snapshot = durability::LoadSnapshot(*maybe_snapshot_path, &storage_->vertices_, &storage_->edges_,
&storage_->epoch_history_, &storage_->name_id_mapper_,
&storage_->edge_count_, storage_->config_.items);
&storage_->edge_count_, storage_->config_);
spdlog::debug("Snapshot loaded successfully");
// If this step is present it should always be the first step of
// the recovery so we use the UUID we read from snasphost

View File

@@ -34,6 +34,8 @@
#include "storage/v2/storage_mode.hpp"
#include "storage/v2/transaction.hpp"
#include "storage/v2/vertex_accessor.hpp"
#include "utils/event_counter.hpp"
#include "utils/event_histogram.hpp"
#include "utils/file.hpp"
#include "utils/logging.hpp"
#include "utils/memory_tracker.hpp"
@@ -41,6 +43,7 @@
#include "utils/rw_lock.hpp"
#include "utils/spin_lock.hpp"
#include "utils/stat.hpp"
#include "utils/timer.hpp"
#include "utils/uuid.hpp"
/// REPLICATION ///
@@ -49,6 +52,13 @@
#include "storage/v2/replication/rpc.hpp"
#include "storage/v2/storage_error.hpp"
namespace memgraph::metrics {
extern const Event SnapshotCreationLatency_us;
extern const Event ActiveLabelIndices;
extern const Event ActiveLabelPropertyIndices;
} // namespace memgraph::metrics
namespace memgraph::storage {
using OOMExceptionEnabler = utils::MemoryTracker::OutOfMemoryExceptionEnabler;
@@ -360,7 +370,7 @@ Storage::Storage(Config config)
if (config_.durability.recover_on_startup) {
auto info = durability::RecoverData(snapshot_directory_, wal_directory_, &uuid_, &epoch_id_, &epoch_history_,
&vertices_, &edges_, &edge_count_, &name_id_mapper_, &indices_, &constraints_,
config_.items, &wal_seq_num_);
config_, &wal_seq_num_);
if (info) {
vertex_id_ = info->next_vertex_id;
edge_id_ = info->next_edge_id;
@@ -1213,6 +1223,9 @@ utils::BasicResult<StorageIndexDefinitionError, void> Storage::CreateIndex(
commit_log_->MarkFinished(commit_timestamp);
last_commit_timestamp_ = commit_timestamp;
// We don't care if there is a replication error because on main node the change will go through
memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveLabelIndices);
if (success) {
return {};
}
@@ -1232,6 +1245,9 @@ utils::BasicResult<StorageIndexDefinitionError, void> Storage::CreateIndex(
commit_log_->MarkFinished(commit_timestamp);
last_commit_timestamp_ = commit_timestamp;
// We don't care if there is a replication error because on main node the change will go through
memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveLabelPropertyIndices);
if (success) {
return {};
}
@@ -1251,6 +1267,9 @@ utils::BasicResult<StorageIndexDefinitionError, void> Storage::DropIndex(
commit_log_->MarkFinished(commit_timestamp);
last_commit_timestamp_ = commit_timestamp;
// We don't care if there is a replication error because on main node the change will go through
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveLabelIndices);
if (success) {
return {};
}
@@ -1272,6 +1291,9 @@ utils::BasicResult<StorageIndexDefinitionError, void> Storage::DropIndex(
commit_log_->MarkFinished(commit_timestamp);
last_commit_timestamp_ = commit_timestamp;
// We don't care if there is a replication error because on main node the change will go through
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveLabelPropertyIndices);
if (success) {
return {};
}
@@ -1943,14 +1965,18 @@ utils::BasicResult<Storage::CreateSnapshotError> Storage::CreateSnapshot(std::op
}
auto snapshot_creator = [this]() {
utils::Timer timer;
auto transaction = CreateTransaction(IsolationLevel::SNAPSHOT_ISOLATION, storage_mode_);
// Create snapshot.
durability::CreateSnapshot(&transaction, snapshot_directory_, wal_directory_,
config_.durability.snapshot_retention_count, &vertices_, &edges_, &name_id_mapper_,
&indices_, &constraints_, config_.items, uuid_, epoch_id_, epoch_history_,
&file_retainer_);
&indices_, &constraints_, config_, uuid_, epoch_id_, epoch_history_, &file_retainer_);
// Finalize snapshot transaction.
commit_log_->MarkFinished(transaction.start_timestamp);
memgraph::metrics::Measure(memgraph::metrics::SnapshotCreationLatency_us,
std::chrono::duration_cast<std::chrono::microseconds>(timer.Elapsed()).count());
};
std::lock_guard snapshot_guard(snapshot_lock_);
@@ -1981,16 +2007,23 @@ utils::BasicResult<Storage::CreateSnapshotError> Storage::CreateSnapshot(std::op
return CreateSnapshotError::ReachedMaxNumTries;
}
bool Storage::LockPath() {
utils::FileRetainer::FileLockerAccessor::ret_type Storage::IsPathLocked() {
auto locker_accessor = global_locker_.Access();
return locker_accessor.IsPathLocked(config_.durability.storage_directory);
}
utils::FileRetainer::FileLockerAccessor::ret_type Storage::LockPath() {
auto locker_accessor = global_locker_.Access();
return locker_accessor.AddPath(config_.durability.storage_directory);
}
bool Storage::UnlockPath() {
utils::FileRetainer::FileLockerAccessor::ret_type Storage::UnlockPath() {
{
auto locker_accessor = global_locker_.Access();
if (!locker_accessor.RemovePath(config_.durability.storage_directory)) {
return false;
const auto ret = locker_accessor.RemovePath(config_.durability.storage_directory);
if (ret.HasError() || !ret.GetValue()) {
// Exit without cleaning the queue
return ret;
}
}
@@ -2174,6 +2207,8 @@ utils::BasicResult<Storage::SetIsolationLevelError> Storage::SetIsolationLevel(I
return {};
}
IsolationLevel Storage::GetIsolationLevel() const noexcept { return isolation_level_; }
void Storage::SetStorageMode(StorageMode storage_mode) {
std::unique_lock main_guard{main_lock_};
storage_mode_ = storage_mode;

View File

@@ -39,6 +39,7 @@
#include "storage/v2/vertex_accessor.hpp"
#include "utils/file_locker.hpp"
#include "utils/on_scope_exit.hpp"
#include "utils/result.hpp"
#include "utils/rw_lock.hpp"
#include "utils/scheduler.hpp"
#include "utils/skip_list.hpp"
@@ -467,8 +468,9 @@ class Storage final {
StorageInfo GetInfo() const;
bool LockPath();
bool UnlockPath();
utils::FileRetainer::FileLockerAccessor::ret_type IsPathLocked();
utils::FileRetainer::FileLockerAccessor::ret_type LockPath();
utils::FileRetainer::FileLockerAccessor::ret_type UnlockPath();
bool SetReplicaRole(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config = {});
@@ -513,6 +515,7 @@ class Storage final {
enum class SetIsolationLevelError : uint8_t { DisabledForAnalyticalMode };
utils::BasicResult<SetIsolationLevelError> SetIsolationLevel(IsolationLevel isolation_level);
IsolationLevel GetIsolationLevel() const noexcept;
void SetStorageMode(StorageMode storage_mode);

View File

@@ -0,0 +1,25 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "storage_mode.hpp"
namespace memgraph::storage {
std::string_view StorageModeToString(memgraph::storage::StorageMode storage_mode) {
switch (storage_mode) {
case memgraph::storage::StorageMode::IN_MEMORY_ANALYTICAL:
return "IN_MEMORY_ANALYTICAL";
case memgraph::storage::StorageMode::IN_MEMORY_TRANSACTIONAL:
return "IN_MEMORY_TRANSACTIONAL";
}
}
} // namespace memgraph::storage

View File

@@ -1,9 +1,23 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <cstdint>
#include <string_view>
namespace memgraph::storage {
enum class StorageMode : std::uint8_t { IN_MEMORY_ANALYTICAL, IN_MEMORY_TRANSACTIONAL };
std::string_view StorageModeToString(memgraph::storage::StorageMode storage_mode);
} // namespace memgraph::storage

View File

@@ -2,6 +2,8 @@ set(utils_src_files
async_timer.cpp
base64.cpp
event_counter.cpp
event_gauge.cpp
event_histogram.cpp
csv_parsing.cpp
file.cpp
file_locker.cpp
@@ -15,7 +17,8 @@ set(utils_src_files
thread_pool.cpp
tsc.cpp
system_info.cpp
uuid.cpp)
uuid.cpp
build_info.cpp)
find_package(Boost REQUIRED)
find_package(fmt REQUIRED)

26
src/utils/build_info.cpp Normal file
View File

@@ -0,0 +1,26 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "build_info.hpp"
namespace memgraph::utils {
BuildInfo GetBuildInfo() {
#ifdef CMAKE_BUILD_TYPE_NAME
constexpr const char *build_info_name = CMAKE_BUILD_TYPE_NAME;
#else
constexpr const char *build_info_name = "unkown";
#endif
BuildInfo info{build_info_name};
return info;
}
} // namespace memgraph::utils

24
src/utils/build_info.hpp Normal file
View File

@@ -0,0 +1,24 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <string>
namespace memgraph::utils {
struct BuildInfo {
std::string build_name;
};
BuildInfo GetBuildInfo();
} // namespace memgraph::utils

View File

@@ -11,69 +11,86 @@
#include "utils/event_counter.hpp"
#define APPLY_FOR_EVENTS(M) \
M(ReadQuery, "Number of read-only queries executed.") \
M(WriteQuery, "Number of write-only queries executed.") \
M(ReadWriteQuery, "Number of read-write queries executed.") \
\
M(OnceOperator, "Number of times Once operator was used.") \
M(CreateNodeOperator, "Number of times CreateNode operator was used.") \
M(CreateExpandOperator, "Number of times CreateExpand operator was used.") \
M(ScanAllOperator, "Number of times ScanAll operator was used.") \
M(ScanAllByLabelOperator, "Number of times ScanAllByLabel operator was used.") \
M(ScanAllByLabelPropertyRangeOperator, "Number of times ScanAllByLabelPropertyRange operator was used.") \
M(ScanAllByLabelPropertyValueOperator, "Number of times ScanAllByLabelPropertyValue operator was used.") \
M(ScanAllByLabelPropertyOperator, "Number of times ScanAllByLabelProperty operator was used.") \
M(ScanAllByIdOperator, "Number of times ScanAllById operator was used.") \
M(ExpandOperator, "Number of times Expand operator was used.") \
M(ExpandVariableOperator, "Number of times ExpandVariable operator was used.") \
M(ConstructNamedPathOperator, "Number of times ConstructNamedPath operator was used.") \
M(FilterOperator, "Number of times Filter operator was used.") \
M(ProduceOperator, "Number of times Produce operator was used.") \
M(DeleteOperator, "Number of times Delete operator was used.") \
M(SetPropertyOperator, "Number of times SetProperty operator was used.") \
M(SetPropertiesOperator, "Number of times SetProperties operator was used.") \
M(SetLabelsOperator, "Number of times SetLabels operator was used.") \
M(RemovePropertyOperator, "Number of times RemoveProperty operator was used.") \
M(RemoveLabelsOperator, "Number of times RemoveLabels operator was used.") \
M(EdgeUniquenessFilterOperator, "Number of times EdgeUniquenessFilter operator was used.") \
M(EmptyResultOperator, "Number of times EmptyResult operator was used.") \
M(AccumulateOperator, "Number of times Accumulate operator was used.") \
M(AggregateOperator, "Number of times Aggregate operator was used.") \
M(SkipOperator, "Number of times Skip operator was used.") \
M(LimitOperator, "Number of times Limit operator was used.") \
M(OrderByOperator, "Number of times OrderBy operator was used.") \
M(MergeOperator, "Number of times Merge operator was used.") \
M(OptionalOperator, "Number of times Optional operator was used.") \
M(UnwindOperator, "Number of times Unwind operator was used.") \
M(DistinctOperator, "Number of times Distinct operator was used.") \
M(UnionOperator, "Number of times Union operator was used.") \
M(CartesianOperator, "Number of times Cartesian operator was used.") \
M(CallProcedureOperator, "Number of times CallProcedure operator was used.") \
M(ForeachOperator, "Number of times Foreach operator was used.") \
M(EvaluatePatternFilterOperator, "Number of times EvaluatePatternFilter operator was used.") \
M(ApplyOperator, "Number of times ApplyOperator operator was used.") \
\
M(FailedQuery, "Number of times executing a query failed.") \
M(LabelIndexCreated, "Number of times a label index was created.") \
M(LabelPropertyIndexCreated, "Number of times a label property index was created.") \
M(StreamsCreated, "Number of Streams created.") \
M(MessagesConsumed, "Number of consumed streamed messages.") \
M(TriggersCreated, "Number of Triggers created.") \
M(TriggersExecuted, "Number of Triggers executed.")
namespace EventCounter {
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define APPLY_FOR_COUNTERS(M) \
M(ReadQuery, QueryType, "Number of read-only queries executed.") \
M(WriteQuery, QueryType, "Number of write-only queries executed.") \
M(ReadWriteQuery, QueryType, "Number of read-write queries executed.") \
\
M(OnceOperator, Operator, "Number of times Once operator was used.") \
M(CreateNodeOperator, Operator, "Number of times CreateNode operator was used.") \
M(CreateExpandOperator, Operator, "Number of times CreateExpand operator was used.") \
M(ScanAllOperator, Operator, "Number of times ScanAll operator was used.") \
M(ScanAllByLabelOperator, Operator, "Number of times ScanAllByLabel operator was used.") \
M(ScanAllByLabelPropertyRangeOperator, Operator, "Number of times ScanAllByLabelPropertyRange operator was used.") \
M(ScanAllByLabelPropertyValueOperator, Operator, "Number of times ScanAllByLabelPropertyValue operator was used.") \
M(ScanAllByLabelPropertyOperator, Operator, "Number of times ScanAllByLabelProperty operator was used.") \
M(ScanAllByIdOperator, Operator, "Number of times ScanAllById operator was used.") \
M(ExpandOperator, Operator, "Number of times Expand operator was used.") \
M(ExpandVariableOperator, Operator, "Number of times ExpandVariable operator was used.") \
M(ConstructNamedPathOperator, Operator, "Number of times ConstructNamedPath operator was used.") \
M(FilterOperator, Operator, "Number of times Filter operator was used.") \
M(ProduceOperator, Operator, "Number of times Produce operator was used.") \
M(DeleteOperator, Operator, "Number of times Delete operator was used.") \
M(SetPropertyOperator, Operator, "Number of times SetProperty operator was used.") \
M(SetPropertiesOperator, Operator, "Number of times SetProperties operator was used.") \
M(SetLabelsOperator, Operator, "Number of times SetLabels operator was used.") \
M(RemovePropertyOperator, Operator, "Number of times RemoveProperty operator was used.") \
M(RemoveLabelsOperator, Operator, "Number of times RemoveLabels operator was used.") \
M(EdgeUniquenessFilterOperator, Operator, "Number of times EdgeUniquenessFilter operator was used.") \
M(EmptyResultOperator, Operator, "Number of times EmptyResult operator was used.") \
M(AccumulateOperator, Operator, "Number of times Accumulate operator was used.") \
M(AggregateOperator, Operator, "Number of times Aggregate operator was used.") \
M(SkipOperator, Operator, "Number of times Skip operator was used.") \
M(LimitOperator, Operator, "Number of times Limit operator was used.") \
M(OrderByOperator, Operator, "Number of times OrderBy operator was used.") \
M(MergeOperator, Operator, "Number of times Merge operator was used.") \
M(OptionalOperator, Operator, "Number of times Optional operator was used.") \
M(UnwindOperator, Operator, "Number of times Unwind operator was used.") \
M(DistinctOperator, Operator, "Number of times Distinct operator was used.") \
M(UnionOperator, Operator, "Number of times Union operator was used.") \
M(CartesianOperator, Operator, "Number of times Cartesian operator was used.") \
M(CallProcedureOperator, Operator, "Number of times CallProcedure operator was used.") \
M(ForeachOperator, Operator, "Number of times Foreach operator was used.") \
M(EvaluatePatternFilterOperator, Operator, "Number of times EvaluatePatternFilter operator was used.") \
M(ApplyOperator, Operator, "Number of times ApplyOperator operator was used.") \
\
M(ActiveLabelIndices, Index, "Number of active label indices in the system.") \
M(ActiveLabelPropertyIndices, Index, "Number of active label property indices in the system<.") \
\
M(StreamsCreated, Stream, "Number of Streams created.") \
M(MessagesConsumed, Stream, "Number of consumed streamed messages.") \
\
M(TriggersCreated, Trigger, "Number of Triggers created.") \
M(TriggersExecuted, Trigger, "Number of Triggers executed.") \
\
M(ActiveSessions, Session, "Number of active connections.") \
M(ActiveBoltSessions, Session, "Number of active Bolt connections.") \
M(ActiveTCPSessions, Session, "Number of active TCP connections.") \
M(ActiveSSLSessions, Session, "Number of active SSL connections.") \
M(ActiveWebSocketSessions, Session, "Number of active websocket connections.") \
M(BoltMessages, Session, "Number of Bolt messages sent.") \
\
M(ActiveTransactions, Transaction, "Number of active transactions.") \
M(CommitedTransactions, Transaction, "Number of committed transactions.") \
M(RollbackedTransactions, Transaction, "Number of rollbacked transactions.") \
M(FailedQuery, Transaction, "Number of times executing a query failed.")
namespace memgraph::metrics {
// define every Event as an index in the array of counters
#define M(NAME, DOCUMENTATION) extern const Event NAME = __COUNTER__;
APPLY_FOR_EVENTS(M)
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION) extern const Event NAME = __COUNTER__;
APPLY_FOR_COUNTERS(M)
#undef M
inline constexpr Event END = __COUNTER__;
// Initialize array for the global counter with all values set to 0
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
Counter global_counters_array[END]{};
// Initialize global counters
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
EventCounters global_counters(global_counters_array);
const Event EventCounters::num_counters = END;
@@ -82,28 +99,45 @@ void EventCounters::Increment(const Event event, Count amount) {
counters_[event].fetch_add(amount, std::memory_order_relaxed);
}
void EventCounters::Decrement(const Event event, Count amount) {
counters_[event].fetch_sub(amount, std::memory_order_relaxed);
}
void IncrementCounter(const Event event, Count amount) { global_counters.Increment(event, amount); }
void DecrementCounter(const Event event, Count amount) { global_counters.Decrement(event, amount); }
const char *GetName(const Event event) {
const char *GetCounterName(const Event event) {
static const char *strings[] = {
#define M(NAME, DOCUMENTATION) #NAME,
APPLY_FOR_EVENTS(M)
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION) #NAME,
APPLY_FOR_COUNTERS(M)
#undef M
};
return strings[event];
}
const char *GetDocumentation(const Event event) {
const char *GetCounterDocumentation(const Event event) {
static const char *strings[] = {
#define M(NAME, DOCUMENTATION) DOCUMENTATION,
APPLY_FOR_EVENTS(M)
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION) DOCUMENTATION,
APPLY_FOR_COUNTERS(M)
#undef M
};
return strings[event];
}
Event End() { return END; }
const char *GetCounterType(const Event event) {
static const char *strings[] = {
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION) #TYPE,
APPLY_FOR_COUNTERS(M)
#undef M
};
} // namespace EventCounter
return strings[event];
}
Event CounterEnd() { return END; }
} // namespace memgraph::metrics

View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -10,11 +10,12 @@
// licenses/APL.txt.
#pragma once
#include <atomic>
#include <cstdlib>
#include <memory>
namespace EventCounter {
namespace memgraph::metrics {
using Event = uint64_t;
using Count = uint64_t;
using Counter = std::atomic<Count>;
@@ -29,19 +30,23 @@ class EventCounters {
void Increment(Event event, Count amount = 1);
void Decrement(Event event, Count amount = 1);
static const Event num_counters;
private:
Counter *counters_;
};
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
extern EventCounters global_counters;
void IncrementCounter(Event event, Count amount = 1);
void DecrementCounter(Event event, Count amount = 1);
const char *GetName(Event event);
const char *GetDocumentation(Event event);
const char *GetCounterName(Event event);
const char *GetCounterDocumentation(Event event);
const char *GetCounterType(Event event);
Event End();
} // namespace EventCounter
Event CounterEnd();
} // namespace memgraph::metrics

76
src/utils/event_gauge.cpp Normal file
View File

@@ -0,0 +1,76 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "utils/event_gauge.hpp"
// We don't have any gauges for now
#define APPLY_FOR_GAUGES(M)
namespace memgraph::metrics {
// define every Event as an index in the array of gauges
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION) extern const Event NAME = __COUNTER__;
APPLY_FOR_GAUGES(M)
#undef M
inline constexpr Event END = __COUNTER__;
// Initialize array for the global gauges with all values set to 0
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
Gauge global_gauges_array[END]{};
// Initialize global counters
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
EventGauges global_gauges(global_gauges_array);
const Event EventGauges::num_gauges = END;
void EventGauges::SetValue(const Event event, Value value) { gauges_[event].store(value, std::memory_order_seq_cst); }
void SetGaugeValue(const Event event, Value value) { global_gauges.SetValue(event, value); }
const char *GetGaugeName(const Event event) {
static const char *strings[] = {
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION) #NAME,
APPLY_FOR_GAUGES(M)
#undef M
};
return strings[event];
}
const char *GetGaugeDocumentation(const Event event) {
static const char *strings[] = {
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION) DOCUMENTATION,
APPLY_FOR_GAUGES(M)
#undef M
};
return strings[event];
}
const char *GetGaugeType(const Event event) {
static const char *strings[] = {
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION) #TYPE,
APPLY_FOR_GAUGES(M)
#undef M
};
return strings[event];
}
Event GaugeEnd() { return END; }
} // namespace memgraph::metrics

49
src/utils/event_gauge.hpp Normal file
View File

@@ -0,0 +1,49 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <atomic>
#include <cstdlib>
#include <memory>
namespace memgraph::metrics {
using Event = uint64_t;
using Value = uint64_t;
using Gauge = std::atomic<Value>;
class EventGauges {
public:
explicit EventGauges(Gauge *allocated_gauges) noexcept : gauges_(allocated_gauges) {}
auto &operator[](const Event event) { return gauges_[event]; }
const auto &operator[](const Event event) const { return gauges_[event]; }
void SetValue(Event event, Value value);
static const Event num_gauges;
private:
Gauge *gauges_;
};
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
extern EventGauges global_gauges;
void SetGaugeValue(Event event, Value value);
const char *GetGaugeName(Event event);
const char *GetGaugeDocumentation(Event event);
const char *GetGaugeType(Event event);
Event GaugeEnd();
} // namespace memgraph::metrics

View File

@@ -0,0 +1,84 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "utils/event_histogram.hpp"
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define APPLY_FOR_HISTOGRAMS(M) \
M(QueryExecutionLatency_us, Query, "Query execution latency in microseconds", 50, 90, 99) \
M(SnapshotCreationLatency_us, Snapshot, "Snapshot creation latency in microseconds", 50, 90, 99) \
M(SnapshotRecoveryLatency_us, Snapshot, "Snapshot recovery latency in microseconds", 50, 90, 99)
namespace memgraph::metrics {
// define every Event as an index in the array of counters
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION, ...) extern const Event NAME = __COUNTER__;
APPLY_FOR_HISTOGRAMS(M)
#undef M
inline constexpr Event END = __COUNTER__;
// Initialize array for the global histogram with all named histograms and their percentiles
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
Histogram global_histograms_array[END]{
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION, ...) Histogram({__VA_ARGS__}),
APPLY_FOR_HISTOGRAMS(M)
#undef M
};
// Initialize global histograms
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
EventHistograms global_histograms(global_histograms_array);
const Event EventHistograms::num_histograms = END;
void Measure(const Event event, Value value) { global_histograms.Measure(event, value); }
void EventHistograms::Measure(const Event event, Value value) { histograms_[event].Measure(value); }
const char *GetHistogramName(const Event event) {
static const char *strings[] = {
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION, ...) #NAME,
APPLY_FOR_HISTOGRAMS(M)
#undef M
};
return strings[event];
}
const char *GetHistogramDocumentation(const Event event) {
static const char *strings[] = {
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION, ...) DOCUMENTATION,
APPLY_FOR_HISTOGRAMS(M)
#undef M
};
return strings[event];
}
const char *GetHistogramType(const Event event) {
static const char *strings[] = {
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define M(NAME, TYPE, DOCUMENTATION, ...) #TYPE,
APPLY_FOR_HISTOGRAMS(M)
#undef M
};
return strings[event];
}
Event HistogramEnd() { return END; }
} // namespace memgraph::metrics

View File

@@ -0,0 +1,171 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <cmath>
#include "utils/logging.hpp"
namespace memgraph::metrics {
using Event = uint64_t;
using Value = uint64_t;
using Measurement = std::atomic<uint64_t>;
// This is a logarithmically bucketing histogram optimized
// for collecting network response latency distributions.
// It "compresses" values by mapping them to a point on a
// logarithmic curve, which serves as the bucket index. This
// compression technique allows for very accurate histograms
// (unlike what is the case for sampling or lossy probabilistic
// approaches) with the trade-off that we sacrifice around 1%
// precision.
//
// properties:
// * roughly 1% precision loss - can be higher for values
// less than 100, so if measuring latency, generally do
// so in microseconds.
// * ~32kb constant space, single allocation per Histogram.
// * Histogram::Percentile() will return 0 if there were no
// samples measured yet.
class Histogram {
// This is the number of buckets that observed values
// will be logarithmically compressed into.
constexpr static auto kSampleLimit = 4096;
// This is roughly 1/error rate, where 100.0 is roughly
// a 1% error bound for measurements. This is less true
// for tiny measurements, but because we tend to measure
// microseconds, it is usually over 100, which is where
// the error bound starts to stabilize a bit. This has
// been tuned to allow the maximum uint64_t to compress
// within 4096 samples while still achieving a high accuracy.
constexpr static auto kPrecision = 92.0;
// samples_ stores per-bucket counts for measurements
// that have been mapped to a specific uint64_t in
// the "compression" logic below.
std::vector<uint64_t> samples_ = {};
std::vector<uint8_t> percentiles_;
// count_ is the number of measurements that have been
// included in this Histogram.
Measurement count_ = 0;
// sum_ is the summed value of all measurements that
// have been included in this Histogram.
Measurement sum_ = 0;
std::mutex samples_mutex_;
public:
Histogram() {
samples_.resize(kSampleLimit, 0);
percentiles_ = {0, 25, 50, 75, 90, 100};
}
explicit Histogram(std::vector<uint8_t> percentiles) : percentiles_(percentiles) { samples_.resize(kSampleLimit, 0); }
uint64_t Count() const { return count_.load(std::memory_order_relaxed); }
uint64_t Sum() const { return sum_.load(std::memory_order_relaxed); }
std::vector<uint8_t> Percentiles() const { return percentiles_; }
void Measure(uint64_t value) {
// "compression" logic
double boosted = 1.0 + static_cast<double>(value);
double ln = std::log(boosted);
double compressed = (kPrecision * ln) + 0.5;
MG_ASSERT(compressed < kSampleLimit, "compressing value {} to {} is invalid", value, compressed);
auto sample_index = static_cast<uint16_t>(compressed);
count_.fetch_add(1, std::memory_order_relaxed);
sum_.fetch_add(value, std::memory_order_relaxed);
{
std::lock_guard<std::mutex> lock(samples_mutex_);
samples_[sample_index]++;
}
}
std::vector<std::pair<uint64_t, uint64_t>> YieldPercentiles() const {
std::vector<std::pair<uint64_t, uint64_t>> percentile_yield;
percentile_yield.reserve(percentiles_.size());
for (const auto percentile : percentiles_) {
percentile_yield.emplace_back(std::make_pair(percentile, Percentile(percentile)));
}
return percentile_yield;
}
uint64_t Percentile(double percentile) const {
MG_ASSERT(percentile <= 100.0, "percentiles must not exceed 100.0");
MG_ASSERT(percentile >= 0.0, "percentiles must be greater than or equal to 0.0");
auto count = Count();
if (count == 0) {
return 0;
}
const auto floated_count = static_cast<double>(count);
const auto target = std::max(floated_count * percentile / 100.0, 1.0);
auto scanned = 0.0;
for (int i = 0; i < kSampleLimit; i++) {
const auto samples_at_index = samples_[i];
scanned += static_cast<double>(samples_at_index);
if (scanned >= target) {
// "decompression" logic
auto floated = static_cast<double>(i);
auto unboosted = floated / kPrecision;
auto decompressed = std::exp(unboosted) - 1.0;
return static_cast<uint64_t>(decompressed);
}
}
LOG_FATAL("bug in Histogram::Percentile where it failed to return the {} percentile", percentile);
return 0;
}
};
class EventHistograms {
public:
explicit EventHistograms(Histogram *allocated_histograms) noexcept : histograms_(allocated_histograms) {}
auto &operator[](const Event event) { return histograms_[event]; }
const auto &operator[](const Event event) const { return histograms_[event]; }
void Measure(Event event, Value value);
static const Event num_histograms;
private:
Histogram *histograms_;
};
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
extern EventHistograms global_histograms;
void Measure(Event event, Value value);
const char *GetHistogramName(Event event);
const char *GetHistogramDocumentation(Event event);
const char *GetHistogramType(Event event);
Event HistogramEnd();
} // namespace memgraph::metrics

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -80,13 +80,14 @@ void FileRetainer::CleanQueue() {
}
////// LockerEntry //////
void FileRetainer::LockerEntry::LockPath(const std::filesystem::path &path) {
bool FileRetainer::LockerEntry::LockPath(const std::filesystem::path &path) {
auto absolute_path = std::filesystem::absolute(path);
if (std::filesystem::is_directory(absolute_path)) {
directories_.emplace(std::move(absolute_path));
return;
const auto [itr, success] = directories_.emplace(std::move(absolute_path));
return success;
}
files_.emplace(std::move(absolute_path));
const auto [itr, success] = files_.emplace(std::move(absolute_path));
return success;
}
bool FileRetainer::LockerEntry::RemovePath(const std::filesystem::path &path) {
@@ -140,13 +141,27 @@ FileRetainer::FileLockerAccessor::FileLockerAccessor(FileRetainer *retainer, siz
file_retainer_->active_accessors_.fetch_add(1);
}
bool FileRetainer::FileLockerAccessor::AddPath(const std::filesystem::path &path) {
if (!std::filesystem::exists(path)) return false;
file_retainer_->lockers_.WithLock([&](auto &lockers) { lockers[locker_id_].LockPath(path); });
return true;
FileRetainer::FileLockerAccessor::ret_type FileRetainer::FileLockerAccessor::IsPathLocked(
const std::filesystem::path &path) {
if (!std::filesystem::exists(path)) {
return Error::NonexistentPath;
}
return file_retainer_->FileLocked(std::filesystem::absolute(path));
}
bool FileRetainer::FileLockerAccessor::RemovePath(const std::filesystem::path &path) {
FileRetainer::FileLockerAccessor::ret_type FileRetainer::FileLockerAccessor::AddPath(
const std::filesystem::path &path) {
if (!std::filesystem::exists(path)) {
return Error::NonexistentPath;
}
return file_retainer_->lockers_.WithLock([&](auto &lockers) { return lockers[locker_id_].LockPath(path); });
}
FileRetainer::FileLockerAccessor::ret_type FileRetainer::FileLockerAccessor::RemovePath(
const std::filesystem::path &path) {
if (!std::filesystem::exists(path)) {
return Error::NonexistentPath;
}
return file_retainer_->lockers_.WithLock([&](auto &lockers) { return lockers[locker_id_].RemovePath(path); });
}

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -18,6 +18,7 @@
#include <unordered_map>
#include "utils/file.hpp"
#include "utils/result.hpp"
#include "utils/rw_lock.hpp"
#include "utils/spin_lock.hpp"
#include "utils/synchronized.hpp"
@@ -114,15 +115,26 @@ class FileRetainer {
struct FileLockerAccessor {
friend FileLocker;
enum class Error : uint8_t {
NonexistentPath = 0,
};
using ret_type = utils::BasicResult<FileRetainer::FileLockerAccessor::Error, bool>;
/**
* Checks if a single path is in the current locker.
*/
ret_type IsPathLocked(const std::filesystem::path &path);
/**
* Add a single path to the current locker.
*/
bool AddPath(const std::filesystem::path &path);
ret_type AddPath(const std::filesystem::path &path);
/**
* Remove a single path form the current locker.
*/
bool RemovePath(const std::filesystem::path &path);
ret_type RemovePath(const std::filesystem::path &path);
FileLockerAccessor(const FileLockerAccessor &) = delete;
FileLockerAccessor(FileLockerAccessor &&) = default;
@@ -182,7 +194,7 @@ class FileRetainer {
class LockerEntry {
public:
void LockPath(const std::filesystem::path &path);
bool LockPath(const std::filesystem::path &path);
bool RemovePath(const std::filesystem::path &path);
[[nodiscard]] bool LocksFile(const std::filesystem::path &path) const;

View File

@@ -0,0 +1,22 @@
#include <string>
#include "query/frontend/ast/ast.hpp"
#include "spdlog/spdlog.h"
namespace memgraph::utils {
// Get ID by which FrameChangeCollector struct can cache in_list.expression2_
inline std::optional<std::string> GetFrameChangeId(memgraph::query::InListOperator &in_list) {
if (in_list.expression2_->GetTypeInfo() == memgraph::query::ListLiteral::kType) {
std::stringstream ss;
ss << static_cast<const void *>(in_list.expression2_);
return ss.str();
}
if (in_list.expression2_->GetTypeInfo() == memgraph::query::Identifier::kType) {
auto *identifier = utils::Downcast<memgraph::query::Identifier>(in_list.expression2_);
return identifier->name_;
}
return {};
};
} // namespace memgraph::utils

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -187,14 +187,19 @@ void *Pool::Allocate() {
for (unsigned char i = 0U; i < blocks_per_chunk_; ++i) {
*(data + (i * block_size_)) = i + 1U;
}
Chunk chunk{data, 0, blocks_per_chunk_};
// Insert the big block in the sorted position.
auto it = std::lower_bound(chunks_.begin(), chunks_.end(), chunk,
[](const auto &a, const auto &b) { return a.data < b.data; });
try {
chunks_.push_back(Chunk{data, 0, blocks_per_chunk_});
it = chunks_.insert(it, chunk);
} catch (...) {
GetUpstreamResource()->Deallocate(data, data_size, alignment);
throw;
}
last_alloc_chunk_ = &chunks_.back();
last_dealloc_chunk_ = &chunks_.back();
last_alloc_chunk_ = &*it;
last_dealloc_chunk_ = &*it;
return allocate_block_from_chunk(last_alloc_chunk_);
}
@@ -223,18 +228,20 @@ void Pool::Deallocate(void *p) {
deallocate_block_from_chunk(last_dealloc_chunk_);
return;
}
// Find the chunk which served this allocation
for (auto &chunk : chunks_) {
if (is_in_chunk(chunk)) {
// Update last_alloc_chunk_ as well because it now has a free block.
// Additionally this corresponds with C++ pattern of allocations and
// deallocations being done in reverse order.
last_alloc_chunk_ = &chunk;
last_dealloc_chunk_ = &chunk;
deallocate_block_from_chunk(&chunk);
return;
}
}
Chunk chunk{reinterpret_cast<unsigned char *>(p) - blocks_per_chunk_ * block_size_, 0, 0};
auto it = std::lower_bound(chunks_.begin(), chunks_.end(), chunk,
[](const auto &a, const auto &b) { return a.data <= b.data; });
MG_ASSERT(it != chunks_.end(), "Failed deallocation in utils::Pool");
MG_ASSERT(is_in_chunk(*it), "Failed deallocation in utils::Pool");
// Update last_alloc_chunk_ as well because it now has a free block.
// Additionally this corresponds with C++ pattern of allocations and
// deallocations being done in reverse order.
last_alloc_chunk_ = &*it;
last_dealloc_chunk_ = &*it;
deallocate_block_from_chunk(last_dealloc_chunk_);
// TODO: We could release the Chunk to upstream memory
}

23
src/utils/pmr/deque.hpp Normal file
View File

@@ -0,0 +1,23 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <deque>
#include "utils/memory.hpp"
namespace memgraph::utils::pmr {
template <class T>
using deque = std::deque<T, utils::Allocator<T>>;
} // namespace memgraph::utils::pmr

View File

@@ -118,8 +118,10 @@ enum class TypeId : uint64_t {
AST_PRIMITIVE_LITERAL,
AST_LIST_LITERAL,
AST_MAP_LITERAL,
AST_MAP_PROJECTION_LITERAL,
AST_IDENTIFIER,
AST_PROPERTY_LOOKUP,
AST_ALL_PROPERTIES_LOOKUP,
AST_LABELS_TEST,
AST_FUNCTION,
AST_REDUCE,

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -60,6 +60,8 @@ BENCHMARK_TEMPLATE(MapLiteral, NewDeleteResource)->Range(512, 1U << 15U)->Unit(b
BENCHMARK_TEMPLATE(MapLiteral, MonotonicBufferResource)->Range(512, 1U << 15U)->Unit(benchmark::kMicrosecond);
// TODO ante benchmark template for MapProjectionLiteral
template <class TMemory>
// NOLINTNEXTLINE(google-runtime-references)
static void AdditionOperator(benchmark::State &state) {

View File

@@ -0,0 +1,12 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>netcoreapp2.1</TargetFramework>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Neo4j.Driver.Simple" Version="4.1.1" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,75 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using Neo4j.Driver;
public class Transactions {
public static void Main(string[] args) {
var driver =
GraphDatabase.Driver("bolt://localhost:7687", AuthTokens.None,
(builder) => builder.WithEncryptionLevel(EncryptionLevel.None));
ClearDatabase(driver);
// Explicit transaction query.
using (var session = driver.Session()) {
Console.WriteLine("Checking explicit transaction metadata...");
var txMetadata = new Dictionary<string, object> {
{ "ver", "transaction" }, { "str", "oho" }, { "num", 456 }
};
using (var tx = session.BeginTransaction(txConfig => txConfig.WithMetadata(txMetadata))) {
tx.Run("MATCH (n) RETURN n LIMIT 1").Consume();
// Check transaction info from another thread
Thread show_tx = new Thread(() => ShowTx(ref driver));
show_tx.Start();
show_tx.Join();
// End current transaction
tx.Commit();
}
}
// Implicit transaction query
using (var session = driver.Session()) {
Console.WriteLine("Checking implicit transaction metadata...");
var txMetadata = new Dictionary<string, object> {
{ "ver", "session" }, { "str", "aha" }, { "num", 123 }
};
CheckMD(session.Run("SHOW TRANSACTIONS", txConfig => txConfig.WithMetadata(txMetadata)));
}
Console.WriteLine("All ok!");
}
private static void ClearDatabase(IDriver driver) {
using (var session = driver.Session()) session.Run("MATCH (n) DETACH DELETE n").Consume();
}
public static void ShowTx(ref IDriver driver) {
using (var session = driver.Session()) {
CheckMD(session.Run("SHOW TRANSACTIONS"));
}
}
public static void CheckMD(IResult tx_md) {
int n = 0;
try {
foreach (var res in tx_md) {
var md = res["metadata"].As<Dictionary<string, object>>();
if (md.Count != 0) {
if (md["ver"].As<string>() == "transaction" && md["str"].As<string>() == "oho" &&
md["num"].As<int>() == 456) {
n = n + 1;
} else if (md["ver"].As<string>() == "session" && md["str"].As<string>() == "aha" &&
md["num"].As<int>() == 123) {
n = n + 1;
}
}
}
} catch {
n = 0;
}
if (n == 0) {
Console.WriteLine("Metadata error!");
Environment.Exit(1);
}
}
}

View File

@@ -0,0 +1,12 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>netcoreapp2.1</TargetFramework>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Neo4j.Driver.Simple" Version="5.8.0" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,35 @@
using System;
using System.Linq;
using Neo4j.Driver;
public class Basic {
public static void Main(string[] args) {
using (var driver = GraphDatabase.Driver(
"bolt://localhost:7687", AuthTokens.None,
(ConfigBuilder builder) => builder.WithEncryptionLevel(
EncryptionLevel.None))) using (var session = driver.Session()) {
session.Run("MATCH (n) DETACH DELETE n;").Consume();
Console.WriteLine("Database cleared.");
session.Run("CREATE (alice:Person {name: \"Alice\", age: 22});").Consume();
Console.WriteLine("Record created.");
var node = (INode)session.Run("MATCH (n) RETURN n;").First()["n"];
Console.WriteLine("Record matched.");
var label = string.Join("", node.Labels);
var name = node["name"];
var age = (long)node["age"];
if (!label.Equals("Person") || !name.Equals("Alice") || !age.Equals(22)) {
Console.WriteLine("Data doesn't match!");
System.Environment.Exit(1);
}
Console.WriteLine("Label: " + label);
Console.WriteLine("name: " + name);
Console.WriteLine("age: " + age);
}
Console.WriteLine("All ok!");
}
}

View File

@@ -0,0 +1,64 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using Neo4j.Driver;
public class Transactions {
public static void Main(string[] args) {
using (var driver = GraphDatabase.Driver(
"bolt://localhost:7687", AuthTokens.None,
(builder) => builder.WithEncryptionLevel(EncryptionLevel.None))) {
ClearDatabase(driver);
// Wrong query.
try {
using (var session = driver.Session()) using (var tx = session.BeginTransaction()) {
CreatePerson(tx, "mirko");
// Incorrectly start CREATE
tx.Run("CREATE (").Consume();
CreatePerson(tx, "slavko");
tx.Commit();
}
} catch (ClientException) {
Console.WriteLine("Rolled back transaction");
}
Trace.Assert(CountNodes(driver) == 0, "Expected transaction was rolled back.");
// Correct query.
using (var session = driver.Session()) using (var tx = session.BeginTransaction()) {
CreatePerson(tx, "mirka");
CreatePerson(tx, "slavka");
tx.Commit();
}
Trace.Assert(CountNodes(driver) == 2, "Expected 2 created nodes.");
ClearDatabase(driver);
using (var session = driver.Session()) {
// Create a lot of nodes so that the next read takes a long time.
session.Run("UNWIND range(1, 100000) AS i CREATE ()").Consume();
try {
Console.WriteLine("Running a long read...");
session.Run("MATCH (a), (b), (c), (d), (e), (f) RETURN COUNT(*) AS cnt").Consume();
} catch (TransientException) {
Console.WriteLine("Transaction timed out");
}
}
}
Console.WriteLine("All ok!");
}
private static void CreatePerson(ITransaction tx, string name) {
var parameters = new Dictionary<string, Object> { { "name", name } };
var result = tx.Run("CREATE (person:Person {name: $name}) RETURN person", parameters);
Console.WriteLine("Created: " + ((INode)result.First()["person"])["name"]);
}
private static void ClearDatabase(IDriver driver) {
using (var session = driver.Session()) session.Run("MATCH (n) DETACH DELETE n").Consume();
}
private static int CountNodes(IDriver driver) {
using (var session = driver.Session()) {
var result = session.Run("MATCH (n) RETURN COUNT(*) AS cnt");
return Convert.ToInt32(result.First()["cnt"]);
}
}
}

Some files were not shown because too many files have changed in this diff Show More