partition_version: Introduce partition_entry::with_detached_versions()

partition_version.cc

@@ -20,6 +20,7 @@
  */
 
 #include <boost/range/algorithm/heap_algorithm.hpp>
+#include <seastar/util/defer.hh>
 
 #include "partition_version.hh"
 
@@ -316,6 +317,31 @@ void partition_entry::apply(const schema& s, partition_entry&& pe, const schema&
     }
 }
 
+
+template<typename Func>
+void partition_entry::with_detached_versions(Func&& func) {
+    partition_version* current = &*_version;
+    auto snapshot = _snapshot;
+    if (snapshot) {
+        snapshot->_version = std::move(_version);
+        snapshot->_entry = nullptr;
+        _snapshot = nullptr;
+    }
+    _version = { };
+
+    auto revert = defer([&] {
+        if (snapshot) {
+            _snapshot = snapshot;
+            snapshot->_entry = this;
+            _version = std::move(snapshot->_version);
+        } else {
+            _version = partition_version_ref(*current);
+        }
+    });
+
+    func(current);
+}
+
 mutation_partition partition_entry::squashed(schema_ptr from, schema_ptr to)
 {
     mutation_partition mp(to);

partition_version.hh

@@ -245,6 +245,11 @@ class partition_entry {
     friend class partition_snapshot;
     friend class cache_entry;
 private:
+    // Detaches all versions temporarily around execution of the function.
+    // The function receives partition_version* pointing to the latest version.
+    template<typename Func>
+    void with_detached_versions(Func&&);
+
     void set_version(partition_version*);
 
     void apply(const schema& s, partition_version* pv, const schema& pv_schema);