Merge "cql3: lift infinite bound check" from Benny & Piotr

"
If the database supports infinite bound range deletions,
CQL layer will no longer throw an error indicating that both ranges
need to be specified.

Fixes scylladb#432

Update test_range_deletion_scenarios unit test accordingly.
"

* 'cql3-lift-infinite-bound-check' of https://github.com/bhalevy/scylla:
  cql3: lift infinite bound check if it's supported
  service: enable infinite bound range deletions with mc
  database: add flag for infinite bound range deletions

cql3/statements/delete_statement.cc

@@ -41,6 +41,7 @@
 
 #include "delete_statement.hh"
 #include "raw/delete_statement.hh"
+#include "database.hh"
 
 namespace cql3 {
 
@@ -102,9 +103,11 @@ delete_statement::prepare_internal(database& db, schema_ptr schema, shared_ptr<v
     }
 
     stmt->process_where_clause(db, _where_clause, std::move(bound_names));
-    if (!stmt->restrictions()->get_clustering_columns_restrictions()->has_bound(bound::START)
-            || !stmt->restrictions()->get_clustering_columns_restrictions()->has_bound(bound::END)) {
-        throw exceptions::invalid_request_exception("A range deletion operation needs to specify both bounds");
+    if (!db.supports_infinite_bound_range_deletions()) {
+        if (!stmt->restrictions()->get_clustering_columns_restrictions()->has_bound(bound::START)
+                || !stmt->restrictions()->get_clustering_columns_restrictions()->has_bound(bound::END)) {
+            throw exceptions::invalid_request_exception("A range deletion operation needs to specify both bounds for clusters without sstable mc format support");
+        }
     }
     if (!schema->is_compound() && stmt->restrictions()->get_clustering_columns_restrictions()->is_slice()) {
         throw exceptions::invalid_request_exception("Range deletions on \"compact storage\" schemas are not supported");

database.hh

@@ -1296,6 +1296,8 @@ private:
     friend db::data_listeners;
     std::unique_ptr<db::data_listeners> _data_listeners;
 
+    bool _supports_infinite_bound_range_deletions = false;
+
     future<> init_commitlog();
     future<> apply_in_memory(const frozen_mutation& m, schema_ptr m_schema, db::rp_handle&&, db::timeout_clock::time_point timeout);
     future<> apply_in_memory(const mutation& m, column_family& cf, db::rp_handle&&, db::timeout_clock::time_point timeout);
@@ -1510,6 +1512,14 @@ public:
     db::data_listeners& data_listeners() const {
         return *_data_listeners;
     }
+
+    void enable_infinite_bound_range_deletions() {
+        _supports_infinite_bound_range_deletions = true;
+    }
+
+    bool supports_infinite_bound_range_deletions() {
+        return _supports_infinite_bound_range_deletions;
+    }
 };
 
 future<> stop_database(sharded<database>& db);

service/storage_service.cc

@@ -184,6 +184,13 @@ storage_service::storage_service(distributed<database>& db, gms::gossiper& gossi
     } else {
         _sstables_format = sstables::sstable_version_types::mc;
     }
+
+    _unbounded_range_tombstones_feature.when_enabled().then([&db] () mutable {
+        slogger.debug("Enabling infinite bound range deletions");
+        db.invoke_on_all([] (database& local_db) mutable {
+            local_db.enable_infinite_bound_range_deletions();
+        });
+    });
 }
 
 void storage_service::enable_all_features() {

tests/cql_query_test.cc

@@ -1064,18 +1064,12 @@ SEASTAR_TEST_CASE(test_range_deletion_scenarios) {
             e.execute_cql(format("insert into cf (p, c, v) values (1, {:d}, 'abc');", i)).get();
         }
 
-        try {
-            e.execute_cql("delete from cf where p = 1 and c <= 3").get();
-            BOOST_FAIL("should've thrown");
-        } catch (...) { }
-        try {
-            e.execute_cql("delete from cf where p = 1 and c >= 0").get();
-            BOOST_FAIL("should've thrown");
-        } catch (...) { }
+        e.execute_cql("delete from cf where p = 1 and c <= 3").get();
+        e.execute_cql("delete from cf where p = 1 and c >= 8").get();
 
-        e.execute_cql("delete from cf where p = 1 and c >= 0 and c <= 3").get();
+        e.execute_cql("delete from cf where p = 1 and c >= 0 and c <= 5").get();
         auto msg = e.execute_cql("select * from cf").get0();
-        assert_that(msg).is_rows().with_size(6);
+        assert_that(msg).is_rows().with_size(2);
         e.execute_cql("delete from cf where p = 1 and c > 3 and c < 10").get();
         msg = e.execute_cql("select * from cf").get0();
         assert_that(msg).is_rows().with_size(0);