Refactor to use table::add_new_sstable() and add test

- Add table::add_new_sstable() method that wraps error handling
- Update streaming/consumer.cc to use new method
- Update streaming/stream_blob.cc to use new method
- Add test_add_new_sstable_cleanup_on_failure test to verify cleanup

Addresses feedback from @tgrabiec to extract common logic into a reusable method.

Co-authored-by: tgrabiec <283695+tgrabiec@users.noreply.github.com>

.github/CODEOWNERS

@@ -1,5 +1,5 @@
 # AUTH
-auth/* @nuivall
+auth/* @nuivall @ptrsmrn
 
 # CACHE
 row_cache* @tgrabiec
@@ -25,11 +25,11 @@ compaction/* @raphaelsc
 transport/*
 
 # CQL QUERY LANGUAGE
-cql3/* @tgrabiec @nuivall
+cql3/* @tgrabiec @nuivall @ptrsmrn
 
 # COUNTERS
-counters* @nuivall
-tests/counter_test* @nuivall
+counters* @nuivall @ptrsmrn
+tests/counter_test* @nuivall @ptrsmrn
 
 # DOCS
 docs/* @annastuchlik @tzach
@@ -57,6 +57,7 @@ repair/* @tgrabiec @asias
 
 # SCHEMA MANAGEMENT
 db/schema_tables* @tgrabiec
+db/legacy_schema_migrator* @tgrabiec
 service/migration* @tgrabiec
 schema* @tgrabiec
 

.github/copilot-instructions.md

@@ -84,14 +84,3 @@ ninja build/<mode>/scylla
 - Strive for simplicity and clarity, add complexity only when clearly justified
 - Question requests: don't blindly implement requests - evaluate trade-offs, identify issues, and suggest better alternatives when appropriate
 - Consider different approaches, weigh pros and cons, and recommend the best fit for the specific context
-
-## Test Philosophy
-- Performance matters. Tests should run as quickly as possible. Sleeps in the code are highly discouraged and should be avoided, to reduce run time and flakiness.
-- Stability matters. Tests should be stable. New tests should be executed 100 times at least to ensure they pass 100 out of 100 times. (use --repeat 100 --max-failures 1 when running it)
-- Unit tests should ideally test one thing and one thing only.
-- Tests for bug fixes should run before the fix - and show the failure and after the fix - and show they now pass.
-- Tests for bug fixes should have in their comments which bug fixes (GitHub or JIRA issue) they test.
-- Tests in debug are always slower, so if needed, reduce number of iterations, rows, data used, cycles, etc. in debug mode.
-- Tests should strive to be repeatable, and not use random input that will make their results unpredictable.
-- Tests should consume as little resources as possible. Prefer running tests on a single node if it is sufficient, for example.
-

.github/scripts/auto-backport.py

@@ -62,7 +62,7 @@ def create_pull_request(repo, new_branch_name, base_branch_name, pr, backport_pr
         if is_draft:
             labels_to_add.append("conflicts")
             pr_comment = f"@{pr.user.login} - This PR was marked as draft because it has conflicts\n"
-            pr_comment += "Please resolve them and remove the 'conflicts' label. The PR will be made ready for review automatically."
+            pr_comment += "Please resolve them and mark this PR as ready for review"
             backport_pr.create_issue_comment(pr_comment)
         
         # Apply all labels at once if we have any

.github/workflows/backport-pr-fixes-validation.yaml

@@ -18,7 +18,7 @@ jobs:
             
             // Regular expression pattern to check for "Fixes" prefix
             // Adjusted to dynamically insert the repository full name
-            const pattern = `Fixes:? ((?:#|${repo.replace('/', '\\/')}#|https://github\\.com/${repo.replace('/', '\\/')}/issues/)(\\d+)|([A-Z]+-\\d+))`;
+            const pattern = `Fixes:? (?:#|${repo.replace('/', '\\/')}#|https://github\\.com/${repo.replace('/', '\\/')}/issues/)(\\d+)`;
             const regex = new RegExp(pattern);
             
             if (!regex.test(body)) {

.github/workflows/call_jira_status_in_progress.yml

@@ -0,0 +1,12 @@
+name: Call Jira Status In Progress
+
+on:
+  pull_request_target:
+    types: [opened]
+
+jobs:
+  call-jira-status-in-progress:
+    uses: scylladb/github-automation/.github/workflows/main_update_jira_status_to_in_progress.yml@main
+    secrets:
+      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
+

.github/workflows/call_jira_status_in_review.yml

@@ -0,0 +1,12 @@
+name: Call Jira Status In Review
+
+on:
+  pull_request_target:
+    types: [ready_for_review, review_requested]
+
+jobs:
+  call-jira-status-in-review:
+    uses: scylladb/github-automation/.github/workflows/main_update_jira_status_to_in_review.yml@main
+    secrets:
+      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
+

.github/workflows/call_jira_status_ready_for_merge.yml

@@ -0,0 +1,12 @@
+name: Call Jira Status Ready For Merge
+
+on:
+  pull_request_target:
+    types: [labeled]
+
+jobs:
+  call-jira-status-update:
+    uses: scylladb/github-automation/.github/workflows/main_update_jira_status_to_ready_for_merge.yml@main
+    secrets:
+      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
+

.github/workflows/call_jira_sync.yml

@@ -1,41 +0,0 @@
-name: Sync Jira Based on PR Events
-
-on:
-  pull_request_target:
-    types: [opened, ready_for_review, review_requested, labeled, unlabeled, closed]
-
-permissions:
-  contents: read
-  pull-requests: write
-  issues: write
-
-jobs:
-  jira-sync-pr-opened:
-    if: github.event.action == 'opened'
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_pr_opened.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
-
-  jira-sync-in-review:
-    if: github.event.action == 'ready_for_review' || github.event.action == 'review_requested'
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_in_review.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
-
-  jira-sync-add-label:
-    if: github.event.action == 'labeled'
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_add_label.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
-
-  jira-status-remove-label:
-    if: github.event.action == 'unlabeled'
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_remove_label.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
-
-  jira-status-pr-closed:
-    if: github.event.action == 'closed' 
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_pr_closed.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}

.github/workflows/call_sync_milestone_to_jira.yml

@@ -1,14 +0,0 @@
-name: Call Jira release creation for new milestone
-
-on:
-  milestone:
-    types: [created]
-
-jobs:
-  sync-milestone-to-jira:
-    uses: scylladb/github-automation/.github/workflows/main_sync_milestone_to_jira_release.yml@main
-    with:
-      # Comma-separated list of Jira project keys
-      jira_project_keys: "SCYLLADB,CUSTOMER"
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}

.github/workflows/call_validate_pr_author_email.yml

@@ -1,13 +0,0 @@
-name: validate_pr_author_email
-
-on:
-  pull_request_target:
-    types:
-      - opened
-      - synchronize
-      - reopened
-
-jobs:
-  validate_pr_author_email:
-    uses: scylladb/github-automation/.github/workflows/validate_pr_author_email.yml@main
-

.github/workflows/codespell.yaml

@@ -13,5 +13,5 @@ jobs:
       - uses: codespell-project/actions-codespell@master
         with:
           only_warn: 1
-          ignore_words_list: "ans,datas,fo,ser,ue,crate,nd,reenable,strat,stap,te,raison,iif,tread"
+          ignore_words_list: "ans,datas,fo,ser,ue,crate,nd,reenable,strat,stap,te,raison"
           skip: "./.git,./build,./tools,*.js,*.lock,./test,./licenses,./redis/lolwut.cc,*.svg"

.github/workflows/docs-pages.yaml

@@ -18,8 +18,6 @@ on:
 
 jobs:
   release:
-    permissions:
-      contents: write
     runs-on: ubuntu-latest
     steps:
       - name: Checkout

.github/workflows/docs-pr.yaml

@@ -2,9 +2,6 @@ name: "Docs / Build PR"
 # For more information,
 # see https://sphinx-theme.scylladb.com/stable/deployment/production.html#available-workflows
 
-permissions:
-  contents: read
-
 env:
   FLAG: ${{ github.repository == 'scylladb/scylla-enterprise' && 'enterprise' || 'opensource' }}
 

.github/workflows/docs-validate-metrics.yml

@@ -1,8 +1,5 @@
 name: Docs / Validate metrics
 
-permissions:
-  contents: read
-
 on:
   pull_request:
     branches:
@@ -10,7 +7,7 @@ on:
       - enterprise
     paths:
       - '**/*.cc'
-      - 'scripts/metrics-config.yml'
+      - 'scripts/metrics-config.yml' 
       - 'scripts/get_description.py'
       - 'docs/_ext/scylladb_metrics.py'
 
@@ -18,20 +15,20 @@ jobs:
   validate-metrics:
     runs-on: ubuntu-latest
     name: Check metrics documentation coverage
-
+    
     steps:
     - name: Checkout code
       uses: actions/checkout@v4
       with:
         submodules: true
-
+      
     - name: Set up Python
       uses: actions/setup-python@v6
       with:
         python-version: '3.10'
-
+        
     - name: Install dependencies
       run: pip install PyYAML
-
+        
     - name: Validate metrics
       run: python3 scripts/get_description.py --validate -c scripts/metrics-config.yml

.github/workflows/read-toolchain.yaml

@@ -10,8 +10,6 @@ on:
 jobs:
   read-toolchain:
     runs-on: ubuntu-latest
-    permissions:
-      contents: read
     outputs:
       image: ${{ steps.read.outputs.image }}
     steps:

.github/workflows/trigger-scylla-ci.yaml

@@ -3,13 +3,10 @@ name: Trigger Scylla CI Route
 on:
   issue_comment:
     types: [created]
-  pull_request_target:
-    types:
-      - unlabeled
 
 jobs:
   trigger-jenkins:
-    if: (github.event.comment.user.login != 'scylladbbot' && contains(github.event.comment.body, '@scylladbbot') && contains(github.event.comment.body, 'trigger-ci')) || github.event.label.name == 'conflicts'
+    if: github.event.comment.user.login != 'scylladbbot' && contains(github.event.comment.body, '@scylladbbot') && contains(github.event.comment.body, 'trigger-ci')
     runs-on: ubuntu-latest
     steps:
       - name: Trigger Scylla-CI-Route Jenkins Job

alternator/CMakeLists.txt

@@ -18,7 +18,6 @@ target_sources(alternator
     consumed_capacity.cc
     ttl.cc
     parsed_expression_cache.cc
-    http_compression.cc
     ${cql_grammar_srcs})
 target_include_directories(alternator
   PUBLIC

alternator/conditions.cc

@@ -42,7 +42,7 @@ comparison_operator_type get_comparison_operator(const rjson::value& comparison_
     if (!comparison_operator.IsString()) {
         throw api_error::validation(fmt::format("Invalid comparison operator definition {}", rjson::print(comparison_operator)));
     }
-    std::string op = rjson::to_string(comparison_operator);
+    std::string op = comparison_operator.GetString();
     auto it = ops.find(op);
     if (it == ops.end()) {
         throw api_error::validation(fmt::format("Unsupported comparison operator {}", op));
@@ -377,8 +377,8 @@ bool check_compare(const rjson::value* v1, const rjson::value& v2, const Compara
         return cmp(unwrap_number(*v1, cmp.diagnostic), unwrap_number(v2, cmp.diagnostic));
     }
     if (kv1.name == "S") {
-        return cmp(rjson::to_string_view(kv1.value),
-                   rjson::to_string_view(kv2.value));
+        return cmp(std::string_view(kv1.value.GetString(), kv1.value.GetStringLength()),
+                   std::string_view(kv2.value.GetString(), kv2.value.GetStringLength()));
     }
     if (kv1.name == "B") {
         auto d_kv1 = unwrap_bytes(kv1.value, v1_from_query);
@@ -470,9 +470,9 @@ static bool check_BETWEEN(const rjson::value* v, const rjson::value& lb, const r
         return check_BETWEEN(unwrap_number(*v, diag), unwrap_number(lb, diag), unwrap_number(ub, diag), bounds_from_query);
     }
     if (kv_v.name == "S") {
-        return check_BETWEEN(rjson::to_string_view(kv_v.value),
-                             rjson::to_string_view(kv_lb.value),
-                             rjson::to_string_view(kv_ub.value),
+        return check_BETWEEN(std::string_view(kv_v.value.GetString(), kv_v.value.GetStringLength()),
+                             std::string_view(kv_lb.value.GetString(), kv_lb.value.GetStringLength()),
+                             std::string_view(kv_ub.value.GetString(), kv_ub.value.GetStringLength()),
                              bounds_from_query);
     }
     if (kv_v.name == "B") {

alternator/consumed_capacity.cc

@@ -8,8 +8,6 @@
 
 #include "consumed_capacity.hh"
 #include "error.hh"
-#include "utils/rjson.hh"
-#include <fmt/format.h>
 
 namespace alternator {
 
@@ -34,12 +32,12 @@ bool consumed_capacity_counter::should_add_capacity(const rjson::value& request)
     if (!return_consumed->IsString()) {
         throw api_error::validation("Non-string ReturnConsumedCapacity field in request");
     }
-    std::string_view consumed = rjson::to_string_view(*return_consumed);
+    std::string consumed = return_consumed->GetString();
     if (consumed == "INDEXES") {
         throw api_error::validation("INDEXES consumed capacity is not supported");
     }
     if (consumed != "TOTAL") {
-        throw api_error::validation(fmt::format("Unknown consumed capacity {}", consumed));
+        throw api_error::validation("Unknown consumed capacity "+ consumed);
     }
     return true;
 }

alternator/controller.cc

@@ -28,7 +28,6 @@ static logging::logger logger("alternator_controller");
 controller::controller(
         sharded<gms::gossiper>& gossiper,
         sharded<service::storage_proxy>& proxy,
-        sharded<service::storage_service>& ss,
         sharded<service::migration_manager>& mm,
         sharded<db::system_distributed_keyspace>& sys_dist_ks,
         sharded<cdc::generation_service>& cdc_gen_svc,
@@ -40,7 +39,6 @@ controller::controller(
     : protocol_server(sg)
     , _gossiper(gossiper)
     , _proxy(proxy)
-    , _ss(ss)
     , _mm(mm)
     , _sys_dist_ks(sys_dist_ks)
     , _cdc_gen_svc(cdc_gen_svc)
@@ -91,7 +89,7 @@ future<> controller::start_server() {
         auto get_timeout_in_ms = [] (const db::config& cfg) -> utils::updateable_value<uint32_t> {
             return cfg.alternator_timeout_in_ms;
         };
-        _executor.start(std::ref(_gossiper), std::ref(_proxy), std::ref(_ss), std::ref(_mm), std::ref(_sys_dist_ks),
+        _executor.start(std::ref(_gossiper), std::ref(_proxy), std::ref(_mm), std::ref(_sys_dist_ks),
                         sharded_parameter(get_cdc_metadata, std::ref(_cdc_gen_svc)), _ssg.value(),
                         sharded_parameter(get_timeout_in_ms, std::ref(_config))).get();
         _server.start(std::ref(_executor), std::ref(_proxy), std::ref(_gossiper), std::ref(_auth_service), std::ref(_sl_controller)).get();
@@ -105,23 +103,11 @@ future<> controller::start_server() {
             alternator_port = _config.alternator_port();
             _listen_addresses.push_back({addr, *alternator_port});
         }
-        std::optional<uint16_t> alternator_port_proxy_protocol;
-        if (_config.alternator_port_proxy_protocol()) {
-            alternator_port_proxy_protocol = _config.alternator_port_proxy_protocol();
-            _listen_addresses.push_back({addr, *alternator_port_proxy_protocol});
-        }
         std::optional<uint16_t> alternator_https_port;
-        std::optional<uint16_t> alternator_https_port_proxy_protocol;
         std::optional<tls::credentials_builder> creds;
-        if (_config.alternator_https_port() || _config.alternator_https_port_proxy_protocol()) {
-            if (_config.alternator_https_port()) {
-                alternator_https_port = _config.alternator_https_port();
-                _listen_addresses.push_back({addr, *alternator_https_port});
-            }
-            if (_config.alternator_https_port_proxy_protocol()) {
-                alternator_https_port_proxy_protocol = _config.alternator_https_port_proxy_protocol();
-                _listen_addresses.push_back({addr, *alternator_https_port_proxy_protocol});
-            }
+        if (_config.alternator_https_port()) {
+            alternator_https_port = _config.alternator_https_port();
+            _listen_addresses.push_back({addr, *alternator_https_port});
             creds.emplace();
             auto opts = _config.alternator_encryption_options();
             if (opts.empty()) {
@@ -147,29 +133,20 @@ future<> controller::start_server() {
             }
         }
         _server.invoke_on_all(
-                [this, addr, alternator_port, alternator_https_port, alternator_port_proxy_protocol, alternator_https_port_proxy_protocol, creds = std::move(creds)] (server& server) mutable {
-            return server.init(addr, alternator_port, alternator_https_port, alternator_port_proxy_protocol, alternator_https_port_proxy_protocol, creds,
+                [this, addr, alternator_port, alternator_https_port, creds = std::move(creds)] (server& server) mutable {
+            return server.init(addr, alternator_port, alternator_https_port, creds,
                     _config.alternator_enforce_authorization,
                     _config.alternator_warn_authorization,
                     _config.alternator_max_users_query_size_in_trace_output,
                     &_memory_limiter.local().get_semaphore(),
                     _config.max_concurrent_requests_per_shard);
-        }).handle_exception([this, addr, alternator_port, alternator_https_port, alternator_port_proxy_protocol, alternator_https_port_proxy_protocol] (std::exception_ptr ep) {
-            logger.error("Failed to set up Alternator HTTP server on {} port {}, TLS port {}, proxy-protocol port {}, TLS proxy-protocol port {}: {}",
-                    addr,
-                    alternator_port ? std::to_string(*alternator_port) : "OFF",
-                    alternator_https_port ? std::to_string(*alternator_https_port) : "OFF",
-                    alternator_port_proxy_protocol ? std::to_string(*alternator_port_proxy_protocol) : "OFF",
-                    alternator_https_port_proxy_protocol ? std::to_string(*alternator_https_port_proxy_protocol) : "OFF",
-                    ep);
+        }).handle_exception([this, addr, alternator_port, alternator_https_port] (std::exception_ptr ep) {
+            logger.error("Failed to set up Alternator HTTP server on {} port {}, TLS port {}: {}",
+                    addr, alternator_port ? std::to_string(*alternator_port) : "OFF", alternator_https_port ? std::to_string(*alternator_https_port) : "OFF", ep);
             return stop_server().then([ep = std::move(ep)] { return make_exception_future<>(ep); });
-        }).then([addr, alternator_port, alternator_https_port, alternator_port_proxy_protocol, alternator_https_port_proxy_protocol] {
-            logger.info("Alternator server listening on {}, HTTP port {}, HTTPS port {}, proxy-protocol port {}, TLS proxy-protocol port {}",
-                    addr,
-                    alternator_port ? std::to_string(*alternator_port) : "OFF",
-                    alternator_https_port ? std::to_string(*alternator_https_port) : "OFF",
-                    alternator_port_proxy_protocol ? std::to_string(*alternator_port_proxy_protocol) : "OFF",
-                    alternator_https_port_proxy_protocol ? std::to_string(*alternator_https_port_proxy_protocol) : "OFF");
+        }).then([addr, alternator_port, alternator_https_port] {
+            logger.info("Alternator server listening on {}, HTTP port {}, HTTPS port {}",
+                    addr, alternator_port ? std::to_string(*alternator_port) : "OFF", alternator_https_port ? std::to_string(*alternator_https_port) : "OFF");
         }).get();
     });
 }
@@ -192,7 +169,7 @@ future<> controller::request_stop_server() {
     });
 }
 
-future<utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>>> controller::get_client_data() {
+future<utils::chunked_vector<client_data>> controller::get_client_data() {
     return _server.local().get_client_data();
 }
 

alternator/controller.hh

@@ -15,7 +15,6 @@
 
 namespace service {
 class storage_proxy;
-class storage_service;
 class migration_manager;
 class memory_limiter;
 }
@@ -58,7 +57,6 @@ class server;
 class controller : public protocol_server {
     sharded<gms::gossiper>& _gossiper;
     sharded<service::storage_proxy>& _proxy;
-    sharded<service::storage_service>& _ss;
     sharded<service::migration_manager>& _mm;
     sharded<db::system_distributed_keyspace>& _sys_dist_ks;
     sharded<cdc::generation_service>& _cdc_gen_svc;
@@ -76,7 +74,6 @@ public:
     controller(
         sharded<gms::gossiper>& gossiper,
         sharded<service::storage_proxy>& proxy,
-        sharded<service::storage_service>& ss,
         sharded<service::migration_manager>& mm,
         sharded<db::system_distributed_keyspace>& sys_dist_ks,
         sharded<cdc::generation_service>& cdc_gen_svc,
@@ -96,7 +93,7 @@ public:
     // This virtual function is called (on each shard separately) when the
     // virtual table "system.clients" is read. It is expected to generate a
     // list of clients connected to this server (on this shard).
-    virtual future<utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>>> get_client_data() override;
+    virtual future<utils::chunked_vector<client_data>> get_client_data() override;
 };
 
 }

alternator/executor.cc

@@ -67,14 +67,6 @@ using namespace std::chrono_literals;
 
 logging::logger elogger("alternator-executor");
 
-namespace std {
-    template <> struct hash<std::pair<sstring, sstring>> {
-        size_t operator () (const std::pair<sstring, sstring>& p) const {
-            return std::hash<sstring>()(p.first) * 1009 + std::hash<sstring>()(p.second) * 3;
-        }
-    };
-}
-
 namespace alternator {
 
 // Alternator-specific table properties stored as hidden table tags:
@@ -256,66 +248,14 @@ static const rjson::value::Member& get_single_member(const rjson::value& v, cons
     return *(v.MemberBegin());
 }
 
-class executor::describe_table_info_manager : public service::migration_listener::empty_listener {
-    executor &_executor;
-
-    struct table_info {
-        utils::simple_value_with_expiry<std::uint64_t> size_in_bytes;
-    };
-    std::unordered_map<std::pair<sstring, sstring>, table_info> info_for_tables;
-    bool active = false;
-
-public:
-    describe_table_info_manager(executor& executor) : _executor(executor) {
-        _executor._proxy.data_dictionary().real_database_ptr()->get_notifier().register_listener(this);
-        active = true;
-    }
-    describe_table_info_manager(const describe_table_info_manager &) = delete;
-    describe_table_info_manager(describe_table_info_manager&&) = delete;
-    ~describe_table_info_manager() {
-        if (active) {
-            on_fatal_internal_error(elogger, "describe_table_info_manager was not stopped before destruction");
-        }
-    }
-
-    describe_table_info_manager &operator = (const describe_table_info_manager &) = delete;
-    describe_table_info_manager &operator = (describe_table_info_manager&&) = delete;
-
-    static std::chrono::high_resolution_clock::time_point now() {
-        return std::chrono::high_resolution_clock::now();
-    }
-
-    std::optional<std::uint64_t> get_cached_size_in_bytes(const sstring &ks_name, const sstring &cf_name) const {
-        auto it = info_for_tables.find({ks_name, cf_name});
-        if (it != info_for_tables.end()) {
-            return it->second.size_in_bytes.get();
-        }
-        return std::nullopt;
-    }
-    void cache_size_in_bytes(sstring ks_name, sstring cf_name, std::uint64_t size_in_bytes, std::chrono::high_resolution_clock::time_point expiry) {
-        info_for_tables[{std::move(ks_name), std::move(cf_name)}].size_in_bytes.set_if_longer_expiry(size_in_bytes, expiry);
-    }
-    future<> stop() {
-        co_await _executor._proxy.data_dictionary().real_database_ptr()->get_notifier().unregister_listener(this);
-        active = false;
-        co_return;
-    }
-    void on_drop_column_family(const sstring& ks_name, const sstring& cf_name) override {
-        if (!ks_name.starts_with(executor::KEYSPACE_NAME_PREFIX)) return;
-        info_for_tables.erase({ks_name, cf_name});
-    }
-};
-
 executor::executor(gms::gossiper& gossiper,
          service::storage_proxy& proxy,
-         service::storage_service& ss,
          service::migration_manager& mm,
          db::system_distributed_keyspace& sdks,
          cdc::metadata& cdc_metadata,
          smp_service_group ssg,
          utils::updateable_value<uint32_t> default_timeout_in_ms)
     : _gossiper(gossiper),
-      _ss(ss),
       _proxy(proxy),
       _mm(mm),
       _sdks(sdks),
@@ -328,7 +268,6 @@ executor::executor(gms::gossiper& gossiper,
         _stats))
 {
     s_default_timeout_in_ms = std::move(default_timeout_in_ms);
-    _describe_table_info_manager = std::make_unique<describe_table_info_manager>(*this);
     register_metrics(_metrics, _stats);
 }
 
@@ -480,7 +419,7 @@ static std::optional<std::string> find_table_name(const rjson::value& request) {
     if (!table_name_value->IsString()) {
         throw api_error::validation("Non-string TableName field in request");
     }
-    std::string table_name = rjson::to_string(*table_name_value);
+    std::string table_name = table_name_value->GetString();
     return table_name;
 }
 
@@ -607,7 +546,7 @@ get_table_or_view(service::storage_proxy& proxy, const rjson::value& request) {
             // does exist but the index does not (ValidationException).
             if (proxy.data_dictionary().has_schema(keyspace_name, orig_table_name)) {
                 throw api_error::validation(
-                    fmt::format("Requested resource not found: Index '{}' for table '{}'", rjson::to_string_view(*index_name), orig_table_name));
+                    fmt::format("Requested resource not found: Index '{}' for table '{}'", index_name->GetString(), orig_table_name));
             } else {
                 throw api_error::resource_not_found(
                     fmt::format("Requested resource not found: Table: {} not found", orig_table_name));
@@ -648,7 +587,7 @@ static std::string get_string_attribute(const rjson::value& value, std::string_v
         throw api_error::validation(fmt::format("Expected string value for attribute {}, got: {}",
                 attribute_name, value));
     }
-    return rjson::to_string(*attribute_value);
+    return std::string(attribute_value->GetString(), attribute_value->GetStringLength());
 }
 
 // Convenience function for getting the value of a boolean attribute, or a
@@ -813,44 +752,12 @@ static future<bool> is_view_built(
 
 }
 
-future<> executor::cache_newly_calculated_size_on_all_shards(schema_ptr schema, std::uint64_t size_in_bytes, std::chrono::nanoseconds ttl) {
-    auto expiry = describe_table_info_manager::now() + ttl;
-    return container().invoke_on_all(
-        [schema, size_in_bytes, expiry] (executor& exec) {
-            exec._describe_table_info_manager->cache_size_in_bytes(schema->ks_name(), schema->cf_name(), size_in_bytes, expiry);
-        });
-}
-
-future<> executor::fill_table_size(rjson::value &table_description, schema_ptr schema, bool deleting) {
-    auto cached_size = _describe_table_info_manager->get_cached_size_in_bytes(schema->ks_name(), schema->cf_name());
-    std::uint64_t total_size = 0;
-    if (cached_size) {
-        total_size = *cached_size;
-    } else {
-        // there's no point in trying to estimate value of table that is being deleted, as other nodes more often than not might
-        // move forward with deletion faster than we calculate the size
-        if (!deleting) {
-            total_size = co_await _ss.estimate_total_sstable_volume(schema->id(), service::storage_service::ignore_errors::yes);
-            const auto expiry = std::chrono::seconds{ _proxy.data_dictionary().get_config().alternator_describe_table_info_cache_validity_in_seconds() };
-            // Note: we don't care when the notification of other shards will finish, as long as it will be done
-            // it's possible to get into race condition (next DescribeTable comes to other shard, that new shard doesn't have
-            // the size yet, so it will calculate it again) - this is not a problem, because it will call cache_newly_calculated_size_on_all_shards
-            // with expiry, which is extremely unlikely to be exactly the same as the previous one, all shards will keep the size coming with expiry that is further into the future.
-            // In case of the same expiry, some shards will have different size, which means DescribeTable will return different values depending on the shard
-            // which is also fine, as the specification doesn't give precision guarantees of any kind.
-            co_await cache_newly_calculated_size_on_all_shards(schema, total_size, expiry);
-        }
-    }
-    rjson::add(table_description, "TableSizeBytes", total_size);
-}
-
-future<rjson::value> executor::fill_table_description(schema_ptr schema, table_status tbl_status, service::client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit)
+static future<rjson::value> fill_table_description(schema_ptr schema, table_status tbl_status, service::storage_proxy& proxy, service::client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit)
 {
     rjson::value table_description = rjson::empty_object();
     auto tags_ptr = db::get_tags_of_table(schema);
 
     rjson::add(table_description, "TableName", rjson::from_string(schema->cf_name()));
-    co_await fill_table_size(table_description, schema, tbl_status == table_status::deleting);
 
     auto creation_timestamp = get_table_creation_time(*schema);
 
@@ -894,7 +801,9 @@ future<rjson::value> executor::fill_table_description(schema_ptr schema, table_s
     rjson::add(table_description["ProvisionedThroughput"], "WriteCapacityUnits", wcu);
     rjson::add(table_description["ProvisionedThroughput"], "NumberOfDecreasesToday", 0);
 
-    data_dictionary::table t = _proxy.data_dictionary().find_column_family(schema);
+
+
+    data_dictionary::table t = proxy.data_dictionary().find_column_family(schema);
 
     if (tbl_status != table_status::deleting) {
         rjson::add(table_description, "CreationDateTime", rjson::value(creation_timestamp));
@@ -931,7 +840,7 @@ future<rjson::value> executor::fill_table_description(schema_ptr schema, table_s
                 // (for a built view) or CREATING+Backfilling (if view building
                 // is in progress).
                 if (!is_lsi) {
-                    if (co_await is_view_built(vptr, _proxy, client_state, trace_state, permit)) {
+                    if (co_await is_view_built(vptr, proxy, client_state, trace_state, permit)) {
                         rjson::add(view_entry, "IndexStatus", "ACTIVE");
                     } else {
                         rjson::add(view_entry, "IndexStatus", "CREATING");
@@ -959,8 +868,9 @@ future<rjson::value> executor::fill_table_description(schema_ptr schema, table_s
         }
         rjson::add(table_description, "AttributeDefinitions", std::move(attribute_definitions));
     }
-    executor::supplement_table_stream_info(table_description, *schema, _proxy);
+    executor::supplement_table_stream_info(table_description, *schema, proxy);
 
+    // FIXME: still missing some response fields (issue #5026)
     co_return table_description;
 }
 
@@ -980,7 +890,7 @@ future<executor::request_return_type> executor::describe_table(client_state& cli
     get_stats_from_schema(_proxy, *schema)->api_operations.describe_table++;
     tracing::add_alternator_table_name(trace_state, schema->cf_name());
 
-    rjson::value table_description = co_await fill_table_description(schema, table_status::active, client_state, trace_state, permit);
+    rjson::value table_description = co_await fill_table_description(schema, table_status::active, _proxy, client_state, trace_state, permit);
     rjson::value response = rjson::empty_object();
     rjson::add(response, "Table", std::move(table_description));
     elogger.trace("returning {}", response);
@@ -1083,7 +993,7 @@ future<executor::request_return_type> executor::delete_table(client_state& clien
     auto& p = _proxy.container();
 
     schema_ptr schema = get_table(_proxy, request);
-    rjson::value table_description = co_await fill_table_description(schema, table_status::deleting, client_state, trace_state, permit);
+    rjson::value table_description = co_await fill_table_description(schema, table_status::deleting, _proxy, client_state, trace_state, permit);
     co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, schema, auth::permission::DROP, _stats);
     co_await _mm.container().invoke_on(0, [&, cs = client_state.move_to_other_shard()] (service::migration_manager& mm) -> future<> {
         size_t retries = mm.get_concurrent_ddl_retries();
@@ -1170,8 +1080,8 @@ static void add_column(schema_builder& builder, const std::string& name, const r
     }
     for (auto it = attribute_definitions.Begin(); it != attribute_definitions.End(); ++it) {
         const rjson::value& attribute_info = *it;
-        if (rjson::to_string_view(attribute_info["AttributeName"]) == name) {
-            std::string_view type = rjson::to_string_view(attribute_info["AttributeType"]);
+        if (attribute_info["AttributeName"].GetString() == name) {
+            auto type = attribute_info["AttributeType"].GetString();
             data_type dt = parse_key_type(type);
             if (computed_column) {
                 // Computed column for GSI (doesn't choose a real column as-is
@@ -1206,7 +1116,7 @@ static std::pair<std::string, std::string> parse_key_schema(const rjson::value&
         throw api_error::validation("First element of KeySchema must be an object");
     }
     const rjson::value *v = rjson::find((*key_schema)[0], "KeyType");
-    if (!v || !v->IsString() || rjson::to_string_view(*v) != "HASH") {
+    if (!v || !v->IsString() || v->GetString() != std::string("HASH")) {
         throw api_error::validation("First key in KeySchema must be a HASH key");
     }
     v = rjson::find((*key_schema)[0], "AttributeName");
@@ -1214,14 +1124,14 @@ static std::pair<std::string, std::string> parse_key_schema(const rjson::value&
         throw api_error::validation("First key in KeySchema must have string AttributeName");
     }
     validate_attr_name_length(supplementary_context, v->GetStringLength(), true, "HASH key in KeySchema - ");
-    std::string hash_key = rjson::to_string(*v);
+    std::string hash_key = v->GetString();
     std::string range_key;
     if (key_schema->Size() == 2) {
         if (!(*key_schema)[1].IsObject()) {
             throw api_error::validation("Second element of KeySchema must be an object");
         }
         v = rjson::find((*key_schema)[1], "KeyType");
-        if (!v || !v->IsString() || rjson::to_string_view(*v) != "RANGE") {
+        if (!v || !v->IsString() || v->GetString() != std::string("RANGE")) {
             throw api_error::validation("Second key in KeySchema must be a RANGE key");
         }
         v = rjson::find((*key_schema)[1], "AttributeName");
@@ -1647,7 +1557,8 @@ static future<> mark_view_schemas_as_built(utils::chunked_vector<mutation>& out,
     }
 }
 
-future<executor::request_return_type> executor::create_table_on_shard0(service::client_state&& client_state, tracing::trace_state_ptr trace_state, rjson::value request, bool enforce_authorization, bool warn_authorization, const db::tablets_mode_t::mode tablets_mode) {
+static future<executor::request_return_type> create_table_on_shard0(service::client_state&& client_state, tracing::trace_state_ptr trace_state, rjson::value request,
+            service::storage_proxy& sp, service::migration_manager& mm, gms::gossiper& gossiper, bool enforce_authorization, bool warn_authorization, stats& stats, const db::tablets_mode_t::mode tablets_mode) {
     SCYLLA_ASSERT(this_shard_id() == 0);
 
     // We begin by parsing and validating the content of the CreateTable
@@ -1834,7 +1745,7 @@ future<executor::request_return_type> executor::create_table_on_shard0(service::
 
     rjson::value* stream_specification = rjson::find(request, "StreamSpecification");
     if (stream_specification && stream_specification->IsObject()) {
-        if (executor::add_stream_options(*stream_specification, builder, _proxy)) {
+        if (executor::add_stream_options(*stream_specification, builder, sp)) {
             validate_cdc_log_name_length(builder.cf_name());
         }
     }
@@ -1853,7 +1764,7 @@ future<executor::request_return_type> executor::create_table_on_shard0(service::
     set_table_creation_time(tags_map, db_clock::now());
     builder.add_extension(db::tags_extension::NAME, ::make_shared<db::tags_extension>(tags_map));
 
-    co_await verify_create_permission(enforce_authorization, warn_authorization, client_state, _stats);
+    co_await verify_create_permission(enforce_authorization, warn_authorization, client_state, stats);
 
     schema_ptr schema = builder.build();
     for (auto& view_builder : view_builders) {
@@ -1869,18 +1780,18 @@ future<executor::request_return_type> executor::create_table_on_shard0(service::
         view_builder.with_view_info(schema, include_all_columns, ""/*where clause*/);
     }
 
-    size_t retries = _mm.get_concurrent_ddl_retries();
+    size_t retries = mm.get_concurrent_ddl_retries();
     for (;;) {
-        auto group0_guard = co_await _mm.start_group0_operation();
+        auto group0_guard = co_await mm.start_group0_operation();
         auto ts = group0_guard.write_timestamp();
         utils::chunked_vector<mutation> schema_mutations;
-        auto ksm = create_keyspace_metadata(keyspace_name, _proxy, _gossiper, ts, tags_map, _proxy.features(), tablets_mode);
+        auto ksm = create_keyspace_metadata(keyspace_name, sp, gossiper, ts, tags_map, sp.features(), tablets_mode);
         // Alternator Streams doesn't yet work when the table uses tablets (#23838)
         if (stream_specification && stream_specification->IsObject()) {
             auto stream_enabled = rjson::find(*stream_specification, "StreamEnabled");
             if (stream_enabled && stream_enabled->IsBool() && stream_enabled->GetBool()) {
                 locator::replication_strategy_params params(ksm->strategy_options(), ksm->initial_tablets(), ksm->consistency_option());
-                const auto& topo = _proxy.local_db().get_token_metadata().get_topology();
+                const auto& topo = sp.local_db().get_token_metadata().get_topology();
                 auto rs = locator::abstract_replication_strategy::create_replication_strategy(ksm->strategy_name(), params, topo);
                 if (rs->uses_tablets()) {
                     co_return api_error::validation("Streams not yet supported on a table using tablets (issue #23838). "
@@ -1888,19 +1799,14 @@ future<executor::request_return_type> executor::create_table_on_shard0(service::
                 }
             }
         }
-        // Creating an index in tablets mode requires the rf_rack_valid_keyspaces option to be enabled.
-        // GSI and LSI indexes are based on materialized views which require this option to avoid consistency issues.
-        if (!view_builders.empty() && ksm->uses_tablets() && !_proxy.data_dictionary().get_config().rf_rack_valid_keyspaces()) {
-            co_return api_error::validation("GlobalSecondaryIndexes and LocalSecondaryIndexes with tablets require the rf_rack_valid_keyspaces option to be enabled.");
-        }
         try {
-            schema_mutations = service::prepare_new_keyspace_announcement(_proxy.local_db(), ksm, ts);
+            schema_mutations = service::prepare_new_keyspace_announcement(sp.local_db(), ksm, ts);
         } catch (exceptions::already_exists_exception&) {
-            if (_proxy.data_dictionary().has_schema(keyspace_name, table_name)) {
+            if (sp.data_dictionary().has_schema(keyspace_name, table_name)) {
                 co_return api_error::resource_in_use(fmt::format("Table {} already exists", table_name));
             }
         }
-        if (_proxy.data_dictionary().try_find_table(schema->id())) {
+        if (sp.data_dictionary().try_find_table(schema->id())) {
             // This should never happen, the ID is supposed to be unique
             co_return api_error::internal(format("Table with ID {} already exists", schema->id()));
         }
@@ -1909,9 +1815,9 @@ future<executor::request_return_type> executor::create_table_on_shard0(service::
         for (schema_builder& view_builder : view_builders) {
             schemas.push_back(view_builder.build());
         }
-        co_await service::prepare_new_column_families_announcement(schema_mutations, _proxy, *ksm, schemas, ts);
+        co_await service::prepare_new_column_families_announcement(schema_mutations, sp, *ksm, schemas, ts);
         if (ksm->uses_tablets()) {
-            co_await mark_view_schemas_as_built(schema_mutations, schemas, ts, _proxy);
+            co_await mark_view_schemas_as_built(schema_mutations, schemas, ts, sp);
         }
 
         // If a role is allowed to create a table, we must give it permissions to
@@ -1936,7 +1842,7 @@ future<executor::request_return_type> executor::create_table_on_shard0(service::
         }
         std::tie(schema_mutations, group0_guard) = co_await std::move(mc).extract();
         try {
-            co_await _mm.announce(std::move(schema_mutations), std::move(group0_guard), fmt::format("alternator-executor: create {} table", table_name));
+            co_await mm.announce(std::move(schema_mutations), std::move(group0_guard), fmt::format("alternator-executor: create {} table", table_name));
             break;
         }  catch (const service::group0_concurrent_modification& ex) {
             elogger.info("Failed to execute CreateTable {} due to concurrent schema modifications. {}.",
@@ -1948,9 +1854,9 @@ future<executor::request_return_type> executor::create_table_on_shard0(service::
         }
     }
 
-    co_await _mm.wait_for_schema_agreement(_proxy.local_db(), db::timeout_clock::now() + 10s, nullptr);
+    co_await mm.wait_for_schema_agreement(sp.local_db(), db::timeout_clock::now() + 10s, nullptr);
     rjson::value status = rjson::empty_object();
-    executor::supplement_table_info(request, *schema, _proxy);
+    executor::supplement_table_info(request, *schema, sp);
     rjson::add(status, "TableDescription", std::move(request));
     co_return rjson::print(std::move(status));
 }
@@ -1959,11 +1865,10 @@ future<executor::request_return_type> executor::create_table(client_state& clien
     _stats.api_operations.create_table++;
     elogger.trace("Creating table {}", request);
 
-    co_return co_await _mm.container().invoke_on(0, [&, tr = tracing::global_trace_state_ptr(trace_state), request = std::move(request), &e = this->container(), client_state_other_shard = client_state.move_to_other_shard(), enforce_authorization = bool(_enforce_authorization), warn_authorization = bool(_warn_authorization)]
+    co_return co_await _mm.container().invoke_on(0, [&, tr = tracing::global_trace_state_ptr(trace_state), request = std::move(request), &sp = _proxy.container(), &g = _gossiper.container(), &e = this->container(), client_state_other_shard = client_state.move_to_other_shard(), enforce_authorization = bool(_enforce_authorization), warn_authorization = bool(_warn_authorization)]
                                         (service::migration_manager& mm) mutable -> future<executor::request_return_type> {
         const db::tablets_mode_t::mode tablets_mode = _proxy.data_dictionary().get_config().tablets_mode_for_new_keyspaces(); // type cast
-        // `invoke_on` hopped us to shard 0, but `this` points to `executor` is from 'old' shard, we need to hop it too.
-        co_return co_await e.local().create_table_on_shard0(client_state_other_shard.get(), tr, std::move(request), enforce_authorization, warn_authorization, std::move(tablets_mode));
+        co_return co_await create_table_on_shard0(client_state_other_shard.get(), tr, std::move(request), sp.local(), mm, g.local(), enforce_authorization, warn_authorization, e.local()._stats, std::move(tablets_mode));
     });
 }
 
@@ -1982,8 +1887,8 @@ future<executor::request_return_type> executor::create_table(client_state& clien
         std::string def_type = type_to_string(def.type);
         for (auto it = attribute_definitions.Begin(); it != attribute_definitions.End(); ++it) {
             const rjson::value& attribute_info = *it;
-            if (rjson::to_string_view(attribute_info["AttributeName"]) == def.name_as_text()) {
-                std::string_view type = rjson::to_string_view(attribute_info["AttributeType"]);
+            if (attribute_info["AttributeName"].GetString() == def.name_as_text()) {
+                auto type = attribute_info["AttributeType"].GetString();
                 if (type != def_type) {
                     throw api_error::validation(fmt::format("AttributeDefinitions redefined {} to {} already a key attribute of type {} in this table", def.name_as_text(), type, def_type));
                 }
@@ -2114,10 +2019,6 @@ future<executor::request_return_type> executor::update_table(client_state& clien
                             co_return api_error::validation(fmt::format(
                                 "LSI {} already exists in table {}, can't use same name for GSI", index_name, table_name));
                         }
-                        if (p.local().local_db().find_keyspace(keyspace_name).get_replication_strategy().uses_tablets() &&
-                                !p.local().data_dictionary().get_config().rf_rack_valid_keyspaces()) {
-                            co_return api_error::validation("GlobalSecondaryIndexes with tablets require the rf_rack_valid_keyspaces option to be enabled.");
-                        }
 
                         elogger.trace("Adding GSI {}", index_name);
                         // FIXME: read and handle "Projection" parameter. This will
@@ -2322,12 +2223,12 @@ void validate_value(const rjson::value& v, const char* caller) {
 
 // The put_or_delete_item class builds the mutations needed by the PutItem and
 // DeleteItem operations - either as stand-alone commands or part of a list
-// of commands in BatchWriteItem.
+// of commands in BatchWriteItems.
 // put_or_delete_item splits each operation into two stages: Constructing the
 // object parses and validates the user input (throwing exceptions if there
 // are input errors). Later, build() generates the actual mutation, with a
 // specified timestamp. This split is needed because of the peculiar needs of
-// BatchWriteItem and LWT. BatchWriteItem needs all parsing to happen before
+// BatchWriteItems and LWT. BatchWriteItems needs all parsing to happen before
 // any writing happens (if one of the commands has an error, none of the
 // writes should be done). LWT makes it impossible for the parse step to
 // generate "mutation" objects, because the timestamp still isn't known.
@@ -2461,7 +2362,7 @@ put_or_delete_item::put_or_delete_item(const rjson::value& item, schema_ptr sche
     _cells = std::vector<cell>();
     _cells->reserve(item.MemberCount());
     for (auto it = item.MemberBegin(); it != item.MemberEnd(); ++it) {
-        bytes column_name = to_bytes(rjson::to_string_view(it->name));
+        bytes column_name = to_bytes(it->name.GetString());
         validate_value(it->value, "PutItem");
         const column_definition* cdef = find_attribute(*schema, column_name);
         validate_attr_name_length("", column_name.size(), cdef && cdef->is_primary_key());
@@ -2838,7 +2739,7 @@ future<executor::request_return_type> rmw_operation::execute(service::storage_pr
     auto read_command = needs_read_before_write ?
             previous_item_read_command(proxy, schema(), _ck, selection) :
             nullptr;
-    return proxy.cas(schema(), std::move(*cas_shard), *this, read_command, to_partition_ranges(*schema(), _pk),
+    return proxy.cas(schema(), std::move(*cas_shard), shared_from_this(), read_command, to_partition_ranges(*schema(), _pk),
             {timeout, std::move(permit), client_state, trace_state},
             db::consistency_level::LOCAL_SERIAL, db::consistency_level::LOCAL_QUORUM, timeout, timeout, true, std::move(cdc_opts)).then([this, read_command, &wcu_total] (bool is_applied) mutable {
         if (!is_applied) {
@@ -2882,10 +2783,10 @@ static void verify_all_are_used(const rjson::value* field,
         return;
     }
     for (auto it = field->MemberBegin(); it != field->MemberEnd(); ++it) {
-        if (!used.contains(rjson::to_string(it->name))) {
+        if (!used.contains(it->name.GetString())) {
             throw api_error::validation(
                 format("{} has spurious '{}', not used in {}",
-                    field_name, rjson::to_string_view(it->name), operation));
+                    field_name, it->name.GetString(), operation));
         }
     }
 }
@@ -3099,7 +3000,7 @@ future<executor::request_return_type> executor::delete_item(client_state& client
 }
 
 static schema_ptr get_table_from_batch_request(const service::storage_proxy& proxy, const rjson::value::ConstMemberIterator& batch_request) {
-    sstring table_name = rjson::to_sstring(batch_request->name); // JSON keys are always strings
+    sstring table_name = batch_request->name.GetString(); // JSON keys are always strings
     try {
         return proxy.data_dictionary().find_schema(sstring(executor::KEYSPACE_NAME_PREFIX) + table_name, table_name);
     } catch(data_dictionary::no_such_column_family&) {
@@ -3125,20 +3026,17 @@ struct primary_key_equal {
 };
 
 // This is a cas_request subclass for applying given put_or_delete_items to
-// one partition using LWT as part as BatchWriteItem. This is a write-only
+// one partition using LWT as part as BatchWriteItems. This is a write-only
 // operation, not needing the previous value of the item (the mutation to be
 // done is known prior to starting the operation). Nevertheless, we want to
 // do this mutation via LWT to ensure that it is serialized with other LWT
 // mutations to the same partition.
-// 
-// The std::vector<put_or_delete_item> must remain alive until the
-// storage_proxy::cas() future is resolved.
 class put_or_delete_item_cas_request : public service::cas_request {
     schema_ptr schema;
-    const std::vector<put_or_delete_item>& _mutation_builders;
+    std::vector<put_or_delete_item> _mutation_builders;
 public:
-    put_or_delete_item_cas_request(schema_ptr s, const std::vector<put_or_delete_item>& b) :
-        schema(std::move(s)), _mutation_builders(b) { }
+    put_or_delete_item_cas_request(schema_ptr s, std::vector<put_or_delete_item>&& b) :
+        schema(std::move(s)), _mutation_builders(std::move(b)) { }
     virtual ~put_or_delete_item_cas_request() = default;
     virtual std::optional<mutation> apply(foreign_ptr<lw_shared_ptr<query::result>> qr, const query::partition_slice& slice, api::timestamp_type ts, cdc::per_request_options& cdc_opts) override {
         std::optional<mutation> ret;
@@ -3154,48 +3052,20 @@ public:
     }
 };
 
-future<> executor::cas_write(schema_ptr schema, service::cas_shard cas_shard, const dht::decorated_key& dk,
-        const std::vector<put_or_delete_item>& mutation_builders, service::client_state& client_state,
-        tracing::trace_state_ptr trace_state, service_permit permit)
-{
-    if (!cas_shard.this_shard()) {
-        _stats.shard_bounce_for_lwt++;
-        return container().invoke_on(cas_shard.shard(), _ssg,
-                    [cs = client_state.move_to_other_shard(),
-                    &mb = mutation_builders,
-                    &dk,
-                    ks = schema->ks_name(),
-                    cf = schema->cf_name(),
-                    gt = tracing::global_trace_state_ptr(trace_state),
-                    permit = std::move(permit)]
-                    (executor& self) mutable {
-            return do_with(cs.get(), [&mb, &dk, ks = std::move(ks), cf = std::move(cf),
-                                    trace_state = tracing::trace_state_ptr(gt), &self]
-                                    (service::client_state& client_state) mutable {
-                auto schema = self._proxy.data_dictionary().find_schema(ks, cf);
-                service::cas_shard cas_shard(*schema, dk.token());
-
-                //FIXME: Instead of passing empty_service_permit() to the background operation,
-                // the current permit's lifetime should be prolonged, so that it's destructed
-                // only after all background operations are finished as well.
-                return self.cas_write(schema, std::move(cas_shard), dk, mb, client_state, std::move(trace_state), empty_service_permit());
-            });
-        });
-    }
-
+static future<> cas_write(service::storage_proxy& proxy, schema_ptr schema, service::cas_shard cas_shard, dht::decorated_key dk, std::vector<put_or_delete_item>&& mutation_builders,
+        service::client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit) {
     auto timeout = executor::default_timeout();
-    auto op = std::make_unique<put_or_delete_item_cas_request>(schema, mutation_builders);
-    auto* op_ptr = op.get();
+    auto op = seastar::make_shared<put_or_delete_item_cas_request>(schema, std::move(mutation_builders));
     auto cdc_opts = cdc::per_request_options{
         .alternator = true,
         .alternator_streams_increased_compatibility =
-                schema->cdc_options().enabled() && _proxy.data_dictionary().get_config().alternator_streams_increased_compatibility(),
+                schema->cdc_options().enabled() && proxy.data_dictionary().get_config().alternator_streams_increased_compatibility(),
     };
-    return _proxy.cas(schema, std::move(cas_shard), *op_ptr, nullptr, to_partition_ranges(dk),
+    return proxy.cas(schema, std::move(cas_shard), op, nullptr, to_partition_ranges(dk),
             {timeout, std::move(permit), client_state, trace_state},
             db::consistency_level::LOCAL_SERIAL, db::consistency_level::LOCAL_QUORUM,
-            timeout, timeout, true, std::move(cdc_opts)).finally([op = std::move(op)]{}).discard_result();
-    // We discarded cas()'s future value ("is_applied") because BatchWriteItem
+            timeout, timeout, true, std::move(cdc_opts)).discard_result();
+    // We discarded cas()'s future value ("is_applied") because BatchWriteItems
     // does not need to support conditional updates.
 }
 
@@ -3217,11 +3087,13 @@ struct schema_decorated_key_equal {
 
 // FIXME: if we failed writing some of the mutations, need to return a list
 // of these failed mutations rather than fail the whole write (issue #5650).
-future<> executor::do_batch_write(
+static future<> do_batch_write(service::storage_proxy& proxy,
+        smp_service_group ssg,
         std::vector<std::pair<schema_ptr, put_or_delete_item>> mutation_builders,
         service::client_state& client_state,
         tracing::trace_state_ptr trace_state,
-        service_permit permit) {
+        service_permit permit,
+        stats& stats) {
     if (mutation_builders.empty()) {
         return make_ready_future<>();
     }
@@ -3243,7 +3115,7 @@ future<> executor::do_batch_write(
             mutations.push_back(b.second.build(b.first, now));
             any_cdc_enabled |= b.first->cdc_options().enabled();
         }
-        return _proxy.mutate(std::move(mutations),
+        return proxy.mutate(std::move(mutations),
                 db::consistency_level::LOCAL_QUORUM,
                 executor::default_timeout(),
                 trace_state,
@@ -3252,48 +3124,55 @@ future<> executor::do_batch_write(
                 false,
                 cdc::per_request_options{
                     .alternator = true,
-                    .alternator_streams_increased_compatibility = any_cdc_enabled && _proxy.data_dictionary().get_config().alternator_streams_increased_compatibility(),
+                    .alternator_streams_increased_compatibility = any_cdc_enabled && proxy.data_dictionary().get_config().alternator_streams_increased_compatibility(),
                 });
     } else {
         // Do the write via LWT:
         // Multiple mutations may be destined for the same partition, adding
         // or deleting different items of one partition. Join them together
         // because we can do them in one cas() call.
-        using map_type = std::unordered_map<schema_decorated_key, 
-            std::vector<put_or_delete_item>, 
-            schema_decorated_key_hash, 
-            schema_decorated_key_equal>;
-        auto key_builders = std::make_unique<map_type>(1, schema_decorated_key_hash{}, schema_decorated_key_equal{});
-        for (auto&& b : std::move(mutation_builders)) {
-            auto [it, added] = key_builders->try_emplace(schema_decorated_key {
-                .schema = b.first,
-                .dk = dht::decorate_key(*b.first, b.second.pk())
-            });
+        std::unordered_map<schema_decorated_key, std::vector<put_or_delete_item>, schema_decorated_key_hash, schema_decorated_key_equal>
+            key_builders(1, schema_decorated_key_hash{}, schema_decorated_key_equal{});
+        for (auto& b : mutation_builders) {
+            auto dk = dht::decorate_key(*b.first, b.second.pk());
+            auto [it, added] = key_builders.try_emplace(schema_decorated_key{b.first, dk});
             it->second.push_back(std::move(b.second));
         }
-        auto* key_builders_ptr = key_builders.get();
-        return parallel_for_each(*key_builders_ptr, [this, &client_state, trace_state, permit = std::move(permit)] (const auto& e) {
-            _stats.write_using_lwt++;
+        return parallel_for_each(std::move(key_builders), [&proxy, &client_state, &stats, trace_state, ssg, permit = std::move(permit)] (auto& e) {
+            stats.write_using_lwt++;
             auto desired_shard = service::cas_shard(*e.first.schema, e.first.dk.token());
-            auto s = e.first.schema;
+            if (desired_shard.this_shard()) {
+                return cas_write(proxy, e.first.schema, std::move(desired_shard), e.first.dk, std::move(e.second), client_state, trace_state, permit);
+            } else {
+                stats.shard_bounce_for_lwt++;
+                return proxy.container().invoke_on(desired_shard.shard(), ssg,
+                            [cs = client_state.move_to_other_shard(),
+                             mb = e.second,
+                             dk = e.first.dk,
+                             ks = e.first.schema->ks_name(),
+                             cf = e.first.schema->cf_name(),
+                             gt =  tracing::global_trace_state_ptr(trace_state),
+                             permit = std::move(permit)]
+                            (service::storage_proxy& proxy) mutable {
+                    return do_with(cs.get(), [&proxy, mb = std::move(mb), dk = std::move(dk), ks = std::move(ks), cf = std::move(cf),
+                                              trace_state = tracing::trace_state_ptr(gt)]
+                                              (service::client_state& client_state) mutable {
+                        auto schema = proxy.data_dictionary().find_schema(ks, cf);
 
-            static const auto* injection_name = "alternator_executor_batch_write_wait";
-            return utils::get_local_injector().inject(injection_name, [s = std::move(s)] (auto& handler) -> future<> {
-                const auto ks = handler.get("keyspace");
-                const auto cf = handler.get("table");
-                const auto shard = std::atoll(handler.get("shard")->data());
-                if (ks == s->ks_name() && cf == s->cf_name() && shard == this_shard_id()) {
-                    elogger.info("{}: hit", injection_name);
-                    co_await handler.wait_for_message(std::chrono::steady_clock::now() + std::chrono::minutes{5});
-                    elogger.info("{}: continue", injection_name);
-                }
-            }).then([&e, desired_shard = std::move(desired_shard),
-                 &client_state, trace_state = std::move(trace_state), permit = std::move(permit), this]() mutable
-            {
-                return cas_write(e.first.schema, std::move(desired_shard), e.first.dk,
-                    std::move(e.second), client_state, std::move(trace_state), std::move(permit));
-            });
-        }).finally([key_builders = std::move(key_builders)]{});
+                        // The desired_shard on the original shard remains alive for the duration
+                        // of cas_write on this shard and prevents any tablet operations.
+                        // However, we need a local instance of cas_shard on this shard
+                        // to pass it to sp::cas, so we just create a new one.
+                        service::cas_shard cas_shard(*schema, dk.token());
+
+                        //FIXME: Instead of passing empty_service_permit() to the background operation,
+                        // the current permit's lifetime should be prolonged, so that it's destructed
+                        // only after all background operations are finished as well.
+                        return cas_write(proxy, schema, std::move(cas_shard), dk, std::move(mb), client_state, std::move(trace_state), empty_service_permit());
+                    });
+                }).finally([desired_shard = std::move(desired_shard)]{});
+            }
+        });
     }
 }
 
@@ -3440,7 +3319,7 @@ future<executor::request_return_type> executor::batch_write_item(client_state& c
     _stats.wcu_total[stats::DELETE_ITEM] += wcu_delete_units;
     _stats.api_operations.batch_write_item_batch_total += total_items;
     _stats.api_operations.batch_write_item_histogram.add(total_items);
-    co_await do_batch_write(std::move(mutation_builders), client_state, trace_state, std::move(permit));
+    co_await do_batch_write(_proxy, _ssg, std::move(mutation_builders), client_state, trace_state, std::move(permit), _stats);
     // FIXME: Issue #5650: If we failed writing some of the updates,
     // need to return a list of these failed updates in UnprocessedItems
     // rather than fail the whole write (issue #5650).
@@ -3485,7 +3364,7 @@ static bool hierarchy_filter(rjson::value& val, const attribute_path_map_node<T>
         }
         rjson::value newv = rjson::empty_object();
         for (auto it = v.MemberBegin(); it != v.MemberEnd(); ++it) {
-            std::string attr = rjson::to_string(it->name);
+            std::string attr = it->name.GetString();
             auto x = members.find(attr);
             if (x != members.end()) {
                 if (x->second) {
@@ -3705,7 +3584,7 @@ static std::optional<attrs_to_get> calculate_attrs_to_get(const rjson::value& re
         const rjson::value& attributes_to_get = req["AttributesToGet"];
         attrs_to_get ret;
         for (auto it = attributes_to_get.Begin(); it != attributes_to_get.End(); ++it) {
-            attribute_path_map_add("AttributesToGet", ret, rjson::to_string(*it));
+            attribute_path_map_add("AttributesToGet", ret, it->GetString());
             validate_attr_name_length("AttributesToGet", it->GetStringLength(), false);
         }
         if (ret.empty()) {
@@ -4371,12 +4250,12 @@ inline void update_item_operation::apply_attribute_updates(const std::unique_ptr
         attribute_collector& modified_attrs, bool& any_updates, bool& any_deletes) const {
     for (auto it = _attribute_updates->MemberBegin(); it != _attribute_updates->MemberEnd(); ++it) {
         // Note that it.key() is the name of the column, *it is the operation
-        bytes column_name = to_bytes(rjson::to_string_view(it->name));
+        bytes column_name = to_bytes(it->name.GetString());
         const column_definition* cdef = _schema->get_column_definition(column_name);
         if (cdef && cdef->is_primary_key()) {
-            throw api_error::validation(format("UpdateItem cannot update key column {}", rjson::to_string_view(it->name)));
+            throw api_error::validation(format("UpdateItem cannot update key column {}", it->name.GetString()));
         }
-        std::string action = rjson::to_string((it->value)["Action"]);
+        std::string action = (it->value)["Action"].GetString();
         if (action == "DELETE") {
             // The DELETE operation can do two unrelated tasks. Without a
             // "Value" option, it is used to delete an attribute. With a
@@ -5573,7 +5452,7 @@ calculate_bounds_conditions(schema_ptr schema, const rjson::value& conditions) {
     std::vector<query::clustering_range> ck_bounds;
 
     for (auto it = conditions.MemberBegin(); it != conditions.MemberEnd(); ++it) {
-        sstring key = rjson::to_sstring(it->name);
+        std::string key = it->name.GetString();
         const rjson::value& condition = it->value;
 
         const rjson::value& comp_definition = rjson::get(condition, "ComparisonOperator");
@@ -5581,13 +5460,13 @@ calculate_bounds_conditions(schema_ptr schema, const rjson::value& conditions) {
 
         const column_definition& pk_cdef = schema->partition_key_columns().front();
         const column_definition* ck_cdef = schema->clustering_key_size() > 0 ? &schema->clustering_key_columns().front() : nullptr;
-        if (key == pk_cdef.name_as_text()) {
+        if (sstring(key) == pk_cdef.name_as_text()) {
             if (!partition_ranges.empty()) {
                 throw api_error::validation("Currently only a single restriction per key is allowed");
             }
             partition_ranges.push_back(calculate_pk_bound(schema, pk_cdef, comp_definition, attr_list));
         }
-        if (ck_cdef && key == ck_cdef->name_as_text()) {
+        if (ck_cdef && sstring(key) == ck_cdef->name_as_text()) {
             if (!ck_bounds.empty()) {
                 throw api_error::validation("Currently only a single restriction per key is allowed");
             }
@@ -5986,14 +5865,9 @@ future<executor::request_return_type> executor::list_tables(client_state& client
     _stats.api_operations.list_tables++;
     elogger.trace("Listing tables {}", request);
 
-    co_await utils::get_local_injector().inject("alternator_list_tables", [] (auto& handler) -> future<> {
-        handler.set("waiting", true);
-        co_await handler.wait_for_message(std::chrono::steady_clock::now() + std::chrono::minutes{5});
-    });
-
     rjson::value* exclusive_start_json = rjson::find(request, "ExclusiveStartTableName");
     rjson::value* limit_json = rjson::find(request, "Limit");
-    std::string exclusive_start = exclusive_start_json ? rjson::to_string(*exclusive_start_json) : "";
+    std::string exclusive_start = exclusive_start_json ? exclusive_start_json->GetString() : "";
     int limit = limit_json ? limit_json->GetInt() : 100;
     if (limit < 1 || limit > 100) {
         co_return api_error::validation("Limit must be greater than 0 and no greater than 100");
@@ -6182,10 +6056,9 @@ future<> executor::start() {
 }
 
 future<> executor::stop() {
-    co_await _describe_table_info_manager->stop();
     // disconnect from the value source, but keep the value unchanged.
     s_default_timeout_in_ms = utils::updateable_value<uint32_t>{s_default_timeout_in_ms()};
-    co_await _parsed_expression_cache->stop();
+    return _parsed_expression_cache->stop();
 }
 
 } // namespace alternator

alternator/executor.hh

@@ -17,13 +17,11 @@
 #include "service/client_state.hh"
 #include "service_permit.hh"
 #include "db/timeout_clock.hh"
-#include "db/config.hh"
 
 #include "alternator/error.hh"
 #include "stats.hh"
 #include "utils/rjson.hh"
 #include "utils/updateable_value.hh"
-#include "utils/simple_value_with_expiry.hh"
 
 #include "tracing/trace_state.hh"
 
@@ -42,8 +40,6 @@ namespace cql3::selection {
 
 namespace service {
     class storage_proxy;
-    class cas_shard;
-    class storage_service;
 }
 
 namespace cdc {
@@ -60,9 +56,7 @@ class schema_builder;
 
 namespace alternator {
 
-enum class table_status;
 class rmw_operation;
-class put_or_delete_item;
 
 schema_ptr get_table(service::storage_proxy& proxy, const rjson::value& request);
 bool is_alternator_keyspace(const sstring& ks_name);
@@ -140,7 +134,6 @@ class expression_cache;
 
 class executor : public peering_sharded_service<executor> {
     gms::gossiper& _gossiper;
-    service::storage_service& _ss;
     service::storage_proxy& _proxy;
     service::migration_manager& _mm;
     db::system_distributed_keyspace& _sdks;
@@ -153,11 +146,6 @@ class executor : public peering_sharded_service<executor> {
 
     std::unique_ptr<parsed::expression_cache> _parsed_expression_cache;
 
-    struct describe_table_info_manager;
-    std::unique_ptr<describe_table_info_manager> _describe_table_info_manager;
-
-    future<> cache_newly_calculated_size_on_all_shards(schema_ptr schema, std::uint64_t size_in_bytes, std::chrono::nanoseconds ttl);
-    future<> fill_table_size(rjson::value &table_description, schema_ptr schema, bool deleting);
 public:
     using client_state = service::client_state;
     // request_return_type is the return type of the executor methods, which
@@ -183,7 +171,6 @@ public:
 
     executor(gms::gossiper& gossiper,
              service::storage_proxy& proxy,
-             service::storage_service& ss,
              service::migration_manager& mm,
              db::system_distributed_keyspace& sdks,
              cdc::metadata& cdc_metadata,
@@ -231,18 +218,6 @@ private:
     friend class rmw_operation;
 
     static void describe_key_schema(rjson::value& parent, const schema&, std::unordered_map<std::string,std::string> * = nullptr, const std::map<sstring, sstring> *tags = nullptr);
-    future<rjson::value> fill_table_description(schema_ptr schema, table_status tbl_status, service::client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit);
-    future<executor::request_return_type> create_table_on_shard0(service::client_state&& client_state, tracing::trace_state_ptr trace_state, rjson::value request, bool enforce_authorization, bool warn_authorization, const db::tablets_mode_t::mode tablets_mode);
-
-    future<> do_batch_write(
-        std::vector<std::pair<schema_ptr, put_or_delete_item>> mutation_builders,
-        service::client_state& client_state,
-        tracing::trace_state_ptr trace_state,
-        service_permit permit);
-
-    future<> cas_write(schema_ptr schema, service::cas_shard cas_shard, const dht::decorated_key& dk,
-        const std::vector<put_or_delete_item>& mutation_builders, service::client_state& client_state,
-        tracing::trace_state_ptr trace_state, service_permit permit);
 
 public:
     static void describe_key_schema(rjson::value& parent, const schema& schema, std::unordered_map<std::string,std::string>&, const std::map<sstring, sstring> *tags = nullptr);

alternator/http_compression.cc

@@ -1,301 +0,0 @@
-/*
- * Copyright 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#include "alternator/http_compression.hh"
-#include "alternator/server.hh"
-#include <seastar/coroutine/maybe_yield.hh>
-#include <zlib.h>
-
-static logging::logger slogger("alternator-http-compression");
-
-namespace alternator {
-
-
-static constexpr size_t compressed_buffer_size = 1024;
-class zlib_compressor {
-    z_stream _zs;
-    temporary_buffer<char> _output_buf;
-    noncopyable_function<future<>(temporary_buffer<char>&&)> _write_func;
-public:
-    zlib_compressor(bool gzip, int compression_level, noncopyable_function<future<>(temporary_buffer<char>&&)> write_func)
-     : _write_func(std::move(write_func)) {
-        memset(&_zs, 0, sizeof(_zs));
-        if (deflateInit2(&_zs, std::clamp(compression_level, Z_NO_COMPRESSION, Z_BEST_COMPRESSION), Z_DEFLATED,
-                (gzip ? 16 : 0) + MAX_WBITS, 8, Z_DEFAULT_STRATEGY) != Z_OK) {
-            // Should only happen if memory allocation fails
-            throw std::bad_alloc();
-        }
-    }
-    ~zlib_compressor() {
-        deflateEnd(&_zs);
-    }
-    future<> close() {
-        return compress(nullptr, 0, true);
-    }
-
-    future<> compress(const char* buf, size_t len, bool is_last_chunk = false) {
-        _zs.next_in = reinterpret_cast<unsigned char*>(const_cast<char*>(buf));
-        _zs.avail_in = (uInt) len;
-        int mode = is_last_chunk ? Z_FINISH : Z_NO_FLUSH;
-        while(_zs.avail_in > 0 || is_last_chunk) {
-            co_await coroutine::maybe_yield();
-            if (_output_buf.empty()) {
-                if (is_last_chunk) {
-                    uint32_t max_buffer_size = 0;
-                    deflatePending(&_zs, &max_buffer_size, nullptr);
-                    max_buffer_size += deflateBound(&_zs, _zs.avail_in) + 1;
-                    _output_buf = temporary_buffer<char>(std::min(compressed_buffer_size, (size_t) max_buffer_size));
-                } else {
-                    _output_buf = temporary_buffer<char>(compressed_buffer_size);
-                }
-                _zs.next_out = reinterpret_cast<unsigned char*>(_output_buf.get_write());
-                _zs.avail_out = compressed_buffer_size;
-            }
-            int e = deflate(&_zs, mode);
-            if (e < Z_OK) {
-                throw api_error::internal("Error during compression of response body");
-            }
-            if (e == Z_STREAM_END || _zs.avail_out < compressed_buffer_size / 4) {
-                _output_buf.trim(compressed_buffer_size - _zs.avail_out);
-                co_await _write_func(std::move(_output_buf));
-                if (e == Z_STREAM_END) {
-                    break;
-                }
-            }
-        }
-    }
-};
-
-// Helper string_view functions for parsing Accept-Encoding header
-struct case_insensitive_cmp_sv {
-    bool operator()(std::string_view s1, std::string_view s2) const {
-        return std::equal(s1.begin(), s1.end(), s2.begin(), s2.end(),
-            [](char a, char b) { return ::tolower(a) == ::tolower(b); });
-    }
-};
-static inline std::string_view trim_left(std::string_view sv) {
-    while (!sv.empty() && std::isspace(static_cast<unsigned char>(sv.front())))
-        sv.remove_prefix(1);
-    return sv;
-}
-static inline std::string_view trim_right(std::string_view sv) {
-    while (!sv.empty() && std::isspace(static_cast<unsigned char>(sv.back())))
-        sv.remove_suffix(1);
-    return sv;
-}
-static inline std::string_view trim(std::string_view sv) {
-    return trim_left(trim_right(sv));
-}
-
-inline std::vector<std::string_view> split(std::string_view text, char separator) {
-    std::vector<std::string_view> tokens;
-    if (text == "") {
-        return tokens;
-    }
-
-    while (true) {
-        auto pos = text.find_first_of(separator);
-        if (pos != std::string_view::npos) {
-            tokens.emplace_back(text.data(), pos);
-            text.remove_prefix(pos + 1);
-        } else {
-            tokens.emplace_back(text);
-            break;
-        }
-    }
-    return tokens;
-}
-
-constexpr response_compressor::compression_type response_compressor::get_compression_type(std::string_view encoding) {
-    for (size_t i = 0; i < static_cast<size_t>(compression_type::count); ++i) {
-        if (case_insensitive_cmp_sv{}(encoding, compression_names[i])) {
-            return static_cast<compression_type>(i);
-        }
-    }
-    return compression_type::unknown;
-}
-
-response_compressor::compression_type response_compressor::find_compression(std::string_view accept_encoding, size_t response_size) {
-    std::optional<float> ct_q[static_cast<size_t>(compression_type::count)];
-    ct_q[static_cast<size_t>(compression_type::none)] = std::numeric_limits<float>::min(); // enabled, but lowest priority
-    compression_type selected_ct = compression_type::none;
-
-    std::vector<std::string_view> entries = split(accept_encoding, ',');
-    for (auto& e : entries) {
-        std::vector<std::string_view> params = split(e, ';');
-        if (params.size() == 0) {
-            continue;
-        }
-        compression_type ct = get_compression_type(trim(params[0]));
-        if (ct == compression_type::unknown) {
-            continue; // ignore unknown encoding types
-        }
-        if (ct_q[static_cast<size_t>(ct)].has_value() && ct_q[static_cast<size_t>(ct)] != 0.0f) {
-            continue; // already processed this encoding
-        }
-        if (response_size < _threshold[static_cast<size_t>(ct)]) {
-            continue; // below threshold treat as unknown
-        }
-        for (size_t i = 1; i < params.size(); ++i) { // find "q=" parameter
-            auto pos = params[i].find("q=");
-            if (pos == std::string_view::npos) {
-                continue;
-            }
-            std::string_view param = params[i].substr(pos + 2);
-            param = trim(param);
-            // parse quality value
-            float q_value = 1.0f;
-            auto [ptr, ec] = std::from_chars(param.data(), param.data() + param.size(), q_value);
-            if (ec != std::errc() || ptr != param.data() + param.size()) {
-                continue;
-            }
-            if (q_value < 0.0) {
-                q_value = 0.0;
-            } else if (q_value > 1.0) {
-                q_value = 1.0;
-            }
-            ct_q[static_cast<size_t>(ct)] = q_value;
-            break; // we parsed quality value
-        }
-        if (!ct_q[static_cast<size_t>(ct)].has_value()) {
-            ct_q[static_cast<size_t>(ct)] = 1.0f; // default quality value
-        }
-        // keep the highest encoding (in the order, unless 'any')
-        if (selected_ct == compression_type::any) {
-            if (ct_q[static_cast<size_t>(ct)] >= ct_q[static_cast<size_t>(selected_ct)]) {
-                selected_ct = ct;
-            }
-        } else {
-            if (ct_q[static_cast<size_t>(ct)] > ct_q[static_cast<size_t>(selected_ct)]) {
-                selected_ct = ct;
-            }
-        }
-    }
-    if (selected_ct == compression_type::any) {
-        // select any not mentioned or highest quality
-        selected_ct = compression_type::none;
-        for (size_t i = 0; i < static_cast<size_t>(compression_type::compressions_count); ++i) {
-            if (!ct_q[i].has_value()) {
-                return static_cast<compression_type>(i);
-            }
-            if (ct_q[i] > ct_q[static_cast<size_t>(selected_ct)]) {
-                selected_ct = static_cast<compression_type>(i);
-            }
-        }
-    }
-    return selected_ct;
-}
-
-static future<chunked_content> compress(response_compressor::compression_type ct, const db::config& cfg, std::string str) {
-    chunked_content compressed;
-    auto write = [&compressed](temporary_buffer<char>&& buf) -> future<> {
-        compressed.push_back(std::move(buf));
-        return make_ready_future<>();
-    };
-    zlib_compressor compressor(ct != response_compressor::compression_type::deflate,
-        cfg.alternator_response_gzip_compression_level(), std::move(write));
-    co_await compressor.compress(str.data(), str.size(), true);
-    co_return compressed;
-}
-
-static sstring flatten(chunked_content&& cc) {
-    size_t total_size = 0;
-    for (const auto& chunk : cc) {
-        total_size += chunk.size();
-    }
-    sstring result = sstring{ sstring::initialized_later{}, total_size };
-    size_t offset = 0;
-    for (const auto& chunk : cc) {
-        std::copy(chunk.begin(), chunk.end(), result.begin() + offset);
-        offset += chunk.size();
-    }
-    return result;
-}
-
-future<std::unique_ptr<http::reply>> response_compressor::generate_reply(std::unique_ptr<http::reply> rep, sstring accept_encoding, const char* content_type, std::string&& response_body) {
-    response_compressor::compression_type ct = find_compression(accept_encoding, response_body.size());
-    if (ct != response_compressor::compression_type::none) {
-        rep->add_header("Content-Encoding", get_encoding_name(ct));
-        rep->set_content_type(content_type);
-        return compress(ct, cfg, std::move(response_body)).then([rep = std::move(rep)] (chunked_content compressed) mutable {
-            rep->_content = flatten(std::move(compressed));
-            return make_ready_future<std::unique_ptr<http::reply>>(std::move(rep));
-        });
-    } else {
-        // Note that despite the move, there is a copy here -
-        // as str is std::string and rep->_content is sstring.
-        rep->_content = std::move(response_body);
-        rep->set_content_type(content_type);
-    }
-    return make_ready_future<std::unique_ptr<http::reply>>(std::move(rep));
-}
-
-template<typename Compressor>
-class compressed_data_sink_impl : public data_sink_impl {
-    output_stream<char> _out;
-    Compressor _compressor;
-public:
-    template<typename... Args>
-    compressed_data_sink_impl(output_stream<char>&& out, Args&&... args)
-     : _out(std::move(out)), _compressor(std::forward<Args>(args)..., [this](temporary_buffer<char>&& buf) {
-        return _out.write(std::move(buf));
-    }) { }
-
-    future<> put(std::span<temporary_buffer<char>> data) override {
-        return data_sink_impl::fallback_put(data, [this] (temporary_buffer<char>&& buf) {
-            return do_put(std::move(buf));
-        });
-    }
-
-private:
-    future<> do_put(temporary_buffer<char> buf) {
-        co_return co_await _compressor.compress(buf.get(), buf.size());
-
-    }
-    future<> close() override {
-        return _compressor.close().then([this] {
-            return _out.close();
-        });
-    }
-};
-
-executor::body_writer compress(response_compressor::compression_type ct, const db::config& cfg, executor::body_writer&& bw) {
-    return [bw = std::move(bw), ct, level = cfg.alternator_response_gzip_compression_level()](output_stream<char>&& out) mutable -> future<> {
-        output_stream_options opts;
-        opts.trim_to_size = true;
-        std::unique_ptr<data_sink_impl> data_sink_impl;
-        switch (ct) {
-            case response_compressor::compression_type::gzip:
-                data_sink_impl = std::make_unique<compressed_data_sink_impl<zlib_compressor>>(std::move(out), true, level);
-                break;
-            case response_compressor::compression_type::deflate:
-                data_sink_impl = std::make_unique<compressed_data_sink_impl<zlib_compressor>>(std::move(out), false, level);
-                break;
-            case response_compressor::compression_type::none:
-            case response_compressor::compression_type::any:
-            case response_compressor::compression_type::unknown:
-                on_internal_error(slogger,"Compression not selected");
-            default:
-                on_internal_error(slogger, "Unsupported compression type for data sink");
-        }
-        return bw(output_stream<char>(data_sink(std::move(data_sink_impl)), compressed_buffer_size, opts));
-    };
-}
-
-future<std::unique_ptr<http::reply>> response_compressor::generate_reply(std::unique_ptr<http::reply> rep, sstring accept_encoding, const char* content_type, executor::body_writer&& body_writer) {
-    response_compressor::compression_type ct = find_compression(accept_encoding, std::numeric_limits<size_t>::max());
-    if (ct != response_compressor::compression_type::none) {
-        rep->add_header("Content-Encoding", get_encoding_name(ct));
-        rep->write_body(content_type, compress(ct, cfg, std::move(body_writer)));
-    } else {
-        rep->write_body(content_type, std::move(body_writer));
-    }
-    return make_ready_future<std::unique_ptr<http::reply>>(std::move(rep));
-}
-
-} // namespace alternator

alternator/http_compression.hh

@@ -1,91 +0,0 @@
-/*
- * Copyright 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#pragma once
-
-#include "alternator/executor.hh"
-#include <seastar/http/httpd.hh>
-#include "db/config.hh"
-
-namespace alternator {
-
-class response_compressor {
-public:
-    enum class compression_type {
-        gzip,
-        deflate,
-        compressions_count,
-        any = compressions_count,
-        none,
-        count,
-        unknown = count
-    };
-    static constexpr std::string_view compression_names[] = {
-        "gzip",
-        "deflate",
-        "*",
-        "identity"
-    };
-
-    static sstring get_encoding_name(compression_type ct) {
-        return sstring(compression_names[static_cast<size_t>(ct)]);
-    }
-    static constexpr compression_type get_compression_type(std::string_view encoding);
-
-    sstring get_accepted_encoding(const http::request& req) {
-        if (get_threshold() == 0) {
-            return "";
-        }
-        return req.get_header("Accept-Encoding");
-    }
-    compression_type find_compression(std::string_view accept_encoding, size_t response_size);
-
-    response_compressor(const db::config& cfg)
-        : cfg(cfg)
-        ,_gzip_level_observer(
-            cfg.alternator_response_gzip_compression_level.observe([this](int v) {
-                    update_threshold();
-                }))
-        ,_gzip_threshold_observer(
-            cfg.alternator_response_compression_threshold_in_bytes.observe([this](uint32_t v) {
-                    update_threshold();
-                }))
-    {
-        update_threshold();
-    }
-    response_compressor(const response_compressor& rhs) : response_compressor(rhs.cfg) {}
-
-private:
-    const db::config& cfg;
-    utils::observable<int>::observer _gzip_level_observer;
-    utils::observable<uint32_t>::observer _gzip_threshold_observer;
-    uint32_t _threshold[static_cast<size_t>(compression_type::count)];
-
-    size_t get_threshold() { return _threshold[static_cast<size_t>(compression_type::any)]; }
-    void update_threshold() {
-        _threshold[static_cast<size_t>(compression_type::none)] = std::numeric_limits<uint32_t>::max();
-        _threshold[static_cast<size_t>(compression_type::any)] = std::numeric_limits<uint32_t>::max();
-        uint32_t gzip = cfg.alternator_response_gzip_compression_level() <= 0 ? std::numeric_limits<uint32_t>::max()
-            : cfg.alternator_response_compression_threshold_in_bytes();
-        _threshold[static_cast<size_t>(compression_type::gzip)] = gzip;
-        _threshold[static_cast<size_t>(compression_type::deflate)] = gzip;
-        for (size_t i = 0; i < static_cast<size_t>(compression_type::compressions_count); ++i) {
-            if (_threshold[i] < _threshold[static_cast<size_t>(compression_type::any)]) {
-                _threshold[static_cast<size_t>(compression_type::any)] = _threshold[i];
-            }
-        }
-    }
-
-public:
-    future<std::unique_ptr<http::reply>> generate_reply(std::unique_ptr<http::reply> rep,
-         sstring accept_encoding, const char* content_type, std::string&& response_body);
-    future<std::unique_ptr<http::reply>> generate_reply(std::unique_ptr<http::reply> rep,
-         sstring accept_encoding, const char* content_type, executor::body_writer&& body_writer);
-};
-
-}

alternator/serialization.cc

@@ -496,7 +496,7 @@ const std::pair<std::string, const rjson::value*> unwrap_set(const rjson::value&
         return {"", nullptr};
     }
     auto it = v.MemberBegin();
-    const std::string it_key = rjson::to_string(it->name);
+    const std::string it_key = it->name.GetString();
     if (it_key != "SS" && it_key != "BS" && it_key != "NS") {
         return {std::move(it_key), nullptr};
     }

alternator/server.cc

@@ -34,7 +34,6 @@
 #include "client_data.hh"
 #include "utils/updateable_value.hh"
 #include <zlib.h>
-#include "alternator/http_compression.hh"
 
 static logging::logger slogger("alternator-server");
 
@@ -112,12 +111,9 @@ class api_handler : public handler_base {
     // type applies to all replies, both success and error.
     static constexpr const char* REPLY_CONTENT_TYPE = "application/x-amz-json-1.0";
 public:
-    api_handler(const std::function<future<executor::request_return_type>(std::unique_ptr<request> req)>& _handle,
-                const db::config& config) : _response_compressor(config), _f_handle(
+    api_handler(const std::function<future<executor::request_return_type>(std::unique_ptr<request> req)>& _handle) : _f_handle(
          [this, _handle](std::unique_ptr<request> req, std::unique_ptr<reply> rep) {
-         sstring accept_encoding = _response_compressor.get_accepted_encoding(*req);
-         return seastar::futurize_invoke(_handle, std::move(req)).then_wrapped(
-            [this, rep = std::move(rep), accept_encoding=std::move(accept_encoding)](future<executor::request_return_type> resf) mutable {
+         return seastar::futurize_invoke(_handle, std::move(req)).then_wrapped([this, rep = std::move(rep)](future<executor::request_return_type> resf) mutable {
              if (resf.failed()) {
                  // Exceptions of type api_error are wrapped as JSON and
                  // returned to the client as expected. Other types of
@@ -137,20 +133,22 @@ public:
                  return make_ready_future<std::unique_ptr<reply>>(std::move(rep));
              }
              auto res = resf.get();
-             return std::visit(overloaded_functor {
+             std::visit(overloaded_functor {
                 [&] (std::string&& str) {
-                    return _response_compressor.generate_reply(std::move(rep), std::move(accept_encoding),
-                                                               REPLY_CONTENT_TYPE, std::move(str));
+                    // Note that despite the move, there is a copy here -
+                    // as str is std::string and rep->_content is sstring.
+                    rep->_content = std::move(str);
+                    rep->set_content_type(REPLY_CONTENT_TYPE);
                 },
                 [&] (executor::body_writer&& body_writer) {
-                    return _response_compressor.generate_reply(std::move(rep), std::move(accept_encoding),
-                                                               REPLY_CONTENT_TYPE, std::move(body_writer));
+                    rep->write_body(REPLY_CONTENT_TYPE, std::move(body_writer));
                 },
                 [&] (const api_error& err) {
                     generate_error_reply(*rep, err);
-                    return make_ready_future<std::unique_ptr<reply>>(std::move(rep));
                 }
              }, std::move(res));
+
+             return make_ready_future<std::unique_ptr<reply>>(std::move(rep));
          });
     }) { }
 
@@ -179,7 +177,6 @@ protected:
         slogger.trace("api_handler error case: {}", rep._content);
     }
 
-    response_compressor _response_compressor;
     future_handler_function _f_handle;
 };
 
@@ -374,40 +371,13 @@ future<std::string> server::verify_signature(const request& req, const chunked_c
     for (const auto& header : signed_headers) {
         signed_headers_map.emplace(header, std::string_view());
     }
-    std::vector<std::string> modified_values;
     for (auto& header : req._headers) {
         std::string header_str;
         header_str.resize(header.first.size());
         std::transform(header.first.begin(), header.first.end(), header_str.begin(), ::tolower);
         auto it = signed_headers_map.find(header_str);
         if (it != signed_headers_map.end()) {
-            // replace multiple spaces in the header value header.second with
-            // a single space, as required by AWS SigV4 header canonization.
-            // If we modify the value, we need to save it in modified_values
-            // to keep it alive.
-            std::string value;
-            value.reserve(header.second.size());
-            bool prev_space = false;
-            bool modified = false;
-            for (char ch : header.second) {
-                if (ch == ' ') {
-                    if (!prev_space) {
-                        value += ch;
-                        prev_space = true;
-                    } else {
-                        modified = true; // skip a space
-                    }
-                } else {
-                    value += ch;
-                    prev_space = false;
-                }
-            }
-            if (modified) {
-                modified_values.emplace_back(std::move(value));
-                it->second = std::string_view(modified_values.back());
-            } else {
-                it->second = std::string_view(header.second);
-            }
+            it->second = std::string_view(header.second);
         }
     }
 
@@ -420,7 +390,6 @@ future<std::string> server::verify_signature(const request& req, const chunked_c
                                                     datestamp = std::move(datestamp),
                                                     signed_headers_str = std::move(signed_headers_str),
                                                     signed_headers_map = std::move(signed_headers_map),
-                                                    modified_values = std::move(modified_values),
                                                     region = std::move(region),
                                                     service = std::move(service),
                                                     user_signature = std::move(user_signature)] (future<key_cache::value_ptr> key_ptr_fut) {
@@ -591,11 +560,11 @@ read_entire_stream(input_stream<char>& inp, size_t length_limit) {
 class safe_gzip_zstream {
     z_stream _zs;
 public:
-    // If gzip is true, decode a gzip header (for "Content-Encoding: gzip").
-    // Otherwise, a zlib header (for "Content-Encoding: deflate").
-    safe_gzip_zstream(bool gzip = true) {
+    safe_gzip_zstream() {
         memset(&_zs, 0, sizeof(_zs));
-        if (inflateInit2(&_zs, gzip ? 16 + MAX_WBITS : MAX_WBITS) != Z_OK) {
+        // The strange 16 + WMAX_BITS tells zlib to expect and decode
+        // a gzip header, not a zlib header.
+        if (inflateInit2(&_zs, 16 + MAX_WBITS) != Z_OK) {
             // Should only happen if memory allocation fails
             throw std::bad_alloc();
         }
@@ -614,21 +583,19 @@ public:
     }
 };
 
-// ungzip() takes a chunked_content of a compressed request body, and returns
-// the uncompressed content as a chunked_content. If gzip is true, we expect
-// gzip header (for "Content-Encoding: gzip"), if gzip is false, we expect a
-// zlib header (for "Content-Encoding: deflate").
+// ungzip() takes a chunked_content with a gzip-compressed request body,
+// uncompresses it, and returns the uncompressed content as a chunked_content.
 // If the uncompressed content exceeds length_limit, an error is thrown.
 static future<chunked_content>
-ungzip(chunked_content&& compressed_body, size_t length_limit, bool gzip = true) {
+ungzip(chunked_content&& compressed_body, size_t length_limit) {
     chunked_content ret;
     // output_buf can be any size - when uncompressing input_buf, it doesn't
     // need to fit in a single output_buf, we'll use multiple output_buf for
     // a single input_buf if needed.
     constexpr size_t OUTPUT_BUF_SIZE = 4096;
     temporary_buffer<char> output_buf;
-    safe_gzip_zstream strm(gzip);
-    bool complete_stream = false; // empty input is not a valid gzip/deflate
+    safe_gzip_zstream strm;
+    bool complete_stream = false; // empty input is not a valid gzip
     size_t total_out_bytes = 0;
     for (const temporary_buffer<char>& input_buf : compressed_body) {
         if (input_buf.empty()) {
@@ -731,8 +698,6 @@ future<executor::request_return_type> server::handle_api_request(std::unique_ptr
     sstring content_encoding = req->get_header("Content-Encoding");
     if (content_encoding == "gzip") {
         content = co_await ungzip(std::move(content), request_content_length_limit);
-    } else if (content_encoding == "deflate") {
-        content = co_await ungzip(std::move(content), request_content_length_limit, false);
     } else if (!content_encoding.empty()) {
         // DynamoDB returns a 500 error for unsupported Content-Encoding.
         // I'm not sure if this is the best error code, but let's do it too.
@@ -743,12 +708,8 @@ future<executor::request_return_type> server::handle_api_request(std::unique_ptr
     // As long as the system_clients_entry object is alive, this request will
     // be visible in the "system.clients" virtual table. When requested, this
     // entry will be formatted by server::ongoing_request::make_client_data().
-    auto user_agent_header = co_await _connection_options_keys_and_values.get_or_load(req->get_header("User-Agent"), [] (const client_options_cache_key_type&) {
-        return make_ready_future<options_cache_value_type>(options_cache_value_type{});
-    });
-
     auto system_clients_entry = _ongoing_requests.emplace(
-        req->get_client_address(), std::move(user_agent_header),
+        req->get_client_address(), req->get_header("User-Agent"),
         username, current_scheduling_group(),
         req->get_protocol_name() == "https");
 
@@ -793,7 +754,7 @@ future<executor::request_return_type> server::handle_api_request(std::unique_ptr
 void server::set_routes(routes& r) {
     api_handler* req_handler = new api_handler([this] (std::unique_ptr<request> req) mutable {
         return handle_api_request(std::move(req));
-    }, _proxy.data_dictionary().get_config());
+    });
 
     r.put(operation_type::POST, "/", req_handler);
     r.put(operation_type::GET, "/", new health_handler(_pending_requests));
@@ -904,9 +865,7 @@ server::server(executor& exec, service::storage_proxy& proxy, gms::gossiper& gos
     } {
 }
 
-future<> server::init(net::inet_address addr, std::optional<uint16_t> port, std::optional<uint16_t> https_port,
-        std::optional<uint16_t> port_proxy_protocol, std::optional<uint16_t> https_port_proxy_protocol,
-        std::optional<tls::credentials_builder> creds,
+future<> server::init(net::inet_address addr, std::optional<uint16_t> port, std::optional<uint16_t> https_port, std::optional<tls::credentials_builder> creds,
         utils::updateable_value<bool> enforce_authorization, utils::updateable_value<bool> warn_authorization, utils::updateable_value<uint64_t> max_users_query_size_in_trace_output,
         semaphore* memory_limiter, utils::updateable_value<uint32_t> max_concurrent_requests) {
     _memory_limiter = memory_limiter;
@@ -914,28 +873,20 @@ future<> server::init(net::inet_address addr, std::optional<uint16_t> port, std:
     _warn_authorization = std::move(warn_authorization);
     _max_concurrent_requests = std::move(max_concurrent_requests);
     _max_users_query_size_in_trace_output = std::move(max_users_query_size_in_trace_output);
-    if (!port && !https_port && !port_proxy_protocol && !https_port_proxy_protocol) {
+    if (!port && !https_port) {
         return make_exception_future<>(std::runtime_error("Either regular port or TLS port"
                 " must be specified in order to init an alternator HTTP server instance"));
     }
-    return seastar::async([this, addr, port, https_port, port_proxy_protocol, https_port_proxy_protocol, creds] {
+    return seastar::async([this, addr, port, https_port, creds] {
         _executor.start().get();
 
-        if (port || port_proxy_protocol) {
+        if (port) {
             set_routes(_http_server._routes);
             _http_server.set_content_streaming(true);
-            if (port) {
-                _http_server.listen(socket_address{addr, *port}).get();
-            }
-            if (port_proxy_protocol) {
-                listen_options lo;
-                lo.reuse_address = true;
-                lo.proxy_protocol = true;
-                _http_server.listen(socket_address{addr, *port_proxy_protocol}, lo).get();
-            }
+            _http_server.listen(socket_address{addr, *port}).get();
             _enabled_servers.push_back(std::ref(_http_server));
         }
-        if (https_port || https_port_proxy_protocol) {
+        if (https_port) {
             set_routes(_https_server._routes);
             _https_server.set_content_streaming(true);
 
@@ -955,15 +906,7 @@ future<> server::init(net::inet_address addr, std::optional<uint16_t> port, std:
             } else {
                 _credentials = creds->build_server_credentials();
             }
-            if (https_port) {
-                _https_server.listen(socket_address{addr, *https_port}, _credentials).get();
-            }
-            if (https_port_proxy_protocol) {
-                listen_options lo;
-                lo.reuse_address = true;
-                lo.proxy_protocol = true;
-                _https_server.listen(socket_address{addr, *https_port_proxy_protocol}, lo, _credentials).get();
-            }
+            _https_server.listen(socket_address{addr, *https_port}, _credentials).get();
             _enabled_servers.push_back(std::ref(_https_server));
         }
     });
@@ -1036,15 +979,16 @@ client_data server::ongoing_request::make_client_data() const {
     // and keep "driver_version" unset.
     cd.driver_name = _user_agent;
     // Leave "protocol_version" unset, it has no meaning in Alternator.
-    // Leave "hostname", "ssl_protocol" and "ssl_cipher_suite" unset for Alternator.
-    // Note: CQL sets ssl_protocol and ssl_cipher_suite via generic_server::connection base class.
+    // Leave "hostname", "ssl_protocol" and "ssl_cipher_suite" unset.
+    // As reported in issue #9216, we never set these fields in CQL
+    // either (see cql_server::connection::make_client_data()).
     return cd;
 }
 
-future<utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>>> server::get_client_data() {
-    utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>> ret;
+future<utils::chunked_vector<client_data>> server::get_client_data() {
+    utils::chunked_vector<client_data> ret;
     co_await _ongoing_requests.for_each_gently([&ret] (const ongoing_request& r) {
-        ret.emplace_back(make_foreign(std::make_unique<client_data>(r.make_client_data())));
+        ret.emplace_back(r.make_client_data());
     });
     co_return ret;
 }

alternator/server.hh

@@ -55,7 +55,6 @@ class server : public peering_sharded_service<server> {
     // though it isn't really relevant for Alternator which defines its own
     // timeouts separately. We can create this object only once.
     updateable_timeout_config _timeout_config;
-    client_options_cache_type _connection_options_keys_and_values;
 
     alternator_callbacks_map _callbacks;
 
@@ -89,7 +88,7 @@ class server : public peering_sharded_service<server> {
     // is called when reading the "system.clients" virtual table.
     struct ongoing_request {
         socket_address _client_address;
-        client_options_cache_entry_type _user_agent;
+        sstring _user_agent;
         sstring _username;
         scheduling_group _scheduling_group;
         bool _is_https;
@@ -100,9 +99,7 @@ class server : public peering_sharded_service<server> {
 public:
     server(executor& executor, service::storage_proxy& proxy, gms::gossiper& gossiper, auth::service& service, qos::service_level_controller& sl_controller);
 
-    future<> init(net::inet_address addr, std::optional<uint16_t> port, std::optional<uint16_t> https_port,
-            std::optional<uint16_t> port_proxy_protocol, std::optional<uint16_t> https_port_proxy_protocol,
-            std::optional<tls::credentials_builder> creds,
+    future<> init(net::inet_address addr, std::optional<uint16_t> port, std::optional<uint16_t> https_port, std::optional<tls::credentials_builder> creds,
             utils::updateable_value<bool> enforce_authorization, utils::updateable_value<bool> warn_authorization, utils::updateable_value<uint64_t> max_users_query_size_in_trace_output,
             semaphore* memory_limiter, utils::updateable_value<uint32_t> max_concurrent_requests);
     future<> stop();
@@ -110,7 +107,7 @@ public:
     // table "system.clients" is read. It is expected to generate a list of
     // clients connected to this server (on this shard). This function is
     // called by alternator::controller::get_client_data().
-    future<utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>>> get_client_data();
+    future<utils::chunked_vector<client_data>> get_client_data();
 private:
     void set_routes(seastar::httpd::routes& r);
     // If verification succeeds, returns the authenticated user's username

alternator/ttl.cc

@@ -93,7 +93,7 @@ future<executor::request_return_type> executor::update_time_to_live(client_state
     if (v->GetStringLength() < 1 || v->GetStringLength() > 255) {
         co_return api_error::validation("The length of AttributeName must be between 1 and 255");
     }
-    sstring attribute_name = rjson::to_sstring(*v);
+    sstring attribute_name(v->GetString(), v->GetStringLength());
 
     co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, schema, auth::permission::ALTER, _stats);
     co_await db::modify_tags(_mm, schema->ks_name(), schema->cf_name(), [&](std::map<sstring, sstring>& tags_map) {

api/CMakeLists.txt

@@ -31,7 +31,6 @@ set(swagger_files
   api-doc/column_family.json
   api-doc/commitlog.json
   api-doc/compaction_manager.json
-  api-doc/client_routes.json
   api-doc/config.json
   api-doc/cql_server_test.json
   api-doc/endpoint_snitch_info.json
@@ -69,7 +68,6 @@ target_sources(api
   PRIVATE
     api.cc
     cache_service.cc
-    client_routes.cc
     collectd.cc
     column_family.cc
     commitlog.cc

api/api-doc/client_routes.def.json

@@ -1,23 +0,0 @@
-    , "client_routes_entry": {
-        "id": "client_routes_entry",
-        "summary": "An entry storing client routes",
-        "properties": {
-            "connection_id": {"type": "string"},
-            "host_id": {"type": "string", "format": "uuid"},
-            "address": {"type": "string"},
-            "port": {"type": "integer"},
-            "tls_port": {"type": "integer"},
-            "alternator_port": {"type": "integer"},
-            "alternator_https_port": {"type": "integer"}
-        },
-        "required": ["connection_id", "host_id", "address"]
-    }
-    , "client_routes_key": {
-        "id": "client_routes_key",
-        "summary": "A key of client_routes_entry",
-        "properties": {
-            "connection_id": {"type": "string"},
-            "host_id": {"type": "string", "format": "uuid"}
-        }
-    }
-

api/api-doc/client_routes.json

@@ -1,74 +0,0 @@
-    , "/v2/client-routes":{
-        "get": {
-            "description":"List all client route entries",
-            "operationId":"get_client_routes",
-            "tags":["client_routes"],
-            "produces":[
-                "application/json"
-            ],
-            "parameters":[],
-            "responses":{
-                "200":{
-                    "schema":{
-                        "type":"array",
-                        "items":{ "$ref":"#/definitions/client_routes_entry" }
-                    }
-                },
-                "default":{
-                    "description":"unexpected error",
-                    "schema":{"$ref":"#/definitions/ErrorModel"}
-                }
-            }
-        },
-        "post": {
-            "description":"Upsert one or more client route entries",
-            "operationId":"set_client_routes",
-            "tags":["client_routes"],
-            "parameters":[
-                {
-                    "name":"body",
-                    "in":"body",
-                    "required":true,
-                    "schema":{
-                        "type":"array",
-                        "items":{ "$ref":"#/definitions/client_routes_entry" }
-                    }
-                }
-            ],
-            "responses":{
-                "200":{ "description": "OK" },
-                "default":{
-                    "description":"unexpected error",
-                    "schema":{ "$ref":"#/definitions/ErrorModel" }
-                }
-            }
-        },
-        "delete": {
-            "description":"Delete one or more client route entries",
-            "operationId":"delete_client_routes",
-            "tags":["client_routes"],
-            "parameters":[
-                {
-                    "name":"body",
-                    "in":"body",
-                    "required":true,
-                    "schema":{
-                        "type":"array",
-                        "items":{ "$ref":"#/definitions/client_routes_key" }
-                    }
-                }
-            ],
-            "responses":{
-                "200":{
-                    "description": "OK"
-                },
-                "default":{
-                    "description":"unexpected error",
-                    "schema":{
-                        "$ref":"#/definitions/ErrorModel"
-                    }
-                }
-            }
-        }
-    }
-

api/api-doc/storage_service.json

@@ -3051,7 +3051,7 @@
                   },
                   {
                      "name":"incremental_mode",
-                     "description":"Set the incremental repair mode. Can be 'disabled', 'incremental', or 'full'. 'incremental': The incremental repair logic is enabled. Unrepaired sstables will be included for repair. Repaired sstables will be skipped. The incremental repair states will be updated after repair. 'full': The incremental repair logic is enabled. Both repaired and unrepaired sstables will be included for repair. The incremental repair states will be updated after repair. 'disabled': The incremental repair logic is disabled completely. The incremental repair states, e.g., repaired_at in sstables and sstables_repaired_at in the system.tablets table, will not be updated after repair. When the option is not provided, it defaults to 'disabled' mode.",
+                     "description":"Set the incremental repair mode. Can be 'disabled', 'incremental', or 'full'. 'incremental': The incremental repair logic is enabled. Unrepaired sstables will be included for repair. Repaired sstables will be skipped. The incremental repair states will be updated after repair. 'full': The incremental repair logic is enabled. Both repaired and unrepaired sstables will be included for repair. The incremental repair states will be updated after repair. 'disabled': The incremental repair logic is disabled completely. The incremental repair states, e.g., repaired_at in sstables and sstables_repaired_at in the system.tablets table, will not be updated after repair. When the option is not provided, it defaults to incremental mode.",
                      "required":false,
                      "allowMultiple":false,
                      "type":"string",

api/api.cc

@@ -37,7 +37,6 @@
 #include "raft.hh"
 #include "gms/gossip_address_map.hh"
 #include "service_levels.hh"
-#include "client_routes.hh"
 
 logging::logger apilog("api");
 
@@ -68,11 +67,9 @@ future<> set_server_init(http_context& ctx) {
         rb02->set_api_doc(r);
         rb02->register_api_file(r, "swagger20_header");
         rb02->register_api_file(r, "metrics");
-        rb02->register_api_file(r, "client_routes");
         rb->register_function(r, "system",
                 "The system related API");
         rb02->add_definitions_file(r, "metrics");
-        rb02->add_definitions_file(r, "client_routes");
         set_system(ctx, r);
         rb->register_function(r, "error_injection",
             "The error injection API");
@@ -132,16 +129,6 @@ future<> unset_server_storage_service(http_context& ctx) {
     return ctx.http_server.set_routes([&ctx] (routes& r) { unset_storage_service(ctx, r); });
 }
 
-future<> set_server_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr) {
-    return ctx.http_server.set_routes([&ctx, &cr] (routes& r) {
-        set_client_routes(ctx, r, cr);
-    });
-}
-
-future<> unset_server_client_routes(http_context& ctx) {
-    return ctx.http_server.set_routes([&ctx] (routes& r) { unset_client_routes(ctx, r); });
-}
-
 future<> set_load_meter(http_context& ctx, service::load_meter& lm) {
     return ctx.http_server.set_routes([&ctx, &lm] (routes& r) { set_load_meter(ctx, r, lm); });
 }

api/api_init.hh

@@ -29,7 +29,6 @@ class storage_proxy;
 class storage_service;
 class raft_group0_client;
 class raft_group_registry;
-class client_routes_service;
 
 } // namespace service
 
@@ -100,8 +99,6 @@ future<> set_server_snitch(http_context& ctx, sharded<locator::snitch_ptr>& snit
 future<> unset_server_snitch(http_context& ctx);
 future<> set_server_storage_service(http_context& ctx, sharded<service::storage_service>& ss, service::raft_group0_client&);
 future<> unset_server_storage_service(http_context& ctx);
-future<> set_server_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr);
-future<> unset_server_client_routes(http_context& ctx);
 future<> set_server_sstables_loader(http_context& ctx, sharded<sstables_loader>& sst_loader);
 future<> unset_server_sstables_loader(http_context& ctx);
 future<> set_server_view_builder(http_context& ctx, sharded<db::view::view_builder>& vb, sharded<gms::gossiper>& g);

api/client_routes.cc

@@ -1,176 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- *
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
- #include <seastar/http/short_streams.hh>
-
-#include "client_routes.hh"
-#include "api/api.hh"
-#include "service/storage_service.hh"
-#include "service/client_routes.hh"
-#include "utils/rjson.hh"
-
-
-#include "api/api-doc/client_routes.json.hh"
-
-using namespace seastar::httpd;
-using namespace std::chrono_literals;
-using namespace json;
-
-extern logging::logger apilog;
-
-namespace api {
-
-static void validate_client_routes_endpoint(sharded<service::client_routes_service>& cr, sstring endpoint_name) {
-    if (!cr.local().get_feature_service().client_routes) {
-        apilog.warn("{}: called before the cluster feature was enabled", endpoint_name);
-        throw std::runtime_error(fmt::format("{} requires all nodes to support the CLIENT_ROUTES cluster feature", endpoint_name));
-    }
-}
-
-static sstring parse_string(const char* name, rapidjson::Value const& v) {
-    const auto it = v.FindMember(name);
-    if (it == v.MemberEnd()) {
-        throw bad_param_exception(fmt::format("Missing '{}'", name));
-    }
-    if (!it->value.IsString()) {
-        throw bad_param_exception(fmt::format("'{}' must be a string", name));
-    }
-    return {it->value.GetString(), it->value.GetStringLength()};
-}
-
-static std::optional<uint32_t> parse_port(const char* name, rapidjson::Value const& v) {
-    const auto it = v.FindMember(name);
-    if (it == v.MemberEnd()) {
-        return std::nullopt;
-    }
-    if (!it->value.IsInt()) {
-        throw bad_param_exception(fmt::format("'{}' must be an integer", name));
-    }
-    auto port = it->value.GetInt();
-    if (port < 1 || port > 65535) {
-        throw bad_param_exception(fmt::format("'{}' value={} is outside the allowed port range", name, port));
-    }
-    return port;
-}
-
-static std::vector<service::client_routes_service::client_route_entry> parse_set_client_array(const rapidjson::Document& root) {
-    if (!root.IsArray()) {
-        throw bad_param_exception("Body must be a JSON array");
-    }
-
-    std::vector<service::client_routes_service::client_route_entry> v;
-    v.reserve(root.GetArray().Size());
-    for (const auto& element : root.GetArray()) {
-        if (!element.IsObject()) { throw bad_param_exception("Each element must be object"); }
-
-        const auto port = parse_port("port", element);
-        const auto tls_port = parse_port("tls_port", element);
-        const auto alternator_port = parse_port("alternator_port", element);
-        const auto alternator_https_port = parse_port("alternator_https_port", element);
-
-        if (!port.has_value() && !tls_port.has_value() && !alternator_port.has_value() && !alternator_https_port.has_value()) {
-            throw bad_param_exception("At least one port field ('port', 'tls_port', 'alternator_port', 'alternator_https_port') must be specified");
-        }
-
-        v.emplace_back(
-            parse_string("connection_id", element),
-            utils::UUID{parse_string("host_id", element)},
-            parse_string("address", element),
-            port,
-            tls_port,
-            alternator_port,
-            alternator_https_port
-        );
-    }
-
-    return v;
-}
-
-static
-future<json::json_return_type>
-rest_set_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr, std::unique_ptr<http::request> req) {
-    validate_client_routes_endpoint(cr, "rest_set_client_routes");
-
-    rapidjson::Document root;
-    auto content = co_await util::read_entire_stream_contiguous(*req->content_stream);
-    root.Parse(content.c_str());
-
-    co_await cr.local().set_client_routes(parse_set_client_array(root));
-    co_return seastar::json::json_void();
-}
-
-static std::vector<service::client_routes_service::client_route_key> parse_delete_client_array(const rapidjson::Document& root) {
-    if (!root.IsArray()) {
-        throw bad_param_exception("Body must be a JSON array");
-    }
-
-    std::vector<service::client_routes_service::client_route_key> v;
-    v.reserve(root.GetArray().Size());
-    for (const auto& element : root.GetArray()) {
-        v.emplace_back(
-            parse_string("connection_id", element),
-            utils::UUID{parse_string("host_id", element)}
-        );
-    }
-
-    return v;
-}
-
-static
-future<json::json_return_type>
-rest_delete_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr, std::unique_ptr<http::request> req) {
-    validate_client_routes_endpoint(cr, "delete_client_routes");
-
-    rapidjson::Document root;
-    auto content = co_await util::read_entire_stream_contiguous(*req->content_stream);
-    root.Parse(content.c_str());
-
-    co_await cr.local().delete_client_routes(parse_delete_client_array(root));
-    co_return seastar::json::json_void();
-}
-
-static
-future<json::json_return_type>
-rest_get_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr, std::unique_ptr<http::request> req) {
-    validate_client_routes_endpoint(cr, "get_client_routes");
-
-    co_return co_await cr.invoke_on(0, [] (service::client_routes_service& cr) -> future<json::json_return_type> {
-        co_return json::json_return_type(stream_range_as_array(co_await cr.get_client_routes(), [](const service::client_routes_service::client_route_entry & entry) {
-            seastar::httpd::client_routes_json::client_routes_entry obj;
-            obj.connection_id = entry.connection_id;
-            obj.host_id = fmt::to_string(entry.host_id);
-            obj.address = entry.address;
-            if (entry.port.has_value()) { obj.port = entry.port.value(); }
-            if (entry.tls_port.has_value()) { obj.tls_port = entry.tls_port.value(); }
-            if (entry.alternator_port.has_value()) { obj.alternator_port = entry.alternator_port.value(); }
-            if (entry.alternator_https_port.has_value()) { obj.alternator_https_port = entry.alternator_https_port.value(); }
-            return obj;
-        }));
-    });
-}
-
-void set_client_routes(http_context& ctx, routes& r, sharded<service::client_routes_service>& cr) {
-    seastar::httpd::client_routes_json::set_client_routes.set(r, [&ctx, &cr] (std::unique_ptr<seastar::http::request> req) {
-        return rest_set_client_routes(ctx, cr, std::move(req));
-    });
-    seastar::httpd::client_routes_json::delete_client_routes.set(r, [&ctx, &cr] (std::unique_ptr<seastar::http::request> req) {
-        return rest_delete_client_routes(ctx, cr, std::move(req));
-    });
-    seastar::httpd::client_routes_json::get_client_routes.set(r, [&ctx, &cr] (std::unique_ptr<seastar::http::request> req) {
-        return rest_get_client_routes(ctx, cr, std::move(req));
-    });
-}
-
-void unset_client_routes(http_context& ctx, routes& r) {
-    seastar::httpd::client_routes_json::set_client_routes.unset(r);
-    seastar::httpd::client_routes_json::delete_client_routes.unset(r);
-    seastar::httpd::client_routes_json::get_client_routes.unset(r);
-}
-
-}

api/client_routes.hh

@@ -1,20 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- *
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-#pragma once
-
-#include <seastar/core/sharded.hh>
-#include <seastar/json/json_elements.hh>
-#include "api/api_init.hh"
-
-namespace api {
-
-void set_client_routes(http_context& ctx, httpd::routes& r, sharded<service::client_routes_service>& cr);
-void unset_client_routes(http_context& ctx, httpd::routes& r);
-
-}

api/storage_service.cc

@@ -547,13 +547,17 @@ void set_view_builder(http_context& ctx, routes& r, sharded<db::view::view_build
                 vp.insert(b.second);
             }
         }
+        std::vector<sstring> res;
         replica::database& db = vb.local().get_db();
         auto uuid = validate_table(db, ks, cf_name);
         replica::column_family& cf = db.find_column_family(uuid);
-        co_return cf.get_index_manager().list_indexes()
-                | std::views::transform([] (const auto& i) { return i.metadata().name(); })
-                | std::views::filter([&vp] (const auto& n) { return vp.contains(secondary_index::index_table_name(n)); })
-                | std::ranges::to<std::vector>();
+        res.reserve(cf.get_index_manager().list_indexes().size());
+        for (auto&& i : cf.get_index_manager().list_indexes()) {
+            if (vp.contains(secondary_index::index_table_name(i.metadata().name()))) {
+                res.emplace_back(i.metadata().name());
+            }
+        }
+        co_return res;
     });
 
 }

api/task_manager.cc

@@ -9,7 +9,6 @@
 #include <seastar/core/chunked_fifo.hh>
 #include <seastar/core/coroutine.hh>
 #include <seastar/coroutine/exception.hh>
-#include <seastar/coroutine/maybe_yield.hh>
 #include <seastar/http/exception.hh>
 
 #include "task_manager.hh"
@@ -265,7 +264,7 @@ void set_task_manager(http_context& ctx, routes& r, sharded<tasks::task_manager>
                 if (id) {
                     module->unregister_task(id);
                 }
-                co_await coroutine::maybe_yield();
+                co_await maybe_yield();
             }
         });
         co_return json_void();

auth/CMakeLists.txt

@@ -9,7 +9,6 @@ target_sources(scylla_auth
     allow_all_authorizer.cc
     authenticated_user.cc
     authenticator.cc
-    cache.cc
     certificate_authenticator.cc
     common.cc
     default_authorizer.cc

auth/allow_all_authenticator.cc

@@ -9,6 +9,7 @@
 #include "auth/allow_all_authenticator.hh"
 
 #include "service/migration_manager.hh"
+#include "utils/alien_worker.hh"
 #include "utils/class_registrator.hh"
 
 namespace auth {
@@ -22,6 +23,6 @@ static const class_registrator<
         cql3::query_processor&,
         ::service::raft_group0_client&,
         ::service::migration_manager&,
-        cache&> registration("org.apache.cassandra.auth.AllowAllAuthenticator");
+        utils::alien_worker&> registration("org.apache.cassandra.auth.AllowAllAuthenticator");
 
 }

auth/allow_all_authenticator.hh

@@ -12,8 +12,8 @@
 
 #include "auth/authenticated_user.hh"
 #include "auth/authenticator.hh"
-#include "auth/cache.hh"
 #include "auth/common.hh"
+#include "utils/alien_worker.hh"
 
 namespace cql3 {
 class query_processor;
@@ -29,7 +29,7 @@ extern const std::string_view allow_all_authenticator_name;
 
 class allow_all_authenticator final : public authenticator {
 public:
-    allow_all_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&) {
+    allow_all_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&) {
     }
 
     virtual future<> start() override {

auth/cache.cc

@@ -1,188 +0,0 @@
-/*
- * Copyright (C) 2017-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#include "auth/cache.hh"
-#include "auth/common.hh"
-#include "auth/roles-metadata.hh"
-#include "cql3/query_processor.hh"
-#include "cql3/untyped_result_set.hh"
-#include "db/consistency_level_type.hh"
-#include "db/system_keyspace.hh"
-#include "schema/schema.hh"
-#include <iterator>
-#include <seastar/core/abort_source.hh>
-#include <seastar/coroutine/maybe_yield.hh>
-#include <seastar/core/format.hh>
-
-namespace auth {
-
-logging::logger logger("auth-cache");
-
-cache::cache(cql3::query_processor& qp, abort_source& as) noexcept
-    : _current_version(0)
-    , _qp(qp)
-    , _loading_sem(1)
-    , _as(as) {
-}
-
-lw_shared_ptr<const cache::role_record> cache::get(const role_name_t& role) const noexcept {
-    auto it = _roles.find(role);
-    if (it == _roles.end()) {
-        return {};
-    }
-    return it->second;
-}
-
-future<lw_shared_ptr<cache::role_record>> cache::fetch_role(const role_name_t& role) const {
-    auto rec = make_lw_shared<role_record>();
-    rec->version = _current_version;
-
-    auto fetch = [this, &role](const sstring& q) {
-        return _qp.execute_internal(q, db::consistency_level::LOCAL_ONE,
-                internal_distributed_query_state(), {role},
-                cql3::query_processor::cache_internal::yes);
-    };
-    // roles
-    {
-        static const sstring q = format("SELECT * FROM {}.{} WHERE role = ?", db::system_keyspace::NAME, meta::roles_table::name);
-        auto rs = co_await fetch(q);
-        if (!rs->empty()) {
-            auto& r = rs->one();
-            rec->is_superuser = r.get_or<bool>("is_superuser", false);
-            rec->can_login = r.get_or<bool>("can_login", false);
-            rec->salted_hash = r.get_or<sstring>("salted_hash", "");
-            if (r.has("member_of")) {
-                auto mo = r.get_set<sstring>("member_of");
-                rec->member_of.insert(
-                        std::make_move_iterator(mo.begin()),
-                        std::make_move_iterator(mo.end()));
-            }
-        } else {
-            // role got deleted
-            co_return nullptr;
-        }
-    }
-    // members
-    {
-        static const sstring q = format("SELECT role, member FROM {}.{} WHERE role = ?", db::system_keyspace::NAME, ROLE_MEMBERS_CF);
-        auto rs = co_await fetch(q);
-        for (const auto& r : *rs) {
-            rec->members.insert(r.get_as<sstring>("member"));
-            co_await coroutine::maybe_yield();
-        }
-    }
-    // attributes
-    {
-        static const sstring q = format("SELECT role, name, value FROM {}.{} WHERE role = ?", db::system_keyspace::NAME, ROLE_ATTRIBUTES_CF);
-        auto rs = co_await fetch(q);
-        for (const auto& r : *rs) {
-            rec->attributes[r.get_as<sstring>("name")] =
-                    r.get_as<sstring>("value");
-            co_await coroutine::maybe_yield();
-        }
-    }
-    // permissions
-    {
-        static const sstring q = format("SELECT role, resource, permissions FROM {}.{} WHERE role = ?", db::system_keyspace::NAME, PERMISSIONS_CF);
-        auto rs = co_await fetch(q);
-        for (const auto& r : *rs) {
-            auto resource = r.get_as<sstring>("resource");
-            auto perms_strings = r.get_set<sstring>("permissions");
-            std::unordered_set<sstring> perms_set(perms_strings.begin(), perms_strings.end());
-            auto pset = permissions::from_strings(perms_set);
-            rec->permissions[std::move(resource)] = std::move(pset);
-            co_await coroutine::maybe_yield();
-        }
-    }
-    co_return rec;
-}
-
-future<> cache::prune_all() noexcept {
-    for (auto it = _roles.begin(); it != _roles.end(); ) {
-        if (it->second->version != _current_version) {
-            _roles.erase(it++);
-            co_await coroutine::maybe_yield();
-        } else {
-            ++it;
-        }
-    }
-    co_return;
-}
-
-future<> cache::load_all() {
-    if (legacy_mode(_qp)) {
-        co_return;
-    }
-    SCYLLA_ASSERT(this_shard_id() == 0);
-    auto units = co_await get_units(_loading_sem, 1, _as);
-
-    ++_current_version;
-
-    logger.info("Loading all roles");
-    const uint32_t page_size = 128;
-    auto loader = [this](const cql3::untyped_result_set::row& r) -> future<stop_iteration> {
-        const auto name = r.get_as<sstring>("role");
-        auto role = co_await fetch_role(name);
-        if (role) {
-            _roles[name] = role;
-        }
-        co_return stop_iteration::no;
-    };
-    co_await _qp.query_internal(format("SELECT * FROM {}.{}",
-            db::system_keyspace::NAME, meta::roles_table::name),
-            db::consistency_level::LOCAL_ONE, {}, page_size, loader);
-
-    co_await prune_all();
-    for (const auto& [name, role] : _roles) {
-        co_await distribute_role(name, role);
-    }
-    co_await container().invoke_on_others([this](cache& c) -> future<> {
-        c._current_version = _current_version;
-        co_await c.prune_all();
-    });
-}
-
-future<> cache::load_roles(std::unordered_set<role_name_t> roles) {
-    if (legacy_mode(_qp)) {
-        co_return;
-    }
-    SCYLLA_ASSERT(this_shard_id() == 0);
-    auto units = co_await get_units(_loading_sem, 1, _as);
-
-    for (const auto& name : roles) {
-        logger.info("Loading role {}", name);
-        auto role = co_await fetch_role(name);
-         if (role) {
-            _roles[name] = role;
-        } else {
-            _roles.erase(name);
-        }
-        co_await distribute_role(name, role);
-    }
-}
-
-future<> cache::distribute_role(const role_name_t& name, lw_shared_ptr<role_record> role) {
-    auto role_ptr = role.get();
-    co_await container().invoke_on_others([&name, role_ptr](cache& c) {
-        if (!role_ptr) {
-            c._roles.erase(name);
-            return;
-        }
-        auto role_copy = make_lw_shared<role_record>(*role_ptr);
-        c._roles[name] = std::move(role_copy);
-    });
-}
-
-bool cache::includes_table(const table_id& id) noexcept {
-    return id == db::system_keyspace::roles()->id()
-            || id == db::system_keyspace::role_members()->id()
-            || id == db::system_keyspace::role_attributes()->id()
-            || id == db::system_keyspace::role_permissions()->id();
-}
-
-} // namespace auth

auth/cache.hh

@@ -1,65 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#pragma once
-
-#include <seastar/core/abort_source.hh>
-#include <unordered_set>
-#include <unordered_map>
-
-#include <seastar/core/sstring.hh>
-#include <seastar/core/future.hh>
-#include <seastar/core/sharded.hh>
-#include <seastar/core/shared_ptr.hh>
-#include <seastar/core/semaphore.hh>
-
-#include <absl/container/flat_hash_map.h>
-
-#include "auth/permission.hh"
-#include "auth/common.hh"
-
-namespace cql3 { class query_processor; }
-
-namespace auth {
-
-class cache : public peering_sharded_service<cache> {
-public:
-    using role_name_t = sstring;
-    using version_tag_t = char;
-
-	struct role_record {
-        bool can_login = false;
-        bool is_superuser = false;
-        std::unordered_set<role_name_t> member_of;
-        std::unordered_set<role_name_t> members;
-        sstring salted_hash;
-        std::unordered_map<sstring, sstring> attributes;
-        std::unordered_map<sstring, permission_set> permissions;
-        version_tag_t version; // used for seamless cache reloads
-    };
-
-    explicit cache(cql3::query_processor& qp, abort_source& as) noexcept;
-    lw_shared_ptr<const role_record> get(const role_name_t& role) const noexcept;
-    future<> load_all();
-    future<> load_roles(std::unordered_set<role_name_t> roles);
-    static bool includes_table(const table_id&) noexcept;
-
-private:
-    using roles_map = absl::flat_hash_map<role_name_t, lw_shared_ptr<role_record>>;
-    roles_map _roles;
-    version_tag_t _current_version;
-    cql3::query_processor& _qp;
-    semaphore _loading_sem;
-    abort_source& _as;
-
-    future<lw_shared_ptr<role_record>> fetch_role(const role_name_t& role) const;
-    future<> prune_all() noexcept;
-    future<> distribute_role(const role_name_t& name, const lw_shared_ptr<role_record> role);
-};
-
-} // namespace auth

auth/certificate_authenticator.cc

@@ -8,7 +8,6 @@
  */
 
 #include "auth/certificate_authenticator.hh"
-#include "auth/cache.hh"
 
 #include <boost/regex.hpp>
 #include <fmt/ranges.h>
@@ -35,13 +34,13 @@ static const class_registrator<auth::authenticator
     , cql3::query_processor&
     , ::service::raft_group0_client&
     , ::service::migration_manager&
-    , auth::cache&> cert_auth_reg(CERT_AUTH_NAME);
+    , utils::alien_worker&> cert_auth_reg(CERT_AUTH_NAME);
 
 enum class auth::certificate_authenticator::query_source {
     subject, altname
 };
 
-auth::certificate_authenticator::certificate_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, auth::cache&)
+auth::certificate_authenticator::certificate_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&)
     : _queries([&] {
         auto& conf = qp.db().get_config();
         auto queries = conf.auth_certificate_role_queries();
@@ -76,9 +75,9 @@ auth::certificate_authenticator::certificate_authenticator(cql3::query_processor
                         throw std::invalid_argument(fmt::format("Invalid source: {}", map.at(cfg_source_attr)));
                     }
                     continue;
-                } catch (const std::out_of_range&) {
+                } catch (std::out_of_range&) {
                     // just fallthrough
-                } catch (const boost::regex_error&) {
+                } catch (boost::regex_error&) {
                     std::throw_with_nested(std::invalid_argument(fmt::format("Invalid query expression: {}", map.at(cfg_query_attr))));
                 }
             }

auth/certificate_authenticator.hh

@@ -10,6 +10,7 @@
 #pragma once
 
 #include "auth/authenticator.hh"
+#include "utils/alien_worker.hh"
 #include <boost/regex_fwd.hpp>  // IWYU pragma: keep
 
 namespace cql3 {
@@ -25,15 +26,13 @@ class raft_group0_client;
 
 namespace auth {
 
-class cache;
-
 extern const std::string_view certificate_authenticator_name;
 
 class certificate_authenticator : public authenticator {
     enum class query_source;
     std::vector<std::pair<query_source, boost::regex>> _queries;
 public:
-    certificate_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&);
+    certificate_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&);
     ~certificate_authenticator();
 
     future<> start() override;

auth/common.cc

@@ -94,7 +94,7 @@ static future<> create_legacy_metadata_table_if_missing_impl(
         try {
             co_return co_await mm.announce(co_await ::service::prepare_new_column_family_announcement(qp.proxy(), table, ts),
                     std::move(group0_guard), format("auth: create {} metadata table", table->cf_name()));
-        } catch (const exceptions::already_exists_exception&) {}
+        } catch (exceptions::already_exists_exception&) {}
     }
 }
 

auth/common.hh

@@ -48,10 +48,6 @@ extern constinit const std::string_view AUTH_PACKAGE_NAME;
 
 } // namespace meta
 
-constexpr std::string_view PERMISSIONS_CF = "role_permissions";
-constexpr std::string_view ROLE_MEMBERS_CF = "role_members";
-constexpr std::string_view ROLE_ATTRIBUTES_CF = "role_attributes";
-
 // This is a helper to check whether auth-v2 is on.
 bool legacy_mode(cql3::query_processor& qp);
 

auth/default_authorizer.cc

@@ -37,6 +37,7 @@ std::string_view default_authorizer::qualified_java_name() const {
 static constexpr std::string_view ROLE_NAME = "role";
 static constexpr std::string_view RESOURCE_NAME = "resource";
 static constexpr std::string_view PERMISSIONS_NAME = "permissions";
+static constexpr std::string_view PERMISSIONS_CF = "role_permissions";
 
 static logging::logger alogger("default_authorizer");
 
@@ -256,7 +257,7 @@ future<> default_authorizer::revoke_all(std::string_view role_name, ::service::g
         } else {
             co_await collect_mutations(_qp, mc, query, {sstring(role_name)});
         }
-    } catch (const exceptions::request_execution_exception& e) {
+    } catch (exceptions::request_execution_exception& e) {
         alogger.warn("CassandraAuthorizer failed to revoke all permissions of {}: {}", role_name, e);
     }
 }
@@ -293,13 +294,13 @@ future<> default_authorizer::revoke_all_legacy(const resource& resource) {
                                 [resource](auto ep) {
                     try {
                         std::rethrow_exception(ep);
-                    } catch (const exceptions::request_execution_exception& e) {
+                    } catch (exceptions::request_execution_exception& e) {
                         alogger.warn("CassandraAuthorizer failed to revoke all permissions on {}: {}", resource, e);
                     }
 
                 });
             });
-        } catch (const exceptions::request_execution_exception& e) {
+        } catch (exceptions::request_execution_exception& e) {
             alogger.warn("CassandraAuthorizer failed to revoke all permissions on {}: {}", resource, e);
             return make_ready_future();
         }

auth/ldap_role_manager.cc

@@ -83,18 +83,17 @@ static const class_registrator<
     ldap_role_manager,
     cql3::query_processor&,
     ::service::raft_group0_client&,
-    ::service::migration_manager&,
-    cache&> registration(ldap_role_manager_full_name);
+    ::service::migration_manager&> registration(ldap_role_manager_full_name);
 
 ldap_role_manager::ldap_role_manager(
         std::string_view query_template, std::string_view target_attr, std::string_view bind_name, std::string_view bind_password,
-        cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm, cache& cache)
-        : _std_mgr(qp, rg0c, mm, cache), _group0_client(rg0c), _query_template(query_template), _target_attr(target_attr), _bind_name(bind_name)
+        cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm)
+        : _std_mgr(qp, rg0c, mm), _group0_client(rg0c), _query_template(query_template), _target_attr(target_attr), _bind_name(bind_name)
         , _bind_password(bind_password)
         , _connection_factory(bind(std::mem_fn(&ldap_role_manager::reconnect), std::ref(*this))) {
 }
 
-ldap_role_manager::ldap_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm, cache& cache)
+ldap_role_manager::ldap_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm)
     : ldap_role_manager(
             qp.db().get_config().ldap_url_template(),
             qp.db().get_config().ldap_attr_role(),
@@ -102,8 +101,7 @@ ldap_role_manager::ldap_role_manager(cql3::query_processor& qp, ::service::raft_
             qp.db().get_config().ldap_bind_passwd(),
             qp,
             rg0c,
-            mm,
-            cache) {
+            mm) {
 }
 
 std::string_view ldap_role_manager::qualified_java_name() const noexcept {

auth/ldap_role_manager.hh

@@ -14,7 +14,6 @@
 
 #include "ent/ldap/ldap_connection.hh"
 #include "standard_role_manager.hh"
-#include "auth/cache.hh"
 
 namespace auth {
 
@@ -44,13 +43,12 @@ class ldap_role_manager : public role_manager {
             std::string_view bind_password, ///< LDAP bind credentials.
             cql3::query_processor& qp, ///< Passed to standard_role_manager.
             ::service::raft_group0_client& rg0c, ///< Passed to standard_role_manager.
-            ::service::migration_manager& mm, ///< Passed to standard_role_manager.
-            cache& cache ///< Passed to standard_role_manager.
+            ::service::migration_manager& mm ///< Passed to standard_role_manager.
     );
 
     /// Retrieves LDAP configuration entries from qp and invokes the other constructor.  Required by
     /// class_registrator<role_manager>.
-    ldap_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm, cache& cache);
+    ldap_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm);
 
     /// Thrown when query-template parsing fails.
     struct url_error : public std::runtime_error {

auth/maintenance_socket_role_manager.cc

@@ -11,7 +11,6 @@
 #include <seastar/core/future.hh>
 #include <stdexcept>
 #include <string_view>
-#include "auth/cache.hh"
 #include "cql3/description.hh"
 #include "utils/class_registrator.hh"
 
@@ -24,8 +23,7 @@ static const class_registrator<
         maintenance_socket_role_manager,
         cql3::query_processor&,
         ::service::raft_group0_client&,
-        ::service::migration_manager&,
-        cache&> registration(sstring{maintenance_socket_role_manager_name});
+        ::service::migration_manager&> registration(sstring{maintenance_socket_role_manager_name});
 
 
 std::string_view maintenance_socket_role_manager::qualified_java_name() const noexcept {

auth/maintenance_socket_role_manager.hh

@@ -8,7 +8,6 @@
 
 #pragma once
 
-#include "auth/cache.hh"
 #include "auth/resource.hh"
 #include "auth/role_manager.hh"
 #include <seastar/core/future.hh>
@@ -30,7 +29,7 @@ extern const std::string_view maintenance_socket_role_manager_name;
 // system_auth keyspace, which may be not yet created when the maintenance socket starts listening.
 class maintenance_socket_role_manager final : public role_manager {
 public:
-    maintenance_socket_role_manager(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&) {}
+    maintenance_socket_role_manager(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&) {}
 
     virtual std::string_view qualified_java_name() const noexcept override;
 

auth/password_authenticator.cc

@@ -49,7 +49,7 @@ static const class_registrator<
         cql3::query_processor&,
         ::service::raft_group0_client&,
         ::service::migration_manager&,
-        cache&> password_auth_reg("org.apache.cassandra.auth.PasswordAuthenticator");
+        utils::alien_worker&> password_auth_reg("org.apache.cassandra.auth.PasswordAuthenticator");
 
 static thread_local auto rng_for_salt = std::default_random_engine(std::random_device{}());
 
@@ -63,13 +63,13 @@ std::string password_authenticator::default_superuser(const db::config& cfg) {
 password_authenticator::~password_authenticator() {
 }
 
-password_authenticator::password_authenticator(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, cache& cache)
+password_authenticator::password_authenticator(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, utils::alien_worker& hashing_worker)
     : _qp(qp)
     , _group0_client(g0)
     , _migration_manager(mm)
-    , _cache(cache)
     , _stopped(make_ready_future<>()) 
     , _superuser(default_superuser(qp.db().get_config()))
+    , _hashing_worker(hashing_worker)
 {}
 
 static bool has_salted_hash(const cql3::untyped_result_set_row& row) {
@@ -315,31 +315,24 @@ future<authenticated_user> password_authenticator::authenticate(
     const sstring password = credentials.at(PASSWORD_KEY);
 
     try {
-        std::optional<sstring> salted_hash;
-        if (legacy_mode(_qp)) {
-            salted_hash = co_await get_password_hash(username);
-            if (!salted_hash) {
-                throw exceptions::authentication_exception("Username and/or password are incorrect");
-            }
-        } else {
-            auto role = _cache.get(username);
-            if (!role || role->salted_hash.empty()) {
-                throw exceptions::authentication_exception("Username and/or password are incorrect");
-            }
-            salted_hash = role->salted_hash;
+        const std::optional<sstring> salted_hash = co_await get_password_hash(username);
+        if (!salted_hash) {
+            throw exceptions::authentication_exception("Username and/or password are incorrect");
         }
-        const bool password_match = co_await passwords::check(password, *salted_hash);
+        const bool password_match = co_await _hashing_worker.submit<bool>([password = std::move(password), salted_hash = std::move(salted_hash)]{
+            return passwords::check(password, *salted_hash);
+        });
         if (!password_match) {
             throw exceptions::authentication_exception("Username and/or password are incorrect");
         }
         co_return username;
-    } catch (const std::system_error &) {
+    } catch (std::system_error &) {
         std::throw_with_nested(exceptions::authentication_exception("Could not verify password"));
-    } catch (const exceptions::request_execution_exception& e) {
+    } catch (exceptions::request_execution_exception& e) {
         std::throw_with_nested(exceptions::authentication_exception(e.what()));
-    } catch (const exceptions::authentication_exception& e) {
+    } catch (exceptions::authentication_exception& e) {
         std::throw_with_nested(e);
-    } catch (const exceptions::unavailable_exception& e) {
+    } catch (exceptions::unavailable_exception& e) {
         std::throw_with_nested(exceptions::authentication_exception(e.get_message()));
     } catch (...) {
         std::throw_with_nested(exceptions::authentication_exception("authentication failed"));

auth/password_authenticator.hh

@@ -16,8 +16,8 @@
 #include "db/consistency_level_type.hh"
 #include "auth/authenticator.hh"
 #include "auth/passwords.hh"
-#include "auth/cache.hh"
 #include "service/raft/raft_group0_client.hh"
+#include "utils/alien_worker.hh"
 
 namespace db {
     class config;
@@ -41,19 +41,19 @@ class password_authenticator : public authenticator {
     cql3::query_processor& _qp;
     ::service::raft_group0_client& _group0_client;
     ::service::migration_manager& _migration_manager;
-    cache& _cache;
     future<> _stopped;
     abort_source _as;
     std::string _superuser; // default superuser name from the config (may or may not be present in roles table)
     shared_promise<> _superuser_created_promise;
     // We used to also support bcrypt, SHA-256, and MD5 (ref. scylladb#24524).
     constexpr static auth::passwords::scheme _scheme = passwords::scheme::sha_512;
+    utils::alien_worker& _hashing_worker;
 
 public:
     static db::consistency_level consistency_for_user(std::string_view role_name);
     static std::string default_superuser(const db::config&);
 
-    password_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&);
+    password_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&);
 
     ~password_authenticator();
 

auth/passwords.cc

@@ -7,8 +7,6 @@
  */
 
 #include "auth/passwords.hh"
-#include "utils/crypt_sha512.hh"
-#include <seastar/core/coroutine.hh>
 
 #include <cerrno>
 
@@ -23,46 +21,25 @@ static thread_local crypt_data tlcrypt = {};
 
 namespace detail {
 
-void verify_hashing_output(const char * res) {
-    if (!res || (res[0] == '*')) {
-        throw std::system_error(errno, std::system_category());
-    }
-}
-
 void verify_scheme(scheme scheme) {
     const sstring random_part_of_salt = "aaaabbbbccccdddd";
 
     const sstring salt = sstring(prefix_for_scheme(scheme)) + random_part_of_salt;
     const char* e = crypt_r("fisk", salt.c_str(), &tlcrypt);
-    try {
-        verify_hashing_output(e);
-    } catch (const std::system_error& ex) {
-        throw no_supported_schemes();
+
+    if (e && (e[0] != '*')) {
+        return;
     }
+
+    throw no_supported_schemes();
 }
 
 sstring hash_with_salt(const sstring& pass, const sstring& salt) {
     auto res = crypt_r(pass.c_str(), salt.c_str(), &tlcrypt);
-    verify_hashing_output(res);
-    return res;
-}
-
-seastar::future<sstring> hash_with_salt_async(const sstring& pass, const sstring& salt) {
-    sstring res;
-    // Only SHA-512 hashes for passphrases shorter than 256 bytes can be computed using
-    // the __crypt_sha512 method. For other computations, we fall back to the
-    // crypt_r implementation from `<crypt.h>`, which can stall.
-    if (salt.starts_with(prefix_for_scheme(scheme::sha_512)) && pass.size() <= 255) {
-        char buf[128];
-        const char * output_ptr = co_await __crypt_sha512(pass.c_str(), salt.c_str(), buf);
-        verify_hashing_output(output_ptr);
-        res = output_ptr;
-    } else {
-        const char * output_ptr = crypt_r(pass.c_str(), salt.c_str(), &tlcrypt);
-        verify_hashing_output(output_ptr);
-        res = output_ptr;
+    if (!res || (res[0] == '*')) {
+        throw std::system_error(errno, std::system_category());
     }
-    co_return res;
+    return res;
 }
 
 std::string_view prefix_for_scheme(scheme c) noexcept {
@@ -81,9 +58,8 @@ no_supported_schemes::no_supported_schemes()
         : std::runtime_error("No allowed hashing schemes are supported on this system") {
 }
 
-seastar::future<bool> check(const sstring& pass, const sstring& salted_hash) {
-    const auto pwd_hash = co_await detail::hash_with_salt_async(pass, salted_hash);
-    co_return pwd_hash == salted_hash;
+bool check(const sstring& pass, const sstring& salted_hash) {
+    return detail::hash_with_salt(pass, salted_hash) == salted_hash;
 }
 
 } // namespace auth::passwords

auth/passwords.hh

@@ -11,7 +11,6 @@
 #include <random>
 #include <stdexcept>
 
-#include <seastar/core/future.hh>
 #include <seastar/core/sstring.hh>
 
 #include "seastarx.hh"
@@ -76,23 +75,11 @@ sstring generate_salt(RandomNumberEngine& g, scheme scheme) {
 
 ///
 /// Hash a password combined with an implementation-specific salt string.
-/// Deprecated in favor of `hash_with_salt_async`. This function is still used
-/// when generating password hashes for storage to ensure that
-/// `hash_with_salt` and `hash_with_salt_async` produce identical results,
-/// preserving backward compatibility.
 ///
 /// \throws \ref std::system_error when an unexpected implementation-specific error occurs.
 ///
 sstring hash_with_salt(const sstring& pass, const sstring& salt);
 
-///
-/// Async version of `hash_with_salt` that returns a future.
-/// If possible, hashing uses `coroutine::maybe_yield` to prevent reactor stalls.
-///
-/// \throws \ref std::system_error when an unexpected implementation-specific error occurs.
-///
-seastar::future<sstring> hash_with_salt_async(const sstring& pass, const sstring& salt);
-
 } // namespace detail
 
 ///
@@ -120,6 +107,6 @@ sstring hash(const sstring& pass, RandomNumberEngine& g, scheme scheme) {
 ///
 /// \throws \ref std::system_error when an unexpected implementation-specific error occurs.
 ///
-seastar::future<bool> check(const sstring& pass, const sstring& salted_hash);
+bool check(const sstring& pass, const sstring& salted_hash);
 
 } // namespace auth::passwords

auth/saslauthd_authenticator.cc

@@ -35,9 +35,9 @@ static const class_registrator<
         cql3::query_processor&,
         ::service::raft_group0_client&,
         ::service::migration_manager&,
-        cache&> saslauthd_auth_reg("com.scylladb.auth.SaslauthdAuthenticator");
+        utils::alien_worker&> saslauthd_auth_reg("com.scylladb.auth.SaslauthdAuthenticator");
 
-saslauthd_authenticator::saslauthd_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, cache&)
+saslauthd_authenticator::saslauthd_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&)
     : _socket_path(qp.db().get_config().saslauthd_socket_path())
 {}
 

auth/saslauthd_authenticator.hh

@@ -11,7 +11,7 @@
 #pragma once
 
 #include "auth/authenticator.hh"
-#include "auth/cache.hh"
+#include "utils/alien_worker.hh"
 
 namespace cql3 {
 class query_processor;
@@ -29,7 +29,7 @@ namespace auth {
 class saslauthd_authenticator : public authenticator {
     sstring _socket_path; ///< Path to the domain socket on which saslauthd is listening.
 public:
-    saslauthd_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&);
+    saslauthd_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&);
 
     future<> start() override;
 

auth/service.cc

@@ -17,7 +17,6 @@
 #include <chrono>
 
 #include <seastar/core/future-util.hh>
-#include <seastar/core/shard_id.hh>
 #include <seastar/core/sharded.hh>
 #include <seastar/core/shared_ptr.hh>
 
@@ -158,7 +157,6 @@ static future<> validate_role_exists(const service& ser, std::string_view role_n
 
 service::service(
         utils::loading_cache_config c,
-        cache& cache,
         cql3::query_processor& qp,
         ::service::raft_group0_client& g0,
         ::service::migration_notifier& mn,
@@ -168,7 +166,6 @@ service::service(
         maintenance_socket_enabled used_by_maintenance_socket)
             : _loading_cache_config(std::move(c))
             , _permissions_cache(nullptr)
-            , _cache(cache)
             , _qp(qp)
             , _group0_client(g0)
             , _mnotifier(mn)
@@ -191,16 +188,15 @@ service::service(
         ::service::migration_manager& mm,
         const service_config& sc,
         maintenance_socket_enabled used_by_maintenance_socket,
-        cache& cache)
+        utils::alien_worker& hashing_worker)
             : service(
                       std::move(c),
-                      cache,
                       qp,
                       g0,
                       mn,
                       create_object<authorizer>(sc.authorizer_java_name, qp, g0, mm),
-                      create_object<authenticator>(sc.authenticator_java_name, qp, g0, mm, cache),
-                      create_object<role_manager>(sc.role_manager_java_name, qp, g0, mm, cache),
+                      create_object<authenticator>(sc.authenticator_java_name, qp, g0, mm, hashing_worker),
+                      create_object<role_manager>(sc.role_manager_java_name, qp, g0, mm),
                       used_by_maintenance_socket) {
 }
 
@@ -225,7 +221,7 @@ future<> service::create_legacy_keyspace_if_missing(::service::migration_manager
             try {
                 co_return co_await mm.announce(::service::prepare_new_keyspace_announcement(db.real_database(), ksm, ts),
                         std::move(group0_guard), seastar::format("auth_service: create {} keyspace", meta::legacy::AUTH_KS));
-            } catch (const ::service::group0_concurrent_modification&) {
+            } catch (::service::group0_concurrent_modification&) {
                 log.info("Concurrent operation is detected while creating {} keyspace, retrying.", meta::legacy::AUTH_KS);
             }
         }
@@ -236,9 +232,6 @@ future<> service::start(::service::migration_manager& mm, db::system_keyspace& s
     auto auth_version = co_await sys_ks.get_auth_version();
     // version is set in query processor to be easily available in various places we call auth::legacy_mode check.
     _qp.auth_version = auth_version;
-    if (this_shard_id() == 0) {
-        co_await _cache.load_all();
-    }
     if (!_used_by_maintenance_socket) {
         // this legacy keyspace is only used by cqlsh
         // it's needed when executing `list roles` or `list users`

auth/service.hh

@@ -21,12 +21,12 @@
 #include "auth/authorizer.hh"
 #include "auth/permission.hh"
 #include "auth/permissions_cache.hh"
-#include "auth/cache.hh"
 #include "auth/role_manager.hh"
 #include "auth/common.hh"
 #include "cql3/description.hh"
 #include "seastarx.hh"
 #include "service/raft/raft_group0_client.hh"
+#include "utils/alien_worker.hh"
 #include "utils/observable.hh"
 #include "utils/serialized_action.hh"
 #include "service/maintenance_mode.hh"
@@ -77,7 +77,6 @@ public:
 class service final : public seastar::peering_sharded_service<service> {
     utils::loading_cache_config _loading_cache_config;
     std::unique_ptr<permissions_cache> _permissions_cache;
-    cache& _cache;
 
     cql3::query_processor& _qp;
 
@@ -108,7 +107,6 @@ class service final : public seastar::peering_sharded_service<service> {
 public:
     service(
             utils::loading_cache_config,
-            cache& cache,
             cql3::query_processor&,
             ::service::raft_group0_client&,
             ::service::migration_notifier&,
@@ -130,7 +128,7 @@ public:
             ::service::migration_manager&,
             const service_config&,
             maintenance_socket_enabled,
-            cache&);
+            utils::alien_worker&);
 
     future<> start(::service::migration_manager&, db::system_keyspace&);
 

auth/standard_role_manager.cc

@@ -41,6 +41,21 @@
 
 namespace auth {
 
+namespace meta {
+
+namespace role_members_table {
+
+constexpr std::string_view name{"role_members" , 12};
+
+}
+
+namespace role_attributes_table {
+
+constexpr std::string_view name{"role_attributes", 15};
+
+}
+
+}
 
 static logging::logger log("standard_role_manager");
 
@@ -49,8 +64,7 @@ static const class_registrator<
         standard_role_manager,
         cql3::query_processor&,
         ::service::raft_group0_client&,
-        ::service::migration_manager&,
-        cache&> registration("org.apache.cassandra.auth.CassandraRoleManager");
+        ::service::migration_manager&> registration("org.apache.cassandra.auth.CassandraRoleManager");
 
 struct record final {
     sstring name;
@@ -107,11 +121,10 @@ static bool has_can_login(const cql3::untyped_result_set_row& row) {
     return row.has("can_login") && !(boolean_type->deserialize(row.get_blob_unfragmented("can_login")).is_null());
 }
 
-standard_role_manager::standard_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, cache& cache)
+standard_role_manager::standard_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm)
     : _qp(qp)
     , _group0_client(g0)
     , _migration_manager(mm)
-    , _cache(cache)
     , _stopped(make_ready_future<>())
     , _superuser(password_authenticator::default_superuser(qp.db().get_config()))
 {}
@@ -123,7 +136,7 @@ std::string_view standard_role_manager::qualified_java_name() const noexcept {
 const resource_set& standard_role_manager::protected_resources() const {
     static const resource_set resources({
             make_data_resource(meta::legacy::AUTH_KS, meta::roles_table::name),
-            make_data_resource(meta::legacy::AUTH_KS, ROLE_MEMBERS_CF)});
+            make_data_resource(meta::legacy::AUTH_KS, meta::role_members_table::name)});
 
     return resources;
 }
@@ -147,7 +160,7 @@ future<> standard_role_manager::create_legacy_metadata_tables_if_missing() const
             "  PRIMARY KEY (role, member)"
             ")",
             meta::legacy::AUTH_KS,
-            ROLE_MEMBERS_CF);
+            meta::role_members_table::name);
     static const sstring create_role_attributes_query = seastar::format(
             "CREATE TABLE {}.{} ("
             "  role text,"
@@ -156,7 +169,7 @@ future<> standard_role_manager::create_legacy_metadata_tables_if_missing() const
             "  PRIMARY KEY(role, name)"
             ")",
             meta::legacy::AUTH_KS,
-            ROLE_ATTRIBUTES_CF);
+            meta::role_attributes_table::name);
     return when_all_succeed(
             create_legacy_metadata_table_if_missing(
                     meta::roles_table::name,
@@ -164,12 +177,12 @@ future<> standard_role_manager::create_legacy_metadata_tables_if_missing() const
                     create_roles_query,
                     _migration_manager),
             create_legacy_metadata_table_if_missing(
-                    ROLE_MEMBERS_CF,
+                    meta::role_members_table::name,
                     _qp,
                     create_role_members_query,
                     _migration_manager),
             create_legacy_metadata_table_if_missing(
-                    ROLE_ATTRIBUTES_CF,
+                    meta::role_attributes_table::name,
                     _qp,
                     create_role_attributes_query,
                     _migration_manager)).discard_result();
@@ -192,7 +205,7 @@ future<> standard_role_manager::legacy_create_default_role_if_missing() {
                 {_superuser},
                 cql3::query_processor::cache_internal::no).discard_result();
         log.info("Created default superuser role '{}'.", _superuser);
-    } catch (const exceptions::unavailable_exception& e) {
+    } catch(const exceptions::unavailable_exception& e) {
         log.warn("Skipped default role setup: some nodes were not ready; will retry");
         throw e;
     }
@@ -416,7 +429,7 @@ future<> standard_role_manager::drop(std::string_view role_name, ::service::grou
     const auto revoke_from_members = [this, role_name, &mc] () -> future<> {
         const sstring query = seastar::format("SELECT member FROM {}.{} WHERE role = ?",
                 get_auth_ks_name(_qp),
-                ROLE_MEMBERS_CF);
+                meta::role_members_table::name);
         const auto members = co_await _qp.execute_internal(
                 query,
                 consistency_for_role(role_name),
@@ -448,7 +461,7 @@ future<> standard_role_manager::drop(std::string_view role_name, ::service::grou
     const auto remove_attributes_of = [this, role_name, &mc] () -> future<> {
         const sstring query = seastar::format("DELETE FROM {}.{} WHERE role = ?",
                 get_auth_ks_name(_qp),
-                ROLE_ATTRIBUTES_CF);
+                meta::role_attributes_table::name);
         if (legacy_mode(_qp)) {
             co_await _qp.execute_internal(query, {sstring(role_name)},
                 cql3::query_processor::cache_internal::yes).discard_result();
@@ -504,7 +517,7 @@ standard_role_manager::legacy_modify_membership(
             case membership_change::add: {
                 const sstring insert_query = seastar::format("INSERT INTO {}.{} (role, member) VALUES (?, ?)",
                         get_auth_ks_name(_qp),
-                        ROLE_MEMBERS_CF);
+                        meta::role_members_table::name);
                 co_return co_await _qp.execute_internal(
                         insert_query,
                         consistency_for_role(role_name),
@@ -516,7 +529,7 @@ standard_role_manager::legacy_modify_membership(
             case membership_change::remove: {
                 const sstring delete_query = seastar::format("DELETE FROM {}.{} WHERE role = ? AND member = ?",
                         get_auth_ks_name(_qp),
-                        ROLE_MEMBERS_CF);
+                        meta::role_members_table::name);
                 co_return co_await _qp.execute_internal(
                         delete_query,
                         consistency_for_role(role_name),
@@ -554,12 +567,12 @@ standard_role_manager::modify_membership(
     case membership_change::add:
         modify_role_members = seastar::format("INSERT INTO {}.{} (role, member) VALUES (?, ?)",
                 get_auth_ks_name(_qp),
-                ROLE_MEMBERS_CF);
+                meta::role_members_table::name);
         break;
     case membership_change::remove:
         modify_role_members = seastar::format("DELETE FROM {}.{} WHERE role = ? AND member = ?",
                 get_auth_ks_name(_qp),
-                ROLE_MEMBERS_CF);
+                meta::role_members_table::name);
         break;
     default:
         on_internal_error(log, format("unknown membership_change value: {}", int(ch)));
@@ -653,7 +666,7 @@ future<role_set> standard_role_manager::query_granted(std::string_view grantee_n
 future<role_to_directly_granted_map> standard_role_manager::query_all_directly_granted(::service::query_state& qs) {
     const sstring query = seastar::format("SELECT * FROM {}.{}",
             get_auth_ks_name(_qp),
-            ROLE_MEMBERS_CF);
+            meta::role_members_table::name);
 
     const auto results = co_await _qp.execute_internal(
             query,
@@ -718,21 +731,15 @@ future<bool> standard_role_manager::is_superuser(std::string_view role_name) {
 }
 
 future<bool> standard_role_manager::can_login(std::string_view role_name) {
-    if (legacy_mode(_qp)) {
-       const auto r = co_await require_record(_qp, role_name);
-       co_return r.can_login;
-    }
-    auto role = _cache.get(sstring(role_name));
-    if (!role) {
-        throw nonexistant_role(role_name);
-    }
-    co_return role->can_login;
+    return require_record(_qp, role_name).then([](record r) {
+        return r.can_login;
+    });
 }
 
 future<std::optional<sstring>> standard_role_manager::get_attribute(std::string_view role_name, std::string_view attribute_name, ::service::query_state& qs) {
     const sstring query = seastar::format("SELECT name, value FROM {}.{} WHERE role = ? AND name = ?",
             get_auth_ks_name(_qp),
-            ROLE_ATTRIBUTES_CF);
+            meta::role_attributes_table::name);
     const auto result_set = co_await _qp.execute_internal(query, db::consistency_level::ONE, qs, {sstring(role_name), sstring(attribute_name)}, cql3::query_processor::cache_internal::yes);
     if (!result_set->empty()) {
         const cql3::untyped_result_set_row &row = result_set->one();
@@ -763,7 +770,7 @@ future<> standard_role_manager::set_attribute(std::string_view role_name, std::s
     }
     const sstring query = seastar::format("INSERT INTO {}.{} (role, name, value)  VALUES (?, ?, ?)",
             get_auth_ks_name(_qp),
-            ROLE_ATTRIBUTES_CF);
+            meta::role_attributes_table::name);
     if (legacy_mode(_qp)) {
         co_await _qp.execute_internal(query, {sstring(role_name), sstring(attribute_name), sstring(attribute_value)}, cql3::query_processor::cache_internal::yes).discard_result();
     } else {
@@ -778,7 +785,7 @@ future<> standard_role_manager::remove_attribute(std::string_view role_name, std
     }
     const sstring query = seastar::format("DELETE FROM {}.{} WHERE role = ? AND name = ?",
             get_auth_ks_name(_qp),
-            ROLE_ATTRIBUTES_CF);
+            meta::role_attributes_table::name);
     if (legacy_mode(_qp)) {
         co_await _qp.execute_internal(query, {sstring(role_name), sstring(attribute_name)}, cql3::query_processor::cache_internal::yes).discard_result();
     } else {

auth/standard_role_manager.hh

@@ -10,7 +10,6 @@
 
 #include "auth/common.hh"
 #include "auth/role_manager.hh"
-#include "auth/cache.hh"
 
 #include <string_view>
 
@@ -37,14 +36,13 @@ class standard_role_manager final : public role_manager {
     cql3::query_processor& _qp;
     ::service::raft_group0_client& _group0_client;
     ::service::migration_manager& _migration_manager;
-    cache& _cache;
     future<> _stopped;
     abort_source _as;
     std::string _superuser;
     shared_promise<> _superuser_created_promise;
 
 public:
-    standard_role_manager(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&);
+    standard_role_manager(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&);
 
     virtual std::string_view qualified_java_name() const noexcept override;
 

auth/transitional.cc

@@ -13,7 +13,6 @@
 #include "auth/authorizer.hh"
 #include "auth/default_authorizer.hh"
 #include "auth/password_authenticator.hh"
-#include "auth/cache.hh"
 #include "auth/permission.hh"
 #include "service/raft/raft_group0_client.hh"
 #include "utils/class_registrator.hh"
@@ -38,8 +37,8 @@ class transitional_authenticator : public authenticator {
 public:
     static const sstring PASSWORD_AUTHENTICATOR_NAME;
 
-    transitional_authenticator(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, cache& cache)
-            : transitional_authenticator(std::make_unique<password_authenticator>(qp, g0, mm, cache)) {
+    transitional_authenticator(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, utils::alien_worker& hashing_worker)
+            : transitional_authenticator(std::make_unique<password_authenticator>(qp, g0, mm, hashing_worker)) {
     }
     transitional_authenticator(std::unique_ptr<authenticator> a)
             : _authenticator(std::move(a)) {
@@ -81,7 +80,7 @@ public:
         }).handle_exception([](auto ep) {
             try {
                 std::rethrow_exception(ep);
-            } catch (const exceptions::authentication_exception&) {
+            } catch (exceptions::authentication_exception&) {
                 // return anon user
                 return make_ready_future<authenticated_user>(anonymous_user());
             }
@@ -126,7 +125,7 @@ public:
             virtual bytes evaluate_response(bytes_view client_response) override {
                 try {
                     return _sasl->evaluate_response(client_response);
-                } catch (const exceptions::authentication_exception&) {
+                } catch (exceptions::authentication_exception&) {
                     _complete = true;
                     return {};
                 }
@@ -141,7 +140,7 @@ public:
                     return _sasl->get_authenticated_user().handle_exception([](auto ep) {
                         try {
                             std::rethrow_exception(ep);
-                        } catch (const exceptions::authentication_exception&) {
+                        } catch (exceptions::authentication_exception&) {
                             // return anon user
                             return make_ready_future<authenticated_user>(anonymous_user());
                         }
@@ -241,7 +240,7 @@ static const class_registrator<
         cql3::query_processor&,
         ::service::raft_group0_client&,
         ::service::migration_manager&,
-        auth::cache&> transitional_authenticator_reg(auth::PACKAGE_NAME + "TransitionalAuthenticator");
+        utils::alien_worker&> transitional_authenticator_reg(auth::PACKAGE_NAME + "TransitionalAuthenticator");
 
 static const class_registrator<
         auth::authorizer,

cdc/generation.cc

@@ -204,7 +204,7 @@ future<topology_description> topology_description::clone_async() const {
 
     for (const auto& entry : _entries) {
         vec.push_back(entry);
-        co_await coroutine::maybe_yield();
+        co_await seastar::maybe_yield();
     }
 
     co_return topology_description{std::move(vec)};

cdc/split.cc

@@ -15,7 +15,7 @@
 #include "mutation/tombstone.hh"
 #include "schema/schema.hh"
 
-#include <seastar/core/sstring.hh>
+#include "seastar/core/sstring.hh"
 #include "types/concrete_types.hh"
 #include "types/types.hh"
 #include "types/user.hh"

client_data.hh

@@ -10,9 +10,7 @@
 #include <seastar/net/inet_address.hh>
 #include <seastar/core/sstring.hh>
 #include "seastarx.hh"
-#include "utils/loading_shared_values.hh"
 
-#include <list>
 #include <optional>
 
 enum class client_type {
@@ -29,20 +27,6 @@ enum class client_connection_stage {
     ready,
 };
 
-// We implement a keys cache using a map-like utils::loading_shared_values container by storing empty values.
-struct options_cache_value_type {};
-using client_options_cache_type = utils::loading_shared_values<sstring, options_cache_value_type>;
-using client_options_cache_entry_type = client_options_cache_type::entry_ptr;
-using client_options_cache_key_type = client_options_cache_type::key_type;
-
-// This struct represents a single OPTION key-value pair from the client's connection options.
-// Both key and value are represented by corresponding "references" to their cached values.
-// Each "reference" is effectively a lw_shared_ptr value.
-struct client_option_key_value_cached_entry {
-    client_options_cache_entry_type key;
-    client_options_cache_entry_type value;
-};
-
 sstring to_string(client_connection_stage ct);
 
 // Representation of a row in `system.clients'. std::optionals are for nullable cells.
@@ -53,8 +37,8 @@ struct client_data {
     client_connection_stage connection_stage = client_connection_stage::established;
     int32_t shard_id;  /// ID of server-side shard which is processing the connection.
 
-    std::optional<client_options_cache_entry_type> driver_name;
-    std::optional<client_options_cache_entry_type> driver_version;
+    std::optional<sstring> driver_name;
+    std::optional<sstring> driver_version;
     std::optional<sstring> hostname;
     std::optional<int32_t> protocol_version;
     std::optional<sstring> ssl_cipher_suite;
@@ -62,7 +46,6 @@ struct client_data {
     std::optional<sstring> ssl_protocol;
     std::optional<sstring> username;
     std::optional<sstring> scheduling_group_name;
-    std::list<client_option_key_value_cached_entry> client_options;
 
     sstring stage_str() const { return to_string(connection_stage); }
     sstring client_type_str() const { return to_string(ct); }

cmake/mode.common.cmake

@@ -125,6 +125,10 @@ if(target_arch)
   add_compile_options("-march=${target_arch}")
 endif()
 
+if(CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
+  add_compile_options("SHELL:-Xclang -fexperimental-assignment-tracking=disabled")
+endif()
+
 function(maybe_limit_stack_usage_in_KB stack_usage_threshold_in_KB config)
   math(EXPR _stack_usage_threshold_in_bytes "${stack_usage_threshold_in_KB} * 1024")
   set(_stack_usage_threshold_flag "-Wstack-usage=${_stack_usage_threshold_in_bytes}")

compaction/compaction_group_view.hh

@@ -12,7 +12,6 @@
 #include <seastar/core/condition-variable.hh>
 
 #include "schema/schema_fwd.hh"
-#include "sstables/open_info.hh"
 #include "compaction_descriptor.hh"
 
 class reader_permit;
@@ -45,7 +44,7 @@ public:
     virtual compaction_strategy_state& get_compaction_strategy_state() noexcept = 0;
     virtual reader_permit make_compaction_reader_permit() const = 0;
     virtual sstables::sstables_manager& get_sstables_manager() noexcept = 0;
-    virtual sstables::shared_sstable make_sstable(sstables::sstable_state) const = 0;
+    virtual sstables::shared_sstable make_sstable() const = 0;
     virtual sstables::sstable_writer_config configure_writer(sstring origin) const = 0;
     virtual api::timestamp_type min_memtable_timestamp() const = 0;
     virtual api::timestamp_type min_memtable_live_timestamp() const = 0;

compaction/compaction_manager.cc

@@ -416,9 +416,7 @@ future<compaction_result> compaction_task_executor::compact_sstables(compaction_
         descriptor.enable_garbage_collection(co_await sstable_set_for_tombstone_gc(t));
     }
     descriptor.creator = [&t] (shard_id) {
-        // All compaction types going through this path will work on normal input sstables only.
-        // Off-strategy, for example, waits until the sstables move out of staging state.
-        return t.make_sstable(sstables::sstable_state::normal);
+        return t.make_sstable();
     };
     descriptor.replacer = [this, &t, &on_replace, offstrategy] (compaction_completion_desc desc) {
         t.get_compaction_strategy().notify_completion(t, desc.old_sstables, desc.new_sstables);
@@ -1849,10 +1847,6 @@ protected:
                 throw make_compaction_stopped_exception();
             }
         }, false);
-        if (utils::get_local_injector().is_enabled("split_sstable_force_stop_exception")) {
-            throw make_compaction_stopped_exception();
-        }
-
         co_return co_await do_rewrite_sstable(std::move(sst));
     }
 };
@@ -2290,16 +2284,12 @@ future<compaction_manager::compaction_stats_opt> compaction_manager::perform_spl
 }
 
 future<std::vector<sstables::shared_sstable>>
-compaction_manager::maybe_split_new_sstable(sstables::shared_sstable sst, compaction_group_view& t, compaction_type_options::split opt) {
+compaction_manager::maybe_split_sstable(sstables::shared_sstable sst, compaction_group_view& t, compaction_type_options::split opt) {
     if (!split_compaction_task_executor::sstable_needs_split(sst, opt)) {
         co_return std::vector<sstables::shared_sstable>{sst};
     }
-    // Throw an error if split cannot be performed due to e.g. out of space prevention.
-    // We don't want to prevent split because compaction is temporarily disabled on a view only for synchronization,
-    // which is unneeded against new sstables that aren't part of any set yet, so never use can_proceed(&t) here.
-    if (is_disabled()) {
-        co_return coroutine::exception(std::make_exception_ptr(std::runtime_error(format("Cannot split {} because manager has compaction disabled, " \
-                                                                                         "reason might be out of space prevention", sst->get_filename()))));
+    if (!can_proceed(&t)) {
+        co_return std::vector<sstables::shared_sstable>{sst};
     }
     std::vector<sstables::shared_sstable> ret;
 
@@ -2307,11 +2297,8 @@ compaction_manager::maybe_split_new_sstable(sstables::shared_sstable sst, compac
     compaction_progress_monitor monitor;
     compaction_data info = create_compaction_data();
     compaction_descriptor desc = split_compaction_task_executor::make_descriptor(sst, opt);
-    desc.creator = [&t, sst] (shard_id _) {
-        // NOTE: preserves the sstable state, since we want the output to be on the same state as the original.
-        // For example, if base table has views, it's important that sstable produced by repair will be
-        // in the staging state.
-        return t.make_sstable(sst->state());
+    desc.creator = [&t] (shard_id _) {
+        return t.make_sstable();
     };
     desc.replacer = [&] (compaction_completion_desc d) {
         std::move(d.new_sstables.begin(), d.new_sstables.end(), std::back_inserter(ret));

compaction/compaction_manager.hh

@@ -376,8 +376,7 @@ public:
     // Splits a single SSTable by segregating all its data according to the classifier.
     // If SSTable doesn't need split, the same input SSTable is returned as output.
     // If SSTable needs split, then output SSTables are returned and the input SSTable is deleted.
-    // Exception is thrown if the input sstable cannot be split due to e.g. out of space prevention.
-    future<std::vector<sstables::shared_sstable>> maybe_split_new_sstable(sstables::shared_sstable sst, compaction_group_view& t, compaction_type_options::split opt);
+    future<std::vector<sstables::shared_sstable>> maybe_split_sstable(sstables::shared_sstable sst, compaction_group_view& t, compaction_type_options::split opt);
 
     // Run a custom job for a given table, defined by a function
     // it completes when future returned by job is ready or returns immediately

conf/scylla.yaml

@@ -571,10 +571,10 @@ commitlog_total_space_in_mb: -1
 #   - "none": auditing is disabled (default)
 #   - "table": save audited events in audit.audit_log column family
 #   - "syslog": send audited events via syslog (depends on OS, but usually to /dev/log)
-audit: "table"
+# audit: "none"
 #
 # List of statement categories that should be audited.
-audit_categories: "DCL,DDL,AUTH,ADMIN"
+# audit_categories: "DCL,DDL,AUTH"
 #
 # List of tables that should be audited.
 # audit_tables: "<keyspace_name>.<table_name>,<keyspace_name>.<table_name>"

configure.py

@@ -368,87 +368,6 @@ def find_ninja():
     sys.exit(1)
 
 
-def find_compiler(name):
-    """
-    Find a compiler by name, skipping ccache wrapper directories.
-
-    This is useful when using sccache to avoid double-caching through ccache.
-
-    Args:
-        name: The compiler name (e.g., 'clang++', 'clang', 'gcc')
-
-    Returns:
-        Path to the compiler, skipping ccache directories, or None if not found.
-    """
-    ccache_dirs = {'/usr/lib/ccache', '/usr/lib64/ccache'}
-    for path_dir in os.environ.get('PATH', '').split(os.pathsep):
-        # Skip ccache wrapper directories
-        if os.path.realpath(path_dir) in ccache_dirs or path_dir in ccache_dirs:
-            continue
-        candidate = os.path.join(path_dir, name)
-        if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
-            return candidate
-    return None
-
-
-def resolve_compilers_for_compiler_cache(args, compiler_cache):
-    """
-    When using a compiler cache, resolve compiler paths to avoid ccache directories.
-
-    This prevents double-caching when ccache symlinks are in PATH.
-
-    Args:
-        args: The argument namespace with cc and cxx attributes.
-        compiler_cache: Path to the compiler cache binary, or None.
-    """
-    if not compiler_cache:
-        return
-    if not os.path.isabs(args.cxx):
-        real_cxx = find_compiler(args.cxx)
-        if real_cxx:
-            args.cxx = real_cxx
-    if not os.path.isabs(args.cc):
-        real_cc = find_compiler(args.cc)
-        if real_cc:
-            args.cc = real_cc
-
-
-def find_compiler_cache(preference):
-    """
-    Find a compiler cache based on the preference.
-
-    Args:
-        preference: One of 'auto', 'sccache', 'ccache', 'none', or a path to a binary.
-
-    Returns:
-        Path to the compiler cache binary, or None if not found/disabled.
-    """
-    if preference == 'none':
-        return None
-
-    if preference == 'auto':
-        # Prefer sccache over ccache
-        for cache in ['sccache', 'ccache']:
-            path = which(cache)
-            if path:
-                return path
-        return None
-
-    if preference in ('sccache', 'ccache'):
-        path = which(preference)
-        if path:
-            return path
-        print(f"Warning: {preference} not found on PATH, disabling compiler cache")
-        return None
-
-    # Assume it's a path to a binary
-    if os.path.isfile(preference) and os.access(preference, os.X_OK):
-        return preference
-
-    print(f"Warning: compiler cache '{preference}' not found or not executable, disabling compiler cache")
-    return None
-
-
 modes = {
     'debug': {
         'cxxflags': '-DDEBUG -DSANITIZE -DDEBUG_LSA_SANITIZER -DSCYLLA_ENABLE_ERROR_INJECTION',
@@ -526,7 +445,6 @@ ldap_tests = set([
 scylla_tests = set([
     'test/boost/combined_tests',
     'test/boost/UUID_test',
-    'test/boost/url_parse_test',
     'test/boost/advanced_rpc_compressor_test',
     'test/boost/allocation_strategy_test',
     'test/boost/alternator_unit_test',
@@ -813,8 +731,6 @@ arg_parser.add_argument('--compiler', action='store', dest='cxx', default='clang
                         help='C++ compiler path')
 arg_parser.add_argument('--c-compiler', action='store', dest='cc', default='clang',
                         help='C compiler path')
-arg_parser.add_argument('--compiler-cache', action='store', dest='compiler_cache', default='auto',
-                        help='Compiler cache to use: auto (default, prefers sccache), sccache, ccache, none, or a path to a binary')
 add_tristate(arg_parser, name='dpdk', dest='dpdk', default=False,
                         help='Use dpdk (from seastar dpdk sources)')
 arg_parser.add_argument('--dpdk-target', action='store', dest='dpdk_target', default='',
@@ -942,7 +858,6 @@ scylla_core = (['message/messaging_service.cc',
                 'utils/alien_worker.cc',
                 'utils/array-search.cc',
                 'utils/base64.cc',
-                'utils/crypt_sha512.cc',
                 'utils/logalloc.cc',
                 'utils/large_bitset.cc',
                 'utils/buffer_input_stream.cc',
@@ -1034,7 +949,6 @@ scylla_core = (['message/messaging_service.cc',
                 'cql3/functions/aggregate_fcts.cc',
                 'cql3/functions/castas_fcts.cc',
                 'cql3/functions/error_injection_fcts.cc',
-                'cql3/functions/vector_similarity_fcts.cc',
                 'cql3/statements/cf_prop_defs.cc',
                 'cql3/statements/cf_statement.cc',
                 'cql3/statements/authentication_statement.cc',
@@ -1092,7 +1006,6 @@ scylla_core = (['message/messaging_service.cc',
                 'cql3/statements/list_service_level_attachments_statement.cc',
                 'cql3/statements/list_effective_service_level_statement.cc',
                 'cql3/statements/describe_statement.cc',
-                'cql3/statements/view_prop_defs.cc',
                 'cql3/update_parameters.cc',
                 'cql3/util.cc',
                 'cql3/ut_name.cc',
@@ -1148,6 +1061,7 @@ scylla_core = (['message/messaging_service.cc',
                 'db/hints/resource_manager.cc',
                 'db/hints/sync_point.cc',
                 'db/large_data_handler.cc',
+                'db/legacy_schema_migrator.cc',
                 'db/marshal/type_parser.cc',
                 'db/per_partition_rate_limit_options.cc',
                 'db/rate_limiter.cc',
@@ -1243,7 +1157,6 @@ scylla_core = (['message/messaging_service.cc',
                 'locator/topology.cc',
                 'locator/util.cc',
                 'service/client_state.cc',
-                'service/client_routes.cc',
                 'service/storage_service.cc',
                 'service/session.cc',
                 'service/task_manager_module.cc',
@@ -1282,7 +1195,6 @@ scylla_core = (['message/messaging_service.cc',
                 'auth/allow_all_authorizer.cc',
                 'auth/authenticated_user.cc',
                 'auth/authenticator.cc',
-                'auth/cache.cc',
                 'auth/common.cc',
                 'auth/default_authorizer.cc',
                 'auth/resource.cc',
@@ -1404,8 +1316,6 @@ api = ['api/api.cc',
        'api/storage_proxy.cc',
        Json2Code('api/api-doc/cache_service.json'),
        'api/cache_service.cc',
-       Json2Code('api/api-doc/client_routes.json'),
-       'api/client_routes.cc',
        Json2Code('api/api-doc/collectd.json'),
        'api/collectd.cc',
        Json2Code('api/api-doc/endpoint_snitch_info.json'),
@@ -1455,7 +1365,6 @@ alternator = [
        'alternator/auth.cc',
        'alternator/streams.cc',
        'alternator/ttl.cc',
-       'alternator/http_compression.cc'
 ]
 
 idls = ['idl/gossip_digest.idl.hh',
@@ -1569,6 +1478,7 @@ deps = {
 
 pure_boost_tests = set([
     'test/boost/anchorless_list_test',
+    'test/boost/auth_passwords_test',
     'test/boost/auth_resource_test',
     'test/boost/big_decimal_test',
     'test/boost/caching_options_test',
@@ -1701,7 +1611,6 @@ deps['test/boost/combined_tests'] += [
     'test/boost/schema_registry_test.cc',
     'test/boost/secondary_index_test.cc',
     'test/boost/sessions_test.cc',
-    'test/boost/simple_value_with_expiry_test.cc',
     'test/boost/sstable_compaction_test.cc',
     'test/boost/sstable_compressor_factory_test.cc',
     'test/boost/sstable_compression_config_test.cc',
@@ -1737,7 +1646,6 @@ deps['test/boost/bytes_ostream_test'] = [
 ]
 deps['test/boost/input_stream_test'] = ['test/boost/input_stream_test.cc']
 deps['test/boost/UUID_test'] = ['clocks-impl.cc', 'utils/UUID_gen.cc', 'test/boost/UUID_test.cc', 'utils/uuid.cc', 'utils/dynamic_bitset.cc', 'utils/hashers.cc', 'utils/on_internal_error.cc']
-deps['test/boost/url_parse_test'] = ['utils/http.cc', 'test/boost/url_parse_test.cc', ]
 deps['test/boost/murmur_hash_test'] = ['bytes.cc', 'utils/murmur_hash.cc', 'test/boost/murmur_hash_test.cc']
 deps['test/boost/allocation_strategy_test'] = ['test/boost/allocation_strategy_test.cc', 'utils/logalloc.cc', 'utils/dynamic_bitset.cc', 'utils/labels.cc']
 deps['test/boost/log_heap_test'] = ['test/boost/log_heap_test.cc']
@@ -1785,18 +1693,6 @@ deps['test/vector_search/vector_store_client_test'] =  ['test/vector_search/vect
 deps['test/vector_search/load_balancer_test'] = ['test/vector_search/load_balancer_test.cc'] + scylla_tests_dependencies
 deps['test/vector_search/client_test'] = ['test/vector_search/client_test.cc'] + scylla_tests_dependencies
 
-boost_tests_prefixes = ["test/boost/", "test/vector_search/", "test/raft/", "test/manual/", "test/ldap/"]
-
-# We need to link these files to all Boost tests to make sure that
-# we can execute `--list_json_content` on them. That will produce
-# a similar result as calling `--list_content={HRF,DOT}`.
-# Unfortunately, to be able to do that, we're forced to link the
-# relevant code by hand.
-for key in deps.keys():
-    for prefix in boost_tests_prefixes:
-        if key.startswith(prefix):
-            deps[key] += ["test/lib/boost_tree_lister_injector.cc", "test/lib/boost_test_tree_lister.cc"]
-
 wasm_deps = {}
 
 wasm_deps['wasm/return_input.wat'] = 'test/resource/wasm/rust/return_input.rs'
@@ -2101,7 +1997,7 @@ def semicolon_separated(*flags):
 def real_relpath(path, start):
     return os.path.relpath(os.path.realpath(path), os.path.realpath(start))
 
-def configure_seastar(build_dir, mode, mode_config, compiler_cache=None):
+def configure_seastar(build_dir, mode, mode_config):
     seastar_cxx_ld_flags = mode_config['cxx_ld_flags']
     # We want to "undo" coverage for seastar if we have it enabled.
     if args.coverage:
@@ -2148,10 +2044,6 @@ def configure_seastar(build_dir, mode, mode_config, compiler_cache=None):
         '-DSeastar_IO_URING=ON',
     ]
 
-    if compiler_cache:
-        seastar_cmake_args += [f'-DCMAKE_CXX_COMPILER_LAUNCHER={compiler_cache}',
-                               f'-DCMAKE_C_COMPILER_LAUNCHER={compiler_cache}']
-
     if args.stack_guards is not None:
         stack_guards = 'ON' if args.stack_guards else 'OFF'
         seastar_cmake_args += ['-DSeastar_STACK_GUARDS={}'.format(stack_guards)]
@@ -2183,7 +2075,7 @@ def configure_seastar(build_dir, mode, mode_config, compiler_cache=None):
     subprocess.check_call(seastar_cmd, shell=False, cwd=cmake_dir)
 
 
-def configure_abseil(build_dir, mode, mode_config, compiler_cache=None):
+def configure_abseil(build_dir, mode, mode_config):
     abseil_cflags = mode_config['lib_cflags']
     cxx_flags = mode_config['cxxflags']
     if '-DSANITIZE' in cxx_flags:
@@ -2209,10 +2101,6 @@ def configure_abseil(build_dir, mode, mode_config, compiler_cache=None):
         '-DABSL_PROPAGATE_CXX_STD=ON',
     ]
 
-    if compiler_cache:
-        abseil_cmake_args += [f'-DCMAKE_CXX_COMPILER_LAUNCHER={compiler_cache}',
-                              f'-DCMAKE_C_COMPILER_LAUNCHER={compiler_cache}']
-
     cmake_args = abseil_cmake_args[:]
     abseil_build_dir = os.path.join(build_dir, mode, 'abseil')
     abseil_cmd = ['cmake', '-G', 'Ninja', real_relpath('abseil', abseil_build_dir)] + cmake_args
@@ -2358,6 +2246,15 @@ def get_extra_cxxflags(mode, mode_config, cxx, debuginfo):
     if debuginfo and mode_config['can_have_debug_info']:
         cxxflags += ['-g', '-gz']
 
+    if 'clang' in cxx:
+        # Since AssignmentTracking was enabled by default in clang
+        # (llvm/llvm-project@de6da6ad55d3ca945195d1cb109cb8efdf40a52a)
+        # coroutine frame debugging info (`coro_frame_ty`) is broken.
+        #
+        # It seems that we aren't losing much by disabling AssigmentTracking,
+        # so for now we choose to disable it to get `coro_frame_ty` back.
+        cxxflags.append('-Xclang -fexperimental-assignment-tracking=disabled')
+
     return cxxflags
 
 
@@ -2385,15 +2282,10 @@ def write_build_file(f,
                      scylla_product,
                      scylla_version,
                      scylla_release,
-                     compiler_cache,
                      args):
     use_precompiled_header = not args.disable_precompiled_header
     warnings = get_warning_options(args.cxx)
     rustc_target = pick_rustc_target('wasm32-wasi', 'wasm32-wasip1')
-    # If compiler cache is available, prefix the compiler with it
-    cxx_with_cache = f'{compiler_cache} {args.cxx}' if compiler_cache else args.cxx
-    # For Rust, sccache is used via RUSTC_WRAPPER environment variable
-    rustc_wrapper = f'RUSTC_WRAPPER={compiler_cache} ' if compiler_cache and 'sccache' in compiler_cache else ''
     f.write(textwrap.dedent('''\
         configure_args = {configure_args}
         builddir = {outdir}
@@ -2456,7 +2348,7 @@ def write_build_file(f,
             command = clang --target=wasm32 --no-standard-libraries -Wl,--export-all -Wl,--no-entry $in -o $out
             description = C2WASM $out
         rule rust2wasm
-            command = {rustc_wrapper}cargo build --target={rustc_target} --example=$example --locked --manifest-path=test/resource/wasm/rust/Cargo.toml --target-dir=$builddir/wasm/ $
+            command = cargo build --target={rustc_target} --example=$example --locked --manifest-path=test/resource/wasm/rust/Cargo.toml --target-dir=$builddir/wasm/ $
                 && wasm-opt -Oz $builddir/wasm/{rustc_target}/debug/examples/$example.wasm -o $builddir/wasm/$example.wasm $
                 && wasm-strip $builddir/wasm/$example.wasm
             description = RUST2WASM $out
@@ -2472,7 +2364,7 @@ def write_build_file(f,
           command = llvm-profdata merge $in -output=$out
         ''').format(configure_args=configure_args,
                     outdir=outdir,
-                    cxx=cxx_with_cache,
+                    cxx=args.cxx,
                     user_cflags=user_cflags,
                     warnings=warnings,
                     defines=defines,
@@ -2480,7 +2372,6 @@ def write_build_file(f,
                     user_ldflags=user_ldflags,
                     libs=libs,
                     rustc_target=rustc_target,
-                    rustc_wrapper=rustc_wrapper,
                     link_pool_depth=link_pool_depth,
                     seastar_path=args.seastar_path,
                     ninja=ninja,
@@ -2565,10 +2456,10 @@ def write_build_file(f,
               description = TEST {mode}
             # This rule is unused for PGO stages. They use the rust lib from the parent mode.
             rule rust_lib.{mode}
-              command = CARGO_BUILD_DEP_INFO_BASEDIR='.' {rustc_wrapper}cargo build --locked --manifest-path=rust/Cargo.toml --target-dir=$builddir/{mode} --profile=rust-{mode} $
+              command = CARGO_BUILD_DEP_INFO_BASEDIR='.' cargo build --locked --manifest-path=rust/Cargo.toml --target-dir=$builddir/{mode} --profile=rust-{mode} $
                         && touch $out
               description = RUST_LIB $out
-            ''').format(mode=mode, antlr3_exec=args.antlr3_exec, fmt_lib=fmt_lib, test_repeat=args.test_repeat, test_timeout=args.test_timeout, rustc_wrapper=rustc_wrapper, **modeval))
+            ''').format(mode=mode, antlr3_exec=args.antlr3_exec, fmt_lib=fmt_lib, test_repeat=args.test_repeat, test_timeout=args.test_timeout, **modeval))
         f.write(
             'build {mode}-build: phony {artifacts} {wasms} {vector_search_validator_bins}\n'.format(
                 mode=mode,
@@ -2632,7 +2523,7 @@ def write_build_file(f,
                 # In debug/sanitize modes, we compile with fsanitizers,
                 # so must use the same options during the link:
                 if '-DSANITIZE' in modes[mode]['cxxflags']:
-                    f.write('   libs = -fsanitize=address -fsanitize=undefined -lubsan\n')
+                    f.write('   libs = -fsanitize=address -fsanitize=undefined\n')
                 else:
                     f.write('   libs =\n')
                 f.write(f'build $builddir/{mode}/{binary}.stripped: strip $builddir/{mode}/{binary}\n')
@@ -2806,35 +2697,38 @@ def write_build_file(f,
 
         seastar_dep = f'$builddir/{mode}/seastar/libseastar.{seastar_lib_ext}'
         seastar_testing_dep = f'$builddir/{mode}/seastar/libseastar_testing.{seastar_lib_ext}'
-        f.write(f'build {seastar_dep}: ninja $builddir/{mode}/seastar/build.ninja | always {profile_dep}\n')
+        f.write('build {seastar_dep}: ninja $builddir/{mode}/seastar/build.ninja | always {profile_dep}\n'
+                .format(**locals()))
         f.write('  pool = submodule_pool\n')
-        f.write(f'  subdir = $builddir/{mode}/seastar\n')
-        f.write('  target = seastar\n')
-        f.write(f'build {seastar_testing_dep}: ninja $builddir/{mode}/seastar/build.ninja | always {profile_dep}\n')
+        f.write('  subdir = $builddir/{mode}/seastar\n'.format(**locals()))
+        f.write('  target = seastar\n'.format(**locals()))
+        f.write('build {seastar_testing_dep}: ninja $builddir/{mode}/seastar/build.ninja | always {profile_dep}\n'
+                .format(**locals()))
         f.write('  pool = submodule_pool\n')
-        f.write(f'  subdir = $builddir/{mode}/seastar\n')
-        f.write('  target = seastar_testing\n')
-        f.write(f'  profile_dep = {profile_dep}\n')
+        f.write('  subdir = $builddir/{mode}/seastar\n'.format(**locals()))
+        f.write('  target = seastar_testing\n'.format(**locals()))
+        f.write('  profile_dep = {profile_dep}\n'.format(**locals()))
 
         for lib in abseil_libs:
-            f.write(f'build $builddir/{mode}/abseil/{lib}: ninja $builddir/{mode}/abseil/build.ninja | always {profile_dep}\n')
-            f.write(f'  pool = submodule_pool\n')
-            f.write(f'  subdir = $builddir/{mode}/abseil\n')
-            f.write(f'  target = {lib}\n')
-            f.write(f'  profile_dep = {profile_dep}\n')
+            f.write('build $builddir/{mode}/abseil/{lib}: ninja $builddir/{mode}/abseil/build.ninja | always {profile_dep}\n'.format(**locals()))
+            f.write('  pool = submodule_pool\n')
+            f.write('  subdir = $builddir/{mode}/abseil\n'.format(**locals()))
+            f.write('  target = {lib}\n'.format(**locals()))
+            f.write('  profile_dep = {profile_dep}\n'.format(**locals()))
 
         f.write(f'build $builddir/{mode}/stdafx.hh.pch: cxx_build_precompiled_header.{mode} stdafx.hh | {profile_dep} {seastar_dep} {abseil_dep} {gen_headers_dep} {pch_dep}\n')
 
-        f.write(f'build $builddir/{mode}/seastar/apps/iotune/iotune: ninja $builddir/{mode}/seastar/build.ninja | $builddir/{mode}/seastar/libseastar.{seastar_lib_ext}\n')
+        f.write('build $builddir/{mode}/seastar/apps/iotune/iotune: ninja $builddir/{mode}/seastar/build.ninja | $builddir/{mode}/seastar/libseastar.{seastar_lib_ext}\n'
+                .format(**locals()))
         f.write('  pool = submodule_pool\n')
-        f.write(f'  subdir = $builddir/{mode}/seastar\n')
-        f.write('  target = iotune\n')
-        f.write(f'  profile_dep = {profile_dep}\n')
-        f.write(textwrap.dedent(f'''\
+        f.write('  subdir = $builddir/{mode}/seastar\n'.format(**locals()))
+        f.write('  target = iotune\n'.format(**locals()))
+        f.write('  profile_dep = {profile_dep}\n'.format(**locals()))
+        f.write(textwrap.dedent('''\
             build $builddir/{mode}/iotune: copy $builddir/{mode}/seastar/apps/iotune/iotune
             build $builddir/{mode}/iotune.stripped: strip $builddir/{mode}/iotune
             build $builddir/{mode}/iotune.debug: phony $builddir/{mode}/iotune.stripped
-            '''))
+            ''').format(**locals()))
         if args.dist_only:
             include_scylla_and_iotune = ''
             include_scylla_and_iotune_stripped = ''
@@ -2843,16 +2737,16 @@ def write_build_file(f,
             include_scylla_and_iotune = f'$builddir/{mode}/scylla $builddir/{mode}/iotune $builddir/{mode}/patchelf'
             include_scylla_and_iotune_stripped = f'$builddir/{mode}/scylla.stripped $builddir/{mode}/iotune.stripped $builddir/{mode}/patchelf.stripped'
             include_scylla_and_iotune_debug = f'$builddir/{mode}/scylla.debug $builddir/{mode}/iotune.debug'
-        f.write(f'build $builddir/{mode}/dist/tar/{scylla_product}-unstripped-{scylla_version}-{scylla_release}.{arch}.tar.gz: package {include_scylla_and_iotune} $builddir/SCYLLA-RELEASE-FILE $builddir/SCYLLA-VERSION-FILE $builddir/debian/debian $builddir/node_exporter/node_exporter | always\n')
-        f.write(f'  mode = {mode}\n')
-        f.write(f'build $builddir/{mode}/dist/tar/{scylla_product}-{scylla_version}-{scylla_release}.{arch}.tar.gz: stripped_package {include_scylla_and_iotune_stripped} $builddir/SCYLLA-RELEASE-FILE $builddir/SCYLLA-VERSION-FILE $builddir/debian/debian $builddir/node_exporter/node_exporter.stripped | always\n')
-        f.write(f'  mode = {mode}\n')
-        f.write(f'build $builddir/{mode}/dist/tar/{scylla_product}-debuginfo-{scylla_version}-{scylla_release}.{arch}.tar.gz: debuginfo_package {include_scylla_and_iotune_debug} $builddir/SCYLLA-RELEASE-FILE $builddir/SCYLLA-VERSION-FILE $builddir/debian/debian $builddir/node_exporter/node_exporter.debug | always\n')
-        f.write(f'  mode = {mode}\n')
-        f.write(f'build $builddir/{mode}/dist/tar/{scylla_product}-package.tar.gz: copy $builddir/{mode}/dist/tar/{scylla_product}-{scylla_version}-{scylla_release}.{arch}.tar.gz\n')
-        f.write(f'  mode = {mode}\n')
-        f.write(f'build $builddir/{mode}/dist/tar/{scylla_product}-{arch}-package.tar.gz: copy $builddir/{mode}/dist/tar/{scylla_product}-{scylla_version}-{scylla_release}.{arch}.tar.gz\n')
-        f.write(f'  mode = {mode}\n')
+        f.write('build $builddir/{mode}/dist/tar/{scylla_product}-unstripped-{scylla_version}-{scylla_release}.{arch}.tar.gz: package {include_scylla_and_iotune} $builddir/SCYLLA-RELEASE-FILE $builddir/SCYLLA-VERSION-FILE $builddir/debian/debian $builddir/node_exporter/node_exporter | always\n'.format(**locals()))
+        f.write('  mode = {mode}\n'.format(**locals()))
+        f.write('build $builddir/{mode}/dist/tar/{scylla_product}-{scylla_version}-{scylla_release}.{arch}.tar.gz: stripped_package {include_scylla_and_iotune_stripped} $builddir/SCYLLA-RELEASE-FILE $builddir/SCYLLA-VERSION-FILE $builddir/debian/debian $builddir/node_exporter/node_exporter.stripped | always\n'.format(**locals()))
+        f.write('  mode = {mode}\n'.format(**locals()))
+        f.write('build $builddir/{mode}/dist/tar/{scylla_product}-debuginfo-{scylla_version}-{scylla_release}.{arch}.tar.gz: debuginfo_package {include_scylla_and_iotune_debug} $builddir/SCYLLA-RELEASE-FILE $builddir/SCYLLA-VERSION-FILE $builddir/debian/debian $builddir/node_exporter/node_exporter.debug | always\n'.format(**locals()))
+        f.write('  mode = {mode}\n'.format(**locals()))
+        f.write('build $builddir/{mode}/dist/tar/{scylla_product}-package.tar.gz: copy $builddir/{mode}/dist/tar/{scylla_product}-{scylla_version}-{scylla_release}.{arch}.tar.gz\n'.format(**locals()))
+        f.write('  mode = {mode}\n'.format(**locals()))
+        f.write('build $builddir/{mode}/dist/tar/{scylla_product}-{arch}-package.tar.gz: copy $builddir/{mode}/dist/tar/{scylla_product}-{scylla_version}-{scylla_release}.{arch}.tar.gz\n'.format(**locals()))
+        f.write('  mode = {mode}\n'.format(**locals()))
 
         f.write(f'build $builddir/dist/{mode}/redhat: rpmbuild $builddir/{mode}/dist/tar/{scylla_product}-unstripped-{scylla_version}-{scylla_release}.{arch}.tar.gz\n')
         f.write(f'  mode = {mode}\n')
@@ -3025,9 +2919,6 @@ def create_build_system(args):
 
     os.makedirs(outdir, exist_ok=True)
 
-    compiler_cache = find_compiler_cache(args.compiler_cache)
-    resolve_compilers_for_compiler_cache(args, compiler_cache)
-
     scylla_product, scylla_version, scylla_release = generate_version(args.date_stamp)
 
     for mode, mode_config in build_modes.items():
@@ -3044,8 +2935,8 @@ def create_build_system(args):
         # {outdir}/{mode}/seastar/build.ninja, and
         # {outdir}/{mode}/seastar/seastar.pc is queried for building flags
         for mode, mode_config in build_modes.items():
-            configure_seastar(outdir, mode, mode_config, compiler_cache)
-            configure_abseil(outdir, mode, mode_config, compiler_cache)
+            configure_seastar(outdir, mode, mode_config)
+            configure_abseil(outdir, mode, mode_config)
         user_cflags += ' -isystem abseil'
 
     for mode, mode_config in build_modes.items():
@@ -3068,7 +2959,6 @@ def create_build_system(args):
                          scylla_product,
                          scylla_version,
                          scylla_release,
-                         compiler_cache,
                          args)
     generate_compdb('compile_commands.json', ninja, args.buildfile, selected_modes)
 
@@ -3111,10 +3001,6 @@ def configure_using_cmake(args):
     selected_modes = args.selected_modes or default_modes
     selected_configs = ';'.join(build_modes[mode].cmake_build_type for mode
                                 in selected_modes)
-
-    compiler_cache = find_compiler_cache(args.compiler_cache)
-    resolve_compilers_for_compiler_cache(args, compiler_cache)
-
     settings = {
         'CMAKE_CONFIGURATION_TYPES': selected_configs,
         'CMAKE_CROSS_CONFIGS': selected_configs,
@@ -3132,14 +3018,6 @@ def configure_using_cmake(args):
         'Scylla_WITH_DEBUG_INFO' : 'ON' if args.debuginfo else 'OFF',
         'Scylla_USE_PRECOMPILED_HEADER': 'OFF' if args.disable_precompiled_header else 'ON',
     }
-
-    if compiler_cache:
-        settings['CMAKE_CXX_COMPILER_LAUNCHER'] = compiler_cache
-        settings['CMAKE_C_COMPILER_LAUNCHER'] = compiler_cache
-        # For Rust, sccache is used via RUSTC_WRAPPER
-        if 'sccache' in compiler_cache:
-            settings['Scylla_RUSTC_WRAPPER'] = compiler_cache
-
     if args.date_stamp:
         settings['Scylla_DATE_STAMP'] = args.date_stamp
     if args.staticboost:
@@ -3171,7 +3049,7 @@ def configure_using_cmake(args):
 
     if not args.dist_only:
         for mode in selected_modes:
-            configure_seastar(build_dir, build_modes[mode].cmake_build_type, modes[mode], compiler_cache)
+            configure_seastar(build_dir, build_modes[mode].cmake_build_type, modes[mode])
 
     cmake_command = ['cmake']
     cmake_command += [f'-D{var}={value}' for var, value in settings.items()]

cql3/CMakeLists.txt

@@ -47,7 +47,6 @@ target_sources(cql3
     functions/aggregate_fcts.cc
     functions/castas_fcts.cc
     functions/error_injection_fcts.cc
-    functions/vector_similarity_fcts.cc
     statements/cf_prop_defs.cc
     statements/cf_statement.cc
     statements/authentication_statement.cc
@@ -105,7 +104,6 @@ target_sources(cql3
     statements/list_service_level_attachments_statement.cc
     statements/list_effective_service_level_statement.cc
     statements/describe_statement.cc
-    statements/view_prop_defs.cc
     update_parameters.cc
     util.cc
     ut_name.cc

cql3/Cql.g

@@ -431,7 +431,6 @@ unaliasedSelector returns [uexpression tmp]
        | K_TTL       '(' c=cident ')'              { tmp = column_mutation_attribute{column_mutation_attribute::attribute_kind::ttl,
                                                                                               unresolved_identifier{std::move(c)}}; }
        | f=functionName args=selectionFunctionArgs { tmp = function_call{std::move(f), std::move(args)}; }
-       | f=similarityFunctionName args=vectorSimilarityArgs            { tmp = function_call{std::move(f), std::move(args)}; }
        | K_CAST      '(' arg=unaliasedSelector K_AS t=native_type ')'  { tmp = cast{.style = cast::cast_style::sql, .arg = std::move(arg), .type = std::move(t)}; }
        )
        ( '.' fi=cident { tmp = field_selection{std::move(tmp), std::move(fi)}; }
@@ -446,18 +445,6 @@ selectionFunctionArgs returns [std::vector<expression> a]
       ')'
     ;
 
-vectorSimilarityArgs returns [std::vector<expression> a]
-    : '(' ')'
-    | '(' v1=vectorSimilarityArg { a.push_back(std::move(v1)); }
-          ( ',' vn=vectorSimilarityArg { a.push_back(std::move(vn)); } )*
-      ')'
-    ;
-
-vectorSimilarityArg returns [uexpression a]
-    : s=unaliasedSelector { a = std::move(s); }
-    | v=value             { a = std::move(v); }
-    ;
-
 countArgument
     : '*'
     | i=INTEGER { if (i->getText() != "1") {
@@ -588,15 +575,6 @@ usingTimeoutServiceLevelClauseObjective[std::unique_ptr<cql3::attributes::raw>&
     | serviceLevel sl_name=serviceLevelOrRoleName { attrs->service_level = std::move(sl_name); }
     ;
 
-usingTimeoutConcurrencyClause[std::unique_ptr<cql3::attributes::raw>& attrs]
-    : K_USING usingTimeoutConcurrencyClauseObjective[attrs] ( K_AND usingTimeoutConcurrencyClauseObjective[attrs] )*
-    ;
-
-usingTimeoutConcurrencyClauseObjective[std::unique_ptr<cql3::attributes::raw>& attrs]
-    : K_TIMEOUT to=term { attrs->timeout = std::move(to); }
-    | K_CONCURRENCY c=term { attrs->concurrency = std::move(c); }
-    ;
-
 /**
  * UPDATE <CF>
  * USING TIMESTAMP <long>
@@ -688,7 +666,7 @@ pruneMaterializedViewStatement returns [std::unique_ptr<raw::select_statement> e
         auto attrs = std::make_unique<cql3::attributes::raw>();
         expression wclause = conjunction{};
     }
-	: K_PRUNE K_MATERIALIZED K_VIEW cf=columnFamilyName (K_WHERE w=whereClause { wclause = std::move(w); } )? ( usingTimeoutConcurrencyClause[attrs] )?
+	: K_PRUNE K_MATERIALIZED K_VIEW cf=columnFamilyName (K_WHERE w=whereClause { wclause = std::move(w); } )? ( usingClause[attrs] )?
 	  {
 	        auto params = make_lw_shared<raw::select_statement::parameters>(std::move(orderings), is_distinct, allow_filtering, statement_subtype, bypass_cache);
 	        return std::make_unique<raw::select_statement>(std::move(cf), std::move(params),
@@ -898,10 +876,6 @@ pkDef[cql3::statements::create_table_statement::raw_statement& expr]
     | '(' k1=ident { l.push_back(k1); } ( ',' kn=ident { l.push_back(kn); } )* ')' { $expr.add_key_aliases(l); }
     ;
 
-cfamProperties[cql3::statements::cf_properties& expr]
-    : cfamProperty[expr] (K_AND cfamProperty[expr])*
-    ;
-
 cfamProperty[cql3::statements::cf_properties& expr]
     : property[*$expr.properties()]
     | K_COMPACT K_STORAGE { $expr.set_compact_storage(); }
@@ -939,22 +913,16 @@ typeColumns[create_type_statement& expr]
  */
 createIndexStatement returns [std::unique_ptr<create_index_statement> expr]
     @init {
-        auto idx_props = make_shared<index_specific_prop_defs>();
-        auto props = index_prop_defs();
+        auto props = make_shared<index_prop_defs>();
         bool if_not_exists = false;
         auto name = ::make_shared<cql3::index_name>();
         std::vector<::shared_ptr<index_target::raw>> targets;
     }
-    : K_CREATE (K_CUSTOM { idx_props->is_custom = true; })? K_INDEX (K_IF K_NOT K_EXISTS { if_not_exists = true; } )?
+    : K_CREATE (K_CUSTOM { props->is_custom = true; })? K_INDEX (K_IF K_NOT K_EXISTS { if_not_exists = true; } )?
         (idxName[*name])? K_ON cf=columnFamilyName '(' (target1=indexIdent { targets.emplace_back(target1); } (',' target2=indexIdent { targets.emplace_back(target2); } )*)? ')'
-        (K_USING cls=STRING_LITERAL { idx_props->custom_class = sstring{$cls.text}; })?
-        (K_WITH cfamProperties[props])?
-      {
-        props.extract_index_specific_properties_to(*idx_props);
-        view_prop_defs view_props = std::move(props).into_view_prop_defs();
-
-        $expr = std::make_unique<create_index_statement>(cf, name, targets, std::move(idx_props), std::move(view_props), if_not_exists);
-      }
+        (K_USING cls=STRING_LITERAL { props->custom_class = sstring{$cls.text}; })?
+        (K_WITH properties[*props])?
+      { $expr = std::make_unique<create_index_statement>(cf, name, targets, props, if_not_exists); }
     ;
 
 indexIdent returns [::shared_ptr<index_target::raw> id]
@@ -1102,9 +1070,9 @@ alterTypeStatement returns [std::unique_ptr<alter_type_statement> expr]
  */
 alterViewStatement returns [std::unique_ptr<alter_view_statement> expr]
     @init {
-        auto props = cql3::statements::view_prop_defs();
+        auto props = cql3::statements::cf_prop_defs();
     }
-    : K_ALTER K_MATERIALIZED K_VIEW cf=columnFamilyName K_WITH properties[*props.properties()]
+    : K_ALTER K_MATERIALIZED K_VIEW cf=columnFamilyName K_WITH properties[props]
     {
         $expr = std::make_unique<alter_view_statement>(std::move(cf), std::move(props));
     }
@@ -1706,10 +1674,6 @@ functionName returns [cql3::functions::function_name s]
     : (ks=keyspaceName '.')? f=allowedFunctionName   { $s.keyspace = std::move(ks); $s.name = std::move(f); }
     ;
 
-similarityFunctionName returns [cql3::functions::function_name s]
-    : f=allowedSimilarityFunctionName { $s = cql3::functions::function_name::native_function(std::move(f)); }
-    ;
-
 allowedFunctionName returns [sstring s]
     : f=IDENT                       { $s = $f.text; std::transform(s.begin(), s.end(), s.begin(), ::tolower); }
     | f=QUOTED_NAME                 { $s = $f.text; }
@@ -1718,11 +1682,6 @@ allowedFunctionName returns [sstring s]
     | K_COUNT                       { $s = "count"; }
     ;
 
-allowedSimilarityFunctionName returns [sstring s]
-    : f=(K_SIMILARITY_COSINE | K_SIMILARITY_EUCLIDEAN | K_SIMILARITY_DOT_PRODUCT)
-      { $s = $f.text; std::transform(s.begin(), s.end(), s.begin(), ::tolower); }
-    ;
-
 functionArgs returns [std::vector<expression> a]
     : '(' ')'
     | '(' t1=term { a.push_back(std::move(t1)); }
@@ -2411,7 +2370,6 @@ K_LIKE:        L I K E;
 
 K_TIMEOUT:     T I M E O U T;
 K_PRUNE:       P R U N E;
-K_CONCURRENCY: C O N C U R R E N C Y;
 
 K_EXECUTE:     E X E C U T E;
 
@@ -2419,10 +2377,6 @@ K_MUTATION_FRAGMENTS:    M U T A T I O N '_' F R A G M E N T S;
 
 K_VECTOR_SEARCH_INDEXING: V E C T O R '_' S E A R C H '_' I N D E X I N G;
 
-K_SIMILARITY_EUCLIDEAN:     S I M I L A R I T Y '_' E U C L I D E A N;
-K_SIMILARITY_COSINE:        S I M I L A R I T Y '_' C O S I N E;
-K_SIMILARITY_DOT_PRODUCT:   S I M I L A R I T Y '_' D O T '_' P R O D U C T;
-
 // Case-insensitive alpha characters
 fragment A: ('a'|'A');
 fragment B: ('b'|'B');

cql3/assignment_testable.hh

@@ -25,11 +25,6 @@ public:
         NOT_ASSIGNABLE,
     };
 
-    struct vector_test_result {
-        test_result result;
-        std::optional<size_t> dimension_opt;
-    };
-
     static bool is_assignable(test_result tr) {
         return tr != test_result::NOT_ASSIGNABLE;
     }
@@ -49,8 +44,6 @@ public:
      */
     virtual test_result test_assignment(data_dictionary::database db, const sstring& keyspace, const schema* schema_opt, const column_specification& receiver) const = 0;
 
-    virtual vector_test_result test_assignment_any_size_float_vector() const = 0;
-
     virtual std::optional<data_type> assignment_testable_type_opt() const = 0;
 
     // for error reporting

cql3/attributes.cc

@@ -20,21 +20,19 @@
 namespace cql3 {
 
 std::unique_ptr<attributes> attributes::none() {
-    return std::unique_ptr<attributes>{new attributes{{}, {}, {}, {}, {}}};
+    return std::unique_ptr<attributes>{new attributes{{}, {}, {}, {}}};
 }
 
 attributes::attributes(std::optional<cql3::expr::expression>&& timestamp,
                        std::optional<cql3::expr::expression>&& time_to_live,
                        std::optional<cql3::expr::expression>&& timeout,
-                       std::optional<sstring> service_level,
-                       std::optional<cql3::expr::expression>&& concurrency)
+                       std::optional<sstring> service_level)
     : _timestamp_unset_guard(timestamp)
     , _timestamp{std::move(timestamp)}
     , _time_to_live_unset_guard(time_to_live)
     , _time_to_live{std::move(time_to_live)}
     , _timeout{std::move(timeout)}
     , _service_level(std::move(service_level))
-    , _concurrency{std::move(concurrency)}
 { }
 
 bool attributes::is_timestamp_set() const {
@@ -53,10 +51,6 @@ bool attributes::is_service_level_set() const {
     return bool(_service_level);
 }
 
-bool attributes::is_concurrency_set() const {
-    return bool(_concurrency);
-}
-
 int64_t attributes::get_timestamp(int64_t now, const query_options& options) {
     if (!_timestamp.has_value() || _timestamp_unset_guard.is_unset(options)) {
         return now;
@@ -129,27 +123,6 @@ qos::service_level_options attributes::get_service_level(qos::service_level_cont
     return sl_controller.get_service_level(sl_name).slo;
 }
 
-std::optional<int32_t> attributes::get_concurrency(const query_options& options) const {
-    if (!_concurrency.has_value()) {
-        return std::nullopt;
-    }
-
-    cql3::raw_value concurrency_raw = expr::evaluate(*_concurrency, options);
-    if (concurrency_raw.is_null()) {
-        throw exceptions::invalid_request_exception("Invalid null value of concurrency");
-    }
-    int32_t concurrency;
-    try {
-        concurrency = concurrency_raw.view().validate_and_deserialize<int32_t>(*int32_type);
-    } catch (marshal_exception& e) {
-        throw exceptions::invalid_request_exception("Invalid concurrency value");
-    }
-    if (concurrency <= 0) {
-        throw exceptions::invalid_request_exception("Concurrency must be a positive integer");
-    }
-    return concurrency;
-}
-
 void attributes::fill_prepare_context(prepare_context& ctx) {
     if (_timestamp.has_value()) {
         expr::fill_prepare_context(*_timestamp, ctx);
@@ -160,13 +133,10 @@ void attributes::fill_prepare_context(prepare_context& ctx) {
     if (_timeout.has_value()) {
         expr::fill_prepare_context(*_timeout, ctx);
     }
-    if (_concurrency.has_value()) {
-        expr::fill_prepare_context(*_concurrency, ctx);
-    }
 }
 
 std::unique_ptr<attributes> attributes::raw::prepare(data_dictionary::database db, const sstring& ks_name, const sstring& cf_name) const {
-    std::optional<expr::expression> ts, ttl, to, conc;
+    std::optional<expr::expression> ts, ttl, to;
 
     if (timestamp.has_value()) {
         ts = prepare_expression(*timestamp, db, ks_name, nullptr, timestamp_receiver(ks_name, cf_name));
@@ -183,12 +153,7 @@ std::unique_ptr<attributes> attributes::raw::prepare(data_dictionary::database d
         verify_no_aggregate_functions(*timeout, "USING clause");
     }
 
-    if (concurrency.has_value()) {
-        conc = prepare_expression(*concurrency, db, ks_name, nullptr, concurrency_receiver(ks_name, cf_name));
-        verify_no_aggregate_functions(*concurrency, "USING clause");
-    }
-
-    return std::unique_ptr<attributes>{new attributes{std::move(ts), std::move(ttl), std::move(to), std::move(service_level), std::move(conc)}};
+    return std::unique_ptr<attributes>{new attributes{std::move(ts), std::move(ttl), std::move(to), std::move(service_level)}};
 }
 
 lw_shared_ptr<column_specification> attributes::raw::timestamp_receiver(const sstring& ks_name, const sstring& cf_name) const {
@@ -203,8 +168,4 @@ lw_shared_ptr<column_specification> attributes::raw::timeout_receiver(const sstr
     return make_lw_shared<column_specification>(ks_name, cf_name, ::make_shared<column_identifier>("[timeout]", true), duration_type);
 }
 
-lw_shared_ptr<column_specification> attributes::raw::concurrency_receiver(const sstring& ks_name, const sstring& cf_name) const {
-    return make_lw_shared<column_specification>(ks_name, cf_name, ::make_shared<column_identifier>("[concurrency]", true), data_type_for<int32_t>());
-}
-
 }

cql3/attributes.hh

@@ -36,15 +36,13 @@ private:
     std::optional<cql3::expr::expression> _time_to_live;
     std::optional<cql3::expr::expression> _timeout;
     std::optional<sstring> _service_level;
-    std::optional<cql3::expr::expression> _concurrency;
 public:
     static std::unique_ptr<attributes> none();
 private:
     attributes(std::optional<cql3::expr::expression>&& timestamp,
                std::optional<cql3::expr::expression>&& time_to_live,
                std::optional<cql3::expr::expression>&& timeout,
-               std::optional<sstring> service_level,
-               std::optional<cql3::expr::expression>&& concurrency);
+               std::optional<sstring> service_level);
 public:
     bool is_timestamp_set() const;
 
@@ -54,8 +52,6 @@ public:
 
     bool is_service_level_set() const;
 
-    bool is_concurrency_set() const;
-
     int64_t get_timestamp(int64_t now, const query_options& options);
 
     std::optional<int32_t> get_time_to_live(const query_options& options);
@@ -64,8 +60,6 @@ public:
 
     qos::service_level_options get_service_level(qos::service_level_controller& sl_controller) const;
 
-    std::optional<int32_t> get_concurrency(const query_options& options) const;
-
     void fill_prepare_context(prepare_context& ctx);
 
     class raw final {
@@ -74,7 +68,6 @@ public:
         std::optional<cql3::expr::expression> time_to_live;
         std::optional<cql3::expr::expression> timeout;
         std::optional<sstring> service_level;
-        std::optional<cql3::expr::expression> concurrency;
 
         std::unique_ptr<attributes> prepare(data_dictionary::database db, const sstring& ks_name, const sstring& cf_name) const;
     private:
@@ -83,8 +76,6 @@ public:
         lw_shared_ptr<column_specification> time_to_live_receiver(const sstring& ks_name, const sstring& cf_name) const;
 
         lw_shared_ptr<column_specification> timeout_receiver(const sstring& ks_name, const sstring& cf_name) const;
-
-        lw_shared_ptr<column_specification> concurrency_receiver(const sstring& ks_name, const sstring& cf_name) const;
     };
 };
 

cql3/expr/prepare_expr.cc

@@ -1434,112 +1434,6 @@ test_assignment(const expression& expr, data_dictionary::database db, const sstr
     }, expr);
 }
 
-template <cql3_type::kind... Kinds>
-assignment_testable::vector_test_result
-test_assignment_any_size_float_vector(const expression& expr) {
-    using test_result = assignment_testable::vector_test_result;
-    const test_result NOT_ASSIGNABLE = {assignment_testable::test_result::NOT_ASSIGNABLE, std::nullopt};
-    const test_result WEAKLY_ASSIGNABLE = {assignment_testable::test_result::WEAKLY_ASSIGNABLE, std::nullopt};
-    auto is_float_or_bind = [] (const expression& e) {
-        return expr::visit(overloaded_functor{
-            [] (const bind_variable&) {
-                return true;
-            },
-            [] (const untyped_constant& uc) {
-                return uc.partial_type == untyped_constant::type_class::floating_point
-                    || uc.partial_type == untyped_constant::type_class::integer;
-            },
-            [] (const constant& value) {
-                auto kind = value.type->as_cql3_type().get_kind();
-                return cql3_type::kind_enum_set::frozen<Kinds...>().contains(kind);
-            },
-            [] (const auto&) {
-                return false;
-            },
-        }, e);
-    };
-    auto validate_assignment = [&] (const data_type& dt) -> test_result {
-         auto vt = dynamic_pointer_cast<const vector_type_impl>(dt->underlying_type());
-            if (!vt) {
-                return NOT_ASSIGNABLE;
-            }
-            auto elem_kind = vt->get_elements_type()->as_cql3_type().get_kind();
-            if (cql3_type::kind_enum_set::frozen<Kinds...>().contains(elem_kind)) {
-                return {assignment_testable::test_result::WEAKLY_ASSIGNABLE, vt->get_dimension()};
-            }
-            return NOT_ASSIGNABLE;
-    };
-    return expr::visit(overloaded_functor{
-        [&] (const constant& value) -> test_result {
-            return validate_assignment(value.type);
-        },
-        [&] (const binary_operator&) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const conjunction&) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const column_value& col_val) -> test_result {
-            return validate_assignment(col_val.col->type);
-        },
-        [&] (const subscript&) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const unresolved_identifier& ui) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const column_mutation_attribute& cma) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const function_call& fc) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const cast& c) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const field_selection& fs) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const bind_variable& bv) -> test_result {
-            return WEAKLY_ASSIGNABLE;
-        },
-        [&] (const untyped_constant& uc) -> test_result {
-            return uc.partial_type == untyped_constant::type_class::null
-                ? WEAKLY_ASSIGNABLE
-                : NOT_ASSIGNABLE;
-        },
-        [&] (const tuple_constructor& tc) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const collection_constructor& c) -> test_result {
-            switch (c.style) {
-            case collection_constructor::style_type::list_or_vector: {
-                if(std::ranges::all_of(c.elements, is_float_or_bind)) {
-                    return {assignment_testable::test_result::WEAKLY_ASSIGNABLE, c.elements.size()};
-                }
-                return NOT_ASSIGNABLE;
-            }
-            case collection_constructor::style_type::set: return NOT_ASSIGNABLE;
-            case collection_constructor::style_type::map: return NOT_ASSIGNABLE;
-            case collection_constructor::style_type::vector:
-                on_internal_error(expr_logger, "vector style type found in test_assignment, should have been introduced post-prepare");
-            }
-            on_internal_error(expr_logger, fmt::format("unexpected collection_constructor style {}", static_cast<unsigned>(c.style)));
-        },
-        [&] (const usertype_constructor& uc) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-        [&] (const temporary& t) -> test_result {
-            return NOT_ASSIGNABLE;
-        },
-    }, expr);
-}
-
-assignment_testable::vector_test_result
-test_assignment_any_size_float_vector(const expression& expr) {
-    return test_assignment_any_size_float_vector<cql3_type::kind::FLOAT, cql3_type::kind::DOUBLE>(expr);
-}
-
 expression
 prepare_expression(const expression& expr, data_dictionary::database db, const sstring& keyspace, const schema* schema_opt, lw_shared_ptr<column_specification> receiver) {
     auto e_opt = try_prepare_expression(expr, db, keyspace, schema_opt, std::move(receiver));
@@ -1573,9 +1467,6 @@ public:
     virtual test_result test_assignment(data_dictionary::database db, const sstring& keyspace, const schema* schema_opt, const column_specification& receiver) const override {
         return expr::test_assignment(_e, db, keyspace, schema_opt, receiver);
     }
-    virtual vector_test_result test_assignment_any_size_float_vector() const override {
-        return expr::test_assignment_any_size_float_vector(_e);
-    }
     virtual sstring assignment_testable_source_context() const override {
         return fmt::format("{}", _e);
     }

cql3/functions/functions.cc

@@ -16,7 +16,6 @@
 #include "cql3/functions/user_function.hh"
 #include "cql3/functions/user_aggregate.hh"
 #include "cql3/functions/uuid_fcts.hh"
-#include "cql3/functions/vector_similarity_fcts.hh"
 #include "data_dictionary/data_dictionary.hh"
 #include "as_json_function.hh"
 #include "cql3/prepare_context.hh"
@@ -399,14 +398,6 @@ functions::get(data_dictionary::database db,
         }
     });
 
-    const auto func_name = name.has_keyspace() ? name : name.as_native_function();
-    if (SIMILARITY_FUNCTIONS.contains(func_name)) {
-        auto arg_types = retrieve_vector_arg_types(func_name, provided_args);
-        auto fun = ::make_shared<vector_similarity_fct>(func_name.name, arg_types);
-        validate_types(db, keyspace, schema.get(), fun, provided_args, receiver_ks, receiver_cf);
-        return fun;
-    }
-
     if (name.has_keyspace()
                 ? name == TOKEN_FUNCTION_NAME
                 : name.name == TOKEN_FUNCTION_NAME.name) {

cql3/functions/vector_similarity_fcts.cc

@@ -1,150 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#include "vector_similarity_fcts.hh"
-#include "types/types.hh"
-#include "types/vector.hh"
-#include "exceptions/exceptions.hh"
-
-namespace cql3 {
-namespace functions {
-namespace {
-
-// The computations of similarity scores match the exact formulas of Cassandra's (jVector's) implementation to ensure compatibility.
-// There exist tests checking the compliance of the results.
-// Reference:
-// https://github.com/datastax/jvector/blob/f967f1c9249035b63b55a566fac7d4dc38380349/jvector-base/src/main/java/io/github/jbellis/jvector/vector/VectorSimilarityFunction.java#L36-L69
-
-// You should only use this function if you need to preserve the original vectors and cannot normalize
-// them in advance.
-float compute_cosine_similarity(const std::vector<data_value>& v1, const std::vector<data_value>& v2) {
-    double dot_product = 0.0;
-    double squared_norm_a = 0.0;
-    double squared_norm_b = 0.0;
-
-    for (size_t i = 0; i < v1.size(); ++i) {
-        double a = value_cast<float>(v1[i]);
-        double b = value_cast<float>(v2[i]);
-
-        dot_product += a * b;
-        squared_norm_a += a * a;
-        squared_norm_b += b * b;
-    }
-
-    if (squared_norm_a == 0 || squared_norm_b == 0) {
-        throw exceptions::invalid_request_exception("Function system.similarity_cosine doesn't support all-zero vectors");
-    }
-
-    // The cosine similarity is in the range [-1, 1].
-    // It is mapped to a similarity score in the range [0, 1] (-1 -> 0, 1 -> 1)
-    // for consistency with other similarity functions.
-    return (1 + (dot_product / (std::sqrt(squared_norm_a * squared_norm_b)))) / 2;
-}
-
-float compute_euclidean_similarity(const std::vector<data_value>& v1, const std::vector<data_value>& v2) {
-    double sum = 0.0;
-
-    for (size_t i = 0; i < v1.size(); ++i) {
-        double a = value_cast<float>(v1[i]);
-        double b = value_cast<float>(v2[i]);
-
-        double diff = a - b;
-        sum += diff * diff;
-    }
-
-    // The squared Euclidean (L2) distance is of range [0, inf).
-    // It is mapped to a similarity score in the range (0, 1] (0 -> 1, inf -> 0)
-    // for consistency with other similarity functions.
-    return (1 / (1 + sum));
-}
-
-// Assumes that both vectors are L2-normalized.
-// This similarity is intended as an optimized way to perform cosine similarity calculation.
-float compute_dot_product_similarity(const std::vector<data_value>& v1, const std::vector<data_value>& v2) {
-    double dot_product = 0.0;
-
-    for (size_t i = 0; i < v1.size(); ++i) {
-        double a = value_cast<float>(v1[i]);
-        double b = value_cast<float>(v2[i]);
-        dot_product += a * b;
-    }
-
-    // The dot product is in the range [-1, 1] for L2-normalized vectors.
-    // It is mapped to a similarity score in the range [0, 1] (-1 -> 0, 1 -> 1)
-    // for consistency with other similarity functions.
-    return ((1 + dot_product) / 2);
-}
-
-} // namespace
-
-thread_local const std::unordered_map<function_name, similarity_function_t> SIMILARITY_FUNCTIONS = {
-        {SIMILARITY_COSINE_FUNCTION_NAME, compute_cosine_similarity},
-        {SIMILARITY_EUCLIDEAN_FUNCTION_NAME, compute_euclidean_similarity},
-        {SIMILARITY_DOT_PRODUCT_FUNCTION_NAME, compute_dot_product_similarity},
-};
-
-std::vector<data_type> retrieve_vector_arg_types(const function_name& name, const std::vector<shared_ptr<assignment_testable>>& provided_args) {
-    if (provided_args.size() != 2) {
-        throw exceptions::invalid_request_exception(fmt::format("Invalid number of arguments for function {}(vector<float, n>, vector<float, n>)", name));
-    }
-
-    auto [first_result, first_dim_opt] = provided_args[0]->test_assignment_any_size_float_vector();
-    auto [second_result, second_dim_opt] = provided_args[1]->test_assignment_any_size_float_vector();
-
-    auto invalid_type_error_message = [&name](const shared_ptr<assignment_testable>& arg) {
-        auto type = arg->assignment_testable_type_opt();
-        const auto& source_context = arg->assignment_testable_source_context();
-        if (type) {
-            return fmt::format("Function {} requires a float vector argument, but found {} of type {}", name, source_context, type.value()->cql3_type_name());
-        } else {
-            return fmt::format("Function {} requires a float vector argument, but found {}", name, source_context);
-        }
-    };
-
-    if (!is_assignable(first_result)) {
-        throw exceptions::invalid_request_exception(invalid_type_error_message(provided_args[0]));
-    }
-    if (!is_assignable(second_result)) {
-        throw exceptions::invalid_request_exception(invalid_type_error_message(provided_args[1]));
-    }
-
-    if (!first_dim_opt && !second_dim_opt) {
-        throw exceptions::invalid_request_exception(fmt::format("Cannot infer type of argument {} for function {}(vector<float, n>, vector<float, n>)",
-                provided_args[0]->assignment_testable_source_context(), name));
-    }
-    if (first_dim_opt && second_dim_opt) {
-        if (*first_dim_opt != *second_dim_opt) {
-            throw exceptions::invalid_request_exception(fmt::format(
-                    "All arguments must have the same vector dimensions, but found vector<float, {}> and vector<float, {}>", *first_dim_opt, *second_dim_opt));
-        }
-    }
-
-    size_t dimension = first_dim_opt ? *first_dim_opt : *second_dim_opt;
-    auto type = vector_type_impl::get_instance(float_type, dimension);
-    return {type, type};
-}
-
-bytes_opt vector_similarity_fct::execute(std::span<const bytes_opt> parameters) {
-    if (std::any_of(parameters.begin(), parameters.end(), [](const auto& param) {
-            return !param;
-        })) {
-        return std::nullopt;
-    }
-
-    const auto& type = arg_types()[0];
-    data_value v1 = type->deserialize(*parameters[0]);
-    data_value v2 = type->deserialize(*parameters[1]);
-    const auto& v1_elements = value_cast<std::vector<data_value>>(v1);
-    const auto& v2_elements = value_cast<std::vector<data_value>>(v2);
-
-    float result = SIMILARITY_FUNCTIONS.at(_name)(v1_elements, v2_elements);
-    return float_type->decompose(result);
-}
-
-} // namespace functions
-} // namespace cql3

cql3/functions/vector_similarity_fcts.hh

@@ -1,37 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#pragma once
-
-#include "native_scalar_function.hh"
-#include "cql3/assignment_testable.hh"
-#include "cql3/functions/function_name.hh"
-
-namespace cql3 {
-namespace functions {
-
-static const function_name SIMILARITY_COSINE_FUNCTION_NAME = function_name::native_function("similarity_cosine");
-static const function_name SIMILARITY_EUCLIDEAN_FUNCTION_NAME = function_name::native_function("similarity_euclidean");
-static const function_name SIMILARITY_DOT_PRODUCT_FUNCTION_NAME = function_name::native_function("similarity_dot_product");
-
-using similarity_function_t = float (*)(const std::vector<data_value>&, const std::vector<data_value>&);
-extern thread_local const std::unordered_map<function_name, similarity_function_t> SIMILARITY_FUNCTIONS;
-
-std::vector<data_type> retrieve_vector_arg_types(const function_name& name, const std::vector<shared_ptr<assignment_testable>>& provided_args);
-
-class vector_similarity_fct : public native_scalar_function {
-public:
-    vector_similarity_fct(const sstring& name, const std::vector<data_type>& arg_types)
-        : native_scalar_function(name, float_type, arg_types) {
-    }
-
-    virtual bytes_opt execute(std::span<const bytes_opt> parameters) override;
-};
-
-} // namespace functions
-} // namespace cql3

cql3/query_processor.cc

@@ -14,7 +14,6 @@
 #include <seastar/core/shared_ptr.hh>
 #include <seastar/coroutine/parallel_for_each.hh>
 #include <seastar/coroutine/as_future.hh>
-#include <seastar/coroutine/try_future.hh>
 
 #include "service/storage_proxy.hh"
 #include "service/migration_manager.hh"
@@ -65,10 +64,6 @@ bool query_processor::topology_global_queue_empty() {
     return remote().first.get().ss.topology_global_queue_empty();
 }
 
-future<bool> query_processor::ongoing_rf_change(const service::group0_guard& guard, sstring ks) {
-    return remote().first.get().ss.ongoing_rf_change(guard, std::move(ks));
-}
-
 static service::query_state query_state_for_internal_call() {
     return {service::client_state::for_internal_calls(), empty_service_permit()};
 }
@@ -994,7 +989,7 @@ query_processor::execute_with_params(
     auto opts = make_internal_options(p, values, cl);
     auto statement = p->statement;
 
-    auto msg = co_await coroutine::try_future(execute_maybe_with_guard(query_state, std::move(statement), opts, &query_processor::do_execute_with_params));
+    auto msg = co_await execute_maybe_with_guard(query_state, std::move(statement), opts, &query_processor::do_execute_with_params);
     co_return ::make_shared<untyped_result_set>(msg);
 }
 
@@ -1004,7 +999,7 @@ query_processor::do_execute_with_params(
         shared_ptr<cql_statement> statement,
         const query_options& options, std::optional<service::group0_guard> guard) {
     statement->validate(*this, service::client_state::for_internal_calls());
-    co_return co_await coroutine::try_future(statement->execute(*this, query_state, options, std::move(guard)));
+    co_return co_await statement->execute(*this, query_state, options, std::move(guard));
 }
 
 

cql3/query_processor.hh

@@ -474,7 +474,6 @@ public:
     void reset_cache();
 
     bool topology_global_queue_empty();
-    future<bool> ongoing_rf_change(const service::group0_guard& guard, sstring ks);
 
     query_options make_internal_options(
             const statements::prepared_statement::checked_weak_ptr& p,

cql3/restrictions/statement_restrictions.cc

@@ -1322,10 +1322,6 @@ const std::vector<expr::expression>& statement_restrictions::index_restrictions(
     return _index_restrictions;
 }
 
-bool statement_restrictions::is_empty() const {
-    return !_where.has_value();
-}
-
 // Current score table:
 // local and restrictions include full partition key: 2
 // global: 1

cql3/restrictions/statement_restrictions.hh

@@ -408,8 +408,6 @@ public:
 
     /// Checks that the primary key restrictions don't contain null values, throws invalid_request_exception otherwise.
     void validate_primary_key(const query_options& options) const;
-
-    bool is_empty() const;
 };
 
 statement_restrictions analyze_statement_restrictions(

cql3/selection/selectable.cc

@@ -32,7 +32,7 @@ bool
 selectable_processes_selection(const expr::expression& selectable) {
     return expr::visit(overloaded_functor{
         [&] (const expr::constant&) -> bool {
-            return true;
+            on_internal_error(slogger, "no way to express SELECT constant in the grammar yet");
         },
         [&] (const expr::conjunction& conj) -> bool {
             on_internal_error(slogger, "no way to express 'SELECT a AND b' in the grammar yet");

cql3/statements/alter_keyspace_statement.cc

@@ -19,7 +19,6 @@
 #include "locator/abstract_replication_strategy.hh"
 #include "mutation/canonical_mutation.hh"
 #include "prepared_statement.hh"
-#include <seastar/coroutine/exception.hh>
 #include "service/migration_manager.hh"
 #include "service/storage_proxy.hh"
 #include "service/topology_mutation.hh"
@@ -139,7 +138,6 @@ bool cql3::statements::alter_keyspace_statement::changes_tablets(query_processor
 future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, cql3::cql_warnings_vec>>
 cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_processor& qp, service::query_state& state, const query_options& options, service::group0_batch& mc) const {
     using namespace cql_transport;
-    bool unknown_keyspace = false;
     try {
         event::schema_change::target_type target_type = event::schema_change::target_type::KEYSPACE;
         auto ks = qp.db().find_keyspace(_name);
@@ -160,12 +158,8 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce
         //       when in reality nothing or only schema is being changed
         if (changes_tablets(qp)) {
             if (!qp.proxy().features().topology_global_request_queue && !qp.topology_global_queue_empty()) {
-                co_await coroutine::return_exception(
-                    exceptions::invalid_request_exception("Another global topology request is ongoing, please retry."));
-            }
-            if (qp.proxy().features().rack_list_rf && co_await qp.ongoing_rf_change(mc.guard(),_name)) {
-                co_await coroutine::return_exception(
-                        exceptions::invalid_request_exception(format("Another RF change for this keyspace {} ongoing, please retry.", _name)));
+                return make_exception_future<std::tuple<::shared_ptr<::cql_transport::event::schema_change>, cql3::cql_warnings_vec>>(
+                        exceptions::invalid_request_exception("Another global topology request is ongoing, please retry."));
             }
             qp.db().real_database().validate_keyspace_update(*ks_md_update);
 
@@ -206,9 +200,8 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce
                 locator::replication_strategy_params(ks_md_update->strategy_options(), ks_md_update->initial_tablets(), ks_md_update->consistency_option()),
                 topo);
 
-        // If RF-rack-validity must be enforced for the keyspace according to `enforce_rf_rack_validity_for_keyspace`,
-        // it's forbidden to perform a schema change that would lead to an RF-rack-invalid keyspace.
-        // Verify that this change does not.
+        // If `rf_rack_valid_keyspaces` is enabled, it's forbidden to perform a schema change that
+        // would lead to an RF-rack-valid keyspace. Verify that this change does not.
         // For more context, see: scylladb/scylladb#23071.
         try {
             // There are two things to note here:
@@ -226,13 +219,13 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce
             //    disturb it (see scylladb/scylladb#23345), but we ignore that.
             locator::assert_rf_rack_valid_keyspace(_name, tmptr, *rs);
         } catch (const std::exception& e) {
-            if (replica::database::enforce_rf_rack_validity_for_keyspace(qp.db().get_config(), *ks_md)) {
+            if (qp.db().get_config().rf_rack_valid_keyspaces()) {
                 // There's no guarantee what the type of the exception will be, so we need to
                 // wrap it manually here in a type that can be passed to the user.
                 throw exceptions::invalid_request_exception(e.what());
             } else {
-                // Even when RF-rack-validity is not enforced for the keyspace, we'd
-                // like to inform the user that the keyspace they're altering will not
+                // Even when the configuration option `rf_rack_valid_keyspaces` is set to false,
+                // we'd like to inform the user that the keyspace they're altering will not
                 // satisfy the restriction after the change--but just as a warning.
                 // For more context, see issue: scylladb/scylladb#23330.
                 warnings.push_back(seastar::format(
@@ -249,15 +242,10 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce
                 target_type,
                 keyspace());
         mc.add_mutations(std::move(muts), "CQL alter keyspace");
-        co_return std::make_tuple(std::move(ret), warnings);
+        return make_ready_future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, cql3::cql_warnings_vec>>(std::make_tuple(std::move(ret), warnings));
     } catch (data_dictionary::no_such_keyspace& e) {
-        unknown_keyspace = true;
+        return make_exception_future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, cql3::cql_warnings_vec>>(exceptions::invalid_request_exception("Unknown keyspace " + _name));
     }
-    if (unknown_keyspace) {
-        co_await coroutine::return_exception(
-                exceptions::invalid_request_exception("Unknown keyspace " + _name));
-    }
-    std::unreachable();
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>

cql3/statements/alter_view_statement.cc

@@ -11,7 +11,6 @@
 #include <seastar/core/coroutine.hh>
 #include "cql3/statements/alter_view_statement.hh"
 #include "cql3/statements/prepared_statement.hh"
-#include "cql3/statements/view_prop_defs.hh"
 #include "service/migration_manager.hh"
 #include "service/storage_proxy.hh"
 #include "validation.hh"
@@ -23,7 +22,7 @@ namespace cql3 {
 
 namespace statements {
 
-alter_view_statement::alter_view_statement(cf_name view_name, std::optional<view_prop_defs> properties)
+alter_view_statement::alter_view_statement(cf_name view_name, std::optional<cf_prop_defs> properties)
         : schema_altering_statement{std::move(view_name)}
         , _properties{std::move(properties)}
 {
@@ -53,8 +52,8 @@ view_ptr alter_view_statement::prepare_view(data_dictionary::database db) const
         throw exceptions::invalid_request_exception("ALTER MATERIALIZED VIEW WITH invoked, but no parameters found");
     }
 
-    auto schema_extensions = _properties->properties()->make_schema_extensions(db.extensions());
-    _properties->validate_raw(view_prop_defs::op_type::alter, db, keyspace(), schema_extensions);
+    auto schema_extensions = _properties->make_schema_extensions(db.extensions());
+    _properties->validate(db, keyspace(), schema_extensions);
 
     bool is_colocated = [&] {
         if (!db.find_keyspace(keyspace()).get_replication_strategy().uses_tablets()) {
@@ -71,15 +70,28 @@ view_ptr alter_view_statement::prepare_view(data_dictionary::database db) const
     }();
 
     if (is_colocated) {
-        auto gc_opts = _properties->properties()->get_tombstone_gc_options(schema_extensions);
+        auto gc_opts = _properties->get_tombstone_gc_options(schema_extensions);
         if (gc_opts && gc_opts->mode() == tombstone_gc_mode::repair) {
             throw exceptions::invalid_request_exception("The 'repair' mode for tombstone_gc is not allowed on co-located materialized view tables.");
         }
     }
 
     auto builder = schema_builder(schema);
-    _properties->apply_to_builder(view_prop_defs::op_type::alter, builder, std::move(schema_extensions),
-            db, keyspace(), is_colocated);
+    _properties->apply_to_builder(builder, std::move(schema_extensions), db, keyspace(), !is_colocated);
+
+    if (builder.get_gc_grace_seconds() == 0) {
+        throw exceptions::invalid_request_exception(
+                "Cannot alter gc_grace_seconds of a materialized view to 0, since this "
+                "value is used to TTL undelivered updates. Setting gc_grace_seconds too "
+                "low might cause undelivered updates to expire before being replayed.");
+    }
+
+    if (builder.default_time_to_live().count() > 0) {
+        throw exceptions::invalid_request_exception(
+                "Cannot set or alter default_time_to_live for a materialized view. "
+                "Data in a materialized view always expire at the same time than "
+                "the corresponding data in the parent table.");
+    }
 
     return view_ptr(builder.build());
 }

cql3/statements/alter_view_statement.hh

@@ -12,8 +12,8 @@
 
 #include <seastar/core/shared_ptr.hh>
 
-#include "cql3/statements/view_prop_defs.hh"
 #include "data_dictionary/data_dictionary.hh"
+#include "cql3/statements/cf_prop_defs.hh"
 #include "cql3/statements/schema_altering_statement.hh"
 
 namespace cql3 {
@@ -26,10 +26,10 @@ namespace statements {
 /** An <code>ALTER MATERIALIZED VIEW</code> parsed from a CQL query statement. */
 class alter_view_statement : public schema_altering_statement {
 private:
-    std::optional<view_prop_defs> _properties;
+    std::optional<cf_prop_defs> _properties;
     view_ptr prepare_view(data_dictionary::database db) const;
 public:
-    alter_view_statement(cf_name view_name, std::optional<view_prop_defs> properties);
+    alter_view_statement(cf_name view_name, std::optional<cf_prop_defs> properties);
 
     virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
 

cql3/statements/batch_statement.cc

@@ -190,7 +190,7 @@ future<utils::chunked_vector<mutation>> batch_statement::get_mutations(query_pro
     co_return vresult;
 }
 
-void batch_statement::verify_batch_size(query_processor& qp, const utils::chunked_vector<mutation>& mutations) const {
+void batch_statement::verify_batch_size(query_processor& qp, const utils::chunked_vector<mutation>& mutations) {
     if (mutations.size() <= 1) {
         return;     // We only warn for batch spanning multiple mutations
     }
@@ -209,9 +209,8 @@ void batch_statement::verify_batch_size(query_processor& qp, const utils::chunke
             for (auto&& m : mutations) {
                 ks_cf_pairs.insert(m.schema()->ks_name() + "." + m.schema()->cf_name());
             }
-            const auto batch_type = _type == type::LOGGED ? "Logged" : "Unlogged";
-            return seastar::format("{} batch modifying {:d} partitions in {} is of size {:d} bytes, exceeding specified {} threshold of {:d} by {:d}.",
-                    batch_type, mutations.size(), fmt::join(ks_cf_pairs, ", "), size, type, threshold, size - threshold);
+            return seastar::format("Batch modifying {:d} partitions in {} is of size {:d} bytes, exceeding specified {} threshold of {:d} by {:d}.",
+                    mutations.size(), fmt::join(ks_cf_pairs, ", "), size, type, threshold, size - threshold);
         };
         if (size > fail_threshold) {
             _logger.error("{}", error("FAIL", fail_threshold).c_str());
@@ -332,7 +331,7 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
     if (!cl_for_paxos) [[unlikely]] {
         return make_exception_future<shared_ptr<cql_transport::messages::result_message>>(std::move(cl_for_paxos).assume_error());
     }
-    std::unique_ptr<cas_request> request;
+    seastar::shared_ptr<cas_request> request;
     schema_ptr schema;
 
     db::timeout_clock::time_point now = db::timeout_clock::now();
@@ -355,9 +354,9 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
         if (keys.empty()) {
             continue;
         }
-        if (!request) {
+        if (request.get() == nullptr) {
             schema = statement.s;
-            request = std::make_unique<cas_request>(schema, std::move(keys));
+            request = seastar::make_shared<cas_request>(schema, std::move(keys));
         } else if (keys.size() != 1 || keys.front().equal(request->key().front(), dht::ring_position_comparator(*schema)) == false) {
             throw exceptions::invalid_request_exception("BATCH with conditions cannot span multiple partitions");
         }
@@ -367,7 +366,7 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
 
         request->add_row_update(statement, std::move(ranges), std::move(json_cache), statement_options);
     }
-    if (!request) {
+    if (request.get() == nullptr) {
         throw exceptions::invalid_request_exception(format("Unrestricted partition key in a conditional BATCH"));
     }
 
@@ -378,10 +377,9 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
             );
     }
 
-    auto* request_ptr = request.get();
-    return qp.proxy().cas(schema, std::move(cas_shard), *request_ptr, request->read_command(qp), request->key(),
+    return qp.proxy().cas(schema, std::move(cas_shard), request, request->read_command(qp), request->key(),
             {read_timeout, qs.get_permit(), qs.get_client_state(), qs.get_trace_state()},
-            std::move(cl_for_paxos).assume_value(), cl_for_learn, batch_timeout, cas_timeout).then([this, request = std::move(request)] (bool is_applied) {
+            std::move(cl_for_paxos).assume_value(), cl_for_learn, batch_timeout, cas_timeout).then([this, request] (bool is_applied) {
         return request->build_cas_result_set(_metadata, _columns_of_cas_result_set, is_applied);
     });
 }

cql3/statements/batch_statement.hh

@@ -116,7 +116,7 @@ public:
      * Checks batch size to ensure threshold is met. If not, a warning is logged.
      * @param cfs ColumnFamilies that will store the batch's mutations.
      */
-    void verify_batch_size(query_processor& qp, const utils::chunked_vector<mutation>& mutations) const;
+    static void verify_batch_size(query_processor& qp, const utils::chunked_vector<mutation>& mutations);
 
     virtual future<shared_ptr<cql_transport::messages::result_message>> execute(
             query_processor& qp, service::query_state& state, const query_options& options, std::optional<service::group0_guard> guard) const override;

cql3/statements/cf_properties.hh

@@ -19,8 +19,7 @@ namespace statements {
 /**
  * Class for common statement properties.
  */
-class cf_properties {
-protected:
+class cf_properties final {
     const ::shared_ptr<cf_prop_defs> _properties = ::make_shared<cf_prop_defs>();
     bool _use_compact_storage = false;
     std::vector<std::pair<::shared_ptr<column_identifier>, bool>> _defined_ordering; // Insertion ordering is important

cql3/statements/create_index_statement.cc

@@ -14,7 +14,6 @@
 #include "db/view/view.hh"
 #include "exceptions/exceptions.hh"
 #include "index/vector_index.hh"
-#include "locator/token_metadata_fwd.hh"
 #include "prepared_statement.hh"
 #include "replica/database.hh"
 #include "types/types.hh"
@@ -219,24 +218,18 @@ view_ptr create_index_statement::create_view_for_index(const schema_ptr schema,
         std::map<sstring, sstring> tags_map = {{db::SYNCHRONOUS_VIEW_UPDATES_TAG_KEY, "true"}};
         builder.add_extension(db::tags_extension::NAME, ::make_shared<db::tags_extension>(tags_map));
     }
-
-    const schema::extensions_map exts = _view_properties.properties()->make_schema_extensions(db.extensions());
-    _view_properties.apply_to_builder(view_prop_defs::op_type::create, builder, exts, db, keyspace(), is_colocated);
-
     return view_ptr{builder.build()};
 }
 
 create_index_statement::create_index_statement(cf_name name,
                                                ::shared_ptr<index_name> index_name,
                                                std::vector<::shared_ptr<index_target::raw>> raw_targets,
-                                               ::shared_ptr<index_specific_prop_defs> idx_properties,
-                                               view_prop_defs view_properties,
+                                               ::shared_ptr<index_prop_defs> properties,
                                                bool if_not_exists)
     : schema_altering_statement(name)
     , _index_name(index_name->get_idx())
     , _raw_targets(raw_targets)
-    , _idx_properties(std::move(idx_properties))
-    , _view_properties(std::move(view_properties))
+    , _properties(properties)
     , _if_not_exists(if_not_exists)
 {
 }
@@ -259,53 +252,14 @@ static sstring target_type_name(index_target::target_type type) {
 void
 create_index_statement::validate(query_processor& qp, const service::client_state& state) const
 {
-    if (_raw_targets.empty() && !_idx_properties->is_custom) {
+    if (_raw_targets.empty() && !_properties->is_custom) {
         throw exceptions::invalid_request_exception("Only CUSTOM indexes can be created without specifying a target column");
     }
 
-    _idx_properties->validate();
-
-    // FIXME: This is ugly and can be improved.
-    const bool is_vector_index = _idx_properties->custom_class && *_idx_properties->custom_class == "vector_index";
-    const bool uses_view_properties = _view_properties.properties()->count() > 0
-            || _view_properties.use_compact_storage()
-            || _view_properties.defined_ordering().size() > 0;
-
-    if (is_vector_index && uses_view_properties) {
-        throw exceptions::invalid_request_exception("You cannot use view properties with a vector index");
-    }
-
-    const schema::extensions_map exts = _view_properties.properties()->make_schema_extensions(qp.db().extensions());
-    _view_properties.validate_raw(view_prop_defs::op_type::create, qp.db(), keyspace(), exts);
-
-    // These keywords are still accepted by other schema entities, but they don't have effect on them.
-    // Since indexes are not bound by any backward compatibility contract in this regard, let's forbid these.
-    static sstring obsolete_keywords[] = {
-        "index_interval",
-        "replicate_on_write",
-        "populate_io_cache_on_flush",
-        "read_repair_chance",
-        "dclocal_read_repair_chance",
-    };
-
-    for (const sstring& keyword : obsolete_keywords) {
-        if (_view_properties.properties()->has_property(keyword)) {
-            // We use the same type of exception and the same error message as would be thrown for
-            // an invalid property via `_view_properties.validate_raw`.
-            throw exceptions::syntax_exception(seastar::format("Unknown property '{}'", keyword));
-        }
-    }
-
-    // FIXME: This is a temporary limitation as it might deserve more attention.
-    if (!_view_properties.defined_ordering().empty()) {
-        throw exceptions::invalid_request_exception("Indexes do not allow for specifying the clustering order");
-    }
+    _properties->validate();
 }
 
-std::pair<std::vector<::shared_ptr<index_target>>, cql3::cql_warnings_vec>
-create_index_statement::validate_while_executing(data_dictionary::database db, locator::token_metadata_ptr tmptr) const {
-    cql3::cql_warnings_vec warnings;
-
+std::vector<::shared_ptr<index_target>> create_index_statement::validate_while_executing(data_dictionary::database db) const {
     auto schema = validation::validate_column_family(db, keyspace(), column_family());
 
     if (schema->is_counter()) {
@@ -325,24 +279,11 @@ create_index_statement::validate_while_executing(data_dictionary::database db, l
         throw exceptions::invalid_request_exception(format("index names shouldn't be more than {:d} characters long (got \"{}\")", schema::NAME_LENGTH, _index_name.c_str()));
     }
 
-    // Regular secondary indexes require rf-rack-validity.
-    // Custom indexes need to validate this property themselves, if they need it.
-    if (!_idx_properties || !_idx_properties->custom_class) {
-        try {
-            db::view::validate_view_keyspace(db, keyspace(), tmptr);
-        } catch (const std::exception& e) {
-            // The type of the thrown exception is not specified, so we need to wrap it here.
-            throw exceptions::invalid_request_exception(e.what());
-        }
-
-        if (db.find_keyspace(keyspace()).uses_tablets()) {
-            warnings.emplace_back(
-                "Creating an index in a keyspace that uses tablets requires "
-                "the keyspace to remain RF-rack-valid while the index exists. "
-                "Some operations will be restricted to enforce this: altering the keyspace's replication "
-                "factor, adding a node in a new rack, and removing or decommissioning a node that would "
-                "eliminate a rack.");
-        }
+    try {
+        db::view::validate_view_keyspace(db, keyspace());
+    } catch (const std::exception& e) {
+        // The type of the thrown exception is not specified, so we need to wrap it here.
+        throw exceptions::invalid_request_exception(e.what());
     }
 
     validate_for_local_index(*schema);
@@ -352,14 +293,14 @@ create_index_statement::validate_while_executing(data_dictionary::database db, l
         targets.emplace_back(raw_target->prepare(*schema));
     }
 
-    if (_idx_properties && _idx_properties->custom_class) {
-        auto custom_index_factory = secondary_index::secondary_index_manager::get_custom_class_factory(*_idx_properties->custom_class);
+    if (_properties && _properties->custom_class) {
+        auto custom_index_factory = secondary_index::secondary_index_manager::get_custom_class_factory(*_properties->custom_class);
         if (!custom_index_factory) {
-            throw exceptions::invalid_request_exception(format("Non-supported custom class \'{}\' provided", *_idx_properties->custom_class));
+            throw exceptions::invalid_request_exception(format("Non-supported custom class \'{}\' provided", *(_properties->custom_class)));
         }
         auto custom_index = (*custom_index_factory)();
-        custom_index->validate(*schema, *_idx_properties, targets, db.features(), db);
-        _idx_properties->index_version = custom_index->index_version(*schema);
+        custom_index->validate(*schema, *_properties, targets, db.features(), db);
+        _properties->index_version = custom_index->index_version(*schema);
     }
 
     if (targets.size() > 1) {
@@ -439,7 +380,7 @@ create_index_statement::validate_while_executing(data_dictionary::database db, l
         }
     }
 
-    return std::make_pair(std::move(targets), std::move(warnings));
+    return targets;
 }
 
 void create_index_statement::validate_for_local_index(const schema& schema) const {
@@ -578,7 +519,7 @@ void create_index_statement::validate_target_column_is_map_if_index_involves_key
 
 void create_index_statement::validate_targets_for_multi_column_index(std::vector<::shared_ptr<index_target>> targets) const
 {
-    if (!_idx_properties->is_custom) {
+    if (!_properties->is_custom) {
         if (targets.size() > 2 || (targets.size() == 2 && std::holds_alternative<index_target::single_column>(targets.front()->value))) {
             throw exceptions::invalid_request_exception("Only CUSTOM indexes support multiple columns");
         }
@@ -592,9 +533,8 @@ void create_index_statement::validate_targets_for_multi_column_index(std::vector
     }
 }
 
-std::pair<std::optional<create_index_statement::base_schema_with_new_index>, cql3::cql_warnings_vec>
-create_index_statement::build_index_schema(data_dictionary::database db, locator::token_metadata_ptr tmptr) const {
-    auto [targets, warnings] = validate_while_executing(db, tmptr);
+std::optional<create_index_statement::base_schema_with_new_index> create_index_statement::build_index_schema(data_dictionary::database db) const {
+    auto targets = validate_while_executing(db);
 
     auto schema = db.find_schema(keyspace(), column_family());
 
@@ -610,8 +550,8 @@ create_index_statement::build_index_schema(data_dictionary::database db, locator
     }
     index_metadata_kind kind;
     index_options_map index_options;
-    if (_idx_properties->custom_class) {
-        index_options = _idx_properties->get_options();
+    if (_properties->custom_class) {
+        index_options = _properties->get_options();
         kind = index_metadata_kind::custom;
     } else {
         kind = schema->is_compound() ? index_metadata_kind::composites : index_metadata_kind::keys;
@@ -620,17 +560,17 @@ create_index_statement::build_index_schema(data_dictionary::database db, locator
     auto existing_index = schema->find_index_noname(index);
     if (existing_index) {
         if (_if_not_exists) {
-            return std::make_pair(std::nullopt, std::move(warnings));
+            return {};
         } else {
             throw exceptions::invalid_request_exception(
                     format("Index {} is a duplicate of existing index {}", index.name(), existing_index.value().name()));
         }
     }
-    bool existing_vector_index = _idx_properties->custom_class && _idx_properties->custom_class == "vector_index" && secondary_index::vector_index::has_vector_index_on_column(*schema, targets[0]->column_name());
-    bool custom_index_with_same_name = _idx_properties->custom_class && db.existing_index_names(keyspace()).contains(_index_name);
+    bool existing_vector_index = _properties->custom_class && _properties->custom_class == "vector_index" && secondary_index::vector_index::has_vector_index_on_column(*schema, targets[0]->column_name());
+    bool custom_index_with_same_name = _properties->custom_class && db.existing_index_names(keyspace()).contains(_index_name);
     if (existing_vector_index || custom_index_with_same_name) {
         if (_if_not_exists) {
-            return std::make_pair(std::nullopt, std::move(warnings));
+            return {};
         } else {
             throw exceptions::invalid_request_exception("There exists a duplicate custom index");
         }
@@ -646,13 +586,13 @@ create_index_statement::build_index_schema(data_dictionary::database db, locator
     schema_builder builder{schema};
     builder.with_index(index);
 
-    return std::make_pair(base_schema_with_new_index{builder.build(), index}, std::move(warnings));
+    return base_schema_with_new_index{builder.build(), index};
 }
 
 future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, utils::chunked_vector<mutation>, cql3::cql_warnings_vec>>
 create_index_statement::prepare_schema_mutations(query_processor& qp, const query_options&, api::timestamp_type ts) const {
     using namespace cql_transport;
-    auto [res, warnings] = build_index_schema(qp.db(), qp.proxy().get_token_metadata_ptr());
+    auto res = build_index_schema(qp.db());
 
     ::shared_ptr<event::schema_change> ret;
     utils::chunked_vector<mutation> muts;
@@ -682,7 +622,7 @@ create_index_statement::prepare_schema_mutations(query_processor& qp, const quer
                 column_family());
     }
 
-    co_return std::make_tuple(std::move(ret), std::move(muts), std::move(warnings));
+    co_return std::make_tuple(std::move(ret), std::move(muts), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>

cql3/statements/create_index_statement.hh

@@ -10,8 +10,6 @@
 
 #pragma once
 
-#include "cql3/statements/index_prop_defs.hh"
-#include "cql3/statements/view_prop_defs.hh"
 #include "schema_altering_statement.hh"
 #include "index_target.hh"
 
@@ -29,25 +27,20 @@ class index_name;
 
 namespace statements {
 
-class index_specific_prop_defs;
+class index_prop_defs;
 
 /** A <code>CREATE INDEX</code> statement parsed from a CQL query. */
 class create_index_statement : public schema_altering_statement {
     const sstring _index_name;
     const std::vector<::shared_ptr<index_target::raw>> _raw_targets;
-
-    // Options specific to this index.
-    const ::shared_ptr<index_specific_prop_defs> _idx_properties;
-    // Options corresponding to the underlying materialized view.
-    const view_prop_defs _view_properties;
-
+    const ::shared_ptr<index_prop_defs> _properties;
     const bool _if_not_exists;
     cql_stats* _cql_stats = nullptr;
 
 public:
     create_index_statement(cf_name name, ::shared_ptr<index_name> index_name,
             std::vector<::shared_ptr<index_target::raw>> raw_targets,
-            ::shared_ptr<index_specific_prop_defs> idx_properties, view_prop_defs view_properties, bool if_not_exists);
+            ::shared_ptr<index_prop_defs> properties, bool if_not_exists);
 
     future<> check_access(query_processor& qp, const service::client_state& state) const override;
     void validate(query_processor&, const service::client_state& state) const override;
@@ -60,7 +53,7 @@ public:
         schema_ptr schema;
         index_metadata index;
     };
-    std::pair<std::optional<base_schema_with_new_index>, cql3::cql_warnings_vec> build_index_schema(data_dictionary::database db, locator::token_metadata_ptr tmptr) const;
+    std::optional<base_schema_with_new_index> build_index_schema(data_dictionary::database db) const;
     view_ptr create_view_for_index(const schema_ptr, const index_metadata& im, const data_dictionary::database&) const;
 private:
     void validate_for_local_index(const schema& schema) const;
@@ -76,7 +69,7 @@ private:
                                               const sstring& name,
                                               index_metadata_kind kind,
                                               const index_options_map& options);
-    std::pair<std::vector<::shared_ptr<index_target>>, cql3::cql_warnings_vec> validate_while_executing(data_dictionary::database db, locator::token_metadata_ptr tmptr) const;
+    std::vector<::shared_ptr<index_target>> validate_while_executing(data_dictionary::database db) const;
 };
 
 }

cql3/statements/create_keyspace_statement.cc

@@ -116,21 +116,21 @@ future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, utils::chun
             warnings.push_back("Keyspace `initial` tablets option is deprecated.  Use per-table tablet options instead.");
         }
 
-        // If RF-rack-validity must be enforced for the keyspace according to `enforce_rf_rack_validity_for_keyspace`,
-        // it's forbidden to create an RF-rack-invalid keyspace. Verify that it's RF-rack-valid.
+        // If `rf_rack_valid_keyspaces` is enabled, it's forbidden to create an RF-rack-invalid keyspace.
+        // Verify that it's RF-rack-valid.
         // For more context, see: scylladb/scylladb#23071.
         try {
             // We hold a group0_guard, so it's correct to check this here.
             // The topology or schema cannot change while we're performing this query.
             locator::assert_rf_rack_valid_keyspace(_name, tmptr, *rs);
         } catch (const std::exception& e) {
-            if (replica::database::enforce_rf_rack_validity_for_keyspace(cfg, *ksm)) {
+            if (cfg.rf_rack_valid_keyspaces()) {
                 // There's no guarantee what the type of the exception will be, so we need to
                 // wrap it manually here in a type that can be passed to the user.
                 throw exceptions::invalid_request_exception(e.what());
             } else {
-                // Even when RF-rack-validity is not enforced for the keyspace, we'd
-                // like to inform the user that the keyspace they're creating does not
+                // Even when the configuration option `rf_rack_valid_keyspaces` is set to false,
+                // we'd like to inform the user that the keyspace they're creating does not
                 // satisfy the restriction--but just as a warning.
                 // For more context, see issue: scylladb/scylladb#23330.
                 warnings.push_back(seastar::format(

cql3/statements/create_view_statement.cc

@@ -8,7 +8,6 @@
  * SPDX-License-Identifier: (LicenseRef-ScyllaDB-Source-Available-1.0 and Apache-2.0)
  */
 
-#include "cql3/statements/view_prop_defs.hh"
 #include "exceptions/exceptions.hh"
 #include "utils/assert.hh"
 #include <unordered_set>
@@ -106,7 +105,7 @@ static bool validate_primary_key(
     return new_non_pk_column;
 }
 
-std::pair<view_ptr, cql3::cql_warnings_vec> create_view_statement::prepare_view(data_dictionary::database db, locator::token_metadata_ptr tmptr) const {
+std::pair<view_ptr, cql3::cql_warnings_vec> create_view_statement::prepare_view(data_dictionary::database db) const {
     // We need to make sure that:
     //  - materialized view name is valid
     //  - primary key includes all columns in base table's primary key
@@ -120,7 +119,15 @@ std::pair<view_ptr, cql3::cql_warnings_vec> create_view_statement::prepare_view(
     cql3::cql_warnings_vec warnings;
 
     auto schema_extensions = _properties.properties()->make_schema_extensions(db.extensions());
-    _properties.validate_raw(view_prop_defs::op_type::create, db, keyspace(), schema_extensions);
+    _properties.validate(db, keyspace(), schema_extensions);
+
+    if (_properties.use_compact_storage()) {
+        throw exceptions::invalid_request_exception(format("Cannot use 'COMPACT STORAGE' when defining a materialized view"));
+    }
+
+    if (_properties.properties()->get_cdc_options(schema_extensions)) {
+        throw exceptions::invalid_request_exception("Cannot enable CDC for a materialized view");
+    }
 
     // View and base tables must be in the same keyspace, to ensure that RF
     // is the same (because we assign a view replica to each base replica).
@@ -146,21 +153,12 @@ std::pair<view_ptr, cql3::cql_warnings_vec> create_view_statement::prepare_view(
     schema_ptr schema = validation::validate_column_family(db, _base_name.get_keyspace(), _base_name.get_column_family());
 
     try {
-        db::view::validate_view_keyspace(db, keyspace(), tmptr);
+        db::view::validate_view_keyspace(db, keyspace());
     } catch (const std::exception& e) {
         // The type of the thrown exception is not specified, so we need to wrap it here.
         throw exceptions::invalid_request_exception(e.what());
     }
 
-    if (db.find_keyspace(keyspace()).uses_tablets()) {
-        warnings.emplace_back(
-            "Creating a materialized view in a keyspaces that uses tablets requires "
-            "the keyspace to remain RF-rack-valid while the materialized view exists. "
-            "Some operations will be restricted to enforce this: altering the keyspace's replication "
-            "factor, adding a node in a new rack, and removing or decommissioning a node that would "
-            "eliminate a rack.");
-    }
-
     if (schema->is_counter()) {
         throw exceptions::invalid_request_exception(format("Materialized views are not supported on counter tables"));
     }
@@ -343,7 +341,16 @@ std::pair<view_ptr, cql3::cql_warnings_vec> create_view_statement::prepare_view(
         warnings.emplace_back(std::move(warning_text));
     }
 
-    schema_builder builder{keyspace(), column_family()};
+    const auto maybe_id = _properties.properties()->get_id();
+    if (maybe_id && db.try_find_table(*maybe_id)) {
+        const auto schema_ptr = db.find_schema(*maybe_id);
+        const auto& ks_name = schema_ptr->ks_name();
+        const auto& cf_name = schema_ptr->cf_name();
+
+        throw exceptions::invalid_request_exception(seastar::format("Table with ID {} already exists: {}.{}", *maybe_id, ks_name, cf_name));
+    }
+
+    schema_builder builder{keyspace(), column_family(), maybe_id};
     auto add_columns = [this, &builder] (std::vector<const column_definition*>& defs, column_kind kind) mutable {
         for (auto* def : defs) {
             auto&& type = _properties.get_reversable_type(*def->column_specification->name, def->type);
@@ -389,8 +396,14 @@ std::pair<view_ptr, cql3::cql_warnings_vec> create_view_statement::prepare_view(
         }
     }
 
-    _properties.apply_to_builder(view_prop_defs::op_type::create, builder, std::move(schema_extensions),
-            db, keyspace(), is_colocated);
+    _properties.properties()->apply_to_builder(builder, std::move(schema_extensions), db, keyspace(), !is_colocated);
+
+    if (builder.default_time_to_live().count() > 0) {
+        throw exceptions::invalid_request_exception(
+                "Cannot set or alter default_time_to_live for a materialized view. "
+                "Data in a materialized view always expire at the same time than "
+                "the corresponding data in the parent table.");
+    }
 
     auto where_clause_text = util::relations_to_where_clause(_where_clause);
     builder.with_view_info(schema, included.empty(), std::move(where_clause_text));
@@ -401,7 +414,7 @@ std::pair<view_ptr, cql3::cql_warnings_vec> create_view_statement::prepare_view(
 future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, utils::chunked_vector<mutation>, cql3::cql_warnings_vec>>
 create_view_statement::prepare_schema_mutations(query_processor& qp, const query_options&, api::timestamp_type ts) const {
     utils::chunked_vector<mutation> m;
-    auto [definition, warnings] = prepare_view(qp.db(), qp.proxy().get_token_metadata_ptr());
+    auto [definition, warnings] = prepare_view(qp.db());
     try {
         m = co_await service::prepare_new_view_announcement(qp.proxy(), std::move(definition), ts);
     } catch (const exceptions::already_exists_exception& e) {

cql3/statements/create_view_statement.hh

@@ -7,9 +7,9 @@
 #pragma once
 
 #include "cql3/statements/schema_altering_statement.hh"
+#include "cql3/statements/cf_properties.hh"
 #include "cql3/cf_name.hh"
 #include "cql3/expr/expression.hh"
-#include "cql3/statements/view_prop_defs.hh"
 
 #include <seastar/core/shared_ptr.hh>
 
@@ -35,7 +35,7 @@ private:
     expr::expression _where_clause;
     std::vector<::shared_ptr<cql3::column_identifier::raw>> _partition_keys;
     std::vector<::shared_ptr<cql3::column_identifier::raw>> _clustering_keys;
-    view_prop_defs _properties;
+    cf_properties _properties;
     bool _if_not_exists;
 
 public:
@@ -48,7 +48,7 @@ public:
             std::vector<::shared_ptr<cql3::column_identifier::raw>> clustering_keys,
             bool if_not_exists);
 
-    std::pair<view_ptr, cql3::cql_warnings_vec> prepare_view(data_dictionary::database db, locator::token_metadata_ptr tmptr) const;
+    std::pair<view_ptr, cql3::cql_warnings_vec> prepare_view(data_dictionary::database db) const;
 
     auto& properties() {
         return _properties;

cql3/statements/describe_statement.cc

@@ -710,12 +710,11 @@ std::vector<lw_shared_ptr<column_specification>> listing_describe_statement::get
 
 future<std::vector<std::vector<managed_bytes_opt>>> listing_describe_statement::describe(cql3::query_processor& qp, const service::client_state& client_state) const {
     auto db = qp.db();
+    auto raw_ks = client_state.get_raw_keyspace();
+
     std::vector<sstring> keyspaces;
-    // For most describe statements we should limit the results to the USEd
-    // keyspace (client_state.get_raw_keyspace()), if any. However for DESC
-    // KEYSPACES we must list all keyspaces, not just the USEd one.
-    if (_element != element_type::keyspace && !client_state.get_raw_keyspace().empty()) {
-        keyspaces.push_back(client_state.get_raw_keyspace());
+    if (!raw_ks.empty()) {
+        keyspaces.push_back(raw_ks);
     } else {
         keyspaces = db.get_all_keyspaces();
         std::ranges::sort(keyspaces);

cql3/statements/index_prop_defs.cc

@@ -11,7 +11,6 @@
 #include <set>
 #include <seastar/core/format.hh>
 #include "index_prop_defs.hh"
-#include "cql3/statements/view_prop_defs.hh"
 #include "index/secondary_index.hh"
 #include "exceptions/exceptions.hh"
 
@@ -22,9 +21,7 @@ static void check_system_option_specified(const index_options_map& options, cons
     }
 }
 
-namespace cql3::statements {
-    
-void index_specific_prop_defs::validate() const {
+void cql3::statements::index_prop_defs::validate() const {
     static std::set<sstring> keywords({ sstring(KW_OPTIONS) });
 
     property_definitions::validate(keywords);
@@ -43,13 +40,13 @@ void index_specific_prop_defs::validate() const {
 }
 
 index_options_map
-index_specific_prop_defs::get_raw_options() const {
+cql3::statements::index_prop_defs::get_raw_options() const {
     auto options = get_map(KW_OPTIONS);
     return !options ? std::unordered_map<sstring, sstring>() : std::unordered_map<sstring, sstring>(options->begin(), options->end());
 }
 
 index_options_map
-index_specific_prop_defs::get_options() const {
+cql3::statements::index_prop_defs::get_options() const {
     auto options = get_raw_options();
     options.emplace(db::index::secondary_index::custom_class_option_name, *custom_class);
     if (index_version.has_value()) {
@@ -57,25 +54,3 @@ index_specific_prop_defs::get_options() const {
     }
     return options;
 }
-
-void index_prop_defs::extract_index_specific_properties_to(index_specific_prop_defs& target) {
-    if (properties()->has_property(index_specific_prop_defs::KW_OPTIONS)) {
-        auto value = properties()->extract_property(index_specific_prop_defs::KW_OPTIONS);
-
-        std::visit([&target] <typename T> (T&& val) {
-            target.add_property(index_specific_prop_defs::KW_OPTIONS, std::forward<T>(val));
-        }, std::move(value));
-    }
-}
-
-view_prop_defs index_prop_defs::into_view_prop_defs() && {
-    if (properties()->has_property(index_specific_prop_defs::KW_OPTIONS)) {
-        utils::on_internal_error(seastar::format(
-                "Precondition has been violated. The property '{}' is still present", index_specific_prop_defs::KW_OPTIONS));
-    }
-
-    view_prop_defs result = std::move(static_cast<view_prop_defs&>(*this));
-    return result;
-}
-
-} // namespace cql3::statements