diff --git a/src/storage/v2/indices.cpp b/src/storage/v2/indices.cpp
index c3ce80211..6a25cec64 100644
--- a/src/storage/v2/indices.cpp
+++ b/src/storage/v2/indices.cpp
@@ -295,23 +295,42 @@ bool CurrentVersionHasLabelProperty(Vertex *vertex, LabelId label,
void LabelIndex::UpdateOnAddLabel(LabelId label, Vertex *vertex,
const Transaction &tx) {
- GetOrCreateStorage(label)->access().insert(Entry{vertex, tx.start_timestamp});
+ auto it = index_.find(label);
+ if (it == index_.end()) return;
+ it->second.access().insert(Entry{vertex, tx.start_timestamp});
+}
+
+bool LabelIndex::CreateIndex(LabelId label,
+ utils::SkipList<Vertex>::Accessor vertices) {
+ auto [it, emplaced] =
+ index_.emplace(std::piecewise_construct, std::forward_as_tuple(label),
+ std::forward_as_tuple());
+ if (!emplaced) {
+ // Index already exists.
+ return false;
+ }
+ auto acc = it->second.access();
+ for (Vertex &vertex : vertices) {
+ if (vertex.deleted || !utils::Contains(vertex.labels, label)) {
+ continue;
+ }
+ acc.insert(Entry{&vertex, 0});
+ }
+ return true;
}
std::vector<LabelId> LabelIndex::ListIndices() const {
std::vector<LabelId> ret;
ret.reserve(index_.size());
- auto acc = index_.access();
- for (const auto &item : acc) {
- ret.push_back(item.label);
+ for (const auto &item : index_) {
+ ret.push_back(item.first);
}
return ret;
}
void LabelIndex::RemoveObsoleteEntries(uint64_t oldest_active_start_timestamp) {
- auto index_acc = index_.access();
- for (auto &label_storage : index_acc) {
- auto vertices_acc = label_storage.vertices.access();
+ for (auto &label_storage : index_) {
+ auto vertices_acc = label_storage.second.access();
for (auto it = vertices_acc.begin(); it != vertices_acc.end();) {
auto next_it = it;
++next_it;
@@ -322,7 +341,7 @@ void LabelIndex::RemoveObsoleteEntries(uint64_t oldest_active_start_timestamp) {
}
if ((next_it != vertices_acc.end() && it->vertex == next_it->vertex) ||
- !AnyVersionHasLabel(it->vertex, label_storage.label,
+ !AnyVersionHasLabel(it->vertex, label_storage.first,
oldest_active_start_timestamp)) {
vertices_acc.remove(*it);
}
@@ -332,18 +351,6 @@ void LabelIndex::RemoveObsoleteEntries(uint64_t oldest_active_start_timestamp) {
}
}
-utils::SkipList<LabelIndex::Entry> *LabelIndex::GetOrCreateStorage(
- LabelId label) {
- auto acc = index_.access();
- auto it = acc.find(label);
- if (it == acc.end()) {
- LabelStorage label_storage{.label = label,
- .vertices = utils::SkipList<Entry>()};
- it = acc.insert(std::move(label_storage)).first;
- }
- return &it->vertices;
-}
-
LabelIndex::Iterable::Iterator::Iterator(
Iterable *self, utils::SkipList<Entry>::Iterator index_iterator)
: self_(self),
diff --git a/src/storage/v2/indices.hpp b/src/storage/v2/indices.hpp
index c11f6138c..3b3e5cc73 100644
--- a/src/storage/v2/indices.hpp
+++ b/src/storage/v2/indices.hpp
@@ -45,6 +45,15 @@ class LabelIndex {
/// @throw std::bad_alloc
void UpdateOnAddLabel(LabelId label, Vertex *vertex, const Transaction &tx);
+ /// @throw std::bad_alloc
+ bool CreateIndex(LabelId label, utils::SkipList<Vertex>::Accessor vertices);
+
+ bool DropIndex(LabelId label) { return index_.erase(label) > 0; }
+
+ bool IndexExists(LabelId label) const {
+ return index_.find(label) != index_.end();
+ }
+
std::vector<LabelId> ListIndices() const;
void RemoveObsoleteEntries(uint64_t oldest_active_start_timestamp);
@@ -90,22 +99,23 @@ class LabelIndex {
};
/// Returns an self with vertices visible from the given transaction.
- /// @throw std::bad_alloc
Iterable Vertices(LabelId label, View view, Transaction *transaction) {
- return Iterable(GetOrCreateStorage(label)->access(), label, view,
- transaction, indices_);
+ auto it = index_.find(label);
+ CHECK(it != index_.end())
+ << "Index for label " << label.AsUint() << " doesn't exist";
+ return Iterable(it->second.access(), label, view, transaction, indices_);
}
int64_t ApproximateVertexCount(LabelId label) {
- return GetOrCreateStorage(label)->size();
+ auto it = index_.find(label);
+ CHECK(it != index_.end())
+ << "Index for label " << label.AsUint() << " doesn't exist";
+ return it->second.size();
}
private:
- utils::SkipList<LabelStorage> index_;
Indices *indices_;
-
- /// @throw std::bad_alloc
- utils::SkipList<Entry> *GetOrCreateStorage(LabelId label);
+ std::map<LabelId, utils::SkipList<Entry>> index_;
};
class LabelPropertyIndex {
@@ -140,7 +150,7 @@ class LabelPropertyIndex {
return index_.erase({label, property}) > 0;
}
- bool IndexExists(LabelId label, PropertyId property) {
+ bool IndexExists(LabelId label, PropertyId property) const {
return index_.find({label, property}) != index_.end();
}
diff --git a/src/storage/v2/storage.hpp b/src/storage/v2/storage.hpp
index 42d21a580..4e2f60e1f 100644
--- a/src/storage/v2/storage.hpp
+++ b/src/storage/v2/storage.hpp
@@ -182,7 +182,6 @@ class Storage final {
&storage_->indices_));
}
- /// @throw std::bad_alloc raised in Index::Vertices
VerticesIterable Vertices(LabelId label, View view);
/// @throw std::bad_alloc raised in Index::Vertices
@@ -297,6 +296,12 @@ class Storage final {
/// @throw std::bad_alloc if unable to insert a new mapping
EdgeTypeId NameToEdgeType(const std::string &name);
+ /// @throw std::bad_alloc
+ bool CreateIndex(LabelId label) {
+ std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+ return indices_.label_index.CreateIndex(label, vertices_.access());
+ }
+
/// @throw std::bad_alloc
bool CreateIndex(LabelId label, PropertyId property) {
std::unique_lock<utils::RWLock> storage_guard(main_lock_);
@@ -304,12 +309,21 @@ class Storage final {
vertices_.access());
}
+ bool DropIndex(LabelId label) {
+ std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+ return indices_.label_index.DropIndex(label);
+ }
+
bool DropIndex(LabelId label, PropertyId property) {
std::unique_lock<utils::RWLock> storage_guard(main_lock_);
return indices_.label_property_index.DropIndex(label, property);
}
- bool LabelPropertyIndexExists(LabelId label, PropertyId property) {
+ bool LabelIndexExists(LabelId label) const {
+ return indices_.label_index.IndexExists(label);
+ }
+
+ bool LabelPropertyIndexExists(LabelId label, PropertyId property) const {
return indices_.label_property_index.IndexExists(label, property);
}
diff --git a/tests/unit/query_plan_v2_create_set_remove_delete.cpp b/tests/unit/query_plan_v2_create_set_remove_delete.cpp
index 2975d5be0..9c0722cf3 100644
--- a/tests/unit/query_plan_v2_create_set_remove_delete.cpp
+++ b/tests/unit/query_plan_v2_create_set_remove_delete.cpp
@@ -93,6 +93,7 @@ TEST(QueryPlan, ScanAll) {
TEST(QueryPlan, ScanAllByLabel) {
storage::Storage db;
auto label = db.NameToLabel("label");
+ ASSERT_TRUE(db.CreateIndex(label));
{
auto dba = db.Access();
// Add some unlabeled vertices
diff --git a/tests/unit/storage_v2_gc.cpp b/tests/unit/storage_v2_gc.cpp
index e34317be0..df7d20612 100644
--- a/tests/unit/storage_v2_gc.cpp
+++ b/tests/unit/storage_v2_gc.cpp
@@ -167,6 +167,8 @@ TEST(StorageV2Gc, Indices) {
storage::Config{.gc = {.type = storage::Config::Gc::Type::PERIODIC,
.interval = std::chrono::milliseconds(100)}});
+ ASSERT_TRUE(storage.CreateIndex(storage.NameToLabel("label")));
+
{
auto acc0 = storage.Access();
for (uint64_t i = 0; i < 1000; ++i) {
diff --git a/tests/unit/storage_v2_indices.cpp b/tests/unit/storage_v2_indices.cpp
index 3e64e5839..9097fb4c4 100644
--- a/tests/unit/storage_v2_indices.cpp
+++ b/tests/unit/storage_v2_indices.cpp
@@ -48,6 +48,154 @@ class IndexTest : public testing::Test {
int vertex_id;
};
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelIndexCreate) {
+ EXPECT_FALSE(storage.LabelIndexExists(label1));
+ EXPECT_EQ(storage.ListAllIndices().label.size(), 0);
+
+ {
+ auto acc = storage.Access();
+ for (int i = 0; i < 10; ++i) {
+ auto vertex = CreateVertex(&acc);
+ ASSERT_NO_ERROR(vertex.AddLabel(i % 2 ? label1 : label2));
+ }
+ ASSERT_NO_ERROR(acc.Commit());
+ }
+
+ EXPECT_TRUE(storage.CreateIndex(label1));
+
+ {
+ auto acc = storage.Access();
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+ UnorderedElementsAre(1, 3, 5, 7, 9));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9));
+ }
+
+ {
+ auto acc = storage.Access();
+ for (int i = 10; i < 20; ++i) {
+ auto vertex = CreateVertex(&acc);
+ ASSERT_NO_ERROR(vertex.AddLabel(i % 2 ? label1 : label2));
+ }
+
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+ UnorderedElementsAre(1, 3, 5, 7, 9));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 11, 13, 15, 17, 19));
+
+ acc.AdvanceCommand();
+
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 11, 13, 15, 17, 19));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 11, 13, 15, 17, 19));
+
+ acc.Abort();
+ }
+
+ {
+ auto acc = storage.Access();
+ for (int i = 10; i < 20; ++i) {
+ auto vertex = CreateVertex(&acc);
+ ASSERT_NO_ERROR(vertex.AddLabel(i % 2 ? label1 : label2));
+ }
+
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+ UnorderedElementsAre(1, 3, 5, 7, 9));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 21, 23, 25, 27, 29));
+
+ acc.AdvanceCommand();
+
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 21, 23, 25, 27, 29));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 21, 23, 25, 27, 29));
+
+ ASSERT_NO_ERROR(acc.Commit());
+ }
+
+ {
+ auto acc = storage.Access();
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 21, 23, 25, 27, 29));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 21, 23, 25, 27, 29));
+
+ acc.AdvanceCommand();
+
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 21, 23, 25, 27, 29));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 21, 23, 25, 27, 29));
+
+ ASSERT_NO_ERROR(acc.Commit());
+ }
+}
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelIndexDrop) {
+ EXPECT_FALSE(storage.LabelIndexExists(label1));
+ EXPECT_EQ(storage.ListAllIndices().label.size(), 0);
+
+ {
+ auto acc = storage.Access();
+ for (int i = 0; i < 10; ++i) {
+ auto vertex = CreateVertex(&acc);
+ ASSERT_NO_ERROR(vertex.AddLabel(i % 2 ? label1 : label2));
+ }
+ ASSERT_NO_ERROR(acc.Commit());
+ }
+
+ EXPECT_TRUE(storage.CreateIndex(label1));
+
+ {
+ auto acc = storage.Access();
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+ UnorderedElementsAre(1, 3, 5, 7, 9));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9));
+ }
+
+ EXPECT_TRUE(storage.DropIndex(label1));
+ EXPECT_FALSE(storage.LabelIndexExists(label1));
+ EXPECT_EQ(storage.ListAllIndices().label.size(), 0);
+
+ EXPECT_FALSE(storage.DropIndex(label1));
+ EXPECT_FALSE(storage.LabelIndexExists(label1));
+ EXPECT_EQ(storage.ListAllIndices().label.size(), 0);
+
+ {
+ auto acc = storage.Access();
+ for (int i = 10; i < 20; ++i) {
+ auto vertex = CreateVertex(&acc);
+ ASSERT_NO_ERROR(vertex.AddLabel(i % 2 ? label1 : label2));
+ }
+ ASSERT_NO_ERROR(acc.Commit());
+ }
+
+ EXPECT_TRUE(storage.CreateIndex(label1));
+ EXPECT_TRUE(storage.LabelIndexExists(label1));
+ EXPECT_THAT(storage.ListAllIndices().label, UnorderedElementsAre(label1));
+
+ {
+ auto acc = storage.Access();
+
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 11, 13, 15, 17, 19));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 11, 13, 15, 17, 19));
+
+ acc.AdvanceCommand();
+
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 11, 13, 15, 17, 19));
+ EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+ UnorderedElementsAre(1, 3, 5, 7, 9, 11, 13, 15, 17, 19));
+ }
+}
+
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_F(IndexTest, LabelIndexBasic) {
// The following steps are performed and index correctness is validated after
@@ -57,14 +205,16 @@ TEST_F(IndexTest, LabelIndexBasic) {
// 3. Remove Label1 from odd numbered vertices, and add it to even numbered
// vertices.
// 4. Delete even numbered vertices.
+ EXPECT_TRUE(storage.CreateIndex(label1));
+ EXPECT_TRUE(storage.CreateIndex(label2));
+
auto acc = storage.Access();
- EXPECT_EQ(storage.ListAllIndices().label.size(), 0);
+ EXPECT_THAT(storage.ListAllIndices().label,
+ UnorderedElementsAre(label1, label2));
EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD), IsEmpty());
EXPECT_THAT(GetIds(acc.Vertices(label2, View::OLD), View::OLD), IsEmpty());
EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW), IsEmpty());
EXPECT_THAT(GetIds(acc.Vertices(label2, View::NEW), View::NEW), IsEmpty());
- EXPECT_THAT(storage.ListAllIndices().label,
- UnorderedElementsAre(label1, label2));
for (int i = 0; i < 10; ++i) {
auto vertex = CreateVertex(&acc);
@@ -133,6 +283,9 @@ TEST_F(IndexTest, LabelIndexDuplicateVersions) {
// By removing labels and adding them again we create duplicate entries for
// the same vertex in the index (they only differ by the timestamp). This test
// checks that duplicates are properly filtered out.
+ EXPECT_TRUE(storage.CreateIndex(label1));
+ EXPECT_TRUE(storage.CreateIndex(label2));
+
{
auto acc = storage.Access();
for (int i = 0; i < 5; ++i) {
@@ -172,6 +325,9 @@ TEST_F(IndexTest, LabelIndexDuplicateVersions) {
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_F(IndexTest, LabelIndexTransactionalIsolation) {
// Check that transactions only see entries they are supposed to see.
+ EXPECT_TRUE(storage.CreateIndex(label1));
+ EXPECT_TRUE(storage.CreateIndex(label2));
+
auto acc_before = storage.Access();
auto acc = storage.Access();
auto acc_after = storage.Access();
@@ -202,6 +358,9 @@ TEST_F(IndexTest, LabelIndexTransactionalIsolation) {
// NOLINTNEXTLINE(hicpp-special-member-functions)
TEST_F(IndexTest, LabelIndexCountEstimate) {
+ EXPECT_TRUE(storage.CreateIndex(label1));
+ EXPECT_TRUE(storage.CreateIndex(label2));
+
auto acc = storage.Access();
for (int i = 0; i < 20; ++i) {
auto vertex = CreateVertex(&acc);