bahman/scylla
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Parameterize migration_manager::announce by type to allow executing different raft commands

Browse Source 
Since ALTER KS requires creating topology_change raft command, some
functions need to be extended to handle it. RAFT commands are recognized
by types, so some functions are just going to be parameterized by type,
i.e. made into templates.
These templates are instantiated already, so that only 1 instances of
each template exists across the whole code base, to avoid compiling it
in each translation unit.
This commit is contained in:
Piotr Smaron
2024-04-20 21:29:40 +02:00
committed by Pavel Emelyanov
parent 5afa3028a3
commit 73b59b244d
2 changed files with 28 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
service/migration_manager.cc
View File

@@ -954,18 +954,19 @@ future<> migration_manager::push_schema_mutation(const gms::inet_address& endpoi
return _messaging.send_definitions_update(id, std::vector<frozen_mutation>{}, std::move(cm));
}
template<typename mutation_type>
future<> migration_manager::announce_with_raft(std::vector<mutation> schema, group0_guard guard, std::string_view description) {
assert(this_shard_id() == 0);
auto schema_features = _feat.cluster_schema_features();
auto adjusted_schema = db::schema_tables::adjust_schema_for_schema_features(std::move(schema), schema_features);
auto group0_cmd = _group0_client.prepare_command(
schema_change{
.mutations{adjusted_schema.begin(), adjusted_schema.end()},
mutation_type {
.mutations{adjusted_schema.begin(), adjusted_schema.end()},
},
guard, std::move(description));
co_return co_await _group0_client.add_entry(std::move(group0_cmd), std::move(guard), &_as, raft_timeout{});
return _group0_client.add_entry(std::move(group0_cmd), std::move(guard), &_as);
}
future<> migration_manager::announce_without_raft(std::vector<mutation> schema, group0_guard guard) {
@@ -1027,6 +1028,7 @@ static void add_committed_by_group0_flag(std::vector<mutation>& schema, const gr
}
// Returns a future on the local application of the schema
template<typename mutation_type>
future<> migration_manager::announce(std::vector<mutation> schema, group0_guard guard, std::string_view description) {
if (_feat.group0_schema_versioning) {
schema.push_back(make_group0_schema_version_mutation(_storage_proxy.data_dictionary(), guard));
@@ -1034,11 +1036,20 @@ future<> migration_manager::announce(std::vector<mutation> schema, group0_guard
}
if (guard.with_raft()) {
return announce_with_raft(std::move(schema), std::move(guard), std::move(description));
return announce_with_raft<mutation_type>(std::move(schema), std::move(guard), std::move(description));
} else {
return announce_without_raft(std::move(schema), std::move(guard));
}
}
template
future<> migration_manager::announce_with_raft<schema_change>(std::vector<mutation> schema, group0_guard, std::string_view description);
template
future<> migration_manager::announce_with_raft<topology_change>(std::vector<mutation> schema, group0_guard, std::string_view description);
template
future<> migration_manager::announce<schema_change>(std::vector<mutation> schema, group0_guard, std::string_view description);
template
future<> migration_manager::announce<topology_change>(std::vector<mutation> schema, group0_guard, std::string_view description);
future<group0_guard> migration_manager::start_group0_operation() {
assert(this_shard_id() == 0);

13
service/migration_manager.hh
View File

@@ -136,6 +136,7 @@ public:
// Apply a group 0 change.
// The future resolves after the change is applied locally.
template<typename mutation_type = schema_change>
future<> announce(std::vector<mutation> schema, group0_guard, std::string_view description);
void passive_announce(table_schema_version version);
@@ -164,6 +165,7 @@ private:
future<> maybe_schedule_schema_pull(const table_schema_version& their_version, const gms::inet_address& endpoint);
template<typename mutation_type = schema_change>
future<> announce_with_raft(std::vector<mutation> schema, group0_guard, std::string_view description);
future<> announce_without_raft(std::vector<mutation> schema, group0_guard);
@@ -193,6 +195,17 @@ public:
void set_concurrent_ddl_retries(size_t);
};
extern template
future<> migration_manager::announce_with_raft<schema_change>(std::vector<mutation> schema, group0_guard, std::string_view description);
extern template
future<> migration_manager::announce_with_raft<topology_change>(std::vector<mutation> schema, group0_guard, std::string_view description);
extern template
future<> migration_manager::announce<schema_change>(std::vector<mutation> schema, group0_guard, std::string_view description);
extern template
future<> migration_manager::announce<topology_change>(std::vector<mutation> schema, group0_guard, std::string_view description);
future<column_mapping> get_column_mapping(db::system_keyspace& sys_ks, table_id, table_schema_version v);
std::vector<mutation> prepare_keyspace_update_announcement(replica::database& db, lw_shared_ptr<keyspace_metadata> ksm, api::timestamp_type ts);