[StorageV2] Implement label and label-property indices

Reviewers: mferencevic, teon.banek

Reviewed By: teon.banek

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D2225

src/storage/v2/CMakeLists.txt

@@ -1,5 +1,6 @@
 set(storage_v2_src_files
     edge_accessor.cpp
+    indices.cpp
     vertex_accessor.cpp
     storage.cpp)
 

src/storage/v2/edge_accessor.cpp

@@ -8,11 +8,11 @@
 namespace storage {
 
 VertexAccessor EdgeAccessor::FromVertex() const {
-  return VertexAccessor{from_vertex_, transaction_};
+  return VertexAccessor{from_vertex_, transaction_, indices_};
 }
 
 VertexAccessor EdgeAccessor::ToVertex() const {
-  return VertexAccessor{to_vertex_, transaction_};
+  return VertexAccessor{to_vertex_, transaction_, indices_};
 }
 
 Result<bool> EdgeAccessor::SetProperty(PropertyId property,

src/storage/v2/edge_accessor.hpp

@@ -12,7 +12,7 @@ namespace storage {
 
 struct Vertex;
 class VertexAccessor;
-class Storage;
+struct Indices;
 
 class EdgeAccessor final {
  private:
@@ -20,12 +20,13 @@ class EdgeAccessor final {
 
  public:
   EdgeAccessor(Edge *edge, EdgeTypeId edge_type, Vertex *from_vertex,
-               Vertex *to_vertex, Transaction *transaction)
+               Vertex *to_vertex, Transaction *transaction, Indices *indices)
       : edge_(edge),
         edge_type_(edge_type),
         from_vertex_(from_vertex),
         to_vertex_(to_vertex),
-        transaction_(transaction) {}
+        transaction_(transaction),
+        indices_(indices) {}
 
   VertexAccessor FromVertex() const;
 
@@ -52,6 +53,7 @@ class EdgeAccessor final {
   Vertex *from_vertex_;
   Vertex *to_vertex_;
   Transaction *transaction_;
+  Indices *indices_;
 };
 
 }  // namespace storage

src/storage/v2/indices.cpp

@@ -0,0 +1,595 @@
+#include "indices.hpp"
+
+#include "storage/v2/mvcc.hpp"
+
+namespace storage {
+
+namespace {
+
+/// We treat doubles and integers the same for label property indices, so
+/// special comparison operators are needed.
+bool PropertyValueLess(const PropertyValue &lhs, const PropertyValue &rhs) {
+  if (lhs.type() == rhs.type()) {
+    return lhs < rhs;
+  }
+  if (lhs.IsInt() && rhs.IsDouble()) {
+    return lhs.ValueInt() < rhs.ValueDouble();
+  }
+  if (lhs.IsDouble() && rhs.IsInt()) {
+    return lhs.ValueDouble() < rhs.ValueInt();
+  }
+  return lhs.type() < rhs.type();
+}
+
+bool PropertyValueEqual(const PropertyValue &lhs, const PropertyValue &rhs) {
+  if (lhs.type() == rhs.type()) {
+    return lhs == rhs;
+  }
+  if (lhs.IsInt() && rhs.IsDouble()) {
+    return lhs.ValueInt() == rhs.ValueDouble();
+  }
+  if (lhs.IsDouble() && rhs.IsInt()) {
+    return lhs.ValueDouble() == rhs.ValueInt();
+  }
+  return false;
+}
+
+/// Traverses deltas visible from transaction with start timestamp greater than
+/// the provided timestamp, and calls the provided callback function for each
+/// delta. If the callback ever returns true, traversal is stopped and the
+/// function returns true. Otherwise, the function returns false.
+template <typename TCallback>
+bool AnyVersionSatisfiesPredicate(uint64_t timestamp, Delta *delta,
+                                  const TCallback &predicate) {
+  while (delta != nullptr) {
+    auto ts = delta->timestamp->load(std::memory_order_acquire);
+    // This is a committed change that we see so we shouldn't undo it.
+    if (ts < timestamp) {
+      break;
+    }
+    if (predicate(*delta)) {
+      return true;
+    }
+    // Move to the next delta.
+    delta = delta->next.load(std::memory_order_acquire);
+  }
+  return false;
+}
+
+/// Helper function for label index garbage collection. Returns true if there's
+/// a reachable version of the vertex that has the given label.
+bool AnyVersionHasLabel(Vertex *vertex, LabelId label, uint64_t timestamp) {
+  bool has_label;
+  bool deleted;
+  Delta *delta;
+  {
+    std::lock_guard<utils::SpinLock> guard(vertex->lock);
+    has_label = utils::Contains(vertex->labels, label);
+    deleted = vertex->deleted;
+    delta = vertex->delta;
+  }
+  if (!deleted && has_label) {
+    return true;
+  }
+  return AnyVersionSatisfiesPredicate(
+      timestamp, delta, [&has_label, &deleted, label](const Delta &delta) {
+        switch (delta.action) {
+          case Delta::Action::ADD_LABEL:
+            if (delta.label == label) {
+              CHECK(!has_label) << "Invalid database state!";
+              has_label = true;
+            }
+            break;
+          case Delta::Action::REMOVE_LABEL:
+            if (delta.label == label) {
+              CHECK(has_label) << "Invalid database state!";
+              has_label = false;
+            }
+            break;
+          case Delta::Action::RECREATE_OBJECT: {
+            CHECK(deleted) << "Invalid database state!";
+            deleted = false;
+            break;
+          }
+          case Delta::Action::DELETE_OBJECT: {
+            CHECK(!deleted) << "Invalid database state!";
+            deleted = true;
+            break;
+          }
+          case Delta::Action::SET_PROPERTY:
+          case Delta::Action::ADD_IN_EDGE:
+          case Delta::Action::ADD_OUT_EDGE:
+          case Delta::Action::REMOVE_IN_EDGE:
+          case Delta::Action::REMOVE_OUT_EDGE:
+            break;
+        }
+        return !deleted && has_label;
+      });
+}
+
+/// Helper function for label-property index garbage collection. Returns true if
+/// there's a reachable version of the vertex that has the given label and
+/// property value.
+bool AnyVersionHasLabelProperty(Vertex *vertex, LabelId label, PropertyId key,
+                                const PropertyValue &value,
+                                uint64_t timestamp) {
+  bool has_label;
+  PropertyValue current_value;
+  bool deleted;
+  Delta *delta;
+  {
+    std::lock_guard<utils::SpinLock> guard(vertex->lock);
+    has_label = utils::Contains(vertex->labels, label);
+    auto it = vertex->properties.find(key);
+    if (it != vertex->properties.end()) {
+      current_value = it->second;
+    }
+    deleted = vertex->deleted;
+    delta = vertex->delta;
+  }
+
+  if (!deleted && has_label && PropertyValueEqual(current_value, value)) {
+    return true;
+  }
+
+  return AnyVersionSatisfiesPredicate(
+      timestamp, delta,
+      [&has_label, &current_value, &deleted, label, key,
+       value](const Delta &delta) {
+        switch (delta.action) {
+          case Delta::Action::ADD_LABEL:
+            if (delta.label == label) {
+              CHECK(!has_label) << "Invalid database state!";
+              has_label = true;
+            }
+            break;
+          case Delta::Action::REMOVE_LABEL:
+            if (delta.label == label) {
+              CHECK(has_label) << "Invalid database state!";
+              has_label = false;
+            }
+            break;
+          case Delta::Action::SET_PROPERTY:
+            if (delta.property.key == key) {
+              current_value = delta.property.value;
+            }
+            break;
+          case Delta::Action::RECREATE_OBJECT: {
+            CHECK(deleted) << "Invalid database state!";
+            deleted = false;
+            break;
+          }
+          case Delta::Action::DELETE_OBJECT: {
+            CHECK(!deleted) << "Invalid database state!";
+            deleted = true;
+            break;
+          }
+          case Delta::Action::ADD_IN_EDGE:
+          case Delta::Action::ADD_OUT_EDGE:
+          case Delta::Action::REMOVE_IN_EDGE:
+          case Delta::Action::REMOVE_OUT_EDGE:
+            break;
+        }
+        return !deleted && has_label &&
+               PropertyValueEqual(current_value, value);
+      });
+}
+
+// Helper function for iterating through label index. Returns true if this
+// transaction can see the given vertex, and the visible version has the given
+// label.
+bool CurrentVersionHasLabel(Vertex *vertex, LabelId label,
+                            Transaction *transaction, View view) {
+  bool deleted;
+  bool has_label;
+  Delta *delta;
+  {
+    std::lock_guard<utils::SpinLock> guard(vertex->lock);
+    deleted = vertex->deleted;
+    has_label = utils::Contains(vertex->labels, label);
+    delta = vertex->delta;
+  }
+  ApplyDeltasForRead(transaction, delta, view,
+                     [&deleted, &has_label, label](const Delta &delta) {
+                       switch (delta.action) {
+                         case Delta::Action::REMOVE_LABEL: {
+                           if (delta.label == label) {
+                             CHECK(has_label) << "Invalid database state!";
+                             has_label = false;
+                           }
+                           break;
+                         }
+                         case Delta::Action::ADD_LABEL: {
+                           if (delta.label == label) {
+                             CHECK(!has_label) << "Invalid database state!";
+                             has_label = true;
+                           }
+                           break;
+                         }
+                         case Delta::Action::DELETE_OBJECT: {
+                           CHECK(!deleted) << "Invalid database state!";
+                           deleted = true;
+                           break;
+                         }
+                         case Delta::Action::RECREATE_OBJECT: {
+                           CHECK(deleted) << "Invalid database state!";
+                           deleted = false;
+                           break;
+                         }
+                         case Delta::Action::SET_PROPERTY:
+                         case Delta::Action::ADD_IN_EDGE:
+                         case Delta::Action::ADD_OUT_EDGE:
+                         case Delta::Action::REMOVE_IN_EDGE:
+                         case Delta::Action::REMOVE_OUT_EDGE:
+                           break;
+                       }
+                     });
+  return !deleted && has_label;
+}
+
+// Helper function for iterating through label-property index. Returns true if
+// this transaction can see the given vertex, and the visible version has the
+// given label and property.
+bool CurrentVersionHasLabelProperty(Vertex *vertex, LabelId label,
+                                    PropertyId key, const PropertyValue &value,
+                                    Transaction *transaction, View view) {
+  bool deleted;
+  bool has_label;
+  PropertyValue current_value;
+  Delta *delta;
+  {
+    std::lock_guard<utils::SpinLock> guard(vertex->lock);
+    deleted = vertex->deleted;
+    has_label = utils::Contains(vertex->labels, label);
+    auto it = vertex->properties.find(key);
+    if (it != vertex->properties.end()) {
+      current_value = it->second;
+    }
+    delta = vertex->delta;
+  }
+  ApplyDeltasForRead(
+      transaction, delta, view,
+      [&deleted, &has_label, &current_value, key, label](const Delta &delta) {
+        switch (delta.action) {
+          case Delta::Action::SET_PROPERTY: {
+            if (delta.property.key == key) {
+              current_value = delta.property.value;
+            }
+            break;
+          }
+          case Delta::Action::DELETE_OBJECT: {
+            CHECK(!deleted) << "Invalid database state!";
+            deleted = true;
+            break;
+          }
+          case Delta::Action::RECREATE_OBJECT: {
+            CHECK(deleted) << "Invalid database state!";
+            deleted = false;
+            break;
+          }
+          case Delta::Action::ADD_LABEL:
+            if (delta.label == label) {
+              CHECK(!has_label) << "Invalid database state!";
+              has_label = true;
+            }
+            break;
+          case Delta::Action::REMOVE_LABEL:
+            if (delta.label == label) {
+              CHECK(has_label) << "Invalid database state!";
+              has_label = false;
+            }
+            break;
+          case Delta::Action::ADD_IN_EDGE:
+          case Delta::Action::ADD_OUT_EDGE:
+          case Delta::Action::REMOVE_IN_EDGE:
+          case Delta::Action::REMOVE_OUT_EDGE:
+            break;
+        }
+      });
+  return !deleted && has_label && PropertyValueEqual(current_value, value);
+}
+
+}  // namespace
+
+void LabelIndex::UpdateOnAddLabel(LabelId label, Vertex *vertex,
+                                  const Transaction &tx) {
+  GetOrCreateStorage(label)->access().insert(Entry{vertex, tx.start_timestamp});
+}
+
+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 it = vertices_acc.begin(); it != vertices_acc.end();) {
+      auto next_it = it;
+      ++next_it;
+
+      if (it->timestamp >= oldest_active_start_timestamp) {
+        it = next_it;
+        continue;
+      }
+
+      if ((next_it != vertices_acc.end() && it->vertex == next_it->vertex) ||
+          !AnyVersionHasLabel(it->vertex, label_storage.label,
+                              oldest_active_start_timestamp)) {
+        vertices_acc.remove(*it);
+      }
+
+      it = next_it;
+    }
+  }
+}
+
+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),
+      index_iterator_(index_iterator),
+      current_vertex_accessor_(nullptr, nullptr, nullptr),
+      current_vertex_(nullptr) {
+  AdvanceUntilValid();
+}
+
+LabelIndex::Iterable::Iterator &LabelIndex::Iterable::Iterator::operator++() {
+  ++index_iterator_;
+  AdvanceUntilValid();
+  return *this;
+}
+
+void LabelIndex::Iterable::Iterator::AdvanceUntilValid() {
+  for (; index_iterator_ != self_->index_accessor_.end(); ++index_iterator_) {
+    if (index_iterator_->vertex == current_vertex_) {
+      continue;
+    }
+    if (CurrentVersionHasLabel(index_iterator_->vertex, self_->label_,
+                               self_->transaction_, self_->view_)) {
+      current_vertex_ = index_iterator_->vertex;
+      current_vertex_accessor_ =
+          VertexAccessor{current_vertex_, self_->transaction_, self_->indices_};
+      break;
+    }
+  }
+}
+
+LabelIndex::Iterable::Iterable(utils::SkipList<Entry>::Accessor index_accessor,
+                               LabelId label, View view,
+                               Transaction *transaction, Indices *indices)
+    : index_accessor_(index_accessor),
+      label_(label),
+      view_(view),
+      transaction_(transaction),
+      indices_(indices) {}
+
+bool LabelPropertyIndex::Entry::operator<(const Entry &rhs) {
+  if (PropertyValueLess(value, rhs.value)) {
+    return true;
+  }
+  if (PropertyValueLess(rhs.value, value)) {
+    return false;
+  }
+  return std::make_tuple(vertex, timestamp) <
+         std::make_tuple(rhs.vertex, rhs.timestamp);
+}
+
+bool LabelPropertyIndex::Entry::operator==(const Entry &rhs) {
+  return PropertyValueEqual(value, rhs.value) && vertex == rhs.vertex &&
+         timestamp == rhs.timestamp;
+}
+
+bool LabelPropertyIndex::Entry::operator<(const PropertyValue &rhs) {
+  return PropertyValueLess(value, rhs);
+}
+
+bool LabelPropertyIndex::Entry::operator==(const PropertyValue &rhs) {
+  return PropertyValueEqual(value, rhs);
+}
+
+void LabelPropertyIndex::UpdateOnAddLabel(LabelId label, Vertex *vertex,
+                                          const Transaction &tx) {
+  for (auto &[label_prop, storage] : index_) {
+    if (label_prop.first != label) {
+      continue;
+    }
+    PropertyValue value;
+    {
+      utils::SpinLock guard(vertex->lock);
+      auto it = vertex->properties.find(label_prop.second);
+      if (it != vertex->properties.end()) {
+        value = it->second;
+      }
+    }
+    if (!value.IsNull()) {
+      storage.access().insert(Entry{value, vertex, tx.start_timestamp});
+    }
+  }
+}
+
+void LabelPropertyIndex::UpdateOnSetProperty(PropertyId property,
+                                             const PropertyValue &value,
+                                             Vertex *vertex,
+                                             const Transaction &tx) {
+  if (value.IsNull()) {
+    return;
+  }
+  for (auto &[label_prop, storage] : index_) {
+    if (label_prop.second != property) {
+      continue;
+    }
+    bool has_label;
+    {
+      utils::SpinLock guard(vertex->lock);
+      has_label = utils::Contains(vertex->labels, label_prop.first);
+    }
+    if (has_label) {
+      storage.access().insert(Entry{value, vertex, tx.start_timestamp});
+    }
+  }
+}
+
+bool LabelPropertyIndex::CreateIndex(
+    LabelId label, PropertyId property,
+    utils::SkipList<Vertex>::Accessor vertices) {
+  auto [it, emplaced] = index_.emplace(std::piecewise_construct,
+                                       std::forward_as_tuple(label, property),
+                                       std::forward_as_tuple());
+  if (!emplaced) {
+    // Index already exists.
+    return false;
+  }
+  auto acc = it->second.access();
+  for (Vertex &vertex : vertices) {
+    if (!utils::Contains(vertex.labels, label)) {
+      continue;
+    }
+    auto it = vertex.properties.find(property);
+    if (it == vertex.properties.end()) {
+      continue;
+    }
+    acc.insert(Entry{it->second, &vertex, 0});
+  }
+  return true;
+}
+
+void LabelPropertyIndex::RemoveObsoleteEntries(
+    uint64_t oldest_active_start_timestamp) {
+  for (auto &[label_property, index] : index_) {
+    auto index_acc = index.access();
+    for (auto it = index_acc.begin(); it != index_acc.end();) {
+      auto next_it = it;
+      ++next_it;
+
+      if (it->timestamp >= oldest_active_start_timestamp) {
+        it = next_it;
+        continue;
+      }
+
+      if ((next_it != index_acc.end() && it->vertex == next_it->vertex &&
+           PropertyValueEqual(it->value, next_it->value)) ||
+          !AnyVersionHasLabelProperty(it->vertex, label_property.first,
+                                      label_property.second, it->value,
+                                      oldest_active_start_timestamp)) {
+        index_acc.remove(*it);
+      }
+      it = next_it;
+    }
+  }
+}
+
+LabelPropertyIndex::Iterable::Iterator::Iterator(
+    Iterable *self, utils::SkipList<Entry>::Iterator index_iterator)
+    : self_(self),
+      index_iterator_(index_iterator),
+      current_vertex_accessor_(nullptr, nullptr, nullptr),
+      current_vertex_(nullptr) {
+  AdvanceUntilValid();
+}
+
+LabelPropertyIndex::Iterable::Iterator &LabelPropertyIndex::Iterable::Iterator::
+operator++() {
+  ++index_iterator_;
+  AdvanceUntilValid();
+  return *this;
+}
+
+void LabelPropertyIndex::Iterable::Iterator::AdvanceUntilValid() {
+  for (; index_iterator_ != self_->index_accessor_.end(); ++index_iterator_) {
+    if (index_iterator_->vertex == current_vertex_) {
+      continue;
+    }
+
+    if (self_->lower_bound_) {
+      if (PropertyValueLess(index_iterator_->value,
+                            self_->lower_bound_->value())) {
+        continue;
+      }
+      if (!self_->lower_bound_->IsInclusive() &&
+          PropertyValueEqual(index_iterator_->value,
+                             self_->lower_bound_->value())) {
+        continue;
+      }
+    }
+    if (self_->upper_bound_) {
+      if (PropertyValueLess(self_->upper_bound_->value(),
+                            index_iterator_->value)) {
+        index_iterator_ = self_->index_accessor_.end();
+        break;
+      }
+      if (!self_->upper_bound_->IsInclusive() &&
+          PropertyValueEqual(index_iterator_->value,
+                             self_->upper_bound_->value())) {
+        index_iterator_ = self_->index_accessor_.end();
+        break;
+      }
+    }
+
+    if (CurrentVersionHasLabelProperty(index_iterator_->vertex, self_->label_,
+                                       self_->property_, index_iterator_->value,
+                                       self_->transaction_, self_->view_)) {
+      current_vertex_ = index_iterator_->vertex;
+      current_vertex_accessor_ =
+          VertexAccessor(current_vertex_, self_->transaction_, self_->indices_);
+      break;
+    }
+  }
+}
+
+LabelPropertyIndex::Iterable::Iterable(
+    utils::SkipList<Entry>::Accessor index_accessor, LabelId label,
+    PropertyId property,
+    const std::optional<utils::Bound<PropertyValue>> &lower_bound,
+    const std::optional<utils::Bound<PropertyValue>> &upper_bound, View view,
+    Transaction *transaction, Indices *indices)
+    : index_accessor_(index_accessor),
+      label_(label),
+      property_(property),
+      lower_bound_(lower_bound),
+      upper_bound_(upper_bound),
+      view_(view),
+      transaction_(transaction),
+      indices_(indices) {}
+
+LabelPropertyIndex::Iterable::Iterator LabelPropertyIndex::Iterable::begin() {
+  auto index_iterator = index_accessor_.begin();
+  if (lower_bound_) {
+    index_iterator =
+        index_accessor_.find_equal_or_greater(lower_bound_->value());
+  }
+  return Iterator(this, index_iterator);
+}
+
+LabelPropertyIndex::Iterable::Iterator LabelPropertyIndex::Iterable::end() {
+  return Iterator(this, index_accessor_.end());
+}
+
+void RemoveObsoleteEntries(Indices *indices,
+                           uint64_t oldest_active_start_timestamp) {
+  indices->label_index.RemoveObsoleteEntries(oldest_active_start_timestamp);
+  indices->label_property_index.RemoveObsoleteEntries(
+      oldest_active_start_timestamp);
+}
+
+void UpdateOnAddLabel(Indices *indices, LabelId label, Vertex *vertex,
+                      const Transaction &tx) {
+  indices->label_index.UpdateOnAddLabel(label, vertex, tx);
+  indices->label_property_index.UpdateOnAddLabel(label, vertex, tx);
+}
+
+void UpdateOnSetProperty(Indices *indices, PropertyId property,
+                         const PropertyValue &value, Vertex *vertex,
+                         const Transaction &tx) {
+  indices->label_property_index.UpdateOnSetProperty(property, value, vertex,
+                                                    tx);
+}
+
+}  // namespace storage

src/storage/v2/indices.hpp

@@ -0,0 +1,233 @@
+#pragma once
+
+#include <optional>
+#include <tuple>
+#include <utility>
+
+#include "storage/v2/property_value.hpp"
+#include "storage/v2/transaction.hpp"
+#include "storage/v2/vertex_accessor.hpp"
+#include "utils/bound.hpp"
+#include "utils/skip_list.hpp"
+
+namespace storage {
+
+struct Indices;
+
+class LabelIndex {
+ private:
+  struct Entry {
+    Vertex *vertex;
+    uint64_t timestamp;
+
+    bool operator<(const Entry &rhs) {
+      return std::make_tuple(vertex, timestamp) <
+             std::make_tuple(rhs.vertex, rhs.timestamp);
+    }
+    bool operator==(const Entry &rhs) {
+      return vertex == rhs.vertex && timestamp == rhs.timestamp;
+    }
+  };
+
+  struct LabelStorage {
+    LabelId label;
+    utils::SkipList<Entry> vertices;
+
+    bool operator<(const LabelStorage &rhs) { return label < rhs.label; }
+    bool operator<(LabelId rhs) { return label < rhs; }
+    bool operator==(const LabelStorage &rhs) { return label == rhs.label; }
+    bool operator==(LabelId rhs) { return label == rhs; }
+  };
+
+ public:
+  explicit LabelIndex(Indices *indices) : indices_(indices) {}
+
+  void UpdateOnAddLabel(LabelId label, Vertex *vertex, const Transaction &tx);
+
+  void RemoveObsoleteEntries(uint64_t oldest_active_start_timestamp);
+
+  class Iterable {
+   public:
+    Iterable(utils::SkipList<Entry>::Accessor index_accessor, LabelId label,
+             View view, Transaction *transaction, Indices *indices);
+
+    class Iterator {
+     public:
+      Iterator(Iterable *self, utils::SkipList<Entry>::Iterator index_iterator);
+
+      VertexAccessor operator*() const { return current_vertex_accessor_; }
+
+      bool operator==(const Iterator &other) const {
+        return index_iterator_ == other.index_iterator_;
+      }
+      bool operator!=(const Iterator &other) const {
+        return index_iterator_ != other.index_iterator_;
+      }
+
+      Iterator &operator++();
+
+     private:
+      void AdvanceUntilValid();
+
+      Iterable *self_;
+      utils::SkipList<Entry>::Iterator index_iterator_;
+      VertexAccessor current_vertex_accessor_;
+      Vertex *current_vertex_;
+    };
+
+    Iterator begin() { return Iterator(this, index_accessor_.begin()); }
+    Iterator end() { return Iterator(this, index_accessor_.end()); }
+
+   private:
+    utils::SkipList<Entry>::Accessor index_accessor_;
+    LabelId label_;
+    View view_;
+    Transaction *transaction_;
+    Indices *indices_;
+  };
+
+  /// Returns an self with vertices visible from the given transaction.
+  Iterable Vertices(LabelId label, View view, Transaction *transaction) {
+    return Iterable(GetOrCreateStorage(label)->access(), label, view,
+                    transaction, indices_);
+  }
+
+ private:
+  utils::SkipList<LabelStorage> index_;
+  Indices *indices_;
+
+  utils::SkipList<Entry> *GetOrCreateStorage(LabelId label);
+};
+
+class LabelPropertyIndex {
+ private:
+  struct Entry {
+    PropertyValue value;
+    Vertex *vertex;
+    uint64_t timestamp;
+
+    bool operator<(const Entry &rhs);
+    bool operator==(const Entry &rhs);
+
+    bool operator<(const PropertyValue &rhs);
+    bool operator==(const PropertyValue &rhs);
+  };
+
+ public:
+  explicit LabelPropertyIndex(Indices *indices) : indices_(indices) {}
+
+  void UpdateOnAddLabel(LabelId label, Vertex *vertex, const Transaction &tx);
+
+  void UpdateOnSetProperty(PropertyId property, const PropertyValue &value,
+                           Vertex *vertex, const Transaction &tx);
+
+  bool CreateIndex(LabelId label, PropertyId property,
+                   utils::SkipList<Vertex>::Accessor vertices);
+
+  bool DropIndex(LabelId label, PropertyId property) {
+    return index_.erase({label, property}) > 0;
+  }
+
+  bool IndexExists(LabelId label, PropertyId property) {
+    return index_.find({label, property}) != index_.end();
+  }
+
+  void RemoveObsoleteEntries(uint64_t oldest_active_start_timestamp);
+
+  class Iterable {
+   public:
+    Iterable(utils::SkipList<Entry>::Accessor index_accessor, LabelId label,
+             PropertyId property,
+             const std::optional<utils::Bound<PropertyValue>> &lower_bound,
+             const std::optional<utils::Bound<PropertyValue>> &upper_bound,
+             View view, Transaction *transaction, Indices *indices);
+
+    class Iterator {
+     public:
+      Iterator(Iterable *self, utils::SkipList<Entry>::Iterator index_iterator);
+
+      VertexAccessor operator*() const { return current_vertex_accessor_; }
+
+      bool operator==(const Iterator &other) const {
+        return index_iterator_ == other.index_iterator_;
+      }
+      bool operator!=(const Iterator &other) const {
+        return index_iterator_ != other.index_iterator_;
+      }
+
+      Iterator &operator++();
+
+     private:
+      void AdvanceUntilValid();
+
+      Iterable *self_;
+      utils::SkipList<Entry>::Iterator index_iterator_;
+      VertexAccessor current_vertex_accessor_;
+      Vertex *current_vertex_;
+    };
+
+    Iterator begin();
+    Iterator end();
+
+   private:
+    utils::SkipList<Entry>::Accessor index_accessor_;
+    LabelId label_;
+    PropertyId property_;
+    std::optional<utils::Bound<PropertyValue>> lower_bound_;
+    std::optional<utils::Bound<PropertyValue>> upper_bound_;
+    View view_;
+    Transaction *transaction_;
+    Indices *indices_;
+  };
+
+  Iterable Vertices(
+      LabelId label, PropertyId property,
+      const std::optional<utils::Bound<PropertyValue>> &lower_bound,
+      const std::optional<utils::Bound<PropertyValue>> &upper_bound, View view,
+      Transaction *transaction) {
+    auto it = index_.find({label, property});
+    CHECK(it != index_.end())
+        << "Index for label " << label.AsUint() << " and property "
+        << property.AsUint() << " doesn't exist";
+    return Iterable(it->second.access(), label, property, lower_bound,
+                    upper_bound, view, transaction, indices_);
+  }
+
+ private:
+  Indices *indices_;
+  std::map<std::pair<LabelId, PropertyId>, utils::SkipList<Entry>> index_;
+};
+
+struct Indices {
+  Indices() : label_index(this), label_property_index(this) {}
+
+  // Disable copy and move because members hold pointer to `this`.
+  Indices(const Indices &) = delete;
+  Indices(Indices &&) = delete;
+  Indices &operator=(const Indices &) = delete;
+  Indices &operator=(Indices &&) = delete;
+  ~Indices() = default;
+
+  LabelIndex label_index;
+  LabelPropertyIndex label_property_index;
+};
+
+/// This function should be called from garbage collection to clean-up the
+/// index.
+void RemoveObsoleteEntries(Indices *indices,
+                           uint64_t oldest_active_start_timestamp);
+
+// Indices are updated whenever an update occurs, instead of only on commit or
+// advance command. This is necessary because we want indices to support `NEW`
+// view for use in Merge.
+
+/// This function should be called whenever a label is added to a vertex.
+void UpdateOnAddLabel(Indices *indices, LabelId label, Vertex *vertex,
+                      const Transaction &tx);
+
+/// This function should be called whenever a property is modified on a vertex.
+void UpdateOnSetProperty(Indices *indices, PropertyId property,
+                         const PropertyValue &value, Vertex *vertex,
+                         const Transaction &tx);
+
+}  // namespace storage

src/storage/v2/storage.cpp

@@ -8,13 +8,11 @@
 
 namespace storage {
 
-namespace {
-
 auto AdvanceToVisibleVertex(utils::SkipList<Vertex>::Iterator it,
                             utils::SkipList<Vertex>::Iterator end,
-                            Transaction *tx, View view) {
+                            Transaction *tx, View view, Indices *indices) {
   while (it != end) {
-    auto maybe_vertex = VertexAccessor::Create(&*it, tx, view);
+    auto maybe_vertex = VertexAccessor::Create(&*it, tx, indices, view);
     if (!maybe_vertex) {
       ++it;
       continue;
@@ -24,22 +22,25 @@ auto AdvanceToVisibleVertex(utils::SkipList<Vertex>::Iterator it,
   return it;
 }
 
-}  // namespace
-
 VerticesIterable::Iterator::Iterator(VerticesIterable *self,
                                      utils::SkipList<Vertex>::Iterator it)
     : self_(self),
       it_(AdvanceToVisibleVertex(it, self->vertices_accessor_.end(),
-                                 self->transaction_, self->view_)) {}
+                                 self->transaction_, self->view_,
+                                 self->indices_)) {}
 
 VertexAccessor VerticesIterable::Iterator::operator*() const {
-  return *VertexAccessor::Create(&*it_, self_->transaction_, self_->view_);
+  // TODO: current vertex accessor could be cached to avoid reconstructing every
+  // time
+  return *VertexAccessor::Create(&*it_, self_->transaction_, self_->indices_,
+                                 self_->view_);
 }
 
 VerticesIterable::Iterator &VerticesIterable::Iterator::operator++() {
   ++it_;
   it_ = AdvanceToVisibleVertex(it_, self_->vertices_accessor_.end(),
-                               self_->transaction_, self_->view_);
+                               self_->transaction_, self_->view_,
+                               self_->indices_);
   return *this;
 }
 
@@ -88,7 +89,7 @@ VertexAccessor Storage::Accessor::CreateVertex() {
   CHECK(inserted) << "The vertex must be inserted here!";
   CHECK(it != acc.end()) << "Invalid Vertex accessor!";
   delta->prev.Set(&*it);
-  return VertexAccessor{&*it, &transaction_};
+  return VertexAccessor{&*it, &transaction_, &storage_->indices_};
 }
 
 std::optional<VertexAccessor> Storage::Accessor::FindVertex(Gid gid,
@@ -96,7 +97,7 @@ std::optional<VertexAccessor> Storage::Accessor::FindVertex(Gid gid,
   auto acc = storage_->vertices_.access();
   auto it = acc.find(gid);
   if (it == acc.end()) return std::nullopt;
-  return VertexAccessor::Create(&*it, &transaction_, view);
+  return VertexAccessor::Create(&*it, &transaction_, &storage_->indices_, view);
 }
 
 Result<bool> Storage::Accessor::DeleteVertex(VertexAccessor *vertex) {
@@ -144,7 +145,8 @@ Result<bool> Storage::Accessor::DetachDeleteVertex(VertexAccessor *vertex) {
 
   for (const auto &item : in_edges) {
     auto [edge_type, from_vertex, edge] = item;
-    EdgeAccessor e{edge, edge_type, from_vertex, vertex_ptr, &transaction_};
+    EdgeAccessor e{edge,       edge_type,     from_vertex,
+                   vertex_ptr, &transaction_, &storage_->indices_};
     auto ret = DeleteEdge(&e);
     if (ret.HasError()) {
       CHECK(ret.GetError() == Error::SERIALIZATION_ERROR)
@@ -154,7 +156,8 @@ Result<bool> Storage::Accessor::DetachDeleteVertex(VertexAccessor *vertex) {
   }
   for (const auto &item : out_edges) {
     auto [edge_type, to_vertex, edge] = item;
-    EdgeAccessor e{edge, edge_type, vertex_ptr, to_vertex, &transaction_};
+    EdgeAccessor e{edge,      edge_type,     vertex_ptr,
+                   to_vertex, &transaction_, &storage_->indices_};
     auto ret = DeleteEdge(&e);
     if (ret.HasError()) {
       CHECK(ret.GetError() == Error::SERIALIZATION_ERROR)
@@ -235,8 +238,9 @@ Result<EdgeAccessor> Storage::Accessor::CreateEdge(VertexAccessor *from,
                      edge_type, from_vertex, edge);
   to_vertex->in_edges.emplace_back(edge_type, from_vertex, edge);
 
-  return Result<EdgeAccessor>{
+  return Result<EdgeAccessor>{EdgeAccessor{edge, edge_type, from_vertex,
-      EdgeAccessor{edge, edge_type, from_vertex, to_vertex, &transaction_}};
+                                           to_vertex, &transaction_,
+                                           &storage_->indices_}};
 }
 
 Result<bool> Storage::Accessor::DeleteEdge(EdgeAccessor *edge) {
@@ -587,6 +591,32 @@ Storage::Accessor Storage::Access() {
   return Accessor{this, transaction_id, start_timestamp};
 }
 
+LabelIndex::Iterable Storage::Accessor::Vertices(LabelId label, View view) {
+  return storage_->indices_.label_index.Vertices(label, view, &transaction_);
+}
+
+LabelPropertyIndex::Iterable Storage::Accessor::Vertices(LabelId label,
+                                                         PropertyId property,
+                                                         View view) {
+  return storage_->indices_.label_property_index.Vertices(
+      label, property, std::nullopt, std::nullopt, view, &transaction_);
+}
+
+LabelPropertyIndex::Iterable Storage::Accessor::Vertices(
+    LabelId label, PropertyId property, const PropertyValue &value, View view) {
+  return storage_->indices_.label_property_index.Vertices(
+      label, property, utils::MakeBoundInclusive(value),
+      utils::MakeBoundInclusive(value), view, &transaction_);
+}
+
+LabelPropertyIndex::Iterable Storage::Accessor::Vertices(
+    LabelId label, PropertyId property,
+    const std::optional<utils::Bound<PropertyValue>> &lower_bound,
+    const std::optional<utils::Bound<PropertyValue>> &upper_bound, View view) {
+  return storage_->indices_.label_property_index.Vertices(
+      label, property, lower_bound, upper_bound, view, &transaction_);
+}
+
 void Storage::CollectGarbage() {
   // Garbage collection must be performed in two phases. In the first phase,
   // deltas that won't be applied by any transaction anymore are unlinked from
@@ -606,9 +636,9 @@ void Storage::CollectGarbage() {
   // garbage_undo_buffers lock.
   std::list<std::pair<uint64_t, std::list<Delta>>> unlinked_undo_buffers;
 
-  // We will free only vertices deleted up until now in this GC cycle.
+  // We will only free vertices deleted up until now in this GC cycle, and we
-  // Otherwise, GC cycle might be prolonged by aborted transactions adding new
+  // will do it after cleaning-up the indices. That way we are sure that all
-  // edges and vertices to the lists over and over again.
+  // vertices that appear in an index also exist in main storage.
   std::list<Gid> current_deleted_edges;
   std::list<Gid> current_deleted_vertices;
   deleted_vertices_->swap(current_deleted_vertices);
@@ -697,10 +727,15 @@ void Storage::CollectGarbage() {
     });
   }
 
+  // After unlinking deltas from vertices, we refresh the indices. That way
+  // we're sure that none of the vertices from `current_deleted_vertices`
+  // appears in an index, and we can safely remove the from the main storage
+  // after the last currently active transaction is finished.
+  RemoveObsoleteEntries(&indices_, oldest_active_start_timestamp);
+
   {
-    uint64_t mark_timestamp;
     std::unique_lock<utils::SpinLock> guard(engine_lock_);
-    mark_timestamp = timestamp_;
+    uint64_t mark_timestamp = timestamp_;
     // Take garbage_undo_buffers lock while holding the engine lock to make
     // sure that entries are sorted by mark timestamp in the list.
     garbage_undo_buffers_.WithLock([&](auto &garbage_undo_buffers) {
@@ -716,6 +751,9 @@ void Storage::CollectGarbage() {
       garbage_undo_buffers.splice(garbage_undo_buffers.end(),
                                   unlinked_undo_buffers);
     });
+    for (auto vertex : current_deleted_vertices) {
+      garbage_vertices_.emplace_back(mark_timestamp, vertex);
+    }
   }
 
   while (true) {
@@ -730,8 +768,11 @@ void Storage::CollectGarbage() {
 
   {
     auto vertex_acc = vertices_.access();
-    for (auto vertex : current_deleted_vertices) {
+    while (!garbage_vertices_.empty() &&
-      CHECK(vertex_acc.remove(vertex)) << "Invalid database state!";
+           garbage_vertices_.front().first < oldest_active_start_timestamp) {
+      CHECK(vertex_acc.remove(garbage_vertices_.front().second))
+          << "Invalid database state!";
+      garbage_vertices_.pop_front();
     }
   }
   {

src/storage/v2/storage.hpp

@@ -6,6 +6,7 @@
 #include "storage/v2/commit_log.hpp"
 #include "storage/v2/edge.hpp"
 #include "storage/v2/edge_accessor.hpp"
+#include "storage/v2/indices.hpp"
 #include "storage/v2/mvcc.hpp"
 #include "storage/v2/name_id_mapper.hpp"
 #include "storage/v2/result.hpp"
@@ -50,6 +51,7 @@ class VerticesIterable final {
   utils::SkipList<Vertex>::Accessor vertices_accessor_;
   Transaction *transaction_;
   View view_;
+  Indices *indices_;
 
   class Iterator final {
     VerticesIterable *self_;
@@ -71,10 +73,11 @@ class VerticesIterable final {
 
  public:
   VerticesIterable(utils::SkipList<Vertex>::Accessor vertices_accessor,
-                   Transaction *transaction, View view)
+                   Transaction *transaction, View view, Indices *indices)
       : vertices_accessor_(std::move(vertices_accessor)),
         transaction_(transaction),
-        view_(view) {}
+        view_(view),
+        indices_(indices) {}
 
   Iterator begin() { return Iterator(this, vertices_accessor_.begin()); }
   Iterator end() { return Iterator(this, vertices_accessor_.end()); }
@@ -109,10 +112,25 @@ class Storage final {
     std::optional<VertexAccessor> FindVertex(Gid gid, View view);
 
     VerticesIterable Vertices(View view) {
-      return VerticesIterable(storage_->vertices_.access(), &transaction_,
+      return VerticesIterable(storage_->vertices_.access(), &transaction_, view,
-                              view);
+                              &storage_->indices_);
     }
 
+    LabelIndex::Iterable Vertices(LabelId label, View view);
+
+    LabelPropertyIndex::Iterable Vertices(LabelId label, PropertyId property,
+                                          View view);
+
+    LabelPropertyIndex::Iterable Vertices(LabelId label, PropertyId property,
+                                          const PropertyValue &value,
+                                          View view);
+
+    LabelPropertyIndex::Iterable Vertices(
+        LabelId label, PropertyId property,
+        const std::optional<utils::Bound<PropertyValue>> &lower_bound,
+        const std::optional<utils::Bound<PropertyValue>> &upper_bound,
+        View view);
+
     Result<bool> DeleteVertex(VertexAccessor *vertex);
 
     Result<bool> DetachDeleteVertex(VertexAccessor *vertex);
@@ -147,6 +165,21 @@ class Storage final {
 
   Accessor Access();
 
+  bool CreateIndex(LabelId label, PropertyId property) {
+    std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+    return indices_.label_property_index.CreateIndex(label, property,
+                                                     vertices_.access());
+  }
+
+  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) {
+    return indices_.label_property_index.IndexExists(label, property);
+  }
+
  private:
   void CollectGarbage();
 
@@ -166,6 +199,8 @@ class Storage final {
 
   NameIdMapper name_id_mapper_;
 
+  Indices indices_;
+
   // Transaction engine
   utils::SpinLock engine_lock_;
   uint64_t timestamp_{kTimestampInitialId};
@@ -188,9 +223,16 @@ class Storage final {
                       utils::SpinLock>
       garbage_undo_buffers_;
 
-  // Vertices that are logically deleted and now are waiting to be removed from
+  // Vertices that are logically deleted but still have to be removed from
-  // the main storage.
+  // indices before removing them from the main storage.
   utils::Synchronized<std::list<Gid>, utils::SpinLock> deleted_vertices_;
+
+  // Vertices that are logically deleted and removed from indices and now wait
+  // to be removed from the main storage.
+  std::list<std::pair<uint64_t, Gid>> garbage_vertices_;
+
+  // Edges that are logically deleted and wait to be removed from the main
+  // storage.
   utils::Synchronized<std::list<Gid>, utils::SpinLock> deleted_edges_;
 };
 

src/storage/v2/vertex_accessor.cpp

@@ -3,12 +3,14 @@
 #include <memory>
 
 #include "storage/v2/edge_accessor.hpp"
+#include "storage/v2/indices.hpp"
 #include "storage/v2/mvcc.hpp"
 
 namespace storage {
 
 std::optional<VertexAccessor> VertexAccessor::Create(Vertex *vertex,
                                                      Transaction *transaction,
+                                                     Indices *indices,
                                                      View view) {
   bool is_visible = true;
   Delta *delta = nullptr;
@@ -39,7 +41,7 @@ std::optional<VertexAccessor> VertexAccessor::Create(Vertex *vertex,
                        }
                      });
   if (!is_visible) return std::nullopt;
-  return VertexAccessor{vertex, transaction};
+  return VertexAccessor{vertex, transaction, indices};
 }
 
 Result<bool> VertexAccessor::AddLabel(LabelId label) {
@@ -57,6 +59,9 @@ Result<bool> VertexAccessor::AddLabel(LabelId label) {
   CreateAndLinkDelta(transaction_, vertex_, Delta::RemoveLabelTag(), label);
 
   vertex_->labels.push_back(label);
+
+  UpdateOnAddLabel(indices_, label, vertex_, *transaction_);
+
   return Result<bool>{true};
 }
 
@@ -203,6 +208,8 @@ Result<bool> VertexAccessor::SetProperty(PropertyId property,
     }
   }
 
+  UpdateOnSetProperty(indices_, property, value, vertex_, *transaction_);
+
   return Result<bool>{existed};
 }
 
@@ -362,7 +369,8 @@ Result<std::vector<EdgeAccessor>> VertexAccessor::InEdges(
     auto [edge_type, from_vertex, edge] = item;
     if (edge_types.empty() || std::find(edge_types.begin(), edge_types.end(),
                                         edge_type) != edge_types.end()) {
-      ret.emplace_back(edge, edge_type, from_vertex, vertex_, transaction_);
+      ret.emplace_back(edge, edge_type, from_vertex, vertex_, transaction_,
+                       indices_);
     }
   }
   return Result<decltype(ret)>(std::move(ret));
@@ -428,7 +436,8 @@ Result<std::vector<EdgeAccessor>> VertexAccessor::OutEdges(
     auto [edge_type, to_vertex, edge] = item;
     if (edge_types.empty() || std::find(edge_types.begin(), edge_types.end(),
                                         edge_type) != edge_types.end()) {
-      ret.emplace_back(edge, edge_type, vertex_, to_vertex, transaction_);
+      ret.emplace_back(edge, edge_type, vertex_, to_vertex, transaction_,
+                       indices_);
     }
   }
   return Result<decltype(ret)>(std::move(ret));

src/storage/v2/vertex_accessor.hpp

@@ -12,18 +12,19 @@ namespace storage {
 
 class EdgeAccessor;
 class Storage;
+struct Indices;
 
 class VertexAccessor final {
  private:
   friend class Storage;
 
  public:
-  VertexAccessor(Vertex *vertex, Transaction *transaction)
+  VertexAccessor(Vertex *vertex, Transaction *transaction, Indices *indices)
-      : vertex_(vertex), transaction_(transaction) {}
+      : vertex_(vertex), transaction_(transaction), indices_(indices) {}
 
   static std::optional<VertexAccessor> Create(Vertex *vertex,
                                               Transaction *transaction,
-                                              View view);
+                                              Indices *indices, View view);
 
   Result<bool> AddLabel(LabelId label);
 
@@ -57,6 +58,7 @@ class VertexAccessor final {
  private:
   Vertex *vertex_;
   Transaction *transaction_;
+  Indices *indices_;
 };
 
 }  // namespace storage

tests/benchmark/storage_v2_gc.cpp

@@ -18,9 +18,6 @@ DEFINE_int32(num_iterations, kNumIterations, "number of iterations");
 std::pair<std::string, storage::StorageGcConfig> TestConfigurations[] = {
     {"NoGc",
      storage::StorageGcConfig{.type = storage::StorageGcConfig::Type::NONE}},
-    {"OnFinishGc",
-     storage::StorageGcConfig{.type =
-                                  storage::StorageGcConfig::Type::ON_FINISH}},
     {"100msPeriodicGc",
      storage::StorageGcConfig{.type = storage::StorageGcConfig::Type::PERIODIC,
                               .interval = std::chrono::milliseconds(100)}},

tests/unit/CMakeLists.txt

@@ -319,6 +319,9 @@ target_link_libraries(${test_prefix}storage_v2 mg-storage-v2)
 add_unit_test(storage_v2_gc.cpp)
 target_link_libraries(${test_prefix}storage_v2_gc mg-storage-v2)
 
+add_unit_test(storage_v2_indices.cpp)
+target_link_libraries(${test_prefix}storage_v2_indices mg-storage-v2)
+
 add_unit_test(storage_v2_name_id_mapper.cpp)
 target_link_libraries(${test_prefix}storage_v2_name_id_mapper mg-storage-v2)
 

tests/unit/storage_v2_gc.cpp

@@ -5,8 +5,9 @@
 
 using testing::UnorderedElementsAre;
 
-// TODO: We should implement a more sophisticated stress test to verify that GC
+// TODO: The point of these is not to test GC fully, these are just simple
-// is working properly in a multithreaded environment.
+// sanity checks. These will be superseded by a more sophisticated stress test
+// which will verify that GC is working properly in a multithreaded environment.
 
 // A simple test trying to get GC to run while a transaction is still alive and
 // then verify that GC didn't delete anything it shouldn't have.
@@ -150,3 +151,45 @@ TEST(StorageV2Gc, Sanity) {
     acc.Commit();
   }
 }
+
+// A simple sanity check for index GC:
+// 1. Start transaction 0, create some vertices, add a label to them and
+//    commit.
+// 2. Start transaction 1.
+// 3. Start transaction 2, remove the labels and commit;
+// 4. Wait for GC. GC shouldn't remove the vertices from index because
+//    transaction 1 can still see them with that label.
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST(StorageV2Gc, Indices) {
+  storage::Storage storage(
+      storage::StorageGcConfig{.type = storage::StorageGcConfig::Type::PERIODIC,
+                               .interval = std::chrono::milliseconds(100)});
+
+  {
+    auto acc0 = storage.Access();
+    for (uint64_t i = 0; i < 1000; ++i) {
+      auto vertex = acc0.CreateVertex();
+      ASSERT_TRUE(*vertex.AddLabel(acc0.NameToLabel("label")));
+    }
+    acc0.Commit();
+  }
+  {
+    auto acc1 = storage.Access();
+
+    auto acc2 = storage.Access();
+    for (auto vertex : acc2.Vertices(storage::View::OLD)) {
+      ASSERT_TRUE(*vertex.RemoveLabel(acc2.NameToLabel("label")));
+    }
+    acc2.Commit();
+
+    // Wait for GC.
+    std::this_thread::sleep_for(std::chrono::milliseconds(300));
+
+    std::set<storage::Gid> gids;
+    for (auto vertex :
+         acc1.Vertices(acc1.NameToLabel("label"), storage::View::OLD)) {
+      gids.insert(vertex.Gid());
+    }
+    EXPECT_EQ(gids.size(), 1000);
+  }
+}

tests/unit/storage_v2_indices.cpp

@@ -0,0 +1,468 @@
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "storage/v2/storage.hpp"
+
+// NOLINTNEXTLINE(google-build-using-namespace)
+using namespace storage;
+
+LabelId nil_label = LabelId::FromUint(0);
+PropertyId nil_property = PropertyId::FromUint(0);
+
+using testing::IsEmpty;
+using testing::UnorderedElementsAre;
+
+// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
+#define ASSERT_NO_ERROR(result) ASSERT_FALSE((result).HasError())
+
+class IndexTest : public testing::Test {
+ protected:
+  IndexTest()
+      : prop_id(nil_property),
+        prop_val(nil_property),
+        label1(nil_label),
+        label2(nil_label) {}
+
+  void SetUp() override {
+    auto acc = storage.Access();
+    prop_id = acc.NameToProperty("id");
+    prop_val = acc.NameToProperty("val");
+    label1 = acc.NameToLabel("label1");
+    label2 = acc.NameToLabel("label2");
+    vertex_id = 0;
+  }
+
+  Storage storage;
+  PropertyId prop_id;
+  PropertyId prop_val;
+  LabelId label1;
+  LabelId label2;
+
+  VertexAccessor CreateVertex(Storage::Accessor *accessor) {
+    VertexAccessor vertex = accessor->CreateVertex();
+    CHECK(!vertex.SetProperty(prop_id, PropertyValue(vertex_id++)).HasError());
+    return vertex;
+  }
+
+  template <class TIterable>
+  std::vector<int64_t> GetIds(TIterable iterable, View view = View::OLD) {
+    std::vector<int64_t> ret;
+    for (auto vertex : iterable) {
+      ret.push_back(vertex.GetProperty(prop_id, view)->ValueInt());
+    }
+    return ret;
+  }
+
+ private:
+  int vertex_id;
+};
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelIndexBasic) {
+  // The following steps are performed and index correctness is validated after
+  // each step:
+  // 1. Create 10 vertices numbered from 0 to 9.
+  // 2. Add Label1 to odd numbered, and Label2 to even numbered vertices.
+  // 3. Remove Label1 from odd numbered vertices, and add it to even numbered
+  //    vertices.
+  // 4. Delete even numbered vertices.
+  auto acc = storage.Access();
+  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());
+
+  for (int i = 0; i < 10; ++i) {
+    auto vertex = CreateVertex(&acc);
+    ASSERT_NO_ERROR(vertex.AddLabel(i % 2 ? 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),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+
+  acc.AdvanceCommand();
+  EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, View::OLD), View::OLD),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+  EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+
+  for (auto vertex : acc.Vertices(View::OLD)) {
+    int64_t id = vertex.GetProperty(prop_id, View::OLD)->ValueInt();
+    if (id % 2) {
+      ASSERT_NO_ERROR(vertex.RemoveLabel(label1));
+    } else {
+      ASSERT_NO_ERROR(vertex.AddLabel(label1));
+    }
+  }
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, View::OLD), View::OLD),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+  EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+
+  for (auto vertex : acc.Vertices(View::OLD)) {
+    int64_t id = vertex.GetProperty(prop_id, View::OLD)->ValueInt();
+    if (id % 2 == 0) {
+      ASSERT_NO_ERROR(acc.DeleteVertex(&vertex));
+    }
+  }
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, View::OLD), View::OLD),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+  EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW), IsEmpty());
+  EXPECT_THAT(GetIds(acc.Vertices(label2, View::NEW), View::NEW), IsEmpty());
+
+  acc.AdvanceCommand();
+
+  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());
+}
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+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.
+  {
+    auto acc = storage.Access();
+    for (int i = 0; i < 5; ++i) {
+      auto vertex = CreateVertex(&acc);
+      ASSERT_NO_ERROR(vertex.AddLabel(label1));
+    }
+
+    EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+
+    acc.Commit();
+  }
+
+  {
+    auto acc = storage.Access();
+    EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+
+    for (auto vertex : acc.Vertices(View::OLD)) {
+      ASSERT_NO_ERROR(vertex.RemoveLabel(label1));
+    }
+
+    EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+    EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW), IsEmpty());
+
+    for (auto vertex : acc.Vertices(View::OLD)) {
+      ASSERT_NO_ERROR(vertex.AddLabel(label1));
+    }
+    EXPECT_THAT(GetIds(acc.Vertices(label1, View::OLD), View::OLD),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+    EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+  }
+}
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelIndexTransactionalIsolation) {
+  // Check that transactions only see entries they are supposed to see.
+  auto acc_before = storage.Access();
+  auto acc = storage.Access();
+  auto acc_after = storage.Access();
+
+  for (int i = 0; i < 5; ++i) {
+    auto vertex = CreateVertex(&acc);
+    ASSERT_NO_ERROR(vertex.AddLabel(label1));
+  }
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 1, 2, 3, 4));
+  EXPECT_THAT(GetIds(acc_before.Vertices(label1, View::NEW), View::NEW),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc_after.Vertices(label1, View::NEW), View::NEW),
+              IsEmpty());
+
+  acc.Commit();
+
+  auto acc_after_commit = storage.Access();
+
+  EXPECT_THAT(GetIds(acc_before.Vertices(label1, View::NEW), View::NEW),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc_after.Vertices(label1, View::NEW), View::NEW),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc_after_commit.Vertices(label1, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 1, 2, 3, 4));
+}
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelPropertyIndexCreateAndDrop) {
+  EXPECT_TRUE(storage.CreateIndex(label1, prop_id));
+  EXPECT_TRUE(storage.LabelPropertyIndexExists(label1, prop_id));
+  EXPECT_FALSE(storage.LabelPropertyIndexExists(label2, prop_id));
+  EXPECT_FALSE(storage.CreateIndex(label1, prop_id));
+
+  EXPECT_TRUE(storage.CreateIndex(label2, prop_id));
+  EXPECT_TRUE(storage.LabelPropertyIndexExists(label2, prop_id));
+
+  EXPECT_TRUE(storage.DropIndex(label1, prop_id));
+  EXPECT_FALSE(storage.LabelPropertyIndexExists(label1, prop_id));
+  EXPECT_FALSE(storage.DropIndex(label1, prop_id));
+
+  EXPECT_TRUE(storage.DropIndex(label2, prop_id));
+  EXPECT_FALSE(storage.LabelPropertyIndexExists(label2, prop_id));
+}
+
+// The following three tests are almost an exact copy-paste of the corresponding
+// label index tests. We request all vertices with given label and property from
+// the index, without range filtering. Range filtering is tested in a separate
+// test.
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelPropertyIndexBasic) {
+  storage.CreateIndex(label1, prop_val);
+  storage.CreateIndex(label2, prop_val);
+
+  auto acc = storage.Access();
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+              IsEmpty());
+
+  for (int i = 0; i < 10; ++i) {
+    auto vertex = CreateVertex(&acc);
+    ASSERT_NO_ERROR(vertex.AddLabel(i % 2 ? label1 : label2));
+    ASSERT_NO_ERROR(vertex.SetProperty(prop_val, PropertyValue(i)));
+  }
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::OLD), View::OLD),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+
+  acc.AdvanceCommand();
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::OLD), View::OLD),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+
+  for (auto vertex : acc.Vertices(View::OLD)) {
+    int64_t id = vertex.GetProperty(prop_id, View::OLD)->ValueInt();
+    if (id % 2) {
+      ASSERT_NO_ERROR(vertex.SetProperty(prop_val, PropertyValue()));
+    } else {
+      ASSERT_NO_ERROR(vertex.AddLabel(label1));
+    }
+  }
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::OLD), View::OLD),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+
+  for (auto vertex : acc.Vertices(View::OLD)) {
+    int64_t id = vertex.GetProperty(prop_id, View::OLD)->ValueInt();
+    if (id % 2 == 0) {
+      ASSERT_NO_ERROR(acc.DeleteVertex(&vertex));
+    }
+  }
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+              UnorderedElementsAre(1, 3, 5, 7, 9));
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::OLD), View::OLD),
+              UnorderedElementsAre(0, 2, 4, 6, 8));
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::NEW), View::NEW),
+              IsEmpty());
+
+  acc.AdvanceCommand();
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::OLD), View::OLD),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+              IsEmpty());
+  EXPECT_THAT(GetIds(acc.Vertices(label2, prop_val, View::NEW), View::NEW),
+              IsEmpty());
+}
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelPropertyIndexDuplicateVersions) {
+  storage.CreateIndex(label1, prop_val);
+  {
+    auto acc = storage.Access();
+    for (int i = 0; i < 5; ++i) {
+      auto vertex = CreateVertex(&acc);
+      ASSERT_NO_ERROR(vertex.AddLabel(label1));
+      ASSERT_NO_ERROR(vertex.SetProperty(prop_val, PropertyValue(i)));
+    }
+
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+
+    acc.Commit();
+  }
+
+  {
+    auto acc = storage.Access();
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+
+    for (auto vertex : acc.Vertices(View::OLD)) {
+      ASSERT_NO_ERROR(vertex.SetProperty(prop_val, PropertyValue()));
+    }
+
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+                IsEmpty());
+
+    for (auto vertex : acc.Vertices(View::OLD)) {
+      ASSERT_NO_ERROR(vertex.SetProperty(prop_val, PropertyValue(42)));
+    }
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::OLD), View::OLD),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+                UnorderedElementsAre(0, 1, 2, 3, 4));
+  }
+}
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelPropertyIndexTransactionalIsolation) {
+  storage.CreateIndex(label1, prop_val);
+
+  auto acc_before = storage.Access();
+  auto acc = storage.Access();
+  auto acc_after = storage.Access();
+
+  for (int i = 0; i < 5; ++i) {
+    auto vertex = CreateVertex(&acc);
+    ASSERT_NO_ERROR(vertex.AddLabel(label1));
+    ASSERT_NO_ERROR(vertex.SetProperty(prop_val, PropertyValue(i)));
+  }
+
+  EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, View::NEW), View::NEW),
+              UnorderedElementsAre(0, 1, 2, 3, 4));
+  EXPECT_THAT(
+      GetIds(acc_before.Vertices(label1, prop_val, View::NEW), View::NEW),
+      IsEmpty());
+  EXPECT_THAT(
+      GetIds(acc_after.Vertices(label1, prop_val, View::NEW), View::NEW),
+      IsEmpty());
+
+  acc.Commit();
+
+  auto acc_after_commit = storage.Access();
+
+  EXPECT_THAT(
+      GetIds(acc_before.Vertices(label1, prop_val, View::NEW), View::NEW),
+      IsEmpty());
+  EXPECT_THAT(
+      GetIds(acc_after.Vertices(label1, prop_val, View::NEW), View::NEW),
+      IsEmpty());
+  EXPECT_THAT(
+      GetIds(acc_after_commit.Vertices(label1, prop_val, View::NEW), View::NEW),
+      UnorderedElementsAre(0, 1, 2, 3, 4));
+}
+
+// NOLINTNEXTLINE(hicpp-special-member-functions)
+TEST_F(IndexTest, LabelPropertyIndexFiltering) {
+  // We insert vertices with values:
+  // 0 0.0 1 1.0 2 2.0 3 3.0 4 4.0
+  // Then we check all combinations of inclusive and exclusive bounds.
+  // We also have a mix of doubles and integers to verify that they are sorted
+  // properly.
+
+  storage.CreateIndex(label1, prop_val);
+
+  {
+    auto acc = storage.Access();
+
+    for (int i = 0; i < 10; ++i) {
+      auto vertex = CreateVertex(&acc);
+      ASSERT_NO_ERROR(vertex.AddLabel(label1));
+      ASSERT_NO_ERROR(vertex.SetProperty(
+          prop_val, i % 2 ? PropertyValue(i / 2) : PropertyValue(i / 2.0)));
+    }
+    acc.Commit();
+  }
+  {
+    auto acc = storage.Access();
+    for (int i = 0; i < 5; ++i) {
+      EXPECT_THAT(
+          GetIds(acc.Vertices(label1, prop_val, PropertyValue(i), View::OLD)),
+          UnorderedElementsAre(2 * i, 2 * i + 1));
+    }
+
+    // [1, +inf>
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val,
+                                    utils::MakeBoundInclusive(PropertyValue(1)),
+                                    std::nullopt, View::OLD)),
+                UnorderedElementsAre(2, 3, 4, 5, 6, 7, 8, 9));
+    // <1, +inf>
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val,
+                                    utils::MakeBoundExclusive(PropertyValue(1)),
+                                    std::nullopt, View::OLD)),
+                UnorderedElementsAre(4, 5, 6, 7, 8, 9));
+
+    // <-inf, 3]
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, std::nullopt,
+                                    utils::MakeBoundInclusive(PropertyValue(3)),
+                                    View::OLD)),
+                UnorderedElementsAre(0, 1, 2, 3, 4, 5, 6, 7));
+    // <-inf, 3>
+    EXPECT_THAT(GetIds(acc.Vertices(label1, prop_val, std::nullopt,
+                                    utils::MakeBoundExclusive(PropertyValue(3)),
+                                    View::OLD)),
+                UnorderedElementsAre(0, 1, 2, 3, 4, 5));
+
+    // [1, 3]
+    EXPECT_THAT(
+        GetIds(acc.Vertices(
+            label1, prop_val, utils::MakeBoundInclusive(PropertyValue(1)),
+            utils::MakeBoundInclusive(PropertyValue(3)), View::OLD)),
+        UnorderedElementsAre(2, 3, 4, 5, 6, 7));
+    // <1, 3]
+    EXPECT_THAT(
+        GetIds(acc.Vertices(
+            label1, prop_val, utils::MakeBoundExclusive(PropertyValue(1)),
+            utils::MakeBoundInclusive(PropertyValue(3)), View::OLD)),
+        UnorderedElementsAre(4, 5, 6, 7));
+    // [1, 3>
+    EXPECT_THAT(
+        GetIds(acc.Vertices(
+            label1, prop_val, utils::MakeBoundInclusive(PropertyValue(1)),
+            utils::MakeBoundExclusive(PropertyValue(3)), View::OLD)),
+        UnorderedElementsAre(2, 3, 4, 5));
+    // <1, 3>
+    EXPECT_THAT(
+        GetIds(acc.Vertices(
+            label1, prop_val, utils::MakeBoundExclusive(PropertyValue(1)),
+            utils::MakeBoundExclusive(PropertyValue(3)), View::OLD)),
+        UnorderedElementsAre(4, 5));
+  }
+}