collection_mutation: move collection_type_impl::is_any_live to collec…

…tion_mutation_view

collection_mutation.cc

@@ -59,8 +59,11 @@ bool collection_mutation_view::is_empty() const {
   });
 }
 
-bool collection_type_impl::is_any_live(collection_mutation_view cm, tombstone tomb, gc_clock::time_point now) const {
-  return cm.data.with_linearized([&] (bytes_view in) {
+bool collection_mutation_view::is_any_live(const abstract_type& type, tombstone tomb, gc_clock::time_point now) const {
+    assert(type.is_collection());
+    auto& ctype = static_cast<const collection_type_impl&>(type);
+
+  return data.with_linearized([&] (bytes_view in) {
     auto has_tomb = read_simple<bool>(in);
     if (has_tomb) {
         auto ts = read_simple<api::timestamp_type>(in);
@@ -72,7 +75,7 @@ bool collection_type_impl::is_any_live(collection_mutation_view cm, tombstone to
         auto ksize = read_simple<uint32_t>(in);
         in.remove_prefix(ksize);
         auto vsize = read_simple<uint32_t>(in);
-        auto value = atomic_cell_view::from_bytes(value_comparator()->imr_state().type_info(), read_simple_bytes(in, vsize));
+        auto value = atomic_cell_view::from_bytes(ctype.value_comparator()->imr_state().type_info(), read_simple_bytes(in, vsize));
         if (value.is_live(tomb, now, false)) {
             return true;
         }

collection_mutation.hh

@@ -72,6 +72,11 @@ public:
     // Is this a noop mutation?
     bool is_empty() const;
 
+    // Is any of the stored cells live (not deleted nor expired) at the time point `tp`,
+    // given the later of the tombstones `t` and the one stored in the mutation (if any)?
+    // Requires a type to reconstruct the structural information.
+    bool is_any_live(const abstract_type&, tombstone t = tombstone(), gc_clock::time_point tp = gc_clock::time_point::min()) const;
+
     // Given a function that operates on a collection_mutation_view_description,
     // calls it on the corresponding description of `this`.
     template <typename F>

mutation_partition.cc

@@ -824,12 +824,11 @@ static void get_compacted_row_slice(const schema& s,
                     write_cell(writer, slice, cell->as_atomic_cell(def));
                 }
             } else {
-                auto&& mut = cell->as_collection_mutation();
-                auto&& ctype = static_pointer_cast<const collection_type_impl>(def.type);
-                if (!ctype->is_any_live(mut)) {
+                auto mut = cell->as_collection_mutation();
+                if (!mut.is_any_live(*def.type)) {
                     writer.add().skip();
                 } else {
-                    write_cell(writer, slice, def.type, mut);
+                    write_cell(writer, slice, def.type, std::move(mut));
                 }
             }
         }
@@ -848,9 +847,8 @@ bool has_any_live_data(const schema& s, column_kind kind, const row& cells, tomb
                 return stop_iteration::yes;
             }
         } else {
-            auto&& cell = cell_or_collection.as_collection_mutation();
-            auto&& ctype = static_pointer_cast<const collection_type_impl>(def.type);
-            if (ctype->is_any_live(cell, tomb, now)) {
+            auto mut = cell_or_collection.as_collection_mutation();
+            if (mut.is_any_live(*def.type, tomb, now)) {
                 any_live = true;
                 return stop_iteration::yes;
             }

types/collection.hh

@@ -51,7 +51,6 @@ public:
     virtual bool is_value_compatible_with_frozen(const collection_type_impl& previous) const = 0;
     template <typename BytesViewIterator>
     static bytes pack(BytesViewIterator start, BytesViewIterator finish, int elements, cql_serialization_format sf);
-    bool is_any_live(collection_mutation_view in, tombstone tomb = tombstone(), gc_clock::time_point now = gc_clock::time_point::min()) const;
     api::timestamp_type last_update(collection_mutation_view in) const;
     virtual bytes to_value(collection_mutation_view_description mut, cql_serialization_format sf) const = 0;
     bytes to_value(collection_mutation_view mut, cql_serialization_format sf) const;