From 3520c786bd8eaf15c556370bce8e37a306738017 Mon Sep 17 00:00:00 2001
From: Benny Halevy <bhalevy@scylladb.com>
Date: Wed, 7 Jun 2023 12:31:26 +0300
Subject: [PATCH] database: create_keyspace_on_all_shards

Part of moving the responsibility for applying
and notifying keyspace schema changes from
schema_tables to the database so that the
database can control the order of applying the changes
across shards and when to notify its listeners.

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
---
 data_dictionary/data_dictionary.cc   |  5 +++++
 data_dictionary/keyspace_metadata.hh |  2 ++
 db/schema_tables.cc                  | 13 +++++--------
 replica/database.cc                  | 13 ++++++++-----
 replica/database.hh                  |  2 +-
 5 files changed, 21 insertions(+), 14 deletions(-)

diff --git a/data_dictionary/data_dictionary.cc b/data_dictionary/data_dictionary.cc
index 35b4b512ad..4dd80e9bb8 100644
--- a/data_dictionary/data_dictionary.cc
+++ b/data_dictionary/data_dictionary.cc
@@ -259,6 +259,11 @@ keyspace_metadata::new_keyspace(std::string_view name,
     return ::make_lw_shared<keyspace_metadata>(name, strategy_name, options, durables_writes, cf_defs, user_types_metadata{}, storage_opts);
 }
 
+lw_shared_ptr<keyspace_metadata>
+keyspace_metadata::new_keyspace(const keyspace_metadata& ksm) {
+    return new_keyspace(ksm.name(), ksm.strategy_name(), ksm.strategy_options(), ksm.durable_writes(), std::vector<schema_ptr>{}, ksm.get_storage_options());
+}
+
 void keyspace_metadata::add_user_type(const user_type ut) {
     _user_types.add_type(ut);
 }
diff --git a/data_dictionary/keyspace_metadata.hh b/data_dictionary/keyspace_metadata.hh
index 273bf0c0e8..05ee218d3d 100644
--- a/data_dictionary/keyspace_metadata.hh
+++ b/data_dictionary/keyspace_metadata.hh
@@ -56,6 +56,8 @@ public:
                  bool durables_writes,
                  std::vector<schema_ptr> cf_defs = std::vector<schema_ptr>{},
                  storage_options storage_opts = {});
+    static lw_shared_ptr<keyspace_metadata>
+    new_keyspace(const keyspace_metadata& ksm);
     void validate(const gms::feature_service&, const locator::topology&) const;
     const sstring& name() const {
         return _name;
diff --git a/db/schema_tables.cc b/db/schema_tables.cc
index a55eaf02f3..b48b6d6ad0 100644
--- a/db/schema_tables.cc
+++ b/db/schema_tables.cc
@@ -1361,14 +1361,11 @@ future<std::set<sstring>> merge_keyspaces(distributed<service::storage_proxy>& p
         altered.emplace_back(key);
     }
     auto& sharded_db = proxy.local().get_db();
-    co_await sharded_db.invoke_on_all([&] (replica::database& db) -> future<> {
-        for (auto&& val : created) {
-            auto scylla_specific_rs = co_await extract_scylla_specific_keyspace_info(proxy, val);
-            auto ksm = create_keyspace_from_schema_partition(val, std::move(scylla_specific_rs));
-            co_await db.create_keyspace(ksm, proxy.local().get_erm_factory());
-            co_await db.get_notifier().create_keyspace(ksm);
-        }
-    });
+    for (auto&& val : created) {
+        auto scylla_specific_rs = co_await extract_scylla_specific_keyspace_info(proxy, val);
+        auto ksm = create_keyspace_from_schema_partition(val, std::move(scylla_specific_rs));
+        co_await replica::database::create_keyspace_on_all_shards(sharded_db, proxy, *ksm);
+    }
     for (auto& name : altered) {
         co_await replica::database::update_keyspace_on_all_shards(sharded_db, proxy, name);
     }
diff --git a/replica/database.cc b/replica/database.cc
index 120c3182c8..960fb56d2a 100644
--- a/replica/database.cc
+++ b/replica/database.cc
@@ -1418,11 +1418,6 @@ future<> database::create_in_memory_keyspace(const lw_shared_ptr<keyspace_metada
     _keyspaces.emplace(ksm->name(), std::move(ks));
 }
 
-future<>
-database::create_keyspace(const lw_shared_ptr<keyspace_metadata>& ksm, locator::effective_replication_map_factory& erm_factory) {
-    return create_keyspace(ksm, erm_factory, system_keyspace::no);
-}
-
 future<>
 database::create_keyspace(const lw_shared_ptr<keyspace_metadata>& ksm, locator::effective_replication_map_factory& erm_factory, system_keyspace system) {
     if (_keyspaces.contains(ksm->name())) {
@@ -1434,6 +1429,14 @@ database::create_keyspace(const lw_shared_ptr<keyspace_metadata>& ksm, locator::
     co_await ks.init_storage();
 }
 
+future<> database::create_keyspace_on_all_shards(sharded<database>& sharded_db, sharded<service::storage_proxy>& proxy, const keyspace_metadata& ks_metadata) {
+    co_await sharded_db.invoke_on_all([&] (replica::database& db) -> future<> {
+        auto ksm = keyspace_metadata::new_keyspace(ks_metadata);
+        co_await db.create_keyspace(ksm, proxy.local().get_erm_factory(), system_keyspace::no);
+        co_await db.get_notifier().create_keyspace(ksm);
+    });
+}
+
 future<>
 database::drop_caches() const {
     std::unordered_map<table_id, lw_shared_ptr<column_family>> tables = get_column_families();
diff --git a/replica/database.hh b/replica/database.hh
index 68d115710a..0c2447cb92 100644
--- a/replica/database.hh
+++ b/replica/database.hh
@@ -1530,7 +1530,7 @@ public:
      *
      * @return ready future when the operation is complete
      */
-    future<> create_keyspace(const lw_shared_ptr<keyspace_metadata>&, locator::effective_replication_map_factory& erm_factory);
+    static future<> create_keyspace_on_all_shards(sharded<database>& sharded_db, sharded<service::storage_proxy>& proxy, const keyspace_metadata& ksm);
     /* below, find_keyspace throws no_such_<type> on fail */
     keyspace& find_keyspace(std::string_view name);
     const keyspace& find_keyspace(std::string_view name) const;