cql3: Reimplement uses of _nonprimary_key_restrictions using expression

All parts of the code that use _nonprimary_key_restrictions
are changed to use _new_nonprimary_key_restrictions instead.
I decided not to split this into multiple commits,
as there isn't a lot of changes and they are
analogous to the ones done before for partition
and clustering columns.

Signed-off-by: Jan Ciolek <jan.ciolek@scylladb.com>

cql3/restrictions/statement_restrictions.cc

@@ -456,7 +456,7 @@ statement_restrictions::statement_restrictions(data_dictionary::database db,
             && !type.is_delete();
     _has_queriable_pk_index = parition_key_restrictions_have_supporting_index(sim, allow_local)
             && !type.is_delete();
-    _has_queriable_regular_index = _nonprimary_key_restrictions->has_supporting_index(sim, allow_local)
+    _has_queriable_regular_index = expr::index_supports_some_column(_new_nonprimary_key_restrictions, sim, allow_local)
             && !type.is_delete();
 
     // At this point, the select statement if fully constructed, but we still have a few things to validate
@@ -526,7 +526,7 @@ statement_restrictions::statement_restrictions(data_dictionary::database db,
 #endif
     }
 
-    if (!_nonprimary_key_restrictions->empty()) {
+    if (!expr::is_empty_restriction(_new_nonprimary_key_restrictions)) {
         if (_has_queriable_regular_index && _partition_range_is_simple) {
             _uses_secondary_indexing = true;
         } else if (!allow_filtering) {
@@ -628,13 +628,8 @@ std::vector<const column_definition*> statement_restrictions::get_column_defs_fo
                 }
             }
         }
-        for (auto&& cdef : _nonprimary_key_restrictions->get_column_defs()) {
-            ::shared_ptr<restriction> cur_restr = _nonprimary_key_restrictions->get_restriction(*cdef);
-            const expr::expression* single_col_restr = nullptr;
-            if (is_single_column_restriction(cur_restr->expression)) {
-                single_col_restr = &cur_restr->expression;
-            }
-            if (!column_uses_indexing(cdef, single_col_restr)) {
+        for (auto&& [cdef, cur_restr] : _single_column_nonprimary_key_restrictions) {
+            if (!column_uses_indexing(cdef, &cur_restr)) {
                 column_defs_for_filtering.emplace_back(cdef);
             }
         }
@@ -1757,7 +1752,7 @@ bool statement_restrictions::need_filtering() const {
         // If there is a token(p1, p2) restriction, no p1, p2 restrictions are allowed in the query.
         // All other restrictions must be on clustering or regular columns.
         int64_t non_pk_restrictions_count = clustering_columns_restrictions_size();
-        non_pk_restrictions_count += _nonprimary_key_restrictions->size();
+        non_pk_restrictions_count += expr::get_sorted_column_defs(_new_nonprimary_key_restrictions).size();
 
         // We are querying using an index, one restriction goes to the index restriction.
         // If there are some restrictions other than token() and index column then we need to do filtering.
@@ -1770,7 +1765,8 @@ bool statement_restrictions::need_filtering() const {
         // Can't calculate the token value, so a naive base-table query must be filtered.  Same for any index tables,
         // except if there's only one restriction supported by an index.
         return !(npart == 1 && _has_queriable_pk_index &&
-                 expr::is_empty_restriction(_clustering_columns_restrictions) && _nonprimary_key_restrictions->empty());
+                 expr::is_empty_restriction(_clustering_columns_restrictions) &&
+                 expr::is_empty_restriction(_new_nonprimary_key_restrictions));
     }
     if (pk_restrictions_need_filtering()) {
         // We most likely cannot calculate token(s).  Neither base-table nor index-table queries can avoid filtering.
@@ -1778,7 +1774,7 @@ bool statement_restrictions::need_filtering() const {
     }
     // Now we know the partition key is either unrestricted or fully restricted.
 
-    const auto nreg = _nonprimary_key_restrictions->size();
+    const auto nreg = expr::get_sorted_column_defs(_new_nonprimary_key_restrictions).size();
     if (nreg > 1 || (nreg == 1 && !_has_queriable_regular_index)) {
         return true; // Regular columns are unsorted in storage and no single index suffices.
     }

cql3/restrictions/statement_restrictions.hh

@@ -463,7 +463,7 @@ public:
      * @return <code>true</code> if the restrictions contain any non-primary key restrictions, <code>false</code> otherwise.
      */
     bool has_non_primary_key_restriction() const {
-        return !_nonprimary_key_restrictions->empty();
+        return !expr::is_empty_restriction(_new_nonprimary_key_restrictions);
     }
 
     bool pk_restrictions_need_filtering() const;
@@ -498,8 +498,8 @@ public:
      /**
       * @return the non-primary key restrictions.
       */
-    const single_column_restrictions::restrictions_map& get_non_pk_restriction() const {
-        return _nonprimary_key_restrictions->restrictions();
+    const expr::single_column_restrictions_map& get_non_pk_restriction() const {
+        return _single_column_nonprimary_key_restrictions;
     }
 
     /**

cql3/selection/selection.cc

@@ -441,7 +441,7 @@ bool result_set_builder::restrictions_filter::do_filter(const selection& selecti
 
     auto static_row_iterator = static_row.iterator();
     auto row_iterator = row ? std::optional<query::result_row_view::iterator_type>(row->iterator()) : std::nullopt;
-    auto non_pk_restrictions_map = _restrictions->get_non_pk_restriction();
+    const expr::single_column_restrictions_map& non_pk_restrictions_map = _restrictions->get_non_pk_restriction();
     for (auto&& cdef : selection.get_columns()) {
         switch (cdef->kind) {
         case column_kind::static_column:
@@ -454,11 +454,11 @@ bool result_set_builder::restrictions_filter::do_filter(const selection& selecti
             if (restr_it == non_pk_restrictions_map.end()) {
                 continue;
             }
-            restrictions::restriction& single_col_restriction = *restr_it->second;
+            const expr::expression& single_col_restriction = restr_it->second;
             // FIXME: push to upper layer so it happens once per row
             auto static_and_regular_columns = expr::get_non_pk_values(selection, static_row, row);
             bool regular_restriction_matches = expr::is_satisfied_by(
-                    single_col_restriction.expression,
+                    single_col_restriction,
                     expr::evaluation_inputs{
                         .partition_key = &partition_key,
                         .clustering_key = &clustering_key,

cql3/statements/create_view_statement.cc

@@ -270,7 +270,7 @@ view_ptr create_view_statement::prepare_view(data_dictionary::database db) const
     // non-pk base column + base column used in view pk)". When the filtered
     // column *is* the base column added to the view pk, we don't have this
     // problem. And this case actually works correctly.
-    auto non_pk_restrictions = restrictions->get_non_pk_restriction();
+    const expr::single_column_restrictions_map& non_pk_restrictions = restrictions->get_non_pk_restriction();
     if (non_pk_restrictions.size() == 1 && has_non_pk_column &&
             target_primary_keys.contains(non_pk_restrictions.cbegin()->first)) {
         // This case (filter by new PK column of the view) works, as explained above

db/view/view.cc

@@ -346,7 +346,7 @@ public:
                 // FIXME: pass dummy_options as nullptr
                 auto dummy_options = cql3::query_options({});
                 return cql3::expr::is_satisfied_by(
-                        r->expression,
+                        r,
                         cql3::expr::evaluation_inputs{
                             .partition_key = &_pk,
                             .clustering_key = &ck,