Compare commits
1 Commits
fix-pre-co
...
disk-stora
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9e1b4a6ef1 |
@@ -154,9 +154,10 @@ struct Delta {
|
|||||||
struct RemoveInEdgeTag {};
|
struct RemoveInEdgeTag {};
|
||||||
struct RemoveOutEdgeTag {};
|
struct RemoveOutEdgeTag {};
|
||||||
|
|
||||||
Delta(DeleteDeserializedObjectTag /*tag*/, std::atomic<uint64_t> *timestamp,
|
Delta(DeleteDeserializedObjectTag /*tag*/, uint64_t timestamp, const std::optional<std::string> &old_disk_key)
|
||||||
const std::optional<std::string> &old_disk_key)
|
: action(Action::DELETE_DESERIALIZED_OBJECT), command_id(0), old_disk_key(old_disk_key) {
|
||||||
: action(Action::DELETE_DESERIALIZED_OBJECT), timestamp(timestamp), command_id(0), old_disk_key(old_disk_key) {}
|
this->timestamp->store(timestamp, std::memory_order_release);
|
||||||
|
}
|
||||||
|
|
||||||
Delta(DeleteObjectTag /*tag*/, std::atomic<uint64_t> *timestamp, uint64_t command_id)
|
Delta(DeleteObjectTag /*tag*/, std::atomic<uint64_t> *timestamp, uint64_t command_id)
|
||||||
: action(Action::DELETE_OBJECT), timestamp(timestamp), command_id(command_id) {}
|
: action(Action::DELETE_OBJECT), timestamp(timestamp), command_id(command_id) {}
|
||||||
|
|||||||
@@ -10,7 +10,10 @@
|
|||||||
// licenses/APL.txt.
|
// licenses/APL.txt.
|
||||||
|
|
||||||
#include "rocksdb_storage.hpp"
|
#include "rocksdb_storage.hpp"
|
||||||
|
|
||||||
#include <string_view>
|
#include <string_view>
|
||||||
|
|
||||||
|
#include "utils/disk_utils.hpp"
|
||||||
#include "utils/rocksdb_serialization.hpp"
|
#include "utils/rocksdb_serialization.hpp"
|
||||||
|
|
||||||
namespace memgraph::storage {
|
namespace memgraph::storage {
|
||||||
@@ -23,12 +26,6 @@ inline rocksdb::Slice StripTimestampFromUserKey(const rocksdb::Slice &user_key,
|
|||||||
return ret;
|
return ret;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// NOTE: Timestamp is encoded as last 8B in user key.
|
|
||||||
inline rocksdb::Slice ExtractTimestampFromUserKey(const rocksdb::Slice &user_key) {
|
|
||||||
assert(user_key.size() >= sizeof(uint64_t));
|
|
||||||
return {user_key.data() + user_key.size() - sizeof(uint64_t), sizeof(uint64_t)};
|
|
||||||
}
|
|
||||||
|
|
||||||
// Extracts global id from user key. User key must be without timestamp.
|
// Extracts global id from user key. User key must be without timestamp.
|
||||||
std::string_view ExtractGidFromUserKey(const rocksdb::Slice &key) {
|
std::string_view ExtractGidFromUserKey(const rocksdb::Slice &key) {
|
||||||
assert(key.size() >= 2);
|
assert(key.size() >= 2);
|
||||||
@@ -51,7 +48,7 @@ int ComparatorWithU64TsImpl::Compare(const rocksdb::Slice &a, const rocksdb::Sli
|
|||||||
// Compare timestamp.
|
// Compare timestamp.
|
||||||
// For the same user key with different timestamps, larger (newer) timestamp
|
// For the same user key with different timestamps, larger (newer) timestamp
|
||||||
// comes first.
|
// comes first.
|
||||||
return CompareTimestamp(ExtractTimestampFromUserKey(b), ExtractTimestampFromUserKey(a));
|
return CompareTimestamp(utils::ExtractTimestampFromUserKey(b), utils::ExtractTimestampFromUserKey(a));
|
||||||
}
|
}
|
||||||
|
|
||||||
int ComparatorWithU64TsImpl::CompareWithoutTimestamp(const rocksdb::Slice &a, bool a_has_ts, const rocksdb::Slice &b,
|
int ComparatorWithU64TsImpl::CompareWithoutTimestamp(const rocksdb::Slice &a, bool a_has_ts, const rocksdb::Slice &b,
|
||||||
|
|||||||
@@ -330,8 +330,9 @@ std::optional<storage::VertexAccessor> DiskStorage::DiskAccessor::LoadVertexToMa
|
|||||||
}
|
}
|
||||||
std::vector<LabelId> labels_id{utils::DeserializeLabelsFromMainDiskStorage(key)};
|
std::vector<LabelId> labels_id{utils::DeserializeLabelsFromMainDiskStorage(key)};
|
||||||
PropertyStore properties{utils::DeserializePropertiesFromMainDiskStorage(value)};
|
PropertyStore properties{utils::DeserializePropertiesFromMainDiskStorage(value)};
|
||||||
|
std::string old_tx_commit_ts = utils::ExtractTimestampFromUserKey(key);
|
||||||
return CreateVertex(main_storage_accessor, gid, std::move(labels_id), std::move(properties),
|
return CreateVertex(main_storage_accessor, gid, std::move(labels_id), std::move(properties),
|
||||||
CreateDeleteDeserializedObjectDelta(&transaction_, key));
|
CreateDeleteDeserializedObjectDelta(&transaction_, std::move(old_tx_commit_ts), key));
|
||||||
}
|
}
|
||||||
|
|
||||||
std::optional<storage::VertexAccessor> DiskStorage::DiskAccessor::LoadVertexToLabelIndexCache(
|
std::optional<storage::VertexAccessor> DiskStorage::DiskAccessor::LoadVertexToLabelIndexCache(
|
||||||
@@ -408,6 +409,7 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(View view) {
|
|||||||
auto it =
|
auto it =
|
||||||
std::unique_ptr<rocksdb::Iterator>(disk_transaction_->GetIterator(ro, disk_storage->kvstore_->vertex_chandle));
|
std::unique_ptr<rocksdb::Iterator>(disk_transaction_->GetIterator(ro, disk_storage->kvstore_->vertex_chandle));
|
||||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||||
|
spdlog::trace("Key data, size: {}, {}", it->key().data(), it->key().size());
|
||||||
LoadVertexToMainMemoryCache(it->key().ToString(), it->value().ToString());
|
LoadVertexToMainMemoryCache(it->key().ToString(), it->value().ToString());
|
||||||
}
|
}
|
||||||
scanned_all_vertices_ = true;
|
scanned_all_vertices_ = true;
|
||||||
@@ -440,9 +442,10 @@ std::unordered_set<Gid> DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWit
|
|||||||
spdlog::trace("Loaded vertex with gid: {} from main index storage to label index",
|
spdlog::trace("Loaded vertex with gid: {} from main index storage to label index",
|
||||||
utils::SerializeIdType(vertex.gid));
|
utils::SerializeIdType(vertex.gid));
|
||||||
/// TODO: here are doing serialization and then later deserialization again -> expensive
|
/// TODO: here are doing serialization and then later deserialization again -> expensive
|
||||||
|
// std::string old_tx_commit_ts = utils::ExtractTimestampFromUserKey();
|
||||||
LoadVertexToLabelIndexCache(label, utils::SerializeVertexAsKeyForLabelIndex(label, vertex.gid),
|
LoadVertexToLabelIndexCache(label, utils::SerializeVertexAsKeyForLabelIndex(label, vertex.gid),
|
||||||
utils::SerializeVertexAsValueForLabelIndex(label, vertex.labels, vertex.properties),
|
utils::SerializeVertexAsValueForLabelIndex(label, vertex.labels, vertex.properties),
|
||||||
CreateDeleteDeserializedIndexObjectDelta(&transaction_, index_deltas, std::nullopt),
|
CreateDeleteDeserializedIndexObjectDelta(index_deltas, "123", std::nullopt),
|
||||||
indexed_vertices->access());
|
indexed_vertices->access());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -469,9 +472,11 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelIndex(LabelId label,
|
|||||||
Gid curr_gid = Gid::FromUint(std::stoull(utils::ExtractGidFromLabelIndexStorage(key)));
|
Gid curr_gid = Gid::FromUint(std::stoull(utils::ExtractGidFromLabelIndexStorage(key)));
|
||||||
spdlog::trace("Loaded vertex with key: {} from label index storage", key);
|
spdlog::trace("Loaded vertex with key: {} from label index storage", key);
|
||||||
if (key.starts_with(serialized_label) && !utils::Contains(gids, curr_gid)) {
|
if (key.starts_with(serialized_label) && !utils::Contains(gids, curr_gid)) {
|
||||||
LoadVertexToLabelIndexCache(label, index_it->key().ToString(), index_it->value().ToString(),
|
std::string old_tx_commit_ts{utils::ExtractTimestampFromUserKey(index_it->key())};
|
||||||
CreateDeleteDeserializedIndexObjectDelta(&transaction_, index_deltas, key),
|
LoadVertexToLabelIndexCache(
|
||||||
indexed_vertices->access());
|
label, index_it->key().ToString(), index_it->value().ToString(),
|
||||||
|
CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::move(old_tx_commit_ts), key),
|
||||||
|
indexed_vertices->access());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -517,8 +522,7 @@ std::unordered_set<Gid> DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWit
|
|||||||
LoadVertexToLabelPropertyIndexCache(
|
LoadVertexToLabelPropertyIndexCache(
|
||||||
label, utils::SerializeVertexAsKeyForLabelPropertyIndex(label, property, vertex.gid),
|
label, utils::SerializeVertexAsKeyForLabelPropertyIndex(label, property, vertex.gid),
|
||||||
utils::SerializeVertexAsValueForLabelPropertyIndex(label, vertex.labels, vertex.properties),
|
utils::SerializeVertexAsValueForLabelPropertyIndex(label, vertex.labels, vertex.properties),
|
||||||
CreateDeleteDeserializedIndexObjectDelta(&transaction_, index_deltas, std::nullopt),
|
CreateDeleteDeserializedIndexObjectDelta(index_deltas, "123", std::nullopt), indexed_vertices->access());
|
||||||
indexed_vertices->access());
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -547,9 +551,11 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndex(LabelId l
|
|||||||
Gid curr_gid = Gid::FromUint(std::stoull(utils::ExtractGidFromLabelPropertyIndexStorage(key)));
|
Gid curr_gid = Gid::FromUint(std::stoull(utils::ExtractGidFromLabelPropertyIndexStorage(key)));
|
||||||
/// TODO: optimize
|
/// TODO: optimize
|
||||||
if (label_property_filter(key, label_property_prefix, gids, curr_gid)) {
|
if (label_property_filter(key, label_property_prefix, gids, curr_gid)) {
|
||||||
LoadVertexToLabelPropertyIndexCache(label, index_it->key().ToString(), index_it->value().ToString(),
|
std::string old_tx_commit_ts{utils::ExtractTimestampFromUserKey(index_it->key())};
|
||||||
CreateDeleteDeserializedIndexObjectDelta(&transaction_, index_deltas, key),
|
LoadVertexToLabelPropertyIndexCache(
|
||||||
indexed_vertices->access());
|
label, index_it->key().ToString(), index_it->value().ToString(),
|
||||||
|
CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::move(old_tx_commit_ts), key),
|
||||||
|
indexed_vertices->access());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -599,9 +605,11 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndexWithPointV
|
|||||||
PropertyStore properties = utils::DeserializePropertiesFromLabelPropertyIndexStorage(it_value);
|
PropertyStore properties = utils::DeserializePropertiesFromLabelPropertyIndexStorage(it_value);
|
||||||
if (key.starts_with(label_property_prefix) && !utils::Contains(gids, curr_gid) &&
|
if (key.starts_with(label_property_prefix) && !utils::Contains(gids, curr_gid) &&
|
||||||
properties.IsPropertyEqual(property, value)) {
|
properties.IsPropertyEqual(property, value)) {
|
||||||
LoadVertexToLabelPropertyIndexCache(label, index_it->key().ToString(), index_it->value().ToString(),
|
std::string old_tx_commit_ts{utils::ExtractTimestampFromUserKey(index_it->key())};
|
||||||
CreateDeleteDeserializedIndexObjectDelta(&transaction_, index_deltas, key),
|
LoadVertexToLabelPropertyIndexCache(
|
||||||
indexed_vertices->access());
|
label, index_it->key().ToString(), index_it->value().ToString(),
|
||||||
|
CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::move(old_tx_commit_ts), key),
|
||||||
|
indexed_vertices->access());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -642,8 +650,7 @@ DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWithLabelPropertyIndexCache
|
|||||||
LoadVertexToLabelPropertyIndexCache(
|
LoadVertexToLabelPropertyIndexCache(
|
||||||
label, utils::SerializeVertexAsKeyForLabelPropertyIndex(label, property, vertex.gid),
|
label, utils::SerializeVertexAsKeyForLabelPropertyIndex(label, property, vertex.gid),
|
||||||
utils::SerializeVertexAsValueForLabelPropertyIndex(label, vertex.labels, vertex.properties),
|
utils::SerializeVertexAsValueForLabelPropertyIndex(label, vertex.labels, vertex.properties),
|
||||||
CreateDeleteDeserializedIndexObjectDelta(&transaction_, index_deltas, std::nullopt),
|
CreateDeleteDeserializedIndexObjectDelta(index_deltas, "123", std::nullopt), indexed_vertices->access());
|
||||||
indexed_vertices->access());
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return gids;
|
return gids;
|
||||||
@@ -678,9 +685,11 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndexForInterva
|
|||||||
!IsPropertyValueWithinInterval(prop_value, lower_bound, upper_bound)) {
|
!IsPropertyValueWithinInterval(prop_value, lower_bound, upper_bound)) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
LoadVertexToLabelPropertyIndexCache(label, index_it->key().ToString(), index_it->value().ToString(),
|
std::string old_tx_commit_ts{utils::ExtractTimestampFromUserKey(index_it->key())};
|
||||||
CreateDeleteDeserializedIndexObjectDelta(&transaction_, index_deltas, key_str),
|
LoadVertexToLabelPropertyIndexCache(
|
||||||
indexed_vertices->access());
|
label, index_it->key().ToString(), index_it->value().ToString(),
|
||||||
|
CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::move(old_tx_commit_ts), key_str),
|
||||||
|
indexed_vertices->access());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1048,7 +1057,8 @@ Result<EdgeAccessor> DiskStorage::DiskAccessor::CreateEdge(const VertexAccessor
|
|||||||
EdgeRef edge(gid);
|
EdgeRef edge(gid);
|
||||||
if (config_.properties_on_edges) {
|
if (config_.properties_on_edges) {
|
||||||
auto acc = edges_.access();
|
auto acc = edges_.access();
|
||||||
auto *delta = CreateDeleteDeserializedObjectDelta(&transaction_, old_disk_key);
|
std::string old_tx_commit_ts{utils::ExtractTimestampFromUserKey(old_disk_key)};
|
||||||
|
auto *delta = CreateDeleteDeserializedObjectDelta(&transaction_, std::move(old_tx_commit_ts), old_disk_key);
|
||||||
auto [it, inserted] = acc.insert(Edge(gid, delta));
|
auto [it, inserted] = acc.insert(Edge(gid, delta));
|
||||||
MG_ASSERT(inserted, "The edge must be inserted here!");
|
MG_ASSERT(inserted, "The edge must be inserted here!");
|
||||||
MG_ASSERT(it != acc.end(), "Invalid Edge accessor!");
|
MG_ASSERT(it != acc.end(), "Invalid Edge accessor!");
|
||||||
@@ -1517,6 +1527,7 @@ utils::BasicResult<StorageDataManipulationError, void> DiskStorage::DiskAccessor
|
|||||||
[](const Delta &delta) { return delta.action == Delta::Action::DELETE_DESERIALIZED_OBJECT; })) {
|
[](const Delta &delta) { return delta.action == Delta::Action::DELETE_DESERIALIZED_OBJECT; })) {
|
||||||
} else {
|
} else {
|
||||||
std::unique_lock<utils::SpinLock> engine_guard(storage_->engine_lock_);
|
std::unique_lock<utils::SpinLock> engine_guard(storage_->engine_lock_);
|
||||||
|
transaction_.EnsureCommitTimestampExists();
|
||||||
commit_timestamp_.emplace(disk_storage->CommitTimestamp(desired_commit_timestamp));
|
commit_timestamp_.emplace(disk_storage->CommitTimestamp(desired_commit_timestamp));
|
||||||
|
|
||||||
if (auto res = FlushMainMemoryCache(); res.HasError()) {
|
if (auto res = FlushMainMemoryCache(); res.HasError()) {
|
||||||
|
|||||||
@@ -114,17 +114,16 @@ inline Delta *CreateDeleteObjectDelta(Transaction *transaction) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// TODO: what if in-memory analytical
|
/// TODO: what if in-memory analytical
|
||||||
inline Delta *CreateDeleteDeserializedObjectDelta(Transaction *transaction, std::optional<std::string> old_disk_key) {
|
inline Delta *CreateDeleteDeserializedObjectDelta(Transaction *transaction, std::string &&old_tx_commit_ts,
|
||||||
transaction->EnsureCommitTimestampExists();
|
std::optional<std::string> old_disk_key) {
|
||||||
return &transaction->deltas.emplace_back(Delta::DeleteDeserializedObjectTag(), transaction->commit_timestamp.get(),
|
return &transaction->deltas.emplace_back(Delta::DeleteDeserializedObjectTag(), std::stoull(old_tx_commit_ts),
|
||||||
old_disk_key);
|
old_disk_key);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// TODO: what if in-memory analytical
|
/// TODO: what if in-memory analytical
|
||||||
inline Delta *CreateDeleteDeserializedIndexObjectDelta(Transaction *transaction, std::list<Delta> &deltas,
|
inline Delta *CreateDeleteDeserializedIndexObjectDelta(std::list<Delta> &deltas, std::string &&old_tx_commit_ts,
|
||||||
std::optional<std::string> old_disk_key) {
|
std::optional<std::string> old_disk_key) {
|
||||||
transaction->EnsureCommitTimestampExists();
|
return &deltas.emplace_back(Delta::DeleteDeserializedObjectTag(), std::stoull(old_tx_commit_ts), old_disk_key);
|
||||||
return &deltas.emplace_back(Delta::DeleteDeserializedObjectTag(), transaction->commit_timestamp.get(), old_disk_key);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// This function creates a delta in the transaction for the object and links
|
/// This function creates a delta in the transaction for the object and links
|
||||||
|
|||||||
@@ -11,6 +11,7 @@
|
|||||||
|
|
||||||
#pragma once
|
#pragma once
|
||||||
|
|
||||||
|
#include <rocksdb/slice.h>
|
||||||
#include "storage/v2/delta.hpp"
|
#include "storage/v2/delta.hpp"
|
||||||
|
|
||||||
namespace memgraph::utils {
|
namespace memgraph::utils {
|
||||||
@@ -25,4 +26,11 @@ inline std::optional<std::string> GetOldDiskKeyOrNull(storage::Delta *head) {
|
|||||||
return std::nullopt;
|
return std::nullopt;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// NOTE: Timestamp is encoded as last 8B in user key.
|
||||||
|
inline std::string ExtractTimestampFromUserKey(const rocksdb::Slice &user_key) {
|
||||||
|
spdlog::trace("User key data, size: {}, {}", user_key.data(), user_key.size());
|
||||||
|
MG_ASSERT(user_key.size() >= sizeof(uint64_t));
|
||||||
|
return {user_key.data() + user_key.size() - sizeof(uint64_t), sizeof(uint64_t)};
|
||||||
|
}
|
||||||
|
|
||||||
} // namespace memgraph::utils
|
} // namespace memgraph::utils
|
||||||
|
|||||||
Reference in New Issue
Block a user