Merge "mvcc: Fix partition_snapshot::merge_partition_versions() to no…

…t leave latest versions unmerged" from Tomasz

"Fixes a bug in partition_snapshot::merge_partition_versions(), which would not
attempt merging if the snapshot is attached to the latest version (in which
case _version is nullptr and _entry is != nullptr). This would cause
partition_version objects to accumulate if there was an older snapshot and it
went away before the latest snapshot. Versions will be removed when the whole
entry goes away (flush or eviction).

May cause performance problems.

Fixes #3402."

* 'tgrabiec/fix-merge_partition_versions' of github.com:tgrabiec/scylla:
  mvcc: Test version merging when snapshots go away
  anchorless_list: Make ranges conform to SinglePassRange
  anchorless_list: Drop deprecated use of std::iterator
  mvcc: Fix partition_snapshot::merge_partition_versions() to not leave latest versions unmerged

partition_version.cc

@@ -169,10 +169,10 @@ void merge_versions(const schema& s, mutation_partition& newer, mutation_partiti
 }
 
 void partition_snapshot::merge_partition_versions() {
-    if (_version && !_version.is_unique_owner()) {
-        auto v = &*_version;
+    partition_version_ref& v = version();
+    if (!v.is_unique_owner()) {
+        auto first_used = &*v;
         _version = { };
-        auto first_used = v;
         while (first_used->prev() && !first_used->is_referenced()) {
             first_used = first_used->prev();
         }

tests/mvcc_test.cc

@@ -23,6 +23,7 @@
 #include <boost/range/adaptor/transformed.hpp>
 #include <boost/range/algorithm/copy.hpp>
 #include <boost/range/algorithm_ext/push_back.hpp>
+#include <boost/range/size.hpp>
 #include <seastar/core/thread.hh>
 
 #include "partition_version.hh"
@@ -722,3 +723,63 @@ SEASTAR_TEST_CASE(test_apply_is_atomic) {
     do_test(random_mutation_generator(random_mutation_generator::generate_counters::yes));
     return make_ready_future<>();
 }
+
+SEASTAR_TEST_CASE(test_versions_are_merged_when_snapshots_go_away) {
+    return seastar::async([] {
+        logalloc::region r;
+        with_allocator(r.allocator(), [&] {
+            random_mutation_generator gen(random_mutation_generator::generate_counters::no);
+            auto s = gen.schema();
+
+            mutation m1 = gen();
+            mutation m2 = gen();
+            mutation m3 = gen();
+
+            m1.partition().make_fully_continuous();
+            m2.partition().make_fully_continuous();
+            m3.partition().make_fully_continuous();
+
+            {
+                auto e = partition_entry(m1.partition());
+                auto snap1 = e.read(r, s, nullptr);
+
+                {
+                    logalloc::reclaim_lock rl(r);
+                    e.apply(*s, m2.partition(), *s);
+                }
+
+                auto snap2 = e.read(r, s, nullptr);
+
+                snap1->merge_partition_versions();
+                snap1 = {};
+
+                snap2->merge_partition_versions();
+                snap2 = {};
+
+                BOOST_REQUIRE_EQUAL(1, boost::size(e.versions()));
+                assert_that(s, e.squashed(*s)).is_equal_to((m1 + m2).partition());
+            }
+
+            {
+                auto e = partition_entry(m1.partition());
+                auto snap1 = e.read(r, s, nullptr);
+
+                {
+                    logalloc::reclaim_lock rl(r);
+                    e.apply(*s, m2.partition(), *s);
+                }
+
+                auto snap2 = e.read(r, s, nullptr);
+
+                snap2->merge_partition_versions();
+                snap2 = {};
+
+                snap1->merge_partition_versions();
+                snap1 = {};
+
+                BOOST_REQUIRE_EQUAL(1, boost::size(e.versions()));
+                assert_that(s, e.squashed(*s)).is_equal_to((m1 + m2).partition());
+            }
+        });
+    });
+}

utils/anchorless_list.hh

@@ -28,12 +28,19 @@ class anchorless_list_base_hook {
     anchorless_list_base_hook<T>* _next = nullptr;
     anchorless_list_base_hook<T>* _prev = nullptr;
 public:
-    class iterator : std::iterator<std::bidirectional_iterator_tag, T> {
+    template <typename ValueType>
+    class iterator {
         anchorless_list_base_hook<T>* _position;
+    public:
+        using iterator_category = std::bidirectional_iterator_tag;
+        using value_type = ValueType;
+        using difference_type = ssize_t;
+        using pointer = ValueType*;
+        using reference = ValueType&;
     public:
         explicit iterator(anchorless_list_base_hook<T>* pos) : _position(pos) { }
-        T& operator*() { return *static_cast<T*>(_position); }
-        T* operator->() { return static_cast<T*>(_position); }
+        ValueType& operator*() { return *static_cast<ValueType*>(_position); }
+        ValueType* operator->() { return static_cast<ValueType*>(_position); }
         iterator& operator++() {
             _position = _position->_next;
             return *this;
@@ -52,26 +59,27 @@ public:
             operator--();
             return it;
         }
-        bool operator==(const iterator& other) {
+        bool operator==(const iterator& other) const {
             return _position == other._position;
         }
-        bool operator!=(const iterator& other) {
+        bool operator!=(const iterator& other) const {
             return !(*this == other);
         }
     };
 
+    template <typename ValueType>
     class reverse_iterator {
         anchorless_list_base_hook<T>* _position;
     public:
         using iterator_category = std::forward_iterator_tag;
-        using value_type = T;
+        using value_type = ValueType;
         using difference_type = ssize_t;
-        using pointer = T*;
-        using reference = T&;
+        using pointer = ValueType*;
+        using reference = ValueType&;
     public:
         explicit reverse_iterator(anchorless_list_base_hook<T>* pos) : _position(pos) { }
-        T& operator*() { return *static_cast<T*>(_position); }
-        T* operator->() { return static_cast<T*>(_position); }
+        ValueType& operator*() { return *static_cast<ValueType*>(_position); }
+        ValueType* operator->() { return static_cast<ValueType*>(_position); }
         reverse_iterator& operator++() {
             _position = _position->_prev;
             return *this;
@@ -81,10 +89,10 @@ public:
             operator++();
             return it;
         }
-        bool operator==(const reverse_iterator& other) {
+        bool operator==(const reverse_iterator& other) const {
             return _position == other._position;
         }
-        bool operator!=(const reverse_iterator& other) {
+        bool operator!=(const reverse_iterator& other) const {
             return !(*this == other);
         }
     };
@@ -93,20 +101,28 @@ public:
         anchorless_list_base_hook<T>* _begin;
         anchorless_list_base_hook<T>* _end;
     public:
+        using iterator = anchorless_list_base_hook::iterator<T>;
+        using const_iterator = anchorless_list_base_hook::iterator<const T>;
         range(anchorless_list_base_hook<T>* b, anchorless_list_base_hook<T>* e)
             : _begin(b), _end(e) { }
         iterator begin() { return iterator(_begin); }
         iterator end() { return iterator(_end); }
+        const_iterator begin() const { return const_iterator(_begin); }
+        const_iterator end() const { return const_iterator(_end); }
     };
 
     class reversed_range {
         anchorless_list_base_hook<T>* _begin;
         anchorless_list_base_hook<T>* _end;
     public:
+        using iterator = anchorless_list_base_hook::reverse_iterator<T>;
+        using const_iterator = anchorless_list_base_hook::reverse_iterator<const T>;
         reversed_range(anchorless_list_base_hook<T>* b, anchorless_list_base_hook<T>* e)
             : _begin(b), _end(e) { }
-        reverse_iterator begin() { return reverse_iterator(_begin); }
-        reverse_iterator end() { return reverse_iterator(_end); }
+        iterator begin() { return iterator(_begin); }
+        iterator end() { return iterator(_end); }
+        const_iterator begin() const { return const_iterator(_begin); }
+        const_iterator end() const { return const_iterator(_end); }
     };
 public:
     anchorless_list_base_hook() = default;
@@ -183,8 +199,8 @@ public:
         }
         return const_cast<T*>(static_cast<const T*>(v));
     }
-    iterator iterator_to() {
-        return iterator(this);
+    iterator<T> iterator_to() {
+        return iterator<T>(this);
     }
     range all_elements() {
         auto begin = this;