Storage::RecordAccessor - added multiple pointers to RecordAcessors and switching capabilities. Rewired GraphDbAccessor and VersionList accordingly. Added tests for new RecordAccessor functions.

Reviewers: teon.banek, dtomicevic, mislav.bradac, dgleich, buda Reviewed By: buda Subscribers: pullbot Differential Revision: https://phabricator.memgraph.io/D207
2017-04-03 11:12:15 +02:00 · 2017-04-03 11:12:15 +02:00 · 9ce2081103
commit 9ce2081103
parent 863f55dc2f
14 changed files with 304 additions and 147 deletions
--- a/src/database/graph_db_accessor.cpp
+++ b/src/database/graph_db_accessor.cpp
@ -38,46 +38,48 @@ void GraphDbAccessor::abort() {

 VertexAccessor GraphDbAccessor::insert_vertex() {
  // create a vertex
-  Vertex *vertex = nullptr;
-  auto vertex_vlist = new mvcc::VersionList<Vertex>(*transaction_, vertex);
+  auto vertex_vlist = new mvcc::VersionList<Vertex>(*transaction_);

  bool success = db_.vertices_.access().insert(vertex_vlist).second;
-  if (success) return VertexAccessor(*vertex_vlist, *vertex, *this);
+  if (success) return VertexAccessor(*vertex_vlist, *this);
  throw CreationException("Unable to create a Vertex after 5 attempts");
 }

 bool GraphDbAccessor::remove_vertex(VertexAccessor &vertex_accessor) {
-  // TODO consider if this works well with MVCC
+  vertex_accessor.SwitchNew();
  if (vertex_accessor.out_degree() > 0 || vertex_accessor.in_degree() > 0)
    return false;

-  vertex_accessor.vlist_->remove(&vertex_accessor.update(), *transaction_);
+  vertex_accessor.vlist_->remove(vertex_accessor.current_, *transaction_);
  return true;
 }

 void GraphDbAccessor::detach_remove_vertex(VertexAccessor &vertex_accessor) {
-  // removing edges via accessors is both safe
-  // and it should remove all the pointers in the relevant
-  // vertices (including this one)
-  vertex_accessor.vlist_->remove(&vertex_accessor.update(), *transaction_);
+  vertex_accessor.SwitchNew();
  for (auto edge_accessor : vertex_accessor.in()) remove_edge(edge_accessor);
+  vertex_accessor.SwitchNew();
  for (auto edge_accessor : vertex_accessor.out()) remove_edge(edge_accessor);
+  vertex_accessor.vlist_->remove(vertex_accessor.SwitchNew().current_, *transaction_);
 }

 EdgeAccessor GraphDbAccessor::insert_edge(VertexAccessor &from,
                                          VertexAccessor &to,
                                          GraphDbTypes::EdgeType edge_type) {
  // create an edge
-  Edge *edge = nullptr;
  auto edge_vlist = new mvcc::VersionList<Edge>(
-      *transaction_, edge, *from.vlist_, *to.vlist_, edge_type);
+      *transaction_, *from.vlist_, *to.vlist_, edge_type);

-  // set the vertex connections to this edge
+  // ensure that the "from" accessor has the latest version
+  from.SwitchNew();
  from.update().out_.emplace_back(edge_vlist);
+  // ensure that the "to" accessor has the latest version
+  // WARNING: must do that after the above "from.update()" for cases when
+  // we are creating a cycle and "from" and "to" are the same vlist
+  to.SwitchNew();
  to.update().in_.emplace_back(edge_vlist);

  bool success = db_.edges_.access().insert(edge_vlist).second;
-  if (success) return EdgeAccessor(*edge_vlist, *edge, *this);
+  if (success) return EdgeAccessor(*edge_vlist, *this);

  throw CreationException("Unable to create an Edge after 5 attempts");
 }
@ -97,7 +99,7 @@ void swap_out_edge(std::vector<mvcc::VersionList<Edge> *> &edges,
 void GraphDbAccessor::remove_edge(EdgeAccessor &edge_accessor) {
  swap_out_edge(edge_accessor.from().update().out_, edge_accessor.vlist_);
  swap_out_edge(edge_accessor.to().update().in_, edge_accessor.vlist_);
-  edge_accessor.vlist_->remove(&edge_accessor.update(), *transaction_);
+  edge_accessor.vlist_->remove(edge_accessor.SwitchNew().current_, *transaction_);
 }

 GraphDbTypes::Label GraphDbAccessor::label(const std::string &label_name) {
--- a/src/database/graph_db_accessor.hpp
+++ b/src/database/graph_db_accessor.hpp
@ -77,15 +77,17 @@ class GraphDbAccessor {
   * visible to the current transaction.
   */
  auto vertices() {
-    // filter out the accessors not visible to the current transaction
-    auto filtered = iter::filter(
-        [this](auto vlist) { return vlist->find(*transaction_) != nullptr; },
-        db_.vertices_.access());
+    // wrap version lists into accessors, which will look for visible versions
+    auto accessors =
+        iter::imap([this](auto vlist) { return VertexAccessor(*vlist, *this); },
+                   db_.vertices_.access());

-    // return accessors of the filtered out vlists
-    return iter::imap(
-        [this](auto vlist) { return VertexAccessor(*vlist, *this); },
-        std::move(filtered));
+    // filter out the accessors not visible to the current transaction
+    return iter::filter(
+        [this](const VertexAccessor &accessor) {
+          return accessor.old_ != nullptr;
+        },
+        std::move(accessors));
  }

  /**
@ -111,15 +113,17 @@ class GraphDbAccessor {
   * visible to the current transaction.
   */
  auto edges() {
-    // filter out the accessors not visible to the current transaction
-    auto filtered = iter::filter(
-        [this](auto vlist) { return vlist->find(*transaction_) != nullptr; },
-        db_.edges_.access());
+    // wrap version lists into accessors, which will look for visible versions
+    auto accessors =
+        iter::imap([this](auto vlist) { return EdgeAccessor(*vlist, *this); },
+                   db_.edges_.access());

-    // return accessors of the filtered out vlists
-    return iter::imap(
-        [this](auto vlist) { return EdgeAccessor(*vlist, *this); },
-        std::move(filtered));
+    // filter out the accessors not visible to the current transaction
+    return iter::filter(
+        [this](const EdgeAccessor &accessor) {
+          return accessor.old_ != nullptr;
+        },
+        std::move(accessors));
  }

  /**
@ -211,22 +215,48 @@ class GraphDbAccessor {
  void abort();

  /**
-   * Init accessor record with vlist.
-   * @args accessor whose record to initialize.
+   * Initializes the record pointers in the given accessor.
+   * The old_ and new_ pointers need to be initialized
+   * with appropriate values, and current_ set to old_
+   * if it exists and to new_ otherwise.
   */
  template <typename TRecord>
-  void init_record(RecordAccessor<TRecord> &accessor) {
-    accessor.record_ = accessor.vlist_->find(*transaction_);
+  void Reconstruct(RecordAccessor<TRecord> &accessor) {
+    accessor.vlist_->find_set_new_old(*transaction_, accessor.old_,
+                                      accessor.new_);
+    accessor.current_ = accessor.old_ ? accessor.old_ : accessor.new_;
+    // TODO review: we should never use a record accessor that
+    // does not have either old_ or new_ (both are null), but
+    // we can't assert that here because we construct such an accessor
+    // and filter it out in GraphDbAccessor::[Vertices|Edges]
+    // any ideas?
  }

  /**
   * Update accessor record with vlist.
+   *
+   * It is not legal
+   * to call this function on a Vertex/Edge that has
+   * been deleted in the current transaction+command.
+   *
   * @args accessor whose record to update if possible.
   */
  template <typename TRecord>
  void update(RecordAccessor<TRecord> &accessor) {
-    if (!accessor.record_->is_visible_write(*transaction_))
-      accessor.record_ = accessor.vlist_->update(*transaction_);
+    // can't update a deleted record if:
+    // - we only have old_ and it hasn't been deleted
+    // - we have new_ and it hasn't been deleted
+    if (!accessor.new_) {
+      debug_assert(
+          !accessor.old_->is_deleted_by(*transaction_),
+          "Can't update a record deleted in the current transaction+command");
+    } else {
+      debug_assert(
+          !accessor.new_->is_deleted_by(*transaction_),
+          "Can't update a record deleted in the current transaction+command");
+    }
+
+    if (!accessor.new_) accessor.new_ = accessor.vlist_->update(*transaction_);
  }

 private:
--- a/src/mvcc/record.hpp
+++ b/src/mvcc/record.hpp
@ -125,6 +125,22 @@ class Record : public Version<T> {
              cmd.exp() >= t.cid)));  // but not before this command,
  }

+  /**
+   * True if this record is created in the current command
+   * of the given transaction.
+   */
+  bool is_created_by(const tx::Transaction &t) {
+    return tx.cre() == t.id && cmd.cre() == t.cid;
+  }
+
+  /**
+   * True if this record is deleted in the current command
+   * of the given transaction.
+   */
+  bool is_deleted_by(const tx::Transaction &t) {
+    return tx.exp() == t.id && cmd.exp() == t.cid;
+  }
+
 protected:
  template <class U>
  bool committed(U &hints, const Id &id, const tx::Transaction &t) {
--- a/src/mvcc/version_list.hpp
+++ b/src/mvcc/version_list.hpp
@ -18,13 +18,18 @@ class VersionList {

  /* @brief Constructor that is used to insert one item into VersionList.
     @param t - transaction
-     @param T item - item which will point to new and only entry in
-     version_list.
-     @param args - args forwarded to constructor of item T.
+     @param args - args forwarded to constructor of item T (for
+     creating the first Record (Version) in this VersionList.
   */
  template <typename... Args>
-  VersionList(tx::Transaction &t, T *&item, Args &&... args) {
-    item = insert(t, std::forward<Args>(args)...);
+  VersionList(tx::Transaction &t, Args &&... args) {
+    // TODO replace 'new' with something better
+    auto v1 = new T(std::forward<Args>(args)...);
+
+    // mark the record as created by the transaction t
+    v1->mark_created(t);
+
+    head.store(v1, std::memory_order_seq_cst);
  }

  VersionList() = delete;
@ -127,6 +132,34 @@ class VersionList {
    return r;
  }

+  /**
+   * Looks for and sets two versions. The 'old' version is the
+   * newest version that is visible by the current transaction+command,
+   * but has not been created by it. The 'new' version is the version
+   * that has been created by current transaction+command.
+   *
+   * It is possible that both, either or neither are found:
+   *     - both are found when an existing record has been modified
+   *     - only old is found when an existing record has not been modified
+   *     - only new is found when the whole vlist was created
+   *     - neither is found when for example the record has been deleted but not
+   *       garbage collected yet
+   *
+   * @param t The transaction
+   */
+  void find_set_new_old(const tx::Transaction &t, T *&old_ref,
+                        T *&new_ref) const {
+    // assume that the sought old record is further down the list
+    // from new record, so that if we found old we can stop looking
+    new_ref = nullptr;
+    old_ref = head.load(std::memory_order_seq_cst);
+    while (old_ref != nullptr && !old_ref->visible(t)) {
+      if (!new_ref && old_ref->is_created_by(t))
+        new_ref = old_ref;
+      old_ref = old_ref->next(std::memory_order_seq_cst);
+    }
+  }
+
  T *update(tx::Transaction &t) {
    debug_assert(head != nullptr, "Head is nullptr on update.");
    auto record = find(t);
@ -190,27 +223,6 @@ class VersionList {
    throw SerializationError();
  }

-  /**
-   * This is private because this should only be called from the constructor.
-   * Otherwise head might be nullptr while version_list exists and it could
-   * interfere with GC.
-   * @tparam Args forwarded to the constructor of T
-   */
-  template <typename... Args>
-  T *insert(tx::Transaction &t, Args &&... args) {
-    debug_assert(head == nullptr, "Head is not nullptr on creation.");
-
-    // create a first version of the record
-    // TODO replace 'new' with something better
-    auto v1 = new T(std::forward<Args>(args)...);
-
-    // mark the record as created by the transaction t
-    v1->mark_created(t);
-
-    head.store(v1, std::memory_order_seq_cst);
-    return v1;
-  }
-
  std::atomic<T *> head{nullptr};
  RecordLock lock;
 };
--- a/src/storage/edge_accessor.cpp
+++ b/src/storage/edge_accessor.cpp
@ -5,12 +5,12 @@ void EdgeAccessor::set_edge_type(GraphDbTypes::EdgeType edge_type) {
  update().edge_type_ = edge_type;
 }

-GraphDbTypes::EdgeType EdgeAccessor::edge_type() const { return view().edge_type_; }
+GraphDbTypes::EdgeType EdgeAccessor::edge_type() const { return current().edge_type_; }

 VertexAccessor EdgeAccessor::from() const {
-  return VertexAccessor(view().from_, db_accessor());
+  return VertexAccessor(current().from_, db_accessor());
 }

 VertexAccessor EdgeAccessor::to() const {
-  return VertexAccessor(view().to_, db_accessor());
+  return VertexAccessor(current().to_, db_accessor());
 }
--- a/src/storage/record_accessor.cpp
+++ b/src/storage/record_accessor.cpp
@ -7,23 +7,14 @@
 template <typename TRecord>
 RecordAccessor<TRecord>::RecordAccessor(mvcc::VersionList<TRecord> &vlist,
                                        GraphDbAccessor &db_accessor)
-    : db_accessor_(&db_accessor), vlist_(&vlist), record_(nullptr) {
-  db_accessor.init_record(*this);
-  debug_assert(record_ != nullptr, "Record is nullptr.");
-}
-
-template <typename TRecord>
-RecordAccessor<TRecord>::RecordAccessor(mvcc::VersionList<TRecord> &vlist,
-                                        TRecord &record,
-                                        GraphDbAccessor &db_accessor)
-    : db_accessor_(&db_accessor), vlist_(&vlist), record_(&record) {
-  debug_assert(record_ != nullptr, "Record is nullptr.");
+    : db_accessor_(&db_accessor), vlist_(&vlist) {
+  Reconstruct();
 }

 template <typename TRecord>
 const PropertyValue &RecordAccessor<TRecord>::PropsAt(
    GraphDbTypes::Property key) const {
-  return view().properties_.at(key);
+  return current().properties_.at(key);
 }

 template <typename TRecord>
@ -39,7 +30,7 @@ void RecordAccessor<TRecord>::PropsClear() {
 template <typename TRecord>
 const PropertyValueStore<GraphDbTypes::Property>
    &RecordAccessor<TRecord>::Properties() const {
-  return view().properties_;
+  return current().properties_;
 }

 template <typename TRecord>
@ -48,7 +39,7 @@ void RecordAccessor<TRecord>::PropertiesAccept(
                       const PropertyValue &prop)>
        handler,
    std::function<void()> finish) const {
-  view().properties_.Accept(handler, finish);
+  current().properties_.Accept(handler, finish);
 }

 template <typename TRecord>
@ -58,35 +49,54 @@ GraphDbAccessor &RecordAccessor<TRecord>::db_accessor() const {

 template <typename TRecord>
 const uint64_t RecordAccessor<TRecord>::temporary_id() const {
-  return (uint64_t) vlist_;
+  return (uint64_t)vlist_;
 }

 template <typename TRecord>
 RecordAccessor<TRecord> &RecordAccessor<TRecord>::SwitchNew() {
-  // TODO implement
+  if (!new_) {
+    // if new_ is not set yet, look for it
+    // we can just Reconstruct the pointers, old_ will get initialized
+    // to the same value as it has now, and the amount of work is the
+    // same as just looking for a new_ record
+    Reconstruct();
+  }
+  // set new if we have it
+  // if we don't then current_ is old_ and remains there
+  if (new_) current_ = new_;
+  debug_assert(current_ != nullptr,
+               "RecordAccessor::SwitchNew - current_ is nullptr");
  return *this;
 }

 template <typename TRecord>
 RecordAccessor<TRecord> &RecordAccessor<TRecord>::SwitchOld() {
-  // TODO implement
+  // if this whole record is new (new version-list) then we don't
+  // have a valid old_ version. in such a situation SwitchOld
+  // is not a legal function call
+  debug_assert(old_ != nullptr,
+               "RecordAccessor.old_ is nullptr and SwitchOld called");
+  current_ = old_;
  return *this;
 }

 template <typename TRecord>
 void RecordAccessor<TRecord>::Reconstruct() {
-  // TODO implement
+  db_accessor().Reconstruct(*this);
 }

 template <typename TRecord>
 TRecord &RecordAccessor<TRecord>::update() {
  db_accessor().update(*this);
-  return *record_;
+  debug_assert(new_ != nullptr, "RecordAccessor.new_ is null after update");
+  return *new_;
 }

 template <typename TRecord>
-const TRecord &RecordAccessor<TRecord>::view() const {
-  return *record_;
+const TRecord &RecordAccessor<TRecord>::current() const {
+  debug_assert(current_ != nullptr,
+               "RecordAccessor.current_ pointer is nullptr");
+  return *current_;
 }

 template class RecordAccessor<Vertex>;
--- a/src/storage/record_accessor.hpp
+++ b/src/storage/record_accessor.hpp
@ -39,18 +39,6 @@ class RecordAccessor {
  RecordAccessor(mvcc::VersionList<TRecord> &vlist,
                 GraphDbAccessor &db_accessor);

-  /**
-   * @param vlist MVCC record that this accessor wraps.
-   * @param record MVCC version (that is viewable from this
-   * db_accessor.transaction)
-   *  of the given record. Slightly more optimal then the constructor that does
-   * not
-   *  accept an already found record.
-   * @param db_accessor The DB accessor that "owns" this record accessor.
-   */
-  RecordAccessor(mvcc::VersionList<TRecord> &vlist, TRecord &record,
-                 GraphDbAccessor &db_accessor);
-
  // this class is default copyable, movable and assignable
  RecordAccessor(const RecordAccessor &other) = default;
  RecordAccessor(RecordAccessor &&other) = default;
@ -160,6 +148,9 @@ class RecordAccessor {
   * Switches this record accessor to use the old
   * (not updated) version visible to the current transaction+command.
   *
+   * It is not legal to call this function on a Vertex/Edge that
+   * was created by the current transaction+command.
+   *
   * @return A reference to this.
   */
  RecordAccessor<TRecord> &SwitchOld();
@ -175,18 +166,21 @@ class RecordAccessor {

 protected:
  /**
-   * Returns the update-ready version of the record.
+   * Ensures there is an updateable version of the record
+   * in the version_list, and that the `new_` pointer
+   * points to it. Returns a reference to that version.
   *
   * @return See above.
   */
  TRecord &update();

  /**
-   * Returns a version of the record that is only for viewing.
+   * Returns the current version (either new_ or old_)
+   * set on this RecordAccessor.
   *
   * @return See above.
   */
-  const TRecord &view() const;
+  const TRecord &current() const;

 private:
  // The database accessor for which this record accessor is created
@ -198,16 +192,34 @@ class RecordAccessor {
  // Immutable, set in the constructor and never changed.
  mvcc::VersionList<TRecord> *vlist_;

-  /* The version of the record currently used in this transaction. Defaults to
-   * the
-   * latest viewable version (set in the constructor). After the first update
-   * done
-   * through this accessor a new, editable version, is created for this
-   * transaction,
-   * and set as the value of this variable.
+  /**
+   * Latest version which is visible to the current transaction+command
+   * but has not been created nor modified by the current transaction+command.
   *
-   * Stored as a pointer due to it's mutability (the update() function changes
-   * it).
+   * Can be null only when the record itself (the version-list) has
+   * been created by the current transaction+command.
   */
-  TRecord *record_;
+  TRecord *old_{nullptr};
+
+  /**
+   * Version that has been modified (created or updated) by the current
+   * transaction+command.
+   *
+   * Can be null when the record has not been modified in the current
+   * transaction+command. It is also possible that the modification
+   * has happened, but this RecordAccessor does not know this. To
+   * ensure correctness, the `SwitchNew` function must check if this
+   * is null, and if it is it must check with the vlist_ if there is
+   * an update.
+   */
+  TRecord *new_{nullptr};
+
+  /**
+   * Pointer to the version (either old_ or new_) that READ operations
+   * in the accessor should take data from. Note that WRITE operations
+   * should always use new_.
+   *
+   * This pointer can never ever be null.
+   */
+  TRecord *current_{nullptr};
 };
--- a/src/storage/vertex_accessor.cpp
+++ b/src/storage/vertex_accessor.cpp
@ -5,12 +5,12 @@
 #include "storage/util.hpp"
 #include "storage/vertex_accessor.hpp"

-size_t VertexAccessor::out_degree() const { return view().out_.size(); }
+size_t VertexAccessor::out_degree() const { return current().out_.size(); }

-size_t VertexAccessor::in_degree() const { return view().in_.size(); }
+size_t VertexAccessor::in_degree() const { return current().in_.size(); }

 bool VertexAccessor::add_label(GraphDbTypes::Label label) {
-  auto &labels_view = view().labels_;
+  auto &labels_view = current().labels_;
  auto found = std::find(labels_view.begin(), labels_view.end(), label);
  if (found != labels_view.end()) return false;

@ -31,10 +31,10 @@ size_t VertexAccessor::remove_label(GraphDbTypes::Label label) {
 }

 bool VertexAccessor::has_label(GraphDbTypes::Label label) const {
-  auto &labels = this->view().labels_;
+  auto &labels = this->current().labels_;
  return std::find(labels.begin(), labels.end(), label) != labels.end();
 }

 const std::vector<GraphDbTypes::Label> &VertexAccessor::labels() const {
-  return this->view().labels_;
+  return this->current().labels_;
 }
--- a/src/storage/vertex_accessor.hpp
+++ b/src/storage/vertex_accessor.hpp
@ -66,10 +66,10 @@ class VertexAccessor : public RecordAccessor<Vertex> {
  /**
   * Returns EdgeAccessors for all incoming edges.
   */
-  auto in() { return make_accessor_iterator<EdgeAccessor>(view().in_, db_accessor()); }
+  auto in() { return make_accessor_iterator<EdgeAccessor>(current().in_, db_accessor()); }

  /**
   * Returns EdgeAccessors for all outgoing edges.
   */
-  auto out() { return make_accessor_iterator<EdgeAccessor>(view().out_, db_accessor()); }
+  auto out() { return make_accessor_iterator<EdgeAccessor>(current().out_, db_accessor()); }
 };
--- a/tests/unit/graph_db_accessor.cpp
+++ b/tests/unit/graph_db_accessor.cpp
@ -162,38 +162,42 @@ TEST(GraphDbAccessorTest, DetachRemoveVertex) {
  Dbms dbms;
  auto dba = dbms.active();

-  // setup (v1)- []->(v2)<-[]-(v3)<-[]-(v4)
-  auto va1 = dba->insert_vertex();
-  auto va2 = dba->insert_vertex();
-  auto va3 = dba->insert_vertex();
-  auto va4 = dba->insert_vertex();
+  // setup (v0)- []->(v1)<-[]-(v2)<-[]-(v3)
+  std::vector<VertexAccessor> vertices;
+  for (int i = 0; i < 4; ++i) vertices.emplace_back(dba->insert_vertex());
+
  auto edge_type = dba->edge_type("type");
-  dba->insert_edge(va1, va2, edge_type);
-  dba->insert_edge(va1, va3, edge_type);
-  dba->insert_edge(va4, va3, edge_type);
+  dba->insert_edge(vertices[0], vertices[1], edge_type);
+  dba->insert_edge(vertices[2], vertices[1], edge_type);
+  dba->insert_edge(vertices[3], vertices[2], edge_type);
+
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();

  // ensure that plain remove does NOT work
  EXPECT_EQ(CountVertices(*dba), 4);
  EXPECT_EQ(CountEdges(*dba), 3);
-  EXPECT_FALSE(dba->remove_vertex(va1));
-  EXPECT_FALSE(dba->remove_vertex(va2));
-  EXPECT_FALSE(dba->remove_vertex(va3));
+  EXPECT_FALSE(dba->remove_vertex(vertices[0]));
+  EXPECT_FALSE(dba->remove_vertex(vertices[1]));
+  EXPECT_FALSE(dba->remove_vertex(vertices[2]));
  EXPECT_EQ(CountVertices(*dba), 4);
  EXPECT_EQ(CountEdges(*dba), 3);

-  dba->detach_remove_vertex(va3);
+  dba->detach_remove_vertex(vertices[2]);
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();

  EXPECT_EQ(CountVertices(*dba), 3);
  EXPECT_EQ(CountEdges(*dba), 1);
-  EXPECT_TRUE(dba->remove_vertex(va4));
+  EXPECT_TRUE(dba->remove_vertex(vertices[3]));
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();

  EXPECT_EQ(CountVertices(*dba), 2);
  EXPECT_EQ(CountEdges(*dba), 1);
  for (auto va : dba->vertices()) EXPECT_FALSE(dba->remove_vertex(va));
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();

  EXPECT_EQ(CountVertices(*dba), 2);
  EXPECT_EQ(CountEdges(*dba), 1);
@ -203,6 +207,7 @@ TEST(GraphDbAccessorTest, DetachRemoveVertex) {
    break;
  }
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();

  EXPECT_EQ(CountVertices(*dba), 1);
  EXPECT_EQ(CountEdges(*dba), 0);
@ -228,12 +233,13 @@ TEST(GraphDbAccessorTest, DetachRemoveVertexMultiple) {
  int N = 7;
  std::vector<VertexAccessor> vertices;
  auto edge_type = dba->edge_type("edge");
-  for (int i = 0; i < N; ++i)
-    vertices.emplace_back(dba->insert_vertex());
+  for (int i = 0; i < N; ++i) vertices.emplace_back(dba->insert_vertex());
  for (int j = 0; j < N; ++j)
    for (int k = 0; k < N; ++k)
      dba->insert_edge(vertices[j], vertices[k], edge_type);
+
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();

  EXPECT_EQ(CountVertices(*dba), N);
  EXPECT_EQ(CountEdges(*dba), N * N);
@ -241,6 +247,7 @@ TEST(GraphDbAccessorTest, DetachRemoveVertexMultiple) {
  // detach delete one edge
  dba->detach_remove_vertex(vertices[0]);
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();
  EXPECT_EQ(CountVertices(*dba), N - 1);
  EXPECT_EQ(CountEdges(*dba), (N - 1) * (N - 1));

@ -248,13 +255,14 @@ TEST(GraphDbAccessorTest, DetachRemoveVertexMultiple) {
  dba->detach_remove_vertex(vertices[1]);
  dba->detach_remove_vertex(vertices[2]);
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();
  EXPECT_EQ(CountVertices(*dba), N - 3);
  EXPECT_EQ(CountEdges(*dba), (N - 3) * (N - 3));

  // detach delete everything, buwahahahaha
-  for (int l = 3; l < N ; ++l)
-    dba->detach_remove_vertex(vertices[l]);
+  for (int l = 3; l < N; ++l) dba->detach_remove_vertex(vertices[l]);
  dba->advance_command();
+  for (auto &vertex : vertices) vertex.Reconstruct();
  EXPECT_EQ(CountVertices(*dba), 0);
  EXPECT_EQ(CountEdges(*dba), 0);
 }
@ -303,6 +311,6 @@ TEST(GraphDbAccessorTest, Properties) {

 int main(int argc, char **argv) {
  ::testing::InitGoogleTest(&argc, argv);
-//  ::testing::GTEST_FLAG(filter) = "*.DetachRemoveVertex";
+  //  ::testing::GTEST_FLAG(filter) = "*.DetachRemoveVertex";
  return RUN_ALL_TESTS();
 }
--- a/tests/unit/interpreter.cpp
+++ b/tests/unit/interpreter.cpp
@ -1000,6 +1000,7 @@ TEST(Interpreter, SetLabels) {
  EXPECT_EQ(2, PullAll(label_set, *dba, symbol_table));

  for (VertexAccessor vertex : dba->vertices()) {
+    vertex.SwitchNew();
    EXPECT_EQ(3, vertex.labels().size());
    EXPECT_TRUE(vertex.has_label(label2));
    EXPECT_TRUE(vertex.has_label(label3));
@ -1083,6 +1084,7 @@ TEST(Interpreter, RemoveLabels) {
  EXPECT_EQ(2, PullAll(label_remove, *dba, symbol_table));

  for (VertexAccessor vertex : dba->vertices()) {
+    vertex.SwitchNew();
    EXPECT_EQ(1, vertex.labels().size());
    EXPECT_FALSE(vertex.has_label(label1));
    EXPECT_FALSE(vertex.has_label(label2));
--- a/tests/unit/mvcc.cpp
+++ b/tests/unit/mvcc.cpp
@ -11,8 +11,7 @@ class Prop : public mvcc::Record<Prop> {};
 TEST(MVCC, Case1Test3) {
  tx::Engine engine;
  auto t1 = engine.begin();
-  Prop *prop;
-  mvcc::VersionList<Prop> version_list(*t1, prop);
+  mvcc::VersionList<Prop> version_list(*t1);
  t1->commit();

  auto t2 = engine.begin();
@ -28,9 +27,8 @@ TEST(MVCC, Case1Test3) {

 TEST(MVCC, InSnapshot) {
  tx::Engine engine;
-  Prop *prop;
  auto t1 = engine.begin();
-  mvcc::VersionList<Prop> version_list(*t1, prop);
+  mvcc::VersionList<Prop> version_list(*t1);
  t1->commit();

  auto t2 = engine.begin();
--- a/tests/unit/mvcc_gc.cpp
+++ b/tests/unit/mvcc_gc.cpp
@ -37,15 +37,14 @@ TEST(VersionList, GcDeleted) {
  std::vector<uint64_t> ids;
  auto t1 = engine.begin();
  std::atomic<int> count{0};
-  Prop *prop;
-  mvcc::VersionList<Prop> version_list(*t1, prop, count);
+  mvcc::VersionList<Prop> version_list(*t1, count);
  ids.push_back(t1->id);
  t1->commit();

  for (int i = 0; i < UPDATES; ++i) {
    auto t2 = engine.begin();
    ids.push_back(t2->id);
-    prop = version_list.update(prop, *t2);
+    version_list.update(*t2);
    t2->commit();
  }

@ -72,9 +71,8 @@ TEST(GarbageCollector, WaitAndClean) {
  gc.Run(std::chrono::seconds(1));

  auto t1 = engine.begin();
-  Prop *prop;
  std::atomic<int> count;
-  auto vl = new mvcc::VersionList<Prop>(*t1, prop, count);
+  auto vl = new mvcc::VersionList<Prop>(*t1, count);

  auto access = skiplist.access();
  access.insert(vl);
@ -99,9 +97,8 @@ TEST(GarbageCollector, WaitAndDontClean) {
                                     // the top one except GC is never run.

  auto t1 = engine.begin();
-  Prop *prop;
  std::atomic<int> count;
-  auto vl = new mvcc::VersionList<Prop>(*t1, prop, count);
+  auto vl = new mvcc::VersionList<Prop>(*t1, count);

  auto access = skiplist.access();
  access.insert(vl);
--- a/tests/unit/record_edge_vertex_accessor.cpp
+++ b/tests/unit/record_edge_vertex_accessor.cpp
@ -77,6 +77,76 @@ TEST(RecordAccessor, RecordLessThan) {
  EXPECT_FALSE(e2 < e2);
 }

+TEST(RecordAccessor, SwitchOldAndSwitchNewMemberFunctionTest) {
+  Dbms dbms;
+
+  // test SwitchOld failure on new record, SwitchNew OK
+  {
+    auto dba = dbms.active();
+    auto v1 = dba->insert_vertex();
+    EXPECT_DEATH(v1.SwitchOld(), "");
+    v1.SwitchNew();
+    dba->commit();
+  }
+
+  // test both Switches work on existing record
+  {
+    auto dba = dbms.active();
+    auto v1 = *dba->vertices().begin();
+    v1.SwitchOld();
+    v1.SwitchNew();
+  }
+
+  // ensure switch exposes the right data
+  {
+    auto dba = dbms.active();
+    auto label = dba->label("label");
+    auto v1 = *dba->vertices().begin();
+
+    EXPECT_FALSE(v1.has_label(label)); // old record
+    v1.add_label(label); // modifying data does not switch to new
+    EXPECT_FALSE(v1.has_label(label)); // old record
+    v1.SwitchNew();
+    EXPECT_TRUE(v1.has_label(label));
+    v1.SwitchOld();
+    EXPECT_FALSE(v1.has_label(label));
+  }
+}
+
+TEST(RecordAccessor, Reconstruct) {
+  Dbms dbms;
+  auto label = dbms.active()->label("label");
+
+  {
+    // we must operate on an old vertex
+    // because otherwise we only have new
+    // so create a vertex and commit it
+    auto dba = dbms.active();
+    dba->insert_vertex();
+    dba->commit();
+  }
+
+  // ensure we don't have label set
+  auto dba = dbms.active();
+  auto v1 = *dba->vertices().begin();
+  v1.SwitchNew();
+  EXPECT_FALSE(v1.has_label(label));
+
+  {
+    // update the record through a different accessor
+    auto v1_other_accessor = *dba->vertices().begin();
+    v1_other_accessor.add_label(label);
+    EXPECT_FALSE(v1.has_label(label));
+    v1_other_accessor.SwitchNew();
+    EXPECT_TRUE(v1_other_accessor.has_label(label));
+  }
+
+  EXPECT_FALSE(v1.has_label(label));
+  v1.Reconstruct();
+  v1.SwitchNew();
+  EXPECT_TRUE(v1.has_label(label));
+}
+
 TEST(RecordAccessor, VertexLabels) {
  Dbms dbms;
  auto dba = dbms.active();