mvcc_test: fix a benign failure of test_apply_to_incomplete_respects_continuity

For performance reasons, mutation_partition_v2::maybe_drop(), and by extension
also mutation_partition_v2::apply_monotonically(mutation_partition_v2&&)
can evict empty row entries, and hence change the continuity of the merged
entry.

For checking that apply_to_incomplete respects continuity,
test_apply_to_incomplete_respects_continuity obtains the continuity of
the partition entry before and after apply_to_incomplete by calling
e.squashed().get_continuity(). But squashed() uses apply_monotonically(),
so in some circumstances the result of squashed() can have smaller
continuity than the argument of squashed(), which messes with the thing
that the test is trying to check, and causes spurious failures.

This patch changes the method of calculating the continuity set,
so that it matches the entry exactly, fixing the test failures.

Fixes scylladb/scylladb#13757

Closes scylladb/scylladb#21459

(cherry picked from commit 35921eb67e)

mutation/mutation_partition_v2.hh

@@ -214,6 +214,7 @@ public:
     mutation_partition as_mutation_partition(const schema&) const;
 private:
     // Erases the entry if it's safe to do so without changing the logical state of the partition.
+    // (It's allowed to evict empty row entries, though).
     rows_type::iterator maybe_drop(const schema&, cache_tracker*, rows_type::iterator, mutation_application_stats&);
     void insert_row(const schema& s, const clustering_key& key, deletable_row&& row);
     void insert_row(const schema& s, const clustering_key& key, const deletable_row& row);

mutation/partition_version.cc

@@ -13,6 +13,7 @@
 #include "partition_snapshot_row_cursor.hh"
 #include "utils/coroutine.hh"
 #include "real_dirty_memory_accounter.hh"
+#include "clustering_interval_set.hh"
 
 static void remove_or_mark_as_unique_owner(partition_version* current, mutation_cleaner* cleaner)
 {
@@ -637,6 +638,15 @@ mutation_partition_v2 partition_entry::squashed_v2(const schema& to, is_evictabl
     return mp;
 }
 
+clustering_interval_set partition_entry::squashed_continuity(const schema& s)
+{
+    clustering_interval_set result;
+    for (auto&& v : _version->all_elements()) {
+        result.add(s, v.partition().as_mutation_partition(*v.get_schema()).get_continuity(s));
+    }
+    return result;
+}
+
 mutation_partition partition_entry::squashed(const schema& s, is_evictable evictable)
 {
     return squashed_v2(s, evictable).as_mutation_partition(s);

mutation/partition_version.hh

@@ -681,6 +681,7 @@ public:
     }
 
     mutation_partition_v2 squashed_v2(const schema& to, is_evictable);
+    clustering_interval_set squashed_continuity(const schema&);
     mutation_partition squashed(const schema&, is_evictable);
     tombstone partition_tombstone() const;
 

test/boost/mvcc_test.cc

@@ -445,7 +445,7 @@ SEASTAR_TEST_CASE(test_apply_to_incomplete_respects_continuity) {
                 }
 
                 auto before = e.squashed();
-                auto e_continuity = before.get_continuity(*s);
+                auto e_continuity = e.entry().squashed_continuity(*s);
 
                 auto expected_to_apply_slice = mutation_partition(*s, to_apply.partition());
                 if (!before.static_row_continuous()) {
@@ -462,7 +462,7 @@ SEASTAR_TEST_CASE(test_apply_to_incomplete_respects_continuity) {
 
                 // After applying to_apply the continuity can be more narrow due to compaction with tombstones
                 // present in to_apply.
-                auto continuity_after = sq.get_continuity(*s);
+                auto continuity_after = e.entry().squashed_continuity(*s);
                 if (!continuity_after.contained_in(e_continuity)) {
                     BOOST_FAIL(format("Expected later continuity to be contained in earlier, later={}\n, earlier={}",
                                       continuity_after, e_continuity));