tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
remove-graphql
memgraph/tests/unit/storage_v2_durability_inmemory.cpp
Andi 78a88737f8
HA: Add automatic failover (#1646) 
Co-authored-by: antoniofilipovic <filipovicantonio1998@gmail.com>
2024-01-29 15:34:00 +01:00

2975 lines
119 KiB
C++
Raw Permalink Blame History

// 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.
#include <gmock/gmock-generated-matchers.h>
#include <gmock/gmock.h>
#include <gtest/gtest-death-test.h>
#include <gtest/gtest.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <algorithm>
#include <chrono>
#include <csignal>
#include <cstdint>
#include <filesystem>
#include <iostream>
#include <thread>
#include <type_traits>
#include "dbms/database.hpp"
#include "replication/state.hpp"
#include "storage/v2/config.hpp"
#include "storage/v2/constraints/constraints.hpp"
#include "storage/v2/constraints/existence_constraints.hpp"
#include "storage/v2/durability/durability.hpp"
#include "storage/v2/durability/marker.hpp"
#include "storage/v2/durability/paths.hpp"
#include "storage/v2/durability/snapshot.hpp"
#include "storage/v2/durability/version.hpp"
#include "storage/v2/durability/wal.hpp"
#include "storage/v2/edge_accessor.hpp"
#include "storage/v2/indices/label_index_stats.hpp"
#include "storage/v2/inmemory/storage.hpp"
#include "storage/v2/inmemory/unique_constraints.hpp"
#include "storage/v2/storage_mode.hpp"
#include "storage/v2/vertex_accessor.hpp"
#include "utils/file.hpp"
#include "utils/logging.hpp"
#include "utils/timer.hpp"
#include "utils/uuid.hpp"
using memgraph::replication_coordination_glue::ReplicationRole;
using testing::Contains;
using testing::UnorderedElementsAre;
class DurabilityTest : public ::testing::TestWithParam<bool> {
protected:
const uint64_t kNumBaseVertices = 1000;
const uint64_t kNumBaseEdges = 10000;
const uint64_t kNumExtendedVertices = 100;
const uint64_t kNumExtendedEdges = 1000;
// We don't want to flush the WAL while we are doing operations because the
// flushing adds a large overhead that slows down execution.
const uint64_t kFlushWalEvery = (kNumBaseVertices + kNumBaseEdges + kNumExtendedVertices + kNumExtendedEdges) * 2;
enum class DatasetType {
ONLY_BASE,
ONLY_BASE_WITH_EXTENDED_INDICES_AND_CONSTRAINTS,
ONLY_EXTENDED,
ONLY_EXTENDED_WITH_BASE_INDICES_AND_CONSTRAINTS,
BASE_WITH_EXTENDED,
};
public:
DurabilityTest()
: base_vertex_gids_(kNumBaseVertices, memgraph::storage::Gid::FromUint(std::numeric_limits<uint64_t>::max())),
base_edge_gids_(kNumBaseEdges, memgraph::storage::Gid::FromUint(std::numeric_limits<uint64_t>::max())),
extended_vertex_gids_(kNumExtendedVertices,
memgraph::storage::Gid::FromUint(std::numeric_limits<uint64_t>::max())),
extended_edge_gids_(kNumExtendedEdges, memgraph::storage::Gid::FromUint(std::numeric_limits<uint64_t>::max())) {
}
void SetUp() override { Clear(); }
void TearDown() override { Clear(); }
void CreateBaseDataset(memgraph::storage::Storage *store, bool properties_on_edges) {
auto label_indexed = store->NameToLabel("base_indexed");
auto label_unindexed = store->NameToLabel("base_unindexed");
auto property_id = store->NameToProperty("id");
auto property_extra = store->NameToProperty("extra");
auto et1 = store->NameToEdgeType("base_et1");
auto et2 = store->NameToEdgeType("base_et2");
{
// Create label index.
auto unique_acc = store->UniqueAccess(ReplicationRole::MAIN);
ASSERT_FALSE(unique_acc->CreateIndex(label_unindexed).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
// Create label index statistics.
auto acc = store->Access(ReplicationRole::MAIN);
acc->SetIndexStats(label_unindexed, memgraph::storage::LabelIndexStats{1, 2});
ASSERT_TRUE(acc->GetIndexStats(label_unindexed));
ASSERT_FALSE(acc->Commit().HasError());
}
{
// Create label+property index.
auto unique_acc = store->UniqueAccess(ReplicationRole::MAIN);
ASSERT_FALSE(unique_acc->CreateIndex(label_indexed, property_id).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
// Create label+property index statistics.
auto acc = store->Access(ReplicationRole::MAIN);
acc->SetIndexStats(label_indexed, property_id, memgraph::storage::LabelPropertyIndexStats{1, 2, 3.4, 5.6, 0.0});
ASSERT_TRUE(acc->GetIndexStats(label_indexed, property_id));
ASSERT_FALSE(acc->Commit().HasError());
}
{
// Create existence constraint.
auto unique_acc = store->UniqueAccess(ReplicationRole::MAIN);
ASSERT_FALSE(unique_acc->CreateExistenceConstraint(label_unindexed, property_id).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
// Create unique constraint.
auto unique_acc = store->UniqueAccess(ReplicationRole::MAIN);
ASSERT_FALSE(unique_acc->CreateUniqueConstraint(label_unindexed, {property_id, property_extra}).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
// Create vertices.
for (uint64_t i = 0; i < kNumBaseVertices; ++i) {
auto acc = store->Access(ReplicationRole::MAIN);
auto vertex = acc->CreateVertex();
base_vertex_gids_[i] = vertex.Gid();
if (i < kNumBaseVertices / 2) {
ASSERT_TRUE(vertex.AddLabel(label_indexed).HasValue());
} else {
ASSERT_TRUE(vertex.AddLabel(label_unindexed).HasValue());
}
if (i < kNumBaseVertices / 3 || i >= kNumBaseVertices / 2) {
ASSERT_TRUE(
vertex.SetProperty(property_id, memgraph::storage::PropertyValue(static_cast<int64_t>(i))).HasValue());
}
ASSERT_FALSE(acc->Commit().HasError());
}
// Create edges.
for (uint64_t i = 0; i < kNumBaseEdges; ++i) {
auto acc = store->Access(ReplicationRole::MAIN);
auto vertex1 = acc->FindVertex(base_vertex_gids_[(i / 2) % kNumBaseVertices], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex1);
auto vertex2 = acc->FindVertex(base_vertex_gids_[(i / 3) % kNumBaseVertices], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex2);
memgraph::storage::EdgeTypeId et;
if (i < kNumBaseEdges / 2) {
et = et1;
} else {
et = et2;
}
auto edgeRes = acc->CreateEdge(&*vertex1, &*vertex2, et);
ASSERT_TRUE(edgeRes.HasValue());
auto edge = std::move(edgeRes.GetValue());
base_edge_gids_[i] = edge.Gid();
if (properties_on_edges) {
ASSERT_TRUE(
edge.SetProperty(property_id, memgraph::storage::PropertyValue(static_cast<int64_t>(i))).HasValue());
} else {
auto ret = edge.SetProperty(property_id, memgraph::storage::PropertyValue(static_cast<int64_t>(i)));
ASSERT_TRUE(ret.HasError());
ASSERT_EQ(ret.GetError(), memgraph::storage::Error::PROPERTIES_DISABLED);
}
ASSERT_FALSE(acc->Commit().HasError());
}
}
void CreateExtendedDataset(memgraph::storage::Storage *store, bool single_transaction = false) {
auto label_indexed = store->NameToLabel("extended_indexed");
auto label_unused = store->NameToLabel("extended_unused");
auto property_count = store->NameToProperty("count");
auto et3 = store->NameToEdgeType("extended_et3");
auto et4 = store->NameToEdgeType("extended_et4");
{
// Create label index.
auto unique_acc = store->UniqueAccess(ReplicationRole::MAIN);
ASSERT_FALSE(unique_acc->CreateIndex(label_unused).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
// Create label index statistics.
auto acc = store->Access(ReplicationRole::MAIN);
acc->SetIndexStats(label_unused, memgraph::storage::LabelIndexStats{123, 9.87});
ASSERT_TRUE(acc->GetIndexStats(label_unused));
ASSERT_FALSE(acc->Commit().HasError());
}
{
// Create label+property index.
auto unique_acc = store->UniqueAccess(ReplicationRole::MAIN);
ASSERT_FALSE(unique_acc->CreateIndex(label_indexed, property_count).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
// Create label+property index statistics.
auto acc = store->Access(ReplicationRole::MAIN);
acc->SetIndexStats(label_indexed, property_count,
memgraph::storage::LabelPropertyIndexStats{456798, 312345, 12312312.2, 123123.2, 67876.9});
ASSERT_TRUE(acc->GetIndexStats(label_indexed, property_count));
ASSERT_FALSE(acc->Commit().HasError());
}
{
// Create existence constraint.
auto unique_acc = store->UniqueAccess(ReplicationRole::MAIN);
ASSERT_FALSE(unique_acc->CreateExistenceConstraint(label_unused, property_count).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
// Create unique constraint.
auto unique_acc = store->UniqueAccess(ReplicationRole::MAIN);
ASSERT_FALSE(unique_acc->CreateUniqueConstraint(label_unused, {property_count}).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
// Storage accessor.
std::unique_ptr<memgraph::storage::Storage::Accessor> acc;
if (single_transaction) acc = store->Access(ReplicationRole::MAIN);
// Create vertices.
for (uint64_t i = 0; i < kNumExtendedVertices; ++i) {
if (!single_transaction) acc = store->Access(ReplicationRole::MAIN);
auto vertex = acc->CreateVertex();
extended_vertex_gids_[i] = vertex.Gid();
if (i < kNumExtendedVertices / 2) {
ASSERT_TRUE(vertex.AddLabel(label_indexed).HasValue());
}
if (i < kNumExtendedVertices / 3 || i >= kNumExtendedVertices / 2) {
ASSERT_TRUE(vertex.SetProperty(property_count, memgraph::storage::PropertyValue("nandare")).HasValue());
}
if (!single_transaction) ASSERT_FALSE(acc->Commit().HasError());
}
// Create edges.
for (uint64_t i = 0; i < kNumExtendedEdges; ++i) {
if (!single_transaction) acc = store->Access(ReplicationRole::MAIN);
auto vertex1 =
acc->FindVertex(extended_vertex_gids_[(i / 5) % kNumExtendedVertices], memgraph::storage::View::NEW);
ASSERT_TRUE(vertex1);
auto vertex2 =
acc->FindVertex(extended_vertex_gids_[(i / 6) % kNumExtendedVertices], memgraph::storage::View::NEW);
ASSERT_TRUE(vertex2);
memgraph::storage::EdgeTypeId et;
if (i < kNumExtendedEdges / 4) {
et = et3;
} else {
et = et4;
}
auto edgeRes = acc->CreateEdge(&*vertex1, &*vertex2, et);
ASSERT_TRUE(edgeRes.HasValue());
auto edge = std::move(edgeRes.GetValue());
extended_edge_gids_[i] = edge.Gid();
if (!single_transaction) ASSERT_FALSE(acc->Commit().HasError());
}
if (single_transaction) ASSERT_FALSE(acc->Commit().HasError());
}
void VerifyDataset(memgraph::storage::Storage *store, DatasetType type, bool properties_on_edges,
bool verify_info = true) {
auto base_label_indexed = store->NameToLabel("base_indexed");
auto base_label_unindexed = store->NameToLabel("base_unindexed");
auto property_id = store->NameToProperty("id");
auto property_extra = store->NameToProperty("extra");
auto et1 = store->NameToEdgeType("base_et1");
auto et2 = store->NameToEdgeType("base_et2");
auto extended_label_indexed = store->NameToLabel("extended_indexed");
auto extended_label_unused = store->NameToLabel("extended_unused");
auto property_count = store->NameToProperty("count");
auto et3 = store->NameToEdgeType("extended_et3");
auto et4 = store->NameToEdgeType("extended_et4");
// Create storage accessor.
auto acc = store->Access(ReplicationRole::MAIN);
// Verify indices info.
{
auto info = acc->ListAllIndices();
switch (type) {
case DatasetType::ONLY_BASE:
ASSERT_THAT(info.label, UnorderedElementsAre(base_label_unindexed));
ASSERT_THAT(info.label_property, UnorderedElementsAre(std::make_pair(base_label_indexed, property_id)));
break;
case DatasetType::ONLY_EXTENDED:
ASSERT_THAT(info.label, UnorderedElementsAre(extended_label_unused));
ASSERT_THAT(info.label_property,
UnorderedElementsAre(std::make_pair(base_label_indexed, property_id),
std::make_pair(extended_label_indexed, property_count)));
break;
case DatasetType::ONLY_BASE_WITH_EXTENDED_INDICES_AND_CONSTRAINTS:
case DatasetType::ONLY_EXTENDED_WITH_BASE_INDICES_AND_CONSTRAINTS:
case DatasetType::BASE_WITH_EXTENDED:
ASSERT_THAT(info.label, UnorderedElementsAre(base_label_unindexed, extended_label_unused));
ASSERT_THAT(info.label_property,
UnorderedElementsAre(std::make_pair(base_label_indexed, property_id),
std::make_pair(extended_label_indexed, property_count)));
break;
}
}
// Verify index statistics
{
switch (type) {
case DatasetType::ONLY_BASE: {
const auto l_stats = acc->GetIndexStats(base_label_unindexed);
ASSERT_TRUE(l_stats);
ASSERT_EQ(l_stats->count, 1);
ASSERT_EQ(l_stats->avg_degree, 2);
const auto lp_stats = acc->GetIndexStats(base_label_indexed, property_id);
ASSERT_TRUE(lp_stats);
ASSERT_EQ(lp_stats->count, 1);
ASSERT_EQ(lp_stats->distinct_values_count, 2);
ASSERT_EQ(lp_stats->statistic, 3.4);
ASSERT_EQ(lp_stats->avg_group_size, 5.6);
ASSERT_EQ(lp_stats->avg_degree, 0.0);
break;
}
case DatasetType::ONLY_EXTENDED: {
const auto l_stats = acc->GetIndexStats(extended_label_unused);
ASSERT_TRUE(l_stats);
ASSERT_EQ(l_stats->count, 123);
ASSERT_EQ(l_stats->avg_degree, 9.87);
const auto lp_stats = acc->GetIndexStats(extended_label_indexed, property_count);
ASSERT_TRUE(lp_stats);
ASSERT_EQ(lp_stats->count, 456798);
ASSERT_EQ(lp_stats->distinct_values_count, 312345);
ASSERT_EQ(lp_stats->statistic, 12312312.2);
ASSERT_EQ(lp_stats->avg_group_size, 123123.2);
ASSERT_EQ(lp_stats->avg_degree, 67876.9);
break;
}
case DatasetType::ONLY_BASE_WITH_EXTENDED_INDICES_AND_CONSTRAINTS:
case DatasetType::ONLY_EXTENDED_WITH_BASE_INDICES_AND_CONSTRAINTS:
case DatasetType::BASE_WITH_EXTENDED: {
const auto l_stats = acc->GetIndexStats(base_label_unindexed);
ASSERT_TRUE(l_stats);
ASSERT_EQ(l_stats->count, 1);
ASSERT_EQ(l_stats->avg_degree, 2);
const auto lp_stats = acc->GetIndexStats(base_label_indexed, property_id);
ASSERT_TRUE(lp_stats);
ASSERT_EQ(lp_stats->count, 1);
ASSERT_EQ(lp_stats->distinct_values_count, 2);
ASSERT_EQ(lp_stats->statistic, 3.4);
ASSERT_EQ(lp_stats->avg_group_size, 5.6);
ASSERT_EQ(lp_stats->avg_degree, 0.0);
const auto l_stats_ex = acc->GetIndexStats(extended_label_unused);
ASSERT_TRUE(l_stats_ex);
ASSERT_EQ(l_stats_ex->count, 123);
ASSERT_EQ(l_stats_ex->avg_degree, 9.87);
const auto lp_stats_ex = acc->GetIndexStats(extended_label_indexed, property_count);
ASSERT_TRUE(lp_stats_ex);
ASSERT_EQ(lp_stats_ex->count, 456798);
ASSERT_EQ(lp_stats_ex->distinct_values_count, 312345);
ASSERT_EQ(lp_stats_ex->statistic, 12312312.2);
ASSERT_EQ(lp_stats_ex->avg_group_size, 123123.2);
ASSERT_EQ(lp_stats_ex->avg_degree, 67876.9);
break;
}
}
}
// Verify constraints info.
{
auto info = acc->ListAllConstraints();
switch (type) {
case DatasetType::ONLY_BASE:
ASSERT_THAT(info.existence, UnorderedElementsAre(std::make_pair(base_label_unindexed, property_id)));
ASSERT_THAT(info.unique, UnorderedElementsAre(
std::make_pair(base_label_unindexed, std::set{property_id, property_extra})));
break;
case DatasetType::ONLY_EXTENDED:
ASSERT_THAT(info.existence, UnorderedElementsAre(std::make_pair(extended_label_unused, property_count)));
ASSERT_THAT(info.unique,
UnorderedElementsAre(std::make_pair(extended_label_unused, std::set{property_count})));
break;
case DatasetType::ONLY_BASE_WITH_EXTENDED_INDICES_AND_CONSTRAINTS:
case DatasetType::ONLY_EXTENDED_WITH_BASE_INDICES_AND_CONSTRAINTS:
case DatasetType::BASE_WITH_EXTENDED:
ASSERT_THAT(info.existence, UnorderedElementsAre(std::make_pair(base_label_unindexed, property_id),
std::make_pair(extended_label_unused, property_count)));
ASSERT_THAT(info.unique,
UnorderedElementsAre(std::make_pair(base_label_unindexed, std::set{property_id, property_extra}),
std::make_pair(extended_label_unused, std::set{property_count})));
break;
}
}
bool have_base_dataset = false;
bool have_extended_dataset = false;
switch (type) {
case DatasetType::ONLY_BASE:
case DatasetType::ONLY_BASE_WITH_EXTENDED_INDICES_AND_CONSTRAINTS:
have_base_dataset = true;
break;
case DatasetType::ONLY_EXTENDED:
case DatasetType::ONLY_EXTENDED_WITH_BASE_INDICES_AND_CONSTRAINTS:
have_extended_dataset = true;
break;
case DatasetType::BASE_WITH_EXTENDED:
have_base_dataset = true;
have_extended_dataset = true;
break;
}
// Verify base dataset.
if (have_base_dataset) {
// Verify vertices.
for (uint64_t i = 0; i < kNumBaseVertices; ++i) {
auto vertex = acc->FindVertex(base_vertex_gids_[i], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex);
auto labels = vertex->Labels(memgraph::storage::View::OLD);
ASSERT_TRUE(labels.HasValue());
if (i < kNumBaseVertices / 2) {
ASSERT_THAT(*labels, UnorderedElementsAre(base_label_indexed));
} else {
ASSERT_THAT(*labels, UnorderedElementsAre(base_label_unindexed));
}
auto properties = vertex->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(properties.HasValue());
if (i < kNumBaseVertices / 3 || i >= kNumBaseVertices / 2) {
ASSERT_EQ(properties->size(), 1);
ASSERT_EQ((*properties)[property_id], memgraph::storage::PropertyValue(static_cast<int64_t>(i)));
} else {
ASSERT_EQ(properties->size(), 0);
}
}
// Verify edges.
for (uint64_t i = 0; i < kNumBaseEdges; ++i) {
auto find_edge = [&](auto &edges) -> std::optional<memgraph::storage::EdgeAccessor> {
for (auto &edge : edges) {
if (edge.Gid() == base_edge_gids_[i]) {
return edge;
}
}
return {};
};
{
auto vertex1 = acc->FindVertex(base_vertex_gids_[(i / 2) % kNumBaseVertices], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex1);
auto out_edges = vertex1->OutEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(out_edges.HasValue());
auto edge1 = find_edge(out_edges->edges);
ASSERT_TRUE(edge1);
if (i < kNumBaseEdges / 2) {
ASSERT_EQ(edge1->EdgeType(), et1);
} else {
ASSERT_EQ(edge1->EdgeType(), et2);
}
auto properties = edge1->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(properties.HasValue());
if (properties_on_edges) {
ASSERT_EQ(properties->size(), 1);
ASSERT_EQ((*properties)[property_id], memgraph::storage::PropertyValue(static_cast<int64_t>(i)));
} else {
ASSERT_EQ(properties->size(), 0);
}
}
{
auto vertex2 = acc->FindVertex(base_vertex_gids_[(i / 3) % kNumBaseVertices], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex2);
auto in_edges = vertex2->InEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(in_edges.HasValue());
auto edge2 = find_edge(in_edges->edges);
ASSERT_TRUE(edge2);
if (i < kNumBaseEdges / 2) {
ASSERT_EQ(edge2->EdgeType(), et1);
} else {
ASSERT_EQ(edge2->EdgeType(), et2);
}
auto properties = edge2->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(properties.HasValue());
if (properties_on_edges) {
ASSERT_EQ(properties->size(), 1);
ASSERT_EQ((*properties)[property_id], memgraph::storage::PropertyValue(static_cast<int64_t>(i)));
} else {
ASSERT_EQ(properties->size(), 0);
}
}
}
// Verify label indices.
{
std::vector<memgraph::storage::VertexAccessor> vertices;
vertices.reserve(kNumBaseVertices / 2);
for (auto vertex : acc->Vertices(base_label_unindexed, memgraph::storage::View::OLD)) {
vertices.push_back(vertex);
}
ASSERT_EQ(vertices.size(), kNumBaseVertices / 2);
std::sort(vertices.begin(), vertices.end(), [](const auto &a, const auto &b) { return a.Gid() < b.Gid(); });
for (uint64_t i = 0; i < kNumBaseVertices / 2; ++i) {
ASSERT_EQ(vertices[i].Gid(), base_vertex_gids_[kNumBaseVertices / 2 + i]);
}
}
// Verify label+property index.
{
std::vector<memgraph::storage::VertexAccessor> vertices;
vertices.reserve(kNumBaseVertices / 3);
for (auto vertex : acc->Vertices(base_label_indexed, property_id, memgraph::storage::View::OLD)) {
vertices.push_back(vertex);
}
ASSERT_EQ(vertices.size(), kNumBaseVertices / 3);
std::sort(vertices.begin(), vertices.end(), [](const auto &a, const auto &b) { return a.Gid() < b.Gid(); });
for (uint64_t i = 0; i < kNumBaseVertices / 3; ++i) {
ASSERT_EQ(vertices[i].Gid(), base_vertex_gids_[i]);
}
}
} else {
// Verify vertices.
for (uint64_t i = 0; i < kNumBaseVertices; ++i) {
auto vertex = acc->FindVertex(base_vertex_gids_[i], memgraph::storage::View::OLD);
ASSERT_FALSE(vertex);
}
if (type == DatasetType::ONLY_EXTENDED_WITH_BASE_INDICES_AND_CONSTRAINTS) {
// Verify label indices.
{
uint64_t count = 0;
auto iterable = acc->Vertices(base_label_unindexed, memgraph::storage::View::OLD);
for (auto it = iterable.begin(); it != iterable.end(); ++it) {
++count;
}
ASSERT_EQ(count, 0);
}
// Verify label+property index.
{
uint64_t count = 0;
auto iterable = acc->Vertices(base_label_indexed, property_id, memgraph::storage::View::OLD);
for (auto it = iterable.begin(); it != iterable.end(); ++it) {
++count;
}
ASSERT_EQ(count, 0);
}
}
}
// Verify extended dataset.
if (have_extended_dataset) {
// Verify vertices.
for (uint64_t i = 0; i < kNumExtendedVertices; ++i) {
auto vertex = acc->FindVertex(extended_vertex_gids_[i], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex);
auto labels = vertex->Labels(memgraph::storage::View::OLD);
ASSERT_TRUE(labels.HasValue());
if (i < kNumExtendedVertices / 2) {
ASSERT_THAT(*labels, UnorderedElementsAre(extended_label_indexed));
}
auto properties = vertex->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(properties.HasValue());
if (i < kNumExtendedVertices / 3 || i >= kNumExtendedVertices / 2) {
ASSERT_EQ(properties->size(), 1);
ASSERT_EQ((*properties)[property_count], memgraph::storage::PropertyValue("nandare"));
} else {
ASSERT_EQ(properties->size(), 0);
}
}
// Verify edges.
for (uint64_t i = 0; i < kNumExtendedEdges; ++i) {
auto find_edge = [&](auto &edges) -> std::optional<memgraph::storage::EdgeAccessor> {
for (auto &edge : edges) {
if (edge.Gid() == extended_edge_gids_[i]) {
return edge;
}
}
return {};
};
{
auto vertex1 =
acc->FindVertex(extended_vertex_gids_[(i / 5) % kNumExtendedVertices], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex1);
auto out_edges = vertex1->OutEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(out_edges.HasValue());
auto edge1 = find_edge(out_edges->edges);
ASSERT_TRUE(edge1);
if (i < kNumExtendedEdges / 4) {
ASSERT_EQ(edge1->EdgeType(), et3);
} else {
ASSERT_EQ(edge1->EdgeType(), et4);
}
auto properties = edge1->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(properties.HasValue());
ASSERT_EQ(properties->size(), 0);
}
{
auto vertex2 =
acc->FindVertex(extended_vertex_gids_[(i / 6) % kNumExtendedVertices], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex2);
auto in_edges = vertex2->InEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(in_edges.HasValue());
auto edge2 = find_edge(in_edges->edges);
ASSERT_TRUE(edge2);
if (i < kNumExtendedEdges / 4) {
ASSERT_EQ(edge2->EdgeType(), et3);
} else {
ASSERT_EQ(edge2->EdgeType(), et4);
}
auto properties = edge2->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(properties.HasValue());
ASSERT_EQ(properties->size(), 0);
}
}
// Verify label indices.
{
std::vector<memgraph::storage::VertexAccessor> vertices;
vertices.reserve(kNumExtendedVertices / 2);
for (auto vertex : acc->Vertices(extended_label_unused, memgraph::storage::View::OLD)) {
vertices.emplace_back(vertex);
}
ASSERT_EQ(vertices.size(), 0);
}
// Verify label+property index.
{
std::vector<memgraph::storage::VertexAccessor> vertices;
vertices.reserve(kNumExtendedVertices / 3);
for (auto vertex : acc->Vertices(extended_label_indexed, property_count, memgraph::storage::View::OLD)) {
vertices.emplace_back(vertex);
}
ASSERT_EQ(vertices.size(), kNumExtendedVertices / 3);
std::sort(vertices.begin(), vertices.end(), [](const auto &a, const auto &b) { return a.Gid() < b.Gid(); });
for (uint64_t i = 0; i < kNumExtendedVertices / 3; ++i) {
ASSERT_EQ(vertices[i].Gid(), extended_vertex_gids_[i]);
}
}
} else {
// Verify vertices.
for (uint64_t i = 0; i < kNumExtendedVertices; ++i) {
auto vertex = acc->FindVertex(extended_vertex_gids_[i], memgraph::storage::View::OLD);
ASSERT_FALSE(vertex);
}
if (type == DatasetType::ONLY_BASE_WITH_EXTENDED_INDICES_AND_CONSTRAINTS) {
// Verify label indices.
{
uint64_t count = 0;
auto iterable = acc->Vertices(extended_label_unused, memgraph::storage::View::OLD);
for (auto it = iterable.begin(); it != iterable.end(); ++it) {
++count;
}
ASSERT_EQ(count, 0);
}
// Verify label+property index.
{
uint64_t count = 0;
auto iterable = acc->Vertices(extended_label_indexed, property_count, memgraph::storage::View::OLD);
for (auto it = iterable.begin(); it != iterable.end(); ++it) {
++count;
}
ASSERT_EQ(count, 0);
}
}
}
if (verify_info) {
auto info = store->GetBaseInfo();
if (have_base_dataset) {
if (have_extended_dataset) {
ASSERT_EQ(info.vertex_count, kNumBaseVertices + kNumExtendedVertices);
ASSERT_EQ(info.edge_count, kNumBaseEdges + kNumExtendedEdges);
} else {
ASSERT_EQ(info.vertex_count, kNumBaseVertices);
ASSERT_EQ(info.edge_count, kNumBaseEdges);
}
} else {
if (have_extended_dataset) {
ASSERT_EQ(info.vertex_count, kNumExtendedVertices);
ASSERT_EQ(info.edge_count, kNumExtendedEdges);
} else {
ASSERT_EQ(info.vertex_count, 0);
ASSERT_EQ(info.edge_count, 0);
}
}
}
}
std::vector<std::filesystem::path> GetSnapshotsList() {
return GetFilesList(storage_directory / memgraph::storage::durability::kSnapshotDirectory);
}
std::vector<std::filesystem::path> GetBackupSnapshotsList() {
return GetFilesList(storage_directory / memgraph::storage::durability::kBackupDirectory /
memgraph::storage::durability::kSnapshotDirectory);
}
std::vector<std::filesystem::path> GetWalsList() {
return GetFilesList(storage_directory / memgraph::storage::durability::kWalDirectory);
}
std::vector<std::filesystem::path> GetBackupWalsList() {
return GetFilesList(storage_directory / memgraph::storage::durability::kBackupDirectory /
memgraph::storage::durability::kWalDirectory);
}
void RestoreBackups() {
{
auto backup_snapshots = GetBackupSnapshotsList();
for (const auto &item : backup_snapshots) {
std::filesystem::rename(
item, storage_directory / memgraph::storage::durability::kSnapshotDirectory / item.filename());
}
}
{
auto backup_wals = GetBackupWalsList();
for (const auto &item : backup_wals) {
std::filesystem::rename(item,
storage_directory / memgraph::storage::durability::kWalDirectory / item.filename());
}
}
}
std::filesystem::path storage_directory{std::filesystem::temp_directory_path() /
"MG_test_unit_storage_v2_durability"};
private:
std::vector<std::filesystem::path> GetFilesList(const std::filesystem::path &path) {
std::vector<std::filesystem::path> ret;
std::error_code ec; // For exception suppression.
for (auto &item : std::filesystem::directory_iterator(path, ec)) {
ret.push_back(item.path());
}
std::sort(ret.begin(), ret.end());
std::reverse(ret.begin(), ret.end());
return ret;
}
void Clear() {
if (!std::filesystem::exists(storage_directory)) return;
std::filesystem::remove_all(storage_directory);
}
std::vector<memgraph::storage::Gid> base_vertex_gids_;
std::vector<memgraph::storage::Gid> base_edge_gids_;
std::vector<memgraph::storage::Gid> extended_vertex_gids_;
std::vector<memgraph::storage::Gid> extended_edge_gids_;
};
void CorruptSnapshot(const std::filesystem::path &path) {
auto info = memgraph::storage::durability::ReadSnapshotInfo(path);
spdlog::info("Destroying snapshot {}", path);
memgraph::utils::OutputFile file;
file.Open(path, memgraph::utils::OutputFile::Mode::OVERWRITE_EXISTING);
file.SetPosition(memgraph::utils::OutputFile::Position::SET, info.offset_vertices);
auto value = static_cast<uint8_t>(memgraph::storage::durability::Marker::TYPE_MAP);
file.Write(&value, sizeof(value));
file.Sync();
file.Close();
}
void DestroyWalFirstDelta(const std::filesystem::path &path) {
auto info = memgraph::storage::durability::ReadWalInfo(path);
spdlog::info("Destroying WAL {}", path);
memgraph::utils::OutputFile file;
file.Open(path, memgraph::utils::OutputFile::Mode::OVERWRITE_EXISTING);
file.SetPosition(memgraph::utils::OutputFile::Position::SET, info.offset_deltas);
auto value = static_cast<uint8_t>(memgraph::storage::durability::Marker::TYPE_MAP);
file.Write(&value, sizeof(value));
file.Sync();
file.Close();
}
void DestroyWalSuffix(const std::filesystem::path &path) {
auto info = memgraph::storage::durability::ReadWalInfo(path);
spdlog::info("Destroying WAL {}", path);
memgraph::utils::OutputFile file;
file.Open(path, memgraph::utils::OutputFile::Mode::OVERWRITE_EXISTING);
ASSERT_LT(info.offset_deltas, file.SetPosition(memgraph::utils::OutputFile::Position::RELATIVE_TO_END, -100));
uint8_t value = 0;
for (size_t i = 0; i < 100; ++i) {
file.Write(&value, sizeof(value));
}
file.Sync();
file.Close();
}
INSTANTIATE_TEST_CASE_P(EdgesWithProperties, DurabilityTest, ::testing::Values(true));
INSTANTIATE_TEST_CASE_P(EdgesWithoutProperties, DurabilityTest, ::testing::Values(false));
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, SnapshotOnExit) {
// Create snapshot.
{
memgraph::storage::Config config{.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = GetParam()}}};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
CreateExtendedDataset(db.storage());
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, SnapshotPeriodic) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode = memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT,
.snapshot_interval = std::chrono::milliseconds(2000)},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
std::this_thread::sleep_for(std::chrono::milliseconds(2500));
}
ASSERT_GE(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, SnapshotFallback) {
// Create snapshot.
std::size_t number_to_save;
{
// DEVNOTE_1: assumes that snapshot disk write takes less than this
auto const expected_write_time = std::chrono::milliseconds(750);
auto const snapshot_interval = std::chrono::milliseconds(3000);
memgraph::storage::Config config{
.durability =
{
.storage_directory = storage_directory,
.snapshot_wal_mode = memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT,
.snapshot_interval = snapshot_interval,
.snapshot_retention_count = 10, // We don't anticipate that we make this many
},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto const ensure_snapshot_is_written = [&](auto &&func) {
auto const pre_count = GetSnapshotsList().size();
func();
// wait long enough to ensure at least one CreateSnapshot has been invoked
// DEVNOTE_2: no guarantee that it completed, see DEVNOTE_1
std::this_thread::sleep_for(snapshot_interval + expected_write_time);
auto const post_count = GetSnapshotsList().size();
// validate at least one snapshot has happened...hence must have written the writes from func
ASSERT_GT(post_count, pre_count) << "No snapshot exists to capture the last transaction";
// TODO: maybe double check by looking at InMemoryStorage's commit log,
// its oldest active should be newer than the transaction used when running `func`
};
ensure_snapshot_is_written([&]() { CreateBaseDataset(db.storage(), GetParam()); });
number_to_save = GetSnapshotsList().size();
ensure_snapshot_is_written([&]() { CreateExtendedDataset(db.storage()); });
}
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Destroy snapshots.
{
auto snapshots = GetSnapshotsList();
// snapshots order newest first, destroy the newest, preserve number_to_save so that we ONLY_BASE
auto it = snapshots.begin();
auto const e = snapshots.end() - number_to_save;
for (; it != e; ++it) {
CorruptSnapshot(*it);
}
}
// Recover snapshot.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, SnapshotEverythingCorrupt) {
// Create unrelated snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t i = 0; i < 1000; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Get unrelated UUID.
std::string unrelated_uuid;
{
auto snapshots = GetSnapshotsList();
ASSERT_EQ(snapshots.size(), 1);
auto info = memgraph::storage::durability::ReadSnapshotInfo(*snapshots.begin());
unrelated_uuid = info.uuid;
}
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode = memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT,
.snapshot_interval = std::chrono::milliseconds(2000)},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
std::this_thread::sleep_for(std::chrono::milliseconds(2500));
CreateExtendedDataset(db.storage());
std::this_thread::sleep_for(std::chrono::milliseconds(2500));
}
ASSERT_GE(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 1);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Restore unrelated snapshots.
RestoreBackups();
ASSERT_GE(GetSnapshotsList().size(), 2);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Destroy all current snapshots.
{
auto snapshots = GetSnapshotsList();
ASSERT_GE(snapshots.size(), 2);
for (const auto &snapshot : snapshots) {
auto info = memgraph::storage::durability::ReadSnapshotInfo(snapshot);
if (info.uuid == unrelated_uuid) {
spdlog::info("Skipping snapshot {}", snapshot);
continue;
}
CorruptSnapshot(snapshot);
}
}
// Recover snapshot.
ASSERT_DEATH(
([&]() {
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
}()) // iile
,
"");
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, SnapshotRetention) {
// Create unrelated snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t i = 0; i < 1000; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_GE(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode = memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT,
.snapshot_interval = std::chrono::milliseconds(2000),
.snapshot_retention_count = 3},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
// Restore unrelated snapshots after the database has been started.
RestoreBackups();
CreateBaseDataset(db.storage(), GetParam());
// Allow approximately 5 snapshots to be created.
std::this_thread::sleep_for(std::chrono::milliseconds(10000));
}
ASSERT_EQ(GetSnapshotsList().size(), 1 + 3);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Verify that exactly 3 snapshots and 1 unrelated snapshot exist.
{
auto snapshots = GetSnapshotsList();
ASSERT_EQ(snapshots.size(), 1 + 3);
std::string uuid;
for (size_t i = 0; i < snapshots.size(); ++i) {
const auto &path = snapshots[i];
// This shouldn't throw.
auto info = memgraph::storage::durability::ReadSnapshotInfo(path);
if (i == 0) uuid = info.uuid;
if (i < snapshots.size() - 1) {
ASSERT_EQ(info.uuid, uuid);
} else {
ASSERT_NE(info.uuid, uuid);
}
}
}
// Recover snapshot.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, SnapshotMixedUUID) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
CreateExtendedDataset(db.storage());
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
}
// Create another snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 1);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Restore unrelated snapshot.
RestoreBackups();
ASSERT_EQ(GetSnapshotsList().size(), 2);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, SnapshotBackup) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t i = 0; i < 1000; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Start storage without recovery.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode = memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT,
.snapshot_interval = std::chrono::minutes(20)},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 1);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_F(DurabilityTest, SnapshotWithoutPropertiesOnEdgesRecoveryWithPropertiesOnEdges) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = false}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), false);
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, false);
CreateExtendedDataset(db.storage());
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, false);
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = true}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, false);
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_F(DurabilityTest, SnapshotWithPropertiesOnEdgesRecoveryWithoutPropertiesOnEdges) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = true}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), true);
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, true);
CreateExtendedDataset(db.storage());
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, true);
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
ASSERT_DEATH(
([&]() {
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = false}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
}()) // iile
,
"");
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_F(DurabilityTest, SnapshotWithPropertiesOnEdgesButUnusedRecoveryWithoutPropertiesOnEdges) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = true}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), true);
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, true);
CreateExtendedDataset(db.storage());
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, true);
// Remove properties from edges.
{
auto acc = db.Access();
for (auto vertex : acc->Vertices(memgraph::storage::View::OLD)) {
auto in_edges = vertex.InEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(in_edges.HasValue());
for (auto &edge : in_edges->edges) {
// TODO (mferencevic): Replace with `ClearProperties()`
auto props = edge.Properties(memgraph::storage::View::NEW);
ASSERT_TRUE(props.HasValue());
for (const auto &prop : *props) {
ASSERT_TRUE(edge.SetProperty(prop.first, memgraph::storage::PropertyValue()).HasValue());
}
}
auto out_edges = vertex.InEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(out_edges.HasValue());
for (auto &edge : out_edges->edges) {
// TODO (mferencevic): Replace with `ClearProperties()`
auto props = edge.Properties(memgraph::storage::View::NEW);
ASSERT_TRUE(props.HasValue());
for (const auto &prop : *props) {
ASSERT_TRUE(edge.SetProperty(prop.first, memgraph::storage::PropertyValue()).HasValue());
}
}
}
ASSERT_FALSE(acc->Commit().HasError());
}
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = false}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, false);
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalBasic) {
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
CreateExtendedDataset(db.storage());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalBackup) {
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t i = 0; i < 1000; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
auto num_wals = GetWalsList().size();
ASSERT_GE(num_wals, 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Start storage without recovery.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20)},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), num_wals);
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalAppendToExisting) {
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
}
// Recover WALs and create more WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.recover_on_startup = true,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateExtendedDataset(db.storage());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 2);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalCreateInSingleTransaction) {
// NOLINTNEXTLINE(readability-isolate-declaration)
memgraph::storage::Gid gid_v1, gid_v2, gid_e1, gid_v3;
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
auto v1 = acc->CreateVertex();
gid_v1 = v1.Gid();
auto v2 = acc->CreateVertex();
gid_v2 = v2.Gid();
auto e1Res = acc->CreateEdge(&v1, &v2, db.storage()->NameToEdgeType("e1"));
ASSERT_TRUE(e1Res.HasValue());
auto e1 = std::move(e1Res.GetValue());
gid_e1 = e1.Gid();
ASSERT_TRUE(v1.AddLabel(db.storage()->NameToLabel("l11")).HasValue());
ASSERT_TRUE(v1.AddLabel(db.storage()->NameToLabel("l12")).HasValue());
ASSERT_TRUE(v1.AddLabel(db.storage()->NameToLabel("l13")).HasValue());
if (GetParam()) {
ASSERT_TRUE(
e1.SetProperty(db.storage()->NameToProperty("test"), memgraph::storage::PropertyValue("nandare")).HasValue());
}
ASSERT_TRUE(v2.AddLabel(db.storage()->NameToLabel("l21")).HasValue());
ASSERT_TRUE(
v2.SetProperty(db.storage()->NameToProperty("hello"), memgraph::storage::PropertyValue("world")).HasValue());
auto v3 = acc->CreateVertex();
gid_v3 = v3.Gid();
ASSERT_TRUE(v3.SetProperty(db.storage()->NameToProperty("v3"), memgraph::storage::PropertyValue(42)).HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
{
auto acc = db.Access();
auto indices = acc->ListAllIndices();
ASSERT_EQ(indices.label.size(), 0);
ASSERT_EQ(indices.label_property.size(), 0);
auto constraints = acc->ListAllConstraints();
ASSERT_EQ(constraints.existence.size(), 0);
ASSERT_EQ(constraints.unique.size(), 0);
{
auto v1 = acc->FindVertex(gid_v1, memgraph::storage::View::OLD);
ASSERT_TRUE(v1);
auto labels = v1->Labels(memgraph::storage::View::OLD);
ASSERT_TRUE(labels.HasValue());
ASSERT_THAT(*labels, UnorderedElementsAre(db.storage()->NameToLabel("l11"), db.storage()->NameToLabel("l12"),
db.storage()->NameToLabel("l13")));
auto props = v1->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(props.HasValue());
ASSERT_EQ(props->size(), 0);
auto in_edges = v1->InEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(in_edges.HasValue());
ASSERT_EQ(in_edges->edges.size(), 0);
auto out_edges = v1->OutEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(out_edges.HasValue());
ASSERT_EQ(out_edges->edges.size(), 1);
const auto &edge = out_edges->edges[0];
ASSERT_EQ(edge.Gid(), gid_e1);
auto edge_props = edge.Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(edge_props.HasValue());
if (GetParam()) {
ASSERT_THAT(*edge_props, UnorderedElementsAre(std::make_pair(db.storage()->NameToProperty("test"),
memgraph::storage::PropertyValue("nandare"))));
} else {
ASSERT_EQ(edge_props->size(), 0);
}
}
{
auto v2 = acc->FindVertex(gid_v2, memgraph::storage::View::OLD);
ASSERT_TRUE(v2);
auto labels = v2->Labels(memgraph::storage::View::OLD);
ASSERT_TRUE(labels.HasValue());
ASSERT_THAT(*labels, UnorderedElementsAre(db.storage()->NameToLabel("l21")));
auto props = v2->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(props.HasValue());
ASSERT_THAT(*props, UnorderedElementsAre(std::make_pair(db.storage()->NameToProperty("hello"),
memgraph::storage::PropertyValue("world"))));
auto in_edges = v2->InEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(in_edges.HasValue());
ASSERT_EQ(in_edges->edges.size(), 1);
const auto &edge = in_edges->edges[0];
ASSERT_EQ(edge.Gid(), gid_e1);
auto edge_props = edge.Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(edge_props.HasValue());
if (GetParam()) {
ASSERT_THAT(*edge_props, UnorderedElementsAre(std::make_pair(db.storage()->NameToProperty("test"),
memgraph::storage::PropertyValue("nandare"))));
} else {
ASSERT_EQ(edge_props->size(), 0);
}
auto out_edges = v2->OutEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(out_edges.HasValue());
ASSERT_EQ(out_edges->edges.size(), 0);
}
{
auto v3 = acc->FindVertex(gid_v3, memgraph::storage::View::OLD);
ASSERT_TRUE(v3);
auto labels = v3->Labels(memgraph::storage::View::OLD);
ASSERT_TRUE(labels.HasValue());
ASSERT_EQ(labels->size(), 0);
auto props = v3->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(props.HasValue());
ASSERT_THAT(*props, UnorderedElementsAre(std::make_pair(db.storage()->NameToProperty("v3"),
memgraph::storage::PropertyValue(42))));
auto in_edges = v3->InEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(in_edges.HasValue());
ASSERT_EQ(in_edges->edges.size(), 0);
auto out_edges = v3->OutEdges(memgraph::storage::View::OLD);
ASSERT_TRUE(out_edges.HasValue());
ASSERT_EQ(out_edges->edges.size(), 0);
}
}
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalCreateAndRemoveEverything) {
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
CreateExtendedDataset(db.storage());
auto indices = [&] {
auto acc = db.Access();
auto res = acc->ListAllIndices();
(void)acc->Commit();
return res;
}(); // iile
for (const auto &index : indices.label) {
auto unique_acc = db.UniqueAccess();
ASSERT_FALSE(unique_acc->DropIndex(index).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
for (const auto &index : indices.label_property) {
auto unique_acc = db.UniqueAccess();
ASSERT_FALSE(unique_acc->DropIndex(index.first, index.second).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
auto constraints = [&] {
auto acc = db.Access();
auto res = acc->ListAllConstraints();
(void)acc->Commit();
return res;
}(); // iile
for (const auto &constraint : constraints.existence) {
auto unique_acc = db.UniqueAccess();
ASSERT_FALSE(unique_acc->DropExistenceConstraint(constraint.first, constraint.second).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
for (const auto &constraint : constraints.unique) {
auto unique_acc = db.UniqueAccess();
ASSERT_EQ(unique_acc->DropUniqueConstraint(constraint.first, constraint.second),
memgraph::storage::UniqueConstraints::DeletionStatus::SUCCESS);
ASSERT_FALSE(unique_acc->Commit().HasError());
}
auto acc = db.Access();
for (auto vertex : acc->Vertices(memgraph::storage::View::OLD)) {
ASSERT_TRUE(acc->DetachDeleteVertex(&vertex).HasValue());
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
{
auto acc = db.Access();
auto indices = acc->ListAllIndices();
ASSERT_EQ(indices.label.size(), 0);
ASSERT_EQ(indices.label_property.size(), 0);
auto constraints = acc->ListAllConstraints();
ASSERT_EQ(constraints.existence.size(), 0);
ASSERT_EQ(constraints.unique.size(), 0);
uint64_t count = 0;
auto iterable = acc->Vertices(memgraph::storage::View::OLD);
for (auto it = iterable.begin(); it != iterable.end(); ++it) {
++count;
}
ASSERT_EQ(count, 0);
}
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalTransactionOrdering) {
// NOLINTNEXTLINE(readability-isolate-declaration)
memgraph::storage::Gid gid1, gid2, gid3;
// Create WAL.
{
memgraph::storage::Config config{
.durability =
{
.storage_directory = storage_directory,
.snapshot_wal_mode = memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 100000,
.wal_file_flush_every_n_tx = kFlushWalEvery,
},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc1 = db.Access();
auto acc2 = db.Access();
// Create vertex in transaction 2.
{
auto vertex2 = acc2->CreateVertex();
gid2 = vertex2.Gid();
ASSERT_TRUE(
vertex2.SetProperty(db.storage()->NameToProperty("id"), memgraph::storage::PropertyValue(2)).HasValue());
}
auto acc3 = db.Access();
// Create vertex in transaction 3.
{
auto vertex3 = acc3->CreateVertex();
gid3 = vertex3.Gid();
ASSERT_TRUE(
vertex3.SetProperty(db.storage()->NameToProperty("id"), memgraph::storage::PropertyValue(3)).HasValue());
}
// Create vertex in transaction 1.
{
auto vertex1 = acc1->CreateVertex();
gid1 = vertex1.Gid();
ASSERT_TRUE(
vertex1.SetProperty(db.storage()->NameToProperty("id"), memgraph::storage::PropertyValue(1)).HasValue());
}
// Commit transaction 3, then 1, then 2.
ASSERT_FALSE(acc3->Commit().HasError());
ASSERT_FALSE(acc1->Commit().HasError());
ASSERT_FALSE(acc2->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Verify WAL data.
{
auto path = GetWalsList().front();
auto info = memgraph::storage::durability::ReadWalInfo(path);
memgraph::storage::durability::Decoder wal;
wal.Initialize(path, memgraph::storage::durability::kWalMagic);
wal.SetPosition(info.offset_deltas);
ASSERT_EQ(info.num_deltas, 9);
std::vector<std::pair<uint64_t, memgraph::storage::durability::WalDeltaData>> data;
for (uint64_t i = 0; i < info.num_deltas; ++i) {
auto timestamp = memgraph::storage::durability::ReadWalDeltaHeader(&wal);
data.emplace_back(timestamp, memgraph::storage::durability::ReadWalDeltaData(&wal));
}
// Verify timestamps.
ASSERT_EQ(data[1].first, data[0].first);
ASSERT_EQ(data[2].first, data[1].first);
ASSERT_GT(data[3].first, data[2].first);
ASSERT_EQ(data[4].first, data[3].first);
ASSERT_EQ(data[5].first, data[4].first);
ASSERT_GT(data[6].first, data[5].first);
ASSERT_EQ(data[7].first, data[6].first);
ASSERT_EQ(data[8].first, data[7].first);
// Verify transaction 3.
ASSERT_EQ(data[0].second.type, memgraph::storage::durability::WalDeltaData::Type::VERTEX_CREATE);
ASSERT_EQ(data[0].second.vertex_create_delete.gid, gid3);
ASSERT_EQ(data[1].second.type, memgraph::storage::durability::WalDeltaData::Type::VERTEX_SET_PROPERTY);
ASSERT_EQ(data[1].second.vertex_edge_set_property.gid, gid3);
ASSERT_EQ(data[1].second.vertex_edge_set_property.property, "id");
ASSERT_EQ(data[1].second.vertex_edge_set_property.value, memgraph::storage::PropertyValue(3));
ASSERT_EQ(data[2].second.type, memgraph::storage::durability::WalDeltaData::Type::TRANSACTION_END);
// Verify transaction 1.
ASSERT_EQ(data[3].second.type, memgraph::storage::durability::WalDeltaData::Type::VERTEX_CREATE);
ASSERT_EQ(data[3].second.vertex_create_delete.gid, gid1);
ASSERT_EQ(data[4].second.type, memgraph::storage::durability::WalDeltaData::Type::VERTEX_SET_PROPERTY);
ASSERT_EQ(data[4].second.vertex_edge_set_property.gid, gid1);
ASSERT_EQ(data[4].second.vertex_edge_set_property.property, "id");
ASSERT_EQ(data[4].second.vertex_edge_set_property.value, memgraph::storage::PropertyValue(1));
ASSERT_EQ(data[5].second.type, memgraph::storage::durability::WalDeltaData::Type::TRANSACTION_END);
// Verify transaction 2.
ASSERT_EQ(data[6].second.type, memgraph::storage::durability::WalDeltaData::Type::VERTEX_CREATE);
ASSERT_EQ(data[6].second.vertex_create_delete.gid, gid2);
ASSERT_EQ(data[7].second.type, memgraph::storage::durability::WalDeltaData::Type::VERTEX_SET_PROPERTY);
ASSERT_EQ(data[7].second.vertex_edge_set_property.gid, gid2);
ASSERT_EQ(data[7].second.vertex_edge_set_property.property, "id");
ASSERT_EQ(data[7].second.vertex_edge_set_property.value, memgraph::storage::PropertyValue(2));
ASSERT_EQ(data[8].second.type, memgraph::storage::durability::WalDeltaData::Type::TRANSACTION_END);
}
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
{
auto acc = db.Access();
for (auto [gid, id] : std::vector<std::pair<memgraph::storage::Gid, int64_t>>{{gid1, 1}, {gid2, 2}, {gid3, 3}}) {
auto vertex = acc->FindVertex(gid, memgraph::storage::View::OLD);
ASSERT_TRUE(vertex);
auto labels = vertex->Labels(memgraph::storage::View::OLD);
ASSERT_TRUE(labels.HasValue());
ASSERT_EQ(labels->size(), 0);
auto props = vertex->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(props.HasValue());
ASSERT_EQ(props->size(), 1);
ASSERT_EQ(props->at(db.storage()->NameToProperty("id")), memgraph::storage::PropertyValue(id));
}
}
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalCreateAndRemoveOnlyBaseDataset) {
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
CreateExtendedDataset(db.storage());
auto label_indexed = db.storage()->NameToLabel("base_indexed");
auto label_unindexed = db.storage()->NameToLabel("base_unindexed");
auto acc = db.Access();
for (auto vertex : acc->Vertices(memgraph::storage::View::OLD)) {
auto has_indexed = vertex.HasLabel(label_indexed, memgraph::storage::View::OLD);
ASSERT_TRUE(has_indexed.HasValue());
auto has_unindexed = vertex.HasLabel(label_unindexed, memgraph::storage::View::OLD);
if (!*has_indexed && !*has_unindexed) continue;
ASSERT_TRUE(acc->DetachDeleteVertex(&vertex).HasValue());
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::ONLY_EXTENDED_WITH_BASE_INDICES_AND_CONSTRAINTS, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalDeathResilience) {
pid_t pid = fork();
if (pid == 0) {
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
// Create one million vertices.
for (uint64_t i = 0; i < 1000000; ++i) {
auto acc = db.Access();
acc->CreateVertex();
MG_ASSERT(!acc->Commit().HasError(), "Couldn't commit transaction!");
}
}
} else if (pid > 0) {
// Wait for WALs to be created.
std::this_thread::sleep_for(std::chrono::seconds(2));
int status;
EXPECT_EQ(waitpid(pid, &status, WNOHANG), 0);
EXPECT_EQ(kill(pid, SIGKILL), 0);
EXPECT_EQ(waitpid(pid, &status, 0), pid);
EXPECT_NE(status, 0);
} else {
LOG_FATAL("Couldn't create process to execute test!");
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs and create more WALs.
const uint64_t kExtraItems = 1000;
uint64_t count = 0;
{
memgraph::storage::Config config{
.durability =
{
.storage_directory = storage_directory,
.recover_on_startup = true,
.snapshot_wal_mode = memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery,
},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
{
auto acc = db.Access();
auto iterable = acc->Vertices(memgraph::storage::View::OLD);
for (auto it = iterable.begin(); it != iterable.end(); ++it) {
++count;
}
ASSERT_GT(count, 0);
}
{
auto acc = db.Access();
for (uint64_t i = 0; i < kExtraItems; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
}
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 2);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
{
uint64_t current = 0;
auto acc = db.Access();
auto iterable = acc->Vertices(memgraph::storage::View::OLD);
for (auto it = iterable.begin(); it != iterable.end(); ++it) {
++current;
}
ASSERT_EQ(count + kExtraItems, current);
}
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalMissingSecond) {
// Create unrelated WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t i = 0; i < 1000; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
uint64_t unrelated_wals = GetWalsList().size();
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
const uint64_t kNumVertices = 1000;
std::vector<memgraph::storage::Gid> gids;
gids.reserve(kNumVertices);
for (uint64_t i = 0; i < kNumVertices; ++i) {
auto acc = db.Access();
auto vertex = acc->CreateVertex();
gids.push_back(vertex.Gid());
ASSERT_FALSE(acc->Commit().HasError());
}
for (uint64_t i = 0; i < kNumVertices; ++i) {
auto acc = db.Access();
auto vertex = acc->FindVertex(gids[i], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex);
ASSERT_TRUE(
vertex->SetProperty(db.storage()->NameToProperty("nandare"), memgraph::storage::PropertyValue("haihaihai!"))
.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_GE(GetBackupWalsList().size(), 1);
// Restore unrelated WALs.
RestoreBackups();
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 2);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Remove second WAL.
{
auto wals = GetWalsList();
ASSERT_GT(wals.size(), unrelated_wals + 2);
const auto &wal_file = wals[wals.size() - unrelated_wals - 2];
spdlog::info("Deleting WAL file {}", wal_file);
ASSERT_TRUE(std::filesystem::remove(wal_file));
}
// Recover WALs.
ASSERT_DEATH(
([&]() {
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
}()) // iile
,
"");
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalCorruptSecond) {
// Create unrelated WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t i = 0; i < 1000; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
uint64_t unrelated_wals = GetWalsList().size();
// Create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
const uint64_t kNumVertices = 1000;
std::vector<memgraph::storage::Gid> gids;
gids.reserve(kNumVertices);
for (uint64_t i = 0; i < kNumVertices; ++i) {
auto acc = db.Access();
auto vertex = acc->CreateVertex();
gids.push_back(vertex.Gid());
ASSERT_FALSE(acc->Commit().HasError());
}
for (uint64_t i = 0; i < kNumVertices; ++i) {
auto acc = db.Access();
auto vertex = acc->FindVertex(gids[i], memgraph::storage::View::OLD);
ASSERT_TRUE(vertex);
ASSERT_TRUE(
vertex->SetProperty(db.storage()->NameToProperty("nandare"), memgraph::storage::PropertyValue("haihaihai!"))
.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_GE(GetBackupWalsList().size(), 1);
// Restore unrelated WALs.
RestoreBackups();
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 2);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Destroy second WAL.
{
auto wals = GetWalsList();
ASSERT_GT(wals.size(), unrelated_wals + 2);
const auto &wal_file = wals[wals.size() - unrelated_wals - 2];
DestroyWalFirstDelta(wal_file);
}
// Recover WALs.
ASSERT_DEATH(
([&]() {
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
}()) // iile
,
"");
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalCorruptLastTransaction) {
// Create WALs
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
CreateExtendedDataset(db.storage(), /* single_transaction = */ true);
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 2);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Destroy last transaction in the latest WAL.
{
auto wals = GetWalsList();
ASSERT_GE(wals.size(), 2);
const auto &wal_file = wals.front();
DestroyWalSuffix(wal_file);
}
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
// The extended dataset shouldn't be recovered because its WAL transaction was
// corrupt.
VerifyDataset(db.storage(), DatasetType::ONLY_BASE_WITH_EXTENDED_INDICES_AND_CONSTRAINTS, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalAllOperationsInSingleTransaction) {
// Create WALs
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
auto vertex1 = acc->CreateVertex();
auto vertex2 = acc->CreateVertex();
ASSERT_TRUE(vertex1.AddLabel(acc->NameToLabel("nandare")).HasValue());
ASSERT_TRUE(vertex2.SetProperty(acc->NameToProperty("haihai"), memgraph::storage::PropertyValue(42)).HasValue());
ASSERT_TRUE(vertex1.RemoveLabel(acc->NameToLabel("nandare")).HasValue());
auto edge1Res = acc->CreateEdge(&vertex1, &vertex2, acc->NameToEdgeType("et1"));
ASSERT_TRUE(edge1Res.HasValue());
auto edge1 = std::move(edge1Res.GetValue());
ASSERT_TRUE(vertex2.SetProperty(acc->NameToProperty("haihai"), memgraph::storage::PropertyValue()).HasValue());
auto vertex3 = acc->CreateVertex();
auto edge2Res = acc->CreateEdge(&vertex3, &vertex3, acc->NameToEdgeType("et2"));
ASSERT_TRUE(edge2Res.HasValue());
auto edge2 = std::move(edge2Res.GetValue());
if (GetParam()) {
ASSERT_TRUE(edge2.SetProperty(acc->NameToProperty("meaning"), memgraph::storage::PropertyValue(true)).HasValue());
ASSERT_TRUE(
edge1.SetProperty(acc->NameToProperty("hello"), memgraph::storage::PropertyValue("world")).HasValue());
ASSERT_TRUE(edge2.SetProperty(acc->NameToProperty("meaning"), memgraph::storage::PropertyValue()).HasValue());
}
ASSERT_TRUE(vertex3.AddLabel(acc->NameToLabel("test")).HasValue());
ASSERT_TRUE(vertex3.SetProperty(acc->NameToProperty("nonono"), memgraph::storage::PropertyValue(-1)).HasValue());
ASSERT_TRUE(vertex3.SetProperty(acc->NameToProperty("nonono"), memgraph::storage::PropertyValue()).HasValue());
if (GetParam()) {
ASSERT_TRUE(edge1.SetProperty(acc->NameToProperty("hello"), memgraph::storage::PropertyValue()).HasValue());
}
ASSERT_TRUE(vertex3.RemoveLabel(acc->NameToLabel("test")).HasValue());
ASSERT_TRUE(acc->DetachDeleteVertex(&vertex1).HasValue());
ASSERT_TRUE(acc->DeleteEdge(&edge2).HasValue());
ASSERT_TRUE(acc->DeleteVertex(&vertex2).HasValue());
ASSERT_TRUE(acc->DeleteVertex(&vertex3).HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
{
auto acc = db.Access();
uint64_t count = 0;
auto iterable = acc->Vertices(memgraph::storage::View::OLD);
for (auto it = iterable.begin(); it != iterable.end(); ++it) {
++count;
}
ASSERT_EQ(count, 0);
}
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalAndSnapshot) {
// Create snapshot and WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::milliseconds(2000),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
std::this_thread::sleep_for(std::chrono::milliseconds(2500));
CreateExtendedDataset(db.storage());
}
ASSERT_GE(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot and WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalAndSnapshotAppendToExistingSnapshot) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
}
// Recover snapshot and create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.recover_on_startup = true,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateExtendedDataset(db.storage());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot and WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalAndSnapshotAppendToExistingSnapshotAndWal) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
}
// Recover snapshot and create WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.recover_on_startup = true,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateExtendedDataset(db.storage());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot and WALs and create more WALs.
memgraph::storage::Gid vertex_gid;
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.recover_on_startup = true,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
auto acc = db.Access();
auto vertex = acc->CreateVertex();
vertex_gid = vertex.Gid();
if (GetParam()) {
ASSERT_TRUE(
vertex.SetProperty(db.storage()->NameToProperty("meaning"), memgraph::storage::PropertyValue(42)).HasValue());
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 2);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot and WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam(),
/* verify_info = */ false);
{
auto acc = db.Access();
auto vertex = acc->FindVertex(vertex_gid, memgraph::storage::View::OLD);
ASSERT_TRUE(vertex);
auto labels = vertex->Labels(memgraph::storage::View::OLD);
ASSERT_TRUE(labels.HasValue());
ASSERT_EQ(labels->size(), 0);
auto props = vertex->Properties(memgraph::storage::View::OLD);
ASSERT_TRUE(props.HasValue());
if (GetParam()) {
ASSERT_THAT(*props, UnorderedElementsAre(std::make_pair(db.storage()->NameToProperty("meaning"),
memgraph::storage::PropertyValue(42))));
} else {
ASSERT_EQ(props->size(), 0);
}
}
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, WalAndSnapshotWalRetention) {
// Create unrelated WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::minutes(20),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = kFlushWalEvery},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t i = 0; i < 1000; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
}
ASSERT_EQ(GetSnapshotsList().size(), 0);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
uint64_t unrelated_wals = GetWalsList().size();
uint64_t items_created = 0;
// Create snapshot and WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::seconds(2),
.wal_file_size_kibibytes = 1,
.wal_file_flush_every_n_tx = 1},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
// Restore unrelated snapshots after the database has been started.
RestoreBackups();
memgraph::utils::Timer timer;
// Allow at least 6 snapshots to be created.
while (timer.Elapsed().count() < 13.0) {
auto acc = db.Access();
acc->CreateVertex();
ASSERT_FALSE(acc->Commit().HasError());
++items_created;
}
}
ASSERT_EQ(GetSnapshotsList().size(), 3);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), unrelated_wals + 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
auto snapshots = GetSnapshotsList();
ASSERT_EQ(snapshots.size(), 3);
for (uint64_t i = 0; i < snapshots.size(); ++i) {
spdlog::info("Recovery attempt {}", i);
// Recover and verify data.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t j = 0; j < items_created; ++j) {
auto vertex = acc->FindVertex(memgraph::storage::Gid::FromUint(j), memgraph::storage::View::OLD);
ASSERT_TRUE(vertex);
}
}
// Destroy current snapshot.
CorruptSnapshot(snapshots[i]);
}
// Recover data after all of the snapshots have been destroyed. The recovery
// shouldn't be possible because the initial WALs are already deleted.
ASSERT_DEATH(
([&]() {
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
}()) // iile
,
"");
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, SnapshotAndWalMixedUUID) {
// Create unrelated snapshot and WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::seconds(2)},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
auto acc = db.Access();
for (uint64_t i = 0; i < 1000; ++i) {
acc->CreateVertex();
}
ASSERT_FALSE(acc->Commit().HasError());
std::this_thread::sleep_for(std::chrono::milliseconds(2500));
}
ASSERT_GE(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Create snapshot and WALs.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL,
.snapshot_interval = std::chrono::seconds(2)},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
std::this_thread::sleep_for(std::chrono::milliseconds(2500));
CreateExtendedDataset(db.storage());
std::this_thread::sleep_for(std::chrono::milliseconds(2500));
}
ASSERT_GE(GetSnapshotsList().size(), 1);
ASSERT_GE(GetBackupSnapshotsList().size(), 1);
ASSERT_GE(GetWalsList().size(), 1);
ASSERT_GE(GetBackupWalsList().size(), 1);
// Restore unrelated snapshots and WALs.
RestoreBackups();
ASSERT_GE(GetSnapshotsList().size(), 2);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_GE(GetWalsList().size(), 2);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot and WALs.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .recover_on_startup = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
// Try to use the storage.
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, ParallelConstraintsRecovery) {
// Create snapshot.
{
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory, .snapshot_on_exit = true, .items_per_batch = 13},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
CreateExtendedDataset(db.storage());
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
// Recover snapshot.
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.recover_on_startup = true,
.snapshot_on_exit = false,
.items_per_batch = 13,
.allow_parallel_index_creation = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
{
auto acc = db.Access();
auto vertex = acc->CreateVertex();
auto edge = acc->CreateEdge(&vertex, &vertex, db.storage()->NameToEdgeType("et"));
ASSERT_TRUE(edge.HasValue());
ASSERT_FALSE(acc->Commit().HasError());
}
}
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_P(DurabilityTest, ConstraintsRecoveryFunctionSetting) {
memgraph::storage::Config config{
.durability = {.storage_directory = storage_directory,
.recover_on_startup = true,
.snapshot_on_exit = false,
.items_per_batch = 13,
.allow_parallel_schema_creation = true},
.salient = {.items = {.properties_on_edges = GetParam()}},
};
// Create snapshot.
{
config.durability.recover_on_startup = false;
config.durability.snapshot_on_exit = true;
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::dbms::Database db{config, repl_state};
CreateBaseDataset(db.storage(), GetParam());
VerifyDataset(db.storage(), DatasetType::ONLY_BASE, GetParam());
CreateExtendedDataset(db.storage());
VerifyDataset(db.storage(), DatasetType::BASE_WITH_EXTENDED, GetParam());
}
ASSERT_EQ(GetSnapshotsList().size(), 1);
ASSERT_EQ(GetBackupSnapshotsList().size(), 0);
ASSERT_EQ(GetWalsList().size(), 0);
ASSERT_EQ(GetBackupWalsList().size(), 0);
config.durability.recover_on_startup = true;
config.durability.snapshot_on_exit = false;
memgraph::replication::ReplicationState repl_state{memgraph::storage::ReplicationStateRootPath(config)};
memgraph::utils::SkipList<memgraph::storage::Vertex> vertices;
memgraph::utils::SkipList<memgraph::storage::Edge> edges;
std::unique_ptr<memgraph::storage::NameIdMapper> name_id_mapper = std::make_unique<memgraph::storage::NameIdMapper>();
std::atomic<uint64_t> edge_count{0};
uint64_t wal_seq_num{0};
std::string uuid{memgraph::utils::GenerateUUID()};
memgraph::storage::Indices indices{config, memgraph::storage::StorageMode::IN_MEMORY_TRANSACTIONAL};
memgraph::storage::Constraints constraints{config, memgraph::storage::StorageMode::IN_MEMORY_TRANSACTIONAL};
memgraph::storage::ReplicationStorageState repl_storage_state;
memgraph::storage::durability::Recovery recovery{
config.durability.storage_directory / memgraph::storage::durability::kSnapshotDirectory,
config.durability.storage_directory / memgraph::storage::durability::kWalDirectory};
// Recover snapshot.
const auto info = recovery.RecoverData(&uuid, repl_storage_state, &vertices, &edges, &edge_count,
name_id_mapper.get(), &indices, &constraints, config, &wal_seq_num);
MG_ASSERT(info.has_value(), "Info doesn't have value present");
const auto par_exec_info = memgraph::storage::durability::GetParallelExecInfo(*info, config);
MG_ASSERT(par_exec_info.has_value(), "Parallel exec info should have value present");
// Unique constraint choose function
auto *mem_unique_constraints =
static_cast<memgraph::storage::InMemoryUniqueConstraints *>(constraints.unique_constraints_.get());
auto variant_unique_constraint_creation_func = mem_unique_constraints->GetCreationFunction(par_exec_info);
const auto *pval = std::get_if<memgraph::storage::InMemoryUniqueConstraints::MultipleThreadsConstraintValidation>(
&variant_unique_constraint_creation_func);
MG_ASSERT(pval, "Chose wrong function for recovery of data");
// Existence constraint choose function
auto *mem_existence_constraint =
static_cast<memgraph::storage::ExistenceConstraints *>(constraints.existence_constraints_.get());
auto variant_existence_constraint_creation_func = mem_existence_constraint->GetCreationFunction(par_exec_info);
const auto *pval_existence =
std::get_if<memgraph::storage::ExistenceConstraints::MultipleThreadsConstraintValidation>(
&variant_existence_constraint_creation_func);
MG_ASSERT(pval_existence, "Chose wrong type of function for recovery of existence constraint data");
}
Reference in New Issue View Git Blame Copy Permalink