From 7ddcd0c91883144635f6af9db4366dddfc519bf9 Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Thu, 20 Apr 2023 09:17:52 +0300
Subject: [PATCH 1/9] view: Add database getters to v._update_generator and
 v._builder

Both services carry database which will be used by auxiliary objects
like view_updates, view_update_builder, consumer, etc in next patches.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 db/view/view_builder.hh          | 2 ++
 db/view/view_update_generator.hh | 2 ++
 2 files changed, 4 insertions(+)

diff --git a/db/view/view_builder.hh b/db/view/view_builder.hh
index f032099028..cd7c0e422a 100644
--- a/db/view/view_builder.hh
+++ b/db/view/view_builder.hh
@@ -189,6 +189,8 @@ public:
     static constexpr size_t batch_size = 128;
     static constexpr size_t batch_memory_max = 1024*1024;
 
+    replica::database& get_db() noexcept { return _db; }
+
 public:
     view_builder(replica::database&, db::system_keyspace&, db::system_distributed_keyspace&, service::migration_notifier&, view_update_generator& vug);
     view_builder(view_builder&&) = delete;
diff --git a/db/view/view_update_generator.hh b/db/view/view_update_generator.hh
index 6b9c0cf5e7..dbc8fd9b7f 100644
--- a/db/view/view_update_generator.hh
+++ b/db/view/view_update_generator.hh
@@ -72,6 +72,8 @@ public:
     future<> stop();
     future<> register_staging_sstable(sstables::shared_sstable sst, lw_shared_ptr<replica::table> table);
 
+    replica::database& get_db() noexcept { return _db; }
+
     future<> mutate_MV(
             dht::token base_token,
             utils::chunked_vector<frozen_mutation_and_schema> view_updates,

From 4a16ab3bd41b4884f3c037221a5e94c1223a5b1e Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Wed, 19 Apr 2023 21:53:50 +0300
Subject: [PATCH 2/9] table: Push view_update_generator arg to affected_views()

Caller already has it to call mutate_MV() on. The method in question
will need the generator in one of the next patches.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 replica/database.hh | 2 +-
 replica/table.cc    | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/replica/database.hh b/replica/database.hh
index b884f21a6b..c072770b84 100644
--- a/replica/database.hh
+++ b/replica/database.hh
@@ -1064,7 +1064,7 @@ public:
 private:
     future<row_locker::lock_holder> do_push_view_replica_updates(shared_ptr<db::view::view_update_generator> gen, schema_ptr s, mutation m, db::timeout_clock::time_point timeout, mutation_source source,
             tracing::trace_state_ptr tr_state, reader_concurrency_semaphore& sem, const io_priority_class& io_priority, query::partition_slice::option_set custom_opts) const;
-    std::vector<view_ptr> affected_views(const schema_ptr& base, const mutation& update) const;
+    std::vector<view_ptr> affected_views(shared_ptr<db::view::view_update_generator> gen, const schema_ptr& base, const mutation& update) const;
     future<> generate_and_propagate_view_updates(shared_ptr<db::view::view_update_generator> gen, const schema_ptr& base,
             reader_permit permit,
             std::vector<db::view::view_and_base>&& views,
diff --git a/replica/table.cc b/replica/table.cc
index a6ab830572..58f27e5064 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -1976,7 +1976,7 @@ const std::vector<view_ptr>& table::views() const {
     return _views;
 }
 
-std::vector<view_ptr> table::affected_views(const schema_ptr& base, const mutation& update) const {
+std::vector<view_ptr> table::affected_views(shared_ptr<db::view::view_update_generator> gen, const schema_ptr& base, const mutation& update) const {
     //FIXME: Avoid allocating a vector here; consider returning the boost iterator.
     return boost::copy_range<std::vector<view_ptr>>(_views | boost::adaptors::filtered([&] (auto&& view) {
         return db::view::partition_key_matches(*base, *view->view_info(), update.decorated_key());
@@ -2634,7 +2634,7 @@ future<row_locker::lock_holder> table::do_push_view_replica_updates(shared_ptr<d
     utils::get_local_injector().inject("table_push_view_replica_updates_stale_time_point", [&now] {
         now -= 10s;
     });
-    auto views = db::view::with_base_info_snapshot(affected_views(base, m));
+    auto views = db::view::with_base_info_snapshot(affected_views(gen, base, m));
     if (views.empty()) {
         co_return row_locker::lock_holder();
     }

From 9d3d533561415b62a40c3b6fd27ccabae150f1ec Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Wed, 19 Apr 2023 21:49:52 +0300
Subject: [PATCH 3/9] view_update_builder: Construct with data dictionary

The caller is table with view-update-generator at hand (it calls
mutate_MV on). Builder here is used as a temporary object that destroys
once the caller coroutine co_return-s, so keeping the database obtained
from the view-update-generator is safe.

Later the v.u.b. object will propagate its data dictionary down the
callstacks.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 db/view/view.cc  | 3 ++-
 db/view/view.hh  | 8 ++++++--
 replica/table.cc | 2 ++
 3 files changed, 10 insertions(+), 3 deletions(-)

diff --git a/db/view/view.cc b/db/view/view.cc
index b9525c70fd..8420b382e2 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -1439,6 +1439,7 @@ future<stop_iteration> view_update_builder::on_results() {
 }
 
 view_update_builder make_view_update_builder(
+        data_dictionary::database db,
         const replica::table& base_table,
         const schema_ptr& base,
         std::vector<view_and_base>&& views_to_update,
@@ -1454,7 +1455,7 @@ view_update_builder make_view_update_builder(
         bool is_index = base_table.get_index_manager().is_index(v.view);
         return view_updates(std::move(v), is_index);
     }));
-    return view_update_builder(base_table, base, std::move(vs), std::move(updates), std::move(existings), now);
+    return view_update_builder(std::move(db), base_table, base, std::move(vs), std::move(updates), std::move(existings), now);
 }
 
 future<query::clustering_row_ranges> calculate_affected_clustering_ranges(const schema& base,
diff --git a/db/view/view.hh b/db/view/view.hh
index a80e117102..691b386085 100644
--- a/db/view/view.hh
+++ b/db/view/view.hh
@@ -14,6 +14,7 @@
 #include "schema/schema_fwd.hh"
 #include "readers/flat_mutation_reader_v2.hh"
 #include "mutation/frozen_mutation.hh"
+#include "data_dictionary/data_dictionary.hh"
 
 class frozen_mutation_and_schema;
 
@@ -244,6 +245,7 @@ private:
 };
 
 class view_update_builder {
+    data_dictionary::database _db;
     const replica::table& _base;
     schema_ptr _schema; // The base schema
     std::vector<view_updates> _view_updates;
@@ -259,12 +261,13 @@ class view_update_builder {
     partition_key _key = partition_key::make_empty();
 public:
 
-    view_update_builder(const replica::table& base, schema_ptr s,
+    view_update_builder(data_dictionary::database db, const replica::table& base, schema_ptr s,
         std::vector<view_updates>&& views_to_update,
         flat_mutation_reader_v2&& updates,
         flat_mutation_reader_v2_opt&& existings,
         gc_clock::time_point now)
-            : _base(base)
+            : _db(std::move(db))
+            , _base(base)
             , _schema(std::move(s))
             , _view_updates(std::move(views_to_update))
             , _updates(std::move(updates))
@@ -298,6 +301,7 @@ private:
 };
 
 view_update_builder make_view_update_builder(
+        data_dictionary::database db,
         const replica::table& base_table,
         const schema_ptr& base_schema,
         std::vector<view_and_base>&& views_to_update,
diff --git a/replica/table.cc b/replica/table.cc
index 58f27e5064..cebf23a814 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -2014,6 +2014,7 @@ future<> table::generate_and_propagate_view_updates(shared_ptr<db::view::view_up
     auto base_token = m.token();
     auto m_schema = m.schema();
     db::view::view_update_builder builder = db::view::make_view_update_builder(
+            gen->get_db().as_data_dictionary(),
             *this,
             base,
             std::move(views),
@@ -2151,6 +2152,7 @@ future<> table::populate_views(
         gc_clock::time_point now) {
     auto schema = reader.schema();
     db::view::view_update_builder builder = db::view::make_view_update_builder(
+            gen->get_db().as_data_dictionary(),
             *this,
             schema,
             std::move(views),

From 1301a99ba357f1e4ccbc7628c9b4151023c9cb53 Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Thu, 20 Apr 2023 09:15:52 +0300
Subject: [PATCH 4/9] view_updates: Carry data_dictionary argument throug
 methods

The goal is to have the dictionary at places that later wrap calls to
view_info::select_statement(). This graph of calls starts at the only
public view_updates::generate_update() method which, in turn, is called
from view_update_builder that already has data dictionary at hand.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 db/view/view.cc | 29 +++++++++++++++--------------
 db/view/view.hh |  8 ++++----
 2 files changed, 19 insertions(+), 18 deletions(-)

diff --git a/db/view/view.cc b/db/view/view.cc
index 8420b382e2..20e0842800 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -876,7 +876,7 @@ static void add_cells_to_view(const schema& base, const schema& view, column_kin
  * Creates a view entry corresponding to the provided base row.
  * This method checks that the base row does match the view filter before applying anything.
  */
-void view_updates::create_entry(const partition_key& base_key, const clustering_or_static_row& update, gc_clock::time_point now) {
+void view_updates::create_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& update, gc_clock::time_point now) {
     if (!matches_view_filter(*_base, _view_info, base_key, update, now)) {
         return;
     }
@@ -900,7 +900,7 @@ void view_updates::create_entry(const partition_key& base_key, const clustering_
  * Deletes the view entry corresponding to the provided base row.
  * This method checks that the base row does match the view filter before bothering.
  */
-void view_updates::delete_old_entry(const partition_key& base_key, const clustering_or_static_row& existing, const clustering_or_static_row& update, gc_clock::time_point now) {
+void view_updates::delete_old_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& existing, const clustering_or_static_row& update, gc_clock::time_point now) {
     // Before deleting an old entry, make sure it was matching the view filter
     // (otherwise there is nothing to delete)
     if (matches_view_filter(*_base, _view_info, base_key, existing, now)) {
@@ -1020,11 +1020,11 @@ bool view_updates::can_skip_view_updates(const clustering_or_static_row& update,
  * This method checks that the base row (before and after) matches the view filter before
  * applying anything.
  */
-void view_updates::update_entry(const partition_key& base_key, const clustering_or_static_row& update, const clustering_or_static_row& existing, gc_clock::time_point now) {
+void view_updates::update_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& update, const clustering_or_static_row& existing, gc_clock::time_point now) {
     // While we know update and existing correspond to the same view entry,
     // they may not match the view filter.
     if (!matches_view_filter(*_base, _view_info, base_key, existing, now)) {
-        create_entry(base_key, update, now);
+        create_entry(db, base_key, update, now);
         return;
     }
     if (!matches_view_filter(*_base, _view_info, base_key, update, now)) {
@@ -1076,6 +1076,7 @@ void view_updates::update_entry_for_computed_column(
 }
 
 void view_updates::generate_update(
+        data_dictionary::database db,
         const partition_key& base_key,
         const clustering_or_static_row& update,
         const std::optional<clustering_or_static_row>& existing,
@@ -1105,12 +1106,12 @@ void view_updates::generate_update(
         // The view key is necessarily the same pre and post update.
         if (existing && existing->is_live(*_base)) {
             if (update.is_live(*_base)) {
-                update_entry(base_key, update, *existing, now);
+                update_entry(db, base_key, update, *existing, now);
             } else {
-                delete_old_entry(base_key, *existing, update, now);
+                delete_old_entry(db, base_key, *existing, update, now);
             }
         } else if (update.is_live(*_base)) {
-            create_entry(base_key, update, now);
+            create_entry(db, base_key, update, now);
         }
         return;
     }
@@ -1168,21 +1169,21 @@ void view_updates::generate_update(
     if (has_old_row) {
         if (has_new_row) {
             if (same_row) {
-                update_entry(base_key, update, *existing, now);
+                update_entry(db, base_key, update, *existing, now);
             } else {
                 // This code doesn't work if the old and new view row have the
                 // same key, because if they do we get both data and tombstone
                 // for the same timestamp (now) and the tombstone wins. This
                 // is why we need the "same_row" case above - it's not just a
                 // performance optimization.
-                delete_old_entry(base_key, *existing, update, now);
-                create_entry(base_key, update, now);
+                delete_old_entry(db, base_key, *existing, update, now);
+                create_entry(db, base_key, update, now);
             }
         } else {
-            delete_old_entry(base_key, *existing, update, now);
+            delete_old_entry(db, base_key, *existing, update, now);
         }
     } else if (has_new_row) {
-        create_entry(base_key, update, now);
+        create_entry(db, base_key, update, now);
     }
 }
 
@@ -1277,7 +1278,7 @@ void view_update_builder::generate_update(clustering_row&& update, std::optional
             ? std::make_optional<clustering_or_static_row>(std::move(*existing))
             : std::optional<clustering_or_static_row>();
     for (auto&& v : _view_updates) {
-        v.generate_update(_key, update_row, existing_row, _now);
+        v.generate_update(_db, _key, update_row, existing_row, _now);
     }
 }
 
@@ -1304,7 +1305,7 @@ void view_update_builder::generate_update(static_row&& update, const tombstone&
             ? std::make_optional<clustering_or_static_row>(std::move(*existing))
             : std::optional<clustering_or_static_row>();
     for (auto&& v : _view_updates) {
-        v.generate_update(_key, update_row, existing_row, _now);
+        v.generate_update(_db, _key, update_row, existing_row, _now);
     }
 }
 
diff --git a/db/view/view.hh b/db/view/view.hh
index 691b386085..e75cceb547 100644
--- a/db/view/view.hh
+++ b/db/view/view.hh
@@ -224,7 +224,7 @@ public:
 
     future<> move_to(utils::chunked_vector<frozen_mutation_and_schema>& mutations);
 
-    void generate_update(const partition_key& base_key, const clustering_or_static_row& update, const std::optional<clustering_or_static_row>& existing, gc_clock::time_point now);
+    void generate_update(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& update, const std::optional<clustering_or_static_row>& existing, gc_clock::time_point now);
 
     size_t op_count() const;
 
@@ -237,10 +237,10 @@ private:
     };
     std::vector<view_row_entry> get_view_rows(const partition_key& base_key, const clustering_or_static_row& update, const std::optional<clustering_or_static_row>& existing);
     bool can_skip_view_updates(const clustering_or_static_row& update, const clustering_or_static_row& existing) const;
-    void create_entry(const partition_key& base_key, const clustering_or_static_row& update, gc_clock::time_point now);
-    void delete_old_entry(const partition_key& base_key, const clustering_or_static_row& existing, const clustering_or_static_row& update, gc_clock::time_point now);
+    void create_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& update, gc_clock::time_point now);
+    void delete_old_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& existing, const clustering_or_static_row& update, gc_clock::time_point now);
     void do_delete_old_entry(const partition_key& base_key, const clustering_or_static_row& existing, const clustering_or_static_row& update, gc_clock::time_point now);
-    void update_entry(const partition_key& base_key, const clustering_or_static_row& update, const clustering_or_static_row& existing, gc_clock::time_point now);
+    void update_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& update, const clustering_or_static_row& existing, gc_clock::time_point now);
     void update_entry_for_computed_column(const partition_key& base_key, const clustering_or_static_row& update, const std::optional<clustering_or_static_row>& existing, gc_clock::time_point now);
 };
 

From 837fde84b10f4cb74bc2e4f5e7204f529dcce387 Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Thu, 20 Apr 2023 09:26:55 +0300
Subject: [PATCH 5/9] view: Carry data_dictionary arg through standalone
 helpers

There's a bunch of functions in view.{hh|cc} that don't belong to any
class and perform view-related claculations for view updates. Lots of
them eventually call view_info::select_statement() which will later need
the dictionary.

By now all those methods' callers have data dictionary at hand and can
share it via argument.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 db/view/view.cc  | 31 ++++++++++++++++---------------
 db/view/view.hh  |  9 +++++----
 replica/table.cc |  4 ++--
 3 files changed, 23 insertions(+), 21 deletions(-)

diff --git a/db/view/view.cc b/db/view/view.cc
index 20e0842800..c2ebf8385a 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -273,7 +273,7 @@ void stats::register_stats() {
     });
 }
 
-bool partition_key_matches(const schema& base, const view_info& view, const dht::decorated_key& key) {
+bool partition_key_matches(data_dictionary::database db, const schema& base, const view_info& view, const dht::decorated_key& key) {
     const cql3::expr::expression& pk_restrictions = view.select_statement().get_restrictions()->get_partition_key_restrictions();
     std::vector<bytes> exploded_pk = key.key().explode();
     std::vector<bytes> exploded_ck;
@@ -298,7 +298,7 @@ bool partition_key_matches(const schema& base, const view_info& view, const dht:
             });
 }
 
-bool clustering_prefix_matches(const schema& base, const view_info& view, const partition_key& key, const clustering_key_prefix& ck) {
+bool clustering_prefix_matches(data_dictionary::database db, const schema& base, const view_info& view, const partition_key& key, const clustering_key_prefix& ck) {
     const cql3::expr::expression& r = view.select_statement().get_restrictions()->get_clustering_columns_restrictions();
     std::vector<bytes> exploded_pk = key.explode();
     std::vector<bytes> exploded_ck = ck.explode();
@@ -322,21 +322,21 @@ bool clustering_prefix_matches(const schema& base, const view_info& view, const
             });
 }
 
-bool may_be_affected_by(const schema& base, const view_info& view, const dht::decorated_key& key, const rows_entry& update) {
+bool may_be_affected_by(data_dictionary::database db, const schema& base, const view_info& view, const dht::decorated_key& key, const rows_entry& update) {
     // We can guarantee that the view won't be affected if:
     //  - the primary key is excluded by the view filter (note that this isn't true of the filter on regular columns:
     //    even if an update don't match a view condition on a regular column, that update can still invalidate a
     //    pre-existing entry) - note that the upper layers should already have checked the partition key;
-    return clustering_prefix_matches(base, view, key.key(), update.key());
+    return clustering_prefix_matches(db, base, view, key.key(), update.key());
 }
 
-static bool update_requires_read_before_write(const schema& base,
+static bool update_requires_read_before_write(data_dictionary::database db, const schema& base,
         const std::vector<view_and_base>& views,
         const dht::decorated_key& key,
         const rows_entry& update) {
     for (auto&& v : views) {
         view_info& vf = *v.view->view_info();
-        if (may_be_affected_by(base, vf, key, update)) {
+        if (may_be_affected_by(db, base, vf, key, update)) {
             return true;
         }
     }
@@ -447,7 +447,7 @@ public:
     }
 };
 
-bool matches_view_filter(const schema& base, const view_info& view, const partition_key& key, const clustering_or_static_row& update, gc_clock::time_point now) {
+bool matches_view_filter(data_dictionary::database db, const schema& base, const view_info& view, const partition_key& key, const clustering_or_static_row& update, gc_clock::time_point now) {
     // TODO: Filtering is only supported in materialized views which don't support
     // static rows yet. Skip the whole function if it is a static row update.
     if (update.is_static_row()) {
@@ -464,7 +464,7 @@ bool matches_view_filter(const schema& base, const view_info& view, const partit
     view_filter_checking_visitor visitor(base, view);
     query::result_view::consume(result, slice, visitor);
 
-    return clustering_prefix_matches(base, view, key, *update.key())
+    return clustering_prefix_matches(db, base, view, key, *update.key())
             && visitor.matches_view_filter();
 }
 
@@ -877,7 +877,7 @@ static void add_cells_to_view(const schema& base, const schema& view, column_kin
  * This method checks that the base row does match the view filter before applying anything.
  */
 void view_updates::create_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& update, gc_clock::time_point now) {
-    if (!matches_view_filter(*_base, _view_info, base_key, update, now)) {
+    if (!matches_view_filter(db, *_base, _view_info, base_key, update, now)) {
         return;
     }
 
@@ -903,7 +903,7 @@ void view_updates::create_entry(data_dictionary::database db, const partition_ke
 void view_updates::delete_old_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& existing, const clustering_or_static_row& update, gc_clock::time_point now) {
     // Before deleting an old entry, make sure it was matching the view filter
     // (otherwise there is nothing to delete)
-    if (matches_view_filter(*_base, _view_info, base_key, existing, now)) {
+    if (matches_view_filter(db, *_base, _view_info, base_key, existing, now)) {
         do_delete_old_entry(base_key, existing, update, now);
     }
 }
@@ -1023,11 +1023,11 @@ bool view_updates::can_skip_view_updates(const clustering_or_static_row& update,
 void view_updates::update_entry(data_dictionary::database db, const partition_key& base_key, const clustering_or_static_row& update, const clustering_or_static_row& existing, gc_clock::time_point now) {
     // While we know update and existing correspond to the same view entry,
     // they may not match the view filter.
-    if (!matches_view_filter(*_base, _view_info, base_key, existing, now)) {
+    if (!matches_view_filter(db, *_base, _view_info, base_key, existing, now)) {
         create_entry(db, base_key, update, now);
         return;
     }
-    if (!matches_view_filter(*_base, _view_info, base_key, update, now)) {
+    if (!matches_view_filter(db, *_base, _view_info, base_key, update, now)) {
         do_delete_old_entry(base_key, existing, update, now);
         return;
     }
@@ -1459,7 +1459,8 @@ view_update_builder make_view_update_builder(
     return view_update_builder(std::move(db), base_table, base, std::move(vs), std::move(updates), std::move(existings), now);
 }
 
-future<query::clustering_row_ranges> calculate_affected_clustering_ranges(const schema& base,
+future<query::clustering_row_ranges> calculate_affected_clustering_ranges(data_dictionary::database db,
+        const schema& base,
         const dht::decorated_key& key,
         const mutation_partition& mp,
         const std::vector<view_and_base>& views) {
@@ -1498,7 +1499,7 @@ future<query::clustering_row_ranges> calculate_affected_clustering_ranges(const
     }
 
     for (auto&& row : mp.clustered_rows()) {
-        if (update_requires_read_before_write(base, views, key, row)) {
+        if (update_requires_read_before_write(db, base, views, key, row)) {
             row_ranges.emplace_back(row.key());
         }
         co_await coroutine::maybe_yield();
@@ -2307,7 +2308,7 @@ public:
         inject_failure("view_builder_load_views");
         for (auto&& vs : _step.build_status) {
             if (_step.current_token() >= vs.next_token) {
-                if (partition_key_matches(*_step.reader.schema(), *vs.view->view_info(), _step.current_key)) {
+                if (partition_key_matches(_builder.get_db().as_data_dictionary(), *_step.reader.schema(), *vs.view->view_info(), _step.current_key)) {
                     _views_to_build.push_back(vs.view);
                 }
                 if (vs.next_token || _step.current_token() != vs.first_token) {
diff --git a/db/view/view.hh b/db/view/view.hh
index e75cceb547..b8d578551f 100644
--- a/db/view/view.hh
+++ b/db/view/view.hh
@@ -125,7 +125,7 @@ public:
  * @return false if we can guarantee that inserting an update for specified key
  * won't affect the view in any way, true otherwise.
  */
-bool partition_key_matches(const schema& base, const view_info& view, const dht::decorated_key& key);
+bool partition_key_matches(data_dictionary::database, const schema& base, const view_info& view, const dht::decorated_key& key);
 
 /**
  * Whether the view might be affected by the provided update.
@@ -140,7 +140,7 @@ bool partition_key_matches(const schema& base, const view_info& view, const dht:
  * @return false if we can guarantee that inserting update for key
  * won't affect the view in any way, true otherwise.
  */
-bool may_be_affected_by(const schema& base, const view_info& view, const dht::decorated_key& key, const rows_entry& update);
+bool may_be_affected_by(data_dictionary::database, const schema& base, const view_info& view, const dht::decorated_key& key, const rows_entry& update);
 
 /**
  * Whether a given base row matches the view filter (and thus if the view should have a corresponding entry).
@@ -159,9 +159,9 @@ bool may_be_affected_by(const schema& base, const view_info& view, const dht::de
  * @param now the current time in seconds (to decide what is live and what isn't).
  * @return whether the base row matches the view filter.
  */
-bool matches_view_filter(const schema& base, const view_info& view, const partition_key& key, const clustering_or_static_row& update, gc_clock::time_point now);
+bool matches_view_filter(data_dictionary::database, const schema& base, const view_info& view, const partition_key& key, const clustering_or_static_row& update, gc_clock::time_point now);
 
-bool clustering_prefix_matches(const schema& base, const partition_key& key, const clustering_key_prefix& ck);
+bool clustering_prefix_matches(data_dictionary::database, const schema& base, const partition_key& key, const clustering_key_prefix& ck);
 
 /*
  * When a base-table update modifies a value in a materialized view's key
@@ -310,6 +310,7 @@ view_update_builder make_view_update_builder(
         gc_clock::time_point now);
 
 future<query::clustering_row_ranges> calculate_affected_clustering_ranges(
+        data_dictionary::database db,
         const schema& base,
         const dht::decorated_key& key,
         const mutation_partition& mp,
diff --git a/replica/table.cc b/replica/table.cc
index cebf23a814..5ecc3c1a74 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -1979,7 +1979,7 @@ const std::vector<view_ptr>& table::views() const {
 std::vector<view_ptr> table::affected_views(shared_ptr<db::view::view_update_generator> gen, const schema_ptr& base, const mutation& update) const {
     //FIXME: Avoid allocating a vector here; consider returning the boost iterator.
     return boost::copy_range<std::vector<view_ptr>>(_views | boost::adaptors::filtered([&] (auto&& view) {
-        return db::view::partition_key_matches(*base, *view->view_info(), update.decorated_key());
+        return db::view::partition_key_matches(gen->get_db().as_data_dictionary(), *base, *view->view_info(), update.decorated_key());
     }));
 }
 
@@ -2640,7 +2640,7 @@ future<row_locker::lock_holder> table::do_push_view_replica_updates(shared_ptr<d
     if (views.empty()) {
         co_return row_locker::lock_holder();
     }
-    auto cr_ranges = co_await db::view::calculate_affected_clustering_ranges(*base, m.decorated_key(), m.partition(), views);
+    auto cr_ranges = co_await db::view::calculate_affected_clustering_ranges(gen->get_db().as_data_dictionary(), *base, m.decorated_key(), m.partition(), views);
     const bool need_regular = !cr_ranges.empty();
     const bool need_static = db::view::needs_static_row(m.partition(), views);
     if (!need_regular && !need_static) {

From 0aff55cdb2c5d78b8c9179340d29959dc10f711e Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Wed, 19 Apr 2023 20:35:14 +0300
Subject: [PATCH 6/9] view_filter_checking_visitor: Construct with
 data_dictionary

The visitor is wait-free helper for matches_view_filter() that has
dictionary as its argument. Later the visitor will pass the dictionary
to view_info::select_statement().

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 db/view/view.cc | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/db/view/view.cc b/db/view/view.cc
index c2ebf8385a..7ed9d3fc54 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -346,6 +346,7 @@ static bool update_requires_read_before_write(data_dictionary::database db, cons
 // Checks if the result matches the provided view filter.
 // It's currently assumed that the result consists of just a single row.
 class view_filter_checking_visitor {
+    data_dictionary::database _db;
     const schema& _base;
     const view_info& _view;
     ::shared_ptr<cql3::selection::selection> _selection;
@@ -353,8 +354,9 @@ class view_filter_checking_visitor {
 
     bool _matches_view_filter = true;
 public:
-    view_filter_checking_visitor(const schema& base, const view_info& view)
-        : _base(base)
+    view_filter_checking_visitor(data_dictionary::database db, const schema& base, const view_info& view)
+        : _db(std::move(db))
+        , _base(base)
         , _view(view)
         , _selection(cql3::selection::selection::for_columns(_base.shared_from_this(),
             boost::copy_range<std::vector<const column_definition*>>(
@@ -461,7 +463,7 @@ bool matches_view_filter(data_dictionary::database db, const schema& base, const
     builder.consume(clustering_row(base, update.as_clustering_row(base)), row_tombstone{}, update.is_live(base, tombstone(), now));
     builder.consume_end_of_partition();
     auto result = builder.consume_end_of_stream();
-    view_filter_checking_visitor visitor(base, view);
+    view_filter_checking_visitor visitor(db, base, view);
     query::result_view::consume(result, slice, visitor);
 
     return clustering_prefix_matches(db, base, view, key, *update.key())

From 4375835cddf3a3759b528da6d7c3d44b0289e7b2 Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Thu, 20 Apr 2023 09:30:13 +0300
Subject: [PATCH 7/9] view_info: Add data_dictionary argument to
 partition_slice() method

The caller is calculate_affected_clustering_ranges() with dictionary
arg, the method needs dictionary to call view_info::select_statement()
later.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 db/view/view.cc | 4 ++--
 view_info.hh    | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/db/view/view.cc b/db/view/view.cc
index 7ed9d3fc54..bb45187cd7 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -109,7 +109,7 @@ cql3::statements::select_statement& view_info::select_statement() const {
     return *_select_statement;
 }
 
-const query::partition_slice& view_info::partition_slice() const {
+const query::partition_slice& view_info::partition_slice(data_dictionary::database db) const {
     if (!_partition_slice) {
         _partition_slice = select_statement().make_partition_slice(cql3::query_options({ }));
     }
@@ -1476,7 +1476,7 @@ future<query::clustering_row_ranges> calculate_affected_clustering_ranges(data_d
                 view_row_ranges.push_back(nonwrapping_range<clustering_key_prefix_view>::make_open_ended_both_sides());
                 break;
             }
-            for (auto&& r : v.view->view_info()->partition_slice().default_row_ranges()) {
+            for (auto&& r : v.view->view_info()->partition_slice(db).default_row_ranges()) {
                 view_row_ranges.push_back(r.transform(std::mem_fn(&clustering_key_prefix::view)));
                 co_await coroutine::maybe_yield();
             }
diff --git a/view_info.hh b/view_info.hh
index 123164db7c..dfccf3458c 100644
--- a/view_info.hh
+++ b/view_info.hh
@@ -49,7 +49,7 @@ public:
     }
 
     cql3::statements::select_statement& select_statement() const;
-    const query::partition_slice& partition_slice() const;
+    const query::partition_slice& partition_slice(data_dictionary::database) const;
     const column_definition* view_column(const schema& base, column_kind kind, column_id base_id) const;
     const column_definition* view_column(const column_definition& base_def) const;
     bool has_base_non_pk_columns_in_view_pk() const;

From 3e4fb7cad649cedc24d13278e8c592a7f2c4359b Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Thu, 20 Apr 2023 09:32:49 +0300
Subject: [PATCH 8/9] view_info: Add data_dictionary argument to
 select_statement()

This method needs data_dictionary to work. Fortunately, all callers of
it already have the dictionary at hand and can just pass it as argument.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 db/view/view.cc | 12 ++++++------
 view_info.hh    |  2 +-
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/db/view/view.cc b/db/view/view.cc
index bb45187cd7..228b5994e6 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -79,7 +79,7 @@ view_info::view_info(const schema& schema, const raw_view_info& raw_view_info)
         , _has_computed_column_depending_on_base_non_primary_key(false)
 { }
 
-cql3::statements::select_statement& view_info::select_statement() const {
+cql3::statements::select_statement& view_info::select_statement(data_dictionary::database db) const {
     if (!_select_statement) {
         std::unique_ptr<cql3::statements::raw::select_statement> raw;
         // FIXME(sarna): legacy code, should be removed after "computed_columns" feature is guaranteed
@@ -111,7 +111,7 @@ cql3::statements::select_statement& view_info::select_statement() const {
 
 const query::partition_slice& view_info::partition_slice(data_dictionary::database db) const {
     if (!_partition_slice) {
-        _partition_slice = select_statement().make_partition_slice(cql3::query_options({ }));
+        _partition_slice = select_statement(db).make_partition_slice(cql3::query_options({ }));
     }
     return *_partition_slice;
 }
@@ -274,7 +274,7 @@ void stats::register_stats() {
 }
 
 bool partition_key_matches(data_dictionary::database db, const schema& base, const view_info& view, const dht::decorated_key& key) {
-    const cql3::expr::expression& pk_restrictions = view.select_statement().get_restrictions()->get_partition_key_restrictions();
+    const cql3::expr::expression& pk_restrictions = view.select_statement(db).get_restrictions()->get_partition_key_restrictions();
     std::vector<bytes> exploded_pk = key.key().explode();
     std::vector<bytes> exploded_ck;
     std::vector<const column_definition*> pk_columns;
@@ -299,7 +299,7 @@ bool partition_key_matches(data_dictionary::database db, const schema& base, con
 }
 
 bool clustering_prefix_matches(data_dictionary::database db, const schema& base, const view_info& view, const partition_key& key, const clustering_key_prefix& ck) {
-    const cql3::expr::expression& r = view.select_statement().get_restrictions()->get_clustering_columns_restrictions();
+    const cql3::expr::expression& r = view.select_statement(db).get_restrictions()->get_clustering_columns_restrictions();
     std::vector<bytes> exploded_pk = key.explode();
     std::vector<bytes> exploded_ck = ck.explode();
     std::vector<const column_definition*> ck_columns;
@@ -384,7 +384,7 @@ public:
     bool check_if_matches(const clustering_key& key, const query::result_row_view& static_row, const query::result_row_view& row) const {
         std::vector<bytes> ck = key.explode();
         return boost::algorithm::all_of(
-            _view.select_statement().get_restrictions()->get_non_pk_restriction() | boost::adaptors::map_values,
+            _view.select_statement(_db).get_restrictions()->get_non_pk_restriction() | boost::adaptors::map_values,
             [&] (auto&& r) {
                 // FIXME: move outside all_of(). However, crashes.
                 auto static_and_regular_columns = cql3::expr::get_non_pk_values(*_selection, static_row, &row);
@@ -1472,7 +1472,7 @@ future<query::clustering_row_ranges> calculate_affected_clustering_ranges(data_d
     if (mp.partition_tombstone() || !mp.row_tombstones().empty()) {
         for (auto&& v : views) {
             // FIXME: #2371
-            if (v.view->view_info()->select_statement().get_restrictions()->has_unrestricted_clustering_columns()) {
+            if (v.view->view_info()->select_statement(db).get_restrictions()->has_unrestricted_clustering_columns()) {
                 view_row_ranges.push_back(nonwrapping_range<clustering_key_prefix_view>::make_open_ended_both_sides());
                 break;
             }
diff --git a/view_info.hh b/view_info.hh
index dfccf3458c..2c24add3f0 100644
--- a/view_info.hh
+++ b/view_info.hh
@@ -48,7 +48,7 @@ public:
         return _raw.where_clause();
     }
 
-    cql3::statements::select_statement& select_statement() const;
+    cql3::statements::select_statement& select_statement(data_dictionary::database) const;
     const query::partition_slice& partition_slice(data_dictionary::database) const;
     const column_definition* view_column(const schema& base, column_kind kind, column_id base_id) const;
     const column_definition* view_column(const column_definition& base_def) const;

From edcce7d8dd02eb3c0c78a401fc49939aa91faab0 Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Thu, 20 Apr 2023 09:35:12 +0300
Subject: [PATCH 9/9] view_info: Drop calls to get_local_storage_proxy()

In both cases the proxy is called to get data_dictionary from. Now its
available as the call argument.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 db/view/view.cc | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/db/view/view.cc b/db/view/view.cc
index 228b5994e6..55f3dc4019 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -85,7 +85,7 @@ cql3::statements::select_statement& view_info::select_statement(data_dictionary:
         // FIXME(sarna): legacy code, should be removed after "computed_columns" feature is guaranteed
         // to be available on every node. Then, we won't need to check if this view is backing a secondary index.
         const column_definition* legacy_token_column = nullptr;
-        if (service::get_local_storage_proxy().local_db().find_column_family(base_id()).get_index_manager().is_global_index(_schema)) {
+        if (db.find_column_family(base_id()).get_index_manager().is_global_index(_schema)) {
            if (!_schema.clustering_key_columns().empty()) {
                legacy_token_column = &_schema.clustering_key_columns().front();
            }
@@ -103,7 +103,7 @@ cql3::statements::select_statement& view_info::select_statement(data_dictionary:
         raw->prepare_keyspace(_schema.ks_name());
         raw->set_bound_variables({});
         cql3::cql_stats ignored;
-        auto prepared = raw->prepare(service::get_local_storage_proxy().data_dictionary(), ignored, true);
+        auto prepared = raw->prepare(db, ignored, true);
         _select_statement = static_pointer_cast<cql3::statements::select_statement>(prepared->statement);
     }
     return *_select_statement;