Compare commits

...

14 Commits

Author SHA1 Message Date
Deda
f3d3275aaa Override version to 2.15.0-rc3 2024-02-22 11:58:11 +01:00
Marko Barišić
312cdaf685 Push successful RC builds to S3 (#1741)
* Add new workflow which calls release build workflows

* Make the workflow build packages only on RC tags

* Change artifact names to include OS name
2024-02-21 17:13:36 +01:00
Marko Budiselić
7f8a4f2a8b Add toolchain-v5 compatibility Revert to C++20 (#587)
* Upgrade cppitertools, spdlog, fmt, rapidcheck
* Make compilation work on both v4 and v5 toolchains
2024-02-21 17:13:36 +01:00
Andi
381feb7b35 Add --experimental-enabled=high-availability (#1720) 2024-02-21 17:13:36 +01:00
Marko Budiselić
710dba6a00 Patch NuRaft for clang-17 compilation (#1733) 2024-02-21 17:13:36 +01:00
Josipmrden
db79d7f55e Add function for property sizes (#1557)
Add function for property sizes
2024-02-21 17:11:57 +01:00
Deda
d1cdf9a0ba Override memgraph version to 2.15.0 2024-02-16 17:33:09 +01:00
Gareth Andrew Lloyd
be8b755673 Fixup memory e2e tests (#1715)
- Remove the e2e that did concurrent mgp_* calls on the same transaction
  (ATM this is unsupported)
- Fix up the concurrent mgp_global_alloc test to be testing it more precisely
- Reduce the memory limit on detach delete test due to recent memory
  optimizations around deltas.
- No longer throw from hook, through jemalloc C, to our C++ on other
  side. This cause mutex unlocks to not happen.
- No longer allocate error messages while inside the hook. This caused
  recursive entry back inside jamalloc which would try to relock a
  non-recursive mutex.
2024-02-16 17:31:59 +01:00
Andi
48db39a16a Forbid having multiple mains in the cluster (#1727) 2024-02-16 17:31:22 +01:00
Antonio Filipovic
0f7d774fc5 HA: Polish flow for replicas from coordinator (#1711) 2024-02-16 17:31:09 +01:00
Andi
dfb3a62cc4 Forbid writing to cluster-managed main on restart (#1717) 2024-02-16 17:30:53 +01:00
Marko Barišić
307c2e0881 Turn e2e tests back on for release build workflows (#1725) 2024-02-15 16:22:36 +01:00
Deda
b9b221cfe2 Update BSL license date 2024-02-15 15:50:41 +01:00
Marko Barišić
6698d48a82 Add rules for rc workflows (#1722) 2024-02-15 15:48:50 +01:00
146 changed files with 2964 additions and 1112 deletions

View File

@@ -336,53 +336,6 @@ jobs:
# multiple paths could be defined
build/logs
experimental_build_ha:
name: "High availability build"
runs-on: [self-hosted, Linux, X64, Diff]
env:
THREADS: 24
MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
steps:
- name: Set up repository
uses: actions/checkout@v4
with:
# Number of commits to fetch. `0` indicates all history for all
# branches and tags. (default: 1)
fetch-depth: 0
- name: Build release binaries
run: |
source /opt/toolchain-v4/activate
./init
cd build
cmake -DCMAKE_BUILD_TYPE=Release -DMG_EXPERIMENTAL_HIGH_AVAILABILITY=ON ..
make -j$THREADS
- name: Run unit tests
run: |
source /opt/toolchain-v4/activate
cd build
ctest -R memgraph__unit --output-on-failure -j$THREADS
- name: Run e2e tests
if: false
run: |
cd tests
./setup.sh /opt/toolchain-v4/activate
source ve3/bin/activate_e2e
cd e2e
./run.sh "Coordinator"
./run.sh "Client initiated failover"
./run.sh "Uninitialized cluster"
- name: Save test data
uses: actions/upload-artifact@v4
if: always()
with:
name: "Test data(High availability build)"
path: |
# multiple paths could be defined
build/logs
release_jepsen_test:
name: "Release Jepsen Test"
runs-on: [self-hosted, Linux, X64, Debian10, JepsenControl]

View File

@@ -0,0 +1,162 @@
name: Release build test
concurrency:
group: ${{ github.workflow }}-${{ github.ref_name }}
cancel-in-progress: true
on:
workflow_dispatch:
inputs:
build_type:
type: choice
description: "Memgraph Build type. Default value is Release."
default: 'Release'
options:
- Release
- RelWithDebInfo
push:
branches:
- "release/**"
tags:
- "v*.*.*-rc*"
- "v*.*-rc*"
schedule:
# UTC
- cron: "0 22 * * *"
env:
THREADS: 24
MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
BUILD_TYPE: ${{ github.event.inputs.build_type || 'Release' }}
jobs:
Debian10:
uses: ./.github/workflows/release_debian10.yaml
with:
build_type: ${{ github.event.inputs.build_type || 'Release' }}
secrets: inherit
Ubuntu20_04:
uses: ./.github/workflows/release_ubuntu2004.yaml
with:
build_type: ${{ github.event.inputs.build_type || 'Release' }}
secrets: inherit
PackageDebian10:
if: github.ref_type == 'tag'
needs: [Debian10]
runs-on: [self-hosted, DockerMgBuild, X64]
timeout-minutes: 60
steps:
- name: "Set up repository"
uses: actions/checkout@v4
with:
fetch-depth: 0 # Required because of release/get_version.py
- name: "Build package"
run: |
./release/package/run.sh package debian-10 $BUILD_TYPE
- name: "Upload package"
uses: actions/upload-artifact@v4
with:
name: debian-10
path: build/output/debian-10/memgraph*.deb
PackageUbuntu20_04:
if: github.ref_type == 'tag'
needs: [Ubuntu20_04]
runs-on: [self-hosted, DockerMgBuild, X64]
timeout-minutes: 60
steps:
- name: "Set up repository"
uses: actions/checkout@v4
with:
fetch-depth: 0 # Required because of release/get_version.py
- name: "Build package"
run: |
./release/package/run.sh package ubuntu-22.04 $BUILD_TYPE
- name: "Upload package"
uses: actions/upload-artifact@v4
with:
name: ubuntu-22.04
path: build/output/ubuntu-22.04/memgraph*.deb
PackageUbuntu20_04_ARM:
if: github.ref_type == 'tag'
needs: [Ubuntu20_04]
runs-on: [self-hosted, DockerMgBuild, ARM64]
# M1 Mac mini is sometimes slower
timeout-minutes: 90
steps:
- name: "Set up repository"
uses: actions/checkout@v4
with:
fetch-depth: 0 # Required because of release/get_version.py
- name: "Build package"
run: |
./release/package/run.sh package ubuntu-22.04-arm $BUILD_TYPE
- name: "Upload package"
uses: actions/upload-artifact@v4
with:
name: ubuntu-22.04-aarch64
path: build/output/ubuntu-22.04-arm/memgraph*.deb
PackageDebian11:
if: github.ref_type == 'tag'
needs: [Debian10, Ubuntu20_04]
runs-on: [self-hosted, DockerMgBuild, X64]
timeout-minutes: 60
steps:
- name: "Set up repository"
uses: actions/checkout@v4
with:
fetch-depth: 0 # Required because of release/get_version.py
- name: "Build package"
run: |
./release/package/run.sh package debian-11 $BUILD_TYPE
- name: "Upload package"
uses: actions/upload-artifact@v4
with:
name: debian-11
path: build/output/debian-11/memgraph*.deb
PackageDebian11_ARM:
if: github.ref_type == 'tag'
needs: [Debian10, Ubuntu20_04]
runs-on: [self-hosted, DockerMgBuild, ARM64]
# M1 Mac mini is sometimes slower
timeout-minutes: 90
steps:
- name: "Set up repository"
uses: actions/checkout@v4
with:
fetch-depth: 0 # Required because of release/get_version.py
- name: "Build package"
run: |
./release/package/run.sh package debian-11-arm $BUILD_TYPE
- name: "Upload package"
uses: actions/upload-artifact@v4
with:
name: debian-11-aarch64
path: build/output/debian-11-arm/memgraph*.deb
PushToS3:
if: github.ref_type == 'tag'
needs: [PackageDebian10, PackageDebian11, PackageDebian11_ARM, PackageUbuntu20_04, PackageUbuntu20_04_ARM]
runs-on: ubuntu-latest
steps:
- name: Download artifacts
uses: actions/download-artifact@v4
with:
# name: # if name input parameter is not provided, all artifacts are downloaded
# and put in directories named after each one.
path: build/output/release
- name: Upload to S3
uses: jakejarvis/s3-sync-action@v0.5.1
env:
AWS_S3_BUCKET: "deps.memgraph.io"
AWS_ACCESS_KEY_ID: ${{ secrets.S3_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.S3_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: "eu-west-1"
SOURCE_DIR: "build/output/release"
DEST_DIR: "memgraph-unofficial/${{ github.ref_name }}/"

View File

@@ -1,6 +1,12 @@
name: Release Debian 10
on:
workflow_call:
inputs:
build_type:
type: string
description: "Memgraph Build type. Default value is Release."
default: 'Release'
workflow_dispatch:
inputs:
build_type:
@@ -11,10 +17,8 @@ on:
- Release
- RelWithDebInfo
schedule:
- cron: "0 22 * * *"
env:
OS: "Debian10"
THREADS: 24
MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
@@ -111,7 +115,7 @@ jobs:
- name: Save code coverage
uses: actions/upload-artifact@v4
with:
name: "Code coverage(Coverage build)"
name: "Code coverage(Coverage build)-${{ env.OS }}"
path: tools/github/generated/code_coverage.tar.gz
debug_build:
@@ -165,7 +169,7 @@ jobs:
- name: Save cppcheck and clang-format errors
uses: actions/upload-artifact@v4
with:
name: "Code coverage(Debug build)"
name: "Code coverage(Debug build)-${{ env.OS }}"
path: tools/github/cppcheck_and_clang_format.txt
debug_integration_test:
@@ -242,7 +246,7 @@ jobs:
- name: Save enterprise DEB package
uses: actions/upload-artifact@v4
with:
name: "Enterprise DEB package"
name: "Enterprise DEB package-${{ env.OS}}"
path: build/output/memgraph*.deb
- name: Run GQL Behave tests
@@ -255,7 +259,7 @@ jobs:
- name: Save quality assurance status
uses: actions/upload-artifact@v4
with:
name: "GQL Behave Status"
name: "GQL Behave Status-${{ env.OS }}"
path: |
tests/gql_behave/gql_behave_status.csv
tests/gql_behave/gql_behave_status.html
@@ -321,7 +325,6 @@ jobs:
--no-strict
release_e2e_test:
if: false
name: "Release End-to-end Test"
runs-on: [self-hosted, Linux, X64, Debian10]
timeout-minutes: 60
@@ -456,5 +459,5 @@ jobs:
uses: actions/upload-artifact@v4
if: ${{ always() }}
with:
name: "Jepsen Report"
name: "Jepsen Report-${{ env.OS }}"
path: tests/jepsen/Jepsen.tar.gz

View File

@@ -1,6 +1,12 @@
name: Release Ubuntu 20.04
on:
workflow_call:
inputs:
build_type:
type: string
description: "Memgraph Build type. Default value is Release."
default: 'Release'
workflow_dispatch:
inputs:
build_type:
@@ -11,10 +17,8 @@ on:
- Release
- RelWithDebInfo
schedule:
- cron: "0 22 * * *"
env:
OS: "Ubuntu 20.04"
THREADS: 24
MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
@@ -107,7 +111,7 @@ jobs:
- name: Save code coverage
uses: actions/upload-artifact@v4
with:
name: "Code coverage(Coverage build)"
name: "Code coverage(Coverage build)-${{ env.OS }}"
path: tools/github/generated/code_coverage.tar.gz
debug_build:
@@ -161,7 +165,7 @@ jobs:
- name: Save cppcheck and clang-format errors
uses: actions/upload-artifact@v4
with:
name: "Code coverage(Debug build)"
name: "Code coverage(Debug build)-${{ env.OS }}"
path: tools/github/cppcheck_and_clang_format.txt
debug_integration_test:
@@ -238,7 +242,7 @@ jobs:
- name: Save enterprise DEB package
uses: actions/upload-artifact@v4
with:
name: "Enterprise DEB package"
name: "Enterprise DEB package-${{ env.OS }}"
path: build/output/memgraph*.deb
- name: Run GQL Behave tests
@@ -251,7 +255,7 @@ jobs:
- name: Save quality assurance status
uses: actions/upload-artifact@v4
with:
name: "GQL Behave Status"
name: "GQL Behave Status-${{ env.OS }}"
path: |
tests/gql_behave/gql_behave_status.csv
tests/gql_behave/gql_behave_status.html
@@ -317,7 +321,6 @@ jobs:
--no-strict
release_e2e_test:
if: false
name: "Release End-to-end Test"
runs-on: [self-hosted, Linux, X64, Ubuntu20.04]
timeout-minutes: 60

View File

@@ -1,4 +1,7 @@
name: Stress test large
concurrency:
group: ${{ github.workflow }}-${{ github.ref_name }}
cancel-in-progress: true
on:
workflow_dispatch:
@@ -10,7 +13,10 @@ on:
options:
- Release
- RelWithDebInfo
push:
tags:
- "v*.*.*-rc*"
- "v*.*-rc*"
schedule:
- cron: "0 22 * * *"

View File

@@ -64,11 +64,11 @@ option(MG_ENTERPRISE "Build Memgraph Enterprise Edition" ON)
# Set the current version here to override the automatic version detection. The
# version must be specified as `X.Y.Z`. Primarily used when building new patch
# versions.
set(MEMGRAPH_OVERRIDE_VERSION "")
set(MEMGRAPH_OVERRIDE_VERSION "2.15.0")
# Custom suffix that this version should have. The suffix can be any arbitrary
# string. Primarily used when building a version for a specific customer.
set(MEMGRAPH_OVERRIDE_VERSION_SUFFIX "")
set(MEMGRAPH_OVERRIDE_VERSION_SUFFIX "rc3")
# Variables used to generate the versions.
if (MG_ENTERPRISE)
@@ -211,8 +211,13 @@ set(CMAKE_CXX_FLAGS_RELWITHDEBINFO
# ** Static linking is allowed only for executables! **
set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -static-libgcc -static-libstdc++")
# Use lld linker to speedup build
add_link_options(-fuse-ld=lld) # TODO: use mold linker
# Use lld linker to speedup build and use less memory.
add_link_options(-fuse-ld=lld)
# NOTE: Moving to latest Clang (probably starting from 15), lld stopped to work
# without explicit link_directories call.
string(REPLACE ":" " " LD_LIBS $ENV{LD_LIBRARY_PATH})
separate_arguments(LD_LIBS)
link_directories(${LD_LIBS})
# release flags
set(CMAKE_CXX_FLAGS_RELEASE "-O2 -DNDEBUG")
@@ -271,18 +276,6 @@ endif()
set(libs_dir ${CMAKE_SOURCE_DIR}/libs)
add_subdirectory(libs EXCLUDE_FROM_ALL)
option(MG_EXPERIMENTAL_HIGH_AVAILABILITY "Feature flag for experimental high availability" OFF)
if (NOT MG_ENTERPRISE AND MG_EXPERIMENTAL_HIGH_AVAILABILITY)
set(MG_EXPERIMENTAL_HIGH_AVAILABILITY OFF)
message(FATAL_ERROR "MG_EXPERIMENTAL_HIGH_AVAILABILITY can only be used with enterpise version of the code.")
endif ()
if (MG_EXPERIMENTAL_HIGH_AVAILABILITY)
add_compile_definitions(MG_EXPERIMENTAL_HIGH_AVAILABILITY)
endif ()
# Optional subproject configuration -------------------------------------------
option(TEST_COVERAGE "Generate coverage reports from running memgraph" OFF)
option(TOOLS "Build tools binaries" ON)
option(QUERY_MODULES "Build query modules containing custom procedures" ON)

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -283,7 +283,7 @@ inline mgp_list *list_all_unique_constraints(mgp_graph *graph, mgp_memory *memor
}
// mgp_graph
inline bool graph_is_transactional(mgp_graph *graph) { return MgInvoke<int>(mgp_graph_is_transactional, graph); }
inline bool graph_is_mutable(mgp_graph *graph) { return MgInvoke<int>(mgp_graph_is_mutable, graph); }

1
libs/.gitignore vendored
View File

@@ -7,3 +7,4 @@
!pulsar.patch
!antlr4.10.1.patch
!rocksdb8.1.1.patch
!nuraft2.1.0.patch

View File

@@ -16,7 +16,7 @@ set(GFLAGS_NOTHREADS OFF)
# NOTE: config/generate.py depends on the gflags help XML format.
find_package(gflags REQUIRED)
find_package(fmt 8.0.1)
find_package(fmt 8.0.1 REQUIRED)
find_package(ZLIB 1.2.11 REQUIRED)
set(LIB_DIR ${CMAKE_CURRENT_SOURCE_DIR})

View File

@@ -5,7 +5,7 @@ index ee9b58c..31359a9 100644
@@ -48,7 +48,7 @@ option(LIBRDTSC_USE_PMU "Enables PMU usage on ARM platforms" OFF)
# | Library Build and Install Properties |
# +--------------------------------------------------------+
-add_library(rdtsc SHARED
+add_library(rdtsc
src/cycles.c
@@ -14,7 +14,7 @@ index ee9b58c..31359a9 100644
@@ -72,15 +72,6 @@ target_include_directories(rdtsc
PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include
)
-# Install directory changes depending on build mode
-if (CMAKE_BUILD_TYPE MATCHES "^[Dd]ebug")
- # During debug, the library will be installed into a local directory
@@ -27,3 +27,15 @@ index ee9b58c..31359a9 100644
# Specifying what to export when installing (GNUInstallDirs required)
install(TARGETS rdtsc
EXPORT librstsc-config
diff --git a/include/librdtsc/common_timer.h b/include/librdtsc/common_timer.h
index a6922d8..080dc77 100644
--- a/include/librdtsc/common_timer.h
+++ b/include/librdtsc/common_timer.h
@@ -2,6 +2,7 @@
#define LIBRDTSC_COMMON_TIMER_H
#include <librdtsc/common.h>
+#include <librdtsc/cycles.h>
extern uint64_t rdtsc_get_tsc_freq_arch();
extern uint64_t rdtsc_get_tsc_freq();

24
libs/nuraft2.1.0.patch Normal file
View File

@@ -0,0 +1,24 @@
diff --git a/include/libnuraft/asio_service_options.hxx b/include/libnuraft/asio_service_options.hxx
index 8fe1ec9..9497355 100644
--- a/include/libnuraft/asio_service_options.hxx
+++ b/include/libnuraft/asio_service_options.hxx
@@ -17,6 +17,7 @@ limitations under the License.
#pragma once
+#include <cstdint>
#include <functional>
#include <string>
#include <system_error>
diff --git a/include/libnuraft/callback.hxx b/include/libnuraft/callback.hxx
index 7b71624..d48c1e2 100644
--- a/include/libnuraft/callback.hxx
+++ b/include/libnuraft/callback.hxx
@@ -18,6 +18,7 @@ limitations under the License.
#ifndef _CALLBACK_H_
#define _CALLBACK_H_
+#include <cstdint>
#include <functional>
#include <string>

View File

@@ -1,21 +0,0 @@
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 6761929..6a369af 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -220,6 +220,7 @@ else()
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -momit-leaf-frame-pointer")
endif()
endif()
+ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-deprecated-copy -Wno-unused-but-set-variable")
endif()
include(CheckCCompilerFlag)
@@ -997,7 +998,7 @@ if(NOT WIN32 OR ROCKSDB_INSTALL_ON_WINDOWS)
if(ROCKSDB_BUILD_SHARED)
install(
- TARGETS ${ROCKSDB_SHARED_LIB}
+ TARGETS ${ROCKSDB_SHARED_LIB} OPTIONAL
EXPORT RocksDBTargets
COMPONENT runtime
ARCHIVE DESTINATION "${CMAKE_INSTALL_LIBDIR}"

View File

@@ -168,12 +168,11 @@ pushd antlr4
git apply ../antlr4.10.1.patch
popd
# cppitertools v2.0 2019-12-23
cppitertools_ref="cb3635456bdb531121b82b4d2e3afc7ae1f56d47"
cppitertools_ref="v2.1" # 2021-01-15
repo_clone_try_double "${primary_urls[cppitertools]}" "${secondary_urls[cppitertools]}" "cppitertools" "$cppitertools_ref"
# rapidcheck
rapidcheck_tag="7bc7d302191a4f3d0bf005692677126136e02f60" # (2020-05-04)
rapidcheck_tag="1c91f40e64d87869250cfb610376c629307bf77d" # (2023-08-15)
repo_clone_try_double "${primary_urls[rapidcheck]}" "${secondary_urls[rapidcheck]}" "rapidcheck" "$rapidcheck_tag"
# google benchmark
@@ -221,7 +220,7 @@ repo_clone_try_double "${primary_urls[pymgclient]}" "${secondary_urls[pymgclient
mgconsole_tag="v1.4.0" # (2023-05-21)
repo_clone_try_double "${primary_urls[mgconsole]}" "${secondary_urls[mgconsole]}" "mgconsole" "$mgconsole_tag" true
spdlog_tag="v1.9.2" # (2021-08-12)
spdlog_tag="v1.12.0" # (2022-11-02)
repo_clone_try_double "${primary_urls[spdlog]}" "${secondary_urls[spdlog]}" "spdlog" "$spdlog_tag" true
# librdkafka
@@ -286,5 +285,6 @@ repo_clone_try_double "${primary_urls[range-v3]}" "${secondary_urls[range-v3]}"
nuraft_tag="v2.1.0"
repo_clone_try_double "${primary_urls[nuraft]}" "${secondary_urls[nuraft]}" "nuraft" "$nuraft_tag" true
pushd nuraft
git apply ../nuraft2.1.0.patch
./prepare.sh
popd

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

View File

@@ -8,10 +8,12 @@
#pragma once
#include <json/json.hpp>
#include <cstdint>
#include <optional>
#include <string>
#include <json/json.hpp>
namespace memgraph::auth {
/// Need to be stable, auth durability depends on this
enum class PasswordHashAlgorithm : uint8_t { BCRYPT = 0, SHA256 = 1, SHA256_MULTIPLE = 2 };

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -15,6 +15,9 @@
#include "communication/bolt/v1/value.hpp"
#include "utils/logging.hpp"
#include "communication/bolt/v1/fmt.hpp"
#include "io/network/fmt.hpp"
namespace {
constexpr uint8_t kBoltV43Version[4] = {0x00, 0x00, 0x03, 0x04};
constexpr uint8_t kEmptyBoltVersion[4] = {0x00, 0x00, 0x00, 0x00};

View File

@@ -0,0 +1,27 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include "communication/bolt/v1/value.hpp"
template <>
class fmt::formatter<memgraph::communication::bolt::Value> : public fmt::ostream_formatter {};
template <>
class fmt::formatter<std::vector<memgraph::communication::bolt::Value>> : public fmt::ostream_formatter {};
template <>
class fmt::formatter<std::map<std::string, memgraph::communication::bolt::Value>> : public fmt::ostream_formatter {};
#endif

20
src/communication/fmt.hpp Normal file
View File

@@ -0,0 +1,20 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include <boost/asio/ip/tcp.hpp>
template <>
class fmt::formatter<boost::asio::ip::tcp::endpoint> : public fmt::ostream_formatter {};
#endif

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -21,6 +21,7 @@
#include <boost/beast/core.hpp>
#include "communication/context.hpp"
#include "communication/fmt.hpp"
#include "communication/http/session.hpp"
#include "utils/spin_lock.hpp"
#include "utils/synchronized.hpp"
@@ -82,7 +83,7 @@ class Listener final : public std::enable_shared_from_this<Listener<TRequestHand
return;
}
spdlog::info("HTTP server is listening on {}:{}", endpoint.address(), endpoint.port());
spdlog::info("HTTP server is listening on {}", endpoint);
}
void DoAccept() {

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -23,6 +23,7 @@
#include "communication/session.hpp"
#include "io/network/epoll.hpp"
#include "io/network/fmt.hpp"
#include "io/network/socket.hpp"
#include "utils/logging.hpp"
#include "utils/signals.hpp"

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -22,6 +22,7 @@
#include "communication/init.hpp"
#include "communication/listener.hpp"
#include "io/network/fmt.hpp"
#include "io/network/socket.hpp"
#include "utils/logging.hpp"
#include "utils/message.hpp"

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -26,6 +26,7 @@
#include <boost/asio/ip/tcp.hpp>
#include "communication/context.hpp"
#include "communication/fmt.hpp"
#include "communication/init.hpp"
#include "communication/v2/listener.hpp"
#include "communication/v2/pool.hpp"
@@ -129,7 +130,7 @@ bool Server<TSession, TSessionContext>::Start() {
listener_->Start();
spdlog::info("{} server is fully armed and operational", service_name_);
spdlog::info("{} listening on {}", service_name_, endpoint_.address());
spdlog::info("{} listening on {}", service_name_, endpoint_);
context_thread_pool_.Run();
return true;

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -47,6 +47,7 @@
#include "communication/buffer.hpp"
#include "communication/context.hpp"
#include "communication/exceptions.hpp"
#include "communication/fmt.hpp"
#include "dbms/global.hpp"
#include "utils/event_counter.hpp"
#include "utils/logging.hpp"
@@ -212,14 +213,11 @@ class WebsocketSession : public std::enable_shared_from_this<WebsocketSession<TS
session_.Execute();
DoRead();
} catch (const SessionClosedException &e) {
spdlog::info("{} client {}:{} closed the connection.", service_name_, remote_endpoint_.address(),
remote_endpoint_.port());
spdlog::info("{} client {} closed the connection.", service_name_, remote_endpoint_);
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::error("Exception was thrown while processing event in {} session associated with {}", service_name_,
remote_endpoint_);
spdlog::debug("Exception message: {}", e.what());
DoClose();
}
@@ -376,8 +374,7 @@ class Session final : public std::enable_shared_from_this<Session<TSession, TSes
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());
spdlog::info("Accepted a connection from {}: {}", service_name_, remote_endpoint_);
}
void DoRead() {
@@ -437,14 +434,11 @@ class Session final : public std::enable_shared_from_this<Session<TSession, TSes
session_.Execute();
DoRead();
} catch (const SessionClosedException &e) {
spdlog::info("{} client {}:{} closed the connection.", service_name_, remote_endpoint_.address(),
remote_endpoint_.port());
spdlog::info("{} client {} closed the connection.", service_name_, remote_endpoint_);
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::error("Exception was thrown while processing event in {} session associated with {}", service_name_,
remote_endpoint_);
spdlog::debug("Exception message: {}", e.what());
DoShutdown();
}

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -10,6 +10,7 @@
// licenses/APL.txt.
#include "communication/websocket/listener.hpp"
#include "communication/fmt.hpp"
namespace memgraph::communication::websocket {
namespace {
@@ -61,7 +62,7 @@ Listener::Listener(boost::asio::io_context &ioc, ServerContext *context, tcp::en
return;
}
spdlog::info("WebSocket server is listening on {}:{}", endpoint.address(), endpoint.port());
spdlog::info("WebSocket server is listening on {}", endpoint);
}
void Listener::DoAccept() {

View File

@@ -11,10 +11,10 @@ target_sources(mg-coordination
include/coordination/coordinator_slk.hpp
include/coordination/coordinator_instance.hpp
include/coordination/coordinator_handlers.hpp
include/coordination/constants.hpp
include/coordination/instance_status.hpp
include/coordination/replication_instance.hpp
include/coordination/raft_state.hpp
include/coordination/rpc_errors.hpp
include/nuraft/coordinator_log_store.hpp
include/nuraft/coordinator_state_machine.hpp

View File

@@ -17,6 +17,7 @@
#include "coordination/coordinator_config.hpp"
#include "coordination/coordinator_rpc.hpp"
#include "replication_coordination_glue/messages.hpp"
#include "utils/result.hpp"
namespace memgraph::coordination {
@@ -45,6 +46,10 @@ auto CoordinatorClient::InstanceDownTimeoutSec() const -> std::chrono::seconds {
return config_.instance_down_timeout_sec;
}
auto CoordinatorClient::InstanceGetUUIDFrequencySec() const -> std::chrono::seconds {
return config_.instance_get_uuid_frequency_sec;
}
void CoordinatorClient::StartFrequentCheck() {
if (instance_checker_.IsRunning()) {
return;
@@ -140,5 +145,31 @@ auto CoordinatorClient::SendUnregisterReplicaRpc(std::string const &instance_nam
return false;
}
auto CoordinatorClient::SendGetInstanceUUIDRpc() const
-> utils::BasicResult<GetInstanceUUIDError, std::optional<utils::UUID>> {
try {
auto stream{rpc_client_.Stream<GetInstanceUUIDRpc>()};
auto res = stream.AwaitResponse();
return res.uuid;
} catch (const rpc::RpcFailedException &) {
spdlog::error("RPC error occured while sending GetInstance UUID RPC");
return GetInstanceUUIDError::RPC_EXCEPTION;
}
}
auto CoordinatorClient::SendEnableWritingOnMainRpc() const -> bool {
try {
auto stream{rpc_client_.Stream<EnableWritingOnMainRpc>()};
if (!stream.AwaitResponse().success) {
spdlog::error("Failed to receive successful RPC response for enabling writing on main!");
return false;
}
return true;
} catch (rpc::RpcFailedException const &) {
spdlog::error("Failed to enable writing on main!");
}
return false;
}
} // namespace memgraph::coordination
#endif

View File

@@ -45,6 +45,18 @@ void CoordinatorHandlers::Register(memgraph::coordination::CoordinatorServer &se
spdlog::info("Received UnregisterReplicaRpc on coordinator server");
CoordinatorHandlers::UnregisterReplicaHandler(replication_handler, req_reader, res_builder);
});
server.Register<coordination::EnableWritingOnMainRpc>(
[&replication_handler](slk::Reader *req_reader, slk::Builder *res_builder) -> void {
spdlog::info("Received EnableWritingOnMainRpc on coordinator server");
CoordinatorHandlers::EnableWritingOnMainHandler(replication_handler, req_reader, res_builder);
});
server.Register<coordination::GetInstanceUUIDRpc>(
[&replication_handler](slk::Reader *req_reader, slk::Builder *res_builder) -> void {
spdlog::info("Received GetInstanceUUIDRpc on coordinator server");
CoordinatorHandlers::GetInstanceUUIDHandler(replication_handler, req_reader, res_builder);
});
}
void CoordinatorHandlers::SwapMainUUIDHandler(replication::ReplicationHandler &replication_handler,
@@ -68,12 +80,6 @@ void CoordinatorHandlers::DemoteMainToReplicaHandler(replication::ReplicationHan
slk::Reader *req_reader, slk::Builder *res_builder) {
spdlog::info("Executing DemoteMainToReplicaHandler");
if (!replication_handler.IsMain()) {
spdlog::error("Setting to replica must be performed on main.");
slk::Save(coordination::DemoteMainToReplicaRes{false}, res_builder);
return;
}
coordination::DemoteMainToReplicaReq req;
slk::Load(&req, req_reader);
@@ -83,11 +89,18 @@ void CoordinatorHandlers::DemoteMainToReplicaHandler(replication::ReplicationHan
if (!replication_handler.SetReplicationRoleReplica(clients_config, std::nullopt)) {
spdlog::error("Demoting main to replica failed!");
slk::Save(coordination::PromoteReplicaToMainRes{false}, res_builder);
slk::Save(coordination::DemoteMainToReplicaRes{false}, res_builder);
return;
}
slk::Save(coordination::PromoteReplicaToMainRes{true}, res_builder);
slk::Save(coordination::DemoteMainToReplicaRes{true}, res_builder);
}
void CoordinatorHandlers::GetInstanceUUIDHandler(replication::ReplicationHandler &replication_handler,
slk::Reader * /*req_reader*/, slk::Builder *res_builder) {
spdlog::info("Executing GetInstanceUUIDHandler");
slk::Save(coordination::GetInstanceUUIDRes{replication_handler.GetReplicaUUID()}, res_builder);
}
void CoordinatorHandlers::PromoteReplicaToMainHandler(replication::ReplicationHandler &replication_handler,
@@ -119,7 +132,7 @@ void CoordinatorHandlers::PromoteReplicaToMainHandler(replication::ReplicationHa
// registering replicas
for (auto const &config : req.replication_clients_info | ranges::views::transform(converter)) {
auto instance_client = replication_handler.RegisterReplica(config, false);
auto instance_client = replication_handler.RegisterReplica(config);
if (instance_client.HasError()) {
using enum memgraph::replication::RegisterReplicaError;
switch (instance_client.GetError()) {
@@ -148,7 +161,7 @@ void CoordinatorHandlers::PromoteReplicaToMainHandler(replication::ReplicationHa
}
}
}
spdlog::error(fmt::format("FICO : Promote replica to main was success {}", std::string(req.main_uuid_)));
spdlog::info("Promote replica to main was success {}", std::string(req.main_uuid_));
slk::Save(coordination::PromoteReplicaToMainRes{true}, res_builder);
}
@@ -184,5 +197,22 @@ void CoordinatorHandlers::UnregisterReplicaHandler(replication::ReplicationHandl
}
}
void CoordinatorHandlers::EnableWritingOnMainHandler(replication::ReplicationHandler &replication_handler,
slk::Reader * /*req_reader*/, slk::Builder *res_builder) {
if (!replication_handler.IsMain()) {
spdlog::error("Enable writing on main must be performed on main!");
slk::Save(coordination::EnableWritingOnMainRes{false}, res_builder);
return;
}
if (!replication_handler.GetReplState().EnableWritingOnMain()) {
spdlog::error("Enabling writing on main failed!");
slk::Save(coordination::EnableWritingOnMainRes{false}, res_builder);
return;
}
slk::Save(coordination::EnableWritingOnMainRes{true}, res_builder);
}
} // namespace memgraph::dbms
#endif

View File

@@ -14,9 +14,11 @@
#include "coordination/coordinator_instance.hpp"
#include "coordination/coordinator_exceptions.hpp"
#include "coordination/fmt.hpp"
#include "nuraft/coordinator_state_machine.hpp"
#include "nuraft/coordinator_state_manager.hpp"
#include "utils/counter.hpp"
#include "utils/functional.hpp"
#include <range/v3/view.hpp>
#include <shared_mutex>
@@ -47,9 +49,12 @@ CoordinatorInstance::CoordinatorInstance()
spdlog::trace("Instance {} performing replica successful callback", repl_instance_name);
auto &repl_instance = find_repl_instance(self, repl_instance_name);
// We need to get replicas UUID from time to time to ensure replica is listening to correct main
// and that it didn't go down for less time than we could notice
// We need to get id of main replica is listening to
// and swap if necessary
if (!repl_instance.EnsureReplicaHasCorrectMainUUID(self->GetMainUUID())) {
spdlog::error(
fmt::format("Failed to swap uuid for replica instance {} which is alive", repl_instance.InstanceName()));
spdlog::error("Failed to swap uuid for replica instance {} which is alive", repl_instance.InstanceName());
return;
}
@@ -61,14 +66,6 @@ CoordinatorInstance::CoordinatorInstance()
spdlog::trace("Instance {} performing replica failure callback", repl_instance_name);
auto &repl_instance = find_repl_instance(self, repl_instance_name);
repl_instance.OnFailPing();
// We need to restart main uuid from instance since it was "down" at least a second
// There is slight delay, if we choose to use isAlive, instance can be down and back up in less than
// our isAlive time difference, which would lead to instance setting UUID to nullopt and stopping accepting any
// incoming RPCs from valid main
// TODO(antoniofilipovic) this needs here more complex logic
// We need to get id of main replica is listening to on successful ping
// and swap it to correct uuid if it failed
repl_instance.ResetMainUUID();
};
main_succ_cb_ = [find_repl_instance](CoordinatorInstance *self, std::string_view repl_instance_name) -> void {
@@ -87,6 +84,11 @@ CoordinatorInstance::CoordinatorInstance()
auto const curr_main_uuid = self->GetMainUUID();
if (curr_main_uuid == repl_instance_uuid.value()) {
if (!repl_instance.EnableWritingOnMain()) {
spdlog::error("Failed to enable writing on main instance {}", repl_instance_name);
return;
}
repl_instance.OnSuccessPing();
return;
}
@@ -125,9 +127,6 @@ CoordinatorInstance::CoordinatorInstance()
auto CoordinatorInstance::ShowInstances() const -> std::vector<InstanceStatus> {
auto const coord_instances = raft_state_.GetAllCoordinators();
std::vector<InstanceStatus> instances_status;
instances_status.reserve(repl_instances_.size() + coord_instances.size());
auto const stringify_repl_role = [](ReplicationInstance const &instance) -> std::string {
if (!instance.IsAlive()) return "unknown";
if (instance.IsMain()) return "main";
@@ -149,8 +148,7 @@ auto CoordinatorInstance::ShowInstances() const -> std::vector<InstanceStatus> {
// CoordinatorState to every instance, we can be smarter about this using our RPC.
};
std::ranges::transform(coord_instances, std::back_inserter(instances_status), coord_instance_to_status);
auto instances_status = utils::fmap(coord_instance_to_status, coord_instances);
{
auto lock = std::shared_lock{coord_instance_lock_};
std::ranges::transform(repl_instances_, std::back_inserter(instances_status), repl_instance_to_status);
@@ -189,10 +187,9 @@ auto CoordinatorInstance::TryFailover() -> void {
}
}
ReplicationClientsInfo repl_clients_info;
repl_clients_info.reserve(repl_instances_.size() - 1);
std::ranges::transform(repl_instances_ | ranges::views::filter(is_not_new_main),
std::back_inserter(repl_clients_info), &ReplicationInstance::ReplicationClientInfo);
auto repl_clients_info = repl_instances_ | ranges::views::filter(is_not_new_main) |
ranges::views::transform(&ReplicationInstance::ReplicationClientInfo) |
ranges::to<ReplicationClientsInfo>();
if (!new_main->PromoteToMain(new_main_uuid, std::move(repl_clients_info), main_succ_cb_, main_fail_cb_)) {
spdlog::warn("Failover failed since promoting replica to main failed!");
@@ -208,6 +205,10 @@ auto CoordinatorInstance::SetReplicationInstanceToMain(std::string instance_name
-> SetInstanceToMainCoordinatorStatus {
auto lock = std::lock_guard{coord_instance_lock_};
if (std::ranges::any_of(repl_instances_, &ReplicationInstance::IsMain)) {
return SetInstanceToMainCoordinatorStatus::MAIN_ALREADY_EXISTS;
}
auto const is_new_main = [&instance_name](ReplicationInstance const &instance) {
return instance.InstanceName() == instance_name;
};

View File

@@ -68,6 +68,35 @@ void UnregisterReplicaRes::Load(UnregisterReplicaRes *self, memgraph::slk::Reade
memgraph::slk::Load(self, reader);
}
void EnableWritingOnMainRes::Save(EnableWritingOnMainRes const &self, memgraph::slk::Builder *builder) {
memgraph::slk::Save(self, builder);
}
void EnableWritingOnMainRes::Load(EnableWritingOnMainRes *self, memgraph::slk::Reader *reader) {
memgraph::slk::Load(self, reader);
}
void EnableWritingOnMainReq::Save(EnableWritingOnMainReq const &self, memgraph::slk::Builder *builder) {}
void EnableWritingOnMainReq::Load(EnableWritingOnMainReq *self, memgraph::slk::Reader *reader) {}
// GetInstanceUUID
void GetInstanceUUIDReq::Save(const GetInstanceUUIDReq &self, memgraph::slk::Builder *builder) {
memgraph::slk::Save(self, builder);
}
void GetInstanceUUIDReq::Load(GetInstanceUUIDReq *self, memgraph::slk::Reader *reader) {
memgraph::slk::Load(self, reader);
}
void GetInstanceUUIDRes::Save(const GetInstanceUUIDRes &self, memgraph::slk::Builder *builder) {
memgraph::slk::Save(self, builder);
}
void GetInstanceUUIDRes::Load(GetInstanceUUIDRes *self, memgraph::slk::Reader *reader) {
memgraph::slk::Load(self, reader);
}
} // namespace coordination
constexpr utils::TypeInfo coordination::PromoteReplicaToMainReq::kType{utils::TypeId::COORD_FAILOVER_REQ,
@@ -89,8 +118,22 @@ constexpr utils::TypeInfo coordination::UnregisterReplicaReq::kType{utils::TypeI
constexpr utils::TypeInfo coordination::UnregisterReplicaRes::kType{utils::TypeId::COORD_UNREGISTER_REPLICA_RES,
"UnregisterReplicaRes", nullptr};
constexpr utils::TypeInfo coordination::EnableWritingOnMainReq::kType{utils::TypeId::COORD_ENABLE_WRITING_ON_MAIN_REQ,
"CoordEnableWritingOnMainReq", nullptr};
constexpr utils::TypeInfo coordination::EnableWritingOnMainRes::kType{utils::TypeId::COORD_ENABLE_WRITING_ON_MAIN_RES,
"CoordEnableWritingOnMainRes", nullptr};
constexpr utils::TypeInfo coordination::GetInstanceUUIDReq::kType{utils::TypeId::COORD_GET_UUID_REQ, "CoordGetUUIDReq",
nullptr};
constexpr utils::TypeInfo coordination::GetInstanceUUIDRes::kType{utils::TypeId::COORD_GET_UUID_RES, "CoordGetUUIDRes",
nullptr};
namespace slk {
// PromoteReplicaToMainRpc
void Save(const memgraph::coordination::PromoteReplicaToMainRes &self, memgraph::slk::Builder *builder) {
memgraph::slk::Save(self.success, builder);
}
@@ -109,6 +152,7 @@ void Load(memgraph::coordination::PromoteReplicaToMainReq *self, memgraph::slk::
memgraph::slk::Load(&self->replication_clients_info, reader);
}
// DemoteMainToReplicaRpc
void Save(const memgraph::coordination::DemoteMainToReplicaReq &self, memgraph::slk::Builder *builder) {
memgraph::slk::Save(self.replication_client_info, builder);
}
@@ -125,6 +169,8 @@ void Load(memgraph::coordination::DemoteMainToReplicaRes *self, memgraph::slk::R
memgraph::slk::Load(&self->success, reader);
}
// UnregisterReplicaRpc
void Save(memgraph::coordination::UnregisterReplicaReq const &self, memgraph::slk::Builder *builder) {
memgraph::slk::Save(self.instance_name, builder);
}
@@ -141,6 +187,32 @@ void Load(memgraph::coordination::UnregisterReplicaRes *self, memgraph::slk::Rea
memgraph::slk::Load(&self->success, reader);
}
void Save(memgraph::coordination::EnableWritingOnMainRes const &self, memgraph::slk::Builder *builder) {
memgraph::slk::Save(self.success, builder);
}
void Load(memgraph::coordination::EnableWritingOnMainRes *self, memgraph::slk::Reader *reader) {
memgraph::slk::Load(&self->success, reader);
}
// GetInstanceUUIDRpc
void Save(const memgraph::coordination::GetInstanceUUIDReq & /*self*/, memgraph::slk::Builder * /*builder*/) {
/* nothing to serialize*/
}
void Load(memgraph::coordination::GetInstanceUUIDReq * /*self*/, memgraph::slk::Reader * /*reader*/) {
/* nothing to serialize*/
}
void Save(const memgraph::coordination::GetInstanceUUIDRes &self, memgraph::slk::Builder *builder) {
memgraph::slk::Save(self.uuid, builder);
}
void Load(memgraph::coordination::GetInstanceUUIDRes *self, memgraph::slk::Reader *reader) {
memgraph::slk::Load(&self->uuid, reader);
}
} // namespace slk
} // namespace memgraph

60
src/coordination/fmt.hpp Normal file
View File

@@ -0,0 +1,60 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include <string>
#include <libnuraft/nuraft.hxx>
#include "utils/logging.hpp"
inline std::string ToString(const nuraft::cmd_result_code &code) {
switch (code) {
case nuraft::cmd_result_code::OK:
return "OK";
case nuraft::cmd_result_code::FAILED:
return "FAILED";
case nuraft::cmd_result_code::RESULT_NOT_EXIST_YET:
return "RESULT_NOT_EXIST_YET";
case nuraft::cmd_result_code::TERM_MISMATCH:
return "TERM_MISMATCH";
case nuraft::cmd_result_code::SERVER_IS_LEAVING:
return "SERVER_IS_LEAVING";
case nuraft::cmd_result_code::CANNOT_REMOVE_LEADER:
return "CANNOT_REMOVE_LEADER";
case nuraft::cmd_result_code::SERVER_NOT_FOUND:
return "SERVER_NOT_FOUND";
case nuraft::cmd_result_code::SERVER_IS_JOINING:
return "SERVER_IS_JOINING";
case nuraft::cmd_result_code::CONFIG_CHANGING:
return "CONFIG_CHANGING";
case nuraft::cmd_result_code::SERVER_ALREADY_EXISTS:
return "SERVER_ALREADY_EXISTS";
case nuraft::cmd_result_code::BAD_REQUEST:
return "BAD_REQUEST";
case nuraft::cmd_result_code::NOT_LEADER:
return "NOT_LEADER";
case nuraft::cmd_result_code::TIMEOUT:
return "TIMEOUT";
case nuraft::cmd_result_code::CANCELLED:
return "CANCELLED";
}
LOG_FATAL("ToString of a nuraft::cmd_result_code -> check missing switch case");
}
inline std::ostream &operator<<(std::ostream &os, const nuraft::cmd_result_code &code) {
os << ToString(code);
return os;
}
template <>
class fmt::formatter<nuraft::cmd_result_code> : public fmt::ostream_formatter {};
#endif

View File

@@ -11,12 +11,14 @@
#pragma once
#include "utils/uuid.hpp"
#ifdef MG_ENTERPRISE
#include "coordination/coordinator_config.hpp"
#include "rpc/client.hpp"
#include "rpc_errors.hpp"
#include "utils/result.hpp"
#include "utils/scheduler.hpp"
#include "utils/uuid.hpp"
namespace memgraph::coordination {
@@ -46,7 +48,7 @@ class CoordinatorClient {
auto SocketAddress() const -> std::string;
[[nodiscard]] auto DemoteToReplica() const -> bool;
// TODO: (andi) Consistent naming
auto SendPromoteReplicaToMainRpc(const utils::UUID &uuid, ReplicationClientsInfo replication_clients_info) const
-> bool;
@@ -54,6 +56,10 @@ class CoordinatorClient {
auto SendUnregisterReplicaRpc(std::string const &instance_name) const -> bool;
auto SendEnableWritingOnMainRpc() const -> bool;
auto SendGetInstanceUUIDRpc() const -> memgraph::utils::BasicResult<GetInstanceUUIDError, std::optional<utils::UUID>>;
auto ReplicationClientInfo() const -> ReplClientInfo;
auto SetCallbacks(HealthCheckCallback succ_cb, HealthCheckCallback fail_cb) -> void;
@@ -62,6 +68,8 @@ class CoordinatorClient {
auto InstanceDownTimeoutSec() const -> std::chrono::seconds;
auto InstanceGetUUIDFrequencySec() const -> std::chrono::seconds;
friend bool operator==(CoordinatorClient const &first, CoordinatorClient const &second) {
return first.config_ == second.config_;
}
@@ -69,7 +77,6 @@ class CoordinatorClient {
private:
utils::Scheduler instance_checker_;
// TODO: (andi) Pimpl?
communication::ClientContext rpc_context_;
mutable rpc::Client rpc_client_;

View File

@@ -30,6 +30,7 @@ struct CoordinatorClientConfig {
uint16_t port{};
std::chrono::seconds instance_health_check_frequency_sec{1};
std::chrono::seconds instance_down_timeout_sec{5};
std::chrono::seconds instance_get_uuid_frequency_sec{10};
auto SocketAddress() const -> std::string { return ip_address + ":" + std::to_string(port); }

View File

@@ -36,6 +36,11 @@ class CoordinatorHandlers {
static void UnregisterReplicaHandler(replication::ReplicationHandler &replication_handler, slk::Reader *req_reader,
slk::Builder *res_builder);
static void EnableWritingOnMainHandler(replication::ReplicationHandler &replication_handler, slk::Reader *req_reader,
slk::Builder *res_builder);
static void GetInstanceUUIDHandler(replication::ReplicationHandler &replication_handler, slk::Reader *req_reader,
slk::Builder *res_builder);
};
} // namespace memgraph::dbms

View File

@@ -111,6 +111,56 @@ struct UnregisterReplicaRes {
using UnregisterReplicaRpc = rpc::RequestResponse<UnregisterReplicaReq, UnregisterReplicaRes>;
struct EnableWritingOnMainReq {
static const utils::TypeInfo kType;
static const utils::TypeInfo &GetTypeInfo() { return kType; }
static void Load(EnableWritingOnMainReq *self, memgraph::slk::Reader *reader);
static void Save(EnableWritingOnMainReq const &self, memgraph::slk::Builder *builder);
EnableWritingOnMainReq() = default;
};
struct EnableWritingOnMainRes {
static const utils::TypeInfo kType;
static const utils::TypeInfo &GetTypeInfo() { return kType; }
static void Load(EnableWritingOnMainRes *self, memgraph::slk::Reader *reader);
static void Save(EnableWritingOnMainRes const &self, memgraph::slk::Builder *builder);
explicit EnableWritingOnMainRes(bool success) : success(success) {}
EnableWritingOnMainRes() = default;
bool success;
};
using EnableWritingOnMainRpc = rpc::RequestResponse<EnableWritingOnMainReq, EnableWritingOnMainRes>;
struct GetInstanceUUIDReq {
static const utils::TypeInfo kType;
static const utils::TypeInfo &GetTypeInfo() { return kType; }
static void Load(GetInstanceUUIDReq *self, memgraph::slk::Reader *reader);
static void Save(const GetInstanceUUIDReq &self, memgraph::slk::Builder *builder);
GetInstanceUUIDReq() = default;
};
struct GetInstanceUUIDRes {
static const utils::TypeInfo kType;
static const utils::TypeInfo &GetTypeInfo() { return kType; }
static void Load(GetInstanceUUIDRes *self, memgraph::slk::Reader *reader);
static void Save(const GetInstanceUUIDRes &self, memgraph::slk::Builder *builder);
explicit GetInstanceUUIDRes(std::optional<utils::UUID> uuid) : uuid(uuid) {}
GetInstanceUUIDRes() = default;
std::optional<utils::UUID> uuid;
};
using GetInstanceUUIDRpc = rpc::RequestResponse<GetInstanceUUIDReq, GetInstanceUUIDRes>;
} // namespace memgraph::coordination
// SLK serialization declarations
@@ -128,12 +178,20 @@ void Load(memgraph::coordination::DemoteMainToReplicaRes *self, memgraph::slk::R
void Save(const memgraph::coordination::DemoteMainToReplicaReq &self, memgraph::slk::Builder *builder);
void Load(memgraph::coordination::DemoteMainToReplicaReq *self, memgraph::slk::Reader *reader);
// GetInstanceUUIDRpc
void Save(const memgraph::coordination::GetInstanceUUIDReq &self, memgraph::slk::Builder *builder);
void Load(memgraph::coordination::GetInstanceUUIDReq *self, memgraph::slk::Reader *reader);
void Save(const memgraph::coordination::GetInstanceUUIDRes &self, memgraph::slk::Builder *builder);
void Load(memgraph::coordination::GetInstanceUUIDRes *self, memgraph::slk::Reader *reader);
// UnregisterReplicaRpc
void Save(memgraph::coordination::UnregisterReplicaRes const &self, memgraph::slk::Builder *builder);
void Load(memgraph::coordination::UnregisterReplicaRes *self, memgraph::slk::Reader *reader);
void Save(memgraph::coordination::UnregisterReplicaReq const &self, memgraph::slk::Builder *builder);
void Load(memgraph::coordination::UnregisterReplicaReq *self, memgraph::slk::Reader *reader);
void Save(memgraph::coordination::EnableWritingOnMainRes const &self, memgraph::slk::Builder *builder);
void Load(memgraph::coordination::EnableWritingOnMainRes *self, memgraph::slk::Reader *reader);
} // namespace memgraph::slk
#endif

View File

@@ -15,8 +15,6 @@
#include <flags/replication.hpp>
#include <optional>
#include <libnuraft/nuraft.hxx>
namespace memgraph::coordination {

View File

@@ -39,6 +39,7 @@ enum class UnregisterInstanceCoordinatorStatus : uint8_t {
enum class SetInstanceToMainCoordinatorStatus : uint8_t {
NO_INSTANCE_WITH_NAME,
MAIN_ALREADY_EXISTS,
NOT_COORDINATOR,
SUCCESS,
COULD_NOT_PROMOTE_TO_MAIN,

View File

@@ -18,6 +18,7 @@
#include "replication_coordination_glue/role.hpp"
#include <libnuraft/nuraft.hxx>
#include "utils/result.hpp"
#include "utils/uuid.hpp"
namespace memgraph::coordination {
@@ -37,6 +38,9 @@ class ReplicationInstance {
auto OnSuccessPing() -> void;
auto OnFailPing() -> bool;
auto IsReadyForUUIDPing() -> bool;
void UpdateReplicaLastResponseUUID();
auto IsAlive() const -> bool;
@@ -62,9 +66,12 @@ class ReplicationInstance {
auto SendSwapAndUpdateUUID(const utils::UUID &new_main_uuid) -> bool;
auto SendUnregisterReplicaRpc(std::string const &instance_name) -> bool;
// TODO: (andi) Inconsistent API
auto SendGetInstanceUUID() -> utils::BasicResult<coordination::GetInstanceUUIDError, std::optional<utils::UUID>>;
auto GetClient() -> CoordinatorClient &;
auto EnableWritingOnMain() -> bool;
auto SetNewMainUUID(utils::UUID const &main_uuid) -> void;
auto ResetMainUUID() -> void;
auto GetMainUUID() const -> const std::optional<utils::UUID> &;
@@ -74,6 +81,7 @@ class ReplicationInstance {
replication_coordination_glue::ReplicationRole replication_role_;
std::chrono::system_clock::time_point last_response_time_{};
bool is_alive_{false};
std::chrono::system_clock::time_point last_check_of_uuid_{};
// for replica this is main uuid of current main
// for "main" main this same as in CoordinatorData

View File

@@ -9,14 +9,6 @@
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
namespace memgraph::coordination {
#ifdef MG_EXPERIMENTAL_HIGH_AVAILABILITY
constexpr bool allow_ha = true;
#else
constexpr bool allow_ha = false;
#endif
enum class GetInstanceUUIDError { NO_RESPONSE, RPC_EXCEPTION };
} // namespace memgraph::coordination

View File

@@ -14,6 +14,7 @@
#include "coordination/replication_instance.hpp"
#include "replication_coordination_glue/handler.hpp"
#include "utils/result.hpp"
namespace memgraph::coordination {
@@ -39,6 +40,11 @@ auto ReplicationInstance::OnFailPing() -> bool {
return is_alive_;
}
auto ReplicationInstance::IsReadyForUUIDPing() -> bool {
return std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now() - last_check_of_uuid_) >
client_.InstanceGetUUIDFrequencySec();
}
auto ReplicationInstance::InstanceName() const -> std::string { return client_.InstanceName(); }
auto ReplicationInstance::SocketAddress() const -> std::string { return client_.SocketAddress(); }
auto ReplicationInstance::IsAlive() const -> bool { return is_alive_; }
@@ -91,10 +97,20 @@ auto ReplicationInstance::ResetMainUUID() -> void { main_uuid_ = std::nullopt; }
auto ReplicationInstance::GetMainUUID() const -> std::optional<utils::UUID> const & { return main_uuid_; }
auto ReplicationInstance::EnsureReplicaHasCorrectMainUUID(utils::UUID const &curr_main_uuid) -> bool {
if (!main_uuid_ || *main_uuid_ != curr_main_uuid) {
return SendSwapAndUpdateUUID(curr_main_uuid);
if (!IsReadyForUUIDPing()) {
return true;
}
return true;
auto res = SendGetInstanceUUID();
if (res.HasError()) {
return false;
}
UpdateReplicaLastResponseUUID();
if (res.GetValue().has_value() && res.GetValue().value() == curr_main_uuid) {
return true;
}
return SendSwapAndUpdateUUID(curr_main_uuid);
}
auto ReplicationInstance::SendSwapAndUpdateUUID(const utils::UUID &new_main_uuid) -> bool {
@@ -109,5 +125,14 @@ auto ReplicationInstance::SendUnregisterReplicaRpc(std::string const &instance_n
return client_.SendUnregisterReplicaRpc(instance_name);
}
auto ReplicationInstance::EnableWritingOnMain() -> bool { return client_.SendEnableWritingOnMainRpc(); }
auto ReplicationInstance::SendGetInstanceUUID()
-> utils::BasicResult<coordination::GetInstanceUUIDError, std::optional<utils::UUID>> {
return client_.SendGetInstanceUUIDRpc();
}
void ReplicationInstance::UpdateReplicaLastResponseUUID() { last_check_of_uuid_ = std::chrono::system_clock::now(); }
} // namespace memgraph::coordination
#endif

View File

@@ -19,6 +19,7 @@
#include "storage/v2/durability/durability.hpp"
#include "storage/v2/durability/snapshot.hpp"
#include "storage/v2/durability/version.hpp"
#include "storage/v2/fmt.hpp"
#include "storage/v2/indices/label_index_stats.hpp"
#include "storage/v2/inmemory/storage.hpp"
#include "storage/v2/inmemory/unique_constraints.hpp"

View File

View File

@@ -10,6 +10,7 @@
// licenses/APL.txt.
#pragma once
#include <algorithm>
#include <atomic>
#include <compare>
#include <cstdint>

View File

@@ -19,13 +19,14 @@
// Bolt server flags.
// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_string(experimental_enabled, "",
"Experimental features to be used, comma seperated. Options [system-replication]");
"Experimental features to be used, comma seperated. Options [system-replication, high-availability]");
using namespace std::string_view_literals;
namespace memgraph::flags {
auto const mapping = std::map{std::pair{"system-replication"sv, Experiments::SYSTEM_REPLICATION}};
auto const mapping = std::map{std::pair{"system-replication"sv, Experiments::SYSTEM_REPLICATION},
std::pair{"high-availability"sv, Experiments::HIGH_AVAILABILITY}};
auto ExperimentsInstance() -> Experiments & {
static auto instance = Experiments{};

View File

@@ -23,6 +23,7 @@ namespace memgraph::flags {
// old experiments can be reused once code cleanup has happened
enum class Experiments : uint8_t {
SYSTEM_REPLICATION = 1 << 0,
HIGH_AVAILABILITY = 1 << 1,
};
bool AreExperimentsEnabled(Experiments experiments);

View File

@@ -22,6 +22,8 @@ DEFINE_uint32(raft_server_id, 0, "Unique ID of the raft server.");
DEFINE_uint32(instance_down_timeout_sec, 5, "Time duration after which an instance is considered down.");
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_uint32(instance_health_check_frequency_sec, 1, "The time duration between two health checks/pings.");
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_uint32(instance_get_uuid_frequency_sec, 10, "The time duration between two instance uuid checks.");
#endif
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)

View File

@@ -24,6 +24,8 @@ DECLARE_uint32(raft_server_id);
DECLARE_uint32(instance_down_timeout_sec);
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DECLARE_uint32(instance_health_check_frequency_sec);
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DECLARE_uint32(instance_get_uuid_frequency_sec);
#endif
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -22,6 +22,7 @@
#include "integrations/constants.hpp"
#include "integrations/kafka/exceptions.hpp"
#include "integrations/kafka/fmt.hpp"
#include "utils/exceptions.hpp"
#include "utils/logging.hpp"
#include "utils/on_scope_exit.hpp"

View File

@@ -0,0 +1,25 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include <librdkafka/rdkafkacpp.h>
inline std::ostream &operator<<(std::ostream &os, const RdKafka::ErrorCode &code) {
os << RdKafka::err2str(code);
return os;
}
template <>
class fmt::formatter<RdKafka::ErrorCode> : public fmt::ostream_formatter {};
#endif

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -15,12 +15,12 @@
#include <chrono>
#include <thread>
#include <fmt/format.h>
#include <pulsar/Client.h>
#include <pulsar/InitialPosition.h>
#include "integrations/constants.hpp"
#include "integrations/pulsar/exceptions.hpp"
#include "integrations/pulsar/fmt.hpp"
#include "utils/concepts.hpp"
#include "utils/logging.hpp"
#include "utils/on_scope_exit.hpp"

View File

@@ -0,0 +1,21 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include "integrations/pulsar/consumer.hpp"
template <>
class fmt::formatter<memgraph::integrations::pulsar::pulsar_client::Result> : public fmt::ostream_formatter {};
#endif

21
src/io/network/fmt.hpp Normal file
View File

@@ -0,0 +1,21 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include "io/network/endpoint.hpp"
template <>
class fmt::formatter<memgraph::io::network::Endpoint> : public fmt::ostream_formatter {};
#endif

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -11,6 +11,7 @@
#pragma once
#include <cstddef>
#include <cstdint>
namespace memgraph::io::network {

View File

@@ -160,13 +160,14 @@ class KVStore final {
* and behaves as if all of those pairs are stored in a single iterable
* collection of std::pair<std::string, std::string>.
*/
class iterator final : public std::iterator<std::input_iterator_tag, // iterator_category
std::pair<std::string, std::string>, // value_type
long, // difference_type
const std::pair<std::string, std::string> *, // pointer
const std::pair<std::string, std::string> & // reference
> {
class iterator final {
public:
using iterator_concept [[maybe_unused]] = std::input_iterator_tag;
using value_type = std::pair<std::string, std::string>;
using difference_type = long;
using pointer = const std::pair<std::string, std::string> *;
using reference = const std::pair<std::string, std::string> &;
explicit iterator(const KVStore *kvstore, const std::string &prefix = "", bool at_end = false);
iterator(const iterator &other) = delete;

View File

@@ -359,6 +359,7 @@ int main(int argc, char **argv) {
#ifdef MG_ENTERPRISE
.instance_down_timeout_sec = std::chrono::seconds(FLAGS_instance_down_timeout_sec),
.instance_health_check_frequency_sec = std::chrono::seconds(FLAGS_instance_health_check_frequency_sec),
.instance_get_uuid_frequency_sec = std::chrono::seconds(FLAGS_instance_get_uuid_frequency_sec),
#endif
.default_kafka_bootstrap_servers = FLAGS_kafka_bootstrap_servers,
.default_pulsar_service_url = FLAGS_pulsar_service_url,

View File

@@ -61,10 +61,12 @@ void *my_alloc(extent_hooks_t *extent_hooks, void *new_addr, size_t size, size_t
// This needs to be before, to throw exception in case of too big alloc
if (*commit) [[likely]] {
if (GetQueriesMemoryControl().IsThreadTracked()) [[unlikely]] {
GetQueriesMemoryControl().TrackAllocOnCurrentThread(size);
bool ok = GetQueriesMemoryControl().TrackAllocOnCurrentThread(size);
if (!ok) return nullptr;
}
// This needs to be here so it doesn't get incremented in case the first TrackAlloc throws an exception
memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(size));
bool ok = memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(size));
if (!ok) return nullptr;
}
auto *ptr = old_hooks->alloc(extent_hooks, new_addr, size, alignment, zero, commit, arena_ind);
@@ -118,10 +120,14 @@ static bool my_commit(extent_hooks_t *extent_hooks, void *addr, size_t size, siz
return err;
}
[[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
if (GetQueriesMemoryControl().IsThreadTracked()) [[unlikely]] {
GetQueriesMemoryControl().TrackAllocOnCurrentThread(length);
bool ok = GetQueriesMemoryControl().TrackAllocOnCurrentThread(length);
DMG_ASSERT(ok);
}
memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(length));
auto ok = memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(length));
DMG_ASSERT(ok);
return false;
}

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -28,6 +28,12 @@ void *newImpl(const std::size_t size) {
return ptr;
}
[[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
auto maybe_msg = memgraph::utils::MemoryErrorStatus().msg();
if (maybe_msg) {
throw memgraph::utils::OutOfMemoryException{std::move(*maybe_msg)};
}
throw std::bad_alloc{};
}
@@ -37,11 +43,21 @@ void *newImpl(const std::size_t size, const std::align_val_t align) {
return ptr;
}
[[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
auto maybe_msg = memgraph::utils::MemoryErrorStatus().msg();
if (maybe_msg) {
throw memgraph::utils::OutOfMemoryException{std::move(*maybe_msg)};
}
throw std::bad_alloc{};
}
void *newNoExcept(const std::size_t size) noexcept { return malloc(size); }
void *newNoExcept(const std::size_t size) noexcept {
[[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
return malloc(size);
}
void *newNoExcept(const std::size_t size, const std::align_val_t align) noexcept {
[[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
return aligned_alloc(size, static_cast<std::size_t>(align));
}

View File

@@ -54,14 +54,14 @@ void QueriesMemoryControl::EraseThreadToTransactionId(const std::thread::id &thr
}
}
void QueriesMemoryControl::TrackAllocOnCurrentThread(size_t size) {
bool QueriesMemoryControl::TrackAllocOnCurrentThread(size_t size) {
auto thread_id_to_transaction_id_accessor = thread_id_to_transaction_id.access();
// we might be just constructing mapping between thread id and transaction id
// so we miss this allocation
auto thread_id_to_transaction_id_elem = thread_id_to_transaction_id_accessor.find(std::this_thread::get_id());
if (thread_id_to_transaction_id_elem == thread_id_to_transaction_id_accessor.end()) {
return;
return true;
}
auto transaction_id_to_tracker_accessor = transaction_id_to_tracker.access();
@@ -71,10 +71,10 @@ void QueriesMemoryControl::TrackAllocOnCurrentThread(size_t size) {
// It can happen that some allocation happens between mapping thread to
// transaction id, so we miss this allocation
if (transaction_id_to_tracker == transaction_id_to_tracker_accessor.end()) [[unlikely]] {
return;
return true;
}
auto &query_tracker = transaction_id_to_tracker->tracker;
query_tracker.TrackAlloc(size);
return query_tracker.TrackAlloc(size);
}
void QueriesMemoryControl::TrackFreeOnCurrentThread(size_t size) {

View File

@@ -62,7 +62,7 @@ class QueriesMemoryControl {
// Find tracker for current thread if exists, track
// query allocation and procedure allocation if
// necessary
void TrackAllocOnCurrentThread(size_t size);
bool TrackAllocOnCurrentThread(size_t size);
// Find tracker for current thread if exists, track
// query allocation and procedure allocation if

View File

@@ -139,6 +139,11 @@ struct NodeId {
std::string id_space;
};
#if FMT_VERSION > 90000
template <>
class fmt::formatter<NodeId> : public fmt::ostream_formatter {};
#endif
bool operator==(const NodeId &a, const NodeId &b) { return a.id == b.id && a.id_space == b.id_space; }
std::ostream &operator<<(std::ostream &stream, const NodeId &node_id) {

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -274,3 +274,8 @@ inline void RestoreError(ExceptionInfo exc_info) {
}
} // namespace memgraph::py
#if FMT_VERSION > 90000
template <>
class fmt::formatter<memgraph::py::ExceptionInfo> : public fmt::ostream_formatter {};
#endif

View File

@@ -19,6 +19,7 @@
#include "query/db_accessor.hpp"
#include "query/exceptions.hpp"
#include "query/fmt.hpp"
#include "query/frontend/ast/ast.hpp"
#include "query/frontend/semantic/symbol.hpp"
#include "query/typed_value.hpp"

View File

@@ -24,6 +24,7 @@ struct InterpreterConfig {
std::chrono::seconds instance_down_timeout_sec{5};
std::chrono::seconds instance_health_check_frequency_sec{1};
std::chrono::seconds instance_get_uuid_frequency_sec{10};
std::string default_kafka_bootstrap_servers;
std::string default_pulsar_service_url;

View File

@@ -54,6 +54,10 @@ class EdgeAccessor final {
return impl_.GetProperty(key, view);
}
storage::Result<uint64_t> GetPropertySize(storage::PropertyId key, storage::View view) const {
return impl_.GetPropertySize(key, view);
}
storage::Result<storage::PropertyValue> SetProperty(storage::PropertyId key, const storage::PropertyValue &value) {
return impl_.SetProperty(key, value);
}
@@ -129,6 +133,10 @@ class VertexAccessor final {
return impl_.GetProperty(key, view);
}
storage::Result<uint64_t> GetPropertySize(storage::PropertyId key, storage::View view) const {
return impl_.GetPropertySize(key, view);
}
storage::Result<storage::PropertyValue> SetProperty(storage::PropertyId key, const storage::PropertyValue &value) {
return impl_.SetProperty(key, value);
}
@@ -268,6 +276,10 @@ class SubgraphVertexAccessor final {
return impl_.GetProperty(view, key);
}
storage::Result<uint64_t> GetPropertySize(storage::PropertyId key, storage::View view) const {
return impl_.GetPropertySize(key, view);
}
storage::Gid Gid() const noexcept { return impl_.Gid(); }
storage::Result<size_t> InDegree(storage::View view) const { return impl_.InDegree(view); }
@@ -529,6 +541,10 @@ class DbAccessor final {
storage::PropertyId NameToProperty(const std::string_view name) { return accessor_->NameToProperty(name); }
std::optional<storage::PropertyId> NameToPropertyIfExists(std::string_view name) const {
return accessor_->NameToPropertyIfExists(name);
}
storage::LabelId NameToLabel(const std::string_view name) { return accessor_->NameToLabel(name); }
storage::EdgeTypeId NameToEdgeType(const std::string_view name) { return accessor_->NameToEdgeType(name); }

23
src/query/fmt.hpp Normal file
View File

@@ -0,0 +1,23 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include "query/typed_value.hpp"
template <>
class fmt::formatter<memgraph::query::TypedValue> : public fmt::ostream_formatter {};
template <>
class fmt::formatter<memgraph::query::TypedValue::Type> : public fmt::ostream_formatter {};
#endif

View File

@@ -442,6 +442,29 @@ TypedValue Size(const TypedValue *args, int64_t nargs, const FunctionContext &ct
}
}
TypedValue PropertySize(const TypedValue *args, int64_t nargs, const FunctionContext &ctx) {
FType<Or<Null, Vertex, Edge>, Or<String>>("propertySize", args, nargs);
auto *dba = ctx.db_accessor;
const auto &property_name = args[1].ValueString();
const auto maybe_property_id = dba->NameToPropertyIfExists(property_name);
if (!maybe_property_id) {
return TypedValue(0, ctx.memory);
}
uint64_t property_size = 0;
const auto &graph_entity = args[0];
if (graph_entity.IsVertex()) {
property_size = graph_entity.ValueVertex().GetPropertySize(*maybe_property_id, ctx.view).GetValue();
} else if (graph_entity.IsEdge()) {
property_size = graph_entity.ValueEdge().GetPropertySize(*maybe_property_id, ctx.view).GetValue();
}
return TypedValue(static_cast<int64_t>(property_size), ctx.memory);
}
TypedValue StartNode(const TypedValue *args, int64_t nargs, const FunctionContext &ctx) {
FType<Or<Null, Edge>>("startNode", args, nargs);
if (args[0].IsNull()) return TypedValue(ctx.memory);
@@ -1325,6 +1348,7 @@ std::function<TypedValue(const TypedValue *, int64_t, const FunctionContext &ctx
if (function_name == "PROPERTIES") return Properties;
if (function_name == "RANDOMUUID") return RandomUuid;
if (function_name == "SIZE") return Size;
if (function_name == "PROPERTYSIZE") return PropertySize;
if (function_name == "STARTNODE") return StartNode;
if (function_name == "TIMESTAMP") return Timestamp;
if (function_name == "TOBOOLEAN") return ToBoolean;

View File

@@ -93,6 +93,7 @@
#include "utils/exceptions.hpp"
#include "utils/file.hpp"
#include "utils/flag_validation.hpp"
#include "utils/functional.hpp"
#include "utils/likely.hpp"
#include "utils/logging.hpp"
#include "utils/memory.hpp"
@@ -108,7 +109,6 @@
#include "utils/variant_helpers.hpp"
#ifdef MG_ENTERPRISE
#include "coordination/constants.hpp"
#include "flags/experimental.hpp"
#endif
@@ -328,7 +328,7 @@ class ReplQueryHandler {
.port = static_cast<uint16_t>(*port),
};
if (!handler_->SetReplicationRoleReplica(config, std::nullopt)) {
if (!handler_->TrySetReplicationRoleReplica(config, std::nullopt)) {
throw QueryRuntimeException("Couldn't set role to replica!");
}
}
@@ -369,7 +369,7 @@ class ReplQueryHandler {
.replica_check_frequency = replica_check_frequency,
.ssl = std::nullopt};
const auto error = handler_->TryRegisterReplica(replication_config, true).HasError();
const auto error = handler_->TryRegisterReplica(replication_config).HasError();
if (error) {
throw QueryRuntimeException(fmt::format("Couldn't register replica '{}'!", name));
@@ -485,8 +485,9 @@ class CoordQueryHandler final : public query::CoordinatorQueryHandler {
void RegisterReplicationInstance(std::string const &coordinator_socket_address,
std::string const &replication_socket_address,
std::chrono::seconds const &instance_check_frequency,
std::chrono::seconds const &instance_down_timeout, std::string const &instance_name,
CoordinatorQuery::SyncMode sync_mode) override {
std::chrono::seconds const &instance_down_timeout,
std::chrono::seconds const &instance_get_uuid_frequency,
std::string const &instance_name, CoordinatorQuery::SyncMode sync_mode) override {
const auto maybe_replication_ip_port =
io::network::Endpoint::ParseSocketOrAddress(replication_socket_address, std::nullopt);
if (!maybe_replication_ip_port) {
@@ -513,6 +514,7 @@ class CoordQueryHandler final : public query::CoordinatorQueryHandler {
.port = coordinator_server_port,
.instance_health_check_frequency_sec = instance_check_frequency,
.instance_down_timeout_sec = instance_down_timeout,
.instance_get_uuid_frequency_sec = instance_get_uuid_frequency,
.replication_client_info = repl_config,
.ssl = std::nullopt};
@@ -560,6 +562,8 @@ class CoordQueryHandler final : public query::CoordinatorQueryHandler {
using enum memgraph::coordination::SetInstanceToMainCoordinatorStatus;
case NO_INSTANCE_WITH_NAME:
throw QueryRuntimeException("No instance with such name!");
case MAIN_ALREADY_EXISTS:
throw QueryRuntimeException("Couldn't set instance to main since there is already a main instance in cluster!");
case NOT_COORDINATOR:
throw QueryRuntimeException("SET INSTANCE TO MAIN query can only be run on a coordinator!");
case COULD_NOT_PROMOTE_TO_MAIN:
@@ -1126,17 +1130,21 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Parameters &parameters,
coordination::CoordinatorState *coordinator_state,
const query::InterpreterConfig &config, std::vector<Notification> *notifications) {
using enum memgraph::flags::Experiments;
if (!license::global_license_checker.IsEnterpriseValidFast()) {
throw QueryRuntimeException("High availability is only available in Memgraph Enterprise.");
}
if (!flags::AreExperimentsEnabled(HIGH_AVAILABILITY)) {
throw QueryRuntimeException(
"High availability is experimental feature. If you want to use it, add high-availability option to the "
"--experimental-enabled flag.");
}
Callback callback;
switch (coordinator_query->action_) {
case CoordinatorQuery::Action::ADD_COORDINATOR_INSTANCE: {
if (!license::global_license_checker.IsEnterpriseValidFast()) {
throw QueryException("Trying to use enterprise feature without a valid license.");
}
if constexpr (!coordination::allow_ha) {
throw QueryRuntimeException(
"High availability is experimental feature. Please set MG_EXPERIMENTAL_HIGH_AVAILABILITY compile flag to "
"be able to use this functionality.");
}
if (!FLAGS_raft_server_id) {
throw QueryRuntimeException("Only coordinator can add coordinator instance!");
}
@@ -1160,15 +1168,6 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
return callback;
}
case CoordinatorQuery::Action::REGISTER_INSTANCE: {
if (!license::global_license_checker.IsEnterpriseValidFast()) {
throw QueryException("Trying to use enterprise feature without a valid license.");
}
if constexpr (!coordination::allow_ha) {
throw QueryRuntimeException(
"High availability is experimental feature. Please set MG_EXPERIMENTAL_HIGH_AVAILABILITY compile flag to "
"be able to use this functionality.");
}
if (!FLAGS_raft_server_id) {
throw QueryRuntimeException("Only coordinator can register coordinator server!");
}
@@ -1184,11 +1183,12 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
instance_health_check_frequency_sec = config.instance_health_check_frequency_sec,
instance_name = coordinator_query->instance_name_,
instance_down_timeout_sec = config.instance_down_timeout_sec,
instance_get_uuid_frequency_sec = config.instance_get_uuid_frequency_sec,
sync_mode = coordinator_query->sync_mode_]() mutable {
handler.RegisterReplicationInstance(std::string(coordinator_socket_address_tv.ValueString()),
std::string(replication_socket_address_tv.ValueString()),
instance_health_check_frequency_sec, instance_down_timeout_sec,
instance_name, sync_mode);
instance_get_uuid_frequency_sec, instance_name, sync_mode);
return std::vector<std::vector<TypedValue>>();
};
@@ -1199,15 +1199,6 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
return callback;
}
case CoordinatorQuery::Action::UNREGISTER_INSTANCE:
if (!license::global_license_checker.IsEnterpriseValidFast()) {
throw QueryException("Trying to use enterprise feature without a valid license.");
}
if constexpr (!coordination::allow_ha) {
throw QueryRuntimeException(
"High availability is experimental feature. Please set MG_EXPERIMENTAL_HIGH_AVAILABILITY compile flag to "
"be able to use this functionality.");
}
if (!FLAGS_raft_server_id) {
throw QueryRuntimeException("Only coordinator can register coordinator server!");
}
@@ -1223,14 +1214,6 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
return callback;
case CoordinatorQuery::Action::SET_INSTANCE_TO_MAIN: {
if (!license::global_license_checker.IsEnterpriseValidFast()) {
throw QueryException("Trying to use enterprise feature without a valid license.");
}
if constexpr (!coordination::allow_ha) {
throw QueryRuntimeException(
"High availability is experimental feature. Please set MG_EXPERIMENTAL_HIGH_AVAILABILITY compile flag to "
"be able to use this functionality.");
}
if (!FLAGS_raft_server_id) {
throw QueryRuntimeException("Only coordinator can register coordinator server!");
}
@@ -1248,14 +1231,6 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
return callback;
}
case CoordinatorQuery::Action::SHOW_INSTANCES: {
if (!license::global_license_checker.IsEnterpriseValidFast()) {
throw QueryException("Trying to use enterprise feature without a valid license.");
}
if constexpr (!coordination::allow_ha) {
throw QueryRuntimeException(
"High availability is experimental feature. Please set MG_EXPERIMENTAL_HIGH_AVAILABILITY compile flag to "
"be able to use this functionality.");
}
if (!FLAGS_raft_server_id) {
throw QueryRuntimeException("Only coordinator can run SHOW INSTANCES.");
}
@@ -1264,17 +1239,13 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
callback.fn = [handler = CoordQueryHandler{*coordinator_state},
replica_nfields = callback.header.size()]() mutable {
auto const instances = handler.ShowInstances();
std::vector<std::vector<TypedValue>> result{};
result.reserve(result.size());
auto const converter = [](const auto &status) -> std::vector<TypedValue> {
return {TypedValue{status.instance_name}, TypedValue{status.raft_socket_address},
TypedValue{status.coord_socket_address}, TypedValue{status.is_alive},
TypedValue{status.cluster_role}};
};
std::ranges::transform(instances, std::back_inserter(result),
[](const auto &status) -> std::vector<TypedValue> {
return {TypedValue{status.instance_name}, TypedValue{status.raft_socket_address},
TypedValue{status.coord_socket_address}, TypedValue{status.is_alive},
TypedValue{status.cluster_role}};
});
return result;
return utils::fmap(converter, instances);
};
return callback;
}
@@ -4404,9 +4375,19 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
UpdateTypeCount(rw_type);
if (interpreter_context_->repl_state->IsReplica() && IsQueryWrite(rw_type)) {
query_execution = nullptr;
throw QueryException("Write query forbidden on the replica!");
bool const write_query = IsQueryWrite(rw_type);
if (write_query) {
if (interpreter_context_->repl_state->IsReplica()) {
query_execution = nullptr;
throw QueryException("Write query forbidden on the replica!");
}
#ifdef MG_ENTERPRISE
if (FLAGS_coordinator_server_port && !interpreter_context_->repl_state->IsMainWriteable()) {
query_execution = nullptr;
throw QueryException(
"Write query forbidden on the main! Coordinator needs to enable writing on main by sending RPC message.");
}
#endif
}
// Set the target db to the current db (some queries have different target from the current db)

View File

@@ -109,6 +109,7 @@ class CoordinatorQueryHandler {
std::string const &replication_socket_address,
std::chrono::seconds const &instance_health_check_frequency,
std::chrono::seconds const &instance_down_timeout,
std::chrono::seconds const &instance_get_uuid_frequency,
std::string const &instance_name, CoordinatorQuery::SyncMode sync_mode) = 0;
/// @throw QueryRuntimeException if an error ocurred.

View File

@@ -313,7 +313,7 @@ void Filters::CollectPatternFilters(Pattern &pattern, SymbolTable &symbol_table,
auto *property_lookup = storage.Create<PropertyLookup>(atom->filter_lambda_.inner_edge, prop_pair.first);
auto *prop_equal = storage.Create<EqualOperator>(property_lookup, prop_pair.second);
// Currently, variable expand has no gains if we set PropertyFilter.
all_filters_.emplace_back(FilterInfo{FilterInfo::Type::Generic, prop_equal, collector.symbols_});
all_filters_.emplace_back(FilterInfo::Type::Generic, prop_equal, collector.symbols_);
}
{
collector.symbols_.clear();
@@ -328,9 +328,9 @@ void Filters::CollectPatternFilters(Pattern &pattern, SymbolTable &symbol_table,
auto *prop_equal = storage.Create<EqualOperator>(property_lookup, prop_pair.second);
// Currently, variable expand has no gains if we set PropertyFilter.
all_filters_.emplace_back(
FilterInfo{FilterInfo::Type::Generic,
storage.Create<All>(identifier, atom->identifier_, storage.Create<Where>(prop_equal)),
collector.symbols_});
FilterInfo::Type::Generic,
storage.Create<All>(identifier, atom->identifier_, storage.Create<Where>(prop_equal)),
collector.symbols_);
}
}
return;
@@ -639,6 +639,12 @@ void AddMatching(const Match &match, SymbolTable &symbol_table, AstStorage &stor
}
}
PatternFilterVisitor::PatternFilterVisitor(SymbolTable &symbol_table, AstStorage &storage)
: symbol_table_(symbol_table), storage_(storage) {}
PatternFilterVisitor::PatternFilterVisitor(const PatternFilterVisitor &) = default;
PatternFilterVisitor::PatternFilterVisitor(PatternFilterVisitor &&) noexcept = default;
PatternFilterVisitor::~PatternFilterVisitor() = default;
void PatternFilterVisitor::Visit(Exists &op) {
std::vector<Pattern *> patterns;
patterns.push_back(op.pattern_);
@@ -652,6 +658,8 @@ void PatternFilterVisitor::Visit(Exists &op) {
matchings_.push_back(std::move(filter_matching));
}
std::vector<FilterMatching> PatternFilterVisitor::getMatchings() { return matchings_; }
static void ParseForeach(query::Foreach &foreach, SingleQueryPart &query_part, AstStorage &storage,
SymbolTable &symbol_table) {
for (auto *clause : foreach.clauses_) {
@@ -723,4 +731,18 @@ QueryParts CollectQueryParts(SymbolTable &symbol_table, AstStorage &storage, Cyp
return QueryParts{query_parts, distinct};
}
FilterInfo::FilterInfo(Type type, Expression *expression, std::unordered_set<Symbol> used_symbols,
std::optional<PropertyFilter> property_filter, std::optional<IdFilter> id_filter)
: type(type),
expression(expression),
used_symbols(std::move(used_symbols)),
property_filter(std::move(property_filter)),
id_filter(std::move(id_filter)),
matchings({}) {}
FilterInfo::FilterInfo(const FilterInfo &) = default;
FilterInfo &FilterInfo::operator=(const FilterInfo &) = default;
FilterInfo::FilterInfo(FilterInfo &&) noexcept = default;
FilterInfo &FilterInfo::operator=(FilterInfo &&) noexcept = default;
FilterInfo::~FilterInfo() = default;
} // namespace memgraph::query::plan

View File

@@ -19,6 +19,7 @@
#include <vector>
#include "query/frontend/ast/ast.hpp"
#include "query/frontend/ast/ast_visitor.hpp"
#include "query/frontend/semantic/symbol_table.hpp"
namespace memgraph::query::plan {
@@ -159,8 +160,12 @@ enum class PatternFilterType { EXISTS };
/// Collects matchings from filters that include patterns
class PatternFilterVisitor : public ExpressionVisitor<void> {
public:
explicit PatternFilterVisitor(SymbolTable &symbol_table, AstStorage &storage)
: symbol_table_(symbol_table), storage_(storage) {}
explicit PatternFilterVisitor(SymbolTable &symbol_table, AstStorage &storage);
PatternFilterVisitor(const PatternFilterVisitor &);
PatternFilterVisitor &operator=(const PatternFilterVisitor &) = delete;
PatternFilterVisitor(PatternFilterVisitor &&) noexcept;
PatternFilterVisitor &operator=(PatternFilterVisitor &&) noexcept = delete;
~PatternFilterVisitor() override;
using ExpressionVisitor<void>::Visit;
@@ -232,7 +237,7 @@ class PatternFilterVisitor : public ExpressionVisitor<void> {
void Visit(RegexMatch &op) override{};
void Visit(PatternComprehension &op) override{};
std::vector<FilterMatching> getMatchings() { return matchings_; }
std::vector<FilterMatching> getMatchings();
SymbolTable &symbol_table_;
AstStorage &storage_;
@@ -298,9 +303,23 @@ struct FilterInfo {
/// elements.
enum class Type { Generic, Label, Property, Id, Pattern };
Type type;
// FilterInfo is tricky because FilterMatching is not yet defined:
// * if no declared constructor -> FilterInfo is std::__is_complete_or_unbounded
// * if any user-declared constructor -> non-aggregate type -> no designated initializers are possible
// * IMPORTANT: Matchings will always be initialized to an empty container.
explicit FilterInfo(Type type = Type::Generic, Expression *expression = nullptr,
std::unordered_set<Symbol> used_symbols = {}, std::optional<PropertyFilter> property_filter = {},
std::optional<IdFilter> id_filter = {});
// All other constructors are also defined in the cpp file because this struct is incomplete here.
FilterInfo(const FilterInfo &);
FilterInfo &operator=(const FilterInfo &);
FilterInfo(FilterInfo &&) noexcept;
FilterInfo &operator=(FilterInfo &&) noexcept;
~FilterInfo();
Type type{Type::Generic};
/// The original filter expression which must be satisfied.
Expression *expression;
Expression *expression{nullptr};
/// Set of used symbols by the filter @c expression.
std::unordered_set<Symbol> used_symbols{};
/// Labels for Type::Label filtering.
@@ -310,7 +329,8 @@ struct FilterInfo {
/// Information for Type::Id filtering.
std::optional<IdFilter> id_filter{};
/// Matchings for filters that include patterns
std::vector<FilterMatching> matchings{};
/// NOTE: The vector is not defined here because FilterMatching is forward declared above.
std::vector<FilterMatching> matchings;
};
/// Stores information on filters used inside the @c Matching of a @c QueryPart.
@@ -329,34 +349,15 @@ class Filters final {
auto empty() const { return all_filters_.empty(); }
auto erase(iterator pos) { return all_filters_.erase(pos); }
auto erase(const_iterator pos) { return all_filters_.erase(pos); }
auto erase(iterator first, iterator last) { return all_filters_.erase(first, last); }
auto erase(const_iterator first, const_iterator last) { return all_filters_.erase(first, last); }
auto erase(iterator pos) -> iterator;
auto erase(const_iterator pos) -> iterator;
auto erase(iterator first, iterator last) -> iterator;
auto erase(const_iterator first, const_iterator last) -> iterator;
void SetFilters(std::vector<FilterInfo> &&all_filters) { all_filters_ = std::move(all_filters); }
auto FilteredLabels(const Symbol &symbol) const {
std::unordered_set<LabelIx> labels;
for (const auto &filter : all_filters_) {
if (filter.type == FilterInfo::Type::Label && utils::Contains(filter.used_symbols, symbol)) {
MG_ASSERT(filter.used_symbols.size() == 1U, "Expected a single used symbol for label filter");
labels.insert(filter.labels.begin(), filter.labels.end());
}
}
return labels;
}
auto FilteredProperties(const Symbol &symbol) const -> std::unordered_set<PropertyIx> {
std::unordered_set<PropertyIx> properties;
for (const auto &filter : all_filters_) {
if (filter.type == FilterInfo::Type::Property && filter.property_filter->symbol_ == symbol) {
properties.insert(filter.property_filter->property_);
}
}
return properties;
}
auto FilteredLabels(const Symbol &symbol) const -> std::unordered_set<LabelIx>;
auto FilteredProperties(const Symbol &symbol) const -> std::unordered_set<PropertyIx>;
/// Remove a filter; may invalidate iterators.
/// Removal is done by comparing only the expression, so that multiple
@@ -370,26 +371,10 @@ class Filters final {
std::vector<Expression *> *removed_filters = nullptr);
/// Returns a vector of FilterInfo for properties.
auto PropertyFilters(const Symbol &symbol) const {
std::vector<FilterInfo> filters;
for (const auto &filter : all_filters_) {
if (filter.type == FilterInfo::Type::Property && filter.property_filter->symbol_ == symbol) {
filters.push_back(filter);
}
}
return filters;
}
auto PropertyFilters(const Symbol &symbol) const -> std::vector<FilterInfo>;
/// Return a vector of FilterInfo for ID equality filtering.
auto IdFilters(const Symbol &symbol) const {
std::vector<FilterInfo> filters;
for (const auto &filter : all_filters_) {
if (filter.type == FilterInfo::Type::Id && filter.id_filter->symbol_ == symbol) {
filters.push_back(filter);
}
}
return filters;
}
auto IdFilters(const Symbol &symbol) const -> std::vector<FilterInfo>;
/// Collects filtering information from a pattern.
///
@@ -459,6 +444,57 @@ struct FilterMatching : Matching {
std::optional<Symbol> symbol;
};
inline auto Filters::erase(Filters::iterator pos) -> iterator { return all_filters_.erase(pos); }
inline auto Filters::erase(Filters::const_iterator pos) -> iterator { return all_filters_.erase(pos); }
inline auto Filters::erase(Filters::iterator first, Filters::iterator last) -> iterator {
return all_filters_.erase(first, last);
}
inline auto Filters::erase(Filters::const_iterator first, Filters::const_iterator last) -> iterator {
return all_filters_.erase(first, last);
}
inline auto Filters::FilteredLabels(const Symbol &symbol) const -> std::unordered_set<LabelIx> {
std::unordered_set<LabelIx> labels;
for (const auto &filter : all_filters_) {
if (filter.type == FilterInfo::Type::Label && utils::Contains(filter.used_symbols, symbol)) {
MG_ASSERT(filter.used_symbols.size() == 1U, "Expected a single used symbol for label filter");
labels.insert(filter.labels.begin(), filter.labels.end());
}
}
return labels;
}
inline auto Filters::FilteredProperties(const Symbol &symbol) const -> std::unordered_set<PropertyIx> {
std::unordered_set<PropertyIx> properties;
for (const auto &filter : all_filters_) {
if (filter.type == FilterInfo::Type::Property && filter.property_filter->symbol_ == symbol) {
properties.insert(filter.property_filter->property_);
}
}
return properties;
}
inline auto Filters::PropertyFilters(const Symbol &symbol) const -> std::vector<FilterInfo> {
std::vector<FilterInfo> filters;
for (const auto &filter : all_filters_) {
if (filter.type == FilterInfo::Type::Property && filter.property_filter->symbol_ == symbol) {
filters.push_back(filter);
}
}
return filters;
}
inline auto Filters::IdFilters(const Symbol &symbol) const -> std::vector<FilterInfo> {
std::vector<FilterInfo> filters;
for (const auto &filter : all_filters_) {
if (filter.type == FilterInfo::Type::Id && filter.id_filter->symbol_ == symbol) {
filters.push_back(filter);
}
}
return filters;
}
/// @brief Represents a read (+ write) part of a query. Parts are split on
/// `WITH` clauses.
///

View File

@@ -0,0 +1,82 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include <string>
#include "mg_procedure.h"
#include "utils/logging.hpp"
inline std::string ToString(const mgp_log_level &log_level) {
switch (log_level) {
case mgp_log_level::MGP_LOG_LEVEL_CRITICAL:
return "CRITICAL";
case mgp_log_level::MGP_LOG_LEVEL_ERROR:
return "ERROR";
case mgp_log_level::MGP_LOG_LEVEL_WARN:
return "WARN";
case mgp_log_level::MGP_LOG_LEVEL_INFO:
return "INFO";
case mgp_log_level::MGP_LOG_LEVEL_DEBUG:
return "DEBUG";
case mgp_log_level::MGP_LOG_LEVEL_TRACE:
return "TRACE";
}
LOG_FATAL("ToString of a wrong mgp_log_level -> check missing switch case");
}
inline std::ostream &operator<<(std::ostream &os, const mgp_log_level &log_level) {
os << ToString(log_level);
return os;
}
template <>
class fmt::formatter<mgp_log_level> : public fmt::ostream_formatter {};
inline std::string ToString(const mgp_error &error) {
switch (error) {
case mgp_error::MGP_ERROR_NO_ERROR:
return "NO ERROR";
case mgp_error::MGP_ERROR_UNKNOWN_ERROR:
return "UNKNOWN ERROR";
case mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE:
return "UNABLE TO ALLOCATE ERROR";
case mgp_error::MGP_ERROR_INSUFFICIENT_BUFFER:
return "INSUFFICIENT BUFFER ERROR";
case mgp_error::MGP_ERROR_OUT_OF_RANGE:
return "OUT OF RANGE ERROR";
case mgp_error::MGP_ERROR_LOGIC_ERROR:
return "LOGIC ERROR";
case mgp_error::MGP_ERROR_DELETED_OBJECT:
return "DELETED OBJECT ERROR";
case mgp_error::MGP_ERROR_INVALID_ARGUMENT:
return "INVALID ARGUMENT ERROR";
case mgp_error::MGP_ERROR_KEY_ALREADY_EXISTS:
return "KEY ALREADY EXISTS ERROR";
case mgp_error::MGP_ERROR_IMMUTABLE_OBJECT:
return "IMMUTABLE OBJECT ERROR";
case mgp_error::MGP_ERROR_VALUE_CONVERSION:
return "VALUE CONVERSION ERROR";
case mgp_error::MGP_ERROR_SERIALIZATION_ERROR:
return "SERIALIZATION ERROR";
case mgp_error::MGP_ERROR_AUTHORIZATION_ERROR:
return "AUTHORIZATION ERROR";
}
LOG_FATAL("ToString of a wrong mgp_error -> check missing switch case");
}
inline std::ostream &operator<<(std::ostream &os, const mgp_error &error) {
os << ToString(error);
return os;
}
template <>
class fmt::formatter<mgp_error> : public fmt::ostream_formatter {};
#endif

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -10,6 +10,7 @@
// licenses/APL.txt.
#include "query/procedure/mg_procedure_helpers.hpp"
#include "query/procedure/fmt.hpp"
namespace memgraph::query::procedure {
MgpUniquePtr<mgp_value> GetStringValueOrSetError(const char *string, mgp_memory *memory, mgp_result *result) {

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -18,6 +18,7 @@
#include <fmt/format.h>
#include "mg_procedure.h"
#include "query/procedure/fmt.hpp"
namespace memgraph::query::procedure {
template <typename TResult, typename TFunc, typename... TArgs>

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -29,6 +29,7 @@
#include "query/db_accessor.hpp"
#include "query/frontend/ast/ast.hpp"
#include "query/procedure/cypher_types.hpp"
#include "query/procedure/fmt.hpp"
#include "query/procedure/mg_procedure_helpers.hpp"
#include "query/stream/common.hpp"
#include "storage/v2/property_value.hpp"
@@ -187,6 +188,7 @@ template <typename TFunc, typename... Args>
spdlog::error("Memory allocation error during mg API call: {}", bae.what());
return mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE;
} catch (const memgraph::utils::OutOfMemoryException &oome) {
[[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
spdlog::error("Memory limit exceeded during mg API call: {}", oome.what());
return mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE;
} catch (const std::out_of_range &oore) {
@@ -198,12 +200,12 @@ template <typename TFunc, typename... Args>
} catch (const std::logic_error &lee) {
spdlog::error("Logic error during mg API call: {}", lee.what());
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::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::MGP_ERROR_INVALID_ARGUMENT;
} catch (const std::exception &e) {
spdlog::error("Unexpected error during mg API call: {}", e.what());
return mgp_error::MGP_ERROR_UNKNOWN_ERROR;
} catch (...) {
spdlog::error("Unexpected error during mg API call");
return mgp_error::MGP_ERROR_UNKNOWN_ERROR;

View File

@@ -49,11 +49,14 @@ struct ReplicationQueryHandler {
virtual bool SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
const std::optional<utils::UUID> &main_uuid) = 0;
virtual bool TrySetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
const std::optional<utils::UUID> &main_uuid) = 0;
// as MAIN, define and connect to REPLICAs
virtual auto TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config, bool send_swap_uuid)
virtual auto TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config)
-> utils::BasicResult<RegisterReplicaError> = 0;
virtual auto RegisterReplica(const memgraph::replication::ReplicationClientConfig &config, bool send_swap_uuid)
virtual auto RegisterReplica(const memgraph::replication::ReplicationClientConfig &config)
-> utils::BasicResult<RegisterReplicaError> = 0;
// as MAIN, remove a REPLICA connection

View File

@@ -19,6 +19,7 @@
#include <string_view>
#include <utility>
#include "query/fmt.hpp"
#include "storage/v2/temporal.hpp"
#include "utils/exceptions.hpp"
#include "utils/fnv.hpp"
@@ -326,13 +327,11 @@ TypedValue::operator storage::PropertyValue() const {
throw TypedValueException("TypedValue is of type '{}', not '{}'", type_, Type::type_enum); \
return field; \
} \
\
const type_param &TypedValue::Value##type_enum() const { \
if (type_ != Type::type_enum) [[unlikely]] \
throw TypedValueException("TypedValue is of type '{}', not '{}'", type_, Type::type_enum); \
return field; \
} \
\
bool TypedValue::Is##type_enum() const { return type_ == Type::type_enum; }
DEFINE_VALUE_AND_TYPE_GETTERS(bool, Bool, bool_v)
@@ -783,10 +782,13 @@ TypedValue operator<(const TypedValue &a, const TypedValue &b) {
return false;
}
};
if (!is_legal(a.type()) || !is_legal(b.type()))
if (!is_legal(a.type()) || !is_legal(b.type())) {
throw TypedValueException("Invalid 'less' operand types({} + {})", a.type(), b.type());
}
if (a.IsNull() || b.IsNull()) return TypedValue(a.GetMemoryResource());
if (a.IsNull() || b.IsNull()) {
return TypedValue(a.GetMemoryResource());
}
if (a.IsString() || b.IsString()) {
if (a.type() != b.type()) {
@@ -956,8 +958,9 @@ inline void EnsureArithmeticallyOk(const TypedValue &a, const TypedValue &b, boo
// checked here because they are handled before this check is performed in
// arithmetic op implementations.
if (!is_legal(a) || !is_legal(b))
if (!is_legal(a) || !is_legal(b)) {
throw TypedValueException("Invalid {} operand types {}, {}", op_name, a.type(), b.type());
}
}
namespace {
@@ -1107,8 +1110,9 @@ TypedValue operator%(const TypedValue &a, const TypedValue &b) {
}
inline void EnsureLogicallyOk(const TypedValue &a, const TypedValue &b, const std::string &op_name) {
if (!((a.IsBool() || a.IsNull()) && (b.IsBool() || b.IsNull())))
if (!((a.IsBool() || a.IsNull()) && (b.IsBool() || b.IsNull()))) {
throw TypedValueException("Invalid {} operand types({} && {})", op_name, a.type(), b.type());
}
}
TypedValue operator&&(const TypedValue &a, const TypedValue &b) {

View File

@@ -39,7 +39,8 @@ enum class RegisterReplicaError : uint8_t { NAME_EXISTS, ENDPOINT_EXISTS, COULD_
struct RoleMainData {
RoleMainData() = default;
explicit RoleMainData(ReplicationEpoch e, std::optional<utils::UUID> uuid = std::nullopt) : epoch_(std::move(e)) {
explicit RoleMainData(ReplicationEpoch e, bool writing_enabled, std::optional<utils::UUID> uuid = std::nullopt)
: epoch_(std::move(e)), writing_enabled_(writing_enabled) {
if (uuid) {
uuid_ = *uuid;
}
@@ -54,6 +55,7 @@ struct RoleMainData {
ReplicationEpoch epoch_;
std::list<ReplicationClient> registered_replicas_{}; // TODO: data race issues
utils::UUID uuid_;
bool writing_enabled_{false};
};
struct RoleReplicaData {
@@ -90,6 +92,21 @@ struct ReplicationState {
bool IsMain() const { return GetRole() == replication_coordination_glue::ReplicationRole::MAIN; }
bool IsReplica() const { return GetRole() == replication_coordination_glue::ReplicationRole::REPLICA; }
auto IsMainWriteable() const -> bool {
if (auto const *main = std::get_if<RoleMainData>(&replication_data_)) {
return main->writing_enabled_;
}
return false;
}
auto EnableWritingOnMain() -> bool {
if (auto *main = std::get_if<RoleMainData>(&replication_data_)) {
main->writing_enabled_ = true;
return true;
}
return false;
}
bool HasDurability() const { return nullptr != durability_; }
bool TryPersistRoleMain(std::string new_epoch, utils::UUID main_uuid);

View File

@@ -10,6 +10,7 @@
// licenses/APL.txt.
#include "replication/replication_client.hpp"
#include "io/network/fmt.hpp"
namespace memgraph::replication {
@@ -30,7 +31,7 @@ ReplicationClient::ReplicationClient(const memgraph::replication::ReplicationCli
ReplicationClient::~ReplicationClient() {
try {
auto const &endpoint = rpc_client_.Endpoint();
spdlog::trace("Closing replication client on {}:{}", endpoint.address, endpoint.port);
spdlog::trace("Closing replication client on {}", endpoint);
} catch (...) {
// Logging can throw. Not a big deal, just ignore.
}

View File

@@ -62,8 +62,9 @@ ReplicationState::ReplicationState(std::optional<std::filesystem::path> durabili
}
#endif
if (std::holds_alternative<RoleReplicaData>(replication_data)) {
spdlog::trace("Recovered main's uuid for replica {}",
std::string(std::get<RoleReplicaData>(replication_data).uuid_.value()));
auto &replica_uuid = std::get<RoleReplicaData>(replication_data).uuid_;
std::string uuid = replica_uuid.has_value() ? std::string(replica_uuid.value()) : "";
spdlog::trace("Recovered main's uuid for replica {}", uuid);
} else {
spdlog::trace("Recovered uuid for main {}", std::string(std::get<RoleMainData>(replication_data).uuid_));
}
@@ -144,8 +145,8 @@ auto ReplicationState::FetchReplicationData() -> FetchReplicationResult_t {
return std::visit(
utils::Overloaded{
[&](durability::MainRole &&r) -> FetchReplicationResult_t {
auto res =
RoleMainData{std::move(r.epoch), r.main_uuid.has_value() ? r.main_uuid.value() : utils::UUID{}};
auto res = RoleMainData{std::move(r.epoch), false,
r.main_uuid.has_value() ? r.main_uuid.value() : utils::UUID{}};
auto b = durability_->begin(durability::kReplicationReplicaPrefix);
auto e = durability_->end(durability::kReplicationReplicaPrefix);
for (; b != e; ++b) {
@@ -253,7 +254,7 @@ bool ReplicationState::SetReplicationRoleMain(const utils::UUID &main_uuid) {
return false;
}
replication_data_ = RoleMainData{ReplicationEpoch{new_epoch}, main_uuid};
replication_data_ = RoleMainData{ReplicationEpoch{new_epoch}, true, main_uuid};
return true;
}

View File

@@ -14,6 +14,7 @@
#include "dbms/dbms_handler.hpp"
#include "flags/experimental.hpp"
#include "replication/include/replication/state.hpp"
#include "replication_handler/system_replication.hpp"
#include "replication_handler/system_rpc.hpp"
#include "utils/result.hpp"
@@ -113,15 +114,19 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
// as REPLICA, become MAIN
bool SetReplicationRoleMain() override;
// as MAIN, become REPLICA
// as MAIN, become REPLICA, can be called on MAIN and REPLICA
bool SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
const std::optional<utils::UUID> &main_uuid) override;
// as MAIN, become REPLICA, can be called only on MAIN
bool TrySetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
const std::optional<utils::UUID> &main_uuid) override;
// as MAIN, define and connect to REPLICAs
auto TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config, bool send_swap_uuid)
auto TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config)
-> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> override;
auto RegisterReplica(const memgraph::replication::ReplicationClientConfig &config, bool send_swap_uuid)
auto RegisterReplica(const memgraph::replication::ReplicationClientConfig &config)
-> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> override;
// as MAIN, remove a REPLICA connection
@@ -137,12 +142,13 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
auto GetReplState() const -> const memgraph::replication::ReplicationState &;
auto GetReplState() -> memgraph::replication::ReplicationState &;
auto GetReplicaUUID() -> std::optional<utils::UUID>;
private:
template <bool AllowReplicaToDivergeFromMain>
auto RegisterReplica_(const memgraph::replication::ReplicationClientConfig &config, bool send_swap_uuid)
template <bool SendSwapUUID>
auto RegisterReplica_(const memgraph::replication::ReplicationClientConfig &config)
-> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> {
MG_ASSERT(repl_state_.IsMain(), "Only main instance can register a replica!");
auto maybe_client = repl_state_.RegisterReplica(config);
if (maybe_client.HasError()) {
switch (maybe_client.GetError()) {
@@ -159,7 +165,6 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
break;
}
}
using enum memgraph::flags::Experiments;
bool system_replication_enabled = flags::AreExperimentsEnabled(SYSTEM_REPLICATION);
if (!system_replication_enabled && dbms_handler_.Count() > 1) {
@@ -167,25 +172,21 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
}
const auto main_uuid =
std::get<memgraph::replication::RoleMainData>(dbms_handler_.ReplicationState().ReplicationData()).uuid_;
if (send_swap_uuid) {
if constexpr (SendSwapUUID) {
if (!memgraph::replication_coordination_glue::SendSwapMainUUIDRpc(maybe_client.GetValue()->rpc_client_,
main_uuid)) {
return memgraph::query::RegisterReplicaError::ERROR_ACCEPTING_MAIN;
}
}
#ifdef MG_ENTERPRISE
// Update system before enabling individual storage <-> replica clients
SystemRestore(*maybe_client.GetValue(), system_, dbms_handler_, main_uuid, auth_);
#endif
const auto dbms_error = HandleRegisterReplicaStatus(maybe_client);
if (dbms_error.has_value()) {
return *dbms_error;
}
auto &instance_client_ptr = maybe_client.GetValue();
bool all_clients_good = true;
// Add database specific clients (NOTE Currently all databases are connected to each replica)
dbms_handler_.ForEach([&](dbms::DatabaseAccess db_acc) {
@@ -195,7 +196,6 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
}
// TODO: ATM only IN_MEMORY_TRANSACTIONAL, fix other modes
if (storage->storage_mode_ != storage::StorageMode::IN_MEMORY_TRANSACTIONAL) return;
all_clients_good &= storage->repl_storage_state_.replication_clients_.WithLock(
[storage, &instance_client_ptr, db_acc = std::move(db_acc),
main_uuid](auto &storage_clients) mutable { // NOLINT
@@ -203,9 +203,9 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
client->Start(storage, std::move(db_acc));
bool const success = std::invoke([state = client->State()]() {
if (state == storage::replication::ReplicaState::DIVERGED_FROM_MAIN) {
return AllowReplicaToDivergeFromMain;
return false;
}
return state != storage::replication::ReplicaState::MAYBE_BEHIND;
return true;
});
if (success) {
@@ -214,14 +214,12 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
return success;
});
});
// NOTE Currently if any databases fails, we revert back
if (!all_clients_good) {
spdlog::error("Failed to register all databases on the REPLICA \"{}\"", config.name);
UnregisterReplica(config.name);
return memgraph::query::RegisterReplicaError::CONNECTION_FAILED;
}
// No client error, start instance level client
#ifdef MG_ENTERPRISE
StartReplicaClient(*instance_client_ptr, system_, dbms_handler_, main_uuid, auth_);
@@ -231,6 +229,57 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
return {};
}
template <bool AllowIdempotency>
bool SetReplicationRoleReplica_(const memgraph::replication::ReplicationServerConfig &config,
const std::optional<utils::UUID> &main_uuid) {
if (repl_state_.IsReplica()) {
if (!AllowIdempotency) {
return false;
}
// We don't want to restart the server if we're already a REPLICA with correct config
auto &replica_data = std::get<memgraph::replication::RoleReplicaData>(repl_state_.ReplicationData());
if (replica_data.config == config) {
return true;
}
repl_state_.SetReplicationRoleReplica(config, main_uuid);
#ifdef MG_ENTERPRISE
return StartRpcServer(dbms_handler_, replica_data, auth_, system_);
#else
return StartRpcServer(dbms_handler_, replica_data);
#endif
}
// TODO StorageState needs to be synched. Could have a dangling reference if someone adds a database as we are
// deleting the replica.
// Remove database specific clients
dbms_handler_.ForEach([&](memgraph::dbms::DatabaseAccess db_acc) {
auto *storage = db_acc->storage();
storage->repl_storage_state_.replication_clients_.WithLock([](auto &clients) { clients.clear(); });
});
// Remove instance level clients
std::get<memgraph::replication::RoleMainData>(repl_state_.ReplicationData()).registered_replicas_.clear();
// Creates the server
repl_state_.SetReplicationRoleReplica(config, main_uuid);
// Start
const auto success =
std::visit(memgraph::utils::Overloaded{[](memgraph::replication::RoleMainData &) {
// ASSERT
return false;
},
[this](memgraph::replication::RoleReplicaData &data) {
#ifdef MG_ENTERPRISE
return StartRpcServer(dbms_handler_, data, auth_, system_);
#else
return StartRpcServer(dbms_handler_, data);
#endif
}},
repl_state_.ReplicationData());
// TODO Handle error (restore to main?)
return success;
}
memgraph::replication::ReplicationState &repl_state_;
memgraph::dbms::DbmsHandler &dbms_handler_;

View File

@@ -192,41 +192,12 @@ bool ReplicationHandler::SetReplicationRoleMain() {
bool ReplicationHandler::SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
const std::optional<utils::UUID> &main_uuid) {
// We don't want to restart the server if we're already a REPLICA
if (repl_state_.IsReplica()) {
spdlog::trace("Instance has already has replica role.");
return false;
}
return SetReplicationRoleReplica_<true>(config, main_uuid);
}
// TODO StorageState needs to be synched. Could have a dangling reference if someone adds a database as we are
// deleting the replica.
// Remove database specific clients
dbms_handler_.ForEach([&](memgraph::dbms::DatabaseAccess db_acc) {
auto *storage = db_acc->storage();
storage->repl_storage_state_.replication_clients_.WithLock([](auto &clients) { clients.clear(); });
});
// Remove instance level clients
std::get<memgraph::replication::RoleMainData>(repl_state_.ReplicationData()).registered_replicas_.clear();
// Creates the server
repl_state_.SetReplicationRoleReplica(config, main_uuid);
// Start
const auto success =
std::visit(memgraph::utils::Overloaded{[](memgraph::replication::RoleMainData &) {
// ASSERT
return false;
},
[this](memgraph::replication::RoleReplicaData &data) {
#ifdef MG_ENTERPRISE
return StartRpcServer(dbms_handler_, data, auth_, system_);
#else
return StartRpcServer(dbms_handler_, data);
#endif
}},
repl_state_.ReplicationData());
// TODO Handle error (restore to main?)
return success;
bool ReplicationHandler::TrySetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
const std::optional<utils::UUID> &main_uuid) {
return SetReplicationRoleReplica_<false>(config, main_uuid);
}
bool ReplicationHandler::DoReplicaToMainPromotion(const utils::UUID &main_uuid) {
@@ -255,16 +226,14 @@ bool ReplicationHandler::DoReplicaToMainPromotion(const utils::UUID &main_uuid)
};
// as MAIN, define and connect to REPLICAs
auto ReplicationHandler::TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config,
bool send_swap_uuid)
auto ReplicationHandler::TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config)
-> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> {
return RegisterReplica_<false>(config, send_swap_uuid);
return RegisterReplica_<true>(config);
}
auto ReplicationHandler::RegisterReplica(const memgraph::replication::ReplicationClientConfig &config,
bool send_swap_uuid)
auto ReplicationHandler::RegisterReplica(const memgraph::replication::ReplicationClientConfig &config)
-> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> {
return RegisterReplica_<true>(config, send_swap_uuid);
return RegisterReplica_<false>(config);
}
auto ReplicationHandler::UnregisterReplica(std::string_view name) -> memgraph::query::UnregisterReplicaResult {
@@ -297,6 +266,11 @@ auto ReplicationHandler::GetRole() const -> memgraph::replication_coordination_g
return repl_state_.GetRole();
}
auto ReplicationHandler::GetReplicaUUID() -> std::optional<utils::UUID> {
MG_ASSERT(repl_state_.IsReplica());
return std::get<RoleReplicaData>(repl_state_.ReplicationData()).uuid_;
}
auto ReplicationHandler::GetReplState() const -> const memgraph::replication::ReplicationState & { return repl_state_; }
auto ReplicationHandler::GetReplState() -> memgraph::replication::ReplicationState & { return repl_state_; }

View File

@@ -27,6 +27,8 @@
#include "utils/on_scope_exit.hpp"
#include "utils/typeinfo.hpp"
#include "io/network/fmt.hpp"
namespace memgraph::rpc {
/// Client is thread safe, but it is recommended to use thread_local clients.

View File

@@ -1278,7 +1278,7 @@ bool DiskStorage::DeleteEdgeFromConnectivityIndex(Transaction *transaction, cons
/// std::map<dst_vertex_gid, ...>
/// Here we also do flushing of too many things, we don't need to serialize edges in read-only txn, check that...
[[nodiscard]] utils::BasicResult<StorageManipulationError, void> DiskStorage::FlushModifiedEdges(
Transaction *transaction, const auto &edge_acc) {
Transaction *transaction, const auto &edges_acc) {
for (const auto &modified_edge : transaction->modified_edges_) {
const std::string edge_gid = modified_edge.first.ToString();
const Delta::Action root_action = modified_edge.second.delta_action;
@@ -1304,8 +1304,8 @@ bool DiskStorage::DeleteEdgeFromConnectivityIndex(Transaction *transaction, cons
return StorageManipulationError{SerializationError{}};
}
const auto &edge = edge_acc.find(modified_edge.first);
MG_ASSERT(edge != edge_acc.end(),
const auto &edge = edges_acc.find(modified_edge.first);
MG_ASSERT(edge != edges_acc.end(),
"Database in invalid state, commit not possible! Please restart your DB and start the import again.");
/// TODO: (andi) I think this is not wrong but it would be better to use AtomicWrites across column families.
@@ -1693,9 +1693,8 @@ utils::BasicResult<StorageManipulationError, void> DiskStorage::DiskAccessor::Co
transaction_.commit_timestamp->store(*commit_timestamp_, std::memory_order_release);
if (edge_import_mode_active) {
if (auto res =
disk_storage->FlushModifiedEdges(&transaction_, disk_storage->edge_import_mode_cache_->AccessToEdges());
res.HasError()) {
auto edges_acc = disk_storage->edge_import_mode_cache_->AccessToEdges();
if (auto res = disk_storage->FlushModifiedEdges(&transaction_, edges_acc); res.HasError()) {
Abort();
return res;
}
@@ -1717,7 +1716,8 @@ utils::BasicResult<StorageManipulationError, void> DiskStorage::DiskAccessor::Co
return del_vertices_res.GetError();
}
if (auto modified_edges_res = disk_storage->FlushModifiedEdges(&transaction_, transaction_.edges_->access());
auto tx_edges_acc = transaction_.edges_->access();
if (auto modified_edges_res = disk_storage->FlushModifiedEdges(&transaction_, tx_edges_acc);
modified_edges_res.HasError()) {
Abort();
return modified_edges_res.GetError();

View File

@@ -195,7 +195,7 @@ class DiskStorage final : public Storage {
[[nodiscard]] utils::BasicResult<StorageManipulationError, void> FlushDeletedVertices(Transaction *transaction);
[[nodiscard]] utils::BasicResult<StorageManipulationError, void> FlushDeletedEdges(Transaction *transaction);
[[nodiscard]] utils::BasicResult<StorageManipulationError, void> FlushModifiedEdges(Transaction *transaction,
const auto &edge_acc);
const auto &edges_acc);
[[nodiscard]] utils::BasicResult<StorageManipulationError, void> ClearDanglingVertices(Transaction *transaction);
/// Writing methods

View File

@@ -22,6 +22,7 @@
#include "storage/v2/edge.hpp"
#include "storage/v2/edge_accessor.hpp"
#include "storage/v2/edge_ref.hpp"
#include "storage/v2/fmt.hpp"
#include "storage/v2/id_types.hpp"
#include "storage/v2/indices/label_index_stats.hpp"
#include "storage/v2/indices/label_property_index_stats.hpp"

View File

@@ -17,6 +17,7 @@
#include "storage/v2/delta.hpp"
#include "storage/v2/mvcc.hpp"
#include "storage/v2/property_store.hpp"
#include "storage/v2/property_value.hpp"
#include "storage/v2/result.hpp"
#include "storage/v2/storage.hpp"
@@ -264,6 +265,27 @@ Result<PropertyValue> EdgeAccessor::GetProperty(PropertyId property, View view)
return *std::move(value);
}
Result<uint64_t> EdgeAccessor::GetPropertySize(PropertyId property, View view) const {
if (!storage_->config_.salient.items.properties_on_edges) return 0;
auto guard = std::shared_lock{edge_.ptr->lock};
Delta *delta = edge_.ptr->delta;
if (!delta) {
return edge_.ptr->properties.PropertySize(property);
}
auto property_result = this->GetProperty(property, view);
if (property_result.HasError()) {
return property_result.GetError();
}
auto property_store = storage::PropertyStore();
property_store.SetProperty(property, *property_result);
return property_store.PropertySize(property);
};
Result<std::map<PropertyId, PropertyValue>> EdgeAccessor::Properties(View view) const {
if (!storage_->config_.salient.items.properties_on_edges) return std::map<PropertyId, PropertyValue>{};
bool exists = true;

View File

@@ -82,6 +82,9 @@ class EdgeAccessor final {
/// @throw std::bad_alloc
Result<PropertyValue> GetProperty(PropertyId property, View view) const;
/// Returns the size of the encoded edge property in bytes.
Result<uint64_t> GetPropertySize(PropertyId property, View view) const;
/// @throw std::bad_alloc
Result<std::map<PropertyId, PropertyValue>> Properties(View view) const;

23
src/storage/v2/fmt.hpp Normal file
View File

@@ -0,0 +1,23 @@
// Copyright 2024 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
#if FMT_VERSION > 90000
#include <fmt/ostream.h>
#include "storage/v2/property_value.hpp"
template <>
class fmt::formatter<memgraph::storage::PropertyValue> : public fmt::ostream_formatter {};
template <>
class fmt::formatter<memgraph::storage::PropertyValue::Type> : public fmt::ostream_formatter {};
#endif

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -83,6 +83,18 @@ class NameIdMapper {
return id;
}
/// This method unlike NameToId does not insert the new property id if not found
/// but just returns either std::nullopt or the value of the property id if it
/// finds it.
virtual std::optional<uint64_t> NameToIdIfExists(const std::string_view name) {
auto name_to_id_acc = name_to_id_.access();
auto found = name_to_id_acc.find(name);
if (found == name_to_id_acc.end()) {
return std::nullopt;
}
return found->id;
}
// NOTE: Currently this function returns a `const std::string &` instead of a
// `std::string` to avoid making unnecessary copies of the string.
// Usually, this wouldn't be correct because the accessor to the

View File

@@ -93,6 +93,19 @@ enum class Size : uint8_t {
INT64 = 0x03,
};
uint64_t SizeToByteSize(Size size) {
switch (size) {
case Size::INT8:
return 1;
case Size::INT16:
return 2;
case Size::INT32:
return 4;
case Size::INT64:
return 8;
}
}
// All of these values must have the lowest 4 bits set to zero because they are
// used to store two `Size` values as described in the comment above.
enum class Type : uint8_t {
@@ -486,6 +499,27 @@ std::optional<TemporalData> DecodeTemporalData(Reader &reader) {
return TemporalData{static_cast<TemporalType>(*type_value), *microseconds_value};
}
std::optional<uint64_t> DecodeTemporalDataSize(Reader &reader) {
uint64_t temporal_data_size = 0;
auto metadata = reader.ReadMetadata();
if (!metadata || metadata->type != Type::TEMPORAL_DATA) return std::nullopt;
temporal_data_size += 1;
auto type_value = reader.ReadUint(metadata->id_size);
if (!type_value) return std::nullopt;
temporal_data_size += SizeToByteSize(metadata->id_size);
auto microseconds_value = reader.ReadInt(metadata->payload_size);
if (!microseconds_value) return std::nullopt;
temporal_data_size += SizeToByteSize(metadata->payload_size);
return temporal_data_size;
}
} // namespace
// Function used to decode a PropertyValue from a byte stream.
@@ -572,6 +606,92 @@ std::optional<TemporalData> DecodeTemporalData(Reader &reader) {
}
}
[[nodiscard]] bool DecodePropertyValueSize(Reader *reader, Type type, Size payload_size, uint64_t &property_size) {
switch (type) {
case Type::EMPTY: {
return false;
}
case Type::NONE:
case Type::BOOL: {
return true;
}
case Type::INT: {
reader->ReadInt(payload_size);
property_size += SizeToByteSize(payload_size);
return true;
}
case Type::DOUBLE: {
reader->ReadDouble(payload_size);
property_size += SizeToByteSize(payload_size);
return true;
}
case Type::STRING: {
auto size = reader->ReadUint(payload_size);
if (!size) return false;
property_size += SizeToByteSize(payload_size);
std::string str_v(*size, '\0');
if (!reader->SkipBytes(*size)) return false;
property_size += *size;
return true;
}
case Type::LIST: {
auto size = reader->ReadUint(payload_size);
if (!size) return false;
uint64_t list_property_size = SizeToByteSize(payload_size);
for (uint64_t i = 0; i < *size; ++i) {
auto metadata = reader->ReadMetadata();
if (!metadata) return false;
list_property_size += 1;
if (!DecodePropertyValueSize(reader, metadata->type, metadata->payload_size, list_property_size)) return false;
}
property_size += list_property_size;
return true;
}
case Type::MAP: {
auto size = reader->ReadUint(payload_size);
if (!size) return false;
uint64_t map_property_size = SizeToByteSize(payload_size);
for (uint64_t i = 0; i < *size; ++i) {
auto metadata = reader->ReadMetadata();
if (!metadata) return false;
map_property_size += 1;
auto key_size = reader->ReadUint(metadata->id_size);
if (!key_size) return false;
map_property_size += SizeToByteSize(metadata->id_size);
std::string key(*key_size, '\0');
if (!reader->ReadBytes(key.data(), *key_size)) return false;
map_property_size += *key_size;
if (!DecodePropertyValueSize(reader, metadata->type, metadata->payload_size, map_property_size)) return false;
}
property_size += map_property_size;
return true;
}
case Type::TEMPORAL_DATA: {
const auto maybe_temporal_data_size = DecodeTemporalDataSize(*reader);
if (!maybe_temporal_data_size) return false;
property_size += *maybe_temporal_data_size;
return true;
}
}
}
// Function used to skip a PropertyValue from a byte stream.
//
// @sa ComparePropertyValue
@@ -788,6 +908,27 @@ enum class ExpectedPropertyStatus {
: ExpectedPropertyStatus::GREATER;
}
[[nodiscard]] ExpectedPropertyStatus DecodeExpectedPropertySize(Reader *reader, PropertyId expected_property,
uint64_t &size) {
auto metadata = reader->ReadMetadata();
if (!metadata) return ExpectedPropertyStatus::MISSING_DATA;
auto property_id = reader->ReadUint(metadata->id_size);
if (!property_id) return ExpectedPropertyStatus::MISSING_DATA;
if (*property_id == expected_property.AsUint()) {
// Add one byte for reading metadata + add the number of bytes for the property key
size += (1 + SizeToByteSize(metadata->id_size));
if (!DecodePropertyValueSize(reader, metadata->type, metadata->payload_size, size))
return ExpectedPropertyStatus::MISSING_DATA;
return ExpectedPropertyStatus::EQUAL;
}
// Don't load the value if this isn't the expected property.
if (!SkipPropertyValue(reader, metadata->type, metadata->payload_size)) return ExpectedPropertyStatus::MISSING_DATA;
return (*property_id < expected_property.AsUint()) ? ExpectedPropertyStatus::SMALLER
: ExpectedPropertyStatus::GREATER;
}
// Function used to check a property exists (PropertyId) from a byte stream.
// It will skip the encoded PropertyValue.
//
@@ -875,6 +1016,13 @@ enum class ExpectedPropertyStatus {
}
}
[[nodiscard]] ExpectedPropertyStatus FindSpecificPropertySize(Reader *reader, PropertyId property, uint64_t &size) {
ExpectedPropertyStatus ret = ExpectedPropertyStatus::SMALLER;
while ((ret = DecodeExpectedPropertySize(reader, property, size)) == ExpectedPropertyStatus::SMALLER) {
}
return ret;
}
// Function used to find if property is set. It relies on the fact that the properties
// are sorted (by ID) in the buffer.
//
@@ -983,6 +1131,31 @@ std::pair<uint64_t, uint8_t *> GetSizeData(const uint8_t *buffer) {
return {size, data};
}
struct BufferInfo {
uint64_t size;
uint8_t *data{nullptr};
bool in_local_buffer;
};
template <size_t N>
BufferInfo GetBufferInfo(const uint8_t (&buffer)[N]) {
uint64_t size = 0;
const uint8_t *data = nullptr;
bool in_local_buffer = false;
std::tie(size, data) = GetSizeData(buffer);
if (size % 8 != 0) {
// We are storing the data in the local buffer.
size = sizeof(buffer) - 1;
data = &buffer[1];
in_local_buffer = true;
}
// NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast)
auto *non_const_data = const_cast<uint8_t *>(data);
return {size, non_const_data, in_local_buffer};
}
void SetSizeData(uint8_t *buffer, uint64_t size, uint8_t *data) {
memcpy(buffer, &size, sizeof(uint64_t));
memcpy(buffer + sizeof(uint64_t), &data, sizeof(uint8_t *));
@@ -1023,30 +1196,27 @@ PropertyStore::~PropertyStore() {
}
PropertyValue PropertyStore::GetProperty(PropertyId property) const {
uint64_t size;
const uint8_t *data;
std::tie(size, data) = GetSizeData(buffer_);
if (size % 8 != 0) {
// We are storing the data in the local buffer.
size = sizeof(buffer_) - 1;
data = &buffer_[1];
}
Reader reader(data, size);
BufferInfo buffer_info = GetBufferInfo(buffer_);
Reader reader(buffer_info.data, buffer_info.size);
PropertyValue value;
if (FindSpecificProperty(&reader, property, value) != ExpectedPropertyStatus::EQUAL) return {};
return value;
}
uint64_t PropertyStore::PropertySize(PropertyId property) const {
auto data_size_localbuffer = GetBufferInfo(buffer_);
Reader reader(data_size_localbuffer.data, data_size_localbuffer.size);
uint64_t property_size = 0;
if (FindSpecificPropertySize(&reader, property, property_size) != ExpectedPropertyStatus::EQUAL) return 0;
return property_size;
}
bool PropertyStore::HasProperty(PropertyId property) const {
uint64_t size;
const uint8_t *data;
std::tie(size, data) = GetSizeData(buffer_);
if (size % 8 != 0) {
// We are storing the data in the local buffer.
size = sizeof(buffer_) - 1;
data = &buffer_[1];
}
Reader reader(data, size);
BufferInfo buffer_info = GetBufferInfo(buffer_);
Reader reader(buffer_info.data, buffer_info.size);
return ExistsSpecificProperty(&reader, property) == ExpectedPropertyStatus::EQUAL;
}
@@ -1081,32 +1251,20 @@ std::optional<std::vector<PropertyValue>> PropertyStore::ExtractPropertyValues(
}
bool PropertyStore::IsPropertyEqual(PropertyId property, const PropertyValue &value) const {
uint64_t size;
const uint8_t *data;
std::tie(size, data) = GetSizeData(buffer_);
if (size % 8 != 0) {
// We are storing the data in the local buffer.
size = sizeof(buffer_) - 1;
data = &buffer_[1];
}
Reader reader(data, size);
BufferInfo buffer_info = GetBufferInfo(buffer_);
Reader reader(buffer_info.data, buffer_info.size);
auto info = FindSpecificPropertyAndBufferInfo(&reader, property);
if (info.property_size == 0) return value.IsNull();
Reader prop_reader(data + info.property_begin, info.property_size);
Reader prop_reader(buffer_info.data + info.property_begin, info.property_size);
if (!CompareExpectedProperty(&prop_reader, property, value)) return false;
return prop_reader.GetPosition() == info.property_size;
}
std::map<PropertyId, PropertyValue> PropertyStore::Properties() const {
uint64_t size;
const uint8_t *data;
std::tie(size, data) = GetSizeData(buffer_);
if (size % 8 != 0) {
// We are storing the data in the local buffer.
size = sizeof(buffer_) - 1;
data = &buffer_[1];
}
Reader reader(data, size);
BufferInfo buffer_info = GetBufferInfo(buffer_);
Reader reader(buffer_info.data, buffer_info.size);
std::map<PropertyId, PropertyValue> props;
while (true) {
PropertyValue value;
@@ -1340,33 +1498,20 @@ bool PropertyStore::InitProperties(std::vector<std::pair<storage::PropertyId, st
}
bool PropertyStore::ClearProperties() {
bool in_local_buffer = false;
uint64_t size;
uint8_t *data;
std::tie(size, data) = GetSizeData(buffer_);
if (size % 8 != 0) {
// We are storing the data in the local buffer.
size = sizeof(buffer_) - 1;
data = &buffer_[1];
in_local_buffer = true;
}
if (!size) return false;
if (!in_local_buffer) delete[] data;
BufferInfo buffer_info = GetBufferInfo(buffer_);
if (!buffer_info.size) return false;
if (!buffer_info.in_local_buffer) delete[] buffer_info.data;
SetSizeData(buffer_, 0, nullptr);
return true;
}
std::string PropertyStore::StringBuffer() const {
uint64_t size = 0;
const uint8_t *data = nullptr;
std::tie(size, data) = GetSizeData(buffer_);
if (size % 8 != 0) { // We are storing the data in the local buffer.
size = sizeof(buffer_) - 1;
data = &buffer_[1];
}
std::string arr(size, ' ');
for (uint i = 0; i < size; ++i) {
arr[i] = static_cast<char>(data[i]);
BufferInfo buffer_info = GetBufferInfo(buffer_);
std::string arr(buffer_info.size, ' ');
for (uint i = 0; i < buffer_info.size; ++i) {
arr[i] = static_cast<char>(buffer_info.data[i]);
}
return arr;
}

View File

@@ -45,6 +45,11 @@ class PropertyStore {
/// @throw std::bad_alloc
PropertyValue GetProperty(PropertyId property) const;
/// Returns the size of the encoded property in bytes.
/// Returns 0 if the property does not exist.
/// The time complexity of this function is O(n).
uint64_t PropertySize(PropertyId property) const;
/// Checks whether the property `property` exists in the store. The time
/// complexity of this function is O(n).
bool HasProperty(PropertyId property) const;

View File

@@ -9,6 +9,8 @@
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include <algorithm>
#include "replication/replication_client.hpp"
#include "storage/v2/inmemory/storage.hpp"
#include "storage/v2/storage.hpp"
@@ -17,7 +19,7 @@
#include "utils/uuid.hpp"
#include "utils/variant_helpers.hpp"
#include <algorithm>
#include "io/network/fmt.hpp"
namespace {
template <typename>

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source

View File

@@ -250,6 +250,10 @@ class Storage {
PropertyId NameToProperty(std::string_view name) { return storage_->NameToProperty(name); }
std::optional<PropertyId> NameToPropertyIfExists(std::string_view name) const {
return storage_->NameToPropertyIfExists(name);
}
EdgeTypeId NameToEdgeType(std::string_view name) { return storage_->NameToEdgeType(name); }
StorageMode GetCreationStorageMode() const noexcept;
@@ -318,6 +322,14 @@ class Storage {
return PropertyId::FromUint(name_id_mapper_->NameToId(name));
}
std::optional<PropertyId> NameToPropertyIfExists(std::string_view name) const {
const auto id = name_id_mapper_->NameToIdIfExists(name);
if (!id) {
return std::nullopt;
}
return PropertyId::FromUint(*id);
}
EdgeTypeId NameToEdgeType(const std::string_view name) const {
return EdgeTypeId::FromUint(name_id_mapper_->NameToId(name));
}

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -438,6 +438,26 @@ Result<PropertyValue> VertexAccessor::GetProperty(PropertyId property, View view
return std::move(value);
}
Result<uint64_t> VertexAccessor::GetPropertySize(PropertyId property, View view) const {
{
auto guard = std::shared_lock{vertex_->lock};
Delta *delta = vertex_->delta;
if (!delta) {
return vertex_->properties.PropertySize(property);
}
}
auto property_result = this->GetProperty(property, view);
if (property_result.HasError()) {
return property_result.GetError();
}
auto property_store = storage::PropertyStore();
property_store.SetProperty(property, *property_result);
return property_store.PropertySize(property);
};
Result<std::map<PropertyId, PropertyValue>> VertexAccessor::Properties(View view) const {
bool exists = true;
bool deleted = false;

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -80,6 +80,9 @@ class VertexAccessor final {
/// @throw std::bad_alloc
Result<PropertyValue> GetProperty(PropertyId property, View view) const;
/// Returns the size of the encoded vertex property in bytes.
Result<uint64_t> GetPropertySize(PropertyId property, View view) const;
/// @throw std::bad_alloc
Result<std::map<PropertyId, PropertyValue>> Properties(View view) const;

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2024 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
@@ -56,12 +56,11 @@ void EraseFlag(uint64_t flag_id) { expiration_flags.access().remove(flag_id); }
std::weak_ptr<std::atomic<bool>> GetFlag(uint64_t flag_id) {
const auto flag_accessor = expiration_flags.access();
const auto it = flag_accessor.find(flag_id);
if (it == flag_accessor.end()) {
const auto iter = flag_accessor.find(flag_id);
if (iter == flag_accessor.end()) {
return {};
}
return it->flag;
return iter->flag;
}
void MarkDone(const uint64_t flag_id) {

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd.
// Copyright 2024 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -42,12 +42,26 @@ namespace memgraph::utils {
class BasicException : public std::exception {
public:
/**
* @brief Constructor (C++ STL strings).
* @brief Constructor (C++ STL strings_view).
*
* @param message The error message.
*/
explicit BasicException(std::string_view message) noexcept : msg_(message) {}
/**
* @brief Constructor (string literal).
*
* @param message The error message.
*/
explicit BasicException(const char *message) noexcept : msg_(message) {}
/**
* @brief Constructor (C++ STL strings).
*
* @param message The error message.
*/
explicit BasicException(std::string message) noexcept : msg_(std::move(message)) {}
/**
* @brief Constructor with format string (C++ STL strings).
*

26
src/utils/functional.hpp Normal file
View File

@@ -0,0 +1,26 @@
// Copyright 2024 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 <vector>
#include <range/v3/view.hpp>
namespace memgraph::utils {
template <class F, class T, class R = typename std::invoke_result<F, T>::type>
auto fmap(F &&f, std::vector<T> const &v) -> std::vector<R> {
return v | ranges::views::transform(std::forward<F>(f)) | ranges::to<std::vector<R>>();
}
} // namespace memgraph::utils

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