replica/table: migrate populate_views() to v2

db/view/view.cc

@@ -56,7 +56,7 @@
 #include "utils/exponential_backoff_retry.hh"
 #include "utils/fb_utilities.hh"
 #include "query-result-writer.hh"
-#include "readers/from_fragments.hh"
+#include "readers/from_fragments_v2.hh"
 
 using namespace std::chrono_literals;
 
@@ -1854,7 +1854,7 @@ private:
     built_views _built_views;
     gc_clock::time_point _now;
     std::vector<view_ptr> _views_to_build;
-    std::deque<mutation_fragment> _fragments;
+    std::deque<mutation_fragment_v2> _fragments;
     // The compact_for_query<> that feeds this consumer is already configured
     // to feed us up to view_builder::batchsize (128) rows and not an entire
     // partition. Still, if rows contain large blobs, saving 128 of them in

replica/database.hh

@@ -1027,7 +1027,7 @@ public:
     future<> populate_views(
             std::vector<db::view::view_and_base>,
             dht::token base_token,
-            flat_mutation_reader&&,
+            flat_mutation_reader_v2&&,
             gc_clock::time_point);
 
     reader_concurrency_semaphore& streaming_read_concurrency_semaphore() {

replica/table.cc

@@ -1810,13 +1810,13 @@ table::local_base_lock(
 future<> table::populate_views(
         std::vector<db::view::view_and_base> views,
         dht::token base_token,
-        flat_mutation_reader&& reader,
+        flat_mutation_reader_v2&& reader,
         gc_clock::time_point now) {
     auto schema = reader.schema();
     db::view::view_update_builder builder = co_await db::view::make_view_update_builder(
             schema,
             std::move(views),
-            upgrade_to_v2(std::move(reader)),
+            std::move(reader),
             { },
             now);