db,view: migrate checking view filter to new is_satisfied_by

In order to unify the interfaces, the is_satisfied_by flavor
for mutations is getting deprecated. In order to be able to remove it,
one of its biggest users, the matches_view_filter() function,
is translated to the other variant.

db/view/view.cc

@@ -82,6 +82,7 @@
 #include "types/map.hh"
 #include "utils/error_injection.hh"
 #include "utils/exponential_backoff_retry.hh"
+#include "query-result-writer.hh"
 
 using namespace std::chrono_literals;
 
@@ -317,6 +318,54 @@ static bool is_partition_key_empty(
     }
 }
 
+// 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 {
+    const schema& _base;
+    const view_info& _view;
+    ::shared_ptr<cql3::selection::selection> _selection;
+    std::vector<bytes> _pk;
+
+    bool _matches_view_filter = true;
+public:
+    view_filter_checking_visitor(const schema& base, const view_info& view)
+        : _base(base)
+        , _view(view)
+        , _selection(cql3::selection::selection::wildcard(_base.shared_from_this()))
+    {}
+
+    void accept_new_partition(const partition_key& key, uint64_t row_count) {
+        _pk = key.explode();
+    }
+    void accept_new_partition(uint64_t row_count) {
+        throw std::logic_error("view_filter_checking_visitor expects an explicit partition key");
+    }
+
+    void accept_new_row(const clustering_key& key, const query::result_row_view& static_row, const query::result_row_view& row) {
+        _matches_view_filter = _matches_view_filter && check_if_matches(key, static_row, row);
+    }
+
+    void accept_new_row(const query::result_row_view& static_row, const query::result_row_view& row) {
+        throw std::logic_error("view_filter_checking_visitor expects an explicit clustering key");
+    }
+    void accept_partition_end(const query::result_row_view& static_row) {}
+
+    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,
+            [&] (auto&& r) {
+                return cql3::expr::is_satisfied_by(
+                        r->expression, _pk, ck, static_row, &row, *_selection, cql3::query_options({ }));
+            }
+        );
+    }
+
+    bool matches_view_filter() const {
+        return _matches_view_filter;
+    }
+};
+
 static query::partition_slice make_partition_slice(const schema& s) {
     query::partition_slice::option_set opts;
     opts.set(query::partition_slice::option::send_partition_key);
@@ -358,13 +407,18 @@ public:
 };
 
 bool matches_view_filter(const schema& base, const view_info& view, const partition_key& key, const clustering_row& update, gc_clock::time_point now) {
+    auto slice = make_partition_slice(base);
+
+    data_query_result_builder builder(base, slice);
+    builder.consume_new_partition(dht::decorate_key(base, key));
+    builder.consume(clustering_row(base, update), 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);
+    query::result_view::consume(result, slice, visitor);
+
     return clustering_prefix_matches(base, view, key, update.key(), now)
-            && boost::algorithm::all_of(
-                view.select_statement().get_restrictions()->get_non_pk_restriction() | boost::adaptors::map_values,
-                [&] (auto&& r) {
-                    return cql3::expr::is_satisfied_by(
-                            r->expression, base, key, update.key(), update.cells(), cql3::query_options({ }), now);
-                });
+            && visitor.matches_view_filter();
 }
 
 void view_updates::move_to(utils::chunked_vector<frozen_mutation_and_schema>& mutations) {