Merge 'strong_consistency: limit available consistency levels in strong consistent requests' from Michał Jadwiszczak

Strong consistent requests take different patch then EC requests and consistency levels don’t map well.
We should limit available consistency levels in SC request to avoid ignoring them silently, which may cause confusion to user.

For writes, there is only one option:
- QUORUM/LOCAL_QUORUM (multi DC is not supported yet, so both of those CLs have the same effect) - we need quorum of replicas to successfully commit new mutations to Raft log.

For reads, there are 2 options:
- QUORUM/LOCAL_QUORUM - if user wants to be sure he sees latest data and the query needs to execute `read_barrier()`, which requires quorum of replicas
- ONE/LOCAL_ONE - if user just wants to read data from one replica without synchronization

All tests were updated to use LOCAL_QUORUM for both read and writes.

Fixes SCYLLADB-1766

SC is in experimental phase and this patch is an improvement, no backport needed.

Closes scylladb/scylladb#29691

* github.com:scylladb/scylladb:
  strong_consistency: allow QUORUM/LOCAL_QUORUM and ONE/LOCAL_ONE for reads
  strong_consistency: allow only QUORUM/LOCAL_QUORUM CL for writes

cql3/statements/strong_consistency/modification_statement.cc

@@ -8,6 +8,7 @@
 
 #include "modification_statement.hh"
 
+#include "db/consistency_level_type.hh"
 #include "db/timeout_clock.hh"
 #include "transport/messages/result_message.hh"
 #include "cql3/query_processor.hh"
@@ -34,10 +35,18 @@ future<shared_ptr<result_message>> modification_statement::execute(query_process
             .then(cql_transport::messages::propagate_exception_as_future<shared_ptr<result_message>>);
 }
 
+static void validate_consistency_level(const db::consistency_level& cl) {
+    if (cl != db::consistency_level::QUORUM && cl != db::consistency_level::LOCAL_QUORUM) {
+        throw exceptions::invalid_request_exception("Strongly consistent writes must use QUORUM/LOCAL_QUORUM consistency level");
+    }
+}
+
 future<shared_ptr<result_message>> modification_statement::execute_without_checking_exception_message(
         query_processor& qp, service::query_state& qs, const query_options& options,
         std::optional<service::group0_guard> guard) const
 {
+    validate_consistency_level(options.get_consistency());
+
     auto timeout = db::timeout_clock::now() + _statement->get_timeout(qs.get_client_state(), options);
     auto json_cache = base_statement::json_cache_opt{};
     const auto keys = _statement->build_partition_keys(options, json_cache);

cql3/statements/strong_consistency/select_statement.cc

@@ -8,6 +8,7 @@
 
 #include "select_statement.hh"
 
+#include "db/consistency_level_type.hh"
 #include "query/query-request.hh"
 #include "cql3/query_processor.hh"
 #include "service/strong_consistency/coordinator.hh"
@@ -17,10 +18,19 @@ namespace cql3::statements::strong_consistency {
 
 using result_message = cql_transport::messages::result_message;
 
+static void validate_consistency_level(const db::consistency_level& cl) {
+    if (cl != db::consistency_level::QUORUM && cl != db::consistency_level::LOCAL_QUORUM &&
+        cl != db::consistency_level::ONE && cl != db::consistency_level::LOCAL_ONE) {
+        throw exceptions::invalid_request_exception("Strongly consistent reads must use QUORUM/LOCAL_QUORUM or ONE/LOCAL_ONE consistency level");
+    }
+}
+
 future<::shared_ptr<result_message>> select_statement::do_execute(query_processor& qp,
         service::query_state& state, 
         const query_options& options) const
 {
+    validate_consistency_level(options.get_consistency());
+
     const auto key_ranges = _restrictions->get_partition_key_ranges(options);
     if (key_ranges.size() != 1 || !query::is_single_partition(key_ranges[0])) {
         throw exceptions::invalid_request_exception("Strongly consistent queries can only target a single partition");

test/cluster/test_strong_consistency.py

@@ -215,7 +215,7 @@ async def test_basic_write_read(manager: ManagerClient):
 
         # Test with prepared statements as well
         insert_stmt = cql.prepare(f"INSERT INTO {ks}.test (pk, c) VALUES (?, ?)")
-        bound_insert_stmt = BoundStatement(insert_stmt, consistency_level=ConsistencyLevel.ONE)
+        bound_insert_stmt = BoundStatement(insert_stmt)
         select_stmt = cql.prepare(f"SELECT * FROM {ks}.test WHERE pk = ?")
         bound_select_stmt = BoundStatement(select_stmt, consistency_level=ConsistencyLevel.ONE)
         bound_select_stmt.bind([10])