database: Use keyspace_metadata for creation functions

Use the keyspace_metadata type for keyspace creation functions. This is
needed to be able to have a mapping from keyspace name to keyspace
metadata for various call-sites.

Signed-off-by: Pekka Enberg <penberg@cloudius-systems.com>

database.cc

@@ -429,11 +429,11 @@ void database::add_keyspace(sstring name, keyspace k) {
 }
 
 future<>
-create_keyspace(distributed<database>& db, sstring name) {
-    return make_directory(db.local()._cfg->data_file_directories() + "/" + name).then([name, &db] {
-        return db.invoke_on_all([&name] (database& db) {
-            auto cfg = db.make_keyspace_config(name);
-            db.add_keyspace(name, keyspace(cfg));
+create_keyspace(distributed<database>& db, const keyspace_metadata& ksm) {
+    return make_directory(db.local()._cfg->data_file_directories() + "/" + ksm.name()).then([&ksm, &db] {
+        return db.invoke_on_all([&ksm] (database& db) {
+            auto cfg = db.make_keyspace_config(ksm);
+            db.add_keyspace(ksm.name(), keyspace(cfg));
         });
     });
     // FIXME: rollback on error, or keyspace directory remains on disk, poisoning
@@ -588,12 +588,12 @@ schema_ptr database::find_schema(const utils::UUID& uuid) const throw (no_such_c
 }
 
 keyspace&
-database::find_or_create_keyspace(const sstring& name) {
-    auto i = _keyspaces.find(name);
+database::find_or_create_keyspace(const keyspace_metadata& ksm) {
+    auto i = _keyspaces.find(ksm.name());
     if (i != _keyspaces.end()) {
         return i->second;
     }
-    return _keyspaces.emplace(name, keyspace(make_keyspace_config(name))).first->second;
+    return _keyspaces.emplace(ksm.name(), keyspace(make_keyspace_config(ksm))).first->second;
 }
 
 void
@@ -755,9 +755,9 @@ future<> database::apply(const frozen_mutation& m) {
 }
 
 keyspace::config
-database::make_keyspace_config(sstring name) const {
+database::make_keyspace_config(const keyspace_metadata& ksm) const {
     keyspace::config cfg;
-    cfg.datadir = sprint("%s/%s", _cfg->data_file_directories(), name);
+    cfg.datadir = sprint("%s/%s", _cfg->data_file_directories(), ksm.name());
     return cfg;
 }
 

database.hh

@@ -255,7 +255,7 @@ public:
     const utils::UUID& find_uuid(const schema_ptr&) const throw (std::out_of_range);
 
     /* below, find* throws no_such_<type> on fail */
-    keyspace& find_or_create_keyspace(const sstring& name);
+    keyspace& find_or_create_keyspace(const keyspace_metadata&);
     keyspace& find_keyspace(const sstring& name) throw (no_such_keyspace);
     const keyspace& find_keyspace(const sstring& name) const throw (no_such_keyspace);
     bool has_keyspace(const sstring& name) const;
@@ -275,14 +275,14 @@ public:
     unsigned shard_of(const frozen_mutation& m);
     future<lw_shared_ptr<query::result>> query(const query::read_command& cmd);
     future<> apply(const frozen_mutation&);
-    keyspace::config make_keyspace_config(sstring name) const;
+    keyspace::config make_keyspace_config(const keyspace_metadata& ksm) const;
     friend std::ostream& operator<<(std::ostream& out, const database& db);
-    friend future<> create_keyspace(distributed<database>&, sstring);
+    friend future<> create_keyspace(distributed<database>&, const keyspace_metadata&);
 };
 
 // Creates a keyspace.  Keyspaces have a non-sharded
 // component (the directory), so a global function is needed.
-future<> create_keyspace(distributed<database>& db, sstring name);
+future<> create_keyspace(distributed<database>& db, const keyspace_metadata&);
 
 // FIXME: stub
 class secondary_index_manager {};

db/legacy_schema_tables.cc

@@ -586,7 +586,7 @@ std::vector<const char*> ALL { KEYSPACES, COLUMNFAMILIES, COLUMNS, TRIGGERS, USE
             return proxy.get_db().invoke_on_all([&created, altered = std::move(altered)] (database& db) {
                 for (auto&& kv : created) {
                     auto ksm = create_keyspace_from_schema_partition(kv);
-                    keyspace k(db.make_keyspace_config(ksm->name()));
+                    keyspace k(db.make_keyspace_config(*ksm));
                     k.create_replication_strategy(*ksm);
                     db.add_keyspace(ksm->name(), std::move(k));
                 }

db/system_keyspace.cc

@@ -1001,7 +1001,12 @@ std::vector<schema_ptr> all_tables() {
 }
 
 void make(database& db) {
-    auto kscfg = db.make_keyspace_config("system");
+    keyspace_metadata ksm{"system",
+            "org.apache.cassandra.locator.SimpleStrategy",
+            std::unordered_map<sstring, sstring>{},
+            false
+            };
+    auto kscfg = db.make_keyspace_config(ksm);
     keyspace ks{std::move(kscfg)};
     db.add_keyspace("system", std::move(ks));
     for (auto&& table : all_tables()) {

tests/urchin/cql_query_test.cc

@@ -791,8 +791,13 @@ SEASTAR_TEST_CASE(test_user_type) {
                 {int32_type, long_type, utf8_type});
     };
     return do_with_cql_env([make_user_type] (cql_test_env& e) {
+        keyspace_metadata ksm{"ks",
+                "org.apache.cassandra.locator.SimpleStrategy",
+                std::unordered_map<sstring, sstring>{},
+                false
+                };
         // We don't have "CREATE TYPE" yet, so we must insert the type manually
-        e.local_db().find_or_create_keyspace("ks")._user_types.add_type(make_user_type());
+        e.local_db().find_or_create_keyspace(ksm)._user_types.add_type(make_user_type());
         return e.create_table([make_user_type] (auto ks_name) {
             // CQL: "create table cf (id int primary key, t ut1)";
             return schema({}, ks_name, "cf",

tests/urchin/cql_test_env.cc

@@ -92,14 +92,14 @@ public:
         return _db->invoke_on_all([schema_maker, id, this] (database& db) {
             auto cf_schema = make_lw_shared(schema_maker(ks_name));
             cf_schema->set_id(id);
-            auto& ks = db.find_or_create_keyspace(ks_name);
-            auto cfg = ks.make_column_family_config(*cf_schema);
-            db.add_column_family(column_family(std::move(cf_schema), std::move(cfg)));
             keyspace_metadata ksm(ks_name,
                     "org.apache.cassandra.locator.SimpleStrategy",
                     std::unordered_map<sstring, sstring>(),
                     false
                     );
+            auto& ks = db.find_or_create_keyspace(ksm);
+            auto cfg = ks.make_column_family_config(*cf_schema);
+            db.add_column_family(column_family(std::move(cf_schema), std::move(cfg)));
             ks.create_replication_strategy(ksm);
         });
     }

thrift/handler.cc

@@ -428,7 +428,12 @@ public:
         boost::for_each(ks_def.cf_defs, [&cf_ids] (auto&&) {
             cf_ids.push_back(utils::UUID_gen::get_time_UUID());
         });
-        create_keyspace(_db, ks_def.name).then([this, ks_def, cf_ids] {
+        keyspace_metadata ksm(to_sstring(ks_def.name),
+                to_sstring(ks_def.strategy_class),
+                std::unordered_map<sstring, sstring>{ks_def.strategy_options.begin(), ks_def.strategy_options.end()},
+                ks_def.durable_writes,
+                std::vector<schema_ptr>{}); // FIXME
+        create_keyspace(_db, ksm).then([this, ks_def, cf_ids] {
             return parallel_for_each(boost::combine(ks_def.cf_defs, cf_ids), [this, ks_def, cf_ids] (auto&& cf_def_and_id) {
                 // We create the directory on the local shard, since the same directory is
                 // used for all shards.