treewide: revamp workload type for service levels

This patch is not backward compatible with its original,
but it's considered fine, since the original workload types were not
yet part of any release.
The changes include:
 - instead of using 'unspecified' for declaring that there's no workload
   type for a particular service level, NULL is used for that purpose;
   NULL is the standard way of representing lack of data
 - introducing a delete marker, which accompanies NULL and makes it
   possible to distinguish between wanting to forcibly reset a workload
   type to unspecified and not wanting to change the previous value
 - updating the tests accordingly

These changes come in as a single patch, because they're intertwined
with each other and the tests for workload types are already in place;
an attempt to split them proved to be more complicated than it's worth.

Tests: unit(release)

Closes #8763

cql3/statements/list_service_level_statement.cc

@@ -88,10 +88,13 @@ list_service_level_statement::execute(query_processor& qp,
                 };
                 auto rs = std::make_unique<result_set>(metadata);
                 for (auto &&[sl_name, slo] : sl_info) {
+                    bytes_opt workload = slo.workload == qos::service_level_options::workload_type::unspecified
+                            ? bytes_opt()
+                            : utf8_type->decompose(qos::service_level_options::to_string(slo.workload));
                     rs->add_row(std::vector<bytes_opt>{
                             utf8_type->decompose(sl_name),
                             d(slo.timeout),
-                            utf8_type->decompose(qos::service_level_options::to_string(slo.workload))});
+                            workload});
                 }
 
                 auto rows = ::make_shared<cql_transport::messages::result_message::rows>(result(std::move(std::move(rs))));

cql3/statements/sl_prop_defs.cc

@@ -63,6 +63,11 @@ void sl_prop_defs::validate() {
             throw exceptions::invalid_request_exception(format("Invalid workload type: {}", *workload_string_opt));
         }
         _slo.workload = *workload;
+        // Explicitly setting a workload type to 'unspecified' should result in resetting
+        // the previous value to 'unspecified, not just keeping it as is
+        if (_slo.workload == qos::service_level_options::workload_type::unspecified) {
+            _slo.workload = qos::service_level_options::workload_type::delete_marker;
+        }
     }
 }
 

db/system_distributed_keyspace.cc

@@ -800,11 +800,14 @@ future<> system_distributed_keyspace::set_service_level(sstring service_level_na
             },
         }, tv);
     };
+    data_value workload = slo.workload == qos::service_level_options::workload_type::unspecified
+            ? data_value::make_null(utf8_type)
+            : data_value(qos::service_level_options::to_string(slo.workload));
     co_await _qp.execute_internal(format("UPDATE {}.{} SET timeout = ?, workload_type = ? WHERE service_level = ?;", NAME, SERVICE_LEVELS),
                 db::consistency_level::ONE,
                 internal_distributed_query_state(),
                 {to_data_value(slo.timeout),
-                    data_value(qos::service_level_options::to_string(slo.workload)),
+                    workload,
                     service_level_name});
 }
 

service/qos/qos_common.cc

@@ -39,8 +39,16 @@ service_level_options service_level_options::replace_defaults(const service_leve
             // leave the value as is
         },
     }, ret.timeout);
-    if (ret.workload == workload_type::unspecified) {
+    switch (ret.workload) {
+    case workload_type::unspecified:
         ret.workload = default_values.workload;
+        break;
+    case workload_type::delete_marker:
+        ret.workload = workload_type::unspecified;
+        break;
+    default:
+        // no-op
+        break;
     }
     return ret;
 }
@@ -74,6 +82,7 @@ std::string_view service_level_options::to_string(const workload_type& wt) {
     case workload_type::unspecified: return "unspecified";
     case workload_type::batch: return "batch";
     case workload_type::interactive: return "interactive";
+    case workload_type::delete_marker: return "delete_marker";
     }
     abort();
 }
@@ -83,7 +92,7 @@ std::ostream& operator<<(std::ostream& os, const service_level_options::workload
 }
 
 std::optional<service_level_options::workload_type> service_level_options::parse_workload_type(std::string_view sv) {
-    if (sv == "unspecified") {
+    if (sv == "null") {
         return workload_type::unspecified;
     } else if (sv == "interactive") {
         return workload_type::interactive;

service/qos/qos_common.hh

@@ -46,7 +46,7 @@ struct service_level_options {
     };
 
     enum class workload_type {
-        unspecified, batch, interactive
+        unspecified, batch, interactive, delete_marker
     };
 
     using timeout_type = std::variant<unset_marker, delete_marker, lowres_clock::duration>;

test/boost/auth_test.cc

@@ -341,15 +341,15 @@ SEASTAR_TEST_CASE(test_alter_with_workload_type) {
 
 	    auto msg = cquery_nofail(e, "SELECT workload_type FROM system_distributed.service_levels");
         assert_that(msg).is_rows().with_rows({{
-            utf8_type->decompose("unspecified")
+            {}
         }});
 
         e.refresh_client_state().get();
         // Default workload type is `unspecified`
         BOOST_REQUIRE_EQUAL(e.local_client_state().get_workload_type(), service::client_state::workload_type::unspecified);
 
         // When multiple per-role timeouts apply, the smallest value is always effective
-        cquery_nofail(e, "CREATE SERVICE LEVEL sl2 WITH workload_type = 'unspecified'");
+        cquery_nofail(e, "CREATE SERVICE LEVEL sl2 WITH workload_type = null");
         cquery_nofail(e, "CREATE SERVICE LEVEL sl3 WITH workload_type = 'batch'");
         cquery_nofail(e, "CREATE SERVICE LEVEL sl4 WITH workload_type = 'interactive'");
         cquery_nofail(e, "ATTACH SERVICE LEVEL sl2 TO user2");

test/boost/cql_query_test.cc

@@ -4889,14 +4889,14 @@ SEASTAR_TEST_CASE(test_user_based_sla_queries) {
         e.execute_cql("CREATE SERVICE_LEVEL sl_1;").get();
         auto msg = e.execute_cql("LIST SERVICE_LEVEL sl_1;").get0();
         assert_that(msg).is_rows().with_rows({
-            {utf8_type->decompose("sl_1"), {}, utf8_type->decompose("unspecified")},
+            {utf8_type->decompose("sl_1"), {}, {}},
         });
         e.execute_cql("CREATE SERVICE_LEVEL sl_2;").get();
         //drop service levels
         e.execute_cql("DROP SERVICE_LEVEL sl_1;").get();
         msg = e.execute_cql("LIST ALL SERVICE_LEVELS;").get0();
         assert_that(msg).is_rows().with_rows({
-            {utf8_type->decompose("sl_2"), {}, utf8_type->decompose("unspecified")},
+            {utf8_type->decompose("sl_2"), {}, {}},
         });
 
         // validate exceptions (illegal requests)

test/cql-pytest/test_service_levels.py

@@ -73,16 +73,26 @@ def test_attached_service_level(scylla_only, cql):
 def test_set_workload_type(scylla_only, cql):
     with new_service_level(cql) as sl:
         res = cql.execute(f"LIST SERVICE LEVEL {sl}")
-        assert res.one().workload_type == 'unspecified'
-        for wt in ['interactive', 'batch', 'unspecified']:
+        assert not res.one().workload_type
+        for wt in ['interactive', 'batch']:
             cql.execute(f"ALTER SERVICE LEVEL {sl} WITH workload_type = '{wt}'")
             res = cql.execute(f"LIST SERVICE LEVEL {sl}")
             assert res.one().workload_type == wt
 
- # Test that workload type input is validated
+# Test that workload type input is validated
 def test_set_invalid_workload_types(scylla_only, cql):
     with new_service_level(cql) as sl:
-        for incorrect in ['', 'null', 'i', 'b', 'dog', 'x'*256]:
+        for incorrect in ['', 'i', 'b', 'dog', 'x'*256]:
             print(f"Checking {incorrect}")
             with pytest.raises(Exception):
-                cql.execute(f"ALTER SERVICE LEVEL {sl} WITH workload_type = '{incorrect}'")
+                cql.execute(f"ALTER SERVICE LEVEL {sl} WITH workload_type = '{incorrect}'")
+
+# Test that resetting an already set workload type by assigning NULL to it works fine
+def test_reset_workload_type(scylla_only, cql):
+    with new_service_level(cql) as sl:
+        cql.execute(f"ALTER SERVICE LEVEL {sl} WITH workload_type = 'interactive'")
+        res = cql.execute(f"LIST SERVICE LEVEL {sl}")
+        assert res.one().workload_type == 'interactive'
+        cql.execute(f"ALTER SERVICE LEVEL {sl} WITH workload_type = null")
+        res = cql.execute(f"LIST SERVICE LEVEL {sl}")
+        assert not res.one().workload_type