tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add logs for loading snapshot and WAL files ()

Browse Source 
* Add logs for loading snapshot and WAL files
This commit is contained in:
János Benjamin Antal 2021-03-26 15:02:35 +01:00 committed by GitHub
parent 50ddd59450
commit 06f761bdf9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 101 additions and 12 deletions
src
storage/v2/durability
durability.cppsnapshot.cppwal.cpp
utils
logging.hpp
tests/e2e/replication
constraints.cpp

44
src/storage/v2/durability/durability.cpp
View File

@ -102,30 +102,47 @@ std::optional<std::vector<WalDurabilityInfo>> GetWalFiles(const std::filesystem:
// recovery process.
void RecoverIndicesAndConstraints(const RecoveredIndicesAndConstraints &indices_constraints, Indices *indices,
Constraints *constraints, utils::SkipList<Vertex> *vertices) {
spdlog::info("Recreating indices from metadata.");
// Recover label indices.
spdlog::info("Recreating {} label indices from metadata.", indices_constraints.indices.label.size());
for (const auto &item : indices_constraints.indices.label) {
if (!indices->label_index.CreateIndex(item, vertices->access()))
throw RecoveryFailure("The label index must be created here!");
spdlog::info("A label index is recreated from metadata.");
}
spdlog::info("Label indices are recreated.");
// Recover label+property indices.
spdlog::info("Recreating {} label+property indices from metadata.",
indices_constraints.indices.label_property.size());
for (const auto &item : indices_constraints.indices.label_property) {
if (!indices->label_property_index.CreateIndex(item.first, item.second, vertices->access()))
throw RecoveryFailure("The label+property index must be created here!");
spdlog::info("A label+property index is recreated from metadata.");
}
spdlog::info("Label+property indices are recreated.");
spdlog::info("Indices are recreated.");
spdlog::info("Recreating constraints from metadata.");
// Recover existence constraints.
spdlog::info("Recreating {} existence constraints from metadata.", indices_constraints.constraints.existence.size());
for (const auto &item : indices_constraints.constraints.existence) {
auto ret = CreateExistenceConstraint(constraints, item.first, item.second, vertices->access());
if (ret.HasError() || !ret.GetValue()) throw RecoveryFailure("The existence constraint must be created here!");
spdlog::info("A existence constraint is recreated from metadata.");
}
spdlog::info("Existence constraints are recreated from metadata.");
// Recover unique constraints.
spdlog::info("Recreating {} unique constraints from metadata.", indices_constraints.constraints.unique.size());
for (const auto &item : indices_constraints.constraints.unique) {
auto ret = constraints->unique_constraints.CreateConstraint(item.first, item.second, vertices->access());
if (ret.HasError() || ret.GetValue() != UniqueConstraints::CreationStatus::SUCCESS)
throw RecoveryFailure("The unique constraint must be created here!");
spdlog::info("A unique constraint is recreated from metadata.");
}
spdlog::info("Unique constraints are recreated from metadata.");
spdlog::info("Constraints are recreated from metadata.");
}
std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_directory,
@ -137,7 +154,12 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
Indices *indices, Constraints *constraints, Config::Items items,
uint64_t *wal_seq_num) {
utils::MemoryTracker::OutOfMemoryExceptionEnabler oom_exception;
if (!utils::DirExists(snapshot_directory) && !utils::DirExists(wal_directory)) return std::nullopt;
spdlog::info("Recovering persisted data using snapshot ({}) and WAL directory ({}).", snapshot_directory,
wal_directory);
if (!utils::DirExists(snapshot_directory) && !utils::DirExists(wal_directory)) {
spdlog::warn("Snapshot or WAL directory don't exist, there is nothing to recover.");
return std::nullopt;
}
auto snapshot_files = GetSnapshotFiles(snapshot_directory);
@ -145,6 +167,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
RecoveredIndicesAndConstraints indices_constraints;
std::optional<uint64_t> snapshot_timestamp;
if (!snapshot_files.empty()) {
spdlog::info("Try recovering from snapshot directory {}.", snapshot_directory);
// Order the files by name
std::sort(snapshot_files.begin(), snapshot_files.end());
@ -157,13 +180,13 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
spdlog::warn("The snapshot file {} isn't related to the latest snapshot file!", path);
continue;
}
spdlog::info("Starting snapshot recovery from {}", path);
spdlog::info("Starting snapshot recovery from {}.", path);
try {
recovered_snapshot = LoadSnapshot(path, vertices, edges, epoch_history, name_id_mapper, edge_count, items);
spdlog::info("Snapshot recovery successful!");
break;
} catch (const RecoveryFailure &e) {
spdlog::warn("Couldn't recover snapshot from {} because of: {}", path, e.what());
spdlog::warn("Couldn't recover snapshot from {} because of: {}.", path, e.what());
continue;
}
}
@ -181,6 +204,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
return recovered_snapshot->recovery_info;
}
} else {
spdlog::info("No snapshot file was found, collecting information from WAL directory {}.", wal_directory);
std::error_code error_code;
if (!utils::DirExists(wal_directory)) return std::nullopt;
// We use this smaller struct that contains only a subset of information
@ -206,7 +230,10 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
}
}
MG_ASSERT(!error_code, "Couldn't recover data because an error occurred: {}!", error_code.message());
if (wal_files.empty()) return std::nullopt;
if (wal_files.empty()) {
spdlog::warn("No snapshot or WAL file found!");
return std::nullopt;
}
std::sort(wal_files.begin(), wal_files.end());
// UUID used for durability is the UUID of the last WAL file.
// Same for the epoch id.
@ -215,7 +242,10 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
}
auto maybe_wal_files = GetWalFiles(wal_directory, *uuid);
if (!maybe_wal_files) return std::nullopt;
if (!maybe_wal_files) {
spdlog::warn("Couldn't get WAL file info from the WAL directory!");
return std::nullopt;
}
// Array of all discovered WAL files, ordered by sequence number.
auto &wal_files = *maybe_wal_files;
@ -232,6 +262,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
"files that match the last WAL file!");
if (!wal_files.empty()) {
spdlog::info("Checking WAL files.");
{
const auto &first_wal = wal_files[0];
if (first_wal.seq_num != 0) {
@ -255,6 +286,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
}
std::optional<uint64_t> previous_seq_num;
auto last_loaded_timestamp = snapshot_timestamp;
spdlog::info("Trying to load WAL files.");
for (auto &wal_file : wal_files) {
if (previous_seq_num && (wal_file.seq_num - *previous_seq_num) > 1) {
LOG_FATAL("You are missing a WAL file with the sequence number {}!", *previous_seq_num + 1);
@ -290,6 +322,8 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
// The sequence number needs to be recovered even though `LoadWal` didn't
// load any deltas from that file.
*wal_seq_num = *previous_seq_num + 1;
spdlog::info("All necessary WAL files are loaded successfully.");
}
RecoverIndicesAndConstraints(indices_constraints, indices, constraints, vertices);

55
src/storage/v2/durability/snapshot.cpp
View File

@ -168,14 +168,15 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
});
// Read snapshot info.
auto info = ReadSnapshotInfo(path);
const auto info = ReadSnapshotInfo(path);
spdlog::info("Recovering {} vertices and {} edges.", info.vertices_count, info.edges_count);
// Check for edges.
bool snapshot_has_edges = info.offset_edges != 0;
// Recover mapper.
std::unordered_map<uint64_t, uint64_t> snapshot_id_map;
{
spdlog::info("Recovering mapper metadata.");
if (!snapshot.SetPosition(info.offset_mapper)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto marker = snapshot.ReadMarker();
@ -191,6 +192,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
if (!name) throw RecoveryFailure("Invalid snapshot data!");
auto my_id = name_id_mapper->NameToId(*name);
snapshot_id_map.emplace(*id, my_id);
SPDLOG_TRACE("Mapping \"{}\"from snapshot id {} to actual id {}.", *name, *id, my_id);
}
}
auto get_label_from_id = [&snapshot_id_map](uint64_t snapshot_id) {
@ -217,10 +219,11 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
auto edge_acc = edges->access();
uint64_t last_edge_gid = 0;
if (snapshot_has_edges) {
spdlog::info("Recovering {} edges.", info.edges_count);
if (!snapshot.SetPosition(info.offset_edges)) throw RecoveryFailure("Couldn't read data from snapshot!");
for (uint64_t i = 0; i < info.edges_count; ++i) {
{
auto marker = snapshot.ReadMarker();
const auto marker = snapshot.ReadMarker();
if (!marker || *marker != Marker::SECTION_EDGE) throw RecoveryFailure("Invalid snapshot data!");
}
@ -230,6 +233,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
if (!gid) throw RecoveryFailure("Invalid snapshot data!");
if (i > 0 && *gid <= last_edge_gid) throw RecoveryFailure("Invalid snapshot data!");
last_edge_gid = *gid;
spdlog::debug("Recovering edge {} with properties.", *gid);
auto [it, inserted] = edge_acc.insert(Edge{Gid::FromUint(*gid), nullptr});
if (!inserted) throw RecoveryFailure("The edge must be inserted here!");
@ -243,6 +247,8 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
if (!key) throw RecoveryFailure("Invalid snapshot data!");
auto value = snapshot.ReadPropertyValue();
if (!value) throw RecoveryFailure("Invalid snapshot data!");
SPDLOG_TRACE("Recovered property \"{}\" with value \"{}\" for edge {}.",
name_id_mapper->IdToName(snapshot_id_map.at(*key)), *value, *gid);
props.SetProperty(get_property_from_id(*key), *value);
}
}
@ -253,6 +259,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
if (i > 0 && *gid <= last_edge_gid) throw RecoveryFailure("Invalid snapshot data!");
last_edge_gid = *gid;
spdlog::debug("Ensuring edge {} doesn't have any properties.", *gid);
// Read properties.
{
auto props_size = snapshot.ReadUint();
@ -264,12 +271,14 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
}
}
}
spdlog::info("Edges are recovered.");
}
// Recover vertices (labels and properties).
if (!snapshot.SetPosition(info.offset_vertices)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto vertex_acc = vertices->access();
uint64_t last_vertex_gid = 0;
spdlog::info("Recovering {} vertices.", info.vertices_count);
for (uint64_t i = 0; i < info.vertices_count; ++i) {
{
auto marker = snapshot.ReadMarker();
@ -283,10 +292,12 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
throw RecoveryFailure("Invalid snapshot data!");
}
last_vertex_gid = *gid;
spdlog::debug("Recovering vertex {}.", *gid);
auto [it, inserted] = vertex_acc.insert(Vertex{Gid::FromUint(*gid), nullptr});
if (!inserted) throw RecoveryFailure("The vertex must be inserted here!");
// Recover labels.
spdlog::trace("Recovering labels for vertex {}.", *gid);
{
auto labels_size = snapshot.ReadUint();
if (!labels_size) throw RecoveryFailure("Invalid snapshot data!");
@ -295,11 +306,14 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
for (uint64_t j = 0; j < *labels_size; ++j) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
SPDLOG_TRACE("Recovered label \"{}\" for vertex {}.", name_id_mapper->IdToName(snapshot_id_map.at(*label)),
*gid);
labels.emplace_back(get_label_from_id(*label));
}
}
// Recover properties.
spdlog::trace("Recovering properties for vertex {}.", *gid);
{
auto props_size = snapshot.ReadUint();
if (!props_size) throw RecoveryFailure("Invalid snapshot data!");
@ -309,6 +323,8 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
if (!key) throw RecoveryFailure("Invalid snapshot data!");
auto value = snapshot.ReadPropertyValue();
if (!value) throw RecoveryFailure("Invalid snapshot data!");
SPDLOG_TRACE("Recovered property \"{}\" with value \"{}\" for vertex {}.",
name_id_mapper->IdToName(snapshot_id_map.at(*key)), *value, *gid);
props.SetProperty(get_property_from_id(*key), *value);
}
}
@ -339,8 +355,10 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
if (!edge_type) throw RecoveryFailure("Invalid snapshot data!");
}
}
spdlog::info("Vertices are recovered.");
// Recover vertices (in/out edges).
spdlog::info("Recovering connectivity.");
if (!snapshot.SetPosition(info.offset_vertices)) throw RecoveryFailure("Couldn't read data from snapshot!");
for (auto &vertex : vertex_acc) {
{
@ -348,6 +366,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
if (!marker || *marker != Marker::SECTION_VERTEX) throw RecoveryFailure("Invalid snapshot data!");
}
spdlog::trace("Recovering connectivity for vertex {}.", vertex.gid.AsUint());
// Check vertex.
auto gid = snapshot.ReadUint();
if (!gid) throw RecoveryFailure("Invalid snapshot data!");
@ -377,6 +396,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
// Recover in edges.
{
spdlog::trace("Recovering inbound edges for vertex {}.", vertex.gid.AsUint());
auto in_size = snapshot.ReadUint();
if (!in_size) throw RecoveryFailure("Invalid snapshot data!");
vertex.in_edges.reserve(*in_size);
@ -404,12 +424,15 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
edge_ref = EdgeRef(&*edge);
}
}
SPDLOG_TRACE("Recovered inbound edge {} with label \"{}\" from vertex {}.", *edge_gid,
name_id_mapper->IdToName(snapshot_id_map.at(*edge_type)), from_vertex->gid.AsUint());
vertex.in_edges.emplace_back(get_edge_type_from_id(*edge_type), &*from_vertex, edge_ref);
}
}
// Recover out edges.
{
spdlog::trace("Recovering outbound edges for vertex {}.", vertex.gid.AsUint());
auto out_size = snapshot.ReadUint();
if (!out_size) throw RecoveryFailure("Invalid snapshot data!");
vertex.out_edges.reserve(*out_size);
@ -437,6 +460,8 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
edge_ref = EdgeRef(&*edge);
}
}
SPDLOG_TRACE("Recovered outbound edge {} with label \"{}\" to vertex {}.", *edge_gid,
name_id_mapper->IdToName(snapshot_id_map.at(*edge_type)), to_vertex->gid.AsUint());
vertex.out_edges.emplace_back(get_edge_type_from_id(*edge_type), &*to_vertex, edge_ref);
}
// Increment edge count. We only increment the count here because the
@ -444,6 +469,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
edge_count->fetch_add(*out_size, std::memory_order_acq_rel);
}
}
spdlog::info("Connectivity is recovered.");
// Set initial values for edge/vertex ID generators.
ret.next_edge_id = last_edge_gid + 1;
@ -452,6 +478,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
// Recover indices.
{
spdlog::info("Recovering metadata of indices.");
if (!snapshot.SetPosition(info.offset_indices)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto marker = snapshot.ReadMarker();
@ -461,18 +488,22 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
{
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
spdlog::info("Recovering metadata of {} label indices.", *size);
for (uint64_t i = 0; i < *size; ++i) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
AddRecoveredIndexConstraint(&indices_constraints.indices.label, get_label_from_id(*label),
"The label index already exists!");
SPDLOG_TRACE("Recovered metadata of label index for :{}", name_id_mapper->IdToName(snapshot_id_map.at(*label)));
}
spdlog::info("Metadata of label indices are recovered.");
}
// Recover label+property indices.
{
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
spdlog::info("Recovering metadata of {} label+property indices.", *size);
for (uint64_t i = 0; i < *size; ++i) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
@ -481,12 +512,18 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
AddRecoveredIndexConstraint(&indices_constraints.indices.label_property,
{get_label_from_id(*label), get_property_from_id(*property)},
"The label+property index already exists!");
SPDLOG_TRACE("Recovered metadata of label+property index for :{}({})",
name_id_mapper->IdToName(snapshot_id_map.at(*label)),
name_id_mapper->IdToName(snapshot_id_map.at(*property)));
}
spdlog::info("Metadata of label+property indices are recovered.");
}
spdlog::info("Metadata of indices are recovered.");
}
// Recover constraints.
{
spdlog::info("Recovering metadata of constraints.");
if (!snapshot.SetPosition(info.offset_constraints)) throw RecoveryFailure("Couldn't read data from snapshot!");
auto marker = snapshot.ReadMarker();
@ -496,6 +533,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
{
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
spdlog::info("Recovering metadata of {} existence constraints.", *size);
for (uint64_t i = 0; i < *size; ++i) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
@ -504,7 +542,11 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
AddRecoveredIndexConstraint(&indices_constraints.constraints.existence,
{get_label_from_id(*label), get_property_from_id(*property)},
"The existence constraint already exists!");
SPDLOG_TRACE("Recovered metadata of existence constraint for :{}({})",
name_id_mapper->IdToName(snapshot_id_map.at(*label)),
name_id_mapper->IdToName(snapshot_id_map.at(*property)));
}
spdlog::info("Metadata of existence constraints are recovered.");
}
// Recover unique constraints.
@ -513,6 +555,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
if (*version >= kUniqueConstraintVersion) {
auto size = snapshot.ReadUint();
if (!size) throw RecoveryFailure("Invalid snapshot data!");
spdlog::info("Recovering metadata of {} unique constraints.", *size);
for (uint64_t i = 0; i < *size; ++i) {
auto label = snapshot.ReadUint();
if (!label) throw RecoveryFailure("Invalid snapshot data!");
@ -526,10 +569,15 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
}
AddRecoveredIndexConstraint(&indices_constraints.constraints.unique, {get_label_from_id(*label), properties},
"The unique constraint already exists!");
SPDLOG_TRACE("Recovered metadata of unique constraints for :{}",
name_id_mapper->IdToName(snapshot_id_map.at(*label)));
}
spdlog::info("Metadata of unique constraints are recovered.");
}
spdlog::info("Metadata of constraints are recovered.");
}
spdlog::info("Recovering metadata.");
// Recover epoch history
{
if (!snapshot.SetPosition(info.offset_epoch_history)) throw RecoveryFailure("Couldn't read data from snapshot!");
@ -555,6 +603,7 @@ RecoveredSnapshot LoadSnapshot(const std::filesystem::path &path, utils::SkipLis
}
}
spdlog::info("Metadata recovered.");
// Recover timestamp.
ret.next_timestamp = info.start_timestamp + 1;

10
src/storage/v2/durability/wal.cpp
View File

@ -610,6 +610,7 @@ RecoveryInfo LoadWal(const std::filesystem::path &path, RecoveredIndicesAndConst
const std::optional<uint64_t> last_loaded_timestamp, utils::SkipList<Vertex> *vertices,
utils::SkipList<Edge> *edges, NameIdMapper *name_id_mapper, std::atomic<uint64_t> *edge_count,
Config::Items items) {
spdlog::info("Trying to load WAL file {}.", path);
RecoveryInfo ret;
Decoder wal;
@ -622,13 +623,17 @@ RecoveryInfo LoadWal(const std::filesystem::path &path, RecoveredIndicesAndConst
ret.last_commit_timestamp = info.to_timestamp;
// Check timestamp.
if (last_loaded_timestamp && info.to_timestamp <= *last_loaded_timestamp) return ret;
if (last_loaded_timestamp && info.to_timestamp <= *last_loaded_timestamp) {
spdlog::info("Skip loading WAL file because it is too old.");
return ret;
}
// Recover deltas.
wal.SetPosition(info.offset_deltas);
uint64_t deltas_applied = 0;
auto edge_acc = edges->access();
auto vertex_acc = vertices->access();
spdlog::info("WAL file contains {} deltas.", info.num_deltas);
for (uint64_t i = 0; i < info.num_deltas; ++i) {
// Read WAL delta header to find out the delta timestamp.
auto timestamp = ReadWalDeltaHeader(&wal);
@ -839,7 +844,8 @@ RecoveryInfo LoadWal(const std::filesystem::path &path, RecoveredIndicesAndConst
}
}
spdlog::info("Applied {} deltas from WAL", deltas_applied, path);
spdlog::info("Applied {} deltas from WAL. Skipped {} deltas, because they were too old.", deltas_applied,
info.num_deltas - deltas_applied);
return ret;
}

2
src/utils/logging.hpp
View File

@ -65,7 +65,7 @@ void Fatal(const char *msg, const Args &...msg_args) {
do { \
spdlog::critical(__VA_ARGS__); \
std::terminate(); \
} while (0);
} while (0)
#ifndef NDEBUG
#define DLOG_FATAL(...) LOG_FATAL(__VA_ARGS__)

2
tests/e2e/replication/constraints.cpp
View File

@ -39,7 +39,7 @@ int main(int argc, char **argv) {
const auto label_name = (*data)[0][1].ValueString();
const auto property_name = (*data)[0][2].ValueList()[0].ValueString();
if (label_name != "Node" || property_name != "id") {
LOG_FATAL("{} does NOT hava valid constraint created.", database_endpoint)
LOG_FATAL("{} does NOT hava valid constraint created.", database_endpoint);
}
} else {
LOG_FATAL("Unable to get CONSTRAINT INFO from {}", database_endpoint);