database,cql3: add STORAGE option to keyspaces

The STORAGE option is designed to hold a map of options
used for customizing storage for given keyspace.
The option is kept in a system_schema.scylla_keyspaces table.
The option is only available if the whole cluster is aware
of it - guarded by a cluster feature.

Example of the table contents:
```
cassandra@cqlsh> select * from system_schema.scylla_keyspaces;

 keyspace_name | storage_options                                | storage_type
---------------+------------------------------------------------+--------------
           ksx | {'bucket': '/tmp/xx', 'endpoint': 'localhost'} |           S3
```

cql3/statements/alter_keyspace_statement.cc

@@ -22,6 +22,7 @@
 #include "cql3/query_processor.hh"
 #include "cql3/statements/ks_prop_defs.hh"
 #include "create_keyspace_statement.hh"
+#include "gms/feature_service.hh"
 
 bool is_system_keyspace(std::string_view keyspace);
 
@@ -50,6 +51,20 @@ void cql3::statements::alter_keyspace_statement::validate(query_processor& qp, c
         if (!bool(_attrs->get_replication_strategy_class()) && !_attrs->get_replication_options().empty()) {
             throw exceptions::configuration_exception("Missing replication strategy class");
         }
+        try {
+            data_dictionary::storage_options current_options = qp.db().find_keyspace(_name).metadata()->get_storage_options();
+            data_dictionary::storage_options new_options = _attrs->get_storage_options();
+            if (!current_options.can_update_to(new_options)) {
+                throw exceptions::invalid_request_exception(format("Cannot alter storage options: {} to {} is not supported",
+                        current_options.type_string(), new_options.type_string()));
+            }
+        } catch (const std::runtime_error& e) {
+            throw exceptions::invalid_request_exception(e.what());
+        }
+        if (!qp.proxy().features().cluster_supports_keyspace_storage_options()
+            && _attrs->get_storage_options().type_string() != "LOCAL") {
+        throw exceptions::invalid_request_exception("Keyspace storage options not supported in the cluster");
+    }
 #if 0
         // The strategy is validated through KSMetaData.validate() in announceKeyspaceUpdate below.
         // However, for backward compatibility with thrift, this doesn't validate unexpected options yet,

cql3/statements/create_keyspace_statement.cc

@@ -23,6 +23,7 @@
 #include "transport/messages/result_message.hh"
 #include "cql3/query_processor.hh"
 #include "db/config.hh"
+#include "gms/feature_service.hh"
 
 #include <regex>
 
@@ -51,7 +52,7 @@ future<> create_keyspace_statement::check_access(query_processor& qp, const serv
     return state.has_all_keyspaces_access(auth::permission::CREATE);
 }
 
-void create_keyspace_statement::validate(query_processor&, const service::client_state& state) const
+void create_keyspace_statement::validate(query_processor& qp, const service::client_state& state) const
 {
     std::string name;
     name.resize(_name.length());
@@ -73,6 +74,15 @@ void create_keyspace_statement::validate(query_processor&, const service::client
     if (!bool(_attrs->get_replication_strategy_class())) {
         throw exceptions::configuration_exception("Missing mandatory replication strategy class");
     }
+    try {
+        _attrs->get_storage_options();
+    } catch (const std::runtime_error& e) {
+        throw exceptions::invalid_request_exception(e.what());
+    }
+    if (!qp.proxy().features().cluster_supports_keyspace_storage_options()
+            && _attrs->get_storage_options().type_string() != "LOCAL") {
+        throw exceptions::invalid_request_exception("Keyspace storage options not supported in the cluster");
+    }
 #if 0
     // The strategy is validated through KSMetaData.validate() in announceNewKeyspace below.
     // However, for backward compatibility with thrift, this doesn't validate unexpected options yet,

cql3/statements/ks_prop_defs.cc

@@ -79,7 +79,7 @@ void ks_prop_defs::validate() {
         return;
     }
 
-    static std::set<sstring> keywords({ sstring(KW_DURABLE_WRITES), sstring(KW_REPLICATION) });
+    static std::set<sstring> keywords({ sstring(KW_DURABLE_WRITES), sstring(KW_REPLICATION), sstring(KW_STORAGE) });
     property_definitions::validate(keywords);
 
     auto replication_options = get_replication_options();
@@ -96,14 +96,28 @@ std::map<sstring, sstring> ks_prop_defs::get_replication_options() const {
     return std::map<sstring, sstring>{};
 }
 
+data_dictionary::storage_options ks_prop_defs::get_storage_options() const {
+    data_dictionary::storage_options opts;
+    auto options_map = get_map(KW_STORAGE);
+    if (options_map) {
+        auto it = options_map->find("type");
+        if (it != options_map->end()) {
+            sstring storage_type = it->second;
+            options_map->erase(it);
+            opts.value = data_dictionary::storage_options::from_map(storage_type, std::move(*options_map));
+        }
+    }
+    return opts;
+}
+
 std::optional<sstring> ks_prop_defs::get_replication_strategy_class() const {
     return _strategy_class;
 }
 
 lw_shared_ptr<data_dictionary::keyspace_metadata> ks_prop_defs::as_ks_metadata(sstring ks_name, const locator::token_metadata& tm) {
     auto sc = get_replication_strategy_class().value();
     return data_dictionary::keyspace_metadata::new_keyspace(ks_name, sc,
-            prepare_options(sc, tm, get_replication_options()), get_boolean(KW_DURABLE_WRITES, true));
+            prepare_options(sc, tm, get_replication_options()), get_boolean(KW_DURABLE_WRITES, true), std::vector<schema_ptr>{}, get_storage_options());
 }
 
 lw_shared_ptr<data_dictionary::keyspace_metadata> ks_prop_defs::as_ks_metadata_update(lw_shared_ptr<data_dictionary::keyspace_metadata> old, const locator::token_metadata& tm) {
@@ -117,7 +131,7 @@ lw_shared_ptr<data_dictionary::keyspace_metadata> ks_prop_defs::as_ks_metadata_u
         options = old_options;
     }
 
-    return data_dictionary::keyspace_metadata::new_keyspace(old->name(), *sc, options, get_boolean(KW_DURABLE_WRITES, true));
+    return data_dictionary::keyspace_metadata::new_keyspace(old->name(), *sc, options, get_boolean(KW_DURABLE_WRITES, true), std::vector<schema_ptr>{}, get_storage_options());
 }
 
 

cql3/statements/ks_prop_defs.hh

@@ -14,6 +14,7 @@
 #pragma once
 
 #include "cql3/statements/property_definitions.hh"
+#include "data_dictionary/storage_options.hh"
 
 #include <seastar/core/shared_ptr.hh>
 #include <seastar/core/sstring.hh>
@@ -38,6 +39,7 @@ class ks_prop_defs : public property_definitions {
 public:
     static constexpr auto KW_DURABLE_WRITES = "durable_writes";
     static constexpr auto KW_REPLICATION = "replication";
+    static constexpr auto KW_STORAGE = "storage";
 
     static constexpr auto REPLICATION_STRATEGY_CLASS_KEY = "class";
     static constexpr auto REPLICATION_FACTOR_KEY = "replication_factor";
@@ -47,6 +49,7 @@ public:
     void validate();
     std::map<sstring, sstring> get_replication_options() const;
     std::optional<sstring> get_replication_strategy_class() const;
+    data_dictionary::storage_options get_storage_options() const;
     lw_shared_ptr<data_dictionary::keyspace_metadata> as_ks_metadata(sstring ks_name, const locator::token_metadata&);
     lw_shared_ptr<data_dictionary::keyspace_metadata> as_ks_metadata_update(lw_shared_ptr<data_dictionary::keyspace_metadata> old, const locator::token_metadata&);
 

data_dictionary/data_dictionary.cc

@@ -11,10 +11,13 @@
 #include "user_types_metadata.hh"
 #include "keyspace_metadata.hh"
 #include "schema.hh"
+#include "utils/overloaded_functor.hh"
 #include <fmt/core.h>
 #include <ostream>
 #include <boost/range/adaptor/map.hpp>
 #include <boost/range/adaptor/filtered.hpp>
+#include <boost/algorithm/string/join.hpp>
+#include <array>
 
 namespace data_dictionary {
 
@@ -196,11 +199,27 @@ keyspace_metadata::keyspace_metadata(std::string_view name,
              bool durable_writes,
              std::vector<schema_ptr> cf_defs,
              user_types_metadata user_types)
+    : keyspace_metadata(name,
+                        strategy_name,
+                        std::move(strategy_options),
+                        durable_writes,
+                        std::move(cf_defs),
+                        user_types_metadata{},
+                        storage_options{}) { }
+
+keyspace_metadata::keyspace_metadata(std::string_view name,
+             std::string_view strategy_name,
+             locator::replication_strategy_config_options strategy_options,
+             bool durable_writes,
+             std::vector<schema_ptr> cf_defs,
+             user_types_metadata user_types,
+             storage_options storage_opts)
     : _name{name}
     , _strategy_name{locator::abstract_replication_strategy::to_qualified_class_name(strategy_name.empty() ? "NetworkTopologyStrategy" : strategy_name)}
     , _strategy_options{std::move(strategy_options)}
     , _durable_writes{durable_writes}
     , _user_types{std::move(user_types)}
+    , _storage_options{std::move(storage_opts)}
 {
     for (auto&& s : cf_defs) {
         _cf_meta_data.emplace(s->cf_name(), s);
@@ -217,9 +236,10 @@ keyspace_metadata::new_keyspace(std::string_view name,
                                 std::string_view strategy_name,
                                 locator::replication_strategy_config_options options,
                                 bool durables_writes,
-                                std::vector<schema_ptr> cf_defs)
+                                std::vector<schema_ptr> cf_defs,
+                                storage_options storage_opts)
 {
-    return ::make_lw_shared<keyspace_metadata>(name, strategy_name, options, durables_writes, cf_defs);
+    return ::make_lw_shared<keyspace_metadata>(name, strategy_name, options, durables_writes, cf_defs, user_types_metadata{}, storage_opts);
 }
 
 void keyspace_metadata::add_user_type(const user_type ut) {
@@ -243,6 +263,64 @@ std::vector<view_ptr> keyspace_metadata::views() const {
             | boost::adaptors::transformed([] (auto&& s) { return view_ptr(s); }));
 }
 
+storage_options::value_type storage_options::from_map(std::string_view type, std::map<sstring, sstring> values) {
+    if (type == "LOCAL") {
+        if (!values.empty()) {
+            throw std::runtime_error("Local storage does not accept any custom options");
+        }
+        return local{};
+    }
+    if (type == "S3") {
+        s3 options;
+        const std::array<std::pair<sstring, sstring*>, 2> allowed_options {
+            std::make_pair("bucket", &options.bucket),
+            std::make_pair("endpoint", &options.endpoint),
+        };
+        for (auto& option : allowed_options) {
+            if (auto it = values.find(option.first); it != values.end()) {
+                *option.second = it->second;
+            } else {
+                throw std::runtime_error(format("Missing S3 option: {}", option.first));
+            }
+        }
+        if (values.size() > allowed_options.size()) {
+            throw std::runtime_error(format("Extraneous options for S3: {}; allowed: {}",
+                    boost::algorithm::join(values | boost::adaptors::map_keys, ","),
+                    boost::algorithm::join(allowed_options | boost::adaptors::map_keys, ",")));
+        }
+        return options;
+    }
+    throw std::runtime_error(format("Unknown storage type: {}", type));
+}
+
+std::string_view storage_options::type_string() const {
+    return std::visit(overloaded_functor {
+        [] (const storage_options::local&) {
+            return "LOCAL";
+        },
+        [] (const storage_options::s3&) {
+            return "S3";
+        }
+    }, value);
+}
+
+std::map<sstring, sstring> storage_options::to_map() const {
+    std::map<sstring, sstring> ret;
+    std::visit(overloaded_functor {
+        [] (const storage_options::local&) {
+        },
+        [&ret] (const storage_options::s3& v) {
+            ret.emplace("bucket", v.bucket);
+            ret.emplace("endpoint", v.endpoint);
+        }
+    }, value);
+    return ret;
+}
+
+bool storage_options::can_update_to(const storage_options& new_options) {
+    return value == new_options.value;
+}
+
 no_such_keyspace::no_such_keyspace(std::string_view ks_name)
     : runtime_error{format("Can't find a keyspace {}", ks_name)}
 {

data_dictionary/keyspace_metadata.hh

@@ -16,6 +16,7 @@
 #include "schema.hh"
 #include "locator/abstract_replication_strategy.hh"
 #include "data_dictionary/user_types_metadata.hh"
+#include "data_dictionary/storage_options.hh"
 
 namespace data_dictionary {
 
@@ -26,6 +27,7 @@ class keyspace_metadata final {
     std::unordered_map<sstring, schema_ptr> _cf_meta_data;
     bool _durable_writes;
     user_types_metadata _user_types;
+    storage_options _storage_options;
 public:
     keyspace_metadata(std::string_view name,
                  std::string_view strategy_name,
@@ -38,12 +40,20 @@ public:
                  bool durable_writes,
                  std::vector<schema_ptr> cf_defs,
                  user_types_metadata user_types);
+    keyspace_metadata(std::string_view name,
+                 std::string_view strategy_name,
+                 locator::replication_strategy_config_options strategy_options,
+                 bool durable_writes,
+                 std::vector<schema_ptr> cf_defs,
+                 user_types_metadata user_types,
+                 storage_options storage_opts);
     static lw_shared_ptr<keyspace_metadata>
     new_keyspace(std::string_view name,
                  std::string_view strategy_name,
                  locator::replication_strategy_config_options options,
                  bool durables_writes,
-                 std::vector<schema_ptr> cf_defs = std::vector<schema_ptr>{});
+                 std::vector<schema_ptr> cf_defs = std::vector<schema_ptr>{},
+                 storage_options storage_opts = {});
     void validate(const locator::topology&) const;
     const sstring& name() const {
         return _name;
@@ -66,6 +76,9 @@ public:
     const user_types_metadata& user_types() const {
         return _user_types;
     }
+    const storage_options& get_storage_options() const {
+        return _storage_options;
+    }
     void add_or_update_column_family(const schema_ptr& s) {
         _cf_meta_data[s->cf_name()] = s;
     }

data_dictionary/storage_options.hh

@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2021-present ScyllaDB
+ */
+
+/*
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+#pragma once
+
+#include <string>
+#include <map>
+#include <seastar/core/sstring.hh>
+
+namespace data_dictionary {
+
+struct storage_options {
+    struct local {
+        friend auto operator<=>(const local&, const local&) = default;
+    };
+    struct s3 {
+        sstring bucket;
+        sstring endpoint;
+
+        friend auto operator<=>(const s3&, const s3&) = default;
+    };
+    using value_type = std::variant<local, s3>;
+    value_type value = local{};
+
+    storage_options() = default;
+
+    std::string_view type_string() const;
+    std::map<sstring, sstring> to_map() const;
+
+    bool can_update_to(const storage_options& new_options);
+
+    static value_type from_map(std::string_view type, std::map<sstring, sstring> values);
+};
+
+} // namespace data_dictionary

db/legacy_schema_migrator.cc

@@ -560,7 +560,7 @@ class migrator {
                             , ks.durable_writes);
 
             // we want separate time stamps for tables/types, so cannot bulk them into the ksm.
-            for (auto&& m : db::schema_tables::make_create_keyspace_mutations(ksm, ks.timestamp.time_since_epoch().count(), false)) {
+            for (auto&& m : db::schema_tables::make_create_keyspace_mutations(schema_features::full(), ksm, ks.timestamp.time_since_epoch().count(), false)) {
                 mutations.emplace_back(std::move(m));
             }
             for (auto& t : ks.tables) {