db: view: Fix incorrect schema access during view building after base table schema changes

The view building process was accessing mutation fragments using
current table's schema. This is not correct, fragments must be
accessed using the schema of the generating reader.

This could lead to undefined behavior when the column set of the base
table changes. out_of_range exceptions could be observed, or data in
the view ending up in the wrong column.

Refs #7061.

The fix has two parts. First, we always use the reader's schema to
access fragments generated by the reader.

Second, when calling populate_views() we upgrade the fragment-wrapping
reader's schema to the base table schema so that it matches the base
table schema of view_and_base snapshots passed to populate_views().

database.hh

@@ -955,8 +955,9 @@ public:
         return *_config.sstables_manager;
     }
 
+    // Reader's schema must be the same as the base schema of each of the views.
     future<> populate_views(
-            std::vector<view_ptr>,
+            std::vector<db::view::view_and_base>,
             dht::token base_token,
             flat_mutation_reader&&,
             gc_clock::time_point);

db/view/view.cc

@@ -1774,7 +1774,7 @@ public:
             return stop_iteration::yes;
         }
 
-        _fragments_memory_usage += cr.memory_usage(*_step.base->schema());
+        _fragments_memory_usage += cr.memory_usage(*_step.reader.schema());
         _fragments.push_back(std::move(cr));
         if (_fragments_memory_usage > batch_memory_max) {
             // Although we have not yet completed the batch of base rows that
@@ -1796,10 +1796,14 @@ public:
         _builder._as.check();
         if (!_fragments.empty()) {
             _fragments.push_front(partition_start(_step.current_key, tombstone()));
+            auto base_schema = _step.base->schema();
+            auto views = with_base_info_snapshot(_views_to_build);
+            auto reader = make_flat_mutation_reader_from_fragments(_step.reader.schema(), std::move(_fragments));
+            reader.upgrade_schema(base_schema);
             _step.base->populate_views(
-                    _views_to_build,
+                    std::move(views),
                     _step.current_token(),
-                    make_flat_mutation_reader_from_fragments(_step.base->schema(), std::move(_fragments)),
+                    std::move(reader),
                     _now).get();
             _fragments.clear();
             _fragments_memory_usage = 0;

table.cc

@@ -1791,16 +1791,14 @@ table::local_base_lock(
  * @return a future that resolves when the updates have been acknowledged by the view replicas
  */
 future<> table::populate_views(
-        std::vector<view_ptr> views,
+        std::vector<db::view::view_and_base> views,
         dht::token base_token,
         flat_mutation_reader&& reader,
         gc_clock::time_point now) {
-    // FIXME: Handle schema version mismatch between view's base_dependent_view_info and reader's schema.
-    // Currently, generate_view_updates() will throw on schema version mismatch.
     auto& schema = reader.schema();
     return db::view::generate_view_updates(
             schema,
-            db::view::with_base_info_snapshot(views),
+            std::move(views),
             std::move(reader),
             { },
             now).then([base_token = std::move(base_token), this] (std::vector<frozen_mutation_and_schema>&& updates) mutable {