cql3/restrictions: Use free functions instead of methods

Instead of `restriction` class methods, use the new free functions.
Specific replacement actions are listed below.

Note that class `restrictions` (plural) remains intact -- both its
methods and its type hierarchy remain intact for now.

Ensure full test coverage of the replacement code with new file
test/boost/restrictions_test.cc and some extra testcases in
test/cql/*.

Drop some existing tests because they codify buggy behaviour
(reference scylladb#6369, scylladb#6382).  Drop others because they forbid relation
combinations that are now allowed (eg, mixing equality and
inequality, comparing to NULL, etc.).

Here are some specific categories of what was replaced:

- restriction::is_foo predicates are replaced by using the free
  function find_if; sometimes it is used transitively (see, eg,
  has_slice)

- restriction::is_multi_column is replaced by dynamic casts (recall
  that the `restrictions` class hierarchy still exists)

- utility methods is_satisfied_by, is_supported_by, to_string, and
  uses_function are replaced by eponymous free functions; note that
  restrictions::uses_function still exists

- restriction::apply_to is replaced by free function
  replace_column_def

- when checking infinite_bound_range_deletions, the has_bound is
  replaced by local free function bounded_ck

- restriction::bounds and restriction::value are replaced by the more
  general free function possible_lhs_values

- using free functions allows us to simplify the
  multi_column_restriction and token_restriction hierarchies; their
  methods merge_with and uses_function became identical in all
  subclasses, so they were moved to the base class

- single_column_primary_key_restrictions<clustering_key>::needs_filtering
  was changed to reuse num_prefix_columns_that_need_not_be_filtered,
  which uses free functions

Fixes scylladb#5799.
Fixes scylladb#6369.
Fixes scylladb#6371.
Fixes scylladb#6372.
Fixes scylladb#6382.

Signed-off-by: Dejan Mircevski <dejan@scylladb.com>

configure.py

@@ -353,6 +353,7 @@ def find_headers(repodir, excluded_dirs):
     'test/boost/range_test',
     'test/boost/range_tombstone_list_test',
     'test/boost/reusable_buffer_test',
+    'test/boost/restrictions_test',
     'test/boost/role_manager_test',
     'test/boost/row_cache_test',
     'test/boost/schema_change_test',

cql3/restrictions/multi_column_restriction.hh

@@ -86,9 +86,9 @@ public:
     }
 
     virtual void merge_with(::shared_ptr<restriction> other) override {
-        statements::request_validations::check_true(other->is_multi_column(),
+        const auto as_pkr = dynamic_pointer_cast<clustering_key_restrictions>(other);
+        statements::request_validations::check_true(bool(as_pkr),
             "Mixing single column relations and multi column relations on clustering columns is not allowed");
-        auto as_pkr = static_pointer_cast<clustering_key_restrictions>(other);
         do_merge_with(as_pkr);
         update_asc_desc_existence();
         expression = make_conjunction(std::move(expression), other->expression);
@@ -109,6 +109,10 @@ public:
         });
     }
 
+    virtual bool uses_function(const sstring& ks_name, const sstring& function_name) const override {
+        return cql3::restrictions::uses_function(expression, ks_name, function_name);
+    }
+
 protected:
     virtual void do_merge_with(::shared_ptr<clustering_key_restrictions> other) = 0;
 
@@ -217,10 +221,6 @@ public:
             std::vector<column_value>(_column_defs.cbegin(), _column_defs.cend()), &operator_type::EQ, _value};
     }
 
-    virtual bool uses_function(const sstring& ks_name, const sstring& function_name) const override {
-        return restriction::term_uses_function(_value, ks_name, function_name);
-    }
-
     virtual bool is_supported_by(const secondary_index::index& index) const override {
         for (auto* cdef : _column_defs) {
             if (index.supports_expression(*cdef, cql3::operator_type::EQ)) {
@@ -386,10 +386,6 @@ public:
             ::make_shared<lists::delayed_value>(_values)};
     }
 
-    virtual bool uses_function(const sstring& ks_name, const sstring& function_name) const override  {
-        return restriction::term_uses_function(_values, ks_name, function_name);
-    }
-
     virtual sstring to_string() const override  {
         return format("IN({})", std::to_string(_values));
     }
@@ -422,10 +418,6 @@ public:
             std::move(marker)};
     }
 
-    virtual bool uses_function(const sstring& ks_name, const sstring& function_name) const override {
-        return false;
-    }
-
     virtual sstring to_string() const override {
         return "IN ?";
     }
@@ -512,26 +504,21 @@ public:
         return _slice.has_bound(b);
     }
 
-    virtual bool uses_function(const sstring& ks_name, const sstring& function_name) const override {
-        return (_slice.has_bound(statements::bound::START) && restriction::term_uses_function(_slice.bound(statements::bound::START), ks_name, function_name))
-                || (_slice.has_bound(statements::bound::END) && restriction::term_uses_function(_slice.bound(statements::bound::END), ks_name, function_name));
-    }
-
     virtual bool is_inclusive(statements::bound b) const override {
         return _slice.is_inclusive(b);
     }
 
     virtual void do_merge_with(::shared_ptr<clustering_key_restrictions> other) override {
         using namespace statements::request_validations;
-        check_true(other->is_slice(),
+        check_true(has_slice(other->expression),
                    "Column \"%s\" cannot be restricted by both an equality and an inequality relation",
                    get_columns_in_commons(other));
         auto other_slice = static_pointer_cast<slice>(other);
 
-        check_false(has_bound(statements::bound::START) && other_slice->has_bound(statements::bound::START),
+        check_false(_slice.has_bound(statements::bound::START) && other_slice->_slice.has_bound(statements::bound::START),
                     "More than one restriction was found for the start bound on %s",
                     get_columns_in_commons(other));
-        check_false(has_bound(statements::bound::END) && other_slice->has_bound(statements::bound::END),
+        check_false(_slice.has_bound(statements::bound::END) && other_slice->_slice.has_bound(statements::bound::END),
                     "More than one restriction was found for the end bound on %s",
                     get_columns_in_commons(other));
 
@@ -560,10 +547,10 @@ private:
     }
 
     std::vector<bytes_opt> read_bound_components(const query_options& options, statements::bound b) const {
-        if (!has_bound(b)) {
+        if (!_slice.has_bound(b)) {
             return {};
         }
-        auto vals = component_bounds(b, options);
+        auto vals = first_multicolumn_bound(expression, options, b);
         for (unsigned i = 0; i < vals.size(); i++) {
             statements::request_validations::check_not_null(vals[i], "Invalid null value in condition for column %s", _column_defs.at(i)->name_as_text());
         }
@@ -578,9 +565,9 @@ private:
      */
     std::vector<bounds_range_type> bounds_ranges_unified_order(const query_options& options) const {
         auto start_prefix = clustering_key_prefix::from_optional_exploded(*_schema, read_bound_components(options, statements::bound::START));
-        auto start_bound = bounds_range_type::bound(std::move(start_prefix), is_inclusive(statements::bound::START));
+        auto start_bound = bounds_range_type::bound(std::move(start_prefix), _slice.is_inclusive(statements::bound::START));
         auto end_prefix = clustering_key_prefix::from_optional_exploded(*_schema, read_bound_components(options, statements::bound::END));
-        auto end_bound = bounds_range_type::bound(std::move(end_prefix), is_inclusive(statements::bound::END));
+        auto end_bound = bounds_range_type::bound(std::move(end_prefix), _slice.is_inclusive(statements::bound::END));
         auto make_range = [&] () {
             if (is_asc_order()) {
                 return bounds_range_type::make(start_bound, end_bound);
@@ -719,8 +706,8 @@ private:
         std::vector<restriction_shared_ptr> ret;
         auto start_components = read_bound_components(options, statements::bound::START);
         auto end_components = read_bound_components(options, statements::bound::END);
-        bool start_inclusive = is_inclusive(statements::bound::START);
-        bool end_inclusive = is_inclusive(statements::bound::END);
+        bool start_inclusive = _slice.is_inclusive(statements::bound::START);
+        bool end_inclusive = _slice.is_inclusive(statements::bound::END);
         std::optional<std::size_t> first_neq_component = std::nullopt;
 
         // find the first index of the first component that is not equal between the tuples.

cql3/restrictions/primary_key_restrictions.hh

@@ -93,8 +93,8 @@ public:
     }
 
     virtual bool needs_filtering(const schema& schema) const {
-        return !empty() && !is_on_token() &&
-                (has_unrestricted_components(schema) || is_contains() || is_slice() || is_LIKE());
+        return !empty() && !has_token(expression) &&
+                (has_unrestricted_components(schema) || has_slice_or_needs_filtering(expression));
     }
 
     // NOTICE(sarna): This function is useless for partition key restrictions,