Compare commits

...

33 Commits

Author SHA1 Message Date
Marko Budiselic
2d109a13d6 WIP 2022-06-06 17:31:34 +02:00
Marko Budiselić
ed71d01aa3 Merge branch 'master' into fix-commit-if-sync-replica-down 2022-06-06 15:20:56 +02:00
Marko Budiselić
21ad5d4328 Fix SHOW REPLICATION ROLE and SHOW REPLICAS (#376) 2022-05-20 20:17:59 -07:00
Marko Budiselić
8e3ab1ad0f Add frequent replica ping (#380) 2022-05-20 19:29:17 -07:00
Marko Budiselić
15afb4b5c2 Merge branch 'master' into fix-commit-if-sync-replica-down 2022-05-19 11:55:14 -07:00
Marko Budiselić
cccf32e79d Add default commented values for Bolt SSL flags (#398) 2022-05-18 08:34:07 +02:00
Jure Bajic
22bd60c613 Fix shutdown call (#395)
* Fix shutdown not called

* Add ssl server tests
2022-05-18 07:50:06 +02:00
János Benjamin Antal
8059a3e653 Improve magic functions docs (#391)
* Improve documentation of magic functions

* Improve wording for decorators
2022-05-17 15:33:28 +02:00
Antonio Andelic
a7f4c98bea Fix module symbol loading (#335)
* Use DEEPBIND

* Add dependency on libstdc++

Co-authored-by: Antonio Andelic <antonio.andelic@memgraph.io>
Co-authored-by: Jure Bajic <jure.bajic@memgraph.com>
2022-05-13 11:43:10 +02:00
Jure Bajic
483f4d04bd Update precommit hooks (#393)
* Add pre-commit

* Add clang pre-commit

* Install pre-commit in init
2022-05-11 21:07:04 +02:00
János Benjamin Antal
3e7aef432f Improve docker-compose setup for running streams e2e tests locally (#388)
* Separate pulsar and kafka docker-compose files

* Add docker-compose for redpanda

* Add small docs how to use docker-compose files
2022-05-10 13:31:45 +02:00
Marko Budiselic
05f3e7d243 Add one more option as comments 2022-05-08 08:04:35 +02:00
Marko Budiselic
7ac7b7c483 Update comment 2022-05-08 07:50:33 +02:00
Marko Budiselić
8734d8d1a9 Merge branch 'master' into fix-commit-if-sync-replica-down 2022-05-07 21:27:17 -07:00
Marko Budiselic
5df607f1e1 Add SYNC to UNABLE_TO_REPLICATE 2022-05-08 06:26:47 +02:00
Jeremy B
10ea9c773e Making mgp_error enum class when compiling c++. Remains enum when com… (#389)
* Making mgp_error enum class when compiling c++. Remains enum when compiling c.
2022-05-05 17:48:17 +02:00
Marko Budiselic
e32adaa18c Improve messages and comments 2022-05-01 04:17:54 +02:00
Marko Budiselic
28eef7edf4 Add CommitError struct 2022-05-01 04:02:27 +02:00
Marko Budiselic
e662206fd1 Add maybe feasible approach example 2022-04-29 05:00:12 +02:00
Jeremy B
b782271be8 Fix shared module loading
* Moving function add_query_module from CMakeLists from tests/e2e/magic_functions to tests/e2e

* Adding failing test copying behavior when loading c module (.so) into memgraph.

* Fixing issue where NO_ERROR status returned  MgpTransAddFixedResult was converted to false

* Removing unnecessary transformation

* removing incorrect parameterization of test

* re-adding parametrized transformation
2022-04-28 20:28:44 +02:00
Jure Bajic
a8ffcfa046 Update license year
* Update license year

Co-authored-by: János Benjamin Antal <antaljanosbenjamin@users.noreply.github.com>

Co-authored-by: János Benjamin Antal <antaljanosbenjamin@users.noreply.github.com>
2022-04-27 13:31:37 +02:00
Jure Bajic
7b78665cd8 Implement Bolt over WebSocket with asio
* Replace server implementation with asio

* Add support for bolt over WebSocket
2022-04-27 10:13:16 +02:00
Marko Budiselic
d47cc290ef Add complexity comments 2022-04-23 19:28:52 +02:00
Josip Matak
4abaf27765 Memgraph magic functions (#345)
* Extend mgp_module with include adding functions

* Add return type to the function API

* Change Cypher grammar

* Add Python support for functions

* Implement error handling

* E2e tests for functions

* Write cpp e2e functions

* Create mg.functions() procedure

* Implement case insensitivity for user-defined Magic Functions.
2022-04-21 15:45:31 +02:00
Kostas Kyrimis
ea2806bd57 Implement foreach clause (#351) 2022-04-11 13:55:34 +03:00
Siniša Šušnjar
c8dbaf5979 Small io network socket fixes (#360)
* Modernize AddrInfo

* Modernize Socket
2022-04-08 14:38:13 +02:00
Jure Bajic
17049ada09 Resolve python dependency issues (#372) 2022-04-07 17:56:18 +02:00
Jure Bajic
1b619f51b2 Add docker release action (#356)
* Add docker release action

* Add debian-11 arm script

* Remove test prefix

* Update package_docker script

* Update release script

* Unify architecture extension
2022-04-07 15:23:18 +02:00
János Benjamin Antal
537855a0b2 Fix usages of constexpr (#367)
* Fix usages of constexpr
2022-03-31 13:52:43 +02:00
Jure Bajic
5822b44b15 Fix CentOS 7 release process (#369) 2022-03-31 07:20:53 +02:00
g-despot
bf01c58ed9 Update README.md 2022-03-22 04:50:01 +01:00
g-despot
89a5566f3f Update README.md 2022-03-22 04:49:03 +01:00
g-despot
29452f8774 Update README.md 2022-03-21 16:49:02 +01:00
194 changed files with 5494 additions and 1406 deletions

View File

@@ -88,4 +88,3 @@ CheckOptions:
- key: modernize-use-nullptr.NullMacros
value: 'NULL'
...

View File

@@ -24,14 +24,6 @@ for file in $modified_files; do
git checkout-index --prefix="$tmpdir/" -- $file
echo "Running clang-format..."
$project_folder/tools/git-clang-format $tmpdir/$file
CODE=$?
if [ $CODE -ne 0 ]; then
break
fi
# Do not break header checker
echo "Running header checker..."
$project_folder/tools/header-checker.py $tmpdir/$file $file --amend-year
@@ -39,7 +31,6 @@ for file in $modified_files; do
if [ $CODE -ne 0 ]; then
FAIL=1
fi
done;
return ${FAIL}

49
.github/workflows/release_docker.yaml vendored Normal file
View File

@@ -0,0 +1,49 @@
name: Publish Docker images
on:
workflow_dispatch:
inputs:
version:
description: "Memgraph binary version to publish on Dockerhub."
required: true
jobs:
docker_publish:
runs-on: ubuntu-latest
env:
DOCKER_ORGANIZATION_NAME: memgraph
DOCKER_REPOSITORY_NAME: memgraph
steps:
- name: Checkout
uses: actions/checkout@v2
- name: Set up QEMU
uses: docker/setup-qemu-action@v1
- name: Set up Docker Buildx
id: buildx
uses: docker/setup-buildx-action@v1
- name: Log in to Docker Hub
uses: docker/login-action@v1
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Download memgraph binary
run: |
cd release/docker
curl -L https://download.memgraph.com/memgraph/v${{ github.event.inputs.version }}/debian-11/memgraph_${{ github.event.inputs.version }}-1_amd64.deb > memgraph-amd64.deb
curl -L https://download.memgraph.com/memgraph/v${{ github.event.inputs.version }}/debian-11-aarch64/memgraph_${{ github.event.inputs.version }}-1_arm64.deb > memgraph-arm64.deb
- name: Build & push docker images
run: |
cd release/docker
docker buildx build \
--build-arg BINARY_NAME="memgraph-" \
--build-arg EXTENSION="deb" \
--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 memgraph_deb.dockerfile \
--push .

24
.pre-commit-config.yaml Normal file
View File

@@ -0,0 +1,24 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v2.3.0
hooks:
- id: check-yaml
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: https://github.com/psf/black
rev: 22.3.0
hooks:
- id: black
args: # arguments to configure black
- --line-length=120
- --include='\.pyi?$'
# these folders wont be formatted by black
- --exclude="""\.git |
\.__pycache__|
build|
libs|
.cache"""
- repo: https://github.com/pre-commit/mirrors-clang-format
rev: v13.0.0
hooks:
- id: clang-format

View File

@@ -184,7 +184,8 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall \
-Werror=switch -Werror=switch-bool -Werror=return-type \
-Werror=return-stack-address \
-Wno-c99-designator")
-Wno-c99-designator \
-DBOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT")
# Don't omit frame pointer in RelWithDebInfo, for additional callchain debug.
set(CMAKE_CXX_FLAGS_RELWITHDEBINFO

View File

@@ -54,6 +54,18 @@ Memgraph is implemented in C/C++ and leverages an in-memory first architecture
to ensure that youre getting the best possible performance consistently and
without surprises. Its also ACID-compliant and highly available.
## :video_game: Memgraph Playground
You don't need to install anything to try out Memgraph. Check out
our **[Memgraph Playground](https://playground.memgraph.com/)** sandboxes in
your browser.
<p align="left">
<a href="https://playground.memgraph.com/">
<img width="450px" alt="Memgraph Playground" src="https://download.memgraph.com/asset/github/memgraph/memgraph-playground.png">
</a>
</p>
## :floppy_disk: Download & Install
### Windows

View File

@@ -47,6 +47,14 @@ modifications:
value: ""
override: false
- name: "bolt_cert_file"
value: "/etc/memgraph/ssl/cert.pem"
override: false
- name: "bolt_key_file"
value: "/etc/memgraph/ssl/key.pem"
override: false
- name: "storage_properties_on_edges"
value: "true"
override: true

View File

@@ -106,7 +106,7 @@ install() {
https://repo.ius.io/ius-release-el7.rpm
yum update -y
yum install -y wget python3 python3-pip
yum install -y git224
yum install -y git
for pkg in $1; do
if [ "$pkg" == libipt ]; then
if ! yum list installed libipt >/dev/null 2>/dev/null; then

104
environment/os/debian-11-arm.sh Executable file
View File

@@ -0,0 +1,104 @@
#!/bin/bash
set -Eeuo pipefail
DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
source "$DIR/../util.sh"
TOOLCHAIN_BUILD_DEPS=(
coreutils gcc g++ build-essential make # generic build tools
wget # used for archive download
gnupg # used for archive signature verification
tar gzip bzip2 xz-utils unzip # used for archive unpacking
zlib1g-dev # zlib library used for all builds
libexpat1-dev liblzma-dev python3-dev texinfo # for gdb
libcurl4-openssl-dev # for cmake
libreadline-dev # for cmake and llvm
libffi-dev libxml2-dev # for llvm
libedit-dev libpcre3-dev automake bison # for swig
curl # snappy
file # for libunwind
libssl-dev # for libevent
libgmp-dev
gperf # for proxygen
git # for fbthrift
)
TOOLCHAIN_RUN_DEPS=(
make # generic build tools
tar gzip bzip2 xz-utils # used for archive unpacking
zlib1g # zlib library used for all builds
libexpat1 liblzma5 python3 # for gdb
libcurl4 # for cmake
file # for CPack
libreadline8 # for cmake and llvm
libffi7 libxml2 # for llvm
libssl-dev # for libevent
)
MEMGRAPH_BUILD_DEPS=(
git # source code control
make pkg-config # build system
curl wget # for downloading libs
uuid-dev default-jre-headless # required by antlr
libreadline-dev # for memgraph console
libpython3-dev python3-dev # for query modules
libssl-dev
libseccomp-dev
netcat # tests are using nc to wait for memgraph
python3 virtualenv python3-virtualenv python3-pip # for qa, macro_benchmark and stress tests
python3-yaml # for the configuration generator
libcurl4-openssl-dev # mg-requests
sbcl # for custom Lisp C++ preprocessing
doxygen graphviz # source documentation generators
mono-runtime mono-mcs zip unzip default-jdk-headless # for driver tests
golang nodejs npm
autoconf # for jemalloc code generation
libtool # for protobuf code generation
)
list() {
echo "$1"
}
check() {
check_all_dpkg "$1"
}
install() {
cat >/etc/apt/sources.list <<EOF
deb http://deb.debian.org/debian bullseye main
deb-src http://deb.debian.org/debian bullseye main
deb http://deb.debian.org/debian-security/ bullseye-security main
deb-src http://deb.debian.org/debian-security/ bullseye-security main
deb http://deb.debian.org/debian bullseye-updates main
deb-src http://deb.debian.org/debian bullseye-updates main
EOF
cd "$DIR"
apt update
# If GitHub Actions runner is installed, append LANG to the environment.
# Python related tests doesn't work the LANG export.
if [ -d "/home/gh/actions-runner" ]; then
echo "LANG=en_US.utf8" >> /home/gh/actions-runner/.env
else
echo "NOTE: export LANG=en_US.utf8"
fi
apt install -y wget
for pkg in $1; do
if [ "$pkg" == dotnet-sdk-3.1 ]; then
if ! dpkg -s "$pkg" 2>/dev/null >/dev/null; then
wget -nv https://packages.microsoft.com/config/debian/10/packages-microsoft-prod.deb -O packages-microsoft-prod.deb
dpkg -i packages-microsoft-prod.deb
apt-get update
apt-get install -y apt-transport-https dotnet-sdk-3.1
fi
continue
fi
apt install -y "$pkg"
done
}
deps=$2"[*]"
"$1" "${!deps}"

View File

@@ -5,6 +5,10 @@ operating_system() {
sort | cut -d '=' -f 2- | sed 's/"//g' | paste -s -d '-'
}
architecture() {
uname -m
}
check_all_yum() {
local missing=""
for pkg in $1; do

File diff suppressed because it is too large Load Diff

View File

@@ -40,6 +40,7 @@ class InvalidContextError(Exception):
"""
Signals using a graph element instance outside of the registered procedure.
"""
pass
@@ -47,6 +48,7 @@ class UnknownError(_mgp.UnknownError):
"""
Signals unspecified failure.
"""
pass
@@ -54,6 +56,7 @@ class UnableToAllocateError(_mgp.UnableToAllocateError):
"""
Signals failed memory allocation.
"""
pass
@@ -61,6 +64,7 @@ class InsufficientBufferError(_mgp.InsufficientBufferError):
"""
Signals that some buffer is not big enough.
"""
pass
@@ -69,6 +73,7 @@ class OutOfRangeError(_mgp.OutOfRangeError):
Signals that an index-like parameter has a value that is outside its
possible values.
"""
pass
@@ -77,6 +82,7 @@ class LogicErrorError(_mgp.LogicErrorError):
Signals faulty logic within the program such as violating logical
preconditions or class invariants and may be preventable.
"""
pass
@@ -84,6 +90,7 @@ class DeletedObjectError(_mgp.DeletedObjectError):
"""
Signals accessing an already deleted object.
"""
pass
@@ -91,6 +98,7 @@ class InvalidArgumentError(_mgp.InvalidArgumentError):
"""
Signals that some of the arguments have invalid values.
"""
pass
@@ -98,6 +106,7 @@ class KeyAlreadyExistsError(_mgp.KeyAlreadyExistsError):
"""
Signals that a key already exists in a container-like object.
"""
pass
@@ -105,6 +114,7 @@ class ImmutableObjectError(_mgp.ImmutableObjectError):
"""
Signals modification of an immutable object.
"""
pass
@@ -112,6 +122,7 @@ class ValueConversionError(_mgp.ValueConversionError):
"""
Signals that the conversion failed between python and cypher values.
"""
pass
@@ -120,12 +131,14 @@ class SerializationError(_mgp.SerializationError):
Signals serialization error caused by concurrent modifications from
different transactions.
"""
pass
class Label:
"""Label of a Vertex."""
__slots__ = ('_name',)
__slots__ = ("_name",)
def __init__(self, name: str):
self._name = name
@@ -145,19 +158,22 @@ class Label:
# Named property value of a Vertex or an Edge.
# It would be better to use typing.NamedTuple with typed fields, but that is
# not available in Python 3.5.
Property = namedtuple('Property', ('name', 'value'))
Property = namedtuple("Property", ("name", "value"))
class Properties:
"""
A collection of properties either on a Vertex or an Edge.
"""
__slots__ = ('_vertex_or_edge', '_len',)
__slots__ = (
"_vertex_or_edge",
"_len",
)
def __init__(self, vertex_or_edge):
if not isinstance(vertex_or_edge, (_mgp.Vertex, _mgp.Edge)):
raise TypeError("Expected '_mgp.Vertex' or '_mgp.Edge', \
got {}".format(type(vertex_or_edge)))
raise TypeError("Expected '_mgp.Vertex' or '_mgp.Edge', got {}".format(type(vertex_or_edge)))
self._len = None
self._vertex_or_edge = vertex_or_edge
@@ -330,7 +346,8 @@ class Properties:
class EdgeType:
"""Type of an Edge."""
__slots__ = ('_name',)
__slots__ = ("_name",)
def __init__(self, name):
self._name = name
@@ -348,7 +365,7 @@ class EdgeType:
if sys.version_info >= (3, 5, 2):
EdgeId = typing.NewType('EdgeId', int)
EdgeId = typing.NewType("EdgeId", int)
else:
EdgeId = int
@@ -360,12 +377,12 @@ class Edge:
a query. You should not globally store an instance of an Edge. Using an
invalid Edge instance will raise InvalidContextError.
"""
__slots__ = ('_edge',)
__slots__ = ("_edge",)
def __init__(self, edge):
if not isinstance(edge, _mgp.Edge):
raise TypeError(
"Expected '_mgp.Edge', got '{}'".format(type(edge)))
raise TypeError("Expected '_mgp.Edge', got '{}'".format(type(edge)))
self._edge = edge
def __deepcopy__(self, memo):
@@ -408,7 +425,7 @@ class Edge:
return EdgeType(self._edge.get_type_name())
@property
def from_vertex(self) -> 'Vertex':
def from_vertex(self) -> "Vertex":
"""
Get the source vertex.
@@ -419,7 +436,7 @@ class Edge:
return Vertex(self._edge.from_vertex())
@property
def to_vertex(self) -> 'Vertex':
def to_vertex(self) -> "Vertex":
"""
Get the destination vertex.
@@ -453,7 +470,7 @@ class Edge:
if sys.version_info >= (3, 5, 2):
VertexId = typing.NewType('VertexId', int)
VertexId = typing.NewType("VertexId", int)
else:
VertexId = int
@@ -465,12 +482,12 @@ class Vertex:
in a query. You should not globally store an instance of a Vertex. Using an
invalid Vertex instance will raise InvalidContextError.
"""
__slots__ = ('_vertex',)
__slots__ = ("_vertex",)
def __init__(self, vertex):
if not isinstance(vertex, _mgp.Vertex):
raise TypeError(
"Expected '_mgp.Vertex', got '{}'".format(type(vertex)))
raise TypeError("Expected '_mgp.Vertex', got '{}'".format(type(vertex)))
self._vertex = vertex
def __deepcopy__(self, memo):
@@ -513,8 +530,7 @@ class Vertex:
"""
if not self.is_valid():
raise InvalidContextError()
return tuple(Label(self._vertex.label_at(i))
for i in range(self._vertex.labels_count()))
return tuple(Label(self._vertex.label_at(i)) for i in range(self._vertex.labels_count()))
def add_label(self, label: str) -> None:
"""
@@ -615,7 +631,8 @@ class Vertex:
class Path:
"""Path containing Vertex and Edge instances."""
__slots__ = ('_path', '_vertices', '_edges')
__slots__ = ("_path", "_vertices", "_edges")
def __init__(self, starting_vertex_or_path: typing.Union[_mgp.Path, Vertex]):
"""Initialize with a starting Vertex.
@@ -636,8 +653,7 @@ class Path:
raise InvalidContextError()
self._path = _mgp.Path.make_with_start(vertex)
else:
raise TypeError("Expected '_mgp.Vertex' or '_mgp.Path', got '{}'"
.format(type(starting_vertex_or_path)))
raise TypeError("Expected '_mgp.Vertex' or '_mgp.Path', got '{}'".format(type(starting_vertex_or_path)))
def __copy__(self):
if not self.is_valid():
@@ -678,8 +694,7 @@ class Path:
extension.
"""
if not isinstance(edge, Edge):
raise TypeError(
"Expected '_mgp.Edge', got '{}'".format(type(edge)))
raise TypeError("Expected '_mgp.Edge', got '{}'".format(type(edge)))
if not self.is_valid() or not edge.is_valid():
raise InvalidContextError()
self._path.expand(edge._edge)
@@ -698,8 +713,7 @@ class Path:
raise InvalidContextError()
if self._vertices is None:
num_vertices = self._path.size() + 1
self._vertices = tuple(Vertex(self._path.vertex_at(i))
for i in range(num_vertices))
self._vertices = tuple(Vertex(self._path.vertex_at(i)) for i in range(num_vertices))
return self._vertices
@property
@@ -713,14 +727,14 @@ class Path:
raise InvalidContextError()
if self._edges is None:
num_edges = self._path.size()
self._edges = tuple(Edge(self._path.edge_at(i))
for i in range(num_edges))
self._edges = tuple(Edge(self._path.edge_at(i)) for i in range(num_edges))
return self._edges
class Record:
"""Represents a record of resulting field values."""
__slots__ = ('fields',)
__slots__ = ("fields",)
def __init__(self, **kwargs):
"""Initialize with name=value fields in kwargs."""
@@ -729,12 +743,12 @@ class Record:
class Vertices:
"""Iterable over vertices in a graph."""
__slots__ = ('_graph', '_len')
__slots__ = ("_graph", "_len")
def __init__(self, graph):
if not isinstance(graph, _mgp.Graph):
raise TypeError(
"Expected '_mgp.Graph', got '{}'".format(type(graph)))
raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
self._graph = graph
self._len = None
@@ -791,12 +805,12 @@ class Vertices:
class Graph:
"""State of the graph database in current ProcCtx."""
__slots__ = ('_graph',)
__slots__ = ("_graph",)
def __init__(self, graph):
if not isinstance(graph, _mgp.Graph):
raise TypeError(
"Expected '_mgp.Graph', got '{}'".format(type(graph)))
raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
self._graph = graph
def __deepcopy__(self, memo):
@@ -885,8 +899,7 @@ class Graph:
raise InvalidContextError()
self._graph.detach_delete_vertex(vertex._vertex)
def create_edge(self, from_vertex: Vertex, to_vertex: Vertex,
edge_type: EdgeType) -> None:
def create_edge(self, from_vertex: Vertex, to_vertex: Vertex, edge_type: EdgeType) -> None:
"""
Create an edge.
@@ -899,8 +912,7 @@ class Graph:
"""
if not self.is_valid():
raise InvalidContextError()
return Edge(self._graph.create_edge(from_vertex._vertex,
to_vertex._vertex, edge_type.name))
return Edge(self._graph.create_edge(from_vertex._vertex, to_vertex._vertex, edge_type.name))
def delete_edge(self, edge: Edge) -> None:
"""
@@ -918,6 +930,7 @@ class Graph:
class AbortError(Exception):
"""Signals that the procedure was asked to abort its execution."""
pass
@@ -927,12 +940,12 @@ class ProcCtx:
Access to a ProcCtx is only valid during a single execution of a procedure
in a query. You should not globally store a ProcCtx instance.
"""
__slots__ = ('_graph',)
__slots__ = ("_graph",)
def __init__(self, graph):
if not isinstance(graph, _mgp.Graph):
raise TypeError(
"Expected '_mgp.Graph', got '{}'".format(type(graph)))
raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
self._graph = Graph(graph)
def is_valid(self) -> bool:
@@ -969,8 +982,7 @@ LocalDateTime = datetime.datetime
Duration = datetime.timedelta
Any = typing.Union[bool, str, Number, Map, Path,
list, Date, LocalTime, LocalDateTime, Duration]
Any = typing.Union[bool, str, Number, Map, Path, list, Date, LocalTime, LocalDateTime, Duration]
List = typing.List
@@ -1003,7 +1015,7 @@ def _typing_to_cypher_type(type_):
Date: _mgp.type_date(),
LocalTime: _mgp.type_local_time(),
LocalDateTime: _mgp.type_local_date_time(),
Duration: _mgp.type_duration()
Duration: _mgp.type_duration(),
}
try:
return simple_types[type_]
@@ -1021,14 +1033,14 @@ def _typing_to_cypher_type(type_):
if type(None) in type_args:
types = tuple(t for t in type_args if t is not type(None)) # noqa E721
if len(types) == 1:
type_arg, = types
(type_arg,) = types
else:
# We cannot do typing.Union[*types], so do the equivalent
# with __getitem__ which does not even need arg unpacking.
type_arg = typing.Union.__getitem__(types)
return _mgp.type_nullable(_typing_to_cypher_type(type_arg))
elif complex_type == list:
type_arg, = type_args
(type_arg,) = type_args
return _mgp.type_list(_typing_to_cypher_type(type_arg))
raise UnsupportedTypingError(type_)
else:
@@ -1038,13 +1050,17 @@ def _typing_to_cypher_type(type_):
# printed the same way. `typing.List[type]` is printed as such, while
# `typing.Optional[type]` is printed as 'typing.Union[type, NoneType]'
def parse_type_args(type_as_str):
return tuple(map(str.strip,
type_as_str[type_as_str.index('[') + 1: -1].split(',')))
return tuple(
map(
str.strip,
type_as_str[type_as_str.index("[") + 1 : -1].split(","),
)
)
def fully_qualified_name(cls):
if cls.__module__ is None or cls.__module__ == 'builtins':
if cls.__module__ is None or cls.__module__ == "builtins":
return cls.__name__
return cls.__module__ + '.' + cls.__name__
return cls.__module__ + "." + cls.__name__
def get_simple_type(type_as_str):
for simple_type, cypher_type in simple_types.items():
@@ -1060,28 +1076,26 @@ def _typing_to_cypher_type(type_):
pass
def parse_typing(type_as_str):
if type_as_str.startswith('typing.Union'):
if type_as_str.startswith("typing.Union"):
type_args_as_str = parse_type_args(type_as_str)
none_type_as_str = type(None).__name__
if none_type_as_str in type_args_as_str:
types = tuple(
t for t in type_args_as_str if t != none_type_as_str)
types = tuple(t for t in type_args_as_str if t != none_type_as_str)
if len(types) == 1:
type_arg_as_str, = types
(type_arg_as_str,) = types
else:
type_arg_as_str = 'typing.Union[' + \
', '.join(types) + ']'
type_arg_as_str = "typing.Union[" + ", ".join(types) + "]"
simple_type = get_simple_type(type_arg_as_str)
if simple_type is not None:
return _mgp.type_nullable(simple_type)
return _mgp.type_nullable(parse_typing(type_arg_as_str))
elif type_as_str.startswith('typing.List'):
elif type_as_str.startswith("typing.List"):
type_arg_as_str = parse_type_args(type_as_str)
if len(type_arg_as_str) > 1:
# Nested object could be a type consisting of a list of types (e.g. mgp.Map)
# so we need to join the parts.
type_arg_as_str = ', '.join(type_arg_as_str)
type_arg_as_str = ", ".join(type_arg_as_str)
else:
type_arg_as_str = type_arg_as_str[0]
@@ -1096,9 +1110,11 @@ def _typing_to_cypher_type(type_):
# Procedure registration
class Deprecated:
"""Annotate a resulting Record's field as deprecated."""
__slots__ = ('field_type',)
__slots__ = ("field_type",)
def __init__(self, type_):
self.field_type = type_
@@ -1106,8 +1122,7 @@ class Deprecated:
def raise_if_does_not_meet_requirements(func: typing.Callable[..., Record]):
if not callable(func):
raise TypeError("Expected a callable object, got an instance of '{}'"
.format(type(func)))
raise TypeError("Expected a callable object, got an instance of '{}'".format(type(func)))
if inspect.iscoroutinefunction(func):
raise TypeError("Callable must not be 'async def' function")
if sys.version_info >= (3, 6):
@@ -1117,24 +1132,25 @@ def raise_if_does_not_meet_requirements(func: typing.Callable[..., Record]):
raise NotImplementedError("Generator functions are not supported")
def _register_proc(func: typing.Callable[..., Record],
is_write: bool):
def _register_proc(func: typing.Callable[..., Record], is_write: bool):
raise_if_does_not_meet_requirements(func)
register_func = (
_mgp.Module.add_write_procedure if is_write
else _mgp.Module.add_read_procedure)
register_func = _mgp.Module.add_write_procedure if is_write else _mgp.Module.add_read_procedure
sig = inspect.signature(func)
params = tuple(sig.parameters.values())
if params and params[0].annotation is ProcCtx:
@wraps(func)
def wrapper(graph, args):
return func(ProcCtx(graph), *args)
params = params[1:]
mgp_proc = register_func(_mgp._MODULE, wrapper)
else:
@wraps(func)
def wrapper(graph, args):
return func(*args)
mgp_proc = register_func(_mgp._MODULE, wrapper)
for param in params:
name = param.name
@@ -1149,8 +1165,7 @@ def _register_proc(func: typing.Callable[..., Record],
if sig.return_annotation is not sig.empty:
record = sig.return_annotation
if not isinstance(record, Record):
raise TypeError("Expected '{}' to return 'mgp.Record', got '{}'"
.format(func.__name__, type(record)))
raise TypeError("Expected '{}' to return 'mgp.Record', got '{}'".format(func.__name__, type(record)))
for name, type_ in record.fields.items():
if isinstance(type_, Deprecated):
cypher_type = _typing_to_cypher_type(type_.field_type)
@@ -1164,16 +1179,15 @@ def read_proc(func: typing.Callable[..., Record]):
"""
Register `func` as a read-only procedure of the current module.
`read_proc` is meant to be used as a decorator function to register module
procedures. The registered `func` needs to be a callable which optionally
takes `ProcCtx` as the first argument. Other arguments of `func` will be
bound to values passed in the cypherQuery. The full signature of `func`
needs to be annotated with types. The return type must be
`Record(field_name=type, ...)` and the procedure must produce either a
complete Record or None. To mark a field as deprecated, use
`Record(field_name=Deprecated(type), ...)`. Multiple records can be
produced by returning an iterable of them. Registering generator functions
is currently not supported.
The decorator `read_proc` is meant to be used to register module procedures.
The registered `func` needs to be a callable which optionally takes
`ProcCtx` as its first argument. Other arguments of `func` will be bound to
values passed in the cypherQuery. The full signature of `func` needs to be
annotated with types. The return type must be `Record(field_name=type, ...)`
and the procedure must produce either a complete Record or None. To mark a
field as deprecated, use `Record(field_name=Deprecated(type), ...)`.
Multiple records can be produced by returning an iterable of them.
Registering generator functions is currently not supported.
Example usage.
@@ -1207,16 +1221,16 @@ def write_proc(func: typing.Callable[..., Record]):
"""
Register `func` as a writeable procedure of the current module.
`write_proc` is meant to be used as a decorator function to register module
The decorator `write_proc` is meant to be used to register module
procedures. The registered `func` needs to be a callable which optionally
takes `ProcCtx` as the first argument. Other arguments of `func` will be
bound to values passed in the cypherQuery. The full signature of `func`
needs to be annotated with types. The return type must be
`Record(field_name=type, ...)` and the procedure must produce either a
complete Record or None. To mark a field as deprecated, use
`Record(field_name=Deprecated(type), ...)`. Multiple records can be
produced by returning an iterable of them. Registering generator functions
is currently not supported.
`Record(field_name=Deprecated(type), ...)`. Multiple records can be produced
by returning an iterable of them. Registering generator functions is
currently not supported.
Example usage.
@@ -1257,20 +1271,22 @@ class InvalidMessageError(Exception):
"""
Signals using a message instance outside of the registered transformation.
"""
pass
SOURCE_TYPE_KAFKA = _mgp.SOURCE_TYPE_KAFKA
SOURCE_TYPE_PULSAR = _mgp.SOURCE_TYPE_PULSAR
class Message:
"""Represents a message from a stream."""
__slots__ = ('_message',)
__slots__ = ("_message",)
def __init__(self, message):
if not isinstance(message, _mgp.Message):
raise TypeError(
"Expected '_mgp.Message', got '{}'".format(type(message)))
raise TypeError("Expected '_mgp.Message', got '{}'".format(type(message)))
self._message = message
def __deepcopy__(self, memo):
@@ -1353,17 +1369,18 @@ class Message:
class InvalidMessagesError(Exception):
"""Signals using a messages instance outside of the registered transformation."""
pass
class Messages:
"""Represents a list of messages from a stream."""
__slots__ = ('_messages',)
__slots__ = ("_messages",)
def __init__(self, messages):
if not isinstance(messages, _mgp.Messages):
raise TypeError(
"Expected '_mgp.Messages', got '{}'".format(type(messages)))
raise TypeError("Expected '_mgp.Messages', got '{}'".format(type(messages)))
self._messages = messages
def __deepcopy__(self, memo):
@@ -1395,12 +1412,12 @@ class TransCtx:
Access to a TransCtx is only valid during a single execution of a transformation.
You should not globally store a TransCtx instance.
"""
__slots__ = ('_graph')
__slots__ = "_graph"
def __init__(self, graph):
if not isinstance(graph, _mgp.Graph):
raise TypeError(
"Expected '_mgp.Graph', got '{}'".format(type(graph)))
raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
self._graph = Graph(graph)
def is_valid(self) -> bool:
@@ -1420,21 +1437,116 @@ def transformation(func: typing.Callable[..., Record]):
params = tuple(sig.parameters.values())
if not params or not params[0].annotation is Messages:
if not len(params) == 2 or not params[1].annotation is Messages:
raise NotImplementedError(
"Valid signatures for transformations are (TransCtx, Messages) or (Messages)")
raise NotImplementedError("Valid signatures for transformations are (TransCtx, Messages) or (Messages)")
if params[0].annotation is TransCtx:
@wraps(func)
def wrapper(graph, messages):
return func(TransCtx(graph), messages)
_mgp._MODULE.add_transformation(wrapper)
else:
@wraps(func)
def wrapper(graph, messages):
return func(messages)
_mgp._MODULE.add_transformation(wrapper)
return func
class FuncCtx:
"""Context of a function being executed.
Access to a FuncCtx is only valid during a single execution of a function in
a query. You should not globally store a FuncCtx instance. The graph object
within the FuncCtx is not mutable.
"""
__slots__ = "_graph"
def __init__(self, graph):
if not isinstance(graph, _mgp.Graph):
raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
self._graph = Graph(graph)
def is_valid(self) -> bool:
return self._graph.is_valid()
def function(func: typing.Callable):
"""
Register `func` as a user-defined function in the current module.
The decorator `function` is meant to be used to register module functions.
The registered `func` needs to be a callable which optionally takes
`FuncCtx` as its first argument. Other arguments of `func` will be bound to
values passed in the Cypher query. Only the funcion arguments need to be
annotated with types. The return type doesn't need to be specified, but it
has to be supported by `mgp.Any`. Registering generator functions is
currently not supported.
Example usage.
```
import mgp
@mgp.function
def func_example(context: mgp.FuncCtx,
required_arg: str,
optional_arg: mgp.Nullable[str] = None
):
return_args = [required_arg]
if optional_arg is not None:
return_args.append(optional_arg)
# Return any kind of result supported by mgp.Any
return return_args
```
The example function above returns a list of provided arguments:
* `required_arg` is always present and its value is the first argument of
the function.
* `optional_arg` is present if the second argument of the function is not
`null`.
Any errors can be reported by raising an Exception.
The function can be invoked in Cypher using the following calls:
RETURN example.func_example("first argument", "second_argument");
RETURN example.func_example("first argument");
Naturally, you may pass in different arguments.
"""
raise_if_does_not_meet_requirements(func)
register_func = _mgp.Module.add_function
sig = inspect.signature(func)
params = tuple(sig.parameters.values())
if params and params[0].annotation is FuncCtx:
@wraps(func)
def wrapper(graph, args):
return func(FuncCtx(graph), *args)
params = params[1:]
mgp_func = register_func(_mgp._MODULE, wrapper)
else:
@wraps(func)
def wrapper(graph, args):
return func(*args)
mgp_func = register_func(_mgp._MODULE, wrapper)
for param in params:
name = param.name
type_ = param.annotation
if type_ is param.empty:
type_ = object
cypher_type = _typing_to_cypher_type(type_)
if param.default is param.empty:
mgp_func.add_arg(name, cypher_type)
else:
mgp_func.add_opt_arg(name, cypher_type, param.default)
return func
def _wrap_exceptions():
def wrap_function(func):
@wraps(func)
@@ -1463,6 +1575,7 @@ def _wrap_exceptions():
raise ValueConversionError(e)
except _mgp.SerializationError as e:
raise SerializationError(e)
return wrapped_func
def wrap_prop_func(func):
@@ -1473,11 +1586,16 @@ def _wrap_exceptions():
if inspect.isfunction(obj):
setattr(cls, name, wrap_function(obj))
elif isinstance(obj, property):
setattr(cls, name, property(
wrap_prop_func(obj.fget),
wrap_prop_func(obj.fset),
wrap_prop_func(obj.fdel),
obj.__doc__))
setattr(
cls,
name,
property(
wrap_prop_func(obj.fget),
wrap_prop_func(obj.fset),
wrap_prop_func(obj.fdel),
obj.__doc__,
),
)
def defined_in_this_module(obj: object):
return getattr(obj, "__module__", "") == __name__

14
init
View File

@@ -65,8 +65,14 @@ else
fi
DISTRO=$(operating_system)
echo "ALL BUILD PACKAGES: $($DIR/environment/os/$DISTRO.sh list MEMGRAPH_BUILD_DEPS)"
$DIR/environment/os/$DISTRO.sh check MEMGRAPH_BUILD_DEPS
ARCHITECTURE=$(architecture)
if [ "${ARCHITECTURE}" = "arm64" ]; then
OS_SCRIPT=$DIR/environment/os/$DISTRO-arm.sh
else
OS_SCRIPT=$DIR/environment/os/$DISTRO.sh
fi
echo "ALL BUILD PACKAGES: $($OS_SCRIPT list MEMGRAPH_BUILD_DEPS)"
$OS_SCRIPT check MEMGRAPH_BUILD_DEPS
echo "All packages are in-place..."
# create a default build directory
@@ -129,3 +135,7 @@ for hook in $(find $DIR/.githooks -type f -printf "%f\n"); do
ln -s -f "$DIR/.githooks/$hook" "$DIR/.git/hooks/$hook"
echo "Added $hook hook"
done;
# Install precommit hook
python3 -m pip install pre-commit
python3 -m pre_commit install

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: 2026-18-02
CHANGE DATE: 2026-27-04
CHANGE LICENSE: Apache License, Version 2.0
For information about alternative licensing arrangements, please visit: https://memgraph.com/legal.

View File

@@ -10,6 +10,14 @@ set(CPACK_PACKAGE_VENDOR "Memgraph Ltd.")
set(CPACK_PACKAGE_DESCRIPTION_SUMMARY
"High performance, in-memory, transactional graph database")
# Setting arhitecture extension for deb packages
set(MG_ARCH_EXTENSION_DEB "all")
if (${MG_ARCH} STREQUAL "x86_64")
set(MG_ARCH_EXTENSION_DEB "amd64")
elseif (${MG_ARCH} STREQUAL "ARM64")
set(MG_ARCH_EXTENSION_DEB "arm64")
endif()
# DEB specific
# Instead of using "name <email>" format, we use "email (name)" to prevent
# errors due to full stop, '.' at the end of "Ltd". (See: RFC 822)
@@ -17,7 +25,7 @@ set(CPACK_DEBIAN_PACKAGE_MAINTAINER "tech@memgraph.com (Memgraph Ltd.)")
set(CPACK_DEBIAN_PACKAGE_SECTION non-free/database)
set(CPACK_DEBIAN_PACKAGE_HOMEPAGE https://memgraph.com)
set(CPACK_DEBIAN_PACKAGE_VERSION "${MEMGRAPH_VERSION_DEB}")
set(CPACK_DEBIAN_FILE_NAME "memgraph_${MEMGRAPH_VERSION_DEB}_amd64.deb")
set(CPACK_DEBIAN_FILE_NAME "memgraph_${MEMGRAPH_VERSION_DEB}_${MG_ARCH_EXTENSION_DEB}.deb")
set(CPACK_DEBIAN_PACKAGE_CONTROL_EXTRA
"${CMAKE_CURRENT_SOURCE_DIR}/debian/conffiles;"
"${CMAKE_CURRENT_SOURCE_DIR}/debian/copyright;"
@@ -33,21 +41,20 @@ set(CPACK_DEBIAN_PACKAGE_DESCRIPTION "${CPACK_PACKAGE_DESCRIPTION_SUMMARY}
applications driver by real-time connected data.")
# Add `openssl` package to dependencies list. Used to generate SSL certificates.
# We also depend on `python3` because we embed it in Memgraph.
set(CPACK_DEBIAN_PACKAGE_DEPENDS "openssl (>= 1.1.0), python3 (>= 3.5.0)")
# RPM specific
set(MG_ARCH_EXTENSION "noarch")
set(CPACK_DEBIAN_PACKAGE_DEPENDS "openssl (>= 1.1.0), python3 (>= 3.5.0), libstdc++6")
# Setting arhitecture extension for rpm packages
set(MG_ARCH_EXTENSION_RPM "noarch")
if (${MG_ARCH} STREQUAL "x86_64")
set(MG_ARCH_EXTENSION "x86_64")
set(MG_ARCH_EXTENSION_RPM "x86_64")
elseif (${MG_ARCH} STREQUAL "ARM64")
set(MG_ARCH_EXTENSION "aarch64")
set(MG_ARCH_EXTENSION_RPM "aarch64")
endif()
# RPM specific
set(CPACK_RPM_PACKAGE_URL https://memgraph.com)
set(CPACK_RPM_PACKAGE_VERSION "${MEMGRAPH_VERSION_RPM}")
set(CPACK_RPM_FILE_NAME "memgraph-${MEMGRAPH_VERSION_RPM}-1.${MG_ARCH_EXTENSION}.rpm")
set(CPACK_RPM_FILE_NAME "memgraph-${MEMGRAPH_VERSION_RPM}-1.${MG_ARCH_EXTENSION_RPM}.rpm")
set(CPACK_RPM_EXCLUDE_FROM_AUTO_FILELIST_ADDITION
/var /var/lib /var/log /etc/logrotate.d
/lib /lib/systemd /lib/systemd/system /lib/systemd/system/memgraph.service)
@@ -60,7 +67,7 @@ It aims to deliver developers the speed, simplicity and scale required to build
the next generation of applications driver by real-time connected data.")
# Add `openssl` package to dependencies list. Used to generate SSL certificates.
# We also depend on `python3` because we embed it in Memgraph.
set(CPACK_RPM_PACKAGE_REQUIRES "openssl >= 1.0.0, curl >= 7.29.0, python3 >= 3.5.0")
set(CPACK_RPM_PACKAGE_REQUIRES "openssl >= 1.0.0, curl >= 7.29.0, python3 >= 3.5.0, libstdc >= 6")
# All variables must be set before including.
include(CPack)

View File

@@ -1,19 +1,21 @@
FROM debian:bullseye
# NOTE: If you change the base distro update release/package as well.
ARG release
ARG BINARY_NAME
ARG EXTENSION
ARG TARGETARCH
RUN apt-get update && apt-get install -y \
openssl libcurl4 libssl1.1 libseccomp2 python3 libpython3.9 python3-pip \
--no-install-recommends \
openssl libcurl4 libssl1.1 libseccomp2 python3 libpython3.9 python3-pip \
--no-install-recommends \
&& rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
RUN pip3 install networkx==2.4 numpy==1.21.4 scipy==1.7.3
COPY ${release} /
COPY "${BINARY_NAME}${TARGETARCH}.${EXTENSION}" /
# Install memgraph package
RUN dpkg -i ${release}
RUN dpkg -i "${BINARY_NAME}${TARGETARCH}.deb"
# Memgraph listens for Bolt Protocol on this port by default.
EXPOSE 7687

View File

@@ -55,7 +55,10 @@ image_name="memgraph:${version}"
image_package_name="memgraph-${version}-docker.tar.gz"
# Build docker image.
docker build -t ${image_name} ${tag_latest} -f ${dockerfile_path} --build-arg release=${package_name}.${extension} .
docker build -t ${image_name} ${tag_latest} -f ${dockerfile_path} \
--build-arg BINARY_NAME=${package_name} \
--build-arg EXTENSION=${extension} \
--build-arg TARGETARCH="" .
docker save ${image_name} ${latest_image} | gzip > ${image_package_name}
rm "${package_name}.${extension}"
echo "Built Docker image at '${working_dir}/${image_package_name}'"

View File

@@ -22,7 +22,7 @@
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_bool(auth_password_permit_null, true, "Set to false to disable null passwords.");
constexpr std::string_view default_password_regex = ".+";
inline constexpr std::string_view default_password_regex = ".+";
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_string(auth_password_strength_regex, default_password_regex.data(),
"The regular expression that should be used to match the entire "

View File

@@ -15,8 +15,8 @@
namespace memgraph::communication::bolt {
static constexpr uint8_t kPreamble[4] = {0x60, 0x60, 0xB0, 0x17};
static constexpr uint8_t kProtocol[4] = {0x00, 0x00, 0x00, 0x01};
inline constexpr uint8_t kPreamble[4] = {0x60, 0x60, 0xB0, 0x17};
inline constexpr uint8_t kProtocol[4] = {0x00, 0x00, 0x00, 0x01};
enum class Signature : uint8_t {
Noop = 0x00,
@@ -95,9 +95,9 @@ enum class Marker : uint8_t {
Struct16 = 0xDD,
};
static constexpr uint8_t MarkerString = 0, MarkerList = 1, MarkerMap = 2;
static constexpr Marker MarkerTiny[3] = {Marker::TinyString, Marker::TinyList, Marker::TinyMap};
static constexpr Marker Marker8[3] = {Marker::String8, Marker::List8, Marker::Map8};
static constexpr Marker Marker16[3] = {Marker::String16, Marker::List16, Marker::Map16};
static constexpr Marker Marker32[3] = {Marker::String32, Marker::List32, Marker::Map32};
inline constexpr uint8_t MarkerString = 0, MarkerList = 1, MarkerMap = 2;
inline constexpr Marker MarkerTiny[3] = {Marker::TinyString, Marker::TinyList, Marker::TinyMap};
inline constexpr Marker Marker8[3] = {Marker::String8, Marker::List8, Marker::Map8};
inline constexpr Marker Marker16[3] = {Marker::String16, Marker::List16, Marker::Map16};
inline constexpr Marker Marker32[3] = {Marker::String32, Marker::List32, Marker::Map32};
} // namespace memgraph::communication::bolt

View File

@@ -19,17 +19,17 @@ namespace memgraph::communication::bolt {
/**
* Sizes related to the chunk defined in Bolt protocol.
*/
static constexpr size_t kChunkHeaderSize = 2;
static constexpr size_t kChunkMaxDataSize = 65535;
static constexpr size_t kChunkWholeSize = kChunkHeaderSize + kChunkMaxDataSize;
inline constexpr size_t kChunkHeaderSize = 2;
inline constexpr size_t kChunkMaxDataSize = 65535;
inline constexpr size_t kChunkWholeSize = kChunkHeaderSize + kChunkMaxDataSize;
/**
* Handshake size defined in the Bolt protocol.
*/
static constexpr size_t kHandshakeSize = 20;
inline constexpr size_t kHandshakeSize = 20;
static constexpr uint16_t kSupportedVersions[] = {0x0100, 0x0400, 0x0401, 0x0403};
inline constexpr uint16_t kSupportedVersions[] = {0x0100, 0x0400, 0x0401, 0x0403};
static constexpr int kPullAll = -1;
static constexpr int kPullLast = -1;
inline constexpr int kPullAll = -1;
inline constexpr int kPullLast = -1;
} // namespace memgraph::communication::bolt

View File

@@ -78,14 +78,18 @@ bool ClientContext::use_ssl() { return use_ssl_; }
ServerContext::ServerContext(const std::string &key_file, const std::string &cert_file, const std::string &ca_file,
bool verify_peer) {
ctx_.emplace(boost::asio::ssl::context::tls_server);
namespace ssl = boost::asio::ssl;
ctx_.emplace(ssl::context::tls_server);
// NOLINTNEXTLINE(hicpp-signed-bitwise)
ctx_->set_options(ssl::context::default_workarounds | ssl::context::no_sslv2 | ssl::context::no_sslv3 |
ssl::context::single_dh_use);
ctx_->set_default_verify_paths();
// TODO: add support for encrypted private keys
// TODO: add certificate revocation list (CRL)
boost::system::error_code ec;
ctx_->use_certificate_chain_file(cert_file, ec);
MG_ASSERT(!ec, "Couldn't load server certificate from file: {}", cert_file);
ctx_->use_private_key_file(key_file, boost::asio::ssl::context::pem, ec);
ctx_->use_private_key_file(key_file, ssl::context::pem, ec);
MG_ASSERT(!ec, "Couldn't load server private key from file: {}", key_file);
ctx_->set_options(SSL_OP_NO_SSLv3, ec);
@@ -100,7 +104,7 @@ ServerContext::ServerContext(const std::string &key_file, const std::string &cer
if (verify_peer) {
// Enable verification of the client certificate.
// NOLINTNEXTLINE(hicpp-signed-bitwise)
ctx_->set_verify_mode(boost::asio::ssl::verify_peer | boost::asio::ssl::verify_fail_if_no_peer_cert, ec);
ctx_->set_verify_mode(ssl::verify_peer | ssl::verify_fail_if_no_peer_cert, ec);
MG_ASSERT(!ec, "Setting SSL verification mode failed!");
}
}

View File

@@ -0,0 +1,135 @@
// Copyright 2022 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 <chrono>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <string_view>
#include <thread>
#include <vector>
#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 <boost/system/detail/error_code.hpp>
#include "communication/context.hpp"
#include "communication/v2/pool.hpp"
#include "communication/v2/session.hpp"
#include "utils/spin_lock.hpp"
#include "utils/synchronized.hpp"
namespace memgraph::communication::v2 {
template <class TSession, class TSessionData>
class Listener final : public std::enable_shared_from_this<Listener<TSession, TSessionData>> {
using tcp = boost::asio::ip::tcp;
using SessionHandler = Session<TSession, TSessionData>;
using std::enable_shared_from_this<Listener<TSession, 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)...)};
}
void Start() { DoAccept(); }
bool IsRunning() const noexcept { return alive_.load(std::memory_order_relaxed); }
private:
Listener(boost::asio::io_context &io_context, TSessionData *data, ServerContext *server_context,
tcp::endpoint &endpoint, const std::string_view service_name, const uint64_t inactivity_timeout_sec)
: io_context_(io_context),
data_(data),
server_context_(server_context),
acceptor_(io_context_),
endpoint_{endpoint},
service_name_{service_name},
inactivity_timeout_{inactivity_timeout_sec} {
boost::system::error_code ec;
// Open the acceptor
acceptor_.open(endpoint.protocol(), ec);
if (ec) {
OnError(ec, "open");
return;
}
// Allow address reuse
acceptor_.set_option(boost::asio::socket_base::reuse_address(true), ec);
if (ec) {
OnError(ec, "set_option");
return;
}
// Bind to the server address
acceptor_.bind(endpoint, ec);
if (ec) {
spdlog::error(
utils::MessageWithLink("Cannot bind to socket on endpoint {}.", endpoint, "https://memgr.ph/socket"));
OnError(ec, "bind");
return;
}
acceptor_.listen(boost::asio::socket_base::max_listen_connections, ec);
if (ec) {
OnError(ec, "listen");
return;
}
}
void DoAccept() {
acceptor_.async_accept(io_context_,
[shared_this = shared_from_this()](auto ec, boost::asio::ip::tcp::socket &&socket) {
shared_this->OnAccept(ec, std::move(socket));
});
}
void OnAccept(boost::system::error_code ec, tcp::socket socket) {
if (ec) {
return OnError(ec, "accept");
}
auto session = SessionHandler::Create(std::move(socket), data_, *server_context_, endpoint_, inactivity_timeout_,
service_name_);
session->Start();
DoAccept();
}
void OnError(const boost::system::error_code &ec, const std::string_view what) {
spdlog::error("Listener failed on {}: {}", what, ec.message());
alive_.store(false, std::memory_order_relaxed);
}
boost::asio::io_context &io_context_;
TSessionData *data_;
ServerContext *server_context_;
tcp::acceptor acceptor_;
tcp::endpoint endpoint_;
std::string_view service_name_;
std::chrono::seconds inactivity_timeout_;
std::atomic<bool> alive_;
};
} // namespace memgraph::communication::v2

View File

@@ -0,0 +1,68 @@
// Copyright 2022 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 <cstddef>
#include <thread>
#include <vector>
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include "utils/logging.hpp"
namespace memgraph::communication::v2 {
class IOContextThreadPool final {
private:
using IOContext = boost::asio::io_context;
using IOContextGuard = boost::asio::executor_work_guard<boost::asio::io_context::executor_type>;
public:
explicit IOContextThreadPool(size_t pool_size) : guard_{io_context_.get_executor()}, pool_size_{pool_size} {
MG_ASSERT(pool_size != 0, "Pool size must be greater then 0!");
}
IOContextThreadPool(const IOContextThreadPool &) = delete;
IOContextThreadPool &operator=(const IOContextThreadPool &) = delete;
IOContextThreadPool(IOContextThreadPool &&) = delete;
IOContextThreadPool &operator=(IOContextThreadPool &&) = delete;
~IOContextThreadPool() = default;
void Run() {
background_threads_.reserve(pool_size_);
for (size_t i = 0; i < pool_size_; ++i) {
background_threads_.emplace_back([this]() { io_context_.run(); });
}
running_ = true;
}
void Shutdown() {
io_context_.stop();
running_ = false;
}
void AwaitShutdown() { background_threads_.clear(); }
bool IsRunning() const noexcept { return running_; }
IOContext &GetIOContext() noexcept { return io_context_; }
private:
/// The pool of io_context.
IOContext io_context_;
IOContextGuard guard_;
size_t pool_size_;
std::vector<std::jthread> background_threads_;
bool running_{false};
};
} // namespace memgraph::communication::v2

View File

@@ -0,0 +1,128 @@
// Copyright 2022 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 <algorithm>
#include <cstdint>
#include <iostream>
#include <memory>
#include <optional>
#include <string>
#include <thread>
#include <vector>
#include <fmt/format.h>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/address.hpp>
#include <boost/asio/ip/tcp.hpp>
#include "communication/context.hpp"
#include "communication/init.hpp"
#include "communication/v2/listener.hpp"
#include "communication/v2/pool.hpp"
#include "utils/logging.hpp"
#include "utils/message.hpp"
#include "utils/thread.hpp"
namespace memgraph::communication::v2 {
using Socket = boost::asio::ip::tcp::socket;
using ServerEndpoint = boost::asio::ip::tcp::endpoint;
/**
* Communication server.
*
* Listens for incoming connections on the server port and assigns them to the
* connection listener. The listener and session are implemented using asio
* async model. Currently the implemented model is thread per core model
* opposed to io_context per core. The reasoning for opting for the former model
* is the robustness to the multiple resource demanding queries that can be split
* across multiple threads, and then a single thread would not block io_context,
* unlike in the latter model where it is possible that thread that accepts
* request is being blocked by demanding query.
* All logic is contained within handlers that are being dispatched
* on a single strand per session. The only exception is write which is
* synchronous since the nature of the clients conenction is synchronous as
* well.
*
* Current Server architecture:
* incoming connection -> server -> listener -> session
*
* @tparam TSession the server can handle different Sessions, each session
* represents a different protocol so the same network infrastructure
* can be used for handling different protocols
* @tparam TSessionData the class with objects that will be forwarded to the
* session
*/
template <typename TSession, typename TSessionData>
class Server final {
using ServerHandler = Server<TSession, TSessionData>;
public:
/**
* Constructs and binds server to endpoint, operates on session data and
* invokes workers_count workers
*/
Server(ServerEndpoint &endpoint, TSessionData *session_data, ServerContext *server_context,
const int inactivity_timeout_sec, const std::string_view service_name,
size_t workers_count = std::thread::hardware_concurrency())
: endpoint_{endpoint},
service_name_{service_name},
context_thread_pool_{workers_count},
listener_{Listener<TSession, TSessionData>::Create(context_thread_pool_.GetIOContext(), session_data,
server_context, endpoint_, service_name_,
inactivity_timeout_sec)} {}
~Server() { MG_ASSERT(!IsRunning(), "Server wasn't shutdown properly"); }
Server(const Server &) = delete;
Server(Server &&) = delete;
Server &operator=(const Server &) = delete;
Server &operator=(Server &&) = delete;
const auto &Endpoint() const {
MG_ASSERT(IsRunning(), "You can't get the server endpoint when it's not running!");
return endpoint_;
}
bool Start() {
if (IsRunning()) {
spdlog::error("The server is already running");
return false;
}
listener_->Start();
spdlog::info("{} server is fully armed and operational", service_name_);
spdlog::info("{} listening on {}", service_name_, endpoint_.address());
context_thread_pool_.Run();
return true;
}
void Shutdown() {
context_thread_pool_.Shutdown();
spdlog::info("{} shutting down...", service_name_);
}
void AwaitShutdown() { context_thread_pool_.AwaitShutdown(); }
bool IsRunning() const noexcept { return context_thread_pool_.IsRunning() && listener_->IsRunning(); }
private:
ServerEndpoint endpoint_;
std::string service_name_;
IOContextThreadPool context_thread_pool_;
std::shared_ptr<Listener<TSession, TSessionData>> listener_;
};
} // namespace memgraph::communication::v2

View File

@@ -0,0 +1,508 @@
// Copyright 2022 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 <chrono>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <deque>
#include <functional>
#include <memory>
#include <string>
#include <string_view>
#include <utility>
#include <variant>
#include <spdlog/spdlog.h>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/read.hpp>
#include <boost/asio/socket_base.hpp>
#include <boost/asio/ssl/stream.hpp>
#include <boost/asio/ssl/stream_base.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/system_context.hpp>
#include <boost/asio/write.hpp>
#include <boost/beast/core/tcp_stream.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/beast/websocket/rfc6455.hpp>
#include <boost/system/detail/error_code.hpp>
#include "communication/context.hpp"
#include "communication/exceptions.hpp"
#include "utils/logging.hpp"
#include "utils/variant_helpers.hpp"
namespace memgraph::communication::v2 {
/**
* This is used to provide input to user Sessions. All Sessions used with the
* network stack should use this class as their input stream.
*/
using InputStream = communication::Buffer::ReadEnd;
using tcp = boost::asio::ip::tcp;
/**
* This is used to provide output from user Sessions. All Sessions used with the
* network stack should use this class for their output stream.
*/
class OutputStream final {
public:
explicit OutputStream(std::function<bool(const uint8_t *, size_t, bool)> write_function)
: write_function_(write_function) {}
OutputStream(const OutputStream &) = delete;
OutputStream(OutputStream &&) = delete;
OutputStream &operator=(const OutputStream &) = delete;
OutputStream &operator=(OutputStream &&) = delete;
~OutputStream() = default;
bool Write(const uint8_t *data, size_t len, bool have_more = false) { return write_function_(data, len, have_more); }
bool Write(const std::string &str, bool have_more = false) {
return Write(reinterpret_cast<const uint8_t *>(str.data()), str.size(), have_more);
}
private:
std::function<bool(const uint8_t *, size_t, bool)> write_function_;
};
/**
* This class is used internally in the communication stack to handle all user
* Websocket Sessions. It handles socket ownership, inactivity timeout and protocol
* wrapping.
*/
template <typename TSession, typename TSessionData>
class WebsocketSession : public std::enable_shared_from_this<WebsocketSession<TSession, TSessionData>> {
using WebSocket = boost::beast::websocket::stream<boost::beast::tcp_stream>;
using std::enable_shared_from_this<WebsocketSession<TSession, TSessionData>>::shared_from_this;
public:
template <typename... Args>
static std::shared_ptr<WebsocketSession> Create(Args &&...args) {
return std::shared_ptr<WebsocketSession>(new WebsocketSession(std::forward<Args>(args)...));
}
// Start the asynchronous accept operation
template <class Body, class Allocator>
void DoAccept(boost::beast::http::request<Body, boost::beast::http::basic_fields<Allocator>> req) {
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));
boost::asio::socket_base::keep_alive option(true);
// Set a decorator to change the Server of the handshake
ws_.set_option(boost::beast::websocket::stream_base::decorator([](boost::beast::websocket::response_type &res) {
res.set(boost::beast::http::field::server, std::string("Memgraph Bolt WS"));
res.set(boost::beast::http::field::sec_websocket_protocol, "binary");
}));
ws_.binary(true);
// Accept the websocket handshake
ws_.async_accept(
req, boost::asio::bind_executor(strand_, std::bind_front(&WebsocketSession::OnAccept, shared_from_this())));
}
bool Write(const uint8_t *data, size_t len) {
if (!IsConnected()) {
return false;
}
boost::system::error_code ec;
ws_.write(boost::asio::buffer(data, len), ec);
if (ec) {
OnError(ec, "write");
return false;
}
return true;
}
private:
// Take ownership of the socket
explicit WebsocketSession(tcp::socket &&socket, TSessionData *data, tcp::endpoint endpoint,
std::string_view service_name)
: ws_(std::move(socket)),
strand_{boost::asio::make_strand(ws_.get_executor())},
output_stream_([this](const uint8_t *data, size_t len, bool /*have_more*/) { return Write(data, len); }),
session_(data, endpoint, input_buffer_.read_end(), &output_stream_),
endpoint_{endpoint},
remote_endpoint_{ws_.next_layer().socket().remote_endpoint()},
service_name_{service_name} {}
void OnAccept(boost::beast::error_code ec) {
if (ec) {
return OnError(ec, "accept");
}
// Read a message
DoRead();
}
void DoRead() {
if (!IsConnected()) {
return;
}
// Read a message into our buffer
auto buffer = input_buffer_.write_end()->Allocate();
ws_.async_read_some(
boost::asio::buffer(buffer.data, buffer.len),
boost::asio::bind_executor(strand_, std::bind_front(&WebsocketSession::OnRead, shared_from_this())));
}
void OnRead(const boost::system::error_code &ec, [[maybe_unused]] const size_t bytes_transferred) {
// This indicates that the WebsocketSession was closed
if (ec == boost::beast::websocket::error::closed) {
return;
}
if (ec) {
OnError(ec, "read");
}
input_buffer_.write_end()->Written(bytes_transferred);
try {
session_.Execute();
DoRead();
} catch (const SessionClosedException &e) {
spdlog::info("{} client {}:{} closed the connection.", service_name_, remote_endpoint_.address(),
remote_endpoint_.port());
DoClose();
} catch (const std::exception &e) {
spdlog::error(
"Exception was thrown while processing event in {} session "
"associated with {}:{}",
service_name_, remote_endpoint_.address(), remote_endpoint_.port());
spdlog::debug("Exception message: {}", e.what());
DoClose();
}
}
void OnError(const boost::system::error_code &ec, const std::string_view action) {
spdlog::error("Websocket Bolt session error: {} on {}", ec.message(), action);
DoClose();
}
void DoClose() {
ws_.async_close(
boost::beast::websocket::close_code::normal,
boost::asio::bind_executor(
strand_, [shared_this = shared_from_this()](boost::beast::error_code ec) { shared_this->OnClose(ec); }));
}
void OnClose(const boost::system::error_code &ec) {
if (!IsConnected()) {
return;
}
if (ec) {
return OnError(ec, "close");
}
}
bool IsConnected() const { return ws_.is_open() && execution_active_; }
WebSocket ws_;
boost::asio::strand<WebSocket::executor_type> strand_;
communication::Buffer input_buffer_;
OutputStream output_stream_;
TSession session_;
tcp::endpoint endpoint_;
tcp::endpoint remote_endpoint_;
std::string_view service_name_;
bool execution_active_{false};
};
/**
* This class is used internally in the communication stack to handle all user
* Sessions. It handles socket ownership, inactivity timeout and protocol
* wrapping.
*/
template <typename TSession, typename TSessionData>
class Session final : public std::enable_shared_from_this<Session<TSession, TSessionData>> {
using TCPSocket = tcp::socket;
using SSLSocket = boost::asio::ssl::stream<TCPSocket>;
using std::enable_shared_from_this<Session<TSession, 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)...));
}
Session(const Session &) = delete;
Session(Session &&) = delete;
Session &operator=(const Session &) = delete;
Session &operator=(Session &&) = delete;
~Session() = default;
bool Start() {
if (execution_active_) {
return false;
}
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_)) {
boost::asio::dispatch(strand_, [shared_this = shared_from_this()] { shared_this->DoHandshake(); });
} else {
boost::asio::dispatch(strand_, [shared_this = shared_from_this()] { shared_this->DoRead(); });
}
return true;
}
bool Write(const uint8_t *data, size_t len, bool have_more = false) {
if (!IsConnected()) {
return false;
}
return std::visit(
utils::Overloaded{[shared_this = shared_from_this(), data, len, have_more](TCPSocket &socket) mutable {
boost::system::error_code ec;
while (len > 0) {
const auto sent = socket.send(boost::asio::buffer(data, len),
MSG_NOSIGNAL | (have_more ? MSG_MORE : 0), ec);
if (ec) {
shared_this->OnError(ec);
return false;
}
data += sent;
len -= sent;
}
return true;
},
[shared_this = shared_from_this(), data, len](SSLSocket &socket) mutable {
boost::system::error_code ec;
while (len > 0) {
const auto sent = socket.write_some(boost::asio::buffer(data, len), ec);
if (ec) {
shared_this->OnError(ec);
return false;
}
data += sent;
len -= sent;
}
return true;
}},
socket_);
}
bool IsConnected() const {
return std::visit([this](const auto &socket) { return execution_active_ && socket.lowest_layer().is_open(); },
socket_);
}
private:
explicit Session(tcp::socket &&socket, TSessionData *data, ServerContext &server_context, tcp::endpoint endpoint,
const std::chrono::seconds inactivity_timeout_sec, std::string_view service_name)
: socket_(CreateSocket(std::move(socket), server_context)),
strand_{boost::asio::make_strand(GetExecutor())},
output_stream_([this](const uint8_t *data, size_t len, bool have_more) { return Write(data, len, have_more); }),
session_(data, endpoint, input_buffer_.read_end(), &output_stream_),
data_{data},
endpoint_{endpoint},
remote_endpoint_{GetRemoteEndpoint()},
service_name_{service_name},
timeout_seconds_(inactivity_timeout_sec),
timeout_timer_(GetExecutor()) {
ExecuteForSocket([](auto &&socket) {
socket.lowest_layer().set_option(tcp::no_delay(true)); // enable PSH
socket.lowest_layer().set_option(boost::asio::socket_base::keep_alive(true)); // enable SO_KEEPALIVE
socket.lowest_layer().non_blocking(false);
});
timeout_timer_.expires_at(boost::asio::steady_timer::time_point::max());
spdlog::info("Accepted a connection from {}:", service_name_, remote_endpoint_.address(), remote_endpoint_.port());
}
void DoRead() {
if (!IsConnected()) {
return;
}
timeout_timer_.expires_after(timeout_seconds_);
ExecuteForSocket([this](auto &&socket) {
auto buffer = input_buffer_.write_end()->Allocate();
socket.async_read_some(
boost::asio::buffer(buffer.data, buffer.len),
boost::asio::bind_executor(strand_, std::bind_front(&Session::OnRead, shared_from_this())));
});
}
bool IsWebsocketUpgrade(boost::beast::http::request_parser<boost::beast::http::string_body> &parser) {
boost::system::error_code error_code_parsing;
parser.put(boost::asio::buffer(input_buffer_.read_end()->data(), input_buffer_.read_end()->size()),
error_code_parsing);
if (error_code_parsing) {
return false;
}
return boost::beast::websocket::is_upgrade(parser.get());
}
void OnRead(const boost::system::error_code &ec, const size_t bytes_transferred) {
if (ec) {
return OnError(ec);
}
input_buffer_.write_end()->Written(bytes_transferred);
// Can be a websocket connection only on the first read, since it is not
// expected from clients to upgrade from tcp to websocket
if (!has_received_msg_) {
has_received_msg_ = true;
boost::beast::http::request_parser<boost::beast::http::string_body> parser;
if (IsWebsocketUpgrade(parser)) {
spdlog::info("Switching {} to websocket connection", remote_endpoint_);
if (std::holds_alternative<TCPSocket>(socket_)) {
auto sock = std::get<TCPSocket>(std::move(socket_));
WebsocketSession<TSession, TSessionData>::Create(std::move(sock), data_, endpoint_, service_name_)
->DoAccept(parser.release());
execution_active_ = false;
return;
}
spdlog::error("Error while upgrading connection to websocket");
DoShutdown();
}
}
try {
session_.Execute();
DoRead();
} catch (const SessionClosedException &e) {
spdlog::info("{} client {}:{} closed the connection.", service_name_, remote_endpoint_.address(),
remote_endpoint_.port());
DoShutdown();
} catch (const std::exception &e) {
spdlog::error(
"Exception was thrown while processing event in {} session "
"associated with {}:{}",
service_name_, remote_endpoint_.address(), remote_endpoint_.port());
spdlog::debug("Exception message: {}", e.what());
DoShutdown();
}
}
void OnError(const boost::system::error_code &ec) {
if (ec == boost::asio::error::operation_aborted) {
return;
}
if (ec == boost::asio::error::eof) {
spdlog::info("Session closed by peer");
} else {
spdlog::error("Session error: {}", ec.message());
}
DoShutdown();
}
void DoShutdown() {
if (!IsConnected()) {
return;
}
execution_active_ = false;
timeout_timer_.cancel();
ExecuteForSocket([](auto &socket) {
boost::system::error_code ec;
auto &lowest_layer = socket.lowest_layer();
lowest_layer.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
if (ec) {
spdlog::error("Session shutdown failed: {}", ec.what());
}
lowest_layer.close();
});
}
void DoHandshake() {
if (!IsConnected()) {
return;
}
if (auto *socket = std::get_if<SSLSocket>(&socket_); socket) {
socket->async_handshake(
boost::asio::ssl::stream_base::server,
boost::asio::bind_executor(strand_, std::bind_front(&Session::OnHandshake, shared_from_this())));
}
}
void OnHandshake(const boost::system::error_code &ec) {
if (ec) {
return OnError(ec);
}
DoRead();
}
void OnClose(const boost::system::error_code &ec) {
if (ec) {
return OnError(ec);
}
}
void OnTimeout() {
if (!IsConnected()) {
return;
}
// Check whether the deadline has passed. We compare the deadline against
// the current time since a new asynchronous operation may have moved the
// deadline before this actor had a chance to run.
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_);
DoShutdown();
} else {
// Put the actor back to sleep.
timeout_timer_.async_wait(
boost::asio::bind_executor(strand_, std::bind(&Session::OnTimeout, shared_from_this())));
}
}
std::variant<TCPSocket, SSLSocket> CreateSocket(tcp::socket &&socket, ServerContext &context) {
if (context.use_ssl()) {
ssl_context_.emplace(context.context_clone());
return SSLSocket{std::move(socket), *ssl_context_};
}
return TCPSocket{std::move(socket)};
}
auto GetExecutor() {
return std::visit(utils::Overloaded{[](auto &&socket) { return socket.get_executor(); }}, socket_);
}
auto GetRemoteEndpoint() const {
return std::visit(utils::Overloaded{[](const auto &socket) { return socket.lowest_layer().remote_endpoint(); }},
socket_);
}
template <typename F>
decltype(auto) ExecuteForSocket(F &&fun) {
return std::visit(utils::Overloaded{std::forward<F>(fun)}, socket_);
}
std::variant<TCPSocket, SSLSocket> socket_;
std::optional<std::reference_wrapper<boost::asio::ssl::context>> ssl_context_;
boost::asio::strand<tcp::socket::executor_type> strand_;
communication::Buffer input_buffer_;
OutputStream output_stream_;
TSession session_;
TSessionData *data_;
tcp::endpoint endpoint_;
tcp::endpoint remote_endpoint_;
std::string_view service_name_;
std::chrono::seconds timeout_seconds_;
boost::asio::steady_timer timeout_timer_;
bool execution_active_{false};
bool has_received_msg_{false};
};
} // namespace memgraph::communication::v2

View File

@@ -11,8 +11,6 @@
#pragma once
#define BOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT
#include <list>
#include <memory>

View File

@@ -44,10 +44,10 @@ class QuoteEscapeFormatter : public spdlog::custom_flag_formatter {
void format(const spdlog::details::log_msg &msg, const std::tm & /*time*/, spdlog::memory_buf_t &dest) override {
for (const auto c : msg.payload) {
if (c == '"') {
constexpr std::string_view escaped_quote = "\\\"";
static constexpr std::string_view escaped_quote = "\\\"";
dest.append(escaped_quote.data(), escaped_quote.data() + escaped_quote.size());
} else if (c == '\n') {
constexpr std::string_view escaped_newline = "\\n";
static constexpr std::string_view escaped_newline = "\\n";
dest.append(escaped_newline.data(), escaped_newline.data() + escaped_newline.size());
} else {
dest.push_back(c);

View File

@@ -11,8 +11,6 @@
#pragma once
#define BOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT
#include <thread>
#include <spdlog/sinks/base_sink.h>

View File

@@ -11,8 +11,6 @@
#pragma once
#define BOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT
#include <deque>
#include <memory>
#include <optional>

View File

@@ -16,10 +16,10 @@
namespace memgraph::integrations {
constexpr int64_t kDefaultCheckBatchLimit{1};
constexpr std::chrono::milliseconds kDefaultCheckTimeout{30000};
constexpr std::chrono::milliseconds kMinimumInterval{1};
constexpr int64_t kMinimumSize{1};
inline constexpr int64_t kDefaultCheckBatchLimit{1};
inline constexpr std::chrono::milliseconds kDefaultCheckTimeout{30000};
inline constexpr std::chrono::milliseconds kMinimumInterval{1};
inline constexpr int64_t kMinimumSize{1};
const std::string kReducted{"<REDUCTED>"};
} // namespace memgraph::integrations

View File

@@ -185,8 +185,10 @@ Consumer::Consumer(ConsumerInfo info, ConsumerFunction consumer_function)
std::inserter(topic_names_from_metadata, topic_names_from_metadata.begin()),
[](const auto topic_metadata) { return topic_metadata->topic(); });
constexpr size_t max_topic_name_length = 249;
constexpr auto is_valid_topic_name = [](const auto c) { return std::isalnum(c) || c == '.' || c == '_' || c == '-'; };
static constexpr size_t max_topic_name_length = 249;
static constexpr auto is_valid_topic_name = [](const auto c) {
return std::isalnum(c) || c == '.' || c == '_' || c == '-';
};
for (const auto &topic_name : info_.topics) {
if (topic_name.size() > max_topic_name_length ||
@@ -351,7 +353,7 @@ void Consumer::StartConsuming() {
}
thread_ = std::thread([this] {
constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize();
static constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize();
const auto full_thread_name = "Cons#" + info_.consumer_name;
utils::ThreadSetName(full_thread_name.substr(0, kMaxThreadNameSize));

View File

@@ -231,7 +231,7 @@ void Consumer::StartConsuming() {
is_running_.store(true);
thread_ = std::thread([this] {
constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize();
static constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize();
const auto full_thread_name = "Cons#" + info_.consumer_name;
utils::ThreadSetName(full_thread_name.substr(0, kMaxThreadNameSize));

View File

@@ -9,34 +9,52 @@
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include <netdb.h>
#include <cstring>
#include "io/network/addrinfo.hpp"
#include <concepts>
#include <iterator>
#include "io/network/network_error.hpp"
namespace memgraph::io::network {
AddrInfo::AddrInfo(struct addrinfo *info) : info(info) {}
static_assert(std::forward_iterator<AddrInfo::Iterator> && std::equality_comparable<AddrInfo::Iterator>);
AddrInfo::~AddrInfo() { freeaddrinfo(info); }
AddrInfo AddrInfo::Get(const char *addr, const char *port) {
struct addrinfo hints;
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = AF_UNSPEC; // IPv4 and IPv6
hints.ai_socktype = SOCK_STREAM; // TCP socket
hints.ai_flags = AI_PASSIVE;
struct addrinfo *result;
auto status = getaddrinfo(addr, port, &hints, &result);
AddrInfo::AddrInfo(const Endpoint &endpoint) : AddrInfo(endpoint.address, endpoint.port) {}
AddrInfo::AddrInfo(const std::string &addr, uint16_t port) : info_{nullptr, nullptr} {
addrinfo hints{
.ai_flags = AI_PASSIVE,
.ai_family = AF_UNSPEC, // IPv4 and IPv6
.ai_socktype = SOCK_STREAM // TCP socket
};
addrinfo *info = nullptr;
auto status = getaddrinfo(addr.c_str(), std::to_string(port).c_str(), &hints, &info);
if (status != 0) throw NetworkError(gai_strerror(status));
return AddrInfo(result);
info_ = std::unique_ptr<addrinfo, decltype(&freeaddrinfo)>(info, &freeaddrinfo);
}
AddrInfo::operator struct addrinfo *() { return info; }
AddrInfo::Iterator::Iterator(addrinfo *p) noexcept : ptr_(p) {}
AddrInfo::Iterator::reference AddrInfo::Iterator::operator*() const noexcept { return *ptr_; }
AddrInfo::Iterator::pointer AddrInfo::Iterator::operator->() const noexcept { return ptr_; }
// NOLINTNEXTLINE(cert-dcl21-cpp)
AddrInfo::Iterator AddrInfo::Iterator::operator++(int) noexcept {
auto it = *this;
++(*this);
return it;
}
AddrInfo::Iterator &AddrInfo::Iterator::operator++() noexcept {
ptr_ = ptr_->ai_next;
return *this;
}
bool operator==(const AddrInfo::Iterator &lhs, const AddrInfo::Iterator &rhs) noexcept { return lhs.ptr_ == rhs.ptr_; };
bool operator!=(const AddrInfo::Iterator &lhs, const AddrInfo::Iterator &rhs) noexcept { return !(lhs == rhs); };
void swap(AddrInfo::Iterator &lhs, AddrInfo::Iterator &rhs) noexcept { std::swap(lhs.ptr_, rhs.ptr_); };
} // namespace memgraph::io::network

View File

@@ -11,6 +11,14 @@
#pragma once
#include <netdb.h>
#include <iterator>
#include <memory>
#include <string>
#include "io/network/endpoint.hpp"
namespace memgraph::io::network {
/**
@@ -18,16 +26,38 @@ namespace memgraph::io::network {
* see: man 3 getaddrinfo
*/
class AddrInfo {
explicit AddrInfo(struct addrinfo *info);
public:
~AddrInfo();
struct Iterator {
using iterator_category = std::forward_iterator_tag;
using value_type = addrinfo;
using difference_type = std::ptrdiff_t;
using pointer = addrinfo *;
using reference = addrinfo &;
static AddrInfo Get(const char *addr, const char *port);
Iterator() = default;
Iterator(const Iterator &) = default;
explicit Iterator(addrinfo *p) noexcept;
Iterator &operator=(const Iterator &) = default;
reference operator*() const noexcept;
pointer operator->() const noexcept;
Iterator operator++(int) noexcept;
Iterator &operator++() noexcept;
operator struct addrinfo *();
friend bool operator==(const Iterator &lhs, const Iterator &rhs) noexcept;
friend bool operator!=(const Iterator &lhs, const Iterator &rhs) noexcept;
friend void swap(Iterator &lhs, Iterator &rhs) noexcept;
private:
addrinfo *ptr_{nullptr};
};
AddrInfo(const std::string &addr, uint16_t port);
explicit AddrInfo(const Endpoint &endpoint);
auto begin() const noexcept { return Iterator(info_.get()); }
auto end() const noexcept { return Iterator{nullptr}; }
private:
struct addrinfo *info;
std::unique_ptr<addrinfo, void (*)(addrinfo *)> info_;
};
} // namespace memgraph::io::network

View File

@@ -9,39 +9,25 @@
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "io/network/socket.hpp"
#include <cstdio>
#include <cstring>
#include <iostream>
#include <stdexcept>
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <poll.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include "io/network/addrinfo.hpp"
#include "io/network/socket.hpp"
#include "utils/likely.hpp"
#include "utils/logging.hpp"
namespace memgraph::io::network {
Socket::Socket(Socket &&other) {
socket_ = other.socket_;
endpoint_ = std::move(other.endpoint_);
Socket::Socket(Socket &&other) noexcept : socket_(other.socket_), endpoint_(std::move(other.endpoint_)) {
other.socket_ = -1;
}
Socket &Socket::operator=(Socket &&other) {
Socket &Socket::operator=(Socket &&other) noexcept {
if (this != &other) {
if (socket_ != -1) close(socket_);
socket_ = other.socket_;
endpoint_ = std::move(other.endpoint_);
other.socket_ = -1;
@@ -49,9 +35,8 @@ Socket &Socket::operator=(Socket &&other) {
return *this;
}
Socket::~Socket() {
if (socket_ == -1) return;
close(socket_);
Socket::~Socket() noexcept {
if (socket_ != -1) close(socket_);
}
void Socket::Close() {
@@ -70,33 +55,27 @@ bool Socket::IsOpen() const { return socket_ != -1; }
bool Socket::Connect(const Endpoint &endpoint) {
if (socket_ != -1) return false;
auto info = AddrInfo::Get(endpoint.address.c_str(), std::to_string(endpoint.port).c_str());
for (struct addrinfo *it = info; it != nullptr; it = it->ai_next) {
int sfd = socket(it->ai_family, it->ai_socktype, it->ai_protocol);
for (const auto &it : AddrInfo{endpoint}) {
int sfd = socket(it.ai_family, it.ai_socktype, it.ai_protocol);
if (sfd == -1) continue;
if (connect(sfd, it->ai_addr, it->ai_addrlen) == 0) {
if (connect(sfd, it.ai_addr, it.ai_addrlen) == 0) {
socket_ = sfd;
endpoint_ = endpoint;
break;
} else {
// If the connect failed close the file descriptor to prevent file
// descriptors being leaked
close(sfd);
}
// If the connect failed close the file descriptor to prevent file
// descriptors being leaked
close(sfd);
}
if (socket_ == -1) return false;
return true;
return !(socket_ == -1);
}
bool Socket::Bind(const Endpoint &endpoint) {
if (socket_ != -1) return false;
auto info = AddrInfo::Get(endpoint.address.c_str(), std::to_string(endpoint.port).c_str());
for (struct addrinfo *it = info; it != nullptr; it = it->ai_next) {
int sfd = socket(it->ai_family, it->ai_socktype, it->ai_protocol);
for (const auto &it : AddrInfo{endpoint}) {
int sfd = socket(it.ai_family, it.ai_socktype, it.ai_protocol);
if (sfd == -1) continue;
int on = 1;
@@ -107,14 +86,13 @@ bool Socket::Bind(const Endpoint &endpoint) {
continue;
}
if (bind(sfd, it->ai_addr, it->ai_addrlen) == 0) {
if (bind(sfd, it.ai_addr, it.ai_addrlen) == 0) {
socket_ = sfd;
break;
} else {
// If the bind failed close the file descriptor to prevent file
// descriptors being leaked
close(sfd);
}
// If the bind failed close the file descriptor to prevent file
// descriptors being leaked
close(sfd);
}
if (socket_ == -1) return false;
@@ -122,7 +100,7 @@ bool Socket::Bind(const Endpoint &endpoint) {
// detect bound port, used when the server binds to a random port
struct sockaddr_in6 portdata;
socklen_t portdatalen = sizeof(portdata);
if (getsockname(socket_, (struct sockaddr *)&portdata, &portdatalen) < 0) {
if (getsockname(socket_, reinterpret_cast<sockaddr *>(&portdata), &portdatalen) < 0) {
// If the getsockname failed close the file descriptor to prevent file
// descriptors being leaked
close(socket_);
@@ -136,36 +114,35 @@ bool Socket::Bind(const Endpoint &endpoint) {
}
void Socket::SetNonBlocking() {
int flags = fcntl(socket_, F_GETFL, 0);
const unsigned flags = fcntl(socket_, F_GETFL);
constexpr unsigned o_nonblock = O_NONBLOCK;
MG_ASSERT(flags != -1, "Can't get socket mode");
flags |= O_NONBLOCK;
MG_ASSERT(fcntl(socket_, F_SETFL, flags) != -1, "Can't set socket nonblocking");
MG_ASSERT(fcntl(socket_, F_SETFL, flags | o_nonblock) != -1, "Can't set socket nonblocking");
}
void Socket::SetKeepAlive() {
int optval = 1;
socklen_t optlen = sizeof(optval);
MG_ASSERT(!setsockopt(socket_, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen), "Can't set socket keep alive");
MG_ASSERT(!setsockopt(socket_, SOL_SOCKET, SO_KEEPALIVE, &optval, sizeof(optval)), "Can't set socket keep alive");
optval = 20; // wait 20s before sending keep-alive packets
MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPIDLE, (void *)&optval, optlen), "Can't set socket keep alive");
MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPIDLE, (void *)&optval, sizeof(optval)),
"Can't set socket keep alive");
optval = 4; // 4 keep-alive packets must fail to close
MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPCNT, (void *)&optval, optlen), "Can't set socket keep alive");
MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPCNT, (void *)&optval, sizeof(optval)), "Can't set socket keep alive");
optval = 15; // send keep-alive packets every 15s
MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPINTVL, (void *)&optval, optlen), "Can't set socket keep alive");
MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPINTVL, (void *)&optval, sizeof(optval)),
"Can't set socket keep alive");
}
void Socket::SetNoDelay() {
int optval = 1;
socklen_t optlen = sizeof(optval);
MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_NODELAY, (void *)&optval, optlen), "Can't set socket no delay");
MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_NODELAY, (void *)&optval, sizeof(optval)), "Can't set socket no delay");
}
void Socket::SetTimeout(long sec, long usec) {
// NOLINTNEXTLINE(readability-make-member-function-const)
void Socket::SetTimeout(int64_t sec, int64_t usec) {
struct timeval tv;
tv.tv_sec = sec;
tv.tv_usec = usec;
@@ -176,7 +153,7 @@ void Socket::SetTimeout(long sec, long usec) {
}
int Socket::ErrorStatus() const {
int optval;
int optval = 0;
socklen_t optlen = sizeof(optval);
auto status = getsockopt(socket_, SOL_SOCKET, SO_ERROR, &optval, &optlen);
MG_ASSERT(!status, "getsockopt failed");
@@ -189,21 +166,22 @@ std::optional<Socket> Socket::Accept() {
sockaddr_storage addr;
socklen_t addr_size = sizeof addr;
char addr_decoded[INET6_ADDRSTRLEN];
void *addr_src;
unsigned short port;
int sfd = accept(socket_, (struct sockaddr *)&addr, &addr_size);
int sfd = accept(socket_, reinterpret_cast<sockaddr *>(&addr), &addr_size);
if (UNLIKELY(sfd == -1)) return std::nullopt;
void *addr_src = nullptr;
uint16_t port = 0;
if (addr.ss_family == AF_INET) {
addr_src = (void *)&(((sockaddr_in *)&addr)->sin_addr);
port = ntohs(((sockaddr_in *)&addr)->sin_port);
addr_src = &reinterpret_cast<sockaddr_in &>(addr).sin_addr;
port = ntohs(reinterpret_cast<sockaddr_in &>(addr).sin_port);
} else {
addr_src = (void *)&(((sockaddr_in6 *)&addr)->sin6_addr);
port = ntohs(((sockaddr_in6 *)&addr)->sin6_port);
addr_src = &reinterpret_cast<sockaddr_in6 &>(addr).sin6_addr;
port = ntohs(reinterpret_cast<sockaddr_in6 &>(addr).sin6_port);
}
inet_ntop(addr.ss_family, addr_src, addr_decoded, INET6_ADDRSTRLEN);
inet_ntop(addr.ss_family, addr_src, addr_decoded, sizeof(addr_decoded));
Endpoint endpoint(addr_decoded, port);
@@ -213,9 +191,11 @@ std::optional<Socket> Socket::Accept() {
bool Socket::Write(const uint8_t *data, size_t len, bool have_more) {
// MSG_NOSIGNAL is here to disable raising a SIGPIPE signal when a
// connection dies mid-write, the socket will only return an EPIPE error.
int flags = MSG_NOSIGNAL | (have_more ? MSG_MORE : 0);
constexpr unsigned msg_nosignal = MSG_NOSIGNAL;
constexpr unsigned msg_more = MSG_MORE;
const unsigned flags = msg_nosignal | (have_more ? msg_more : 0);
while (len > 0) {
auto written = send(socket_, data, len, flags);
auto written = send(socket_, data, len, static_cast<int>(flags));
if (written == -1) {
if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
// Terminal error, return failure.
@@ -253,7 +233,8 @@ bool Socket::WaitForReadyRead() {
// event occurs.
int ret = poll(&p, 1, -1);
if (ret < 1) return false;
return p.revents & POLLIN;
constexpr unsigned pollin = POLLIN;
return static_cast<unsigned>(p.revents) & pollin;
}
bool Socket::WaitForReadyWrite() {
@@ -265,7 +246,8 @@ bool Socket::WaitForReadyWrite() {
// event occurs.
int ret = poll(&p, 1, -1);
if (ret < 1) return false;
return p.revents & POLLOUT;
constexpr unsigned pollout = POLLOUT;
return static_cast<unsigned>(p.revents) & pollout;
}
} // namespace memgraph::io::network

View File

@@ -27,12 +27,12 @@ namespace memgraph::io::network {
*/
class Socket {
public:
Socket() = default;
Socket() noexcept = default;
Socket(const Socket &) = delete;
Socket &operator=(const Socket &) = delete;
Socket(Socket &&);
Socket &operator=(Socket &&);
~Socket();
Socket(Socket &&) noexcept;
Socket &operator=(Socket &&) noexcept;
~Socket() noexcept;
/**
* Closes the socket if it is open.
@@ -118,7 +118,7 @@ class Socket {
* @param sec timeout seconds value
* @param usec timeout microseconds value
*/
void SetTimeout(long sec, long usec);
void SetTimeout(int64_t sec, int64_t usec);
/**
* Checks if there are any errors on a socket. Returns 0 if there are none.

View File

@@ -81,8 +81,8 @@
#include "communication/bolt/v1/exceptions.hpp"
#include "communication/bolt/v1/session.hpp"
#include "communication/init.hpp"
#include "communication/server.hpp"
#include "communication/session.hpp"
#include "communication/v2/server.hpp"
#include "communication/v2/session.hpp"
#include "glue/communication.hpp"
#include "auth/auth.hpp"
@@ -252,6 +252,11 @@ DEFINE_double(query_execution_timeout_sec, 600,
"Maximum allowed query execution time. Queries exceeding this "
"limit will be aborted. Value of 0 means no limit.");
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_uint64(replication_replica_check_frequency_sec, 1,
"The time duration between two replica checks/pings. If < 1, replicas will NOT be checked at all. NOTE: "
"The MAIN instance allocates a new thread for each REPLICA.");
// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_uint64(
memory_limit, 0,
@@ -260,7 +265,7 @@ DEFINE_uint64(
namespace {
using namespace std::literals;
constexpr std::array isolation_level_mappings{
inline constexpr std::array isolation_level_mappings{
std::pair{"SNAPSHOT_ISOLATION"sv, memgraph::storage::IsolationLevel::SNAPSHOT_ISOLATION},
std::pair{"READ_COMMITTED"sv, memgraph::storage::IsolationLevel::READ_COMMITTED},
std::pair{"READ_UNCOMMITTED"sv, memgraph::storage::IsolationLevel::READ_UNCOMMITTED}};
@@ -348,7 +353,7 @@ DEFINE_bool(also_log_to_stderr, false, "Log messages go to stderr in addition to
DEFINE_string(log_file, "", "Path to where the log should be stored.");
namespace {
constexpr std::array log_level_mappings{
inline constexpr std::array log_level_mappings{
std::pair{"TRACE"sv, spdlog::level::trace}, std::pair{"DEBUG"sv, spdlog::level::debug},
std::pair{"INFO"sv, spdlog::level::info}, std::pair{"WARNING"sv, spdlog::level::warn},
std::pair{"ERROR"sv, spdlog::level::err}, std::pair{"CRITICAL"sv, spdlog::level::critical}};
@@ -385,7 +390,7 @@ spdlog::level::level_enum ParseLogLevel() {
}
// 5 weeks * 7 days
constexpr auto log_retention_count = 35;
inline constexpr auto log_retention_count = 35;
void CreateLoggerFromSink(const auto &sinks, const auto log_level) {
auto logger = std::make_shared<spdlog::logger>("memgraph_log", sinks.begin(), sinks.end());
logger->set_level(log_level);
@@ -456,7 +461,7 @@ struct SessionData {
#endif
};
constexpr std::string_view default_user_role_regex = "[a-zA-Z0-9_.+-@]+";
inline constexpr std::string_view default_user_role_regex = "[a-zA-Z0-9_.+-@]+";
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_string(auth_user_or_role_name_regex, default_user_role_regex.data(),
"Set to the regular expression that each user or role name must fulfill.");
@@ -842,13 +847,14 @@ class AuthChecker final : public memgraph::query::AuthChecker {
memgraph::utils::Synchronized<memgraph::auth::Auth, memgraph::utils::WritePrioritizedRWLock> *auth_;
};
class BoltSession final : public memgraph::communication::bolt::Session<memgraph::communication::InputStream,
memgraph::communication::OutputStream> {
class BoltSession final : public memgraph::communication::bolt::Session<memgraph::communication::v2::InputStream,
memgraph::communication::v2::OutputStream> {
public:
BoltSession(SessionData *data, const memgraph::io::network::Endpoint &endpoint,
memgraph::communication::InputStream *input_stream, memgraph::communication::OutputStream *output_stream)
: memgraph::communication::bolt::Session<memgraph::communication::InputStream,
memgraph::communication::OutputStream>(input_stream, output_stream),
BoltSession(SessionData *data, const memgraph::communication::v2::ServerEndpoint &endpoint,
memgraph::communication::v2::InputStream *input_stream,
memgraph::communication::v2::OutputStream *output_stream)
: memgraph::communication::bolt::Session<memgraph::communication::v2::InputStream,
memgraph::communication::v2::OutputStream>(input_stream, output_stream),
db_(data->db),
interpreter_(data->interpreter_context),
auth_(data->auth),
@@ -858,8 +864,8 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
endpoint_(endpoint) {
}
using memgraph::communication::bolt::Session<memgraph::communication::InputStream,
memgraph::communication::OutputStream>::TEncoder;
using memgraph::communication::bolt::Session<memgraph::communication::v2::InputStream,
memgraph::communication::v2::OutputStream>::TEncoder;
void BeginTransaction() override { interpreter_.BeginTransaction(); }
@@ -877,7 +883,8 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
}
#ifdef MG_ENTERPRISE
if (memgraph::utils::license::global_license_checker.IsValidLicenseFast()) {
audit_log_->Record(endpoint_.address, user_ ? *username : "", query, memgraph::storage::PropertyValue(params_pv));
audit_log_->Record(endpoint_.address().to_string(), user_ ? *username : "", query,
memgraph::storage::PropertyValue(params_pv));
}
#endif
try {
@@ -996,10 +1003,10 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
#ifdef MG_ENTERPRISE
memgraph::audit::Log *audit_log_;
#endif
memgraph::io::network::Endpoint endpoint_;
memgraph::communication::v2::ServerEndpoint endpoint_;
};
using ServerT = memgraph::communication::Server<BoltSession, SessionData>;
using ServerT = memgraph::communication::v2::Server<BoltSession, SessionData>;
using memgraph::communication::ServerContext;
// Needed to correctly handle memgraph destruction from a signal handler.
@@ -1068,6 +1075,22 @@ int main(int argc, char **argv) {
if (maybe_exc) {
spdlog::error(memgraph::utils::MessageWithLink("Unable to load support for embedded Python: {}.", *maybe_exc,
"https://memgr.ph/python"));
} else {
// Change how we load dynamic libraries on Python by using RTLD_NOW and
// RTLD_DEEPBIND flags. This solves an issue with using the wrong version of
// libstd.
auto gil = memgraph::py::EnsureGIL();
// NOLINTNEXTLINE(hicpp-signed-bitwise)
auto *flag = PyLong_FromLong(RTLD_NOW | RTLD_DEEPBIND);
auto *setdl = PySys_GetObject("setdlopenflags");
MG_ASSERT(setdl);
auto *arg = PyTuple_New(1);
MG_ASSERT(arg);
MG_ASSERT(PyTuple_SetItem(arg, 0, flag) == 0);
PyObject_CallObject(setdl, arg);
Py_DECREF(flag);
Py_DECREF(setdl);
Py_DECREF(arg);
}
} else {
spdlog::error(
@@ -1198,6 +1221,7 @@ int main(int argc, char **argv) {
&db,
{.query = {.allow_load_csv = FLAGS_allow_load_csv},
.execution_timeout_sec = FLAGS_query_execution_timeout_sec,
.replication_replica_check_frequency = std::chrono::seconds(FLAGS_replication_replica_check_frequency_sec),
.default_kafka_bootstrap_servers = FLAGS_kafka_bootstrap_servers,
.default_pulsar_service_url = FLAGS_pulsar_service_url,
.stream_transaction_conflict_retries = FLAGS_stream_transaction_conflict_retries,
@@ -1241,8 +1265,10 @@ int main(int argc, char **argv) {
memgraph::utils::MessageWithLink("Using non-secure Bolt connection (without SSL).", "https://memgr.ph/ssl"));
}
ServerT server({FLAGS_bolt_address, static_cast<uint16_t>(FLAGS_bolt_port)}, &session_data, &context,
FLAGS_bolt_session_inactivity_timeout, service_name, FLAGS_bolt_num_workers);
auto server_endpoint = memgraph::communication::v2::ServerEndpoint{
boost::asio::ip::address::from_string(FLAGS_bolt_address), static_cast<uint16_t>(FLAGS_bolt_port)};
ServerT server(server_endpoint, &session_data, &context, FLAGS_bolt_session_inactivity_timeout, service_name,
FLAGS_bolt_num_workers);
// Setup telemetry
std::optional<memgraph::telemetry::Telemetry> telemetry;

View File

@@ -21,6 +21,8 @@ struct InterpreterConfig {
// The default execution timeout is 10 minutes.
double execution_timeout_sec{600.0};
// The same as \ref memgraph::storage::replication::ReplicationClientConfig
std::chrono::seconds replication_replica_check_frequency{1};
std::string default_kafka_bootstrap_servers;
std::string default_pulsar_service_url;

View File

@@ -14,6 +14,6 @@
#include <string>
namespace memgraph::query {
constexpr uint16_t kDefaultReplicationPort = 10000;
constexpr auto *kDefaultReplicationServerIp = "0.0.0.0";
inline constexpr uint16_t kDefaultReplicationPort = 10000;
inline constexpr auto *kDefaultReplicationServerIp = "0.0.0.0";
} // namespace memgraph::query

View File

@@ -324,7 +324,7 @@ class DbAccessor final {
void AdvanceCommand() { accessor_->AdvanceCommand(); }
utils::BasicResult<storage::ConstraintViolation, void> Commit() { return accessor_->Commit(); }
utils::BasicResult<storage::CommitError, void> Commit() { return accessor_->Commit(); }
void Abort() { accessor_->Abort(); }

View File

@@ -852,7 +852,9 @@ cpp<#
: arguments_(arguments),
function_name_(function_name),
function_(NameToFunction(function_name_)) {
DMG_ASSERT(function_, "Unexpected missing function: {}", function_name_);
if (!function_) {
throw SemanticException("Function '{}' doesn't exist.", function_name);
}
}
cpp<#)
(:private
@@ -2629,5 +2631,39 @@ cpp<#
(:serialize (:slk))
(:clone))
(lcp:define-class foreach (clause)
((named_expression "NamedExpression *" :initval "nullptr" :scope :public
:slk-save #'slk-save-ast-pointer
:slk-load (slk-load-ast-pointer "Expression"))
(clauses "std::vector<Clause *>"
:scope :public
:slk-save #'slk-save-ast-vector
:slk-load (slk-load-ast-vector "Clause")))
(:public
#>cpp
Foreach() = default;
bool Accept(HierarchicalTreeVisitor &visitor) override {
if (visitor.PreVisit(*this)) {
named_expression_->Accept(visitor);
for (auto &clause : clauses_) {
clause->Accept(visitor);
}
}
return visitor.PostVisit(*this);
}
cpp<#)
(:protected
#>cpp
Foreach(NamedExpression *expression, std::vector<Clause *> clauses)
: named_expression_(expression), clauses_(clauses) {}
cpp<#)
(:private
#>cpp
friend class AstStorage;
cpp<#)
(:serialize (:slk))
(:clone))
(lcp:pop-namespace) ;; namespace query
(lcp:pop-namespace) ;; namespace memgraph

View File

@@ -93,6 +93,7 @@ class CreateSnapshotQuery;
class StreamQuery;
class SettingQuery;
class VersionQuery;
class Foreach;
using TreeCompositeVisitor = utils::CompositeVisitor<
SingleQuery, CypherUnion, NamedExpression, OrOperator, XorOperator, AndOperator, NotOperator, AdditionOperator,
@@ -101,7 +102,7 @@ using TreeCompositeVisitor = utils::CompositeVisitor<
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>;
RemoveProperty, RemoveLabels, Merge, Unwind, RegexMatch, LoadCsv, Foreach>;
using TreeLeafVisitor = utils::LeafVisitor<Identifier, PrimitiveLiteral, ParameterLookup>;

View File

@@ -37,6 +37,7 @@
#include "utils/exceptions.hpp"
#include "utils/logging.hpp"
#include "utils/string.hpp"
#include "utils/typeinfo.hpp"
namespace memgraph::query::frontend {
@@ -645,28 +646,28 @@ antlrcpp::Any CypherMainVisitor::visitKafkaCreateStreamConfig(MemgraphCypher::Ka
if (ctx->TOPICS()) {
ThrowIfExists(memory_, KafkaConfigKey::TOPICS);
constexpr auto topics_key = static_cast<uint8_t>(KafkaConfigKey::TOPICS);
static constexpr auto topics_key = static_cast<uint8_t>(KafkaConfigKey::TOPICS);
GetTopicNames(memory_[topics_key], ctx->topicNames(), *this);
return {};
}
if (ctx->CONSUMER_GROUP()) {
ThrowIfExists(memory_, KafkaConfigKey::CONSUMER_GROUP);
constexpr auto consumer_group_key = static_cast<uint8_t>(KafkaConfigKey::CONSUMER_GROUP);
static constexpr auto consumer_group_key = static_cast<uint8_t>(KafkaConfigKey::CONSUMER_GROUP);
memory_[consumer_group_key] = JoinSymbolicNamesWithDotsAndMinus(*this, *ctx->consumerGroup);
return {};
}
if (ctx->CONFIGS()) {
ThrowIfExists(memory_, KafkaConfigKey::CONFIGS);
constexpr auto configs_key = static_cast<uint8_t>(KafkaConfigKey::CONFIGS);
static constexpr auto configs_key = static_cast<uint8_t>(KafkaConfigKey::CONFIGS);
memory_.emplace(configs_key, ctx->configsMap->accept(this).as<std::unordered_map<Expression *, Expression *>>());
return {};
}
if (ctx->CREDENTIALS()) {
ThrowIfExists(memory_, KafkaConfigKey::CREDENTIALS);
constexpr auto credentials_key = static_cast<uint8_t>(KafkaConfigKey::CREDENTIALS);
static constexpr auto credentials_key = static_cast<uint8_t>(KafkaConfigKey::CREDENTIALS);
memory_.emplace(credentials_key,
ctx->credentialsMap->accept(this).as<std::unordered_map<Expression *, Expression *>>());
return {};
@@ -956,7 +957,8 @@ antlrcpp::Any CypherMainVisitor::visitSingleQuery(MemgraphCypher::SingleQueryCon
utils::IsSubtype(clause_type, SetProperty::kType) ||
utils::IsSubtype(clause_type, SetProperties::kType) || utils::IsSubtype(clause_type, SetLabels::kType) ||
utils::IsSubtype(clause_type, RemoveProperty::kType) ||
utils::IsSubtype(clause_type, RemoveLabels::kType) || utils::IsSubtype(clause_type, Merge::kType)) {
utils::IsSubtype(clause_type, RemoveLabels::kType) || utils::IsSubtype(clause_type, Merge::kType) ||
utils::IsSubtype(clause_type, Foreach::kType)) {
if (has_return) {
throw SemanticException("Update clause can't be used after RETURN.");
}
@@ -1036,6 +1038,9 @@ antlrcpp::Any CypherMainVisitor::visitClause(MemgraphCypher::ClauseContext *ctx)
if (ctx->loadCsv()) {
return static_cast<Clause *>(ctx->loadCsv()->accept(this).as<LoadCsv *>());
}
if (ctx->foreach ()) {
return static_cast<Clause *>(ctx->foreach ()->accept(this).as<Foreach *>());
}
// TODO: implement other clauses.
throw utils::NotYetImplemented("clause '{}'", ctx->getText());
return 0;
@@ -2104,13 +2109,30 @@ antlrcpp::Any CypherMainVisitor::visitFunctionInvocation(MemgraphCypher::Functio
storage_->Create<Aggregation>(expressions[1], expressions[0], Aggregation::Op::COLLECT_MAP));
}
auto function = NameToFunction(function_name);
if (!function) throw SemanticException("Function '{}' doesn't exist.", function_name);
auto is_user_defined_function = [](const std::string &function_name) {
// Dots are present only in user-defined functions, since modules are case-sensitive, so must be user-defined
// functions. Builtin functions should be case insensitive.
return function_name.find('.') != std::string::npos;
};
// Don't cache queries which call user-defined functions. User-defined function's return
// types can vary depending on whether the module is reloaded, therefore the cache would
// be invalid.
if (is_user_defined_function(function_name)) {
query_info_.is_cacheable = false;
}
return static_cast<Expression *>(storage_->Create<Function>(function_name, expressions));
}
antlrcpp::Any CypherMainVisitor::visitFunctionName(MemgraphCypher::FunctionNameContext *ctx) {
return utils::ToUpperCase(ctx->getText());
auto function_name = ctx->getText();
// Dots are present only in user-defined functions, since modules are case-sensitive, so must be user-defined
// functions. Builtin functions should be case insensitive.
if (function_name.find('.') != std::string::npos) {
return function_name;
}
return utils::ToUpperCase(function_name);
}
antlrcpp::Any CypherMainVisitor::visitDoubleLiteral(MemgraphCypher::DoubleLiteralContext *ctx) {
@@ -2283,6 +2305,37 @@ antlrcpp::Any CypherMainVisitor::visitFilterExpression(MemgraphCypher::FilterExp
return 0;
}
antlrcpp::Any CypherMainVisitor::visitForeach(MemgraphCypher::ForeachContext *ctx) {
auto *for_each = storage_->Create<Foreach>();
auto *named_expr = storage_->Create<NamedExpression>();
named_expr->expression_ = ctx->expression()->accept(this);
named_expr->name_ = std::string(ctx->variable()->accept(this).as<std::string>());
for_each->named_expression_ = named_expr;
for (auto *update_clause_ctx : ctx->updateClause()) {
if (auto *set = update_clause_ctx->set(); set) {
auto set_items = visitSet(set).as<std::vector<Clause *>>();
std::copy(set_items.begin(), set_items.end(), std::back_inserter(for_each->clauses_));
} else if (auto *remove = update_clause_ctx->remove(); remove) {
auto remove_items = visitRemove(remove).as<std::vector<Clause *>>();
std::copy(remove_items.begin(), remove_items.end(), std::back_inserter(for_each->clauses_));
} else if (auto *merge = update_clause_ctx->merge(); merge) {
for_each->clauses_.push_back(visitMerge(merge).as<Merge *>());
} else if (auto *create = update_clause_ctx->create(); create) {
for_each->clauses_.push_back(visitCreate(create).as<Create *>());
} else if (auto *cypher_delete = update_clause_ctx->cypherDelete(); cypher_delete) {
for_each->clauses_.push_back(visitCypherDelete(cypher_delete).as<Delete *>());
} else {
auto *nested_for_each = update_clause_ctx->foreach ();
MG_ASSERT(nested_for_each != nullptr, "Unexpected clause in FOREACH");
for_each->clauses_.push_back(visitForeach(nested_for_each).as<Foreach *>());
}
}
return for_each;
}
LabelIx CypherMainVisitor::AddLabel(const std::string &name) { return storage_->GetLabelIx(name); }
PropertyIx CypherMainVisitor::AddProperty(const std::string &name) { return storage_->GetPropertyIx(name); }

View File

@@ -844,6 +844,11 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
*/
antlrcpp::Any visitFilterExpression(MemgraphCypher::FilterExpressionContext *) override;
/**
* @return Foreach*
*/
antlrcpp::Any visitForeach(MemgraphCypher::ForeachContext *ctx) override;
public:
Query *query() { return query_; }
const static std::string kAnonPrefix;

View File

@@ -279,7 +279,7 @@ idInColl : variable IN expression ;
functionInvocation : functionName '(' ( DISTINCT )? ( expression ( ',' expression )* )? ')' ;
functionName : symbolicName ;
functionName : symbolicName ( '.' symbolicName )* ;
listComprehension : '[' filterExpression ( '|' expression )? ']' ;

View File

@@ -47,6 +47,7 @@ memgraphCypherKeyword : cypherKeyword
| DUMP
| EXECUTE
| FOR
| FOREACH
| FREE
| FROM
| GLOBAL
@@ -163,8 +164,19 @@ clause : cypherMatch
| cypherReturn
| callProcedure
| loadCsv
| foreach
;
updateClause : set
| remove
| create
| merge
| cypherDelete
| foreach
;
foreach : FOREACH '(' variable IN expression '|' updateClause+ ')' ;
streamQuery : checkStream
| createStream
| dropStream

View File

@@ -54,6 +54,7 @@ DUMP : D U M P ;
DURABILITY : D U R A B I L I T Y ;
EXECUTE : E X E C U T E ;
FOR : F O R ;
FOREACH : F O R E A C H;
FREE : F R E E ;
FREE_MEMORY : F R E E UNDERSCORE M E M O R Y ;
FROM : F R O M ;

View File

@@ -15,7 +15,9 @@
#include "query/frontend/semantic/symbol_generator.hpp"
#include <algorithm>
#include <optional>
#include <ranges>
#include <unordered_set>
#include <variant>
@@ -39,35 +41,56 @@ std::unordered_map<std::string, Identifier *> GeneratePredefinedIdentifierMap(
} // namespace
SymbolGenerator::SymbolGenerator(SymbolTable *symbol_table, const std::vector<Identifier *> &predefined_identifiers)
: symbol_table_(symbol_table), predefined_identifiers_{GeneratePredefinedIdentifierMap(predefined_identifiers)} {}
: symbol_table_(symbol_table),
predefined_identifiers_{GeneratePredefinedIdentifierMap(predefined_identifiers)},
scopes_(1, Scope()) {}
auto SymbolGenerator::CreateSymbol(const std::string &name, bool user_declared, Symbol::Type type, int token_position) {
auto symbol = symbol_table_->CreateSymbol(name, user_declared, type, token_position);
scope_.symbols[name] = symbol;
return symbol;
}
auto SymbolGenerator::GetOrCreateSymbol(const std::string &name, bool user_declared, Symbol::Type type) {
auto search = scope_.symbols.find(name);
if (search != scope_.symbols.end()) {
auto symbol = search->second;
std::optional<Symbol> SymbolGenerator::FindSymbolInScope(const std::string &name, const Scope &scope,
Symbol::Type type) {
if (auto it = scope.symbols.find(name); it != scope.symbols.end()) {
const auto &symbol = it->second;
// Unless we have `ANY` type, check that types match.
if (type != Symbol::Type::ANY && symbol.type() != Symbol::Type::ANY && type != symbol.type()) {
throw TypeMismatchError(name, Symbol::TypeToString(symbol.type()), Symbol::TypeToString(type));
}
return search->second;
return symbol;
}
return std::nullopt;
}
auto SymbolGenerator::CreateSymbol(const std::string &name, bool user_declared, Symbol::Type type, int token_position) {
auto symbol = symbol_table_->CreateSymbol(name, user_declared, type, token_position);
scopes_.back().symbols[name] = symbol;
return symbol;
}
auto SymbolGenerator::GetOrCreateSymbolLocalScope(const std::string &name, bool user_declared, Symbol::Type type) {
auto &scope = scopes_.back();
if (auto maybe_symbol = FindSymbolInScope(name, scope, type); maybe_symbol) {
return *maybe_symbol;
}
return CreateSymbol(name, user_declared, type);
}
auto SymbolGenerator::GetOrCreateSymbol(const std::string &name, bool user_declared, Symbol::Type type) {
// NOLINTNEXTLINE
for (auto scope = scopes_.rbegin(); scope != scopes_.rend(); ++scope) {
if (auto maybe_symbol = FindSymbolInScope(name, *scope, type); maybe_symbol) {
return *maybe_symbol;
}
}
return CreateSymbol(name, user_declared, type);
}
void SymbolGenerator::VisitReturnBody(ReturnBody &body, Where *where) {
auto &scope = scopes_.back();
for (auto &expr : body.named_expressions) {
expr->Accept(*this);
}
std::vector<Symbol> user_symbols;
if (body.all_identifiers) {
// Carry over user symbols because '*' appeared.
for (auto sym_pair : scope_.symbols) {
for (const auto &sym_pair : scope.symbols) {
if (!sym_pair.second.user_declared()) {
continue;
}
@@ -81,18 +104,18 @@ void SymbolGenerator::VisitReturnBody(ReturnBody &body, Where *where) {
// declares only those established through named expressions. New declarations
// must not be visible inside named expressions themselves.
bool removed_old_names = false;
if ((!where && body.order_by.empty()) || scope_.has_aggregation) {
if ((!where && body.order_by.empty()) || scope.has_aggregation) {
// WHERE and ORDER BY need to see both the old and new symbols, unless we
// have an aggregation. Therefore, we can clear the symbols immediately if
// there is neither ORDER BY nor WHERE, or we have an aggregation.
scope_.symbols.clear();
scope.symbols.clear();
removed_old_names = true;
}
// Create symbols for named expressions.
std::unordered_set<std::string> new_names;
for (const auto &user_sym : user_symbols) {
new_names.insert(user_sym.name());
scope_.symbols[user_sym.name()] = user_sym;
scope.symbols[user_sym.name()] = user_sym;
}
for (auto &named_expr : body.named_expressions) {
const auto &name = named_expr->name_;
@@ -103,35 +126,35 @@ void SymbolGenerator::VisitReturnBody(ReturnBody &body, Where *where) {
// new symbol would have a more specific type.
named_expr->MapTo(CreateSymbol(name, true, Symbol::Type::ANY, named_expr->token_position_));
}
scope_.in_order_by = true;
scope.in_order_by = true;
for (const auto &order_pair : body.order_by) {
order_pair.expression->Accept(*this);
}
scope_.in_order_by = false;
scope.in_order_by = false;
if (body.skip) {
scope_.in_skip = true;
scope.in_skip = true;
body.skip->Accept(*this);
scope_.in_skip = false;
scope.in_skip = false;
}
if (body.limit) {
scope_.in_limit = true;
scope.in_limit = true;
body.limit->Accept(*this);
scope_.in_limit = false;
scope.in_limit = false;
}
if (where) where->Accept(*this);
if (!removed_old_names) {
// We have an ORDER BY or WHERE, but no aggregation, which means we didn't
// clear the old symbols, so do it now. We cannot just call clear, because
// we've added new symbols.
for (auto sym_it = scope_.symbols.begin(); sym_it != scope_.symbols.end();) {
for (auto sym_it = scope.symbols.begin(); sym_it != scope.symbols.end();) {
if (new_names.find(sym_it->first) == new_names.end()) {
sym_it = scope_.symbols.erase(sym_it);
sym_it = scope.symbols.erase(sym_it);
} else {
sym_it++;
}
}
}
scope_.has_aggregation = false;
scopes_.back().has_aggregation = false;
}
// Query
@@ -145,7 +168,7 @@ bool SymbolGenerator::PreVisit(SingleQuery &) {
// Union
bool SymbolGenerator::PreVisit(CypherUnion &) {
scope_ = Scope();
scopes_.back() = Scope();
return true;
}
@@ -166,11 +189,11 @@ bool SymbolGenerator::PostVisit(CypherUnion &cypher_union) {
// Clauses
bool SymbolGenerator::PreVisit(Create &) {
scope_.in_create = true;
scopes_.back().in_create = true;
return true;
}
bool SymbolGenerator::PostVisit(Create &) {
scope_.in_create = false;
scopes_.back().in_create = false;
return true;
}
@@ -183,7 +206,7 @@ bool SymbolGenerator::PreVisit(CallProcedure &call_proc) {
bool SymbolGenerator::PostVisit(CallProcedure &call_proc) {
for (auto *ident : call_proc.result_identifiers_) {
if (HasSymbol(ident->name_)) {
if (HasSymbolLocalScope(ident->name_)) {
throw RedeclareVariableError(ident->name_);
}
ident->MapTo(CreateSymbol(ident->name_, true));
@@ -194,7 +217,7 @@ bool SymbolGenerator::PostVisit(CallProcedure &call_proc) {
bool SymbolGenerator::PreVisit(LoadCsv &load_csv) { return false; }
bool SymbolGenerator::PostVisit(LoadCsv &load_csv) {
if (HasSymbol(load_csv.row_var_->name_)) {
if (HasSymbolLocalScope(load_csv.row_var_->name_)) {
throw RedeclareVariableError(load_csv.row_var_->name_);
}
load_csv.row_var_->MapTo(CreateSymbol(load_csv.row_var_->name_, true));
@@ -202,45 +225,47 @@ bool SymbolGenerator::PostVisit(LoadCsv &load_csv) {
}
bool SymbolGenerator::PreVisit(Return &ret) {
scope_.in_return = true;
auto &scope = scopes_.back();
scope.in_return = true;
VisitReturnBody(ret.body_);
scope_.in_return = false;
scope.in_return = false;
return false; // We handled the traversal ourselves.
}
bool SymbolGenerator::PostVisit(Return &) {
for (const auto &name_symbol : scope_.symbols) curr_return_names_.insert(name_symbol.first);
for (const auto &name_symbol : scopes_.back().symbols) curr_return_names_.insert(name_symbol.first);
return true;
}
bool SymbolGenerator::PreVisit(With &with) {
scope_.in_with = true;
auto &scope = scopes_.back();
scope.in_with = true;
VisitReturnBody(with.body_, with.where_);
scope_.in_with = false;
scope.in_with = false;
return false; // We handled the traversal ourselves.
}
bool SymbolGenerator::PreVisit(Where &) {
scope_.in_where = true;
scopes_.back().in_where = true;
return true;
}
bool SymbolGenerator::PostVisit(Where &) {
scope_.in_where = false;
scopes_.back().in_where = false;
return true;
}
bool SymbolGenerator::PreVisit(Merge &) {
scope_.in_merge = true;
scopes_.back().in_merge = true;
return true;
}
bool SymbolGenerator::PostVisit(Merge &) {
scope_.in_merge = false;
scopes_.back().in_merge = false;
return true;
}
bool SymbolGenerator::PostVisit(Unwind &unwind) {
const auto &name = unwind.named_expression_->name_;
if (HasSymbol(name)) {
if (HasSymbolLocalScope(name)) {
throw RedeclareVariableError(name);
}
unwind.named_expression_->MapTo(CreateSymbol(name, true));
@@ -248,55 +273,70 @@ bool SymbolGenerator::PostVisit(Unwind &unwind) {
}
bool SymbolGenerator::PreVisit(Match &) {
scope_.in_match = true;
scopes_.back().in_match = true;
return true;
}
bool SymbolGenerator::PostVisit(Match &) {
scope_.in_match = false;
auto &scope = scopes_.back();
scope.in_match = false;
// Check variables in property maps after visiting Match, so that they can
// reference symbols out of bind order.
for (auto &ident : scope_.identifiers_in_match) {
if (!HasSymbol(ident->name_) && !ConsumePredefinedIdentifier(ident->name_))
for (auto &ident : scope.identifiers_in_match) {
if (!HasSymbolLocalScope(ident->name_) && !ConsumePredefinedIdentifier(ident->name_))
throw UnboundVariableError(ident->name_);
ident->MapTo(scope_.symbols[ident->name_]);
ident->MapTo(scope.symbols[ident->name_]);
}
scope_.identifiers_in_match.clear();
scope.identifiers_in_match.clear();
return true;
}
bool SymbolGenerator::PreVisit(Foreach &for_each) {
const auto &name = for_each.named_expression_->name_;
scopes_.emplace_back(Scope());
scopes_.back().in_foreach = true;
for_each.named_expression_->MapTo(
CreateSymbol(name, true, Symbol::Type::ANY, for_each.named_expression_->token_position_));
return true;
}
bool SymbolGenerator::PostVisit([[maybe_unused]] Foreach &for_each) {
scopes_.pop_back();
return true;
}
// Expressions
SymbolGenerator::ReturnType SymbolGenerator::Visit(Identifier &ident) {
if (scope_.in_skip || scope_.in_limit) {
throw SemanticException("Variables are not allowed in {}.", scope_.in_skip ? "SKIP" : "LIMIT");
auto &scope = scopes_.back();
if (scope.in_skip || scope.in_limit) {
throw SemanticException("Variables are not allowed in {}.", scope.in_skip ? "SKIP" : "LIMIT");
}
Symbol symbol;
if (scope_.in_pattern && !(scope_.in_node_atom || scope_.visiting_edge)) {
if (scope.in_pattern && !(scope.in_node_atom || scope.visiting_edge)) {
// If we are in the pattern, and outside of a node or an edge, the
// identifier is the pattern name.
symbol = GetOrCreateSymbol(ident.name_, ident.user_declared_, Symbol::Type::PATH);
} else if (scope_.in_pattern && scope_.in_pattern_atom_identifier) {
symbol = GetOrCreateSymbolLocalScope(ident.name_, ident.user_declared_, Symbol::Type::PATH);
} else if (scope.in_pattern && scope.in_pattern_atom_identifier) {
// Patterns used to create nodes and edges cannot redeclare already
// established bindings. Declaration only happens in single node
// patterns and in edge patterns. OpenCypher example,
// `MATCH (n) CREATE (n)` should throw an error that `n` is already
// declared. While `MATCH (n) CREATE (n) -[:R]-> (n)` is allowed,
// since `n` now references the bound node instead of declaring it.
if ((scope_.in_create_node || scope_.in_create_edge) && HasSymbol(ident.name_)) {
if ((scope.in_create_node || scope.in_create_edge) && HasSymbolLocalScope(ident.name_)) {
throw RedeclareVariableError(ident.name_);
}
auto type = Symbol::Type::VERTEX;
if (scope_.visiting_edge) {
if (scope.visiting_edge) {
// Edge referencing is not allowed (like in Neo4j):
// `MATCH (n) - [r] -> (n) - [r] -> (n) RETURN r` is not allowed.
if (HasSymbol(ident.name_)) {
if (HasSymbolLocalScope(ident.name_)) {
throw RedeclareVariableError(ident.name_);
}
type = scope_.visiting_edge->IsVariable() ? Symbol::Type::EDGE_LIST : Symbol::Type::EDGE;
type = scope.visiting_edge->IsVariable() ? Symbol::Type::EDGE_LIST : Symbol::Type::EDGE;
}
symbol = GetOrCreateSymbol(ident.name_, ident.user_declared_, type);
} else if (scope_.in_pattern && !scope_.in_pattern_atom_identifier && scope_.in_match) {
if (scope_.in_edge_range && scope_.visiting_edge->identifier_->name_ == ident.name_) {
symbol = GetOrCreateSymbolLocalScope(ident.name_, ident.user_declared_, type);
} else if (scope.in_pattern && !scope.in_pattern_atom_identifier && scope.in_match) {
if (scope.in_edge_range && scope.visiting_edge->identifier_->name_ == ident.name_) {
// Prevent variable path bounds to reference the identifier which is bound
// by the variable path itself.
throw UnboundVariableError(ident.name_);
@@ -304,30 +344,30 @@ SymbolGenerator::ReturnType SymbolGenerator::Visit(Identifier &ident) {
// Variables in property maps or bounds of variable length path during MATCH
// can reference symbols bound later in the same MATCH. We collect them
// here, so that they can be checked after visiting Match.
scope_.identifiers_in_match.emplace_back(&ident);
scope.identifiers_in_match.emplace_back(&ident);
} else {
// Everything else references a bound symbol.
if (!HasSymbol(ident.name_) && !ConsumePredefinedIdentifier(ident.name_)) throw UnboundVariableError(ident.name_);
symbol = scope_.symbols[ident.name_];
symbol = GetOrCreateSymbol(ident.name_, ident.user_declared_, Symbol::Type::ANY);
}
ident.MapTo(symbol);
return true;
}
bool SymbolGenerator::PreVisit(Aggregation &aggr) {
auto &scope = scopes_.back();
// Check if the aggregation can be used in this context. This check should
// probably move to a separate phase, which checks if the query is well
// formed.
if ((!scope_.in_return && !scope_.in_with) || scope_.in_order_by || scope_.in_skip || scope_.in_limit ||
scope_.in_where) {
if ((!scope.in_return && !scope.in_with) || scope.in_order_by || scope.in_skip || scope.in_limit || scope.in_where) {
throw SemanticException("Aggregation functions are only allowed in WITH and RETURN.");
}
if (scope_.in_aggregation) {
if (scope.in_aggregation) {
throw SemanticException(
"Using aggregation functions inside aggregation functions is not "
"allowed.");
}
if (scope_.num_if_operators) {
if (scope.num_if_operators) {
// Neo allows aggregations here and produces very interesting behaviors.
// To simplify implementation at this moment we decided to completely
// disallow aggregations inside of the CASE.
@@ -341,23 +381,23 @@ bool SymbolGenerator::PreVisit(Aggregation &aggr) {
// Currently, we only have aggregation operators which return numbers.
auto aggr_name = Aggregation::OpToString(aggr.op_) + std::to_string(aggr.symbol_pos_);
aggr.MapTo(CreateSymbol(aggr_name, false, Symbol::Type::NUMBER));
scope_.in_aggregation = true;
scope_.has_aggregation = true;
scope.in_aggregation = true;
scope.has_aggregation = true;
return true;
}
bool SymbolGenerator::PostVisit(Aggregation &) {
scope_.in_aggregation = false;
scopes_.back().in_aggregation = false;
return true;
}
bool SymbolGenerator::PreVisit(IfOperator &) {
++scope_.num_if_operators;
++scopes_.back().num_if_operators;
return true;
}
bool SymbolGenerator::PostVisit(IfOperator &) {
--scope_.num_if_operators;
--scopes_.back().num_if_operators;
return true;
}
@@ -401,33 +441,36 @@ bool SymbolGenerator::PreVisit(Extract &extract) {
// Pattern and its subparts.
bool SymbolGenerator::PreVisit(Pattern &pattern) {
scope_.in_pattern = true;
if ((scope_.in_create || scope_.in_merge) && pattern.atoms_.size() == 1U) {
auto &scope = scopes_.back();
scope.in_pattern = true;
if ((scope.in_create || scope.in_merge) && pattern.atoms_.size() == 1U) {
MG_ASSERT(utils::IsSubtype(*pattern.atoms_[0], NodeAtom::kType), "Expected a single NodeAtom in Pattern");
scope_.in_create_node = true;
scope.in_create_node = true;
}
return true;
}
bool SymbolGenerator::PostVisit(Pattern &) {
scope_.in_pattern = false;
scope_.in_create_node = false;
auto &scope = scopes_.back();
scope.in_pattern = false;
scope.in_create_node = false;
return true;
}
bool SymbolGenerator::PreVisit(NodeAtom &node_atom) {
auto check_node_semantic = [&node_atom, this](const bool props_or_labels) {
auto &scope = scopes_.back();
auto check_node_semantic = [&node_atom, &scope, this](const bool props_or_labels) {
const auto &node_name = node_atom.identifier_->name_;
if ((scope_.in_create || scope_.in_merge) && props_or_labels && HasSymbol(node_name)) {
if ((scope.in_create || scope.in_merge) && props_or_labels && HasSymbolLocalScope(node_name)) {
throw SemanticException("Cannot create node '" + node_name +
"' with labels or properties, because it is already declared.");
}
scope_.in_pattern_atom_identifier = true;
scope.in_pattern_atom_identifier = true;
node_atom.identifier_->Accept(*this);
scope_.in_pattern_atom_identifier = false;
scope.in_pattern_atom_identifier = false;
};
scope_.in_node_atom = true;
scope.in_node_atom = true;
if (auto *properties = std::get_if<std::unordered_map<PropertyIx, Expression *>>(&node_atom.properties_)) {
bool props_or_labels = !properties->empty() || !node_atom.labels_.empty();
@@ -447,20 +490,21 @@ bool SymbolGenerator::PreVisit(NodeAtom &node_atom) {
}
bool SymbolGenerator::PostVisit(NodeAtom &) {
scope_.in_node_atom = false;
scopes_.back().in_node_atom = false;
return true;
}
bool SymbolGenerator::PreVisit(EdgeAtom &edge_atom) {
scope_.visiting_edge = &edge_atom;
if (scope_.in_create || scope_.in_merge) {
scope_.in_create_edge = true;
auto &scope = scopes_.back();
scope.visiting_edge = &edge_atom;
if (scope.in_create || scope.in_merge) {
scope.in_create_edge = true;
if (edge_atom.edge_types_.size() != 1U) {
throw SemanticException(
"A single relationship type must be specified "
"when creating an edge.");
}
if (scope_.in_create && // Merge allows bidirectionality
if (scope.in_create && // Merge allows bidirectionality
edge_atom.direction_ == EdgeAtom::Direction::BOTH) {
throw SemanticException(
"Bidirectional relationship are not supported "
@@ -480,15 +524,15 @@ bool SymbolGenerator::PreVisit(EdgeAtom &edge_atom) {
std::get<ParameterLookup *>(edge_atom.properties_)->Accept(*this);
}
if (edge_atom.IsVariable()) {
scope_.in_edge_range = true;
scope.in_edge_range = true;
if (edge_atom.lower_bound_) {
edge_atom.lower_bound_->Accept(*this);
}
if (edge_atom.upper_bound_) {
edge_atom.upper_bound_->Accept(*this);
}
scope_.in_edge_range = false;
scope_.in_pattern = false;
scope.in_edge_range = false;
scope.in_pattern = false;
if (edge_atom.filter_lambda_.expression) {
VisitWithIdentifiers(edge_atom.filter_lambda_.expression,
{edge_atom.filter_lambda_.inner_edge, edge_atom.filter_lambda_.inner_node});
@@ -505,34 +549,36 @@ bool SymbolGenerator::PreVisit(EdgeAtom &edge_atom) {
VisitWithIdentifiers(edge_atom.weight_lambda_.expression,
{edge_atom.weight_lambda_.inner_edge, edge_atom.weight_lambda_.inner_node});
}
scope_.in_pattern = true;
scope.in_pattern = true;
}
scope_.in_pattern_atom_identifier = true;
scope.in_pattern_atom_identifier = true;
edge_atom.identifier_->Accept(*this);
scope_.in_pattern_atom_identifier = false;
scope.in_pattern_atom_identifier = false;
if (edge_atom.total_weight_) {
if (HasSymbol(edge_atom.total_weight_->name_)) {
if (HasSymbolLocalScope(edge_atom.total_weight_->name_)) {
throw RedeclareVariableError(edge_atom.total_weight_->name_);
}
edge_atom.total_weight_->MapTo(GetOrCreateSymbol(edge_atom.total_weight_->name_,
edge_atom.total_weight_->user_declared_, Symbol::Type::NUMBER));
edge_atom.total_weight_->MapTo(GetOrCreateSymbolLocalScope(
edge_atom.total_weight_->name_, edge_atom.total_weight_->user_declared_, Symbol::Type::NUMBER));
}
return false;
}
bool SymbolGenerator::PostVisit(EdgeAtom &) {
scope_.visiting_edge = nullptr;
scope_.in_create_edge = false;
auto &scope = scopes_.back();
scope.visiting_edge = nullptr;
scope.in_create_edge = false;
return true;
}
void SymbolGenerator::VisitWithIdentifiers(Expression *expr, const std::vector<Identifier *> &identifiers) {
auto &scope = scopes_.back();
std::vector<std::pair<std::optional<Symbol>, Identifier *>> prev_symbols;
// Collect previous symbols if they exist.
for (const auto &identifier : identifiers) {
std::optional<Symbol> prev_symbol;
auto prev_symbol_it = scope_.symbols.find(identifier->name_);
if (prev_symbol_it != scope_.symbols.end()) {
auto prev_symbol_it = scope.symbols.find(identifier->name_);
if (prev_symbol_it != scope.symbols.end()) {
prev_symbol = prev_symbol_it->second;
}
identifier->MapTo(CreateSymbol(identifier->name_, identifier->user_declared_));
@@ -545,14 +591,20 @@ void SymbolGenerator::VisitWithIdentifiers(Expression *expr, const std::vector<I
const auto &prev_symbol = prev.first;
const auto &identifier = prev.second;
if (prev_symbol) {
scope_.symbols[identifier->name_] = *prev_symbol;
scope.symbols[identifier->name_] = *prev_symbol;
} else {
scope_.symbols.erase(identifier->name_);
scope.symbols.erase(identifier->name_);
}
}
}
bool SymbolGenerator::HasSymbol(const std::string &name) { return scope_.symbols.find(name) != scope_.symbols.end(); }
bool SymbolGenerator::HasSymbol(const std::string &name) const {
return std::ranges::any_of(scopes_, [&name](const auto &scope) { return scope.symbols.contains(name); });
}
bool SymbolGenerator::HasSymbolLocalScope(const std::string &name) const {
return scopes_.back().symbols.contains(name);
}
bool SymbolGenerator::ConsumePredefinedIdentifier(const std::string &name) {
auto it = predefined_identifiers_.find(name);

View File

@@ -15,6 +15,9 @@
#pragma once
#include <optional>
#include <vector>
#include "query/exceptions.hpp"
#include "query/frontend/ast/ast.hpp"
#include "query/frontend/semantic/symbol_table.hpp"
@@ -59,6 +62,8 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
bool PostVisit(Unwind &) override;
bool PreVisit(Match &) override;
bool PostVisit(Match &) override;
bool PreVisit(Foreach &) override;
bool PostVisit(Foreach &) override;
// Expressions
ReturnType Visit(Identifier &) override;
@@ -107,6 +112,7 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
bool in_order_by{false};
bool in_where{false};
bool in_match{false};
bool in_foreach{false};
// True when visiting a pattern atom (node or edge) identifier, which can be
// reused or created in the pattern itself.
bool in_pattern_atom_identifier{false};
@@ -125,7 +131,10 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
int num_if_operators{0};
};
bool HasSymbol(const std::string &name);
static std::optional<Symbol> FindSymbolInScope(const std::string &name, const Scope &scope, Symbol::Type type);
bool HasSymbol(const std::string &name) const;
bool HasSymbolLocalScope(const std::string &name) const;
// @return true if it added a predefined identifier with that name
bool ConsumePredefinedIdentifier(const std::string &name);
@@ -135,9 +144,10 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
auto CreateSymbol(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY,
int token_position = -1);
auto GetOrCreateSymbol(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY);
// Returns the symbol by name. If the mapping already exists, checks if the
// types match. Otherwise, returns a new symbol.
auto GetOrCreateSymbol(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY);
auto GetOrCreateSymbolLocalScope(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY);
void VisitReturnBody(ReturnBody &body, Where *where = nullptr);
@@ -148,7 +158,7 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
// Identifiers which are injected from outside the query. Each identifier
// is mapped by its name.
std::unordered_map<std::string, Identifier *> predefined_identifiers_;
Scope scope_;
std::vector<Scope> scopes_;
std::unordered_set<std::string> prev_return_names_;
std::unordered_set<std::string> curr_return_names_;
};

View File

@@ -204,7 +204,8 @@ const trie::Trie kKeywords = {"union",
"pulsar",
"service_url",
"version",
"websocket"};
"websocket"
"foreach"};
// Unicode codepoints that are allowed at the start of the unescaped name.
const std::bitset<kBitsetSize> kUnescapedNameAllowedStarts(

View File

@@ -22,6 +22,9 @@
#include "query/db_accessor.hpp"
#include "query/exceptions.hpp"
#include "query/procedure/cypher_types.hpp"
#include "query/procedure/mg_procedure_impl.hpp"
#include "query/procedure/module.hpp"
#include "query/typed_value.hpp"
#include "utils/string.hpp"
#include "utils/temporal.hpp"
@@ -307,9 +310,9 @@ void FType(const char *name, const TypedValue *args, int64_t nargs, int64_t pos
}
return;
}
constexpr int64_t required_args = FTypeRequiredArgs<ArgType, ArgTypes...>();
constexpr int64_t optional_args = FTypeOptionalArgs<ArgType, ArgTypes...>();
constexpr int64_t total_args = required_args + optional_args;
static constexpr int64_t required_args = FTypeRequiredArgs<ArgType, ArgTypes...>();
static constexpr int64_t optional_args = FTypeOptionalArgs<ArgType, ArgTypes...>();
static constexpr int64_t total_args = required_args + optional_args;
if constexpr (optional_args > 0) {
if (nargs < required_args || nargs > total_args) {
throw QueryRuntimeException("'{}' requires between {} and {} arguments.", name, required_args, total_args);
@@ -810,7 +813,7 @@ TypedValue StringMatchOperator(const TypedValue *args, int64_t nargs, const Func
// Check if s1 starts with s2.
struct StartsWithPredicate {
constexpr static const char *name = "startsWith";
static constexpr const char *name = "startsWith";
bool operator()(const TypedValue::TString &s1, const TypedValue::TString &s2) const {
if (s1.size() < s2.size()) return false;
return std::equal(s2.begin(), s2.end(), s1.begin());
@@ -820,7 +823,7 @@ auto StartsWith = StringMatchOperator<StartsWithPredicate>;
// Check if s1 ends with s2.
struct EndsWithPredicate {
constexpr static const char *name = "endsWith";
static constexpr const char *name = "endsWith";
bool operator()(const TypedValue::TString &s1, const TypedValue::TString &s2) const {
if (s1.size() < s2.size()) return false;
return std::equal(s2.rbegin(), s2.rend(), s1.rbegin());
@@ -830,7 +833,7 @@ auto EndsWith = StringMatchOperator<EndsWithPredicate>;
// Check if s1 contains s2.
struct ContainsPredicate {
constexpr static const char *name = "contains";
static constexpr const char *name = "contains";
bool operator()(const TypedValue::TString &s1, const TypedValue::TString &s2) const {
if (s1.size() < s2.size()) return false;
return s1.find(s2) != std::string::npos;
@@ -1174,6 +1177,53 @@ TypedValue Duration(const TypedValue *args, int64_t nargs, const FunctionContext
MapNumericParameters<Number>(parameter_mappings, args[0].ValueMap());
return TypedValue(utils::Duration(duration_parameters), ctx.memory);
}
std::function<TypedValue(const TypedValue *, const int64_t, const FunctionContext &)> UserFunction(
const mgp_func &func, const std::string &fully_qualified_name) {
return [func, fully_qualified_name](const TypedValue *args, int64_t nargs, const FunctionContext &ctx) -> TypedValue {
/// Find function is called to aquire the lock on Module pointer while user-defined function is executed
const auto &maybe_found =
procedure::FindFunction(procedure::gModuleRegistry, fully_qualified_name, utils::NewDeleteResource());
if (!maybe_found) {
throw QueryRuntimeException(
"Function '{}' has been unloaded. Please check query modules to confirm that function is loaded in Memgraph.",
fully_qualified_name);
}
/// Explicit extraction of module pointer, to clearly state that the lock is aquired.
// NOLINTNEXTLINE(clang-diagnostic-unused-variable)
const auto &module_ptr = (*maybe_found).first;
const auto &func_cb = func.cb;
mgp_memory memory{ctx.memory};
mgp_func_context functx{ctx.db_accessor, ctx.view};
auto graph = mgp_graph::NonWritableGraph(*ctx.db_accessor, ctx.view);
std::vector<TypedValue> args_list;
args_list.reserve(nargs);
for (std::size_t i = 0; i < nargs; ++i) {
args_list.emplace_back(args[i]);
}
auto function_argument_list = mgp_list(ctx.memory);
procedure::ConstructArguments(args_list, func, fully_qualified_name, function_argument_list, graph);
mgp_func_result maybe_res;
func_cb(&function_argument_list, &functx, &maybe_res, &memory);
if (maybe_res.error_msg) {
throw QueryRuntimeException(*maybe_res.error_msg);
}
if (!maybe_res.value) {
throw QueryRuntimeException(
"Function '{}' didn't set the result nor the error message. Please either set the result by using "
"mgp_func_result_set_value or the error by using mgp_func_result_set_error_msg.",
fully_qualified_name);
}
return {*(maybe_res.value), ctx.memory};
};
}
} // namespace
std::function<TypedValue(const TypedValue *, int64_t, const FunctionContext &ctx)> NameToFunction(
@@ -1259,6 +1309,14 @@ std::function<TypedValue(const TypedValue *, int64_t, const FunctionContext &ctx
if (function_name == "LOCALDATETIME") return LocalDateTime;
if (function_name == "DURATION") return Duration;
const auto &maybe_found =
procedure::FindFunction(procedure::gModuleRegistry, function_name, utils::NewDeleteResource());
if (maybe_found) {
const auto *func = (*maybe_found).second;
return UserFunction(*func, function_name);
}
return nullptr;
}

View File

@@ -160,7 +160,8 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
/// @throw QueryRuntimeException if an error ocurred.
void RegisterReplica(const std::string &name, const std::string &socket_address,
const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout) override {
const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout,
const std::chrono::seconds replica_check_frequency) override {
if (db_->GetReplicationRole() == storage::ReplicationRole::REPLICA) {
// replica can't register another replica
throw QueryRuntimeException("Replica can't register another replica!");
@@ -182,8 +183,9 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
io::network::Endpoint::ParseSocketOrIpAddress(socket_address, query::kDefaultReplicationPort);
if (maybe_ip_and_port) {
auto [ip, port] = *maybe_ip_and_port;
auto ret =
db_->RegisterReplica(name, {std::move(ip), port}, repl_mode, {.timeout = timeout, .ssl = std::nullopt});
auto ret = db_->RegisterReplica(
name, {std::move(ip), port}, repl_mode,
{.timeout = timeout, .replica_check_frequency = replica_check_frequency, .ssl = std::nullopt});
if (ret.HasError()) {
throw QueryRuntimeException(fmt::format("Couldn't register replica '{}'!", name));
}
@@ -448,7 +450,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
return callback;
}
case ReplicationQuery::Action::SHOW_REPLICATION_ROLE: {
callback.header = {"replication mode"};
callback.header = {"replication role"};
callback.fn = [handler = ReplQueryHandler{interpreter_context->db}] {
auto mode = handler.ShowReplicationRole();
switch (mode) {
@@ -467,6 +469,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
const auto &sync_mode = repl_query->sync_mode_;
auto socket_address = repl_query->socket_address_->Accept(evaluator);
auto timeout = EvaluateOptionalExpression(repl_query->timeout_, &evaluator);
const auto replica_check_frequency = interpreter_context->config.replication_replica_check_frequency;
std::optional<double> maybe_timeout;
if (timeout.IsDouble()) {
maybe_timeout = timeout.ValueDouble();
@@ -474,8 +477,9 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
maybe_timeout = static_cast<double>(timeout.ValueInt());
}
callback.fn = [handler = ReplQueryHandler{interpreter_context->db}, name, socket_address, sync_mode,
maybe_timeout]() mutable {
handler.RegisterReplica(name, std::string(socket_address.ValueString()), sync_mode, maybe_timeout);
maybe_timeout, replica_check_frequency]() mutable {
handler.RegisterReplica(name, std::string(socket_address.ValueString()), sync_mode, maybe_timeout,
replica_check_frequency);
return std::vector<std::vector<TypedValue>>();
};
notifications->emplace_back(SeverityLevel::INFO, NotificationCode::REGISTER_REPLICA,
@@ -512,7 +516,6 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
typed_replica.emplace_back(TypedValue("async"));
break;
}
typed_replica.emplace_back(TypedValue(static_cast<int64_t>(replica.sync_mode)));
if (replica.timeout) {
typed_replica.emplace_back(TypedValue(*replica.timeout));
} else {
@@ -554,7 +557,7 @@ std::vector<std::string> EvaluateTopicNames(ExpressionEvaluator &evaluator,
Callback::CallbackFunction GetKafkaCreateCallback(StreamQuery *stream_query, ExpressionEvaluator &evaluator,
InterpreterContext *interpreter_context,
const std::string *username) {
constexpr std::string_view kDefaultConsumerGroup = "mg_consumer";
static constexpr std::string_view kDefaultConsumerGroup = "mg_consumer";
std::string consumer_group{stream_query->consumer_group_.empty() ? kDefaultConsumerGroup
: stream_query->consumer_group_};
@@ -899,7 +902,7 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *strea
// Set up temporary memory for a single Pull. Initial memory comes from the
// stack. 256 KiB should fit on the stack and should be more than enough for a
// single `Pull`.
constexpr size_t stack_size = 256 * 1024;
static constexpr size_t stack_size = 256UL * 1024UL;
char stack_data[stack_size];
utils::ResourceWithOutOfMemoryException resource_with_exception;
utils::MonotonicBufferResource monotonic_memory(&stack_data[0], stack_size, &resource_with_exception);
@@ -2206,25 +2209,39 @@ void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, Interpret
continue;
}
auto maybe_constraint_violation = db_accessor.Commit();
if (maybe_constraint_violation.HasError()) {
const auto &constraint_violation = maybe_constraint_violation.GetError();
switch (constraint_violation.type) {
case storage::ConstraintViolation::Type::EXISTENCE: {
const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
MG_ASSERT(constraint_violation.properties.size() == 1U);
const auto &property_name = db_accessor.PropertyToName(*constraint_violation.properties.begin());
spdlog::warn("Trigger '{}' failed to commit due to existence constraint violation on :{}({})", trigger.Name(),
label_name, property_name);
auto maybe_commit_error = db_accessor.Commit();
if (maybe_commit_error.HasError()) {
const auto &commit_error = maybe_commit_error.GetError();
switch (commit_error.type) {
case storage::CommitError::Type::UNABLE_TO_SYNC_REPLICATE: {
// TODO(gitbuda): This is tricky because this is an internal
// operation. Consider stopping main Memgraph instance here.
spdlog::warn("Trigger '{}' failed to commit due to inability to replicate data to SYNC replica",
trigger.Name());
break;
}
case storage::ConstraintViolation::Type::UNIQUE: {
const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
std::stringstream property_names_stream;
utils::PrintIterable(property_names_stream, constraint_violation.properties, ", ",
[&](auto &stream, const auto &prop) { stream << db_accessor.PropertyToName(prop); });
spdlog::warn("Trigger '{}' failed to commit due to unique constraint violation on :{}({})", trigger.Name(),
label_name, property_names_stream.str());
case storage::CommitError::Type::CONSTRAINT_VIOLATION: {
MG_ASSERT(commit_error.maybe_constraint_violation.has_value());
const auto &constraint_violation = *commit_error.maybe_constraint_violation;
switch (constraint_violation.type) {
case storage::ConstraintViolation::Type::EXISTENCE: {
const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
MG_ASSERT(constraint_violation.properties.size() == 1U);
const auto &property_name = db_accessor.PropertyToName(*constraint_violation.properties.begin());
spdlog::warn("Trigger '{}' failed to commit due to existence constraint violation on :{}({})",
trigger.Name(), label_name, property_name);
break;
}
case storage::ConstraintViolation::Type::UNIQUE: {
const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
std::stringstream property_names_stream;
utils::PrintIterable(property_names_stream, constraint_violation.properties, ", ",
[&](auto &stream, const auto &prop) { stream << db_accessor.PropertyToName(prop); });
spdlog::warn("Trigger '{}' failed to commit due to unique constraint violation on :{}({})",
trigger.Name(), label_name, property_names_stream.str());
break;
}
}
break;
}
}
@@ -2269,28 +2286,40 @@ void Interpreter::Commit() {
trigger_context_collector_.reset();
};
auto maybe_constraint_violation = db_accessor_->Commit();
if (maybe_constraint_violation.HasError()) {
const auto &constraint_violation = maybe_constraint_violation.GetError();
switch (constraint_violation.type) {
case storage::ConstraintViolation::Type::EXISTENCE: {
auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
MG_ASSERT(constraint_violation.properties.size() == 1U);
auto property_name = execution_db_accessor_->PropertyToName(*constraint_violation.properties.begin());
auto maybe_commit_error = db_accessor_->Commit();
if (maybe_commit_error.HasError()) {
const auto &commit_error = maybe_commit_error.GetError();
switch (commit_error.type) {
case storage::CommitError::Type::UNABLE_TO_SYNC_REPLICATE: {
reset_necessary_members();
throw QueryException("Unable to commit due to existence constraint violation on :{}({})", label_name,
property_name);
throw QueryException("Unable to commit due to inability to replicate to SYNC replica");
break;
}
case storage::ConstraintViolation::Type::UNIQUE: {
auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
std::stringstream property_names_stream;
utils::PrintIterable(
property_names_stream, constraint_violation.properties, ", ",
[this](auto &stream, const auto &prop) { stream << execution_db_accessor_->PropertyToName(prop); });
reset_necessary_members();
throw QueryException("Unable to commit due to unique constraint violation on :{}({})", label_name,
property_names_stream.str());
case storage::CommitError::Type::CONSTRAINT_VIOLATION: {
MG_ASSERT(commit_error.maybe_constraint_violation.has_value());
const auto &constraint_violation = *commit_error.maybe_constraint_violation;
switch (constraint_violation.type) {
case storage::ConstraintViolation::Type::EXISTENCE: {
auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
MG_ASSERT(constraint_violation.properties.size() == 1U);
auto property_name = execution_db_accessor_->PropertyToName(*constraint_violation.properties.begin());
reset_necessary_members();
throw QueryException("Unable to commit due to existence constraint violation on :{}({})", label_name,
property_name);
break;
}
case storage::ConstraintViolation::Type::UNIQUE: {
auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
std::stringstream property_names_stream;
utils::PrintIterable(
property_names_stream, constraint_violation.properties, ", ",
[this](auto &stream, const auto &prop) { stream << execution_db_accessor_->PropertyToName(prop); });
reset_necessary_members();
throw QueryException("Unable to commit due to unique constraint violation on :{}({})", label_name,
property_names_stream.str());
break;
}
}
break;
}
}

View File

@@ -47,7 +47,7 @@ extern const Event FailedQuery;
namespace memgraph::query {
static constexpr size_t kExecutionMemoryBlockSize = 1U * 1024U * 1024U;
inline constexpr size_t kExecutionMemoryBlockSize = 1UL * 1024UL * 1024UL;
class AuthQueryHandler {
public:
@@ -137,7 +137,8 @@ class ReplicationQueryHandler {
/// @throw QueryRuntimeException if an error ocurred.
virtual void RegisterReplica(const std::string &name, const std::string &socket_address,
const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout) = 0;
const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout,
const std::chrono::seconds replica_check_frequency) = 0;
/// @throw QueryRuntimeException if an error ocurred.
virtual void DropReplica(const std::string &replica_name) = 0;

View File

@@ -61,6 +61,7 @@ class CostEstimator : public HierarchicalLogicalOperatorVisitor {
static constexpr double kFilter{1.5};
static constexpr double kEdgeUniquenessFilter{1.5};
static constexpr double kUnwind{1.3};
static constexpr double kForeach{1.0};
};
struct CardParam {
@@ -72,6 +73,7 @@ class CostEstimator : public HierarchicalLogicalOperatorVisitor {
struct MiscParam {
static constexpr double kUnwindNoLiteral{10.0};
static constexpr double kForeachNoLiteral{10.0};
};
using HierarchicalLogicalOperatorVisitor::PostVisit;
@@ -193,6 +195,23 @@ class CostEstimator : public HierarchicalLogicalOperatorVisitor {
return true;
}
bool PostVisit(Foreach &foreach) override {
// Foreach cost depends both on the number elements in the list that get unwound
// as well as the total clauses that get called for each unwounded element.
// First estimate cardinality and then increment the cost.
double foreach_elements{0};
if (auto *literal = utils::Downcast<query::ListLiteral>(foreach.expression_)) {
foreach_elements = literal->elements_.size();
} else {
foreach_elements = MiscParam::kForeachNoLiteral;
}
cardinality_ *= foreach_elements;
IncrementCost(CostParam::kForeach);
return true;
}
bool Visit(Once &) override { return true; }
auto cost() const { return cost_; }

View File

@@ -45,6 +45,7 @@
#include "utils/fnv.hpp"
#include "utils/likely.hpp"
#include "utils/logging.hpp"
#include "utils/memory.hpp"
#include "utils/pmr/unordered_map.hpp"
#include "utils/pmr/unordered_set.hpp"
#include "utils/pmr/vector.hpp"
@@ -105,6 +106,7 @@ extern const Event DistinctOperator;
extern const Event UnionOperator;
extern const Event CartesianOperator;
extern const Event CallProcedureOperator;
extern const Event ForeachOperator;
} // namespace EventCounter
namespace memgraph::query::plan {
@@ -3703,46 +3705,12 @@ void CallCustomProcedure(const std::string_view &fully_qualified_procedure_name,
"containers aware of that");
// Build and type check procedure arguments.
mgp_list proc_args(memory);
proc_args.elems.reserve(args.size());
if (args.size() < proc.args.size() ||
// Rely on `||` short circuit so we can avoid potential overflow of
// proc.args.size() + proc.opt_args.size() by subtracting.
(args.size() - proc.args.size() > proc.opt_args.size())) {
if (proc.args.empty() && proc.opt_args.empty()) {
throw QueryRuntimeException("'{}' requires no arguments.", fully_qualified_procedure_name);
} else if (proc.opt_args.empty()) {
throw QueryRuntimeException("'{}' requires exactly {} {}.", fully_qualified_procedure_name, proc.args.size(),
proc.args.size() == 1U ? "argument" : "arguments");
} else {
throw QueryRuntimeException("'{}' requires between {} and {} arguments.", fully_qualified_procedure_name,
proc.args.size(), proc.args.size() + proc.opt_args.size());
}
}
for (size_t i = 0; i < args.size(); ++i) {
auto arg = args[i]->Accept(*evaluator);
std::string_view name;
const query::procedure::CypherType *type{nullptr};
if (proc.args.size() > i) {
name = proc.args[i].first;
type = proc.args[i].second;
} else {
MG_ASSERT(proc.opt_args.size() > i - proc.args.size());
name = std::get<0>(proc.opt_args[i - proc.args.size()]);
type = std::get<1>(proc.opt_args[i - proc.args.size()]);
}
if (!type->SatisfiesType(arg)) {
throw QueryRuntimeException("'{}' argument named '{}' at position {} must be of type {}.",
fully_qualified_procedure_name, name, i, type->GetPresentableName());
}
proc_args.elems.emplace_back(std::move(arg), &graph);
}
// Fill missing optional arguments with their default values.
MG_ASSERT(args.size() >= proc.args.size());
size_t passed_in_opt_args = args.size() - proc.args.size();
MG_ASSERT(passed_in_opt_args <= proc.opt_args.size());
for (size_t i = passed_in_opt_args; i < proc.opt_args.size(); ++i) {
proc_args.elems.emplace_back(std::get<2>(proc.opt_args[i]), &graph);
std::vector<TypedValue> args_list;
args_list.reserve(args.size());
for (auto *expression : args) {
args_list.emplace_back(expression->Accept(*evaluator));
}
procedure::ConstructArguments(args_list, proc, fully_qualified_procedure_name, proc_args, graph);
if (memory_limit) {
SPDLOG_INFO("Running '{}' with memory limit of {}", fully_qualified_procedure_name,
utils::GetReadableSize(*memory_limit));
@@ -3830,7 +3798,7 @@ class CallProcedureCursor : public Cursor {
// generator like procedures which yield a new result on each invocation.
auto *memory = context.evaluation_context.memory;
auto memory_limit = EvaluateMemoryLimit(&evaluator, self_->memory_limit_, self_->memory_scale_);
mgp_graph graph{context.db_accessor, graph_view, &context};
auto graph = mgp_graph::WritableGraph(*context.db_accessor, graph_view, context);
CallCustomProcedure(self_->procedure_name_, *proc, self_->arguments_, graph, &evaluator, memory, memory_limit,
&result_);
@@ -4024,4 +3992,85 @@ UniqueCursorPtr LoadCsv::MakeCursor(utils::MemoryResource *mem) const {
return MakeUniqueCursorPtr<LoadCsvCursor>(mem, this, mem);
};
class ForeachCursor : public Cursor {
public:
explicit ForeachCursor(const Foreach &foreach, utils::MemoryResource *mem)
: loop_variable_symbol_(foreach.loop_variable_symbol_),
input_(foreach.input_->MakeCursor(mem)),
updates_(foreach.update_clauses_->MakeCursor(mem)),
expression(foreach.expression_) {}
bool Pull(Frame &frame, ExecutionContext &context) override {
SCOPED_PROFILE_OP(op_name_);
if (!input_->Pull(frame, context)) {
return false;
}
ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, context.db_accessor,
storage::View::NEW);
TypedValue expr_result = expression->Accept(evaluator);
if (expr_result.IsNull()) {
return true;
}
if (!expr_result.IsList()) {
throw QueryRuntimeException("FOREACH expression must resolve to a list, but got '{}'.", expr_result.type());
}
const auto &cache_ = expr_result.ValueList();
for (const auto &index : cache_) {
frame[loop_variable_symbol_] = index;
while (updates_->Pull(frame, context)) {
}
ResetUpdates();
}
return true;
}
void Shutdown() override { input_->Shutdown(); }
void ResetUpdates() { updates_->Reset(); }
void Reset() override {
input_->Reset();
ResetUpdates();
}
private:
const Symbol loop_variable_symbol_;
const UniqueCursorPtr input_;
const UniqueCursorPtr updates_;
Expression *expression;
const char *op_name_{"Foreach"};
};
Foreach::Foreach(std::shared_ptr<LogicalOperator> input, std::shared_ptr<LogicalOperator> updates, Expression *expr,
Symbol loop_variable_symbol)
: input_(input ? std::move(input) : std::make_shared<Once>()),
update_clauses_(std::move(updates)),
expression_(expr),
loop_variable_symbol_(loop_variable_symbol) {}
UniqueCursorPtr Foreach::MakeCursor(utils::MemoryResource *mem) const {
EventCounter::IncrementCounter(EventCounter::ForeachOperator);
return MakeUniqueCursorPtr<ForeachCursor>(mem, *this, mem);
}
std::vector<Symbol> Foreach::ModifiedSymbols(const SymbolTable &table) const {
auto symbols = input_->ModifiedSymbols(table);
symbols.emplace_back(loop_variable_symbol_);
return symbols;
}
bool Foreach::Accept(HierarchicalLogicalOperatorVisitor &visitor) {
if (visitor.PreVisit(*this)) {
input_->Accept(visitor);
update_clauses_->Accept(visitor);
}
return visitor.PostVisit(*this);
}
} // namespace memgraph::query::plan

View File

@@ -131,6 +131,7 @@ class Union;
class Cartesian;
class CallProcedure;
class LoadCsv;
class Foreach;
using LogicalOperatorCompositeVisitor = utils::CompositeVisitor<
Once, CreateNode, CreateExpand, ScanAll, ScanAllByLabel,
@@ -139,7 +140,7 @@ using LogicalOperatorCompositeVisitor = utils::CompositeVisitor<
Expand, ExpandVariable, ConstructNamedPath, Filter, Produce, Delete,
SetProperty, SetProperties, SetLabels, RemoveProperty, RemoveLabels,
EdgeUniquenessFilter, Accumulate, Aggregate, Skip, Limit, OrderBy, Merge,
Optional, Unwind, Distinct, Union, Cartesian, CallProcedure, LoadCsv>;
Optional, Unwind, Distinct, Union, Cartesian, CallProcedure, LoadCsv, Foreach>;
using LogicalOperatorLeafVisitor = utils::LeafVisitor<Once>;
@@ -2261,6 +2262,42 @@ at once. Instead, each call of the callback should return a single row of the ta
(:serialize (:slk))
(:clone))
(lcp:define-class foreach (logical-operator)
((input "std::shared_ptr<LogicalOperator>" :scope :public
:slk-save #'slk-save-operator-pointer
:slk-load #'slk-load-operator-pointer)
(update-clauses "std::shared_ptr<LogicalOperator>" :scope :public
:slk-save #'slk-save-operator-pointer
:slk-load #'slk-load-operator-pointer)
(expression "Expression *" :scope :public
:slk-save #'slk-save-ast-pointer
:slk-load (slk-load-ast-pointer "Expression"))
(loop-variable-symbol "Symbol" :scope :public))
(:documentation
"Iterates over a collection of elements and applies one or more update
clauses.
")
(:public
#>cpp
Foreach() = default;
Foreach(std::shared_ptr<LogicalOperator> input,
std::shared_ptr<LogicalOperator> updates,
Expression *named_expr,
Symbol loop_variable_symbol);
bool Accept(HierarchicalLogicalOperatorVisitor &visitor) override;
UniqueCursorPtr MakeCursor(utils::MemoryResource *) const override;
std::vector<Symbol> ModifiedSymbols(const SymbolTable &) const override;
bool HasSingleInput() const override { return true; }
std::shared_ptr<LogicalOperator> input() const override { return input_; }
void set_input(std::shared_ptr<LogicalOperator> input) override {
input_ = std::move(input);
}
cpp<#)
(:serialize (:slk))
(:clone))
(lcp:pop-namespace) ;; plan
(lcp:pop-namespace) ;; query
(lcp:pop-namespace) ;; memgraph

View File

@@ -17,8 +17,10 @@
#include <variant>
#include "query/exceptions.hpp"
#include "query/frontend/ast/ast.hpp"
#include "query/frontend/ast/ast_visitor.hpp"
#include "query/plan/preprocess.hpp"
#include "utils/typeinfo.hpp"
namespace memgraph::query::plan {
@@ -526,6 +528,18 @@ void Filters::AnalyzeAndStoreFilter(Expression *expr, const SymbolTable &symbol_
// as `expr1 < n.prop AND n.prop < expr2`.
}
static void ParseForeach(query::Foreach &foreach, SingleQueryPart &query_part, AstStorage &storage,
SymbolTable &symbol_table) {
for (auto *clause : foreach.clauses_) {
if (auto *merge = utils::Downcast<query::Merge>(clause)) {
query_part.merge_matching.emplace_back(Matching{});
AddMatching({merge->pattern_}, nullptr, symbol_table, storage, query_part.merge_matching.back());
} else if (auto *nested = utils::Downcast<query::Foreach>(clause)) {
ParseForeach(*nested, query_part, storage, symbol_table);
}
}
}
// Converts a Query to multiple QueryParts. In the process new Ast nodes may be
// created, e.g. filter expressions.
std::vector<SingleQueryPart> CollectSingleQueryParts(SymbolTable &symbol_table, AstStorage &storage,
@@ -546,6 +560,8 @@ std::vector<SingleQueryPart> CollectSingleQueryParts(SymbolTable &symbol_table,
if (auto *merge = utils::Downcast<query::Merge>(clause)) {
query_part->merge_matching.emplace_back(Matching{});
AddMatching({merge->pattern_}, nullptr, symbol_table, storage, query_part->merge_matching.back());
} else if (auto *foreach = utils::Downcast<query::Foreach>(clause)) {
ParseForeach(*foreach, *query_part, storage, symbol_table);
} else if (utils::IsSubtype(*clause, With::kType) || utils::IsSubtype(*clause, query::Unwind::kType) ||
utils::IsSubtype(*clause, query::CallProcedure::kType) ||
utils::IsSubtype(*clause, query::LoadCsv::kType)) {

View File

@@ -306,6 +306,10 @@ struct Matching {
/// will produce the second `merge_matching` element. This way, if someone
/// traverses `remaining_clauses`, the order of appearance of `Merge` clauses is
/// in the same order as their respective `merge_matching` elements.
/// An exception to the above rule is Foreach. Its update clauses will not be contained in
/// the `remaining_clauses`, but rather inside the foreach itself. The order guarantee is not
/// violated because the update clauses of the foreach are immediately processed in
/// the `RuleBasedPlanner` as if as they were pushed into the `remaining_clauses`.
struct SingleQueryPart {
/// @brief All `MATCH` clauses merged into one @c Matching.
Matching matching;
@@ -320,6 +324,10 @@ struct SingleQueryPart {
///
/// Since @c Merge is contained in `remaining_clauses`, this vector contains
/// matching in the same order as @c Merge appears.
//
/// Foreach @c does not violate this gurantee. However, update clauses are not stored
/// in the `remaining_clauses` but rather in the `Foreach` itself and are guranteed
/// to be processed in the same order by the semantics of the `RuleBasedPlanner`.
std::vector<Matching> merge_matching{};
/// @brief All the remaining clauses (without @c Match).
std::vector<Clause *> remaining_clauses{};

View File

@@ -241,6 +241,12 @@ bool PlanPrinter::PreVisit(query::plan::Cartesian &op) {
return false;
}
bool PlanPrinter::PreVisit(query::plan::Foreach &op) {
WithPrintLn([](auto &out) { out << "* Foreach"; });
Branch(*op.update_clauses_);
op.input_->Accept(*this);
return false;
}
#undef PRE_VISIT
bool PlanPrinter::DefaultPreVisit() {
@@ -883,6 +889,21 @@ bool PlanToJsonVisitor::PreVisit(Cartesian &op) {
output_ = std::move(self);
return false;
}
bool PlanToJsonVisitor::PreVisit(Foreach &op) {
json self;
self["name"] = "Foreach";
self["loop_variable_symbol"] = ToJson(op.loop_variable_symbol_);
self["expression"] = ToJson(op.expression_);
op.input_->Accept(*this);
self["input"] = PopOutput();
op.update_clauses_->Accept(*this);
self["update_clauses"] = PopOutput();
output_ = std::move(self);
return false;
}
} // namespace impl

View File

@@ -92,6 +92,7 @@ class PlanPrinter : public virtual HierarchicalLogicalOperatorVisitor {
bool PreVisit(Unwind &) override;
bool PreVisit(CallProcedure &) override;
bool PreVisit(LoadCsv &) override;
bool PreVisit(Foreach &) override;
bool Visit(Once &) override;
@@ -204,6 +205,7 @@ class PlanToJsonVisitor : public virtual HierarchicalLogicalOperatorVisitor {
bool PreVisit(Union &) override;
bool PreVisit(Unwind &) override;
bool PreVisit(Foreach &) override;
bool PreVisit(CallProcedure &) override;
bool PreVisit(LoadCsv &) override;

View File

@@ -79,6 +79,11 @@ bool ReadWriteTypeChecker::PreVisit(CallProcedure &op) {
return true;
}
bool ReadWriteTypeChecker::PreVisit([[maybe_unused]] Foreach &op) {
UpdateType(RWType::RW);
return false;
}
#undef PRE_VISIT
bool ReadWriteTypeChecker::Visit(Once &op) { return false; }

View File

@@ -84,6 +84,7 @@ class ReadWriteTypeChecker : public virtual HierarchicalLogicalOperatorVisitor {
bool PreVisit(Unwind &) override;
bool PreVisit(CallProcedure &) override;
bool PreVisit(Foreach &) override;
bool Visit(Once &) override;

View File

@@ -433,6 +433,16 @@ class IndexLookupRewriter final : public HierarchicalLogicalOperatorVisitor {
return true;
}
bool PreVisit(Foreach &op) override {
prev_ops_.push_back(&op);
return false;
}
bool PostVisit(Foreach &) override {
prev_ops_.pop_back();
return true;
}
std::shared_ptr<LogicalOperator> new_root_;
private:

View File

@@ -22,6 +22,7 @@
#include "query/plan/operator.hpp"
#include "query/plan/preprocess.hpp"
#include "utils/logging.hpp"
#include "utils/typeinfo.hpp"
namespace memgraph::query::plan {
@@ -223,6 +224,10 @@ class RuleBasedPlanner {
input_op =
std::make_unique<plan::LoadCsv>(std::move(input_op), load_csv->file_, load_csv->with_header_,
load_csv->ignore_bad_, load_csv->delimiter_, load_csv->quote_, row_sym);
} else if (auto *foreach = utils::Downcast<query::Foreach>(clause)) {
is_write = true;
input_op = HandleForeachClause(foreach, std::move(input_op), *context.symbol_table, context.bound_symbols,
query_part, merge_id);
} else {
throw utils::NotYetImplemented("clause '{}' conversion to operator(s)", clause->GetTypeInfo().name);
}
@@ -530,6 +535,27 @@ class RuleBasedPlanner {
}
return std::make_unique<plan::Merge>(std::move(input_op), std::move(on_match), std::move(on_create));
}
std::unique_ptr<LogicalOperator> HandleForeachClause(query::Foreach *foreach,
std::unique_ptr<LogicalOperator> input_op,
const SymbolTable &symbol_table,
std::unordered_set<Symbol> &bound_symbols,
const SingleQueryPart &query_part, uint64_t &merge_id) {
const auto &symbol = symbol_table.at(*foreach->named_expression_);
bound_symbols.insert(symbol);
std::unique_ptr<LogicalOperator> op = std::make_unique<plan::Once>();
for (auto *clause : foreach->clauses_) {
if (auto *nested_for_each = utils::Downcast<query::Foreach>(clause)) {
op = HandleForeachClause(nested_for_each, std::move(op), symbol_table, bound_symbols, query_part, merge_id);
} else if (auto *merge = utils::Downcast<query::Merge>(clause)) {
op = GenMerge(*merge, std::move(op), query_part.merge_matching[merge_id++]);
} else {
op = HandleWriteClause(clause, op, symbol_table, bound_symbols);
}
}
return std::make_unique<plan::Foreach>(std::move(input_op), std::move(op), foreach->named_expression_->expression_,
symbol);
}
};
} // namespace memgraph::query::plan

View File

@@ -71,9 +71,9 @@ class ScopedProfile {
private:
query::ExecutionContext *context_;
ProfilingStats *root_;
ProfilingStats *stats_;
unsigned long long start_time_;
ProfilingStats *root_{nullptr};
ProfilingStats *stats_{nullptr};
unsigned long long start_time_{0};
};
} // namespace memgraph::query::plan

View File

@@ -24,7 +24,7 @@ MgpUniquePtr<mgp_value> GetStringValueOrSetError(const char *string, mgp_memory
}
bool InsertResultOrSetError(mgp_result *result, mgp_result_record *record, const char *result_name, mgp_value *value) {
if (const auto err = mgp_result_record_insert(record, result_name, value); err != MGP_ERROR_NO_ERROR) {
if (const auto err = mgp_result_record_insert(record, result_name, value); err != mgp_error::MGP_ERROR_NO_ERROR) {
const auto error_msg = fmt::format("Unable to set the result for {}, error = {}", result_name, err);
static_cast<void>(mgp_result_set_error_msg(result, error_msg.c_str()));
return false;

View File

@@ -25,7 +25,7 @@ TResult Call(TFunc func, TArgs... args) {
static_assert(std::is_trivially_copyable_v<TFunc>);
static_assert((std::is_trivially_copyable_v<std::remove_reference_t<TArgs>> && ...));
TResult result{};
MG_ASSERT(func(args..., &result) == MGP_ERROR_NO_ERROR);
MG_ASSERT(func(args..., &result) == mgp_error::MGP_ERROR_NO_ERROR);
return result;
}
@@ -50,10 +50,10 @@ mgp_error CreateMgpObject(MgpUniquePtr<TObj> &obj, TFunc func, TArgs &&...args)
template <typename Fun>
[[nodiscard]] bool TryOrSetError(Fun &&func, mgp_result *result) {
if (const auto err = func(); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = func(); err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
static_cast<void>(mgp_result_set_error_msg(result, "Not enough memory!"));
return false;
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
const auto error_msg = fmt::format("Unexpected error ({})!", err);
static_cast<void>(mgp_result_set_error_msg(result, error_msg.c_str()));
return false;

View File

@@ -143,51 +143,54 @@ template <typename TFunc, typename... Args>
WrapExceptionsHelper(std::forward<TFunc>(func), std::forward<Args>(args)...);
} catch (const DeletedObjectException &neoe) {
spdlog::error("Deleted object error during mg API call: {}", neoe.what());
return MGP_ERROR_DELETED_OBJECT;
return mgp_error::MGP_ERROR_DELETED_OBJECT;
} catch (const KeyAlreadyExistsException &kaee) {
spdlog::error("Key already exists error during mg API call: {}", kaee.what());
return MGP_ERROR_KEY_ALREADY_EXISTS;
return mgp_error::MGP_ERROR_KEY_ALREADY_EXISTS;
} catch (const InsufficientBufferException &ibe) {
spdlog::error("Insufficient buffer error during mg API call: {}", ibe.what());
return MGP_ERROR_INSUFFICIENT_BUFFER;
return mgp_error::MGP_ERROR_INSUFFICIENT_BUFFER;
} catch (const ImmutableObjectException &ioe) {
spdlog::error("Immutable object error during mg API call: {}", ioe.what());
return MGP_ERROR_IMMUTABLE_OBJECT;
return mgp_error::MGP_ERROR_IMMUTABLE_OBJECT;
} catch (const ValueConversionException &vce) {
spdlog::error("Value converion error during mg API call: {}", vce.what());
return MGP_ERROR_VALUE_CONVERSION;
return mgp_error::MGP_ERROR_VALUE_CONVERSION;
} catch (const SerializationException &se) {
spdlog::error("Serialization error during mg API call: {}", se.what());
return MGP_ERROR_SERIALIZATION_ERROR;
return mgp_error::MGP_ERROR_SERIALIZATION_ERROR;
} catch (const std::bad_alloc &bae) {
spdlog::error("Memory allocation error during mg API call: {}", bae.what());
return MGP_ERROR_UNABLE_TO_ALLOCATE;
return mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE;
} catch (const memgraph::utils::OutOfMemoryException &oome) {
spdlog::error("Memory limit exceeded during mg API call: {}", oome.what());
return MGP_ERROR_UNABLE_TO_ALLOCATE;
return mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE;
} catch (const std::out_of_range &oore) {
spdlog::error("Out of range error during mg API call: {}", oore.what());
return MGP_ERROR_OUT_OF_RANGE;
return mgp_error::MGP_ERROR_OUT_OF_RANGE;
} catch (const std::invalid_argument &iae) {
spdlog::error("Invalid argument error during mg API call: {}", iae.what());
return MGP_ERROR_INVALID_ARGUMENT;
return mgp_error::MGP_ERROR_INVALID_ARGUMENT;
} catch (const std::logic_error &lee) {
spdlog::error("Logic error during mg API call: {}", lee.what());
return MGP_ERROR_LOGIC_ERROR;
return mgp_error::MGP_ERROR_LOGIC_ERROR;
} catch (const std::exception &e) {
spdlog::error("Unexpected error during mg API call: {}", e.what());
return MGP_ERROR_UNKNOWN_ERROR;
return mgp_error::MGP_ERROR_UNKNOWN_ERROR;
} catch (const memgraph::utils::temporal::InvalidArgumentException &e) {
spdlog::error("Invalid argument was sent to an mg API call for temporal types: {}", e.what());
return MGP_ERROR_INVALID_ARGUMENT;
return mgp_error::MGP_ERROR_INVALID_ARGUMENT;
} catch (...) {
spdlog::error("Unexpected error during mg API call");
return MGP_ERROR_UNKNOWN_ERROR;
return mgp_error::MGP_ERROR_UNKNOWN_ERROR;
}
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
bool MgpGraphIsMutable(const mgp_graph &graph) noexcept { return graph.view == memgraph::storage::View::NEW; }
// Graph mutations
bool MgpGraphIsMutable(const mgp_graph &graph) noexcept {
return graph.view == memgraph::storage::View::NEW && graph.ctx != nullptr;
}
bool MgpVertexIsMutable(const mgp_vertex &vertex) { return MgpGraphIsMutable(*vertex.graph); }
@@ -289,6 +292,7 @@ mgp_value_type FromTypedValueType(memgraph::query::TypedValue::Type type) {
return MGP_VALUE_TYPE_DURATION;
}
}
} // namespace
memgraph::query::TypedValue ToTypedValue(const mgp_value &val, memgraph::utils::MemoryResource *memory) {
switch (val.type) {
@@ -345,8 +349,6 @@ memgraph::query::TypedValue ToTypedValue(const mgp_value &val, memgraph::utils::
}
}
} // namespace
mgp_value::mgp_value(memgraph::utils::MemoryResource *m) noexcept : type(MGP_VALUE_TYPE_NULL), memory(m) {}
mgp_value::mgp_value(bool val, memgraph::utils::MemoryResource *m) noexcept
@@ -844,7 +846,7 @@ mgp_value_type MgpValueGetType(const mgp_value &val) noexcept { return val.type;
mgp_error mgp_value_get_type(mgp_value *val, mgp_value_type *result) {
static_assert(noexcept(MgpValueGetType(*val)));
*result = MgpValueGetType(*val);
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
@@ -852,7 +854,7 @@ mgp_error mgp_value_get_type(mgp_value *val, mgp_value_type *result) {
mgp_error mgp_value_is_##type_lowercase(mgp_value *val, int *result) { \
static_assert(noexcept(MgpValueGetType(*val))); \
*result = MgpValueGetType(*val) == MGP_VALUE_TYPE_##type_uppercase; \
return MGP_ERROR_NO_ERROR; \
return mgp_error::MGP_ERROR_NO_ERROR; \
}
DEFINE_MGP_VALUE_IS(null, NULL)
@@ -872,27 +874,27 @@ DEFINE_MGP_VALUE_IS(duration, DURATION)
mgp_error mgp_value_get_bool(mgp_value *val, int *result) {
*result = val->bool_v ? 1 : 0;
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_value_get_int(mgp_value *val, int64_t *result) {
*result = val->int_v;
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_value_get_double(mgp_value *val, double *result) {
*result = val->double_v;
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_value_get_string(mgp_value *val, const char **result) {
static_assert(noexcept(val->string_v.c_str()));
*result = val->string_v.c_str();
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define DEFINE_MGP_VALUE_GET(type) \
mgp_error mgp_value_get_##type(mgp_value *val, mgp_##type **result) { \
*result = val->type##_v; \
return MGP_ERROR_NO_ERROR; \
return mgp_error::MGP_ERROR_NO_ERROR; \
}
DEFINE_MGP_VALUE_GET(list)
@@ -938,13 +940,13 @@ mgp_error mgp_list_append_extend(mgp_list *list, mgp_value *val) {
mgp_error mgp_list_size(mgp_list *list, size_t *result) {
static_assert(noexcept(list->elems.size()));
*result = list->elems.size();
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_list_capacity(mgp_list *list, size_t *result) {
static_assert(noexcept(list->elems.capacity()));
*result = list->elems.capacity();
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_list_at(mgp_list *list, size_t i, mgp_value **result) {
@@ -976,7 +978,7 @@ mgp_error mgp_map_insert(mgp_map *map, const char *key, mgp_value *value) {
mgp_error mgp_map_size(mgp_map *map, size_t *result) {
static_assert(noexcept(map->items.size()));
*result = map->items.size();
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_map_at(mgp_map *map, const char *key, mgp_value **result) {
@@ -1087,7 +1089,7 @@ size_t MgpPathSize(const mgp_path &path) noexcept { return path.edges.size(); }
mgp_error mgp_path_size(mgp_path *path, size_t *result) {
*result = MgpPathSize(*path);
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_path_vertex_at(mgp_path *path, size_t i, mgp_vertex **result) {
@@ -1451,6 +1453,14 @@ mgp_error mgp_result_record_insert(mgp_result_record *record, const char *field_
});
}
mgp_error mgp_func_result_set_error_msg(mgp_func_result *res, const char *msg, mgp_memory *memory) {
return WrapExceptions([=] { res->error_msg.emplace(msg, memory->impl); });
}
mgp_error mgp_func_result_set_value(mgp_func_result *res, mgp_value *value, mgp_memory *memory) {
return WrapExceptions([=] { res->value = ToTypedValue(*value, memory->impl); });
}
/// Graph Constructs
void mgp_properties_iterator_destroy(mgp_properties_iterator *it) { DeleteRawMgpObject(it); }
@@ -1680,7 +1690,7 @@ mgp_error mgp_vertex_equal(mgp_vertex *v1, mgp_vertex *v2, int *result) {
// NOLINTNEXTLINE(clang-diagnostic-unevaluated-expression)
static_assert(noexcept(*result = *v1 == *v2 ? 1 : 0));
*result = *v1 == *v2 ? 1 : 0;
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_vertex_labels_count(mgp_vertex *v, size_t *result) {
@@ -1940,7 +1950,7 @@ mgp_error mgp_edge_equal(mgp_edge *e1, mgp_edge *e2, int *result) {
// NOLINTNEXTLINE(clang-diagnostic-unevaluated-expression)
static_assert(noexcept(*result = *e1 == *e2 ? 1 : 0));
*result = *e1 == *e2 ? 1 : 0;
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_edge_get_type(mgp_edge *e, mgp_edge_type *result) {
@@ -1957,12 +1967,12 @@ mgp_error mgp_edge_get_type(mgp_edge *e, mgp_edge_type *result) {
mgp_error mgp_edge_get_from(mgp_edge *e, mgp_vertex **result) {
*result = &e->from;
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_edge_get_to(mgp_edge *e, mgp_vertex **result) {
*result = &e->to;
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_edge_get_property(mgp_edge *e, const char *name, mgp_memory *memory, mgp_value **result) {
@@ -2072,7 +2082,7 @@ mgp_error mgp_graph_get_vertex_by_id(mgp_graph *graph, mgp_vertex_id id, mgp_mem
mgp_error mgp_graph_is_mutable(mgp_graph *graph, int *result) {
*result = MgpGraphIsMutable(*graph) ? 1 : 0;
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
};
mgp_error mgp_graph_create_vertex(struct mgp_graph *graph, mgp_memory *memory, mgp_vertex **result) {
@@ -2382,31 +2392,53 @@ mgp_error mgp_module_add_write_procedure(mgp_module *module, const char *name, m
return WrapExceptions([=] { return mgp_module_add_procedure(module, name, cb, {.is_write = true}); }, result);
}
mgp_error mgp_proc_add_arg(mgp_proc *proc, const char *name, mgp_type *type) {
return WrapExceptions([=] {
if (!IsValidIdentifierName(name)) {
throw std::invalid_argument{fmt::format("Invalid argument name for procedure '{}': {}", proc->name, name)};
namespace {
template <typename T>
concept IsCallable = memgraph::utils::SameAsAnyOf<T, mgp_proc, mgp_func>;
template <IsCallable TCall>
mgp_error MgpAddArg(TCall &callable, const std::string &name, mgp_type &type) {
return WrapExceptions([&]() mutable {
static constexpr std::string_view type_name = std::invoke([]() constexpr {
if constexpr (std::is_same_v<TCall, mgp_proc>) {
return "procedure";
} else if constexpr (std::is_same_v<TCall, mgp_func>) {
return "function";
}
});
if (!IsValidIdentifierName(name.c_str())) {
throw std::invalid_argument{fmt::format("Invalid argument name for {} '{}': {}", type_name, callable.name, name)};
}
if (!proc->opt_args.empty()) {
throw std::logic_error{fmt::format(
"Cannot add required argument '{}' to procedure '{}' after adding any optional one", name, proc->name)};
if (!callable.opt_args.empty()) {
throw std::logic_error{fmt::format("Cannot add required argument '{}' to {} '{}' after adding any optional one",
name, type_name, callable.name)};
}
proc->args.emplace_back(name, type->impl.get());
callable.args.emplace_back(name, type.impl.get());
});
}
mgp_error mgp_proc_add_opt_arg(mgp_proc *proc, const char *name, mgp_type *type, mgp_value *default_value) {
return WrapExceptions([=] {
if (!IsValidIdentifierName(name)) {
throw std::invalid_argument{fmt::format("Invalid argument name for procedure '{}': {}", proc->name, name)};
template <IsCallable TCall>
mgp_error MgpAddOptArg(TCall &callable, const std::string name, mgp_type &type, mgp_value &default_value) {
return WrapExceptions([&]() mutable {
static constexpr std::string_view type_name = std::invoke([]() constexpr {
if constexpr (std::is_same_v<TCall, mgp_proc>) {
return "procedure";
} else if constexpr (std::is_same_v<TCall, mgp_func>) {
return "function";
}
});
if (!IsValidIdentifierName(name.c_str())) {
throw std::invalid_argument{fmt::format("Invalid argument name for {} '{}': {}", type_name, callable.name, name)};
}
switch (MgpValueGetType(*default_value)) {
switch (MgpValueGetType(default_value)) {
case MGP_VALUE_TYPE_VERTEX:
case MGP_VALUE_TYPE_EDGE:
case MGP_VALUE_TYPE_PATH:
// default_value must not be a graph element.
throw ValueConversionException{
"Default value of argument '{}' of procedure '{}' name must not be a graph element!", name, proc->name};
throw ValueConversionException{"Default value of argument '{}' of {} '{}' name must not be a graph element!",
name, type_name, callable.name};
case MGP_VALUE_TYPE_NULL:
case MGP_VALUE_TYPE_BOOL:
case MGP_VALUE_TYPE_INT:
@@ -2421,16 +2453,32 @@ mgp_error mgp_proc_add_opt_arg(mgp_proc *proc, const char *name, mgp_type *type,
break;
}
// Default value must be of required `type`.
if (!type->impl->SatisfiesType(*default_value)) {
throw std::logic_error{
fmt::format("The default value of argument '{}' for procedure '{}' doesn't satisfy type '{}'", name,
proc->name, type->impl->GetPresentableName())};
if (!type.impl->SatisfiesType(default_value)) {
throw std::logic_error{fmt::format("The default value of argument '{}' for {} '{}' doesn't satisfy type '{}'",
name, type_name, callable.name, type.impl->GetPresentableName())};
}
auto *memory = proc->opt_args.get_allocator().GetMemoryResource();
proc->opt_args.emplace_back(memgraph::utils::pmr::string(name, memory), type->impl.get(),
ToTypedValue(*default_value, memory));
auto *memory = callable.opt_args.get_allocator().GetMemoryResource();
callable.opt_args.emplace_back(memgraph::utils::pmr::string(name, memory), type.impl.get(),
ToTypedValue(default_value, memory));
});
}
} // namespace
mgp_error mgp_proc_add_arg(mgp_proc *proc, const char *name, mgp_type *type) {
return MgpAddArg(*proc, std::string(name), *type);
}
mgp_error mgp_proc_add_opt_arg(mgp_proc *proc, const char *name, mgp_type *type, mgp_value *default_value) {
return MgpAddOptArg(*proc, std::string(name), *type, *default_value);
}
mgp_error mgp_func_add_arg(mgp_func *func, const char *name, mgp_type *type) {
return MgpAddArg(*func, std::string(name), *type);
}
mgp_error mgp_func_add_opt_arg(mgp_func *func, const char *name, mgp_type *type, mgp_value *default_value) {
return MgpAddOptArg(*func, std::string(name), *type, *default_value);
}
namespace {
@@ -2459,7 +2507,7 @@ mgp_error mgp_proc_add_result(mgp_proc *proc, const char *name, mgp_type *type)
mgp_error MgpTransAddFixedResult(mgp_trans *trans) noexcept {
if (const auto err = AddResultToProp(trans, "query", Call<mgp_type *>(mgp_type_string), false);
err != MGP_ERROR_NO_ERROR) {
err != mgp_error::MGP_ERROR_NO_ERROR) {
return err;
}
return AddResultToProp(trans, "parameters", Call<mgp_type *>(mgp_type_nullable, Call<mgp_type *>(mgp_type_map)),
@@ -2545,6 +2593,22 @@ void PrintProcSignature(const mgp_proc &proc, std::ostream *stream) {
(*stream) << ")";
}
void PrintFuncSignature(const mgp_func &func, std::ostream &stream) {
stream << func.name << "(";
utils::PrintIterable(stream, func.args, ", ", [](auto &stream, const auto &arg) {
stream << arg.first << " :: " << arg.second->GetPresentableName();
});
if (!func.args.empty() && !func.opt_args.empty()) {
stream << ", ";
}
utils::PrintIterable(stream, func.opt_args, ", ", [](auto &stream, const auto &arg) {
const auto &[name, type, default_val] = arg;
stream << name << " = ";
PrintValue(default_val, &stream) << " :: " << type->GetPresentableName();
});
stream << ")";
}
bool IsValidIdentifierName(const char *name) {
if (!name) return false;
std::regex regex("[_[:alpha:]][_[:alnum:]]*");
@@ -2690,7 +2754,7 @@ mgp_error mgp_message_offset(struct mgp_message *message, int64_t *result) {
mgp_error mgp_messages_size(mgp_messages *messages, size_t *result) {
static_assert(noexcept(messages->messages.size()));
*result = messages->messages.size();
return MGP_ERROR_NO_ERROR;
return mgp_error::MGP_ERROR_NO_ERROR;
}
mgp_error mgp_messages_at(mgp_messages *messages, size_t index, mgp_message **result) {
@@ -2716,3 +2780,19 @@ mgp_error mgp_module_add_transformation(mgp_module *module, const char *name, mg
module->transformations.emplace(name, mgp_trans(name, cb, memory));
});
}
mgp_error mgp_module_add_function(mgp_module *module, const char *name, mgp_func_cb cb, mgp_func **result) {
return WrapExceptions(
[=] {
if (!IsValidIdentifierName(name)) {
throw std::invalid_argument{fmt::format("Invalid function name: {}", name)};
}
if (module->functions.find(name) != module->functions.end()) {
throw std::logic_error{fmt::format("Function with similar name already exists '{}'", name)};
};
auto *memory = module->functions.get_allocator().GetMemoryResource();
return &module->functions.emplace(name, mgp_func(name, cb, memory)).first->second;
},
result);
}

View File

@@ -562,14 +562,36 @@ struct mgp_result {
std::optional<memgraph::utils::pmr::string> error_msg;
};
struct mgp_func_result {
mgp_func_result() {}
/// Return Magic function result. If user forgets it, the error is raised
std::optional<memgraph::query::TypedValue> value;
/// Return Magic function result with potential error
std::optional<memgraph::utils::pmr::string> error_msg;
};
struct mgp_graph {
memgraph::query::DbAccessor *impl;
memgraph::storage::View view;
// TODO: Merge `mgp_graph` and `mgp_memory` into a single `mgp_context`. The
// `ctx` field is out of place here.
memgraph::query::ExecutionContext *ctx;
static mgp_graph WritableGraph(memgraph::query::DbAccessor &acc, memgraph::storage::View view,
memgraph::query::ExecutionContext &ctx) {
return mgp_graph{&acc, view, &ctx};
}
static mgp_graph NonWritableGraph(memgraph::query::DbAccessor &acc, memgraph::storage::View view) {
return mgp_graph{&acc, view, nullptr};
}
};
// Prevents user to use ExecutionContext in writable callables
struct mgp_func_context {
memgraph::query::DbAccessor *impl;
memgraph::storage::View view;
};
struct mgp_properties_iterator {
using allocator_type = memgraph::utils::Allocator<mgp_properties_iterator>;
@@ -779,18 +801,69 @@ struct mgp_trans {
results;
};
struct mgp_func {
using allocator_type = memgraph::utils::Allocator<mgp_func>;
/// @throw std::bad_alloc
/// @throw std::length_error
mgp_func(const char *name, mgp_func_cb cb, memgraph::utils::MemoryResource *memory)
: name(name, memory), cb(cb), args(memory), opt_args(memory) {}
/// @throw std::bad_alloc
/// @throw std::length_error
mgp_func(const char *name, std::function<void(mgp_list *, mgp_func_context *, mgp_func_result *, mgp_memory *)> cb,
memgraph::utils::MemoryResource *memory)
: name(name, memory), cb(cb), args(memory), opt_args(memory) {}
/// @throw std::bad_alloc
/// @throw std::length_error
mgp_func(const mgp_func &other, memgraph::utils::MemoryResource *memory)
: name(other.name, memory), cb(other.cb), args(other.args, memory), opt_args(other.opt_args, memory) {}
mgp_func(mgp_func &&other, memgraph::utils::MemoryResource *memory)
: name(std::move(other.name), memory),
cb(std::move(other.cb)),
args(std::move(other.args), memory),
opt_args(std::move(other.opt_args), memory) {}
mgp_func(const mgp_func &other) = default;
mgp_func(mgp_func &&other) = default;
mgp_func &operator=(const mgp_func &) = delete;
mgp_func &operator=(mgp_func &&) = delete;
~mgp_func() = default;
/// Name of the function.
memgraph::utils::pmr::string name;
/// Entry-point for the function.
std::function<void(mgp_list *, mgp_func_context *, mgp_func_result *, mgp_memory *)> cb;
/// Required, positional arguments as a (name, type) pair.
memgraph::utils::pmr::vector<std::pair<memgraph::utils::pmr::string, const memgraph::query::procedure::CypherType *>>
args;
/// Optional positional arguments as a (name, type, default_value) tuple.
memgraph::utils::pmr::vector<std::tuple<memgraph::utils::pmr::string, const memgraph::query::procedure::CypherType *,
memgraph::query::TypedValue>>
opt_args;
};
mgp_error MgpTransAddFixedResult(mgp_trans *trans) noexcept;
struct mgp_module {
using allocator_type = memgraph::utils::Allocator<mgp_module>;
explicit mgp_module(memgraph::utils::MemoryResource *memory) : procedures(memory), transformations(memory) {}
explicit mgp_module(memgraph::utils::MemoryResource *memory)
: procedures(memory), transformations(memory), functions(memory) {}
mgp_module(const mgp_module &other, memgraph::utils::MemoryResource *memory)
: procedures(other.procedures, memory), transformations(other.transformations, memory) {}
: procedures(other.procedures, memory),
transformations(other.transformations, memory),
functions(other.functions, memory) {}
mgp_module(mgp_module &&other, memgraph::utils::MemoryResource *memory)
: procedures(std::move(other.procedures), memory), transformations(std::move(other.transformations), memory) {}
: procedures(std::move(other.procedures), memory),
transformations(std::move(other.transformations), memory),
functions(std::move(other.functions), memory) {}
mgp_module(const mgp_module &) = default;
mgp_module(mgp_module &&) = default;
@@ -802,6 +875,7 @@ struct mgp_module {
memgraph::utils::pmr::map<memgraph::utils::pmr::string, mgp_proc> procedures;
memgraph::utils::pmr::map<memgraph::utils::pmr::string, mgp_trans> transformations;
memgraph::utils::pmr::map<memgraph::utils::pmr::string, mgp_func> functions;
};
namespace memgraph::query::procedure {
@@ -811,6 +885,11 @@ namespace memgraph::query::procedure {
/// @throw anything std::ostream::operator<< may throw.
void PrintProcSignature(const mgp_proc &, std::ostream *);
/// @throw std::bad_alloc
/// @throw std::length_error
/// @throw anything std::ostream::operator<< may throw.
void PrintFuncSignature(const mgp_func &, std::ostream &);
bool IsValidIdentifierName(const char *name);
} // namespace memgraph::query::procedure
@@ -839,3 +918,5 @@ struct mgp_messages {
storage_type messages;
};
memgraph::query::TypedValue ToTypedValue(const mgp_value &val, memgraph::utils::MemoryResource *memory);

View File

@@ -52,6 +52,8 @@ class BuiltinModule final : public Module {
const std::map<std::string, mgp_trans, std::less<>> *Transformations() const override;
const std::map<std::string, mgp_func, std::less<>> *Functions() const override;
void AddProcedure(std::string_view name, mgp_proc proc);
void AddTransformation(std::string_view name, mgp_trans trans);
@@ -62,6 +64,7 @@ class BuiltinModule final : public Module {
/// Registered procedures
std::map<std::string, mgp_proc, std::less<>> procedures_;
std::map<std::string, mgp_trans, std::less<>> transformations_;
std::map<std::string, mgp_func, std::less<>> functions_;
};
BuiltinModule::BuiltinModule() {}
@@ -75,6 +78,7 @@ const std::map<std::string, mgp_proc, std::less<>> *BuiltinModule::Procedures()
const std::map<std::string, mgp_trans, std::less<>> *BuiltinModule::Transformations() const {
return &transformations_;
}
const std::map<std::string, mgp_func, std::less<>> *BuiltinModule::Functions() const { return &functions_; }
void BuiltinModule::AddProcedure(std::string_view name, mgp_proc proc) { procedures_.emplace(name, std::move(proc)); }
@@ -117,18 +121,18 @@ void RegisterMgLoad(ModuleRegistry *module_registry, utils::RWLock *lock, Builti
bool succ = false;
WithUpgradedLock(lock, [&]() {
const char *arg_as_string{nullptr};
if (const auto err = mgp_value_get_string(arg, &arg_as_string); err != MGP_ERROR_NO_ERROR) {
if (const auto err = mgp_value_get_string(arg, &arg_as_string); err != mgp_error::MGP_ERROR_NO_ERROR) {
succ = false;
} else {
succ = module_registry->LoadOrReloadModuleFromName(arg_as_string);
}
});
if (!succ) {
MG_ASSERT(mgp_result_set_error_msg(result, "Failed to (re)load the module.") == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_result_set_error_msg(result, "Failed to (re)load the module.") == mgp_error::MGP_ERROR_NO_ERROR);
}
};
mgp_proc load("load", load_cb, utils::NewDeleteResource());
MG_ASSERT(mgp_proc_add_arg(&load, "module_name", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&load, "module_name", Call<mgp_type *>(mgp_type_string)) == mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("load", std::move(load));
}
@@ -231,11 +235,16 @@ void RegisterMgProcedures(
}
};
mgp_proc procedures("procedures", procedures_cb, utils::NewDeleteResource());
MG_ASSERT(mgp_proc_add_result(&procedures, "name", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "signature", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "is_write", Call<mgp_type *>(mgp_type_bool)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "path", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "is_editable", Call<mgp_type *>(mgp_type_bool)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "name", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "signature", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "is_write", Call<mgp_type *>(mgp_type_bool)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "path", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "is_editable", Call<mgp_type *>(mgp_type_bool)) ==
mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("procedures", std::move(procedures));
}
@@ -294,15 +303,98 @@ void RegisterMgTransformations(const std::map<std::string, std::unique_ptr<Modul
}
};
mgp_proc procedures("transformations", transformations_cb, utils::NewDeleteResource());
MG_ASSERT(mgp_proc_add_result(&procedures, "name", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "path", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "is_editable", Call<mgp_type *>(mgp_type_bool)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "name", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "path", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&procedures, "is_editable", Call<mgp_type *>(mgp_type_bool)) ==
mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("transformations", std::move(procedures));
}
void RegisterMgFunctions(
// We expect modules to be sorted by name.
const std::map<std::string, std::unique_ptr<Module>, std::less<>> *all_modules, BuiltinModule *module) {
auto functions_cb = [all_modules](mgp_list * /*args*/, mgp_graph * /*graph*/, mgp_result *result,
mgp_memory *memory) {
// Iterating over all_modules assumes that the standard mechanism of magic
// functions invocations takes the ModuleRegistry::lock_ with READ access.
for (const auto &[module_name, module] : *all_modules) {
// Return the results in sorted order by module and by function_name.
static_assert(std::is_same_v<decltype(module->Functions()), const std::map<std::string, mgp_func, std::less<>> *>,
"Expected module magic functions to be sorted by name");
const auto path = module->Path();
const auto path_string = GetPathString(path);
const auto is_editable = IsFileEditable(path);
for (const auto &[func_name, func] : *module->Functions()) {
mgp_result_record *record{nullptr};
if (!TryOrSetError([&] { return mgp_result_new_record(result, &record); }, result)) {
return;
}
const auto path_value = GetStringValueOrSetError(path_string.c_str(), memory, result);
if (!path_value) {
return;
}
MgpUniquePtr<mgp_value> is_editable_value{nullptr, mgp_value_destroy};
if (!TryOrSetError([&] { return CreateMgpObject(is_editable_value, mgp_value_make_bool, is_editable, memory); },
result)) {
return;
}
utils::pmr::string full_name(module_name, memory->impl);
full_name.append(1, '.');
full_name.append(func_name);
const auto name_value = GetStringValueOrSetError(full_name.c_str(), memory, result);
if (!name_value) {
return;
}
std::stringstream ss;
ss << module_name << ".";
PrintFuncSignature(func, ss);
const auto signature = ss.str();
const auto signature_value = GetStringValueOrSetError(signature.c_str(), memory, result);
if (!signature_value) {
return;
}
if (!InsertResultOrSetError(result, record, "name", name_value.get())) {
return;
}
if (!InsertResultOrSetError(result, record, "signature", signature_value.get())) {
return;
}
if (!InsertResultOrSetError(result, record, "path", path_value.get())) {
return;
}
if (!InsertResultOrSetError(result, record, "is_editable", is_editable_value.get())) {
return;
}
}
}
};
mgp_proc functions("functions", functions_cb, utils::NewDeleteResource());
MG_ASSERT(mgp_proc_add_result(&functions, "name", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&functions, "signature", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&functions, "path", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&functions, "is_editable", Call<mgp_type *>(mgp_type_bool)) ==
mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("functions", std::move(functions));
}
namespace {
bool IsAllowedExtension(const auto &extension) {
constexpr std::array<std::string_view, 1> allowed_extensions{".py"};
static constexpr std::array<std::string_view, 1> allowed_extensions{".py"};
return std::any_of(allowed_extensions.begin(), allowed_extensions.end(),
[&](const auto allowed_extension) { return allowed_extension == extension; });
}
@@ -389,9 +481,10 @@ void RegisterMgGetModuleFiles(ModuleRegistry *module_registry, BuiltinModule *mo
mgp_proc get_module_files("get_module_files", get_module_files_cb, utils::NewDeleteResource(),
{.required_privilege = AuthQuery::Privilege::MODULE_READ});
MG_ASSERT(mgp_proc_add_result(&get_module_files, "path", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&get_module_files, "path", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&get_module_files, "is_editable", Call<mgp_type *>(mgp_type_bool)) ==
MGP_ERROR_NO_ERROR);
mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("get_module_files", std::move(get_module_files));
}
@@ -450,8 +543,10 @@ void RegisterMgGetModuleFile(ModuleRegistry *module_registry, BuiltinModule *mod
};
mgp_proc get_module_file("get_module_file", std::move(get_module_file_cb), utils::NewDeleteResource(),
{.required_privilege = AuthQuery::Privilege::MODULE_READ});
MG_ASSERT(mgp_proc_add_arg(&get_module_file, "path", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&get_module_file, "content", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&get_module_file, "path", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&get_module_file, "content", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("get_module_file", std::move(get_module_file));
}
@@ -529,9 +624,12 @@ void RegisterMgCreateModuleFile(ModuleRegistry *module_registry, utils::RWLock *
};
mgp_proc create_module_file("create_module_file", std::move(create_module_file_cb), utils::NewDeleteResource(),
{.required_privilege = AuthQuery::Privilege::MODULE_WRITE});
MG_ASSERT(mgp_proc_add_arg(&create_module_file, "filename", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&create_module_file, "content", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&create_module_file, "path", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&create_module_file, "filename", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&create_module_file, "content", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&create_module_file, "path", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("create_module_file", std::move(create_module_file));
}
@@ -584,8 +682,10 @@ void RegisterMgUpdateModuleFile(ModuleRegistry *module_registry, utils::RWLock *
};
mgp_proc update_module_file("update_module_file", std::move(update_module_file_cb), utils::NewDeleteResource(),
{.required_privilege = AuthQuery::Privilege::MODULE_WRITE});
MG_ASSERT(mgp_proc_add_arg(&update_module_file, "path", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&update_module_file, "content", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&update_module_file, "path", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&update_module_file, "content", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("update_module_file", std::move(update_module_file));
}
@@ -641,7 +741,8 @@ void RegisterMgDeleteModuleFile(ModuleRegistry *module_registry, utils::RWLock *
};
mgp_proc delete_module_file("delete_module_file", std::move(delete_module_file_cb), utils::NewDeleteResource(),
{.required_privilege = AuthQuery::Privilege::MODULE_WRITE});
MG_ASSERT(mgp_proc_add_arg(&delete_module_file, "path", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&delete_module_file, "path", Call<mgp_type *>(mgp_type_string)) ==
mgp_error::MGP_ERROR_NO_ERROR);
module->AddProcedure("delete_module_file", std::move(delete_module_file));
}
@@ -650,10 +751,10 @@ void RegisterMgDeleteModuleFile(ModuleRegistry *module_registry, utils::RWLock *
// `mgp_module::transformations into `proc_map`. The return value of WithModuleRegistration
// is the same as that of `fun`. Note, the return value need only be convertible to `bool`,
// it does not have to be `bool` itself.
template <class TProcMap, class TTransMap, class TFun>
auto WithModuleRegistration(TProcMap *proc_map, TTransMap *trans_map, const TFun &fun) {
template <class TProcMap, class TTransMap, class TFuncMap, class TFun>
auto WithModuleRegistration(TProcMap *proc_map, TTransMap *trans_map, TFuncMap *func_map, const TFun &fun) {
// We probably don't need more than 256KB for module initialization.
constexpr size_t stack_bytes = 256 * 1024;
static constexpr size_t stack_bytes = 256UL * 1024UL;
unsigned char stack_memory[stack_bytes];
utils::MonotonicBufferResource monotonic_memory(stack_memory, stack_bytes);
mgp_memory memory{&monotonic_memory};
@@ -664,6 +765,8 @@ auto WithModuleRegistration(TProcMap *proc_map, TTransMap *trans_map, const TFun
for (const auto &proc : module_def.procedures) proc_map->emplace(proc);
// Copy transformations into resulting trans_map.
for (const auto &trans : module_def.transformations) trans_map->emplace(trans);
// Copy functions into resulting func_map.
for (const auto &func : module_def.functions) func_map->emplace(func);
}
return res;
}
@@ -687,6 +790,8 @@ class SharedLibraryModule final : public Module {
const std::map<std::string, mgp_trans, std::less<>> *Transformations() const override;
const std::map<std::string, mgp_func, std::less<>> *Functions() const override;
std::optional<std::filesystem::path> Path() const override { return file_path_; }
private:
@@ -702,6 +807,8 @@ class SharedLibraryModule final : public Module {
std::map<std::string, mgp_proc, std::less<>> procedures_;
/// Registered transformations
std::map<std::string, mgp_trans, std::less<>> transformations_;
/// Registered functions
std::map<std::string, mgp_func, std::less<>> functions_;
};
SharedLibraryModule::SharedLibraryModule() : handle_(nullptr) {}
@@ -715,7 +822,8 @@ bool SharedLibraryModule::Load(const std::filesystem::path &file_path) {
spdlog::info("Loading module {}...", file_path);
file_path_ = file_path;
dlerror(); // Clear any existing error.
handle_ = dlopen(file_path.c_str(), RTLD_NOW | RTLD_LOCAL);
// NOLINTNEXTLINE(hicpp-signed-bitwise)
handle_ = dlopen(file_path.c_str(), RTLD_NOW | RTLD_LOCAL | RTLD_DEEPBIND);
if (!handle_) {
spdlog::error(
utils::MessageWithLink("Unable to load module {}; {}.", file_path, dlerror(), "https://memgr.ph/modules"));
@@ -746,8 +854,8 @@ bool SharedLibraryModule::Load(const std::filesystem::path &file_path) {
return with_error(error);
}
for (auto &trans : module_def->transformations) {
const bool was_result_added = MgpTransAddFixedResult(&trans.second);
if (!was_result_added) {
const bool success = mgp_error::MGP_ERROR_NO_ERROR == MgpTransAddFixedResult(&trans.second);
if (!success) {
const auto error =
fmt::format("Unable to add result to transformation in module {}; add result failed", file_path);
return with_error(error);
@@ -755,7 +863,7 @@ bool SharedLibraryModule::Load(const std::filesystem::path &file_path) {
}
return true;
};
if (!WithModuleRegistration(&procedures_, &transformations_, module_cb)) {
if (!WithModuleRegistration(&procedures_, &transformations_, &functions_, module_cb)) {
return false;
}
// Get optional mgp_shutdown_module
@@ -801,6 +909,13 @@ const std::map<std::string, mgp_trans, std::less<>> *SharedLibraryModule::Transf
return &transformations_;
}
const std::map<std::string, mgp_func, std::less<>> *SharedLibraryModule::Functions() const {
MG_ASSERT(handle_,
"Attempting to access functions of a module that has not "
"been loaded...");
return &functions_;
}
class PythonModule final : public Module {
public:
PythonModule();
@@ -816,6 +931,7 @@ class PythonModule final : public Module {
const std::map<std::string, mgp_proc, std::less<>> *Procedures() const override;
const std::map<std::string, mgp_trans, std::less<>> *Transformations() const override;
const std::map<std::string, mgp_func, std::less<>> *Functions() const override;
std::optional<std::filesystem::path> Path() const override { return file_path_; }
private:
@@ -823,6 +939,7 @@ class PythonModule final : public Module {
py::Object py_module_;
std::map<std::string, mgp_proc, std::less<>> procedures_;
std::map<std::string, mgp_trans, std::less<>> transformations_;
std::map<std::string, mgp_func, std::less<>> functions_;
};
PythonModule::PythonModule() {}
@@ -846,14 +963,14 @@ bool PythonModule::Load(const std::filesystem::path &file_path) {
auto module_cb = [&](auto *module_def, auto * /*memory*/) {
auto result = ImportPyModule(file_path.stem().c_str(), module_def);
for (auto &trans : module_def->transformations) {
succ = MgpTransAddFixedResult(&trans.second) == MGP_ERROR_NO_ERROR;
succ = MgpTransAddFixedResult(&trans.second) == mgp_error::MGP_ERROR_NO_ERROR;
if (!succ) {
return result;
}
};
return result;
};
py_module_ = WithModuleRegistration(&procedures_, &transformations_, module_cb);
py_module_ = WithModuleRegistration(&procedures_, &transformations_, &functions_, module_cb);
if (py_module_) {
spdlog::info("Loaded module {}", file_path);
@@ -877,6 +994,7 @@ bool PythonModule::Close() {
auto gil = py::EnsureGIL();
procedures_.clear();
transformations_.clear();
functions_.clear();
// Delete the module from the `sys.modules` directory so that the module will
// be properly imported if imported again.
py::Object sys(PyImport_ImportModule("sys"));
@@ -906,6 +1024,13 @@ const std::map<std::string, mgp_trans, std::less<>> *PythonModule::Transformatio
"not been loaded...");
return &transformations_;
}
const std::map<std::string, mgp_func, std::less<>> *PythonModule::Functions() const {
MG_ASSERT(py_module_,
"Attempting to access functions of a module that has "
"not been loaded...");
return &functions_;
}
namespace {
std::unique_ptr<Module> LoadModuleFromFile(const std::filesystem::path &path) {
@@ -954,6 +1079,7 @@ ModuleRegistry::ModuleRegistry() {
auto module = std::make_unique<BuiltinModule>();
RegisterMgProcedures(&modules_, module.get());
RegisterMgTransformations(&modules_, module.get());
RegisterMgFunctions(&modules_, module.get());
RegisterMgLoad(this, &lock_, module.get());
RegisterMgGetModuleFiles(this, module.get());
RegisterMgGetModuleFile(this, module.get());
@@ -1083,7 +1209,7 @@ std::optional<std::pair<std::string_view, std::string_view>> FindModuleNameAndPr
}
template <typename T>
concept ModuleProperties = utils::SameAsAnyOf<T, mgp_proc, mgp_trans>;
concept ModuleProperties = utils::SameAsAnyOf<T, mgp_proc, mgp_trans, mgp_func>;
template <ModuleProperties T>
std::optional<std::pair<ModulePtr, const T *>> MakePairIfPropFound(const ModuleRegistry &module_registry,
@@ -1092,8 +1218,10 @@ std::optional<std::pair<ModulePtr, const T *>> MakePairIfPropFound(const ModuleR
auto prop_fun = [](auto &module) {
if constexpr (std::is_same_v<T, mgp_proc>) {
return module->Procedures();
} else {
} else if constexpr (std::is_same_v<T, mgp_trans>) {
return module->Transformations();
} else if constexpr (std::is_same_v<T, mgp_func>) {
return module->Functions();
}
};
auto result = FindModuleNameAndProp(module_registry, fully_qualified_name, memory);
@@ -1121,4 +1249,10 @@ std::optional<std::pair<ModulePtr, const mgp_trans *>> FindTransformation(
return MakePairIfPropFound<mgp_trans>(module_registry, fully_qualified_transformation_name, memory);
}
std::optional<std::pair<ModulePtr, const mgp_func *>> FindFunction(const ModuleRegistry &module_registry,
std::string_view fully_qualified_function_name,
utils::MemoryResource *memory) {
return MakePairIfPropFound<mgp_func>(module_registry, fully_qualified_function_name, memory);
}
} // namespace memgraph::query::procedure

View File

@@ -13,6 +13,7 @@
/// API for loading and registering modules providing custom oC procedures
#pragma once
#include <dlfcn.h>
#include <filesystem>
#include <functional>
#include <optional>
@@ -21,6 +22,7 @@
#include <string_view>
#include <unordered_map>
#include "query/procedure/cypher_types.hpp"
#include "query/procedure/mg_procedure_impl.hpp"
#include "utils/memory.hpp"
#include "utils/rw_lock.hpp"
@@ -45,6 +47,8 @@ class Module {
virtual const std::map<std::string, mgp_proc, std::less<>> *Procedures() const = 0;
/// Returns registered transformations of this module
virtual const std::map<std::string, mgp_trans, std::less<>> *Transformations() const = 0;
// /// Returns registered functions of this module
virtual const std::map<std::string, mgp_func, std::less<>> *Functions() const = 0;
virtual std::optional<std::filesystem::path> Path() const = 0;
};
@@ -125,6 +129,40 @@ class ModuleRegistry final {
const std::filesystem::path &InternalModuleDir() const noexcept;
private:
class SharedLibraryHandle {
public:
SharedLibraryHandle(const std::string &shared_library, int mode) : handle_{dlopen(shared_library.c_str(), mode)} {}
SharedLibraryHandle(const SharedLibraryHandle &) = delete;
SharedLibraryHandle(SharedLibraryHandle &&) = delete;
SharedLibraryHandle operator=(const SharedLibraryHandle &) = delete;
SharedLibraryHandle operator=(SharedLibraryHandle &&) = delete;
~SharedLibraryHandle() {
if (handle_) {
dlclose(handle_);
}
}
private:
void *handle_;
};
#if __has_feature(address_sanitizer)
// This is why we need RTLD_NODELETE and we must not use RTLD_DEEPBIND with
// ASAN: https://github.com/google/sanitizers/issues/89
SharedLibraryHandle libstd_handle{"libstdc++.so.6", RTLD_NOW | RTLD_LOCAL | RTLD_NODELETE};
#else
// The reason behind opening share library during runtime is to avoid issues
// with loading symbols from stdlib. We have encounter issues with locale
// that cause std::cout not being printed and issues when python libraries
// would call stdlib (e.g. pytorch).
// The way that those issues were solved was
// by using RTLD_DEEPBIND. RTLD_DEEPBIND ensures that the lookup for the
// mentioned library will be first performed in the already existing binded
// libraries and then the global namespace.
// RTLD_DEEPBIND => https://linux.die.net/man/3/dlopen
SharedLibraryHandle libstd_handle{"libstdc++.so.6", RTLD_NOW | RTLD_LOCAL | RTLD_DEEPBIND};
#endif
std::vector<std::filesystem::path> modules_dirs_;
std::filesystem::path internal_module_dir_;
};
@@ -147,4 +185,62 @@ std::optional<std::pair<procedure::ModulePtr, const mgp_proc *>> FindProcedure(
std::optional<std::pair<procedure::ModulePtr, const mgp_trans *>> FindTransformation(
const ModuleRegistry &module_registry, const std::string_view fully_qualified_transformation_name,
utils::MemoryResource *memory);
/// Return the ModulePtr and `mgp_func *` of the found function after resolving
/// `fully_qualified_function_name` if found. If there is no such function
/// std::nullopt is returned. `memory` is used for temporary allocations
/// inside this function. ModulePtr must be kept alive to make sure it won't be unloaded.
std::optional<std::pair<procedure::ModulePtr, const mgp_func *>> FindFunction(
const ModuleRegistry &module_registry, const std::string_view fully_qualified_function_name,
utils::MemoryResource *memory);
template <typename T>
concept IsCallable = utils::SameAsAnyOf<T, mgp_proc, mgp_func>;
template <IsCallable TCall>
void ConstructArguments(const std::vector<TypedValue> &args, const TCall &callable,
const std::string_view fully_qualified_name, mgp_list &args_list, mgp_graph &graph) {
const auto n_args = args.size();
const auto c_args_sz = callable.args.size();
const auto c_opt_args_sz = callable.opt_args.size();
if (n_args < c_args_sz || (n_args - c_args_sz > c_opt_args_sz)) {
if (callable.args.empty() && callable.opt_args.empty()) {
throw QueryRuntimeException("'{}' requires no arguments.", fully_qualified_name);
}
if (callable.opt_args.empty()) {
throw QueryRuntimeException("'{}' requires exactly {} {}.", fully_qualified_name, c_args_sz,
c_args_sz == 1U ? "argument" : "arguments");
}
throw QueryRuntimeException("'{}' requires between {} and {} arguments.", fully_qualified_name, c_args_sz,
c_args_sz + c_opt_args_sz);
}
args_list.elems.reserve(n_args);
auto is_not_optional_arg = [c_args_sz](int i) { return c_args_sz > i; };
for (size_t i = 0; i < n_args; ++i) {
auto arg = args[i];
std::string_view name;
const query::procedure::CypherType *type;
if (is_not_optional_arg(i)) {
name = callable.args[i].first;
type = callable.args[i].second;
} else {
name = std::get<0>(callable.opt_args[i - c_args_sz]);
type = std::get<1>(callable.opt_args[i - c_args_sz]);
}
if (!type->SatisfiesType(arg)) {
throw QueryRuntimeException("'{}' argument named '{}' at position {} must be of type {}.", fully_qualified_name,
name, i, type->GetPresentableName());
}
args_list.elems.emplace_back(std::move(arg), &graph);
}
// Fill missing optional arguments with their default values.
const size_t passed_in_opt_args = n_args - c_args_sz;
for (size_t i = passed_in_opt_args; i < c_opt_args_sz; ++i) {
args_list.elems.emplace_back(std::get<2>(callable.opt_args[i]), &graph);
}
}
} // namespace memgraph::query::procedure

View File

@@ -55,49 +55,49 @@ PyObject *gMgpSerializationError{nullptr}; // NOLINT(cppcoreguidelines-avo
// Returns true if an exception is raised
bool RaiseExceptionFromErrorCode(const mgp_error error) {
switch (error) {
case MGP_ERROR_NO_ERROR:
case mgp_error::MGP_ERROR_NO_ERROR:
return false;
case MGP_ERROR_UNKNOWN_ERROR: {
case mgp_error::MGP_ERROR_UNKNOWN_ERROR: {
PyErr_SetString(gMgpUnknownError, "Unknown error happened.");
return true;
}
case MGP_ERROR_UNABLE_TO_ALLOCATE: {
case mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE: {
PyErr_SetString(gMgpUnableToAllocateError, "Unable to allocate memory.");
return true;
}
case MGP_ERROR_INSUFFICIENT_BUFFER: {
case mgp_error::MGP_ERROR_INSUFFICIENT_BUFFER: {
PyErr_SetString(gMgpInsufficientBufferError, "Insufficient buffer.");
return true;
}
case MGP_ERROR_OUT_OF_RANGE: {
case mgp_error::MGP_ERROR_OUT_OF_RANGE: {
PyErr_SetString(gMgpOutOfRangeError, "Out of range.");
return true;
}
case MGP_ERROR_LOGIC_ERROR: {
case mgp_error::MGP_ERROR_LOGIC_ERROR: {
PyErr_SetString(gMgpLogicErrorError, "Logic error.");
return true;
}
case MGP_ERROR_DELETED_OBJECT: {
case mgp_error::MGP_ERROR_DELETED_OBJECT: {
PyErr_SetString(gMgpDeletedObjectError, "Accessing deleted object.");
return true;
}
case MGP_ERROR_INVALID_ARGUMENT: {
case mgp_error::MGP_ERROR_INVALID_ARGUMENT: {
PyErr_SetString(gMgpInvalidArgumentError, "Invalid argument.");
return true;
}
case MGP_ERROR_KEY_ALREADY_EXISTS: {
case mgp_error::MGP_ERROR_KEY_ALREADY_EXISTS: {
PyErr_SetString(gMgpKeyAlreadyExistsError, "Key already exists.");
return true;
}
case MGP_ERROR_IMMUTABLE_OBJECT: {
case mgp_error::MGP_ERROR_IMMUTABLE_OBJECT: {
PyErr_SetString(gMgpImmutableObjectError, "Cannot modify immutable object.");
return true;
}
case MGP_ERROR_VALUE_CONVERSION: {
case mgp_error::MGP_ERROR_VALUE_CONVERSION: {
PyErr_SetString(gMgpValueConversionError, "Value conversion failed.");
return true;
}
case MGP_ERROR_SERIALIZATION_ERROR: {
case mgp_error::MGP_ERROR_SERIALIZATION_ERROR: {
PyErr_SetString(gMgpSerializationError, "Operation cannot be serialized.");
return true;
}
@@ -447,62 +447,94 @@ PyObject *MakePyCypherType(mgp_type *type) {
// clang-format off
struct PyQueryProc {
PyObject_HEAD
mgp_proc *proc;
mgp_proc *callable;
};
// clang-format on
PyObject *PyQueryProcAddArg(PyQueryProc *self, PyObject *args) {
MG_ASSERT(self->proc);
// clang-format off
struct PyMagicFunc{
PyObject_HEAD
mgp_func *callable;
};
// clang-format on
template <typename T>
concept IsCallable = utils::SameAsAnyOf<T, PyQueryProc, PyMagicFunc>;
template <IsCallable TCall>
PyObject *PyCallableAddArg(TCall *self, PyObject *args) {
MG_ASSERT(self->callable);
const char *name = nullptr;
PyCypherType *py_type = nullptr;
if (!PyArg_ParseTuple(args, "sO!", &name, &PyCypherTypeType, &py_type)) return nullptr;
auto *type = py_type->type;
if (RaiseExceptionFromErrorCode(mgp_proc_add_arg(self->proc, name, type))) {
return nullptr;
if constexpr (std::is_same_v<TCall, PyQueryProc>) {
if (RaiseExceptionFromErrorCode(mgp_proc_add_arg(self->callable, name, type))) {
return nullptr;
}
} else if constexpr (std::is_same_v<TCall, PyMagicFunc>) {
if (RaiseExceptionFromErrorCode(mgp_func_add_arg(self->callable, name, type))) {
return nullptr;
}
}
Py_RETURN_NONE;
}
PyObject *PyQueryProcAddOptArg(PyQueryProc *self, PyObject *args) {
MG_ASSERT(self->proc);
template <IsCallable TCall>
PyObject *PyCallableAddOptArg(TCall *self, PyObject *args) {
MG_ASSERT(self->callable);
const char *name = nullptr;
PyCypherType *py_type = nullptr;
PyObject *py_value = nullptr;
if (!PyArg_ParseTuple(args, "sO!O", &name, &PyCypherTypeType, &py_type, &py_value)) return nullptr;
auto *type = py_type->type;
mgp_memory memory{self->proc->opt_args.get_allocator().GetMemoryResource()};
mgp_memory memory{self->callable->opt_args.get_allocator().GetMemoryResource()};
mgp_value *value = PyObjectToMgpValueWithPythonExceptions(py_value, &memory);
if (value == nullptr) {
return nullptr;
}
if (RaiseExceptionFromErrorCode(mgp_proc_add_opt_arg(self->proc, name, type, value))) {
mgp_value_destroy(value);
return nullptr;
if constexpr (std::is_same_v<TCall, PyQueryProc>) {
if (RaiseExceptionFromErrorCode(mgp_proc_add_opt_arg(self->callable, name, type, value))) {
mgp_value_destroy(value);
return nullptr;
}
} else if constexpr (std::is_same_v<TCall, PyMagicFunc>) {
if (RaiseExceptionFromErrorCode(mgp_func_add_opt_arg(self->callable, name, type, value))) {
mgp_value_destroy(value);
return nullptr;
}
}
mgp_value_destroy(value);
Py_RETURN_NONE;
}
PyObject *PyQueryProcAddArg(PyQueryProc *self, PyObject *args) { return PyCallableAddArg(self, args); }
PyObject *PyQueryProcAddOptArg(PyQueryProc *self, PyObject *args) { return PyCallableAddOptArg(self, args); }
PyObject *PyQueryProcAddResult(PyQueryProc *self, PyObject *args) {
MG_ASSERT(self->proc);
MG_ASSERT(self->callable);
const char *name = nullptr;
PyCypherType *py_type = nullptr;
if (!PyArg_ParseTuple(args, "sO!", &name, &PyCypherTypeType, &py_type)) return nullptr;
auto *type = reinterpret_cast<PyCypherType *>(py_type)->type;
if (RaiseExceptionFromErrorCode(mgp_proc_add_result(self->proc, name, type))) {
if (RaiseExceptionFromErrorCode(mgp_proc_add_result(self->callable, name, type))) {
return nullptr;
}
Py_RETURN_NONE;
}
PyObject *PyQueryProcAddDeprecatedResult(PyQueryProc *self, PyObject *args) {
MG_ASSERT(self->proc);
MG_ASSERT(self->callable);
const char *name = nullptr;
PyCypherType *py_type = nullptr;
if (!PyArg_ParseTuple(args, "sO!", &name, &PyCypherTypeType, &py_type)) return nullptr;
auto *type = reinterpret_cast<PyCypherType *>(py_type)->type;
if (RaiseExceptionFromErrorCode(mgp_proc_add_deprecated_result(self->proc, name, type))) {
if (RaiseExceptionFromErrorCode(mgp_proc_add_deprecated_result(self->callable, name, type))) {
return nullptr;
}
Py_RETURN_NONE;
@@ -532,6 +564,33 @@ static PyTypeObject PyQueryProcType = {
};
// clang-format on
PyObject *PyMagicFuncAddArg(PyMagicFunc *self, PyObject *args) { return PyCallableAddArg(self, args); }
PyObject *PyMagicFuncAddOptArg(PyMagicFunc *self, PyObject *args) { return PyCallableAddOptArg(self, args); }
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
static PyMethodDef PyMagicFuncMethods[] = {
{"__reduce__", reinterpret_cast<PyCFunction>(DisallowPickleAndCopy), METH_NOARGS, "__reduce__ is not supported"},
{"add_arg", reinterpret_cast<PyCFunction>(PyMagicFuncAddArg), METH_VARARGS,
"Add a required argument to a function."},
{"add_opt_arg", reinterpret_cast<PyCFunction>(PyMagicFuncAddOptArg), METH_VARARGS,
"Add an optional argument with a default value to a function."},
{nullptr},
};
// clang-format off
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
static PyTypeObject PyMagicFuncType = {
PyVarObject_HEAD_INIT(nullptr, 0)
.tp_name = "_mgp.Func",
.tp_basicsize = sizeof(PyMagicFunc),
// NOLINTNEXTLINE(hicpp-signed-bitwise)
.tp_flags = Py_TPFLAGS_DEFAULT,
.tp_doc = "Wraps struct mgp_func.",
.tp_methods = PyMagicFuncMethods,
};
// clang-format on
// clang-format off
struct PyQueryModule {
PyObject_HEAD
@@ -796,7 +855,6 @@ py::Object MgpListToPyTuple(mgp_list *list, PyObject *py_graph) {
}
namespace {
std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Object py_record) {
py::Object py_mgp(PyImport_ImportModule("mgp"));
if (!py_mgp) return py::FetchError();
@@ -844,7 +902,7 @@ std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Obj
if (field_val == nullptr) {
return py::FetchError();
}
if (mgp_result_record_insert(record, field_name, field_val) != MGP_ERROR_NO_ERROR) {
if (mgp_result_record_insert(record, field_name, field_val) != mgp_error::MGP_ERROR_NO_ERROR) {
std::stringstream ss;
ss << "Unable to insert field '" << py::Object::FromBorrow(key) << "' with value: '"
<< py::Object::FromBorrow(val) << "'; did you set the correct field type?";
@@ -870,6 +928,33 @@ std::optional<py::ExceptionInfo> AddMultipleRecordsFromPython(mgp_result *result
return std::nullopt;
}
std::function<void()> PyObjectCleanup(py::Object &py_object) {
return [py_object]() {
// Run `gc.collect` (reference cycle-detection) explicitly, so that we are
// sure the procedure cleaned up everything it held references to. If the
// user stored a reference to one of our `_mgp` instances then the
// internally used `mgp_*` structs will stay unfreed and a memory leak
// will be reported at the end of the query execution.
py::Object gc(PyImport_ImportModule("gc"));
if (!gc) {
LOG_FATAL(py::FetchError().value());
}
if (!gc.CallMethod("collect")) {
LOG_FATAL(py::FetchError().value());
}
// After making sure all references from our side have been cleared,
// invalidate the `_mgp.Graph` object. If the user kept a reference to one
// of our `_mgp` instances then this will prevent them from using those
// objects (whose internal `mgp_*` pointers are now invalid and would cause
// a crash).
if (!py_object.CallMethod("invalidate")) {
LOG_FATAL(py::FetchError().value());
}
};
}
void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *graph, mgp_result *result,
mgp_memory *memory) {
auto gil = py::EnsureGIL();
@@ -895,31 +980,6 @@ void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *gra
}
};
auto cleanup = [](py::Object py_graph) {
// Run `gc.collect` (reference cycle-detection) explicitly, so that we are
// sure the procedure cleaned up everything it held references to. If the
// user stored a reference to one of our `_mgp` instances then the
// internally used `mgp_*` structs will stay unfreed and a memory leak
// will be reported at the end of the query execution.
py::Object gc(PyImport_ImportModule("gc"));
if (!gc) {
LOG_FATAL(py::FetchError().value());
}
if (!gc.CallMethod("collect")) {
LOG_FATAL(py::FetchError().value());
}
// After making sure all references from our side have been cleared,
// invalidate the `_mgp.Graph` object. If the user kept a reference to one
// of our `_mgp` instances then this will prevent them from using those
// objects (whose internal `mgp_*` pointers are now invalid and would cause
// a crash).
if (!py_graph.CallMethod("invalidate")) {
LOG_FATAL(py::FetchError().value());
}
};
// It is *VERY IMPORTANT* to note that this code takes great care not to keep
// any extra references to any `_mgp` instances (except for `_mgp.Graph`), so
// as not to introduce extra reference counts and prevent their deallocation.
@@ -932,14 +992,9 @@ void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *gra
std::optional<std::string> maybe_msg;
{
py::Object py_graph(MakePyGraph(graph, memory));
utils::OnScopeExit clean_up(PyObjectCleanup(py_graph));
if (py_graph) {
try {
maybe_msg = error_to_msg(call(py_graph));
cleanup(py_graph);
} catch (...) {
cleanup(py_graph);
throw;
}
maybe_msg = error_to_msg(call(py_graph));
} else {
maybe_msg = error_to_msg(py::FetchError());
}
@@ -972,32 +1027,58 @@ void CallPythonTransformation(const py::Object &py_cb, mgp_messages *msgs, mgp_g
return AddRecordFromPython(result, py_res);
};
auto cleanup = [](py::Object py_graph, py::Object py_messages) {
// Run `gc.collect` (reference cycle-detection) explicitly, so that we are
// sure the procedure cleaned up everything it held references to. If the
// user stored a reference to one of our `_mgp` instances then the
// internally used `mgp_*` structs will stay unfreed and a memory leak
// will be reported at the end of the query execution.
py::Object gc(PyImport_ImportModule("gc"));
if (!gc) {
LOG_FATAL(py::FetchError().value());
}
// It is *VERY IMPORTANT* to note that this code takes great care not to keep
// any extra references to any `_mgp` instances (except for `_mgp.Graph`), so
// as not to introduce extra reference counts and prevent their deallocation.
// In particular, the `ExceptionInfo` object has a `traceback` field that
// contains references to the Python frames and their arguments, and therefore
// our `_mgp` instances as well. Within this code we ensure not to keep the
// `ExceptionInfo` object alive so that no extra reference counts are
// introduced. We only fetch the error message and immediately destroy the
// object.
std::optional<std::string> maybe_msg;
{
py::Object py_graph(MakePyGraph(graph, memory));
py::Object py_messages(MakePyMessages(msgs, memory));
if (!gc.CallMethod("collect")) {
LOG_FATAL(py::FetchError().value());
}
utils::OnScopeExit clean_up_graph(PyObjectCleanup(py_graph));
utils::OnScopeExit clean_up_messages(PyObjectCleanup(py_messages));
// After making sure all references from our side have been cleared,
// invalidate the `_mgp.Graph` object. If the user kept a reference to one
// of our `_mgp` instances then this will prevent them from using those
// objects (whose internal `mgp_*` pointers are now invalid and would cause
// a crash).
if (!py_graph.CallMethod("invalidate")) {
LOG_FATAL(py::FetchError().value());
if (py_graph && py_messages) {
maybe_msg = error_to_msg(call(py_graph, py_messages));
} else {
maybe_msg = error_to_msg(py::FetchError());
}
if (!py_messages.CallMethod("invalidate")) {
LOG_FATAL(py::FetchError().value());
}
if (maybe_msg) {
static_cast<void>(mgp_result_set_error_msg(result, maybe_msg->c_str()));
}
}
void CallPythonFunction(const py::Object &py_cb, mgp_list *args, mgp_graph *graph, mgp_func_result *result,
mgp_memory *memory) {
auto gil = py::EnsureGIL();
auto error_to_msg = [](const std::optional<py::ExceptionInfo> &exc_info) -> std::optional<std::string> {
if (!exc_info) return std::nullopt;
// Here we tell the traceback formatter to skip the first line of the
// traceback because that line will always be our wrapper function in our
// internal `mgp.py` file. With that line skipped, the user will always
// get only the relevant traceback that happened in his Python code.
return py::FormatException(*exc_info, /* skip_first_line = */ true);
};
auto call = [&](py::Object py_graph) -> utils::BasicResult<std::optional<py::ExceptionInfo>, mgp_value *> {
py::Object py_args(MgpListToPyTuple(args, py_graph.Ptr()));
if (!py_args) return {py::FetchError()};
auto py_res = py_cb.Call(py_graph, py_args);
if (!py_res) return {py::FetchError()};
mgp_value *ret_val = PyObjectToMgpValueWithPythonExceptions(py_res.Ptr(), memory);
if (ret_val == nullptr) {
return {py::FetchError()};
}
return ret_val;
};
// It is *VERY IMPORTANT* to note that this code takes great care not to keep
@@ -1012,22 +1093,22 @@ void CallPythonTransformation(const py::Object &py_cb, mgp_messages *msgs, mgp_g
std::optional<std::string> maybe_msg;
{
py::Object py_graph(MakePyGraph(graph, memory));
py::Object py_messages(MakePyMessages(msgs, memory));
if (py_graph && py_messages) {
try {
maybe_msg = error_to_msg(call(py_graph, py_messages));
cleanup(py_graph, py_messages);
} catch (...) {
cleanup(py_graph, py_messages);
throw;
utils::OnScopeExit clean_up(PyObjectCleanup(py_graph));
if (py_graph) {
auto maybe_result = call(py_graph);
if (!maybe_result.HasError()) {
static_cast<void>(mgp_func_result_set_value(result, maybe_result.GetValue(), memory));
return;
}
maybe_msg = error_to_msg(maybe_result.GetError());
} else {
maybe_msg = error_to_msg(py::FetchError());
}
}
if (maybe_msg) {
static_cast<void>(mgp_result_set_error_msg(result, maybe_msg->c_str()));
static_cast<void>(
mgp_func_result_set_error_msg(result, maybe_msg->c_str(), memory)); // No error fetching if this fails
}
}
@@ -1056,9 +1137,9 @@ PyObject *PyQueryModuleAddProcedure(PyQueryModule *self, PyObject *cb, bool is_w
PyErr_SetString(PyExc_ValueError, "Already registered a procedure with the same name.");
return nullptr;
}
auto *py_proc = PyObject_New(PyQueryProc, &PyQueryProcType);
auto *py_proc = PyObject_New(PyQueryProc, &PyQueryProcType); // NOLINT(cppcoreguidelines-pro-type-cstyle-cast)
if (!py_proc) return nullptr;
py_proc->proc = &proc_it->second;
py_proc->callable = &proc_it->second;
return reinterpret_cast<PyObject *>(py_proc);
}
} // namespace
@@ -1100,6 +1181,39 @@ PyObject *PyQueryModuleAddTransformation(PyQueryModule *self, PyObject *cb) {
Py_RETURN_NONE;
}
PyObject *PyQueryModuleAddFunction(PyQueryModule *self, PyObject *cb) {
MG_ASSERT(self->module);
if (!PyCallable_Check(cb)) {
PyErr_SetString(PyExc_TypeError, "Expected a callable object.");
return nullptr;
}
auto py_cb = py::Object::FromBorrow(cb);
py::Object py_name(py_cb.GetAttr("__name__"));
const auto *name = PyUnicode_AsUTF8(py_name.Ptr());
if (!name) return nullptr;
if (!IsValidIdentifierName(name)) {
PyErr_SetString(PyExc_ValueError, "Function name is not a valid identifier");
return nullptr;
}
auto *memory = self->module->functions.get_allocator().GetMemoryResource();
mgp_func func(
name,
[py_cb](mgp_list *args, mgp_func_context *func_ctx, mgp_func_result *result, mgp_memory *memory) {
auto graph = mgp_graph::NonWritableGraph(*(func_ctx->impl), func_ctx->view);
return CallPythonFunction(py_cb, args, &graph, result, memory);
},
memory);
const auto [func_it, did_insert] = self->module->functions.emplace(name, std::move(func));
if (!did_insert) {
PyErr_SetString(PyExc_ValueError, "Already registered a function with the same name.");
return nullptr;
}
auto *py_func = PyObject_New(PyMagicFunc, &PyMagicFuncType); // NOLINT(cppcoreguidelines-pro-type-cstyle-cast)
if (!py_func) return nullptr;
py_func->callable = &func_it->second;
return reinterpret_cast<PyObject *>(py_func);
}
static PyMethodDef PyQueryModuleMethods[] = {
{"__reduce__", reinterpret_cast<PyCFunction>(DisallowPickleAndCopy), METH_NOARGS, "__reduce__ is not supported"},
{"add_read_procedure", reinterpret_cast<PyCFunction>(PyQueryModuleAddReadProcedure), METH_O,
@@ -1108,6 +1222,8 @@ static PyMethodDef PyQueryModuleMethods[] = {
"Register a writeable procedure with this module."},
{"add_transformation", reinterpret_cast<PyCFunction>(PyQueryModuleAddTransformation), METH_O,
"Register a transformation with this module."},
{"add_function", reinterpret_cast<PyCFunction>(PyQueryModuleAddFunction), METH_O,
"Register a function with this module."},
{nullptr},
};
@@ -1980,6 +2096,7 @@ PyObject *PyInitMgpModule() {
if (!register_type(&PyGraphType, "Graph")) return nullptr;
if (!register_type(&PyEdgeType, "Edge")) return nullptr;
if (!register_type(&PyQueryProcType, "Proc")) return nullptr;
if (!register_type(&PyMagicFuncType, "Func")) return nullptr;
if (!register_type(&PyQueryModuleType, "Module")) return nullptr;
if (!register_type(&PyVertexType, "Vertex")) return nullptr;
if (!register_type(&PyPathType, "Path")) return nullptr;
@@ -2164,9 +2281,10 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
auto py_seq_to_list = [memory](PyObject *seq, Py_ssize_t len, const auto &py_seq_get_item) {
static_assert(std::numeric_limits<Py_ssize_t>::max() <= std::numeric_limits<size_t>::max());
MgpUniquePtr<mgp_list> list{nullptr, &mgp_list_destroy};
if (const auto err = CreateMgpObject(list, mgp_list_make_empty, len, memory); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = CreateMgpObject(list, mgp_list_make_empty, len, memory);
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during making mgp_list"};
}
for (Py_ssize_t i = 0; i < len; ++i) {
@@ -2175,17 +2293,17 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
v = PyObjectToMgpValue(e, memory);
const auto err = mgp_list_append(list.get(), v);
mgp_value_destroy(v);
if (err != MGP_ERROR_NO_ERROR) {
if (err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (err != mgp_error::MGP_ERROR_NO_ERROR) {
if (err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
}
throw std::runtime_error{"Unexpected error during appending to mgp_list"};
}
}
mgp_value *v{nullptr};
if (const auto err = mgp_value_make_list(list.get(), &v); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_value_make_list(list.get(), &v); err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during making mgp_value"};
}
static_cast<void>(list.release());
@@ -2217,7 +2335,7 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
};
mgp_value *mgp_v{nullptr};
mgp_error last_error{MGP_ERROR_NO_ERROR};
mgp_error last_error{mgp_error::MGP_ERROR_NO_ERROR};
if (o == Py_None) {
last_error = mgp_value_make_null(memory, &mgp_v);
@@ -2243,10 +2361,10 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
MgpUniquePtr<mgp_map> map{nullptr, mgp_map_destroy};
const auto map_err = CreateMgpObject(map, mgp_map_make_empty, memory);
if (map_err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (map_err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
}
if (map_err != MGP_ERROR_NO_ERROR) {
if (map_err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during creating mgp_map"};
}
@@ -2267,16 +2385,16 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
MgpUniquePtr<mgp_value> v{PyObjectToMgpValue(value, memory), mgp_value_destroy};
if (const auto err = mgp_map_insert(map.get(), k, v.get()); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_map_insert(map.get(), k, v.get()); err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during inserting an item to mgp_map"};
}
}
if (const auto err = mgp_value_make_map(map.get(), &mgp_v); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_value_make_map(map.get(), &mgp_v); err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during creating mgp_value"};
}
static_cast<void>(map.release());
@@ -2285,14 +2403,14 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
// Copy the edge and pass the ownership to the created mgp_value.
if (const auto err = CreateMgpObject(e, mgp_edge_copy, reinterpret_cast<PyEdge *>(o)->edge, memory);
err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during copying mgp_edge"};
}
if (const auto err = mgp_value_make_edge(e.get(), &mgp_v); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_value_make_edge(e.get(), &mgp_v); err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during copying mgp_edge"};
}
static_cast<void>(e.release());
@@ -2301,14 +2419,14 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
// Copy the edge and pass the ownership to the created mgp_value.
if (const auto err = CreateMgpObject(p, mgp_path_copy, reinterpret_cast<PyPath *>(o)->path, memory);
err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during copying mgp_path"};
}
if (const auto err = mgp_value_make_path(p.get(), &mgp_v); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_value_make_path(p.get(), &mgp_v); err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during copying mgp_path"};
}
static_cast<void>(p.release());
@@ -2317,14 +2435,14 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
// Copy the edge and pass the ownership to the created mgp_value.
if (const auto err = CreateMgpObject(v, mgp_vertex_copy, reinterpret_cast<PyVertex *>(o)->vertex, memory);
err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during copying mgp_vertex"};
}
if (const auto err = mgp_value_make_vertex(v.get(), &mgp_v); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_value_make_vertex(v.get(), &mgp_v); err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error during copying mgp_vertex"};
}
static_cast<void>(v.release());
@@ -2357,14 +2475,14 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
MgpUniquePtr<mgp_date> date{nullptr, mgp_date_destroy};
if (const auto err = CreateMgpObject(date, mgp_date_from_parameters, &parameters, memory);
err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_date"};
}
if (const auto err = mgp_value_make_date(date.get(), &mgp_v); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_value_make_date(date.get(), &mgp_v); err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_value"};
}
static_cast<void>(date.release());
@@ -2382,14 +2500,15 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
MgpUniquePtr<mgp_local_time> local_time{nullptr, mgp_local_time_destroy};
if (const auto err = CreateMgpObject(local_time, mgp_local_time_from_parameters, &parameters, memory);
err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_local_time"};
}
if (const auto err = mgp_value_make_local_time(local_time.get(), &mgp_v); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_value_make_local_time(local_time.get(), &mgp_v);
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_value"};
}
static_cast<void>(local_time.release());
@@ -2414,20 +2533,21 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
MgpUniquePtr<mgp_local_date_time> local_date_time{nullptr, mgp_local_date_time_destroy};
if (const auto err = CreateMgpObject(local_date_time, mgp_local_date_time_from_parameters, &parameters, memory);
err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_local_date_time"};
}
if (const auto err = mgp_value_make_local_date_time(local_date_time.get(), &mgp_v);
err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_value"};
}
static_cast<void>(local_date_time.release());
} else if (PyDelta_CheckExact(o)) {
constexpr int64_t microseconds_in_days = static_cast<std::chrono::microseconds>(std::chrono::days{1}).count();
static constexpr int64_t microseconds_in_days =
static_cast<std::chrono::microseconds>(std::chrono::days{1}).count();
const auto days =
PyDateTime_DELTA_GET_DAYS(o); // NOLINT(cppcoreguidelines-pro-type-cstyle-cast,hicpp-signed-bitwise)
auto microseconds =
@@ -2440,14 +2560,15 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
MgpUniquePtr<mgp_duration> duration{nullptr, mgp_duration_destroy};
if (const auto err = CreateMgpObject(duration, mgp_duration_from_microseconds, microseconds, memory);
err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_duration"};
}
if (const auto err = mgp_value_make_duration(duration.get(), &mgp_v); err == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (const auto err = mgp_value_make_duration(duration.get(), &mgp_v);
err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
} else if (err != MGP_ERROR_NO_ERROR) {
} else if (err != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_value"};
}
static_cast<void>(duration.release());
@@ -2455,10 +2576,10 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
throw std::invalid_argument("Unsupported PyObject conversion");
}
if (last_error == MGP_ERROR_UNABLE_TO_ALLOCATE) {
if (last_error == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
throw std::bad_alloc{};
}
if (last_error != MGP_ERROR_NO_ERROR) {
if (last_error != mgp_error::MGP_ERROR_NO_ERROR) {
throw std::runtime_error{"Unexpected error while creating mgp_value"};
}

View File

@@ -23,8 +23,8 @@
namespace memgraph::query::stream {
constexpr std::chrono::milliseconds kDefaultBatchInterval{100};
constexpr int64_t kDefaultBatchSize{1000};
inline constexpr std::chrono::milliseconds kDefaultBatchInterval{100};
inline constexpr int64_t kDefaultBatchSize{1000};
template <typename TMessage>
using ConsumerFunction = std::function<void(const std::vector<TMessage> &)>;

View File

@@ -42,7 +42,7 @@ extern const Event MessagesConsumed;
namespace memgraph::query::stream {
namespace {
constexpr auto kExpectedTransformationResultSize = 2;
inline constexpr auto kExpectedTransformationResultSize = 2;
const utils::pmr::string query_param_name{"query", utils::NewDeleteResource()};
const utils::pmr::string params_param_name{"parameters", utils::NewDeleteResource()};
@@ -172,50 +172,50 @@ void Streams::RegisterProcedures() {
void Streams::RegisterKafkaProcedures() {
{
constexpr std::string_view proc_name = "kafka_set_stream_offset";
auto set_stream_offset = [this, proc_name](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result,
mgp_memory * /*memory*/) {
static constexpr std::string_view proc_name = "kafka_set_stream_offset";
auto set_stream_offset = [this](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result,
mgp_memory * /*memory*/) {
auto *arg_stream_name = procedure::Call<mgp_value *>(mgp_list_at, args, 0);
const auto *stream_name = procedure::Call<const char *>(mgp_value_get_string, arg_stream_name);
auto *arg_offset = procedure::Call<mgp_value *>(mgp_list_at, args, 1);
const auto offset = procedure::Call<int64_t>(mgp_value_get_int, arg_offset);
auto lock_ptr = streams_.Lock();
auto it = GetStream(*lock_ptr, std::string(stream_name));
std::visit(utils::Overloaded{
[&](StreamData<KafkaStream> &kafka_stream) {
auto stream_source_ptr = kafka_stream.stream_source->Lock();
const auto error = stream_source_ptr->SetStreamOffset(offset);
if (error.HasError()) {
MG_ASSERT(mgp_result_set_error_msg(result, error.GetError().c_str()) == MGP_ERROR_NO_ERROR,
"Unable to set procedure error message of procedure: {}", proc_name);
}
},
[proc_name](auto && /*other*/) {
throw QueryRuntimeException("'{}' can be only used for Kafka stream sources", proc_name);
}},
std::visit(utils::Overloaded{[&](StreamData<KafkaStream> &kafka_stream) {
auto stream_source_ptr = kafka_stream.stream_source->Lock();
const auto error = stream_source_ptr->SetStreamOffset(offset);
if (error.HasError()) {
MG_ASSERT(mgp_result_set_error_msg(result, error.GetError().c_str()) ==
mgp_error::MGP_ERROR_NO_ERROR,
"Unable to set procedure error message of procedure: {}", proc_name);
}
},
[](auto && /*other*/) {
throw QueryRuntimeException("'{}' can be only used for Kafka stream sources",
proc_name);
}},
it->second);
};
mgp_proc proc(proc_name, set_stream_offset, utils::NewDeleteResource());
MG_ASSERT(mgp_proc_add_arg(&proc, "stream_name", procedure::Call<mgp_type *>(mgp_type_string)) ==
MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&proc, "offset", procedure::Call<mgp_type *>(mgp_type_int)) == MGP_ERROR_NO_ERROR);
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_arg(&proc, "offset", procedure::Call<mgp_type *>(mgp_type_int)) ==
mgp_error::MGP_ERROR_NO_ERROR);
procedure::gModuleRegistry.RegisterMgProcedure(proc_name, std::move(proc));
}
{
constexpr std::string_view proc_name = "kafka_stream_info";
static constexpr std::string_view proc_name = "kafka_stream_info";
constexpr std::string_view consumer_group_result_name = "consumer_group";
constexpr std::string_view topics_result_name = "topics";
constexpr std::string_view bootstrap_servers_result_name = "bootstrap_servers";
constexpr std::string_view configs_result_name = "configs";
constexpr std::string_view credentials_result_name = "credentials";
static constexpr std::string_view consumer_group_result_name = "consumer_group";
static constexpr std::string_view topics_result_name = "topics";
static constexpr std::string_view bootstrap_servers_result_name = "bootstrap_servers";
static constexpr std::string_view configs_result_name = "configs";
static constexpr std::string_view credentials_result_name = "credentials";
auto get_stream_info = [this, proc_name, consumer_group_result_name, topics_result_name,
bootstrap_servers_result_name, configs_result_name, credentials_result_name](
mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
auto get_stream_info = [this](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
auto *arg_stream_name = procedure::Call<mgp_value *>(mgp_list_at, args, 0);
const auto *stream_name = procedure::Call<const char *>(mgp_value_get_string, arg_stream_name);
auto lock_ptr = streams_.Lock();
@@ -339,7 +339,7 @@ void Streams::RegisterKafkaProcedures() {
return;
}
},
[proc_name](auto && /*other*/) {
[](auto && /*other*/) {
throw QueryRuntimeException("'{}' can be only used for Kafka stream sources", proc_name);
}},
it->second);
@@ -347,19 +347,19 @@ void Streams::RegisterKafkaProcedures() {
mgp_proc proc(proc_name, get_stream_info, utils::NewDeleteResource());
MG_ASSERT(mgp_proc_add_arg(&proc, "stream_name", procedure::Call<mgp_type *>(mgp_type_string)) ==
MGP_ERROR_NO_ERROR);
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&proc, consumer_group_result_name.data(),
procedure::Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
procedure::Call<mgp_type *>(mgp_type_string)) == mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(
mgp_proc_add_result(&proc, topics_result_name.data(),
procedure::Call<mgp_type *>(mgp_type_list, procedure::Call<mgp_type *>(mgp_type_string))) ==
MGP_ERROR_NO_ERROR);
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&proc, bootstrap_servers_result_name.data(),
procedure::Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
procedure::Call<mgp_type *>(mgp_type_string)) == mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&proc, configs_result_name.data(), procedure::Call<mgp_type *>(mgp_type_map)) ==
MGP_ERROR_NO_ERROR);
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&proc, credentials_result_name.data(), procedure::Call<mgp_type *>(mgp_type_map)) ==
MGP_ERROR_NO_ERROR);
mgp_error::MGP_ERROR_NO_ERROR);
procedure::gModuleRegistry.RegisterMgProcedure(proc_name, std::move(proc));
}
@@ -367,11 +367,10 @@ void Streams::RegisterKafkaProcedures() {
void Streams::RegisterPulsarProcedures() {
{
constexpr std::string_view proc_name = "pulsar_stream_info";
constexpr std::string_view service_url_result_name = "service_url";
constexpr std::string_view topics_result_name = "topics";
auto get_stream_info = [this, proc_name, service_url_result_name, topics_result_name](
mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
static constexpr std::string_view proc_name = "pulsar_stream_info";
static constexpr std::string_view service_url_result_name = "service_url";
static constexpr std::string_view topics_result_name = "topics";
auto get_stream_info = [this](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
auto *arg_stream_name = procedure::Call<mgp_value *>(mgp_list_at, args, 0);
const auto *stream_name = procedure::Call<const char *>(mgp_value_get_string, arg_stream_name);
auto lock_ptr = streams_.Lock();
@@ -427,7 +426,7 @@ void Streams::RegisterPulsarProcedures() {
return;
}
},
[proc_name](auto && /*other*/) {
[](auto && /*other*/) {
throw QueryRuntimeException("'{}' can be only used for Pulsar stream sources", proc_name);
}},
it->second);
@@ -435,14 +434,14 @@ void Streams::RegisterPulsarProcedures() {
mgp_proc proc(proc_name, get_stream_info, utils::NewDeleteResource());
MG_ASSERT(mgp_proc_add_arg(&proc, "stream_name", procedure::Call<mgp_type *>(mgp_type_string)) ==
MGP_ERROR_NO_ERROR);
mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&proc, service_url_result_name.data(),
procedure::Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
procedure::Call<mgp_type *>(mgp_type_string)) == mgp_error::MGP_ERROR_NO_ERROR);
MG_ASSERT(
mgp_proc_add_result(&proc, topics_result_name.data(),
procedure::Call<mgp_type *>(mgp_type_list, procedure::Call<mgp_type *>(mgp_type_string))) ==
MGP_ERROR_NO_ERROR);
mgp_error::MGP_ERROR_NO_ERROR);
procedure::gModuleRegistry.RegisterMgProcedure(proc_name, std::move(proc));
}

View File

@@ -219,7 +219,7 @@ void Trigger::Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution
// Set up temporary memory for a single Pull. Initial memory comes from the
// stack. 256 KiB should fit on the stack and should be more than enough for a
// single `Pull`.
constexpr size_t stack_size = 256 * 1024;
static constexpr size_t stack_size = 256UL * 1024UL;
char stack_data[stack_size];
// We can throw on every query because a simple queries for deleting will use only
@@ -251,7 +251,7 @@ void Trigger::Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution
namespace {
// When the format of the persisted trigger is changed, increase this version
constexpr uint64_t kVersion{2};
inline constexpr uint64_t kVersion{2};
} // namespace
TriggerStore::TriggerStore(std::filesystem::path directory) : storage_{std::move(directory)} {}

View File

@@ -660,8 +660,8 @@ double ToDouble(const TypedValue &value) {
namespace {
bool IsTemporalType(const TypedValue::Type type) {
constexpr std::array temporal_types{TypedValue::Type::Date, TypedValue::Type::LocalTime,
TypedValue::Type::LocalDateTime, TypedValue::Type::Duration};
static constexpr std::array temporal_types{TypedValue::Type::Date, TypedValue::Type::LocalTime,
TypedValue::Type::LocalDateTime, TypedValue::Type::Duration};
return std::any_of(temporal_types.begin(), temporal_types.end(),
[type](const auto temporal_type) { return temporal_type == type; });
};

View File

@@ -1,5 +1,6 @@
set(storage_v2_src_files
commit_log.cpp
commit_error.cpp
constraints.cpp
temporal.cpp
durability/durability.cpp

View File

@@ -0,0 +1,20 @@
// Copyright 2022 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/v2/commit_error.hpp"
namespace memgraph::storage {
bool operator==(const CommitError &lhs, const CommitError &rhs) {
return lhs.type == rhs.type && lhs.maybe_constraint_violation == rhs.maybe_constraint_violation;
}
} // namespace memgraph::storage

View File

@@ -0,0 +1,32 @@
// Copyright 2022 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 <optional>
#include "storage/v2/constraints.hpp"
namespace memgraph::storage {
struct CommitError {
enum class Type {
CONSTRAINT_VIOLATION,
UNABLE_TO_SYNC_REPLICATE,
};
Type type;
std::optional<ConstraintViolation> maybe_constraint_violation;
};
bool operator==(const CommitError &lhs, const CommitError &rhs);
} // namespace memgraph::storage

View File

@@ -16,6 +16,10 @@
namespace memgraph::storage::replication {
struct ReplicationClientConfig {
std::optional<double> timeout;
// The default delay between main checking/pinging replicas is 1s because
// that seems like a reasonable timeframe in which main should notice a
// replica is down.
std::chrono::seconds replica_check_frequency{1};
struct SSL {
std::string key_file = "";

View File

@@ -41,12 +41,49 @@ Storage::ReplicationClient::ReplicationClient(std::string name, Storage *storage
}
rpc_client_.emplace(endpoint, &*rpc_context_);
TryInitializeClient();
TryInitializeClientSync();
if (config.timeout && replica_state_ != replication::ReplicaState::INVALID) {
timeout_.emplace(*config.timeout);
timeout_dispatcher_.emplace();
}
// Help the user to get the most accurate replica state possible.
if (config.replica_check_frequency > std::chrono::seconds(0)) {
replica_checker_.Run("Replica Checker", config.replica_check_frequency, [&] { FrequentCheck(); });
}
}
void Storage::ReplicationClient::TryInitializeClientAsync() {
thread_pool_.AddTask([this] {
rpc_client_->Abort();
this->TryInitializeClientSync();
});
}
void Storage::ReplicationClient::FrequentCheck() {
const auto is_success = std::invoke([this]() {
try {
auto stream{rpc_client_->Stream<replication::FrequentHeartbeatRpc>()};
const auto response = stream.AwaitResponse();
return response.success;
} catch (const rpc::RpcFailedException &) {
return false;
}
});
// States: READY, REPLICATING, RECOVERY, INVALID
// If success && ready, replicating, recovery -> stay the same because something good is going on.
// If success && INVALID -> [it's possible that replica came back to life] -> TryInitializeClient.
// If fail -> [replica is not reachable at all] -> INVALID state.
// NOTE: TryInitializeClient might return nothing if there is a branching point.
// NOTE: The early return pattern simplified the code, but the behavior should be as explained.
if (!is_success) {
replica_state_.store(replication::ReplicaState::INVALID);
return;
}
if (replica_state_.load() == replication::ReplicaState::INVALID) {
TryInitializeClientAsync();
}
}
/// @throws rpc::RpcFailedException
@@ -100,7 +137,7 @@ void Storage::ReplicationClient::InitializeClient() {
}
}
void Storage::ReplicationClient::TryInitializeClient() {
void Storage::ReplicationClient::TryInitializeClientSync() {
try {
InitializeClient();
} catch (const rpc::RpcFailedException &) {
@@ -113,10 +150,7 @@ void Storage::ReplicationClient::TryInitializeClient() {
void Storage::ReplicationClient::HandleRpcFailure() {
spdlog::error(utils::MessageWithLink("Couldn't replicate data to {}.", name_, "https://memgr.ph/replication"));
thread_pool_.AddTask([this] {
rpc_client_->Abort();
this->TryInitializeClient();
});
TryInitializeClientAsync();
}
replication::SnapshotRes Storage::ReplicationClient::TransferSnapshot(const std::filesystem::path &path) {
@@ -193,17 +227,18 @@ void Storage::ReplicationClient::IfStreamingTransaction(const std::function<void
}
}
void Storage::ReplicationClient::FinalizeTransactionReplication() {
std::optional<bool> Storage::ReplicationClient::FinalizeTransactionReplication() {
// We can only check the state because it guarantees to be only
// valid during a single transaction replication (if the assumption
// that this and other transaction replication functions can only be
// called from a one thread stands)
if (replica_state_ != replication::ReplicaState::REPLICATING) {
return;
return std::nullopt;
}
if (mode_ == replication::ReplicationMode::ASYNC) {
thread_pool_.AddTask([this] { this->FinalizeTransactionReplicationInternal(); });
thread_pool_.AddTask([this] { [[maybe_unused]] auto finalized = this->FinalizeTransactionReplicationInternal(); });
return true;
} else if (timeout_) {
MG_ASSERT(mode_ == replication::ReplicationMode::SYNC, "Only SYNC replica can have a timeout.");
MG_ASSERT(timeout_dispatcher_, "Timeout thread is missing");
@@ -211,7 +246,7 @@ void Storage::ReplicationClient::FinalizeTransactionReplication() {
timeout_dispatcher_->active = true;
thread_pool_.AddTask([&, this] {
this->FinalizeTransactionReplicationInternal();
[[maybe_unused]] auto finalized = this->FinalizeTransactionReplicationInternal();
std::unique_lock main_guard(timeout_dispatcher_->main_lock);
// TimerThread can finish waiting for timeout
timeout_dispatcher_->active = false;
@@ -239,12 +274,13 @@ void Storage::ReplicationClient::FinalizeTransactionReplication() {
// and acces the `active` variable`
thread_pool_.AddTask([this] { timeout_dispatcher_.reset(); });
}
return true;
} else {
FinalizeTransactionReplicationInternal();
return FinalizeTransactionReplicationInternal();
}
}
void Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
bool Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
MG_ASSERT(replica_stream_, "Missing stream for transaction deltas");
try {
auto response = replica_stream_->Finalize();
@@ -255,6 +291,7 @@ void Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
thread_pool_.AddTask([&, this] { this->RecoverReplica(response.current_commit_timestamp); });
} else {
replica_state_.store(replication::ReplicaState::READY);
return true;
}
} catch (const rpc::RpcFailedException &) {
replica_stream_.reset();
@@ -264,6 +301,7 @@ void Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
}
HandleRpcFailure();
}
return false;
}
void Storage::ReplicationClient::RecoverReplica(uint64_t replica_commit) {

View File

@@ -103,7 +103,10 @@ class Storage::ReplicationClient {
// StartTransactionReplication, stream is created.
void IfStreamingTransaction(const std::function<void(ReplicaStream &handler)> &callback);
void FinalizeTransactionReplication();
// Return none -> OK
// Return true -> OK
// Return false -> FAIL
std::optional<bool> FinalizeTransactionReplication();
// Transfer the snapshot file.
// @param path Path of the snapshot file.
@@ -125,7 +128,7 @@ class Storage::ReplicationClient {
const auto &Endpoint() const { return rpc_client_->Endpoint(); }
private:
void FinalizeTransactionReplicationInternal();
[[nodiscard]] bool FinalizeTransactionReplicationInternal();
void RecoverReplica(uint64_t replica_commit);
@@ -142,16 +145,14 @@ class Storage::ReplicationClient {
std::vector<RecoveryStep> GetRecoverySteps(uint64_t replica_commit, utils::FileRetainer::FileLocker *file_locker);
void FrequentCheck();
void InitializeClient();
void TryInitializeClient();
void TryInitializeClientSync();
void TryInitializeClientAsync();
void HandleRpcFailure();
std::string name_;
Storage *storage_;
std::optional<communication::ClientContext> rpc_context_;
std::optional<rpc::Client> rpc_client_;
@@ -198,6 +199,8 @@ class Storage::ReplicationClient {
// to ignore concurrency problems inside the client.
utils::ThreadPool thread_pool_{1};
std::atomic<replication::ReplicaState> replica_state_{replication::ReplicaState::INVALID};
utils::Scheduler replica_checker_;
};
} // namespace memgraph::storage

View File

@@ -60,6 +60,10 @@ Storage::ReplicationServer::ReplicationServer(Storage *storage, io::network::End
spdlog::debug("Received HeartbeatRpc");
this->HeartbeatHandler(req_reader, res_builder);
});
rpc_server_->Register<replication::FrequentHeartbeatRpc>([](auto *req_reader, auto *res_builder) {
spdlog::debug("Received FrequentHeartbeatRpc");
FrequentHeartbeatHandler(req_reader, res_builder);
});
rpc_server_->Register<replication::AppendDeltasRpc>([this](auto *req_reader, auto *res_builder) {
spdlog::debug("Received AppendDeltasRpc");
this->AppendDeltasHandler(req_reader, res_builder);
@@ -86,6 +90,13 @@ void Storage::ReplicationServer::HeartbeatHandler(slk::Reader *req_reader, slk::
slk::Save(res, res_builder);
}
void Storage::ReplicationServer::FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
replication::FrequentHeartbeatReq req;
slk::Load(&req, req_reader);
replication::FrequentHeartbeatRes res{true};
slk::Save(res, res_builder);
}
void Storage::ReplicationServer::AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
replication::AppendDeltasReq req;
slk::Load(&req, req_reader);

View File

@@ -29,6 +29,7 @@ class Storage::ReplicationServer {
private:
// RPC handlers
void HeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder);
static void FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder);
void AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder);
void SnapshotHandler(slk::Reader *req_reader, slk::Builder *res_builder);
void WalFilesHandler(slk::Reader *req_reader, slk::Builder *res_builder);

View File

@@ -43,6 +43,12 @@ cpp<#
(current-commit-timestamp :uint64_t)
(epoch-id "std::string"))))
;; FrequentHearthbeat is required because calling Heartbeat takes the storage lock.
;; Configured by `replication_replica_check_delay`.
(lcp:define-rpc frequent-heartbeat
(:request ())
(:response ((success :bool))))
(lcp:define-rpc snapshot
(:request ())
(:response

View File

@@ -40,6 +40,7 @@
#include "utils/uuid.hpp"
/// REPLICATION ///
#include "storage/v2/commit_error.hpp"
#include "storage/v2/replication/replication_client.hpp"
#include "storage/v2/replication/replication_server.hpp"
#include "storage/v2/replication/rpc.hpp"
@@ -49,7 +50,7 @@ namespace memgraph::storage {
using OOMExceptionEnabler = utils::MemoryTracker::OutOfMemoryExceptionEnabler;
namespace {
[[maybe_unused]] constexpr uint16_t kEpochHistoryRetention = 1000;
inline constexpr uint16_t kEpochHistoryRetention = 1000;
} // namespace
auto AdvanceToVisibleVertex(utils::SkipList<Vertex>::Iterator it, utils::SkipList<Vertex>::Iterator end,
@@ -813,7 +814,7 @@ EdgeTypeId Storage::Accessor::NameToEdgeType(const std::string_view &name) { ret
void Storage::Accessor::AdvanceCommand() { ++transaction_.command_id; }
utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
utils::BasicResult<CommitError, void> Storage::Accessor::Commit(
const std::optional<uint64_t> desired_commit_timestamp) {
MG_ASSERT(is_transaction_active_, "The transaction is already terminated!");
MG_ASSERT(!transaction_.must_abort, "The transaction can't be committed!");
@@ -836,7 +837,8 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
auto validation_result = ValidateExistenceConstraints(*prev.vertex, storage_->constraints_);
if (validation_result) {
Abort();
return *validation_result;
return CommitError{.type = CommitError::Type::CONSTRAINT_VIOLATION,
.maybe_constraint_violation = *validation_result};
}
}
@@ -848,6 +850,29 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
// Save these so we can mark them used in the commit log.
uint64_t start_timestamp = transaction_.start_timestamp;
// Aboring here and after the locked block obviously has an issue if
// replica goes down and back up during the execution of the locked block of
// code.
// Not enough, before the actual commit check all SYNC replicas for availability.
bool unable_to_sync_replicate = false;
const auto check_replicas = [&]() {
storage_->replication_clients_.WithLock([&](auto &clients) {
for (auto &client : clients) {
// Exclusively SYNC replicas have to be available to commit the transaction.
if (client->Mode() == replication::ReplicationMode::SYNC && !client->Timeout().has_value() &&
client->State() == replication::ReplicaState::INVALID) {
unable_to_sync_replicate = true;
}
}
});
};
check_replicas();
if (unable_to_sync_replicate) {
Abort();
return CommitError{.type = CommitError::Type::UNABLE_TO_SYNC_REPLICATE};
}
{
std::unique_lock<utils::SpinLock> engine_guard(storage_->engine_lock_);
commit_timestamp_.emplace(storage_->CommitTimestamp(desired_commit_timestamp));
@@ -893,6 +918,8 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
// Replica can log only the write transaction received from Main
// so the Wal files are consistent
if (storage_->replication_role_ == ReplicationRole::MAIN || desired_commit_timestamp.has_value()) {
// TODO(gitbuda): Possible to abort data operation because in this context there is an abort operation.
// TODO(gitbuda): If AppendToWal returns false, we can exit this block and Abort.
storage_->AppendToWal(transaction_, *commit_timestamp_);
}
@@ -915,13 +942,27 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
engine_guard.unlock();
});
// NOTE: This will finish/commit the transaction.
storage_->commit_log_->MarkFinished(start_timestamp);
// TODO(gitbuda): Maybe the solution here is to 1. mark 2. check all
// the replicas 3. unmark if required... it's not possible to unmark
// easily because mark contains mark + update_latest_active which is
// not reversable operation
}
}
// TODO(gitbuda): This doesn't have any effect because (it will only if the
// constraints are also violated).
check_replicas();
if (unable_to_sync_replicate) {
Abort();
return CommitError{.type = CommitError::Type::UNABLE_TO_SYNC_REPLICATE};
}
if (unique_constraint_violation) {
Abort();
return *unique_constraint_violation;
return CommitError{.type = CommitError::Type::CONSTRAINT_VIOLATION,
.maybe_constraint_violation = *unique_constraint_violation};
}
}
is_transaction_active_ = false;
@@ -1124,6 +1165,18 @@ EdgeTypeId Storage::NameToEdgeType(const std::string_view &name) {
return EdgeTypeId::FromUint(name_id_mapper_.NameToId(name));
}
// TODO(gitbuda): Hard to abort global operations in SYNC replication mode
// because there is no an abort op for that yet, one idea is to just apply
// reverse operation, e.g., CreateIndex <-> DropIndex.
//
// Another idea is to double check that all replicas have relevant data prior
// to calling MarkFinished. That approach would work in both data replication
// and global operation cases.
//
// EDGE CASE 1: What if the first SYNC replica is alive, receives the delta
// object, while the second SYNC replica is dead? (replication clients are
// stored in a vector and accessed one by one)
bool Storage::CreateIndex(LabelId label, const std::optional<uint64_t> desired_commit_timestamp) {
std::unique_lock<utils::RWLock> storage_guard(main_lock_);
if (!indices_.label_index.CreateIndex(label, vertices_.access())) return false;
@@ -1572,6 +1625,9 @@ void Storage::FinalizeWalFile() {
}
}
// TODO(gitbuda): Hard to abort data operation in SYNC replication mode because:
// * Just calling Abort inside AppendToWal for some reason causes infinite loop.
void Storage::AppendToWal(const Transaction &transaction, uint64_t final_commit_timestamp) {
if (!InitializeWalFile()) return;
// Traverse deltas and append them to the WAL file.
@@ -1743,10 +1799,15 @@ void Storage::AppendToWal(const Transaction &transaction, uint64_t final_commit_
FinalizeWalFile();
replication_clients_.WithLock([&](auto &clients) {
bool all_sync_replicas_ok = true;
for (auto &client : clients) {
// TODO(gitbuda): SEMI-SYNC should be exculded from here.
if (client->Mode() == replication::ReplicationMode::SYNC)
client->IfStreamingTransaction([&](auto &stream) { stream.AppendTransactionEnd(final_commit_timestamp); });
// TODO(gitbuda): FinalizeTransactionReplication should also indicate that eveything went well for SYNC replicas.
client->FinalizeTransactionReplication();
}
return all_sync_replicas_ok;
});
}

View File

@@ -43,6 +43,7 @@
/// REPLICATION ///
#include "rpc/server.hpp"
#include "storage/v2/commit_error.hpp"
#include "storage/v2/replication/config.hpp"
#include "storage/v2/replication/enums.hpp"
#include "storage/v2/replication/rpc.hpp"
@@ -308,11 +309,12 @@ class Storage final {
void AdvanceCommand();
/// Commit returns `ConstraintViolation` if the changes made by this
/// transaction violate an existence or unique constraint. In that case the
/// transaction is automatically aborted. Otherwise, void is returned.
/// Commit returns `CommitError` if the changes made by this transaction
/// violate an existence, unique constraint or data could NOT be replicated
/// to SYNC replica. In that case the transaction is automatically aborted.
/// Otherwise, void is returned.
/// @throw std::bad_alloc
utils::BasicResult<ConstraintViolation, void> Commit(std::optional<uint64_t> desired_commit_timestamp = {});
utils::BasicResult<CommitError, void> Commit(std::optional<uint64_t> desired_commit_timestamp = {});
/// @throw std::bad_alloc
void Abort();

View File

@@ -25,7 +25,7 @@
namespace {
constexpr uint64_t kInvalidFlagId = 0U;
inline constexpr uint64_t kInvalidFlagId = 0U;
// std::numeric_limits<time_t>::max() cannot be represented precisely as a double, so the next smallest value is the
// maximum number of seconds the timer can be used with
const double max_seconds_as_double = std::nexttoward(std::numeric_limits<time_t>::max(), 0.0);
@@ -143,7 +143,7 @@ AsyncTimer::AsyncTimer(double seconds)
MG_ASSERT(timer_create(CLOCK_MONOTONIC, &notification_settings, &timer_id_) == 0, "Couldn't create timer: ({}) {}",
errno, strerror(errno));
constexpr auto kSecondsToNanos = 1000 * 1000 * 1000;
static constexpr auto kSecondsToNanos = 1000 * 1000 * 1000;
// Casting will truncate down, but that's exactly what we want.
const auto second_as_time_t = static_cast<time_t>(seconds);
const auto remaining_nano_seconds = static_cast<time_t>((seconds - second_as_time_t) * kSecondsToNanos);

View File

@@ -45,7 +45,7 @@ namespace {
// two sets of base64 characters needs to be chosen.
// They differ in their last two characters.
//
constexpr std::array base64_chars = {
inline constexpr std::array base64_chars = {
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz"
"0123456789"

View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 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
@@ -49,6 +49,7 @@
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(FailedQuery, "Number of times executing a query failed.") \
M(LabelIndexCreated, "Number of times a label index was created.") \
@@ -65,7 +66,7 @@ namespace EventCounter {
APPLY_FOR_EVENTS(M)
#undef M
constexpr Event END = __COUNTER__;
inline constexpr Event END = __COUNTER__;
// Initialize array for the global counter with all values set to 0
Counter global_counters_array[END]{};

View File

@@ -68,7 +68,7 @@ bool RenamePath(const std::filesystem::path &src, const std::filesystem::path &d
/// `write` for each of our (very small) logical reads/writes. Because of that,
/// `read` or `write` is only called when the buffer is full and/or needs
/// emptying.
constexpr size_t kFileBufferSize = 262144;
inline constexpr size_t kFileBufferSize = 262144;
/// This class implements a file handler that is used to read binary files. It
/// was developed because the C++ standard library has an awful API and makes

View File

@@ -67,8 +67,8 @@ struct FnvCollection {
template <typename TA, typename TB, typename TAHash = std::hash<TA>, typename TBHash = std::hash<TB>>
struct HashCombine {
size_t operator()(const TA &a, const TB &b) const {
constexpr size_t fnv_prime = 1099511628211UL;
constexpr size_t fnv_offset = 14695981039346656037UL;
static constexpr size_t fnv_prime = 1099511628211UL;
static constexpr size_t fnv_offset = 14695981039346656037UL;
size_t ret = fnv_offset;
ret ^= TAHash()(a);
ret *= fnv_prime;

View File

@@ -30,7 +30,7 @@
namespace memgraph::utils::license {
namespace {
constexpr std::string_view license_key_prefix = "mglk-";
inline constexpr std::string_view license_key_prefix = "mglk-";
std::optional<License> GetLicense(const std::string &license_key) {
if (license_key.empty()) {

View File

@@ -28,8 +28,8 @@ struct License {
bool operator==(const License &) const = default;
};
constexpr std::string_view kEnterpriseLicenseSettingKey = "enterprise.license";
constexpr std::string_view kOrganizationNameSettingKey = "organization.name";
inline constexpr std::string_view kEnterpriseLicenseSettingKey = "enterprise.license";
inline constexpr std::string_view kOrganizationNameSettingKey = "organization.name";
enum class LicenseCheckError : uint8_t { INVALID_LICENSE_KEY_STRING, INVALID_ORGANIZATION_NAME, EXPIRED_LICENSE };

View File

@@ -24,7 +24,7 @@ static_assert(std::is_same_v<uint64_t, unsigned long>,
/// This function computes the log2 function on integer types. It is faster than
/// the cmath `log2` function because it doesn't use floating point values for
/// calculation.
constexpr inline uint64_t Log2(uint64_t val) {
constexpr uint64_t Log2(uint64_t val) {
// The `clz` function is undefined when the passed value is 0 and the value of
// `log` is `-inf` so we special case it here.
if (val == 0) return 0;
@@ -35,12 +35,12 @@ constexpr inline uint64_t Log2(uint64_t val) {
}
/// Return `true` if `val` is a power of 2.
constexpr inline bool IsPow2(uint64_t val) noexcept { return val != 0ULL && (val & (val - 1ULL)) == 0ULL; }
constexpr bool IsPow2(uint64_t val) noexcept { return val != 0ULL && (val & (val - 1ULL)) == 0ULL; }
/// Return `val` if it is power of 2, otherwise get the next power of 2 value.
/// If `val` is sufficiently large, the next power of 2 value may not fit into
/// the result type and you will get a wrapped value to 1ULL.
constexpr inline uint64_t Ceil2(uint64_t val) noexcept {
constexpr uint64_t Ceil2(uint64_t val) noexcept {
if (val == 0ULL || val == 1ULL) return 1ULL;
return 1ULL << (Log2(val - 1ULL) + 1ULL);
}
@@ -53,7 +53,7 @@ constexpr inline uint64_t Ceil2(uint64_t val) noexcept {
/// RoundUint64ToMultiple(5, 8) == 8
/// RoundUint64ToMultiple(8, 8) == 8
/// RoundUint64ToMultiple(9, 8) == 16
constexpr inline std::optional<uint64_t> RoundUint64ToMultiple(uint64_t val, uint64_t multiple) noexcept {
constexpr std::optional<uint64_t> RoundUint64ToMultiple(uint64_t val, uint64_t multiple) noexcept {
if (multiple == 0) return std::nullopt;
uint64_t numerator = val + multiple - 1;
// Check for overflow.

View File

@@ -19,8 +19,8 @@ namespace memgraph::utils {
std::string GetReadableSize(double size) {
// TODO (antonio2368): Add support for base 1000 (KB, GB, TB...)
constexpr std::array units = {"B", "KiB", "MiB", "GiB", "TiB"};
constexpr double delimiter = 1024;
static constexpr std::array units = {"B", "KiB", "MiB", "GiB", "TiB"};
static constexpr double delimiter = 1024;
size_t i = 0;
for (; i + 1 < units.size() && size >= delimiter; ++i) {

View File

@@ -325,7 +325,7 @@ class SmallVectorTemplateBase<T, true> : public SmallVectorTemplateCommon<T> {
};
template <typename T>
constexpr bool is_pod = std::is_standard_layout_v<T> &&std::is_trivial_v<T>;
inline constexpr bool is_pod = std::is_standard_layout_v<T> &&std::is_trivial_v<T>;
/// This class consists of common code factored out of the SmallVector class to
/// reduce code duplication based on the SmallVector 'n' template parameter.

View File

@@ -90,7 +90,7 @@ LocalDateTime CurrentLocalDateTime() {
}
namespace {
constexpr auto *kSupportedDateFormatsHelpMessage = R"help(
inline constexpr auto *kSupportedDateFormatsHelpMessage = R"help(
String representing the date should be in one of the following formats:
- YYYY-MM-DD
@@ -112,7 +112,7 @@ std::pair<DateParameters, bool> ParseDateParameters(std::string_view date_string
// https://en.wikipedia.org/wiki/ISO_8601#Dates
// Date string with the '-' as separator are in the EXTENDED format,
// otherwise they are in a BASIC format
constexpr std::array valid_sizes{
static constexpr std::array valid_sizes{
10, // YYYY-MM-DD
8, // YYYYMMDD
7 // YYYY-MM
@@ -184,7 +184,7 @@ size_t DateHash::operator()(const Date &date) const {
}
namespace {
constexpr auto *kSupportedTimeFormatsHelpMessage = R"help(
inline constexpr auto *kSupportedTimeFormatsHelpMessage = R"help(
String representing the time should be in one of the following formats:
- [T]hh:mm:ss
@@ -388,7 +388,7 @@ size_t LocalTimeHash::operator()(const LocalTime &local_time) const {
}
namespace {
constexpr auto *kSupportedLocalDateTimeFormatsHelpMessage = R"help(
inline constexpr auto *kSupportedLocalDateTimeFormatsHelpMessage = R"help(
String representing the LocalDateTime should be in one of the following formats:
- YYYY-MM-DDThh:mm:ss
@@ -458,7 +458,7 @@ std::pair<DateParameters, LocalTimeParameters> ParseLocalDateTimeParameters(std:
LocalDateTime::LocalDateTime(const int64_t microseconds) {
auto chrono_microseconds = std::chrono::microseconds(microseconds);
constexpr int64_t one_day_in_microseconds = std::chrono::microseconds{std::chrono::days{1}}.count();
static constexpr int64_t one_day_in_microseconds = std::chrono::microseconds{std::chrono::days{1}}.count();
if (microseconds < 0 && (microseconds % one_day_in_microseconds != 0)) {
date = Date(microseconds - one_day_in_microseconds);
} else {

View File

@@ -235,7 +235,8 @@ struct LocalTime {
auto abs = [](auto value) { return (value >= 0) ? value : -value; };
const auto lhs = local_time.MicrosecondsSinceEpoch();
if (rhs < 0 && lhs < abs(rhs)) {
constexpr int64_t one_day_in_microseconds = chrono::duration_cast<chrono::microseconds>(chrono::days(1)).count();
static constexpr int64_t one_day_in_microseconds =
chrono::duration_cast<chrono::microseconds>(chrono::days(1)).count();
rhs = one_day_in_microseconds + rhs;
}
auto result = chrono::microseconds(lhs + rhs);

View File

@@ -18,7 +18,7 @@
namespace memgraph::utils {
void ThreadSetName(const std::string &name) {
constexpr auto max_name_length = GetMaxThreadNameSize();
static constexpr auto max_name_length = GetMaxThreadNameSize();
MG_ASSERT(name.size() <= max_name_length, "Thread name '{}' is too long", max_name_length);
if (prctl(PR_SET_NAME, name.c_str()) != 0) {

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