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

Change caching mechanism.

Browse Source 
Instead of trying to read into the index caches directly, read into the
main chache, and merge the index and main caches when appropriate. This
is needed because of several index modifying calls within one
transaction makes the indexing process convoluted, and should be
propertly addressed when the On-disk LRU cache will be properly
implemented.
This commit is contained in:
gvolfing
2023-10-16 13:22:22 +02:00
parent 646818a0b4
commit 8feefc20b7
2 changed files with 104 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
src/storage/v2/disk/storage.cpp
View File

@@ -552,7 +552,6 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, View view) {
std::list<storage::Delta> *index_deltas{nullptr};
auto merge_with_main_cache = [&](auto &index, auto &index_delta_storage) -> std::unordered_set<Gid> {
index[cache_key] = utils::SkipList<Vertex>();
indexed_vertices = &index[cache_key];
index_delta_storage.emplace_back();
index_deltas = &index_delta_storage.back();
@@ -560,24 +559,36 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, View view) {
return MergeVerticesFromMainCacheWithLabelIndexCache(cache_key, view, *index_deltas, *indexed_vertices);
};
auto &label_index_cache = transaction_.label_index_cache_;
auto &label_index_cache_ci = transaction_.label_index_cache_ci_;
auto &cache = transaction_.label_index_cache_;
auto &cache_ci = transaction_.label_index_cache_ci_;
auto &index_delta_storage = transaction_.index_deltas_storage_;
if (transaction_.command_id > label_index_cache_ci) {
label_index_cache_ci = transaction_.command_id;
SyncDeletedVertices(cache, cache_key, index_delta_storage, view, merge_with_main_cache);
auto gids = merge_with_main_cache(label_index_cache, index_delta_storage);
LoadVerticesFromDiskLabelIndex(gids, cache_key, *index_deltas, *indexed_vertices);
if (transaction_.command_id > cache_ci) {
cache_ci = transaction_.command_id;
} else {
if (label_index_cache.contains(cache_key) && view == View::OLD) {
merge_with_main_cache(label_index_cache, index_delta_storage);
} else {
auto gids = merge_with_main_cache(label_index_cache, index_delta_storage);
LoadVerticesFromDiskLabelIndex(gids, cache_key, *index_deltas, *indexed_vertices);
if (cache.contains(cache_key)) {
if (view == View::OLD) {
// TODO
// we do not need this merge if we can make sure that
// the removing and re-adding the same label to a given
// vertex within one transaction is not permitted.
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
}
cache[cache_key] = utils::SkipList<Vertex>();
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
}
cache[cache_key] = utils::SkipList<Vertex>();
index_delta_storage.emplace_back();
index_deltas = &index_delta_storage.back();
LoadVerticesFromDiskLabelIndex({}, cache_key, *index_deltas, transaction_.vertices_);
}
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
}
@@ -620,22 +631,35 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
auto &label_property_index_cache_ci = transaction_.label_property_index_cache_ci_;
auto &index_delta_storage = transaction_.index_deltas_storage_;
SyncDeletedVertices(label_property_index_cache, cache_key, index_delta_storage, view, merge_with_main_cache);
if (transaction_.command_id > label_property_index_cache_ci) {
label_property_index_cache_ci = transaction_.command_id;
auto gids = merge_with_main_cache(label_property_index_cache, index_delta_storage);
LoadVerticesFromDiskLabelPropertyIndex(gids, label, property, *index_deltas, *indexed_vertices,
disk_label_property_filter);
} else {
if (label_property_index_cache.contains(cache_key) && view == View::OLD) {
if (label_property_index_cache.contains(cache_key)) {
if (view == View::OLD) {
// TODO
// we do not need this merge if we can make sure that
// the removing and re-adding the same label to a given
// vertex within one transaction is not permitted.
merge_with_main_cache(label_property_index_cache, index_delta_storage);
return VerticesIterable(
AllVerticesIterable(label_property_index_cache[cache_key].access(), storage_, &transaction_, view));
}
label_property_index_cache[cache_key] = utils::SkipList<Vertex>();
merge_with_main_cache(label_property_index_cache, index_delta_storage);
} else {
auto gids = merge_with_main_cache(label_property_index_cache, index_delta_storage);
LoadVerticesFromDiskLabelPropertyIndex(gids, label, property, *index_deltas, *indexed_vertices,
disk_label_property_filter);
return VerticesIterable(
AllVerticesIterable(label_property_index_cache[cache_key].access(), storage_, &transaction_, view));
}
label_property_index_cache[cache_key] = utils::SkipList<Vertex>();
index_delta_storage.emplace_back();
index_deltas = &index_delta_storage.back();
LoadVerticesFromDiskLabelPropertyIndex({}, label, property, *index_deltas, transaction_.vertices_,
disk_label_property_filter);
}
merge_with_main_cache(label_property_index_cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
}
@@ -671,26 +695,37 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
*indexed_vertices, label_property_filter);
};
auto &label_property_val_index_cache = transaction_.label_property_val_index_cache_;
auto &label_property_val_index_cache_ci = transaction_.label_property_val_index_cache_ci_;
auto &cache = transaction_.label_property_val_index_cache_;
auto &cache_ci = transaction_.label_property_val_index_cache_ci_;
auto &index_delta_storage = transaction_.index_deltas_storage_;
if (transaction_.command_id > label_property_val_index_cache_ci) {
label_property_val_index_cache_ci = transaction_.command_id;
SyncDeletedVertices(cache, cache_key, index_delta_storage, view, merge_with_main_cache);
auto gids = merge_with_main_cache(label_property_val_index_cache, index_delta_storage);
LoadVerticesFromDiskLabelPropertyIndexWithPointValueLookup(gids, label, property, value, *index_deltas,
*indexed_vertices);
if (transaction_.command_id > cache_ci) {
cache_ci = transaction_.command_id;
} else {
if (label_property_val_index_cache.contains(cache_key) && view == View::OLD) {
merge_with_main_cache(label_property_val_index_cache, index_delta_storage);
} else {
auto gids = merge_with_main_cache(label_property_val_index_cache, index_delta_storage);
LoadVerticesFromDiskLabelPropertyIndexWithPointValueLookup(gids, label, property, value, *index_deltas,
*indexed_vertices);
if (cache.contains(cache_key)) {
if (view == View::OLD) {
// TODO
// we do not need this merge if we can make sure that
// the removing and re-adding the same label to a given
// vertex within one transaction is not permitted.
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
}
cache[cache_key] = utils::SkipList<Vertex>();
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
}
cache[cache_key] = utils::SkipList<Vertex>();
index_delta_storage.emplace_back();
index_deltas = &index_delta_storage.back();
LoadVerticesFromDiskLabelPropertyIndexWithPointValueLookup({}, label, property, value, *index_deltas,
transaction_.vertices_);
}
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
}
@@ -720,26 +755,37 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
label, property, view, lower_bound, upper_bound, *index_deltas, *indexed_vertices);
};
auto &label_property_range_index_cache = transaction_.label_property_range_index_cache_;
auto &label_property_range_index_cache_ci = transaction_.label_property_range_index_cache_ci_;
auto &cache = transaction_.label_property_range_index_cache_;
auto &cache_ci = transaction_.label_property_range_index_cache_ci_;
auto &index_delta_storage = transaction_.index_deltas_storage_;
if (transaction_.command_id > label_property_range_index_cache_ci) {
label_property_range_index_cache_ci = transaction_.command_id;
SyncDeletedVertices(cache, cache_key, index_delta_storage, view, merge_with_main_cache);
auto gids = merge_with_main_cache(label_property_range_index_cache, index_delta_storage);
LoadVerticesFromDiskLabelPropertyIndexForIntervalSearch(gids, label, property, lower_bound, upper_bound,
*index_deltas, *indexed_vertices);
if (transaction_.command_id > cache_ci) {
cache_ci = transaction_.command_id;
} else {
if (label_property_range_index_cache.contains(cache_key) && view == View::OLD) {
merge_with_main_cache(label_property_range_index_cache, index_delta_storage);
} else {
auto gids = merge_with_main_cache(label_property_range_index_cache, index_delta_storage);
LoadVerticesFromDiskLabelPropertyIndexForIntervalSearch(gids, label, property, lower_bound, upper_bound,
*index_deltas, *indexed_vertices);
if (cache.contains(cache_key)) {
if (view == View::OLD) {
// TODO
// we do not need this merge if we can make sure that
// the removing and re-adding the same label to a given
// vertex within one transaction is not permitted.
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
}
cache[cache_key] = utils::SkipList<Vertex>();
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
}
cache[cache_key] = utils::SkipList<Vertex>();
index_delta_storage.emplace_back();
index_deltas = &index_delta_storage.back();
LoadVerticesFromDiskLabelPropertyIndexForIntervalSearch({}, label, property, lower_bound, upper_bound,
*index_deltas, transaction_.vertices_);
}
merge_with_main_cache(cache, index_delta_storage);
return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
}

11
src/storage/v2/disk/storage.hpp
View File

@@ -94,6 +94,17 @@ class DiskStorage final : public Storage {
const std::optional<utils::Bound<PropertyValue>> &upper_bound, std::list<Delta> &index_deltas,
utils::SkipList<Vertex> &indexed_vertices);
template <typename TIndex, typename TIndexCacheKey, typename TMergeFunc>
void SyncDeletedVertices(TIndex &index, TIndexCacheKey &cache_key,
std::vector<std::list<Delta>> &index_delta_storage, View view, TMergeFunc &merge_func) {
if (!transaction_.vertices_to_delete_.empty()) {
index[cache_key] = utils::SkipList<Vertex>();
if (view == View::OLD) {
merge_func(index, index_delta_storage);
}
}
}
public:
DiskAccessor(const DiskAccessor &) = delete;
DiskAccessor &operator=(const DiskAccessor &) = delete;