cql3: fix fetching clustering key columns for filtering

When a column is not present in the select clause, but used for
filtering, it usually needs to be fetched from replicas.
Sometimes it can be avoided, e.g. if primary key columns form a valid
prefix - then, they will be optimized out before filtering itself.
However, clustering key prefix can only be qualified for this
optimization if the whole partition key is restricted - otherwise
the clustering columns still need to be present for filtering.

This commit also fixes tests in cql_query_test suite, because they now
expect more values - columns fetched for filtering will be present as
well (only internally, the clients receive only data they asked for).

Fixes #4541
Message-Id: <f08ebae5562d570ece2bb7ee6c84e647345dfe48.1560410018.git.sarna@scylladb.com>

(cherry picked from commit adeea0a)

cql3/restrictions/statement_restrictions.cc

@@ -392,8 +392,9 @@ std::vector<const column_definition*> statement_restrictions::get_column_defs_fo
                 }
             }
         }
-        if (_clustering_columns_restrictions->needs_filtering(*_schema)) {
-            column_id first_filtering_id = _schema->clustering_key_columns().begin()->id +
+        const bool pk_has_unrestricted_components = _partition_key_restrictions->has_unrestricted_components(*_schema);
+        if (pk_has_unrestricted_components || _clustering_columns_restrictions->needs_filtering(*_schema)) {
+            column_id first_filtering_id = pk_has_unrestricted_components ? 0 : _schema->clustering_key_columns().begin()->id +
                     _clustering_columns_restrictions->num_prefix_columns_that_need_not_be_filtered();
             for (auto&& cdef : _clustering_columns_restrictions->get_column_defs()) {
                 if (cdef->id >= first_filtering_id && !column_uses_indexing(cdef)) {

tests/cql_query_test.cc

@@ -2481,8 +2481,8 @@ SEASTAR_TEST_CASE(test_in_restriction) {
             return e.execute_cql("select r1 from tir2 where (c1,r1) in ((0, 1),(1,2),(0,1),(1,2),(3,3)) ALLOW FILTERING;");
         }).then([&e] (shared_ptr<cql_transport::messages::result_message> msg) {
             assert_that(msg).is_rows().with_rows({
-                {int32_type->decompose(1)},
-                {int32_type->decompose(2)},
+                {int32_type->decompose(1), int32_type->decompose(0)},
+                {int32_type->decompose(2), int32_type->decompose(1)},
             });
         }).then([&e] {
              return e.prepare("select r1 from tir2 where (c1,r1) in ? ALLOW FILTERING;");
@@ -2507,8 +2507,8 @@ SEASTAR_TEST_CASE(test_in_restriction) {
             return e.execute_prepared(prepared_id,raw_values);
         }).then([&e] (shared_ptr<cql_transport::messages::result_message> msg) {
             assert_that(msg).is_rows().with_rows({
-                {int32_type->decompose(1)},
-                {int32_type->decompose(2)},
+                {int32_type->decompose(1), int32_type->decompose(0)},
+                {int32_type->decompose(2), int32_type->decompose(1)},
             });
         });
     });