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

Add recovery and GraphDbAccessor interface for UniqueLabelPropertyConstraint

Browse Source 
Reviewers: msantl, ipaljak

Reviewed By: msantl

Subscribers: mferencevic, pullbot

Differential Revision: https://phabricator.memgraph.io/D1925
This commit is contained in:
Vinko Kasljevic 2019-03-27 11:27:24 +01:00
parent 49b552408b
commit 21bd8b492f
22 changed files with 554 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@ -5,6 +5,8 @@
### Breaking Changes
* `indexInfo()` function replaced with `SHOW INDEX INFO` syntax.
* Write-ahead log format changed (not backward compatible).
* Snapshot format changed (not backward compatible).
### Major Features and Improvements

2
src/database/single_node/graph_db.cpp
View File

@ -43,6 +43,8 @@ GraphDb::GraphDb(Config config) : config_(config) {
durability::RecoverIndexes(this, recovery_data.indexes);
durability::RecoverExistenceConstraints(
this, recovery_data.existence_constraints);
durability::RecoverUniqueConstraints(
this, recovery_data.unique_constraints);
}
}

105
src/database/single_node/graph_db_accessor.cpp
View File

@ -182,12 +182,88 @@ void GraphDbAccessor::DeleteIndex(storage::Label label,
}
}
void GraphDbAccessor::BuildUniqueConstraint(storage::Label label,
storage::Property property) {
try {
auto dba =
db_.AccessBlocking(std::experimental::make_optional(transaction().id_));
if (!db_.storage().unique_label_property_constraints_.AddConstraint(
label, property, dba->transaction())) {
// Already exists
return;
}
for (auto v : dba->Vertices(false)) {
if (std::find(v.labels().begin(), v.labels().end(), label) !=
v.labels().end()) {
db_.storage().unique_label_property_constraints_.Update(
v, dba->transaction());
}
}
dba->wal().Emplace(database::StateDelta::BuildUniqueConstraint(
dba->transaction().id_, label, dba->LabelName(label),
std::vector<storage::Property>{property},
std::vector<std::string>{dba->PropertyName(property)}));
dba->Commit();
} catch (const IndexConstraintViolationException &) {
db_.storage().unique_label_property_constraints_.RemoveConstraint(label,
property);
throw IndexConstraintViolationException(
"Constraint cannot be built due to existing unique constraint "
"violation!");
} catch (const tx::TransactionEngineError &e) {
db_.storage().unique_label_property_constraints_.RemoveConstraint(label,
property);
throw IndexTransactionException(e.what());
} catch (...) {
db_.storage().unique_label_property_constraints_.RemoveConstraint(label,
property);
throw;
}
}
void GraphDbAccessor::DeleteUniqueConstraint(storage::Label label,
storage::Property property) {
try {
auto dba =
db_.AccessBlocking(std::experimental::make_optional(transaction().id_));
if (!db_.storage().unique_label_property_constraints_.RemoveConstraint(
label, property)) {
// Nothing to do
return;
}
dba->wal().Emplace(database::StateDelta::DropUniqueConstraint(
dba->transaction().id_, label, dba->LabelName(label),
std::vector<storage::Property>{property},
std::vector<std::string>{dba->PropertyName(property)}));
dba->Commit();
} catch (const tx::TransactionEngineError &e) {
throw IndexTransactionException(e.what());
}
}
bool GraphDbAccessor::UniqueConstraintExists(storage::Label label,
storage::Property property) const {
return db_.storage().unique_label_property_constraints_.Exists(label,
property);
}
std::vector<storage::constraints::LabelProperty>
GraphDbAccessor::ListUniqueLabelPropertyConstraints() const {
return db_.storage().unique_label_property_constraints_.ListConstraints();
}
void GraphDbAccessor::UpdateLabelIndices(storage::Label label,
const VertexAccessor &vertex_accessor,
const Vertex *vertex) {
DCHECK(!commited_ && !aborted_) << "Accessor committed or aborted";
auto *vlist_ptr = vertex_accessor.address();
db_.storage().unique_label_property_constraints_.UpdateOnAddLabel(
label, vertex_accessor, transaction());
if (!db_.storage().label_property_index_.UpdateOnLabel(label, vlist_ptr,
vertex)) {
throw IndexConstraintViolationException(
@ -202,10 +278,20 @@ void GraphDbAccessor::UpdateLabelIndices(storage::Label label,
db_.storage().labels_index_.Update(label, vlist_ptr, vertex);
}
void GraphDbAccessor::UpdateOnRemoveLabel(
storage::Label label, const RecordAccessor<Vertex> &accessor) {
db_.storage().unique_label_property_constraints_.UpdateOnRemoveLabel(
label, accessor, transaction());
}
void GraphDbAccessor::UpdatePropertyIndex(
storage::Property property, const RecordAccessor<Vertex> &vertex_accessor,
const Vertex *vertex) {
storage::Property property, const PropertyValue &value,
const RecordAccessor<Vertex> &vertex_accessor, const Vertex *vertex) {
DCHECK(!commited_ && !aborted_) << "Accessor committed or aborted";
db_.storage().unique_label_property_constraints_.UpdateOnAddProperty(
property, value, vertex_accessor, transaction());
if (!db_.storage().label_property_index_.UpdateOnProperty(
property, vertex_accessor.address(), vertex)) {
throw IndexConstraintViolationException(
@ -213,9 +299,12 @@ void GraphDbAccessor::UpdatePropertyIndex(
}
}
void GraphDbAccessor::UpdateOnPropertyRemove(storage::Property property,
const Vertex *vertex,
const RecordAccessor<Vertex> &accessor) {
void GraphDbAccessor::UpdateOnPropertyRemove(
storage::Property property, const Vertex *vertex,
const RecordAccessor<Vertex> &accessor) {
db_.storage().unique_label_property_constraints_.UpdateOnRemoveProperty(
property, accessor, transaction());
if (!db_.storage().existence_constraints_.CheckOnRemoveProperty(vertex,
property)) {
throw IndexConstraintViolationException(
@ -237,7 +326,8 @@ void GraphDbAccessor::BuildExistenceConstraint(
}
}
db_.storage().existence_constraints_.AddConstraint(rule);
if (!db_.storage().existence_constraints_.AddConstraint(rule)) return;
std::vector<std::string> property_names(properties.size());
std::transform(properties.begin(), properties.end(), property_names.begin(),
[&dba](auto p) { return dba->PropertyName(p); });
@ -253,7 +343,8 @@ void GraphDbAccessor::DeleteExistenceConstraint(
auto dba =
db_.AccessBlocking(std::experimental::make_optional(transaction().id_));
ExistenceRule rule{label, properties};
db_.storage().existence_constraints_.RemoveConstraint(rule);
if (!db_.storage().existence_constraints_.RemoveConstraint(rule)) return;
std::vector<std::string> property_names(properties.size());
std::transform(properties.begin(), properties.end(), property_names.begin(),
[&dba](auto p) { return dba->PropertyName(p); });

34
src/database/single_node/graph_db_accessor.hpp
View File

@ -446,6 +446,33 @@ class GraphDbAccessor {
/// Writes Index (key) creation to wal, marks it as ready for usage
void EnableIndex(const LabelPropertyIndex::Key &key);
/**
* Creates new unique constraint that consists of label and property.
* If the constraint already exists, this method does nothing.
*
* @throws IndexConstraintViolationException if constraint couldn't be build
* due to existing constraint violation.
*/
void BuildUniqueConstraint(storage::Label label, storage::Property property);
/**
* Deletes existing unique constraint.
* If the constraint doesn't exist, this method does nothing.
*/
void DeleteUniqueConstraint(storage::Label label, storage::Property property);
/**
* Checks if unique constraint exists.
*/
bool UniqueConstraintExists(storage::Label label,
storage::Property property) const;
/**
* Returns a list of currently active unique constraints.
*/
std::vector<storage::constraints::LabelProperty>
ListUniqueLabelPropertyConstraints() const;
/**
* @brief - Returns true if the given label+property index already exists and
* is ready for use.
@ -659,6 +686,12 @@ class GraphDbAccessor {
bool commited_{false};
bool aborted_{false};
/**
* Notifies storage about change.
*/
void UpdateOnRemoveLabel(storage::Label label,
const RecordAccessor<Vertex> &accessor);
/**
* Notifies storage about change.
*/
@ -674,6 +707,7 @@ class GraphDbAccessor {
* @param vertex - vertex to insert
*/
void UpdatePropertyIndex(storage::Property property,
const PropertyValue &value,
const RecordAccessor<Vertex> &vertex_accessor,
const Vertex *const vertex);
};

93
src/durability/single_node/recovery.cpp
View File

@ -175,6 +175,31 @@ bool RecoverSnapshot(const fs::path &snapshot_file, database::GraphDb *db,
ExistenceConstraintRecoveryData{label, std::move(properties), true});
}
// Read a list of unique constraints
RETURN_IF_NOT(decoder.ReadValue(&dv, Value::Type::List));
auto unique_constraints = dv.ValueList();
for (auto it = unique_constraints.begin();
it != unique_constraints.end();) {
RETURN_IF_NOT(it->IsString());
auto label = it->ValueString();
++it;
RETURN_IF_NOT(it != unique_constraints.end());
RETURN_IF_NOT(it->IsInt());
auto prop_size = it->ValueInt();
++it;
std::vector<std::string> properties;
properties.reserve(prop_size);
for (size_t i = 0; i < prop_size; ++i) {
RETURN_IF_NOT(it != unique_constraints.end());
RETURN_IF_NOT(it->IsString());
properties.emplace_back(it->ValueString());
++it;
}
recovery_data->unique_constraints.emplace_back(
UniqueConstraintRecoveryData{label, std::move(properties), true});
}
auto dba = db->Access();
std::unordered_map<uint64_t, VertexAccessor> vertices;
for (int64_t i = 0; i < vertex_count; ++i) {
@ -509,8 +534,50 @@ void RecoverWal(const fs::path &durability_dir, database::GraphDb *db,
recovery_data->existence_constraints.erase(build_it);
} else {
recovery_data->existence_constraints.emplace_back(
ExistenceConstraintRecoveryData{
delta.label_name, delta.property_names, false});
ExistenceConstraintRecoveryData{delta.label_name,
delta.property_names, false});
}
break;
}
case database::StateDelta::Type::BUILD_UNIQUE_CONSTRAINT: {
auto drop_it = std::find_if(
recovery_data->unique_constraints.begin(),
recovery_data->unique_constraints.end(),
[&delta](const UniqueConstraintRecoveryData &data) {
return data.label == delta.label_name &&
std::is_permutation(data.properties.begin(),
data.properties.end(),
delta.property_names.begin()) &&
data.create == false;
});
if (drop_it != recovery_data->unique_constraints.end()) {
recovery_data->unique_constraints.erase(drop_it);
} else {
recovery_data->unique_constraints.emplace_back(
UniqueConstraintRecoveryData{delta.label_name,
delta.property_names, true});
}
break;
}
case database::StateDelta::Type::DROP_UNIQUE_CONSTRAINT: {
auto build_it = std::find_if(
recovery_data->unique_constraints.begin(),
recovery_data->unique_constraints.end(),
[&delta](const UniqueConstraintRecoveryData &data) {
return data.label == delta.label_name &&
std::is_permutation(data.properties.begin(),
data.properties.end(),
delta.property_names.begin()) &&
data.create == true;
});
if (build_it != recovery_data->unique_constraints.end()) {
recovery_data->unique_constraints.erase(build_it);
} else {
recovery_data->unique_constraints.emplace_back(
UniqueConstraintRecoveryData{delta.label_name,
delta.property_names, false});
}
break;
}
@ -562,4 +629,26 @@ void RecoverExistenceConstraints(
}
}
}
void RecoverUniqueConstraints(
database::GraphDb *db,
const std::vector<UniqueConstraintRecoveryData> &constraints) {
auto dba = db->Access();
for (auto &constraint : constraints) {
auto label = dba->Label(constraint.label);
std::vector<storage::Property> properties;
properties.reserve(constraint.properties.size());
for (auto &prop : constraint.properties) {
properties.push_back(dba->Property(prop));
}
DCHECK(properties.size() == 1)
<< "Unique constraint with multiple properties is not supported";
if (constraint.create) {
dba->BuildUniqueConstraint(label, properties[0]);
} else {
dba->DeleteUniqueConstraint(label, properties[0]);
}
}
}
} // namespace durability

14
src/durability/single_node/recovery.hpp
View File

@ -50,6 +50,13 @@ struct ExistenceConstraintRecoveryData {
bool create;
};
struct UniqueConstraintRecoveryData {
std::string label;
std::vector<std::string> properties;
// distinguish between creating and dropping unique constraint
bool create;
};
// A data structure for exchanging info between main recovery function and
// snapshot and WAL recovery functions.
struct RecoveryData {
@ -61,11 +68,14 @@ struct RecoveryData {
std::vector<IndexRecoveryData> indexes;
std::vector<ExistenceConstraintRecoveryData> existence_constraints;
std::vector<UniqueConstraintRecoveryData> unique_constraints;
void Clear() {
snapshooter_tx_id = 0;
snapshooter_tx_snapshot.clear();
indexes.clear();
existence_constraints.clear();
unique_constraints.clear();
}
};
@ -161,4 +171,8 @@ void RecoverExistenceConstraints(
database::GraphDb *db,
const std::vector<ExistenceConstraintRecoveryData> &constraints);
void RecoverUniqueConstraints(
database::GraphDb *db,
const std::vector<UniqueConstraintRecoveryData> &constraints);
} // namespace durability

14
src/durability/single_node/snapshooter.cpp
View File

@ -17,7 +17,7 @@ namespace fs = std::experimental::filesystem;
namespace durability {
// Snapshot layout is described in durability/version.hpp
static_assert(durability::kVersion == 8,
static_assert(durability::kVersion == 9,
"Wrong snapshot version, please update!");
namespace {
@ -69,6 +69,18 @@ bool Encode(const fs::path &snapshot_file, database::GraphDb &db,
encoder.WriteList(existence_constraints);
}
// Write unique constraints to snapshoot
{
std::vector<communication::bolt::Value> unique_constraints;
for (const auto &rule : dba.ListUniqueLabelPropertyConstraints()) {
unique_constraints.emplace_back(dba.LabelName(rule.label));
// UniqueLabelPropertyConstraint has label and single property rule
unique_constraints.emplace_back(1);
unique_constraints.emplace_back(dba.PropertyName(rule.property));
}
encoder.WriteList(unique_constraints);
}
for (const auto &vertex : dba.Vertices(false)) {
encoder.WriteVertex(glue::ToBoltVertex(vertex));
vertex_num++;

56
src/durability/single_node/state_delta.cpp
View File

@ -152,6 +152,32 @@ StateDelta StateDelta::DropExistenceConstraint(
return op;
}
StateDelta StateDelta::BuildUniqueConstraint(
tx::TransactionId tx_id, storage::Label label,
const std::string &label_name,
const std::vector<storage::Property> &properties,
const std::vector<std::string> &property_names) {
StateDelta op(StateDelta::Type::BUILD_UNIQUE_CONSTRAINT, tx_id);
op.label = label;
op.label_name = label_name;
op.properties = properties;
op.property_names = property_names;
return op;
}
StateDelta StateDelta::DropUniqueConstraint(
tx::TransactionId tx_id, storage::Label label,
const std::string &label_name,
const std::vector<storage::Property> &properties,
const std::vector<std::string> &property_names) {
StateDelta op(StateDelta::Type::DROP_UNIQUE_CONSTRAINT, tx_id);
op.label = label;
op.label_name = label_name;
op.properties = properties;
op.property_names = property_names;
return op;
}
void StateDelta::Encode(
HashedFileWriter &writer,
communication::bolt::BaseEncoder<HashedFileWriter> &encoder) const {
@ -211,6 +237,7 @@ void StateDelta::Encode(
encoder.WriteString(property_name);
break;
case Type::BUILD_EXISTENCE_CONSTRAINT:
case Type::BUILD_UNIQUE_CONSTRAINT:
encoder.WriteInt(label.Id());
encoder.WriteString(label_name);
encoder.WriteInt(properties.size());
@ -222,6 +249,7 @@ void StateDelta::Encode(
}
break;
case Type::DROP_EXISTENCE_CONSTRAINT:
case Type::DROP_UNIQUE_CONSTRAINT:
encoder.WriteInt(label.Id());
encoder.WriteString(label_name);
encoder.WriteInt(properties.size());
@ -313,7 +341,8 @@ std::experimental::optional<StateDelta> StateDelta::Decode(
DECODE_MEMBER_CAST(property, ValueInt, storage::Property)
DECODE_MEMBER(property_name, ValueString)
break;
case Type::BUILD_EXISTENCE_CONSTRAINT: {
case Type::BUILD_EXISTENCE_CONSTRAINT:
case Type::BUILD_UNIQUE_CONSTRAINT: {
DECODE_MEMBER_CAST(label, ValueInt, storage::Label)
DECODE_MEMBER(label_name, ValueString)
if (!decoder.ReadValue(&dv)) return nullopt;
@ -329,7 +358,8 @@ std::experimental::optional<StateDelta> StateDelta::Decode(
}
break;
}
case Type::DROP_EXISTENCE_CONSTRAINT: {
case Type::DROP_EXISTENCE_CONSTRAINT:
case Type::DROP_UNIQUE_CONSTRAINT: {
DECODE_MEMBER_CAST(label, ValueInt, storage::Label)
DECODE_MEMBER(label_name, ValueString)
if (!decoder.ReadValue(&dv)) return nullopt;
@ -432,6 +462,28 @@ void StateDelta::Apply(GraphDbAccessor &dba) const {
dba.DeleteExistenceConstraint(dba.Label(label_name), properties);
} break;
case Type::BUILD_UNIQUE_CONSTRAINT: {
std::vector<storage::Property> properties;
properties.reserve(property_names.size());
for (auto &p : property_names) {
properties.push_back(dba.Property(p));
}
DCHECK(properties.size() == 1)
<< "Unique constraint with multiple properties is not supported";
dba.BuildUniqueConstraint(dba.Label(label_name), properties[0]);
} break;
case Type::DROP_UNIQUE_CONSTRAINT: {
std::vector<storage::Property> properties;
properties.reserve(property_names.size());
for (auto &p : property_names) {
properties.push_back(dba.Property(p));
}
DCHECK(properties.size() == 1)
<< "Unique constraint with multiple properties is not supported";
dba.DeleteUniqueConstraint(dba.Label(label_name), properties[0]);
} break;
}
}

12
src/durability/single_node/state_delta.lcp
View File

@ -75,6 +75,8 @@ in StateDeltas.")
drop-index ;; label, label_name, property, property_name
build-existence-constraint ;; label, label_name, properties, property_names
drop-existence-constraint ;; label, label_name, properties, property_names
build-unique_constraint ;; label, label_name, properties, property_names
drop-unique_constraint ;; label, label_name, properties, property_names
)
(:documentation
"Defines StateDelta type. For each type the comment indicates which values
@ -143,6 +145,16 @@ omitted in the comment."))
const std::string &label_name,
const std::vector<storage::Property> &property,
const std::vector<std::string> &property_names);
static StateDelta BuildUniqueConstraint(
tx::TransactionId tx_id, storage::Label label,
const std::string &label_name,
const std::vector<storage::Property> &properties,
const std::vector<std::string> &property_names);
static StateDelta DropUniqueConstraint(
tx::TransactionId tx_id, storage::Label label,
const std::string &label_name,
const std::vector<storage::Property> &property,
const std::vector<std::string> &property_names);
/// Applies CRUD delta to database accessor. Fails on other types of deltas
void Apply(GraphDbAccessor &dba) const;

10
src/durability/single_node/version.hpp
View File

@ -15,7 +15,7 @@ constexpr std::array<uint8_t, 4> kSnapshotMagic{{'M', 'G', 's', 'n'}};
constexpr std::array<uint8_t, 4> kWalMagic{{'M', 'G', 'w', 'l'}};
// The current default version of snapshot and WAL encoding / decoding.
constexpr int64_t kVersion{8};
constexpr int64_t kVersion{9};
// Snapshot format (version 8):
// 1) Magic number + snapshot version
@ -29,13 +29,15 @@ constexpr int64_t kVersion{8};
//
// 5) A list of existence constraints
//
// 6) Bolt encoded nodes. Each node is written in the following format:
// 6) A list of unique constraints
//
// 7) Bolt encoded nodes. Each node is written in the following format:
// * gid, labels, properties
// 7) Bolt encoded edges. Each edge is written in the following format:
// 8) Bolt encoded edges. Each edge is written in the following format:
// * gid
// * from, to
// * edge_type
// * properties
//
// 8) Snapshot summary (number of nodes, number of edges, hash)
// 9) Snapshot summary (number of nodes, number of edges, hash)
} // namespace durability

2
src/durability/single_node/wal.cpp
View File

@ -149,6 +149,8 @@ bool WriteAheadLog::IsStateDeltaTransactionEnd(
case database::StateDelta::Type::DROP_INDEX:
case database::StateDelta::Type::BUILD_EXISTENCE_CONSTRAINT:
case database::StateDelta::Type::DROP_EXISTENCE_CONSTRAINT:
case database::StateDelta::Type::BUILD_UNIQUE_CONSTRAINT:
case database::StateDelta::Type::DROP_UNIQUE_CONSTRAINT:
return false;
}
}

10
src/storage/single_node/constraints/existence_constraints.cpp
View File

@ -20,19 +20,23 @@ bool CheckIfSatisfiesExistenceRule(const Vertex *vertex,
return true;
}
void ExistenceConstraints::AddConstraint(const ExistenceRule &rule) {
bool ExistenceConstraints::AddConstraint(const ExistenceRule &rule) {
auto found = std::find(constraints_.begin(), constraints_.end(), rule);
if (found != constraints_.end()) return;
if (found != constraints_.end()) return false;
constraints_.push_back(rule);
return true;
}
void ExistenceConstraints::RemoveConstraint(const ExistenceRule &rule) {
bool ExistenceConstraints::RemoveConstraint(const ExistenceRule &rule) {
auto found = std::find(constraints_.begin(), constraints_.end(), rule);
if (found != constraints_.end()) {
std::swap(*found, constraints_.back());
constraints_.pop_back();
return true;
}
return false;
}
bool ExistenceConstraints::Exists(const ExistenceRule &rule) const {

8
src/storage/single_node/constraints/existence_constraints.hpp
View File

@ -55,12 +55,16 @@ class ExistenceConstraints {
/// nothing. This method doesn't check if any of the existing vertices breaks
/// this constraint. Caller must do that instead. Caller must also ensure
/// that no other transaction is running in parallel.
void AddConstraint(const ExistenceRule &rule);
///
/// @returns true if constraint was added, false otherwise
bool AddConstraint(const ExistenceRule &rule);
/// Removes existing constraint, if the constraint doesn't exist this method
/// does nothing. Caller must ensure that no other transaction is running in
/// parallel.
void RemoveConstraint(const ExistenceRule &rule);
///
/// @returns true if constraint was removed, false otherwise
bool RemoveConstraint(const ExistenceRule &rule);
/// Checks whether given constraint is visible.
bool Exists(const ExistenceRule &rule) const;

62
src/storage/single_node/constraints/unique_label_property_constraint.cpp
View File

@ -1,6 +1,7 @@
#include "storage/single_node/constraints/unique_label_property_constraint.hpp"
#include "storage/single_node/vertex_accessor.hpp"
#include "storage/single_node/record_accessor.hpp"
#include "storage/single_node/vertex.hpp"
#include "utils/algorithm.hpp"
namespace storage::constraints {
@ -12,17 +13,27 @@ auto FindIn(storage::Label label, storage::Property property,
});
}
void UniqueLabelPropertyConstraint::AddConstraint(storage::Label label,
bool UniqueLabelPropertyConstraint::AddConstraint(storage::Label label,
storage::Property property,
const tx::Transaction &t) {
auto found = FindIn(label, property, constraints_);
if (found == constraints_.end()) constraints_.emplace_back(label, property);
if (found == constraints_.end()) {
constraints_.emplace_back(label, property);
return true;
} else {
return false;
}
}
void UniqueLabelPropertyConstraint::RemoveConstraint(
bool UniqueLabelPropertyConstraint::RemoveConstraint(
storage::Label label, storage::Property property) {
auto found = FindIn(label, property, constraints_);
if (found != constraints_.end()) constraints_.erase(found);
if (found != constraints_.end()) {
constraints_.erase(found);
return true;
} else {
return false;
}
}
bool UniqueLabelPropertyConstraint::Exists(storage::Label label,
@ -41,8 +52,32 @@ std::vector<LabelProperty> UniqueLabelPropertyConstraint::ListConstraints()
return constraints;
}
void UniqueLabelPropertyConstraint::Update(
const RecordAccessor<Vertex> &accessor, const tx::Transaction &t) {
auto &vertex = accessor.current();
std::lock_guard<std::mutex> guard(lock_);
for (auto &constraint : constraints_) {
bool contains_label = utils::Contains(vertex.labels_, constraint.label);
auto value = vertex.properties_.at(constraint.property);
if (contains_label && !value.IsNull()) {
bool found = false;
for (auto &p : constraint.version_pairs) {
if (p.value == value) {
p.record.Insert(accessor.gid(), t);
found = true;
break;
}
}
if (!found) {
constraint.version_pairs.emplace_back(accessor.gid(), value, t);
}
}
}
}
void UniqueLabelPropertyConstraint::UpdateOnAddLabel(
storage::Label label, const VertexAccessor &accessor,
storage::Label label, const RecordAccessor<Vertex> &accessor,
const tx::Transaction &t) {
auto &vertex = accessor.current();
std::lock_guard<std::mutex> guard(lock_);
@ -65,7 +100,7 @@ void UniqueLabelPropertyConstraint::UpdateOnAddLabel(
}
}
void UniqueLabelPropertyConstraint::UpdateOnRemoveLabel(
storage::Label label, const VertexAccessor &accessor,
storage::Label label, const RecordAccessor<Vertex> &accessor,
const tx::Transaction &t) {
auto &vertex = accessor.current();
std::lock_guard<std::mutex> guard(lock_);
@ -84,7 +119,7 @@ void UniqueLabelPropertyConstraint::UpdateOnRemoveLabel(
void UniqueLabelPropertyConstraint::UpdateOnAddProperty(
storage::Property property, const PropertyValue &value,
const VertexAccessor &accessor, const tx::Transaction &t) {
const RecordAccessor<Vertex> &accessor, const tx::Transaction &t) {
auto &vertex = accessor.current();
std::lock_guard<std::mutex> guard(lock_);
for (auto &constraint : constraints_) {
@ -107,16 +142,17 @@ void UniqueLabelPropertyConstraint::UpdateOnAddProperty(
}
void UniqueLabelPropertyConstraint::UpdateOnRemoveProperty(
storage::Property property, const PropertyValue &value,
const VertexAccessor &accessor, const tx::Transaction &t) {
storage::Property property, const RecordAccessor<Vertex> &accessor,
const tx::Transaction &t) {
auto &vertex = accessor.current();
auto gid = accessor.gid();
std::lock_guard<std::mutex> guard(lock_);
for (auto &constraint : constraints_) {
if (constraint.property == property &&
utils::Contains(vertex.labels_, constraint.label)) {
for (auto &p : constraint.version_pairs) {
if (p.value == value) {
p.record.Remove(accessor.gid(), t);
if (p.record.curr_gid == gid && p.record.tx_id_exp == 0) {
p.record.Remove(gid, t);
break;
}
}
@ -133,7 +169,7 @@ void UniqueLabelPropertyConstraint::Refresh(const tx::Snapshot &snapshot,
auto exp_id = p->record.tx_id_exp;
auto cre_id = p->record.tx_id_cre;
if ((exp_id != 0 && exp_id < snapshot.back() &&
engine.Info(exp_id).is_committed() && !snapshot.contains(exp_id)) ||
engine.Info(exp_id).is_committed() && !snapshot.contains(exp_id)) ||
(cre_id < snapshot.back() && engine.Info(cre_id).is_aborted())) {
constraint.version_pairs.erase(p);
}

26
src/storage/single_node/constraints/unique_label_property_constraint.hpp
View File

@ -13,7 +13,10 @@ namespace tx {
class Snapshot;
};
class VertexAccessor;
class Vertex;
template <typename TRecord>
class RecordAccessor;
namespace storage::constraints {
namespace impl {
@ -73,13 +76,13 @@ class UniqueLabelPropertyConstraint {
/// nothing. This method doesn't check if any of the existing vertices breaks
/// this constraint. Caller must do that instead. Caller must also ensure that
/// no other transaction is running in parallel.
void AddConstraint(storage::Label label, storage::Property property,
bool AddConstraint(storage::Label label, storage::Property property,
const tx::Transaction &t);
/// Removes existing unique constraint, if the constraint doesn't exist this
/// method does nothing. Caller must ensure that no other transaction is
/// running in parallel.
void RemoveConstraint(storage::Label label, storage::Property property);
bool RemoveConstraint(storage::Label label, storage::Property property);
/// Checks whether given unique constraint is visible.
bool Exists(storage::Label label, storage::Property property) const;
@ -91,13 +94,21 @@ class UniqueLabelPropertyConstraint {
///
/// @throws IndexConstraintViolationException
/// @throws SerializationError
void UpdateOnAddLabel(storage::Label label, const VertexAccessor &accessor,
void Update(const RecordAccessor<Vertex> &accessor, const tx::Transaction &t);
/// Updates unique constraint versions.
///
/// @throws IndexConstraintViolationException
/// @throws SerializationError
void UpdateOnAddLabel(storage::Label label,
const RecordAccessor<Vertex> &accessor,
const tx::Transaction &t);
/// Updates unique constraint versions.
///
/// @throws SerializationError
void UpdateOnRemoveLabel(storage::Label label, const VertexAccessor &accessor,
void UpdateOnRemoveLabel(storage::Label label,
const RecordAccessor<Vertex> &accessor,
const tx::Transaction &t);
/// Updates unique constraint versions.
@ -106,15 +117,14 @@ class UniqueLabelPropertyConstraint {
/// @throws SerializationError
void UpdateOnAddProperty(storage::Property property,
const PropertyValue &value,
const VertexAccessor &accessor,
const RecordAccessor<Vertex> &accessor,
const tx::Transaction &t);
/// Updates unique constraint versions.
///
/// @throws SerializationError
void UpdateOnRemoveProperty(storage::Property property,
const PropertyValue &value,
const VertexAccessor &accessor,
const RecordAccessor<Vertex> &accessor,
const tx::Transaction &t);
/// Removes records that are no longer visible.

2
src/storage/single_node/record_accessor.cpp
View File

@ -26,7 +26,7 @@ void RecordAccessor<Vertex>::PropsSet(storage::Property key,
auto delta = StateDelta::PropsSetVertex(dba.transaction_id(), gid(), key,
dba.PropertyName(key), value);
update().properties_.set(key, value);
dba.UpdatePropertyIndex(key, *this, &update());
dba.UpdatePropertyIndex(key, value, *this, &update());
db_accessor().wal().Emplace(delta);
}

11
src/storage/single_node/storage.hpp
View File

@ -4,13 +4,14 @@
#include <experimental/optional>
#include "data_structures/concurrent/concurrent_map.hpp"
#include "storage/single_node/mvcc/version_list.hpp"
#include "storage/common/types/types.hpp"
#include "storage/common/kvstore/kvstore.hpp"
#include "storage/single_node/edge.hpp"
#include "storage/common/types/types.hpp"
#include "storage/single_node/constraints/existence_constraints.hpp"
#include "storage/single_node/constraints/unique_label_property_constraint.hpp"
#include "storage/single_node/edge.hpp"
#include "storage/single_node/indexes/key_index.hpp"
#include "storage/single_node/indexes/label_property_index.hpp"
#include "storage/single_node/mvcc/version_list.hpp"
#include "storage/single_node/vertex.hpp"
#include "transactions/type.hpp"
@ -79,6 +80,10 @@ class Storage {
// existence constraints
ExistenceConstraints existence_constraints_;
// unique constraints
storage::constraints::UniqueLabelPropertyConstraint
unique_label_property_constraints_;
std::vector<std::string> properties_on_disk_;
/// Gets the Vertex/Edge main storage map.

28
tests/manual/snapshot_explorer.cpp
View File

@ -20,7 +20,7 @@ int main(int argc, char *argv[]) {
// This makes sure we update the explorer when we bump the snapshot version.
// Snapshot layout is described in durability/version.hpp
static_assert(durability::kVersion == 8,
static_assert(durability::kVersion == 9,
"Wrong snapshot version, please update!");
fs::path snapshot_path(FLAGS_snapshot_file);
@ -108,6 +108,32 @@ int main(int argc, char *argv[]) {
LOG(INFO) << log;
}
decoder.ReadValue(&dv, Value::Type::List);
auto unique_constraint = dv.ValueList();
for (auto it = unique_constraint.begin(); it != unique_constraint.end();) {
std::string log("Adding unique constraint: ");
CHECK(it->IsString()) << "Label is not a string!";
log.append(it->ValueString());
log.append(" -> [");
++it;
CHECK(it->IsInt()) << "Number of properties is not an int!";
int64_t prop_size = it->ValueInt();
++it;
for (size_t i = 0; i < prop_size; ++i) {
CHECK(it->IsString()) << "Property is not a string!";
log.append(it->ValueString());
if (i != prop_size -1) {
log.append(", ");
} else {
log.append("]");
}
++it;
}
LOG(INFO) << log;
}
for (int64_t i = 0; i < vertex_count; ++i) {
auto vertex = decoder.ReadValue(&dv, Value::Type::Vertex);
CHECK(vertex) << "Failed to read vertex " << i;

4
tests/manual/wal_explorer.cpp
View File

@ -49,6 +49,10 @@ std::string StateDeltaTypeToString(database::StateDelta::Type type) {
return "BUILD_EXISTENCE_CONSTRAINT";
case database::StateDelta::Type::DROP_EXISTENCE_CONSTRAINT:
return "DROP_EXISTENCE_CONSTRAINT";
case database::StateDelta::Type::BUILD_UNIQUE_CONSTRAINT:
return "BUILD_UNIQUE_CONSTRAINT";
case database::StateDelta::Type::DROP_UNIQUE_CONSTRAINT:
return "DROP_UNIQUE_CONSTRAINT";
}
}

102
tests/unit/durability.cpp
View File

@ -196,6 +196,21 @@ bool CompareExistenceConstraints(const database::GraphDbAccessor &dba1,
});
}
bool CompareUniqueConstraints(const database::GraphDbAccessor &dba1,
const database::GraphDbAccessor &dba2) {
auto c1 = dba1.ListUniqueLabelPropertyConstraints();
auto c2 = dba2.ListUniqueLabelPropertyConstraints();
return c1.size() == c2.size() &&
std::is_permutation(c1.begin(), c1.end(), c2.begin(),
[&dba1, &dba2](auto &r1, auto &r2) {
return dba1.LabelName(r1.label) ==
dba2.LabelName(r2.label) &&
dba1.PropertyName(r1.property) ==
dba2.PropertyName(r2.property);
});
}
/** Checks if the given databases have the same contents (indices,
* vertices and edges). */
void CompareDbs(database::GraphDb &a, database::GraphDb &b) {
@ -212,6 +227,7 @@ void CompareDbs(database::GraphDb &a, database::GraphDb &b) {
<< utils::Join(index_b, ", ");
}
EXPECT_TRUE(CompareExistenceConstraints(*dba_a, *dba_b));
EXPECT_TRUE(CompareUniqueConstraints(*dba_a, *dba_b));
auto is_permutation_props = [&dba_a, &dba_b](const auto &p1_id,
const auto &p2_id) {
@ -521,6 +537,11 @@ TEST_F(Durability, SnapshotEncoding) {
ASSERT_TRUE(dv.IsList());
ASSERT_EQ(dv.ValueList().size(), 0);
// Unique constraints
decoder.ReadValue(&dv);
ASSERT_TRUE(dv.IsList());
ASSERT_EQ(dv.ValueList().size(), 0);
std::map<gid::Gid, communication::bolt::Vertex> decoded_vertices;
// Decode vertices.
@ -944,7 +965,7 @@ TEST_F(Durability, MoveToBackupWal) {
ASSERT_TRUE(durability::ContainsDurabilityFiles(backup_dir_));
}
TEST_F(Durability, UniqueConstraintRecoverySnapshotAndWal) {
TEST_F(Durability, UniqueIndexRecoverySnapshotAndWal) {
auto config = DbConfig();
config.durability_enabled = true;
database::GraphDb db{config};
@ -988,7 +1009,7 @@ TEST_F(Durability, UniqueConstraintRecoverySnapshotAndWal) {
}
}
TEST_F(Durability, UniqueConstraintRecoveryWal) {
TEST_F(Durability, UniqueIndexRecoveryWal) {
auto config = DbConfig();
config.durability_enabled = true;
database::GraphDb db{config};
@ -1073,7 +1094,7 @@ TEST_F(Durability, ExistenceConstraintRecoverySnapshotAndWal) {
dba->DeleteExistenceConstraint(l1, p1);
dba->Commit();
}
// create snapshot with build index and vertex
// create snapshot with build existence constraint
MakeSnapshot(db);
{
auto dba = db.Access();
@ -1095,6 +1116,81 @@ TEST_F(Durability, ExistenceConstraintRecoverySnapshotAndWal) {
}
}
TEST_F(Durability, UniqueConstraintRecoveryWal) {
auto config = DbConfig(true, false);
database::GraphDb db{config};
{
// Fill database with some data
auto dba = db.Access();
DbGenerator gen(*dba);
dba->InsertVertex();
gen.InsertVertex();
gen.InsertVertex();
auto l1 = dba->Label("l1");
auto p1 = dba->Property("p1");
dba->BuildUniqueConstraint(l1, p1);
gen.InsertEdge();
auto l2 = dba->Label("l2");
auto p2 = dba->Property("p2");
dba->BuildUniqueConstraint(l2, p2);
gen.InsertVertex();
gen.InsertEdge();
dba->DeleteUniqueConstraint(l1, p1);
dba->Commit();
}
{
// Recover and compare
db.wal().Flush();
auto recovered_config = DbConfig(false, true);
database::GraphDb recovered_db{recovered_config};
CompareDbs(db, recovered_db);
}
}
TEST_F(Durability, UniqueConstraintRecoverySnapshotAndWal) {
auto config = DbConfig(true, false);
database::GraphDb db{config};
{
// Fill database with some data
auto dba = db.Access();
DbGenerator gen(*dba);
dba->InsertVertex();
gen.InsertVertex();
gen.InsertVertex();
auto l1 = dba->Label("l1");
auto p1 = dba->Property("p1");
dba->BuildUniqueConstraint(l1, p1);
gen.InsertEdge();
auto l2 = dba->Label("l2");
auto p2 = dba->Property("p2");
dba->BuildUniqueConstraint(l2, p2);
gen.InsertVertex();
gen.InsertEdge();
dba->DeleteUniqueConstraint(l1, p1);
dba->Commit();
}
// create snapshot with build unique constraint
MakeSnapshot(db);
{
auto dba = db.Access();
DbGenerator gen(*dba);
gen.InsertVertex();
gen.InsertVertex();
auto l3 = dba->Label("l3");
auto p3 = dba->Property("p3");
dba->BuildUniqueConstraint(l3, p3);
dba->Commit();
}
{
// Recover and compare
db.wal().Flush();
auto recovered_config = DbConfig(false, true);
database::GraphDb recovered_db{recovered_config};
CompareDbs(db, recovered_db);
}
}
int main(int argc, char **argv) {
::testing::InitGoogleTest(&argc, argv);
google::InitGoogleLogging(argv[0]);

15
tests/unit/unique_label_property_constraint.cpp
View File

@ -55,11 +55,8 @@ TEST_F(UniqueLabelPropertyTest, BuildWithViolation) {
auto dba2 = db_.Access();
auto v3 = dba2->FindVertex(v1.gid(), false);
auto v4 = dba2->FindVertex(v2.gid(), false);
constraint_.UpdateOnAddLabel(label_, v3, dba2->transaction());
EXPECT_THROW(constraint_.UpdateOnAddLabel(label_, v4, dba2->transaction()),
database::IndexConstraintViolationException);
EXPECT_THROW(constraint_.UpdateOnAddProperty(property_, value_, v4,
dba2->transaction()),
constraint_.Update(v3, dba2->transaction());
EXPECT_THROW(constraint_.Update(v4, dba2->transaction()),
database::IndexConstraintViolationException);
}
@ -148,8 +145,7 @@ TEST_F(UniqueLabelPropertyTest, InsertRemoveAbortInsert) {
auto dba = db_.Access();
auto v = dba->FindVertex(gid, false);
v.PropsErase(property_);
constraint_.UpdateOnRemoveProperty(property_, value_, v,
dba->transaction());
constraint_.UpdateOnRemoveProperty(property_, v, dba->transaction());
dba->Abort();
}
{
@ -219,8 +215,7 @@ TEST_F(UniqueLabelPropertyTest, InsertRemoveInsert) {
auto dba = db_.Access();
auto v = dba->FindVertex(gid, false);
v.PropsErase(property_);
constraint_.UpdateOnRemoveProperty(property_, value_, v,
dba->transaction());
constraint_.UpdateOnRemoveProperty(property_, v, dba->transaction());
dba->Commit();
}
{
@ -241,7 +236,7 @@ TEST_F(UniqueLabelPropertyTest, InsertRemoveInsertSameTransaction) {
v.PropsSet(property_, value_);
constraint_.UpdateOnAddProperty(property_, value_, v, dba->transaction());
v.PropsErase(property_);
constraint_.UpdateOnRemoveProperty(property_, value_, v, dba->transaction());
constraint_.UpdateOnRemoveProperty(property_, v, dba->transaction());
v.PropsSet(property_, value_);
constraint_.UpdateOnAddProperty(property_, value_, v, dba->transaction());
dba->Commit();

3
tools/src/mg_import_csv/main.cpp
View File

@ -20,7 +20,7 @@
#include "utils/timer.hpp"
// Snapshot layout is described in durability/version.hpp
static_assert(durability::kVersion == 8,
static_assert(durability::kVersion == 9,
"Wrong snapshot version, please update!");
bool ValidateNotEmpty(const char *flagname, const std::string &value) {
@ -425,6 +425,7 @@ void Convert(const std::vector<std::string> &nodes,
encoder.WriteList({}); // Transactional snapshot.
encoder.WriteList({}); // Label + property indexes.
encoder.WriteList({}); // Existence constraints
encoder.WriteList({}); // Unique constraints
// PassNodes streams vertices to the encoder.
for (const auto &nodes_file : nodes) {
node_count +=