From 3468cbd66b3a466ef42ecbb9bf2100263e9c9a0e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Patryk=20J=C4=99drzejczak?= Date: Tue, 25 Jul 2023 17:41:10 +0200 Subject: [PATCH 1/5] service: migration_manager: change the prepare_ methods to functions The migration_manager service is responsible for schema convergence in the cluster - pushing schema changes to other nodes and pulling schema when a version mismatch is observed. However, there is also a part of migration_manager that doesn't really belong there - creating mutations for schema updates. These are the functions with prepare_ prefix. They don't modify any state and don't exchange any messages. They only need to read the local database. We take these functions out of migration_manager and make them separate functions to reduce the dependency of other modules (especially query_processor and CQL statements) on migration_manager. Since all of these functions only need access to storage_proxy (or even only replica::database), doing such a refactor is not complicated. We just have to add one parameter, either storage_proxy or database and both of them are easily accessible in the places where these functions are called. --- alternator/executor.cc | 8 +- auth/common.cc | 2 +- auth/service.cc | 2 +- cql3/statements/alter_keyspace_statement.cc | 2 +- cql3/statements/alter_table_statement.cc | 2 +- cql3/statements/alter_type_statement.cc | 6 +- cql3/statements/alter_view_statement.cc | 2 +- cql3/statements/create_aggregate_statement.cc | 2 +- cql3/statements/create_function_statement.cc | 2 +- cql3/statements/create_index_statement.cc | 2 +- cql3/statements/create_keyspace_statement.cc | 2 +- cql3/statements/create_table_statement.cc | 2 +- cql3/statements/create_type_statement.cc | 2 +- cql3/statements/create_view_statement.cc | 2 +- cql3/statements/drop_aggregate_statement.cc | 2 +- cql3/statements/drop_function_statement.cc | 2 +- cql3/statements/drop_index_statement.cc | 2 +- cql3/statements/drop_keyspace_statement.cc | 2 +- cql3/statements/drop_table_statement.cc | 2 +- cql3/statements/drop_type_statement.cc | 2 +- cql3/statements/drop_view_statement.cc | 2 +- db/system_distributed_keyspace.cc | 10 +- db/tags/utils.cc | 3 +- redis/keyspace_utils.cc | 8 +- service/migration_manager.cc | 116 +++++++++--------- service/migration_manager.hh | 78 ++++++------ table_helper.cc | 4 +- test/boost/cql_query_test.cc | 2 +- test/boost/database_test.cc | 4 +- test/boost/group0_cmd_merge_test.cc | 2 +- test/boost/memtable_test.cc | 8 +- test/boost/row_cache_test.cc | 2 +- test/boost/schema_change_test.cc | 26 ++-- test/lib/cql_test_env.cc | 2 +- thrift/handler.cc | 18 +-- 35 files changed, 165 insertions(+), 170 deletions(-) diff --git a/alternator/executor.cc b/alternator/executor.cc index c3c7ad7887..9f5be7a59a 100644 --- a/alternator/executor.cc +++ b/alternator/executor.cc @@ -573,8 +573,8 @@ future executor::delete_table(client_state& clien throw api_error::resource_not_found(format("Requested resource not found: Table: {} not found", table_name)); } - auto m = co_await mm.prepare_column_family_drop_announcement(keyspace_name, table_name, group0_guard.write_timestamp(), service::migration_manager::drop_views::yes); - auto m2 = co_await mm.prepare_keyspace_drop_announcement(keyspace_name, group0_guard.write_timestamp()); + auto m = co_await service::prepare_column_family_drop_announcement(_proxy, keyspace_name, table_name, group0_guard.write_timestamp(), service::drop_views::yes); + auto m2 = co_await service::prepare_keyspace_drop_announcement(_proxy.local_db(), keyspace_name, group0_guard.write_timestamp()); std::move(m2.begin(), m2.end(), std::back_inserter(m)); @@ -1208,7 +1208,7 @@ future executor::update_table(client_state& clien auto schema = builder.build(); - auto m = co_await mm.prepare_column_family_update_announcement(schema, false, std::vector(), group0_guard.write_timestamp()); + auto m = co_await service::prepare_column_family_update_announcement(p.local(), schema, false, std::vector(), group0_guard.write_timestamp()); co_await mm.announce(std::move(m), std::move(group0_guard)); @@ -4480,7 +4480,7 @@ static future> create_keyspace(std::string_view keyspace_n auto opts = get_network_topology_options(sp, gossiper, rf); auto ksm = keyspace_metadata::new_keyspace(keyspace_name_str, "org.apache.cassandra.locator.NetworkTopologyStrategy", std::move(opts), true); - co_return mm.prepare_new_keyspace_announcement(ksm, ts); + co_return service::prepare_new_keyspace_announcement(sp.local_db(), ksm, ts); } future<> executor::start() { diff --git a/auth/common.cc b/auth/common.cc index 7158b99b1f..de1d15ebbf 100644 --- a/auth/common.cc +++ b/auth/common.cc @@ -71,7 +71,7 @@ static future<> create_metadata_table_if_missing_impl( auto group0_guard = co_await mm.start_group0_operation(); auto ts = group0_guard.write_timestamp(); try { - co_return co_await mm.announce(co_await mm.prepare_new_column_family_announcement(table, ts), std::move(group0_guard)); + co_return co_await mm.announce(co_await ::service::prepare_new_column_family_announcement(qp.proxy(), table, ts), std::move(group0_guard)); } catch (exceptions::already_exists_exception&) {} } } diff --git a/auth/service.cc b/auth/service.cc index 77e3d8c21d..4aaa6f4ba0 100644 --- a/auth/service.cc +++ b/auth/service.cc @@ -178,7 +178,7 @@ future<> service::create_keyspace_if_missing(::service::migration_manager& mm) c opts, true); - co_return co_await mm.announce(mm.prepare_new_keyspace_announcement(ksm, ts), std::move(group0_guard)); + co_return co_await mm.announce(::service::prepare_new_keyspace_announcement(db.real_database(), ksm, ts), std::move(group0_guard)); } } } diff --git a/cql3/statements/alter_keyspace_statement.cc b/cql3/statements/alter_keyspace_statement.cc index 40705e00e7..bbe74dfd38 100644 --- a/cql3/statements/alter_keyspace_statement.cc +++ b/cql3/statements/alter_keyspace_statement.cc @@ -80,7 +80,7 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce auto old_ksm = qp.db().find_keyspace(_name).metadata(); const auto& tm = *qp.proxy().get_token_metadata_ptr(); - auto m = mm.prepare_keyspace_update_announcement(_attrs->as_ks_metadata_update(old_ksm, tm), ts); + auto m = service::prepare_keyspace_update_announcement(qp.db().real_database(), _attrs->as_ks_metadata_update(old_ksm, tm), ts); using namespace cql_transport; auto ret = ::make_shared( diff --git a/cql3/statements/alter_table_statement.cc b/cql3/statements/alter_table_statement.cc index f6dbef635b..c39035a566 100644 --- a/cql3/statements/alter_table_statement.cc +++ b/cql3/statements/alter_table_statement.cc @@ -384,7 +384,7 @@ future, std::vector alter_table_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { data_dictionary::database db = qp.db(); auto [cfm, view_updates] = prepare_schema_update(db); - auto m = co_await mm.prepare_column_family_update_announcement(cfm.build(), false, std::move(view_updates), ts); + auto m = co_await service::prepare_column_family_update_announcement(qp.proxy(), cfm.build(), false, std::move(view_updates), ts); using namespace cql_transport; auto ret = ::make_shared( diff --git a/cql3/statements/alter_type_statement.cc b/cql3/statements/alter_type_statement.cc index 708fee1c1f..7464a4fe67 100644 --- a/cql3/statements/alter_type_statement.cc +++ b/cql3/statements/alter_type_statement.cc @@ -69,7 +69,7 @@ future> alter_type_statement::prepare_announcement_mutatio auto&& updated = make_updated_type(db, to_update->second); // Now, we need to announce the type update to basically change it for new tables using this type, // but we also need to find all existing user types and CF using it and change them. - auto res = co_await mm.prepare_update_type_announcement(updated, ts); + auto res = co_await service::prepare_update_type_announcement(mm.get_storage_proxy(), updated, ts); std::move(res.begin(), res.end(), std::back_inserter(m)); for (auto&& schema : ks.metadata()->cf_meta_data() | boost::adaptors::map_values) { @@ -85,10 +85,10 @@ future> alter_type_statement::prepare_announcement_mutatio } if (modified) { if (schema->is_view()) { - auto res = co_await mm.prepare_view_update_announcement(view_ptr(cfm.build()), ts); + auto res = co_await service::prepare_view_update_announcement(mm.get_storage_proxy(), view_ptr(cfm.build()), ts); std::move(res.begin(), res.end(), std::back_inserter(m)); } else { - auto res = co_await mm.prepare_column_family_update_announcement(cfm.build(), false, {}, ts); + auto res = co_await service::prepare_column_family_update_announcement(mm.get_storage_proxy(), cfm.build(), false, {}, ts); std::move(res.begin(), res.end(), std::back_inserter(m)); } } diff --git a/cql3/statements/alter_view_statement.cc b/cql3/statements/alter_view_statement.cc index 6ab2984ded..55518f3919 100644 --- a/cql3/statements/alter_view_statement.cc +++ b/cql3/statements/alter_view_statement.cc @@ -77,7 +77,7 @@ view_ptr alter_view_statement::prepare_view(data_dictionary::database db) const } future, std::vector, cql3::cql_warnings_vec>> alter_view_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { - auto m = co_await mm.prepare_view_update_announcement(prepare_view(qp.db()), ts); + auto m = co_await service::prepare_view_update_announcement(qp.proxy(), prepare_view(qp.db()), ts); using namespace cql_transport; auto ret = ::make_shared( diff --git a/cql3/statements/create_aggregate_statement.cc b/cql3/statements/create_aggregate_statement.cc index 94cdb794a6..3fcca4a8ca 100644 --- a/cql3/statements/create_aggregate_statement.cc +++ b/cql3/statements/create_aggregate_statement.cc @@ -90,7 +90,7 @@ create_aggregate_statement::prepare_schema_mutations(query_processor& qp, servic auto aggregate = dynamic_pointer_cast(co_await validate_while_executing(qp)); if (aggregate) { - m = co_await mm.prepare_new_aggregate_announcement(aggregate, ts); + m = co_await service::prepare_new_aggregate_announcement(qp.proxy(), aggregate, ts); ret = create_schema_change(*aggregate, true); } diff --git a/cql3/statements/create_function_statement.cc b/cql3/statements/create_function_statement.cc index 302ba96c12..a4606e6543 100644 --- a/cql3/statements/create_function_statement.cc +++ b/cql3/statements/create_function_statement.cc @@ -73,7 +73,7 @@ create_function_statement::prepare_schema_mutations(query_processor& qp, service auto func = dynamic_pointer_cast(co_await validate_while_executing(qp)); if (func) { - m = co_await mm.prepare_new_function_announcement(func, ts); + m = co_await service::prepare_new_function_announcement(qp.proxy(), func, ts); ret = create_schema_change(*func, true); } diff --git a/cql3/statements/create_index_statement.cc b/cql3/statements/create_index_statement.cc index 2ee64c906d..5a6e7fe009 100644 --- a/cql3/statements/create_index_statement.cc +++ b/cql3/statements/create_index_statement.cc @@ -384,7 +384,7 @@ create_index_statement::prepare_schema_mutations(query_processor& qp, service::m std::vector m; if (schema) { - m = co_await mm.prepare_column_family_update_announcement(std::move(schema), false, {}, ts); + m = co_await service::prepare_column_family_update_announcement(qp.proxy(), std::move(schema), false, {}, ts); ret = ::make_shared( event::schema_change::change_type::UPDATED, diff --git a/cql3/statements/create_keyspace_statement.cc b/cql3/statements/create_keyspace_statement.cc index 524fc7a493..5cfa5b2c52 100644 --- a/cql3/statements/create_keyspace_statement.cc +++ b/cql3/statements/create_keyspace_statement.cc @@ -100,7 +100,7 @@ future, std::vector std::vector m; try { - m = mm.prepare_new_keyspace_announcement(_attrs->as_ks_metadata(_name, tm), ts); + m = service::prepare_new_keyspace_announcement(qp.db().real_database(), _attrs->as_ks_metadata(_name, tm), ts); ret = ::make_shared( event::schema_change::change_type::CREATED, diff --git a/cql3/statements/create_table_statement.cc b/cql3/statements/create_table_statement.cc index e1f9e5b654..ab8e48efad 100644 --- a/cql3/statements/create_table_statement.cc +++ b/cql3/statements/create_table_statement.cc @@ -76,7 +76,7 @@ create_table_statement::prepare_schema_mutations(query_processor& qp, service::m std::vector m; try { - m = co_await mm.prepare_new_column_family_announcement(get_cf_meta_data(qp.db()), ts); + m = co_await service::prepare_new_column_family_announcement(qp.proxy(), get_cf_meta_data(qp.db()), ts); using namespace cql_transport; ret = ::make_shared( diff --git a/cql3/statements/create_type_statement.cc b/cql3/statements/create_type_statement.cc index a526465b7c..d756366cca 100644 --- a/cql3/statements/create_type_statement.cc +++ b/cql3/statements/create_type_statement.cc @@ -124,7 +124,7 @@ future, std::vector try { auto t = make_type(qp); if (t) { - m = co_await mm.prepare_new_type_announcement(*t, ts); + m = co_await service::prepare_new_type_announcement(qp.proxy(), *t, ts); using namespace cql_transport; ret = ::make_shared( diff --git a/cql3/statements/create_view_statement.cc b/cql3/statements/create_view_statement.cc index d181f1dd44..a0e7358c2d 100644 --- a/cql3/statements/create_view_statement.cc +++ b/cql3/statements/create_view_statement.cc @@ -367,7 +367,7 @@ create_view_statement::prepare_schema_mutations(query_processor& qp, service::mi std::vector m; auto [definition, warnings] = prepare_view(qp.db()); try { - m = co_await mm.prepare_new_view_announcement(std::move(definition), ts); + m = co_await service::prepare_new_view_announcement(qp.proxy(), std::move(definition), ts); using namespace cql_transport; ret = ::make_shared( event::schema_change::change_type::CREATED, diff --git a/cql3/statements/drop_aggregate_statement.cc b/cql3/statements/drop_aggregate_statement.cc index 8573d11657..ca44fef5f2 100644 --- a/cql3/statements/drop_aggregate_statement.cc +++ b/cql3/statements/drop_aggregate_statement.cc @@ -34,7 +34,7 @@ drop_aggregate_statement::prepare_schema_mutations(query_processor& qp, service: if (!user_aggr) { throw exceptions::invalid_request_exception(format("'{}' is not a user defined aggregate", func)); } - m = co_await mm.prepare_aggregate_drop_announcement(user_aggr, ts); + m = co_await service::prepare_aggregate_drop_announcement(qp.proxy(), user_aggr, ts); ret = create_schema_change(*func, false); } diff --git a/cql3/statements/drop_function_statement.cc b/cql3/statements/drop_function_statement.cc index 798975c06f..04cbc3e024 100644 --- a/cql3/statements/drop_function_statement.cc +++ b/cql3/statements/drop_function_statement.cc @@ -38,7 +38,7 @@ drop_function_statement::prepare_schema_mutations(query_processor& qp, service:: if (auto aggregate = functions::functions::used_by_user_aggregate(user_func)) { throw exceptions::invalid_request_exception(format("Cannot delete function {}, as it is used by user-defined aggregate {}", func, *aggregate)); } - m = co_await mm.prepare_function_drop_announcement(user_func, ts); + m = co_await service::prepare_function_drop_announcement(qp.proxy(), user_func, ts); ret = create_schema_change(*func, false); } diff --git a/cql3/statements/drop_index_statement.cc b/cql3/statements/drop_index_statement.cc index f496e7a31f..d3089eaa14 100644 --- a/cql3/statements/drop_index_statement.cc +++ b/cql3/statements/drop_index_statement.cc @@ -79,7 +79,7 @@ drop_index_statement::prepare_schema_mutations(query_processor& qp, service::mig auto cfm = make_drop_idex_schema(qp); if (cfm) { - m = co_await mm.prepare_column_family_update_announcement(cfm, false, {}, ts); + m = co_await service::prepare_column_family_update_announcement(qp.proxy(), cfm, false, {}, ts); using namespace cql_transport; ret = ::make_shared(event::schema_change::change_type::UPDATED, diff --git a/cql3/statements/drop_keyspace_statement.cc b/cql3/statements/drop_keyspace_statement.cc index de9bdacbda..ab8d10e3fb 100644 --- a/cql3/statements/drop_keyspace_statement.cc +++ b/cql3/statements/drop_keyspace_statement.cc @@ -52,7 +52,7 @@ drop_keyspace_statement::prepare_schema_mutations(query_processor& qp, service:: ::shared_ptr ret; try { - m = co_await mm.prepare_keyspace_drop_announcement(_keyspace, ts); + m = co_await service::prepare_keyspace_drop_announcement(qp.db().real_database(), _keyspace, ts); using namespace cql_transport; ret = ::make_shared( diff --git a/cql3/statements/drop_table_statement.cc b/cql3/statements/drop_table_statement.cc index 7e2b0d6347..6814e83d6c 100644 --- a/cql3/statements/drop_table_statement.cc +++ b/cql3/statements/drop_table_statement.cc @@ -45,7 +45,7 @@ drop_table_statement::prepare_schema_mutations(query_processor& qp, service::mig } try { - m = co_await mm.prepare_column_family_drop_announcement(keyspace(), column_family(), ts); + m = co_await service::prepare_column_family_drop_announcement(qp.proxy(), keyspace(), column_family(), ts); using namespace cql_transport; ret = ::make_shared( diff --git a/cql3/statements/drop_type_statement.cc b/cql3/statements/drop_type_statement.cc index ad4b611f77..2b2ea23cea 100644 --- a/cql3/statements/drop_type_statement.cc +++ b/cql3/statements/drop_type_statement.cc @@ -137,7 +137,7 @@ drop_type_statement::prepare_schema_mutations(query_processor& qp, service::migr // Can happen with if_exists if (to_drop != all_types.end()) { - m = co_await mm.prepare_type_drop_announcement(to_drop->second, ts); + m = co_await service::prepare_type_drop_announcement(qp.proxy(), to_drop->second, ts); using namespace cql_transport; ret = ::make_shared( diff --git a/cql3/statements/drop_view_statement.cc b/cql3/statements/drop_view_statement.cc index 18da6e7bf1..92170207a7 100644 --- a/cql3/statements/drop_view_statement.cc +++ b/cql3/statements/drop_view_statement.cc @@ -47,7 +47,7 @@ drop_view_statement::prepare_schema_mutations(query_processor& qp, service::migr std::vector m; try { - m = co_await mm.prepare_view_drop_announcement(keyspace(), column_family(), ts); + m = co_await service::prepare_view_drop_announcement(qp.proxy(), keyspace(), column_family(), ts); using namespace cql_transport; ret = ::make_shared( diff --git a/db/system_distributed_keyspace.cc b/db/system_distributed_keyspace.cc index ef3afa0aa2..8206d7db1e 100644 --- a/db/system_distributed_keyspace.cc +++ b/db/system_distributed_keyspace.cc @@ -253,8 +253,8 @@ static future<> add_new_columns_if_missing(replica::database& db, ::service::mig schema_ptr table = b.build(); try { auto ts = group0_guard.write_timestamp(); - co_return co_await mm.announce(co_await mm.prepare_column_family_update_announcement(table, false, std::vector(), ts), - std::move(group0_guard), "Add new columns to system_distributed.service_levels"); + co_return co_await mm.announce(co_await service::prepare_column_family_update_announcement(mm.get_storage_proxy(), table, false, + std::vector(), ts), std::move(group0_guard), "Add new columns to system_distributed.service_levels"); } catch (...) {} } } catch (...) { @@ -282,7 +282,7 @@ future<> system_distributed_keyspace::start() { "org.apache.cassandra.locator.SimpleStrategy", {{"replication_factor", "3"}}, true /* durable_writes */); - co_await _mm.announce(_mm.prepare_new_keyspace_announcement(ksm, ts), std::move(group0_guard), + co_await _mm.announce(service::prepare_new_keyspace_announcement(_sp.local_db(), ksm, ts), std::move(group0_guard), "Create system_distributed keyspace"); } catch (exceptions::already_exists_exception&) {} } else { @@ -299,7 +299,7 @@ future<> system_distributed_keyspace::start() { "org.apache.cassandra.locator.EverywhereStrategy", {}, true /* durable_writes */); - co_await _mm.announce(_mm.prepare_new_keyspace_announcement(ksm, ts), std::move(group0_guard), + co_await _mm.announce(service::prepare_new_keyspace_announcement(_sp.local_db(), ksm, ts), std::move(group0_guard), "Create system_distributed_everywhere keyspace"); } catch (exceptions::already_exists_exception&) {} } else { @@ -318,7 +318,7 @@ future<> system_distributed_keyspace::start() { auto m = co_await map_reduce(tables, /* Mapper */ [this, ts] (auto&& table) -> future> { try { - co_return co_await _mm.prepare_new_column_family_announcement(std::move(table), ts); + co_return co_await service::prepare_new_column_family_announcement(_sp, std::move(table), ts); } catch (exceptions::already_exists_exception&) { co_return std::vector(); } diff --git a/db/tags/utils.cc b/db/tags/utils.cc index 16d443cc50..3874f634d4 100644 --- a/db/tags/utils.cc +++ b/db/tags/utils.cc @@ -62,7 +62,8 @@ future<> modify_tags(service::migration_manager& mm, sstring ks, sstring cf, schema_builder builder(s); builder.add_extension(tags_extension::NAME, ::make_shared(tags)); - auto m = co_await mm.prepare_column_family_update_announcement(builder.build(), false, std::vector(), group0_guard.write_timestamp()); + auto m = co_await service::prepare_column_family_update_announcement(mm.get_storage_proxy(), + builder.build(), false, std::vector(), group0_guard.write_timestamp()); co_await mm.announce(std::move(m), std::move(group0_guard)); }); diff --git a/redis/keyspace_utils.cc b/redis/keyspace_utils.cc index 6e399f8d32..34384dc061 100644 --- a/redis/keyspace_utils.cc +++ b/redis/keyspace_utils.cc @@ -198,7 +198,7 @@ future<> create_keyspace_if_not_exists_impl(seastar::sharded create_keyspace_if_not_exists_impl(seastar::sharded table_mutations; auto table_gen = std::bind_front( - [] (data_dictionary::database db, service::migration_manager& mml, std::vector& table_mutations, + [] (data_dictionary::database db, service::storage_proxy& sp, std::vector& table_mutations, api::timestamp_type ts, sstring ks_name, sstring cf_name, schema_ptr schema) -> future<> { if (db.has_schema(ks_name, cf_name)) { co_return; } logger.info("Create keyspace: {}, table: {} for redis.", ks_name, cf_name); - auto muts = co_await mml.prepare_new_column_family_announcement(schema, ts); + auto muts = co_await service::prepare_new_column_family_announcement(sp, schema, ts); std::move(muts.begin(), muts.end(), std::back_inserter(table_mutations)); - }, db, std::ref(mml), std::ref(table_mutations), group0_guard.write_timestamp()); + }, db, std::ref(proxy.local()), std::ref(table_mutations), group0_guard.write_timestamp()); co_await coroutine::parallel_for_each(ks_names, [table_gen = std::move(table_gen)] (const sstring& ks_name) mutable { return parallel_for_each(tables, [ks_name, table_gen = std::move(table_gen)] (table t) { diff --git a/service/migration_manager.cc b/service/migration_manager.cc index 03ae07ec11..f2175375e7 100644 --- a/service/migration_manager.cc +++ b/service/migration_manager.cc @@ -644,38 +644,32 @@ public void notifyDropAggregate(UDAggregate udf) } #endif -std::vector migration_manager::prepare_keyspace_update_announcement(lw_shared_ptr ksm, api::timestamp_type ts) { - auto& proxy = _storage_proxy; - auto& db = proxy.get_db().local(); - +std::vector prepare_keyspace_update_announcement(replica::database& db, lw_shared_ptr ksm, api::timestamp_type ts) { db.validate_keyspace_update(*ksm); mlogger.info("Update Keyspace: {}", ksm); return db::schema_tables::make_create_keyspace_mutations(db.features().cluster_schema_features(), ksm, ts); } -std::vector migration_manager::prepare_new_keyspace_announcement(lw_shared_ptr ksm, api::timestamp_type timestamp) { - auto& proxy = _storage_proxy; - auto& db = proxy.get_db().local(); - +std::vector prepare_new_keyspace_announcement(replica::database& db, lw_shared_ptr ksm, api::timestamp_type timestamp) { db.validate_new_keyspace(*ksm); mlogger.info("Create new Keyspace: {}", ksm); return db::schema_tables::make_create_keyspace_mutations(db.features().cluster_schema_features(), ksm, timestamp); } -future> migration_manager::include_keyspace( - const keyspace_metadata& keyspace, std::vector mutations) { +static future> include_keyspace( + storage_proxy& sp, const keyspace_metadata& keyspace, std::vector mutations) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - mutation m = co_await db::schema_tables::read_keyspace_mutation(_storage_proxy.container(), keyspace.name()); + mutation m = co_await db::schema_tables::read_keyspace_mutation(sp.container(), keyspace.name()); mutations.push_back(std::move(m)); co_return std::move(mutations); } -future> migration_manager::prepare_new_column_family_announcement(schema_ptr cfm, api::timestamp_type timestamp) { +future> prepare_new_column_family_announcement(storage_proxy& sp, schema_ptr cfm, api::timestamp_type timestamp) { #if 0 cfm.validate(); #endif try { - auto& db = _storage_proxy.get_db().local(); + auto& db = sp.get_db().local(); auto&& keyspace = db.find_keyspace(cfm->ks_name()); if (db.has_schema(cfm->ks_name(), cfm->cf_name())) { throw exceptions::already_exists_exception(cfm->ks_name(), cfm->cf_name()); @@ -687,25 +681,26 @@ future> migration_manager::prepare_new_column_family_annou mlogger.info("Create new ColumnFamily: {}", cfm); auto ksm = keyspace.metadata(); - return seastar::async([this, cfm, timestamp, ksm] { + return seastar::async([&db, cfm, timestamp, ksm] { auto mutations = db::schema_tables::make_create_table_mutations(cfm, timestamp); - get_notifier().before_create_column_family(*cfm, mutations, timestamp); + db.get_notifier().before_create_column_family(*cfm, mutations, timestamp); return mutations; - }).then([this, ksm](std::vector mutations) { - return include_keyspace(*ksm, std::move(mutations)); + }).then([&sp, ksm](std::vector mutations) { + return include_keyspace(sp, *ksm, std::move(mutations)); }); } catch (const replica::no_such_keyspace& e) { throw exceptions::configuration_exception(format("Cannot add table '{}' to non existing keyspace '{}'.", cfm->cf_name(), cfm->ks_name())); } } -future> migration_manager::prepare_column_family_update_announcement(schema_ptr cfm, bool from_thrift, std::vector view_updates, api::timestamp_type ts) { +future> prepare_column_family_update_announcement(storage_proxy& sp, + schema_ptr cfm, bool from_thrift, std::vector view_updates, api::timestamp_type ts) { warn(unimplemented::cause::VALIDATION); #if 0 cfm.validate(); #endif try { - auto& db = _storage_proxy.get_db().local(); + auto& db = sp.local_db(); auto&& old_schema = db.find_column_family(cfm->ks_name(), cfm->cf_name()).schema(); // FIXME: Should we lookup by id? #if 0 oldCfm.validateCompatility(cfm); @@ -722,9 +717,9 @@ future> migration_manager::prepare_column_family_update_an co_await coroutine::maybe_yield(); } co_await seastar::async([&] { - get_notifier().before_update_column_family(*cfm, *old_schema, mutations, ts); + db.get_notifier().before_update_column_family(*cfm, *old_schema, mutations, ts); }); - co_return co_await include_keyspace(*keyspace, std::move(mutations)); + co_return co_await include_keyspace(sp, *keyspace, std::move(mutations)); } catch (const replica::no_such_column_family& e) { auto&& ex = std::make_exception_ptr(exceptions::configuration_exception(format("Cannot update non existing table '{}' in keyspace '{}'.", cfm->cf_name(), cfm->ks_name()))); @@ -732,69 +727,68 @@ future> migration_manager::prepare_column_family_update_an } } -future> migration_manager::do_prepare_new_type_announcement(user_type new_type, api::timestamp_type ts) { - auto& db = _storage_proxy.get_db().local(); +static future> do_prepare_new_type_announcement(storage_proxy& sp, user_type new_type, api::timestamp_type ts) { + auto& db = sp.local_db(); auto&& keyspace = db.find_keyspace(new_type->_keyspace); auto mutations = db::schema_tables::make_create_type_mutations(keyspace.metadata(), new_type, ts); - return include_keyspace(*keyspace.metadata(), std::move(mutations)); + return include_keyspace(sp, *keyspace.metadata(), std::move(mutations)); } -future> migration_manager::prepare_new_type_announcement(user_type new_type, api::timestamp_type ts) { +future> prepare_new_type_announcement(storage_proxy& sp, user_type new_type, api::timestamp_type ts) { mlogger.info("Prepare Create new User Type: {}", new_type->get_name_as_string()); - return do_prepare_new_type_announcement(std::move(new_type), ts); + return do_prepare_new_type_announcement(sp, std::move(new_type), ts); } -future> migration_manager::prepare_update_type_announcement(user_type updated_type, api::timestamp_type ts) { +future> prepare_update_type_announcement(storage_proxy& sp, user_type updated_type, api::timestamp_type ts) { mlogger.info("Prepare Update User Type: {}", updated_type->get_name_as_string()); - return do_prepare_new_type_announcement(updated_type, ts); + return do_prepare_new_type_announcement(sp, updated_type, ts); } -future> migration_manager::prepare_new_function_announcement(shared_ptr func, api::timestamp_type ts) { - auto& db = _storage_proxy.get_db().local(); +future> prepare_new_function_announcement(storage_proxy& sp, shared_ptr func, api::timestamp_type ts) { + auto& db = sp.local_db(); auto&& keyspace = db.find_keyspace(func->name().keyspace); auto mutations = db::schema_tables::make_create_function_mutations(func, ts); - return include_keyspace(*keyspace.metadata(), std::move(mutations)); + return include_keyspace(sp, *keyspace.metadata(), std::move(mutations)); } -future> migration_manager::prepare_function_drop_announcement(shared_ptr func, api::timestamp_type ts) { - auto& db = _storage_proxy.get_db().local(); +future> prepare_function_drop_announcement(storage_proxy& sp, shared_ptr func, api::timestamp_type ts) { + auto& db = sp.local_db(); auto&& keyspace = db.find_keyspace(func->name().keyspace); auto mutations = db::schema_tables::make_drop_function_mutations(func, ts); - return include_keyspace(*keyspace.metadata(), std::move(mutations)); + return include_keyspace(sp, *keyspace.metadata(), std::move(mutations)); } -future> migration_manager::prepare_new_aggregate_announcement(shared_ptr aggregate, api::timestamp_type ts) { - auto& db = _storage_proxy.get_db().local(); +future> prepare_new_aggregate_announcement(storage_proxy& sp, shared_ptr aggregate, api::timestamp_type ts) { + auto& db = sp.local_db(); auto&& keyspace = db.find_keyspace(aggregate->name().keyspace); auto mutations = db::schema_tables::make_create_aggregate_mutations(db.features().cluster_schema_features(), aggregate, ts); - return include_keyspace(*keyspace.metadata(), std::move(mutations)); + return include_keyspace(sp, *keyspace.metadata(), std::move(mutations)); } -future> migration_manager::prepare_aggregate_drop_announcement(shared_ptr aggregate, api::timestamp_type ts) { - auto& db = _storage_proxy.get_db().local(); +future> prepare_aggregate_drop_announcement(storage_proxy& sp, shared_ptr aggregate, api::timestamp_type ts) { + auto& db = sp.local_db(); auto&& keyspace = db.find_keyspace(aggregate->name().keyspace); auto mutations = db::schema_tables::make_drop_aggregate_mutations(db.features().cluster_schema_features(), aggregate, ts); - return include_keyspace(*keyspace.metadata(), std::move(mutations)); + return include_keyspace(sp, *keyspace.metadata(), std::move(mutations)); } -future> migration_manager::prepare_keyspace_drop_announcement(const sstring& ks_name, api::timestamp_type ts) { - auto& db = _storage_proxy.get_db().local(); +future> prepare_keyspace_drop_announcement(replica::database& db, const sstring& ks_name, api::timestamp_type ts) { if (!db.has_keyspace(ks_name)) { throw exceptions::configuration_exception(format("Cannot drop non existing keyspace '{}'.", ks_name)); } auto& keyspace = db.find_keyspace(ks_name); mlogger.info("Drop Keyspace '{}'", ks_name); - return seastar::async([this, &db, &keyspace, ts, ks_name] { + return seastar::async([&db, &keyspace, ts, ks_name] { auto mutations = db::schema_tables::make_drop_keyspace_mutations(db.features().cluster_schema_features(), keyspace.metadata(), ts); - get_notifier().before_drop_keyspace(ks_name, mutations, ts); + db.get_notifier().before_drop_keyspace(ks_name, mutations, ts); return mutations; }); } -future> migration_manager::prepare_column_family_drop_announcement(const sstring& ks_name, - const sstring& cf_name, api::timestamp_type ts, drop_views drop_views) { +future> prepare_column_family_drop_announcement(storage_proxy& sp, + const sstring& ks_name, const sstring& cf_name, api::timestamp_type ts, drop_views drop_views) { try { - auto& db = _storage_proxy.get_db().local(); + auto& db = sp.local_db(); auto& old_cfm = db.find_column_family(ks_name, cf_name); auto& schema = old_cfm.schema(); if (schema->is_view()) { @@ -832,29 +826,29 @@ future> migration_manager::prepare_column_family_drop_anno // notifiers must run in seastar thread co_await seastar::async([&] { - get_notifier().before_drop_column_family(*schema, mutations, ts); + db.get_notifier().before_drop_column_family(*schema, mutations, ts); }); - co_return co_await include_keyspace(*keyspace, std::move(mutations)); + co_return co_await include_keyspace(sp, *keyspace, std::move(mutations)); } catch (const replica::no_such_column_family& e) { auto&& ex = std::make_exception_ptr(exceptions::configuration_exception(format("Cannot drop non existing table '{}' in keyspace '{}'.", cf_name, ks_name))); co_return coroutine::exception(std::move(ex)); } } -future> migration_manager::prepare_type_drop_announcement(user_type dropped_type, api::timestamp_type ts) { - auto& db = _storage_proxy.get_db().local(); +future> prepare_type_drop_announcement(storage_proxy& sp, user_type dropped_type, api::timestamp_type ts) { + auto& db = sp.local_db(); auto&& keyspace = db.find_keyspace(dropped_type->_keyspace); mlogger.info("Drop User Type: {}", dropped_type->get_name_as_string()); auto mutations = db::schema_tables::make_drop_type_mutations(keyspace.metadata(), dropped_type, ts); - return include_keyspace(*keyspace.metadata(), std::move(mutations)); + return include_keyspace(sp, *keyspace.metadata(), std::move(mutations)); } -future> migration_manager::prepare_new_view_announcement(view_ptr view, api::timestamp_type ts) { +future> prepare_new_view_announcement(storage_proxy& sp, view_ptr view, api::timestamp_type ts) { #if 0 view.metadata.validate(); #endif - auto& db = _storage_proxy.get_db().local(); + auto& db = sp.local_db(); try { auto&& keyspace = db.find_keyspace(view->ks_name()).metadata(); if (keyspace->cf_meta_data().contains(view->cf_name())) { @@ -862,18 +856,18 @@ future> migration_manager::prepare_new_view_announcement(v } mlogger.info("Create new view: {}", view); auto mutations = db::schema_tables::make_create_view_mutations(keyspace, std::move(view), ts); - co_return co_await include_keyspace(*keyspace, std::move(mutations)); + co_return co_await include_keyspace(sp, *keyspace, std::move(mutations)); } catch (const replica::no_such_keyspace& e) { auto&& ex = std::make_exception_ptr(exceptions::configuration_exception(format("Cannot add view '{}' to non existing keyspace '{}'.", view->cf_name(), view->ks_name()))); co_return coroutine::exception(std::move(ex)); } } -future> migration_manager::prepare_view_update_announcement(view_ptr view, api::timestamp_type ts) { +future> prepare_view_update_announcement(storage_proxy& sp, view_ptr view, api::timestamp_type ts) { #if 0 view.metadata.validate(); #endif - auto db = _storage_proxy.data_dictionary(); + auto db = sp.data_dictionary(); try { auto&& keyspace = db.find_keyspace(view->ks_name()).metadata(); auto& old_view = keyspace->cf_meta_data().at(view->cf_name()); @@ -885,7 +879,7 @@ future> migration_manager::prepare_view_update_announcemen #endif mlogger.info("Update view '{}.{}' From {} To {}", view->ks_name(), view->cf_name(), *old_view, *view); auto mutations = db::schema_tables::make_update_view_mutations(keyspace, view_ptr(old_view), std::move(view), ts, true); - co_return co_await include_keyspace(*keyspace, std::move(mutations)); + co_return co_await include_keyspace(sp, *keyspace, std::move(mutations)); } catch (const std::out_of_range& e) { auto&& ex = std::make_exception_ptr(exceptions::configuration_exception(format("Cannot update non existing materialized view '{}' in keyspace '{}'.", view->cf_name(), view->ks_name()))); @@ -893,8 +887,8 @@ future> migration_manager::prepare_view_update_announcemen } } -future> migration_manager::prepare_view_drop_announcement(const sstring& ks_name, const sstring& cf_name, api::timestamp_type ts) { - auto& db = _storage_proxy.get_db().local(); +future> prepare_view_drop_announcement(storage_proxy& sp, const sstring& ks_name, const sstring& cf_name, api::timestamp_type ts) { + auto& db = sp.local_db(); try { auto& view = db.find_column_family(ks_name, cf_name).schema(); if (!view->is_view()) { @@ -906,7 +900,7 @@ future> migration_manager::prepare_view_drop_announcement( auto keyspace = db.find_keyspace(ks_name).metadata(); mlogger.info("Drop view '{}.{}'", view->ks_name(), view->cf_name()); auto mutations = db::schema_tables::make_drop_view_mutations(keyspace, view_ptr(std::move(view)), ts); - return include_keyspace(*keyspace, std::move(mutations)); + return include_keyspace(sp, *keyspace, std::move(mutations)); } catch (const replica::no_such_column_family& e) { throw exceptions::configuration_exception(format("Cannot drop non existing materialized view '{}' in keyspace '{}'.", cf_name, ks_name)); diff --git a/service/migration_manager.hh b/service/migration_manager.hh index d0ef183c49..ca2b3548b1 100644 --- a/service/migration_manager.hh +++ b/service/migration_manager.hh @@ -132,43 +132,6 @@ public: bool should_pull_schema_from(const gms::inet_address& endpoint); bool has_compatible_schema_tables_version(const gms::inet_address& endpoint); - std::vector prepare_keyspace_update_announcement(lw_shared_ptr ksm, api::timestamp_type); - - std::vector prepare_new_keyspace_announcement(lw_shared_ptr ksm, api::timestamp_type); - - - // The timestamp parameter can be used to ensure that all nodes update their internal tables' schemas - // with identical timestamps, which can prevent an undeeded schema exchange - future> prepare_column_family_update_announcement(schema_ptr cfm, bool from_thrift, std::vector view_updates, api::timestamp_type ts); - - future> prepare_new_column_family_announcement(schema_ptr cfm, api::timestamp_type timestamp); - - future> prepare_new_type_announcement(user_type new_type, api::timestamp_type); - - future> prepare_new_function_announcement(shared_ptr func, api::timestamp_type); - - future> prepare_new_aggregate_announcement(shared_ptr aggregate, api::timestamp_type); - - future> prepare_function_drop_announcement(shared_ptr func, api::timestamp_type); - - future> prepare_aggregate_drop_announcement(shared_ptr aggregate, api::timestamp_type); - - future> prepare_update_type_announcement(user_type updated_type, api::timestamp_type); - - future> prepare_keyspace_drop_announcement(const sstring& ks_name, api::timestamp_type); - - class drop_views_tag; - using drop_views = bool_class; - future> prepare_column_family_drop_announcement(const sstring& ks_name, const sstring& cf_name, api::timestamp_type, drop_views drop_views = drop_views::no); - - future> prepare_type_drop_announcement(user_type dropped_type, api::timestamp_type); - - future> prepare_new_view_announcement(view_ptr view, api::timestamp_type); - - future> prepare_view_update_announcement(view_ptr view, api::timestamp_type); - - future> prepare_view_drop_announcement(const sstring& ks_name, const sstring& cf_name, api::timestamp_type); - // The function needs to be called if the user wants to read most up-to-date group 0 state (including schema state) // (the function ensures that all previously finished group0 operations are visible on this node) or to write it. // @@ -202,9 +165,6 @@ public: private: future<> uninit_messaging_service(); - future> include_keyspace(const keyspace_metadata& keyspace, std::vector mutations); - future> do_prepare_new_type_announcement(user_type new_type, api::timestamp_type); - future<> push_schema_mutation(const gms::inet_address& endpoint, const std::vector& schema); future<> passive_announce(); @@ -245,4 +205,42 @@ public: future get_column_mapping(table_id, table_schema_version v); +std::vector prepare_keyspace_update_announcement(replica::database& db, lw_shared_ptr ksm, api::timestamp_type ts); + +std::vector prepare_new_keyspace_announcement(replica::database& db, lw_shared_ptr ksm, api::timestamp_type timestamp); + +// The timestamp parameter can be used to ensure that all nodes update their internal tables' schemas +// with identical timestamps, which can prevent an undeeded schema exchange +future> prepare_column_family_update_announcement(storage_proxy& sp, + schema_ptr cfm, bool from_thrift, std::vector view_updates, api::timestamp_type ts); + +future> prepare_new_column_family_announcement(storage_proxy& sp, schema_ptr cfm, api::timestamp_type timestamp); + +future> prepare_new_type_announcement(storage_proxy& sp, user_type new_type, api::timestamp_type ts); + +future> prepare_new_function_announcement(storage_proxy& sp, shared_ptr func, api::timestamp_type ts); + +future> prepare_new_aggregate_announcement(storage_proxy& sp, shared_ptr aggregate, api::timestamp_type ts); + +future> prepare_function_drop_announcement(storage_proxy& sp, shared_ptr func, api::timestamp_type ts); + +future> prepare_aggregate_drop_announcement(storage_proxy& sp, shared_ptr aggregate, api::timestamp_type ts); + +future> prepare_update_type_announcement(storage_proxy& sp, user_type updated_type, api::timestamp_type ts); + +future> prepare_keyspace_drop_announcement(replica::database& db, const sstring& ks_name, api::timestamp_type ts); + +class drop_views_tag; +using drop_views = bool_class; +future> prepare_column_family_drop_announcement(storage_proxy& sp, + const sstring& ks_name, const sstring& cf_name, api::timestamp_type ts, drop_views drop_views = drop_views::no); + +future> prepare_type_drop_announcement(storage_proxy& sp, user_type dropped_type, api::timestamp_type ts); + +future> prepare_new_view_announcement(storage_proxy& sp, view_ptr view, api::timestamp_type ts); + +future> prepare_view_update_announcement(storage_proxy& sp, view_ptr view, api::timestamp_type ts); + +future> prepare_view_drop_announcement(storage_proxy& sp, const sstring& ks_name, const sstring& cf_name, api::timestamp_type ts); + } diff --git a/table_helper.cc b/table_helper.cc index e11c3ed98d..70d9fc5647 100644 --- a/table_helper.cc +++ b/table_helper.cc @@ -52,7 +52,7 @@ future<> table_helper::setup_table(cql3::query_processor& qp, service::migration // The important thing is that it will converge eventually (some traces may // be lost in a process but that's ok). try { - co_return co_await mm.announce(co_await mm.prepare_new_column_family_announcement(b.build(), ts), std::move(group0_guard)); + co_return co_await mm.announce(co_await service::prepare_new_column_family_announcement(qp.proxy(), b.build(), ts), std::move(group0_guard)); } catch (...) {} } @@ -136,7 +136,7 @@ future<> table_helper::setup_keyspace(cql3::query_processor& qp, service::migrat std::map opts; opts["replication_factor"] = replication_factor; auto ksm = keyspace_metadata::new_keyspace(keyspace_name, "org.apache.cassandra.locator.SimpleStrategy", std::move(opts), true); - co_await mm.announce(mm.prepare_new_keyspace_announcement(ksm, ts), std::move(group0_guard)); + co_await mm.announce(service::prepare_new_keyspace_announcement(db.real_database(), ksm, ts), std::move(group0_guard)); } } diff --git a/test/boost/cql_query_test.cc b/test/boost/cql_query_test.cc index 12a9354e20..d547d0b7d1 100644 --- a/test/boost/cql_query_test.cc +++ b/test/boost/cql_query_test.cc @@ -4046,7 +4046,7 @@ SEASTAR_TEST_CASE(test_view_with_two_regular_base_columns_in_key) { auto& mm = e.migration_manager().local(); auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_view_announcement(view_ptr(view_schema), ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_view_announcement(mm.get_storage_proxy(), view_ptr(view_schema), ts).get(), std::move(group0_guard)).get(); // Verify that deleting and restoring columns behaves as expected - i.e. the row is deleted and regenerated cquery_nofail(e, "INSERT INTO t (p, c, v1, v2) VALUES (1, 2, 3, 4)"); diff --git a/test/boost/database_test.cc b/test/boost/database_test.cc index 360188ddde..54a5653579 100644 --- a/test/boost/database_test.cc +++ b/test/boost/database_test.cc @@ -266,13 +266,13 @@ static void test_database(void (*run_tests)(populate_fn_ex, bool), unsigned cgs) auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); e.local_db().find_column_family(s->ks_name(), s->cf_name()); - mm.announce(mm.prepare_column_family_drop_announcement(s->ks_name(), s->cf_name(), ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_column_family_drop_announcement(mm.get_storage_proxy(), s->ks_name(), s->cf_name(), ts).get(), std::move(group0_guard)).get(); } catch (const replica::no_such_column_family&) { // expected } auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_column_family_announcement(s, ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s, ts).get(), std::move(group0_guard)).get(); replica::column_family& cf = e.local_db().find_column_family(s); auto uuid = cf.schema()->id(); for (auto&& m : partitions) { diff --git a/test/boost/group0_cmd_merge_test.cc b/test/boost/group0_cmd_merge_test.cc index 4d08b23e6f..d5cf700790 100644 --- a/test/boost/group0_cmd_merge_test.cc +++ b/test/boost/group0_cmd_merge_test.cc @@ -87,7 +87,7 @@ SEASTAR_TEST_CASE(test_group0_cmd_merge) { }; std::vector cms; size_t size = 0; - auto muts = mm.prepare_keyspace_drop_announcement("ks", api::new_timestamp()).get0(); + auto muts = service::prepare_keyspace_drop_announcement(env.local_db(), "ks", api::new_timestamp()).get0(); // Maximum mutation size is 1/3 of commitlog segment size wich we set // to 1M. Make one command a little bit larger than third of the max size. while (size < 150*1024) { diff --git a/test/boost/memtable_test.cc b/test/boost/memtable_test.cc index b7e758af9d..b622e2d1fb 100644 --- a/test/boost/memtable_test.cc +++ b/test/boost/memtable_test.cc @@ -1001,7 +1001,7 @@ SEASTAR_TEST_CASE(sstable_compaction_does_not_resurrect_data) { .build(); auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_column_family_announcement(s, ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s, ts).get(), std::move(group0_guard)).get(); replica::table& t = db.find_column_family(ks_name, table_name); @@ -1066,7 +1066,7 @@ SEASTAR_TEST_CASE(failed_flush_prevents_writes) { schema_ptr s = ss.schema(); auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_column_family_announcement(s, ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s, ts).get(), std::move(group0_guard)).get(); replica::table& t = db.find_column_family("ks", "cf"); auto memtables = t.active_memtables(); @@ -1134,7 +1134,7 @@ SEASTAR_TEST_CASE(flushing_rate_is_reduced_if_compaction_doesnt_keep_up) { return env.migration_manager().invoke_on(0, [s = global_schema_ptr(std::move(s))] (service::migration_manager& mm) -> future<> { auto group0_guard = co_await mm.start_group0_operation(); auto ts = group0_guard.write_timestamp(); - auto announcement = co_await mm.prepare_new_column_family_announcement(s, ts); + auto announcement = co_await service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s, ts); co_await mm.announce(std::move(announcement), std::move(group0_guard)); }); } @@ -1143,7 +1143,7 @@ SEASTAR_TEST_CASE(flushing_rate_is_reduced_if_compaction_doesnt_keep_up) { return env.migration_manager().invoke_on(0, [shard = this_shard_id()] (service::migration_manager& mm) -> future<> { auto group0_guard = co_await mm.start_group0_operation(); auto ts = group0_guard.write_timestamp(); - auto announcement = co_await mm.prepare_column_family_drop_announcement(ks_name(), cf_name(shard), ts); + auto announcement = co_await service::prepare_column_family_drop_announcement(mm.get_storage_proxy(), ks_name(), cf_name(shard), ts); co_await mm.announce(std::move(announcement), std::move(group0_guard)); }); } diff --git a/test/boost/row_cache_test.cc b/test/boost/row_cache_test.cc index a216659b32..36969d70f7 100644 --- a/test/boost/row_cache_test.cc +++ b/test/boost/row_cache_test.cc @@ -4168,7 +4168,7 @@ SEASTAR_TEST_CASE(row_cache_is_populated_using_compacting_sstable_reader) { .with_column(to_bytes("id"), int32_type) .build(); mm.announce( - mm.prepare_new_column_family_announcement(s, api::new_timestamp()).get(), + service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s, api::new_timestamp()).get(), mm.start_group0_operation().get() ).get(); diff --git a/test/boost/schema_change_test.cc b/test/boost/schema_change_test.cc index cab8e7c85d..e2d4abd4f5 100644 --- a/test/boost/schema_change_test.cc +++ b/test/boost/schema_change_test.cc @@ -47,7 +47,7 @@ SEASTAR_TEST_CASE(test_new_schema_with_no_structural_change_is_propagated) { { auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_column_family_announcement(old_schema, ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_column_family_announcement(mm.get_storage_proxy(), old_schema, ts).get(), std::move(group0_guard)).get(); } auto old_table_version = e.db().local().find_schema(old_schema->id())->version(); @@ -58,7 +58,8 @@ SEASTAR_TEST_CASE(test_new_schema_with_no_structural_change_is_propagated) { auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_column_family_update_announcement(new_schema, false, std::vector(), ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_column_family_update_announcement(mm.get_storage_proxy(), + new_schema, false, std::vector(), ts).get(), std::move(group0_guard)).get(); BOOST_REQUIRE_NE(e.db().local().find_schema(old_schema->id())->version(), old_table_version); BOOST_REQUIRE_NE(e.db().local().get_version(), old_node_version); @@ -81,7 +82,7 @@ SEASTAR_TEST_CASE(test_schema_is_updated_in_keyspace) { { auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_column_family_announcement(old_schema, ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_column_family_announcement(mm.get_storage_proxy(), old_schema, ts).get(), std::move(group0_guard)).get(); } auto s = e.local_db().find_schema(old_schema->id()); @@ -94,7 +95,8 @@ SEASTAR_TEST_CASE(test_schema_is_updated_in_keyspace) { auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_column_family_update_announcement(new_schema, false, std::vector(), ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_column_family_update_announcement(mm.get_storage_proxy(), + new_schema, false, std::vector(), ts).get(), std::move(group0_guard)).get(); s = e.local_db().find_schema(old_schema->id()); BOOST_REQUIRE_NE(*old_schema, *s); @@ -118,7 +120,7 @@ SEASTAR_TEST_CASE(test_tombstones_are_ignored_in_version_calculation) { auto& mm = e.migration_manager().local(); auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_column_family_announcement(table_schema, ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_column_family_announcement(mm.get_storage_proxy(), table_schema, ts).get(), std::move(group0_guard)).get(); auto old_table_version = e.db().local().find_schema(table_schema->id())->version(); auto old_node_version = e.db().local().get_version(); @@ -169,7 +171,7 @@ SEASTAR_TEST_CASE(test_concurrent_column_addition) { { auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_column_family_announcement(s1, ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s1, ts).get(), std::move(group0_guard)).get(); } auto old_version = e.db().local().find_schema(s1->id())->version(); @@ -320,7 +322,7 @@ SEASTAR_TEST_CASE(test_combined_column_add_and_drop) { { auto group0_guard = mm.start_group0_operation().get(); auto ts = group0_guard.write_timestamp(); - mm.announce(mm.prepare_new_column_family_announcement(s1, ts).get(), std::move(group0_guard)).get(); + mm.announce(service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s1, ts).get(), std::move(group0_guard)).get(); } auto&& keyspace = e.db().local().find_keyspace(s1->ks_name()).metadata(); @@ -386,8 +388,8 @@ SEASTAR_TEST_CASE(test_concurrent_table_creation_with_different_schema) { .with_column("v1", bytes_type) .build(); - auto ann1 = mm.prepare_new_column_family_announcement(s1, api::new_timestamp()).get(); - auto ann2 = mm.prepare_new_column_family_announcement(s2, api::new_timestamp()).get(); + auto ann1 = service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s1, api::new_timestamp()).get(); + auto ann2 = service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s2, api::new_timestamp()).get(); { auto group0_guard = mm.start_group0_operation().get(); @@ -477,7 +479,7 @@ SEASTAR_TEST_CASE(test_merging_creates_a_table_even_if_keyspace_was_recreated) { { auto group0_guard = mm.start_group0_operation().get(); const auto ts = group0_guard.write_timestamp(); - auto muts = e.migration_manager().local().prepare_keyspace_drop_announcement("ks", ts).get0(); + auto muts = service::prepare_keyspace_drop_announcement(e.local_db(), "ks", ts).get0(); boost::copy(muts, std::back_inserter(all_muts)); mm.announce(muts, std::move(group0_guard)).get(); } @@ -487,7 +489,7 @@ SEASTAR_TEST_CASE(test_merging_creates_a_table_even_if_keyspace_was_recreated) { const auto ts = group0_guard.write_timestamp(); // all_muts contains keyspace drop. - auto muts = e.migration_manager().local().prepare_new_keyspace_announcement(keyspace, ts); + auto muts = service::prepare_new_keyspace_announcement(e.db().local(), keyspace, ts); boost::copy(muts, std::back_inserter(all_muts)); mm.announce(muts, std::move(group0_guard)).get(); } @@ -496,7 +498,7 @@ SEASTAR_TEST_CASE(test_merging_creates_a_table_even_if_keyspace_was_recreated) { auto group0_guard = mm.start_group0_operation().get(); const auto ts = group0_guard.write_timestamp(); - auto muts = e.migration_manager().local().prepare_new_column_family_announcement(s0, ts).get0(); + auto muts = service::prepare_new_column_family_announcement(mm.get_storage_proxy(), s0, ts).get0(); boost::copy(muts, std::back_inserter(all_muts)); mm.announce(all_muts, std::move(group0_guard)).get(); diff --git a/test/lib/cql_test_env.cc b/test/lib/cql_test_env.cc index 4ae2cb2fa1..2e6f1e2bfc 100644 --- a/test/lib/cql_test_env.cc +++ b/test/lib/cql_test_env.cc @@ -310,7 +310,7 @@ public: auto s = builder.build(schema_builder::compact_storage::no); auto group0_guard = co_await _mm.local().start_group0_operation(); auto ts = group0_guard.write_timestamp(); - co_return co_await _mm.local().announce(co_await _mm.local().prepare_new_column_family_announcement(s, ts), std::move(group0_guard)); + co_return co_await _mm.local().announce(co_await service::prepare_new_column_family_announcement(_proxy.local(), s, ts), std::move(group0_guard)); } virtual future<> require_keyspace_exists(const sstring& ks_name) override { diff --git a/thrift/handler.cc b/thrift/handler.cc index 584258a24e..333f4d1cbc 100644 --- a/thrift/handler.cc +++ b/thrift/handler.cc @@ -885,7 +885,7 @@ public: co_await t._query_state.get_client_state().has_keyspace_access(cf_def.keyspace, auth::permission::CREATE); - co_return co_await t.execute_schema_command([&cf_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + co_return co_await t.execute_schema_command([&p = t._proxy.local(), &cf_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { if (!db.has_keyspace(cf_def.keyspace)) { throw NotFoundException(); } @@ -894,7 +894,7 @@ public: } auto s = schema_from_thrift(cf_def, cf_def.keyspace); - co_return co_await mm.prepare_new_column_family_announcement(std::move(s), ts); + co_return co_await service::prepare_new_column_family_announcement(p, std::move(s), ts); }); }); } @@ -906,7 +906,7 @@ public: co_await t._query_state.get_client_state().has_column_family_access(t.current_keyspace(), column_family, auth::permission::DROP); co_return co_await t.execute_schema_command( - [&column_family, ¤t_keyspace = t.current_keyspace()] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + [&p = t._proxy.local(), &column_family, ¤t_keyspace = t.current_keyspace()] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { auto cf = db.find_table(current_keyspace, column_family); if (cf.schema()->is_view()) { throw make_exception("Cannot drop Materialized Views from Thrift"); @@ -915,7 +915,7 @@ public: throw make_exception("Cannot drop table with Materialized Views {}", column_family); } - co_return co_await mm.prepare_column_family_drop_announcement(current_keyspace, column_family, ts); + co_return co_await service::prepare_column_family_drop_announcement(p, current_keyspace, column_family, ts); }); }); } @@ -929,7 +929,7 @@ public: co_await t._query_state.get_client_state().has_all_keyspaces_access(auth::permission::CREATE); co_return co_await t.execute_schema_command([&ks_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { - co_return mm.prepare_new_keyspace_announcement(keyspace_from_thrift(ks_def), ts); + co_return service::prepare_new_keyspace_announcement(db.real_database(), keyspace_from_thrift(ks_def), ts); }); }); } @@ -948,7 +948,7 @@ public: throw NotFoundException(); } - co_return co_await mm.prepare_keyspace_drop_announcement(keyspace, ts); + co_return co_await service::prepare_keyspace_drop_announcement(db.real_database(), keyspace, ts); }); }); } @@ -971,7 +971,7 @@ public: } auto ksm = keyspace_from_thrift(ks_def); - co_return mm.prepare_keyspace_update_announcement(std::move(ksm), ts); + co_return service::prepare_keyspace_update_announcement(db.real_database(), std::move(ksm), ts); }); }); } @@ -984,7 +984,7 @@ public: co_await t._query_state.get_client_state().has_schema_access(cf_def.keyspace, cf_def.name, auth::permission::ALTER); - co_return co_await t.execute_schema_command([&cf_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + co_return co_await t.execute_schema_command([&p = t._proxy.local(), &cf_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { auto cf = db.find_table(cf_def.keyspace, cf_def.name); auto schema = cf.schema(); @@ -1006,7 +1006,7 @@ public: if (schema->thrift().is_dynamic() != s->thrift().is_dynamic()) { fail(unimplemented::cause::MIXED_CF); } - co_return co_await mm.prepare_column_family_update_announcement(std::move(s), true, std::vector(), ts); + co_return co_await service::prepare_column_family_update_announcement(p, std::move(s), true, std::vector(), ts); }); }); } From 928ee9616c8ab4c13c5da0a5971589cd5eb02103 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Patryk=20J=C4=99drzejczak?= Date: Wed, 26 Jul 2023 12:07:07 +0200 Subject: [PATCH 2/5] alternator: executor::create_keyspace: remove an unused parameter After changing the prepare_ methods of migration_manager to functions, the migration_manager& parameter of executor::create_key has been unused. --- alternator/executor.cc | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/alternator/executor.cc b/alternator/executor.cc index 9f5be7a59a..7b94803a8e 100644 --- a/alternator/executor.cc +++ b/alternator/executor.cc @@ -81,7 +81,7 @@ static sstring_view table_status_to_sstring(table_status tbl_status) { return "UKNOWN"; } -static future> create_keyspace(std::string_view keyspace_name, service::storage_proxy& sp, service::migration_manager& mm, gms::gossiper& gossiper, api::timestamp_type); +static future> create_keyspace(std::string_view keyspace_name, service::storage_proxy& sp, gms::gossiper& gossiper, api::timestamp_type); static map_type attrs_type() { static thread_local auto t = map_type_impl::get_instance(utf8_type, bytes_type, true); @@ -1120,7 +1120,7 @@ static future create_table_on_shard0(tracing::tra auto ts = group0_guard.write_timestamp(); std::vector schema_mutations; try { - schema_mutations = co_await create_keyspace(keyspace_name, sp, mm, gossiper, ts); + schema_mutations = co_await create_keyspace(keyspace_name, sp, gossiper, ts); } catch (exceptions::already_exists_exception&) { if (sp.data_dictionary().has_schema(keyspace_name, table_name)) { co_return api_error::resource_in_use(format("Table {} already exists", table_name)); @@ -4468,7 +4468,7 @@ future executor::describe_continuous_backups(clie // of nodes in the cluster: A cluster with 3 or more live nodes, gets RF=3. // A smaller cluster (presumably, a test only), gets RF=1. The user may // manually create the keyspace to override this predefined behavior. -static future> create_keyspace(std::string_view keyspace_name, service::storage_proxy& sp, service::migration_manager& mm, gms::gossiper& gossiper, api::timestamp_type ts) { +static future> create_keyspace(std::string_view keyspace_name, service::storage_proxy& sp, gms::gossiper& gossiper, api::timestamp_type ts) { sstring keyspace_name_str(keyspace_name); int endpoint_count = gossiper.get_endpoint_states().size(); int rf = 3; From b6ead8de10abf8d29fb51e3199f396f7f1df7c97 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Patryk=20J=C4=99drzejczak?= Date: Wed, 26 Jul 2023 15:26:19 +0200 Subject: [PATCH 3/5] cql3: alter_type_statement::prepare_announcement_mutations: change parameters After changing the prepare_ methods of migration_manager to functions, the migration_manager& parameter of alter_type_statement::prepare_announcement_mutations has become unneeded. However, the function needs access to service::storage_proxy and data_dictionary::database. Passing storage_proxy& to it is enough. --- cql3/statements/alter_type_statement.cc | 14 +++++++------- cql3/statements/alter_type_statement.hh | 2 +- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/cql3/statements/alter_type_statement.cc b/cql3/statements/alter_type_statement.cc index 7464a4fe67..1c9781171c 100644 --- a/cql3/statements/alter_type_statement.cc +++ b/cql3/statements/alter_type_statement.cc @@ -47,9 +47,9 @@ const sstring& alter_type_statement::keyspace() const return _name.get_keyspace(); } -future> alter_type_statement::prepare_announcement_mutations(data_dictionary::database db, service::migration_manager& mm, api::timestamp_type ts) const { +future> alter_type_statement::prepare_announcement_mutations(service::storage_proxy& sp, api::timestamp_type ts) const { std::vector m; - auto&& ks = db.find_keyspace(keyspace()); + auto&& ks = sp.data_dictionary().find_keyspace(keyspace()); auto&& all_types = ks.metadata()->user_types().get_all_types(); auto to_update = all_types.find(_name.get_user_type_name()); // Shouldn't happen, unless we race with a drop @@ -66,10 +66,10 @@ future> alter_type_statement::prepare_announcement_mutatio } } - auto&& updated = make_updated_type(db, to_update->second); + auto&& updated = make_updated_type(sp.data_dictionary(), to_update->second); // Now, we need to announce the type update to basically change it for new tables using this type, // but we also need to find all existing user types and CF using it and change them. - auto res = co_await service::prepare_update_type_announcement(mm.get_storage_proxy(), updated, ts); + auto res = co_await service::prepare_update_type_announcement(sp, updated, ts); std::move(res.begin(), res.end(), std::back_inserter(m)); for (auto&& schema : ks.metadata()->cf_meta_data() | boost::adaptors::map_values) { @@ -85,10 +85,10 @@ future> alter_type_statement::prepare_announcement_mutatio } if (modified) { if (schema->is_view()) { - auto res = co_await service::prepare_view_update_announcement(mm.get_storage_proxy(), view_ptr(cfm.build()), ts); + auto res = co_await service::prepare_view_update_announcement(sp, view_ptr(cfm.build()), ts); std::move(res.begin(), res.end(), std::back_inserter(m)); } else { - auto res = co_await service::prepare_column_family_update_announcement(mm.get_storage_proxy(), cfm.build(), false, {}, ts); + auto res = co_await service::prepare_column_family_update_announcement(sp, cfm.build(), false, {}, ts); std::move(res.begin(), res.end(), std::back_inserter(m)); } } @@ -100,7 +100,7 @@ future> alter_type_statement::prepare_announcement_mutatio future, std::vector, cql3::cql_warnings_vec>> alter_type_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { try { - auto m = co_await prepare_announcement_mutations(qp.db(), mm, ts); + auto m = co_await prepare_announcement_mutations(qp.proxy(), ts); using namespace cql_transport; auto ret = ::make_shared( diff --git a/cql3/statements/alter_type_statement.hh b/cql3/statements/alter_type_statement.hh index 15678cebdf..9c9405ceb8 100644 --- a/cql3/statements/alter_type_statement.hh +++ b/cql3/statements/alter_type_statement.hh @@ -48,7 +48,7 @@ private: virtual future<> operator()(schema_ptr cfm, bool from_thrift, std::vector&& view_updates, std::optional ts_opt) = 0; }; - future> prepare_announcement_mutations(data_dictionary::database db, service::migration_manager& mm, api::timestamp_type) const; + future> prepare_announcement_mutations(service::storage_proxy& sp, api::timestamp_type) const; }; class alter_type_statement::add_or_alter : public alter_type_statement { From ffc3c1302e556ec6fb5b0a07abc0e7e6e5aa6049 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Patryk=20J=C4=99drzejczak?= Date: Wed, 26 Jul 2023 15:59:08 +0200 Subject: [PATCH 4/5] cql3: schema_altering_statement::prepare_schema_mutations: remove an unused parameter After changing the prepare_ methods of migration_manager to functions, the migration_manager& parameter of schema_altering_statement::prepare_schema_mutations has been unused by all classes inheriting from schema_altering_statement. --- cql3/query_processor.cc | 2 +- cql3/statements/alter_keyspace_statement.cc | 2 +- cql3/statements/alter_keyspace_statement.hh | 2 +- cql3/statements/alter_table_statement.cc | 2 +- cql3/statements/alter_table_statement.hh | 2 +- cql3/statements/alter_type_statement.cc | 2 +- cql3/statements/alter_type_statement.hh | 2 +- cql3/statements/alter_view_statement.cc | 2 +- cql3/statements/alter_view_statement.hh | 2 +- cql3/statements/create_aggregate_statement.cc | 2 +- cql3/statements/create_aggregate_statement.hh | 2 +- cql3/statements/create_function_statement.cc | 2 +- cql3/statements/create_function_statement.hh | 2 +- cql3/statements/create_index_statement.cc | 2 +- cql3/statements/create_index_statement.hh | 2 +- cql3/statements/create_keyspace_statement.cc | 2 +- cql3/statements/create_keyspace_statement.hh | 2 +- cql3/statements/create_table_statement.cc | 2 +- cql3/statements/create_table_statement.hh | 2 +- cql3/statements/create_type_statement.cc | 2 +- cql3/statements/create_type_statement.hh | 2 +- cql3/statements/create_view_statement.cc | 2 +- cql3/statements/create_view_statement.hh | 2 +- cql3/statements/drop_aggregate_statement.cc | 2 +- cql3/statements/drop_aggregate_statement.hh | 2 +- cql3/statements/drop_function_statement.cc | 2 +- cql3/statements/drop_function_statement.hh | 2 +- cql3/statements/drop_index_statement.cc | 2 +- cql3/statements/drop_index_statement.hh | 2 +- cql3/statements/drop_keyspace_statement.cc | 2 +- cql3/statements/drop_keyspace_statement.hh | 2 +- cql3/statements/drop_table_statement.cc | 2 +- cql3/statements/drop_table_statement.hh | 2 +- cql3/statements/drop_type_statement.cc | 2 +- cql3/statements/drop_type_statement.hh | 2 +- cql3/statements/drop_view_statement.cc | 2 +- cql3/statements/drop_view_statement.hh | 2 +- cql3/statements/schema_altering_statement.hh | 6 +----- 38 files changed, 38 insertions(+), 42 deletions(-) diff --git a/cql3/query_processor.cc b/cql3/query_processor.cc index 4df19499d7..6de4c19d00 100644 --- a/cql3/query_processor.cc +++ b/cql3/query_processor.cc @@ -890,7 +890,7 @@ query_processor::execute_schema_statement(const statements::schema_altering_stat try { auto group0_guard = co_await mm.start_group0_operation(); - auto [ret, m, cql_warnings] = co_await stmt.prepare_schema_mutations(*this, mm, group0_guard.write_timestamp()); + auto [ret, m, cql_warnings] = co_await stmt.prepare_schema_mutations(*this, group0_guard.write_timestamp()); warnings = std::move(cql_warnings); if (!m.empty()) { diff --git a/cql3/statements/alter_keyspace_statement.cc b/cql3/statements/alter_keyspace_statement.cc index bbe74dfd38..1328ea6e21 100644 --- a/cql3/statements/alter_keyspace_statement.cc +++ b/cql3/statements/alter_keyspace_statement.cc @@ -75,7 +75,7 @@ void cql3::statements::alter_keyspace_statement::validate(query_processor& qp, c } future, std::vector, cql3::cql_warnings_vec>> -cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { try { auto old_ksm = qp.db().find_keyspace(_name).metadata(); const auto& tm = *qp.proxy().get_token_metadata_ptr(); diff --git a/cql3/statements/alter_keyspace_statement.hh b/cql3/statements/alter_keyspace_statement.hh index 0d40b50877..4abddd960d 100644 --- a/cql3/statements/alter_keyspace_statement.hh +++ b/cql3/statements/alter_keyspace_statement.hh @@ -33,7 +33,7 @@ public: future<> check_access(query_processor& qp, const service::client_state& state) const override; void validate(query_processor& qp, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; virtual future<::shared_ptr> execute(query_processor& qp, service::query_state& state, const query_options& options) const override; }; diff --git a/cql3/statements/alter_table_statement.cc b/cql3/statements/alter_table_statement.cc index c39035a566..cb9243fe63 100644 --- a/cql3/statements/alter_table_statement.cc +++ b/cql3/statements/alter_table_statement.cc @@ -381,7 +381,7 @@ std::pair> alter_table_statement::prepare_ } future, std::vector, cql3::cql_warnings_vec>> -alter_table_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +alter_table_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { data_dictionary::database db = qp.db(); auto [cfm, view_updates] = prepare_schema_update(db); auto m = co_await service::prepare_column_family_update_announcement(qp.proxy(), cfm.build(), false, std::move(view_updates), ts); diff --git a/cql3/statements/alter_table_statement.hh b/cql3/statements/alter_table_statement.hh index a509f3e06b..991e44fb4f 100644 --- a/cql3/statements/alter_table_statement.hh +++ b/cql3/statements/alter_table_statement.hh @@ -54,7 +54,7 @@ public: virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; virtual future<::shared_ptr> execute(query_processor& qp, service::query_state& state, const query_options& options) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; private: void add_column(const schema& schema, data_dictionary::table cf, schema_builder& cfm, std::vector& view_updates, const column_identifier& column_name, const cql3_type validator, const column_definition* def, bool is_static) const; void alter_column(const schema& schema, data_dictionary::table cf, schema_builder& cfm, std::vector& view_updates, const column_identifier& column_name, const cql3_type validator, const column_definition* def, bool is_static) const; diff --git a/cql3/statements/alter_type_statement.cc b/cql3/statements/alter_type_statement.cc index 1c9781171c..08675bafc3 100644 --- a/cql3/statements/alter_type_statement.cc +++ b/cql3/statements/alter_type_statement.cc @@ -98,7 +98,7 @@ future> alter_type_statement::prepare_announcement_mutatio } future, std::vector, cql3::cql_warnings_vec>> -alter_type_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +alter_type_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { try { auto m = co_await prepare_announcement_mutations(qp.proxy(), ts); diff --git a/cql3/statements/alter_type_statement.hh b/cql3/statements/alter_type_statement.hh index 9c9405ceb8..a6cf7939a6 100644 --- a/cql3/statements/alter_type_statement.hh +++ b/cql3/statements/alter_type_statement.hh @@ -35,7 +35,7 @@ public: virtual const sstring& keyspace() const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; class add_or_alter; class renames; diff --git a/cql3/statements/alter_view_statement.cc b/cql3/statements/alter_view_statement.cc index 55518f3919..1e7d385724 100644 --- a/cql3/statements/alter_view_statement.cc +++ b/cql3/statements/alter_view_statement.cc @@ -76,7 +76,7 @@ view_ptr alter_view_statement::prepare_view(data_dictionary::database db) const return view_ptr(builder.build()); } -future, std::vector, cql3::cql_warnings_vec>> alter_view_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +future, std::vector, cql3::cql_warnings_vec>> alter_view_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { auto m = co_await service::prepare_view_update_announcement(qp.proxy(), prepare_view(qp.db()), ts); using namespace cql_transport; diff --git a/cql3/statements/alter_view_statement.hh b/cql3/statements/alter_view_statement.hh index b16f25cbd1..5f246ea596 100644 --- a/cql3/statements/alter_view_statement.hh +++ b/cql3/statements/alter_view_statement.hh @@ -33,7 +33,7 @@ public: virtual future<> check_access(query_processor& qp, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; }; diff --git a/cql3/statements/create_aggregate_statement.cc b/cql3/statements/create_aggregate_statement.cc index 3fcca4a8ca..2e5110255e 100644 --- a/cql3/statements/create_aggregate_statement.cc +++ b/cql3/statements/create_aggregate_statement.cc @@ -84,7 +84,7 @@ std::unique_ptr create_aggregate_statement::prepare(data_dic } future, std::vector, cql3::cql_warnings_vec>> -create_aggregate_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +create_aggregate_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; diff --git a/cql3/statements/create_aggregate_statement.hh b/cql3/statements/create_aggregate_statement.hh index 4d5699b472..1a0419903d 100644 --- a/cql3/statements/create_aggregate_statement.hh +++ b/cql3/statements/create_aggregate_statement.hh @@ -25,7 +25,7 @@ namespace statements { class create_aggregate_statement final : public create_function_statement_base { virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual future<> check_access(query_processor& qp, const service::client_state& state) const override; virtual seastar::future> create(query_processor& qp, db::functions::function* old) const override; diff --git a/cql3/statements/create_function_statement.cc b/cql3/statements/create_function_statement.cc index a4606e6543..21fc98b08f 100644 --- a/cql3/statements/create_function_statement.cc +++ b/cql3/statements/create_function_statement.cc @@ -66,7 +66,7 @@ std::unique_ptr create_function_statement::prepare(data_dict } future, std::vector, cql3::cql_warnings_vec>> -create_function_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +create_function_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; diff --git a/cql3/statements/create_function_statement.hh b/cql3/statements/create_function_statement.hh index 278cf4db44..3571ad980e 100644 --- a/cql3/statements/create_function_statement.hh +++ b/cql3/statements/create_function_statement.hh @@ -23,7 +23,7 @@ namespace statements { class create_function_statement final : public create_function_statement_base { virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual seastar::future> create(query_processor& qp, db::functions::function* old) const override; sstring _language; diff --git a/cql3/statements/create_index_statement.cc b/cql3/statements/create_index_statement.cc index 5a6e7fe009..11f76cf92d 100644 --- a/cql3/statements/create_index_statement.cc +++ b/cql3/statements/create_index_statement.cc @@ -376,7 +376,7 @@ schema_ptr create_index_statement::build_index_schema(query_processor& qp) const } future, std::vector, cql3::cql_warnings_vec>> -create_index_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +create_index_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { using namespace cql_transport; auto schema = build_index_schema(qp); diff --git a/cql3/statements/create_index_statement.hh b/cql3/statements/create_index_statement.hh index 294f1192ab..66cb468880 100644 --- a/cql3/statements/create_index_statement.hh +++ b/cql3/statements/create_index_statement.hh @@ -47,7 +47,7 @@ public: future<> check_access(query_processor& qp, const service::client_state& state) const override; void validate(query_processor&, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; diff --git a/cql3/statements/create_keyspace_statement.cc b/cql3/statements/create_keyspace_statement.cc index 5cfa5b2c52..062caf189a 100644 --- a/cql3/statements/create_keyspace_statement.cc +++ b/cql3/statements/create_keyspace_statement.cc @@ -93,7 +93,7 @@ void create_keyspace_statement::validate(query_processor& qp, const service::cli #endif } -future, std::vector, cql3::cql_warnings_vec>> create_keyspace_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +future, std::vector, cql3::cql_warnings_vec>> create_keyspace_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { using namespace cql_transport; const auto& tm = *qp.proxy().get_token_metadata_ptr(); ::shared_ptr ret; diff --git a/cql3/statements/create_keyspace_statement.hh b/cql3/statements/create_keyspace_statement.hh index 5d5c4161fb..55d14ae361 100644 --- a/cql3/statements/create_keyspace_statement.hh +++ b/cql3/statements/create_keyspace_statement.hh @@ -64,7 +64,7 @@ public: virtual void validate(query_processor&, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; diff --git a/cql3/statements/create_table_statement.cc b/cql3/statements/create_table_statement.cc index ab8e48efad..cef65944a5 100644 --- a/cql3/statements/create_table_statement.cc +++ b/cql3/statements/create_table_statement.cc @@ -71,7 +71,7 @@ std::vector create_table_statement::get_columns() const } future, std::vector, cql3::cql_warnings_vec>> -create_table_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +create_table_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; diff --git a/cql3/statements/create_table_statement.hh b/cql3/statements/create_table_statement.hh index 9ff88ea86d..6836b6145c 100644 --- a/cql3/statements/create_table_statement.hh +++ b/cql3/statements/create_table_statement.hh @@ -70,7 +70,7 @@ public: virtual future<> check_access(query_processor& qp, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; diff --git a/cql3/statements/create_type_statement.cc b/cql3/statements/create_type_statement.cc index d756366cca..444c224bf9 100644 --- a/cql3/statements/create_type_statement.cc +++ b/cql3/statements/create_type_statement.cc @@ -118,7 +118,7 @@ std::optional create_type_statement::make_type(query_processor& qp) c return type; } -future, std::vector, cql3::cql_warnings_vec>> create_type_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +future, std::vector, cql3::cql_warnings_vec>> create_type_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; try { diff --git a/cql3/statements/create_type_statement.hh b/cql3/statements/create_type_statement.hh index 777bc91218..fe3252a45b 100644 --- a/cql3/statements/create_type_statement.hh +++ b/cql3/statements/create_type_statement.hh @@ -37,7 +37,7 @@ public: virtual const sstring& keyspace() const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; diff --git a/cql3/statements/create_view_statement.cc b/cql3/statements/create_view_statement.cc index a0e7358c2d..cfcbcc43a7 100644 --- a/cql3/statements/create_view_statement.cc +++ b/cql3/statements/create_view_statement.cc @@ -362,7 +362,7 @@ std::pair create_view_statement::prepare_view( } future, std::vector, cql3::cql_warnings_vec>> -create_view_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +create_view_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; auto [definition, warnings] = prepare_view(qp.db()); diff --git a/cql3/statements/create_view_statement.hh b/cql3/statements/create_view_statement.hh index a3142d8ccb..cbe279ceb3 100644 --- a/cql3/statements/create_view_statement.hh +++ b/cql3/statements/create_view_statement.hh @@ -57,7 +57,7 @@ public: // Functions we need to override to subclass schema_altering_statement virtual future<> check_access(query_processor& qp, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; diff --git a/cql3/statements/drop_aggregate_statement.cc b/cql3/statements/drop_aggregate_statement.cc index ca44fef5f2..7c2c559fee 100644 --- a/cql3/statements/drop_aggregate_statement.cc +++ b/cql3/statements/drop_aggregate_statement.cc @@ -24,7 +24,7 @@ std::unique_ptr drop_aggregate_statement::prepare(data_dicti } future, std::vector, cql3::cql_warnings_vec>> -drop_aggregate_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +drop_aggregate_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; diff --git a/cql3/statements/drop_aggregate_statement.hh b/cql3/statements/drop_aggregate_statement.hh index ba94ad2f50..7cad55ffef 100644 --- a/cql3/statements/drop_aggregate_statement.hh +++ b/cql3/statements/drop_aggregate_statement.hh @@ -15,7 +15,7 @@ class query_processor; namespace statements { class drop_aggregate_statement final : public drop_function_statement_base { virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; public: drop_aggregate_statement(functions::function_name name, std::vector> arg_types, diff --git a/cql3/statements/drop_function_statement.cc b/cql3/statements/drop_function_statement.cc index 04cbc3e024..7f1c23097a 100644 --- a/cql3/statements/drop_function_statement.cc +++ b/cql3/statements/drop_function_statement.cc @@ -24,7 +24,7 @@ std::unique_ptr drop_function_statement::prepare(data_dictio } future, std::vector, cql3::cql_warnings_vec>> -drop_function_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +drop_function_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; diff --git a/cql3/statements/drop_function_statement.hh b/cql3/statements/drop_function_statement.hh index 6ca52a62cb..63db8d72ae 100644 --- a/cql3/statements/drop_function_statement.hh +++ b/cql3/statements/drop_function_statement.hh @@ -15,7 +15,7 @@ class query_processor; namespace statements { class drop_function_statement final : public drop_function_statement_base { virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; public: drop_function_statement(functions::function_name name, std::vector> arg_types, diff --git a/cql3/statements/drop_index_statement.cc b/cql3/statements/drop_index_statement.cc index d3089eaa14..dc26cd9fdc 100644 --- a/cql3/statements/drop_index_statement.cc +++ b/cql3/statements/drop_index_statement.cc @@ -73,7 +73,7 @@ schema_ptr drop_index_statement::make_drop_idex_schema(query_processor& qp) cons } future, std::vector, cql3::cql_warnings_vec>> -drop_index_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +drop_index_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; auto cfm = make_drop_idex_schema(qp); diff --git a/cql3/statements/drop_index_statement.hh b/cql3/statements/drop_index_statement.hh index 9f71858df6..e25b93500c 100644 --- a/cql3/statements/drop_index_statement.hh +++ b/cql3/statements/drop_index_statement.hh @@ -44,7 +44,7 @@ public: virtual void validate(query_processor&, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; private: diff --git a/cql3/statements/drop_keyspace_statement.cc b/cql3/statements/drop_keyspace_statement.cc index ab8d10e3fb..d67564d545 100644 --- a/cql3/statements/drop_keyspace_statement.cc +++ b/cql3/statements/drop_keyspace_statement.cc @@ -47,7 +47,7 @@ const sstring& drop_keyspace_statement::keyspace() const } future, std::vector, cql3::cql_warnings_vec>> -drop_keyspace_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +drop_keyspace_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { std::vector m; ::shared_ptr ret; diff --git a/cql3/statements/drop_keyspace_statement.hh b/cql3/statements/drop_keyspace_statement.hh index 0dfcfbf873..9f2a6b4603 100644 --- a/cql3/statements/drop_keyspace_statement.hh +++ b/cql3/statements/drop_keyspace_statement.hh @@ -30,7 +30,7 @@ public: virtual const sstring& keyspace() const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; }; diff --git a/cql3/statements/drop_table_statement.cc b/cql3/statements/drop_table_statement.cc index 6814e83d6c..1a2f2807b1 100644 --- a/cql3/statements/drop_table_statement.cc +++ b/cql3/statements/drop_table_statement.cc @@ -33,7 +33,7 @@ future<> drop_table_statement::check_access(query_processor& qp, const service:: } future, std::vector, cql3::cql_warnings_vec>> -drop_table_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +drop_table_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; diff --git a/cql3/statements/drop_table_statement.hh b/cql3/statements/drop_table_statement.hh index 5ba9ceebdf..a7823f64ec 100644 --- a/cql3/statements/drop_table_statement.hh +++ b/cql3/statements/drop_table_statement.hh @@ -26,7 +26,7 @@ public: virtual future<> check_access(query_processor& qp, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; }; diff --git a/cql3/statements/drop_type_statement.cc b/cql3/statements/drop_type_statement.cc index 2b2ea23cea..51ecb0a647 100644 --- a/cql3/statements/drop_type_statement.cc +++ b/cql3/statements/drop_type_statement.cc @@ -121,7 +121,7 @@ const sstring& drop_type_statement::keyspace() const } future, std::vector, cql3::cql_warnings_vec>> -drop_type_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +drop_type_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { validate_while_executing(qp); data_dictionary::database db = qp.db(); diff --git a/cql3/statements/drop_type_statement.hh b/cql3/statements/drop_type_statement.hh index cd9589bf80..d81059e053 100644 --- a/cql3/statements/drop_type_statement.hh +++ b/cql3/statements/drop_type_statement.hh @@ -29,7 +29,7 @@ public: virtual const sstring& keyspace() const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; diff --git a/cql3/statements/drop_view_statement.cc b/cql3/statements/drop_view_statement.cc index 92170207a7..0f068a54fb 100644 --- a/cql3/statements/drop_view_statement.cc +++ b/cql3/statements/drop_view_statement.cc @@ -42,7 +42,7 @@ future<> drop_view_statement::check_access(query_processor& qp, const service::c } future, std::vector, cql3::cql_warnings_vec>> -drop_view_statement::prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type ts) const { +drop_view_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const { ::shared_ptr ret; std::vector m; diff --git a/cql3/statements/drop_view_statement.hh b/cql3/statements/drop_view_statement.hh index 4fd4bdd81c..10d072b2c2 100644 --- a/cql3/statements/drop_view_statement.hh +++ b/cql3/statements/drop_view_statement.hh @@ -32,7 +32,7 @@ public: virtual future<> check_access(query_processor& qp, const service::client_state& state) const override; - future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const override; + future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override; virtual std::unique_ptr prepare(data_dictionary::database db, cql_stats& stats) override; }; diff --git a/cql3/statements/schema_altering_statement.hh b/cql3/statements/schema_altering_statement.hh index ccc644351d..fbdd4965bb 100644 --- a/cql3/statements/schema_altering_statement.hh +++ b/cql3/statements/schema_altering_statement.hh @@ -20,10 +20,6 @@ class mutation; -namespace service { -class migration_manager; -} - namespace cql3 { class query_processor; @@ -62,7 +58,7 @@ protected: execute(query_processor& qp, service::query_state& state, const query_options& options) const override; public: - virtual future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, service::migration_manager& mm, api::timestamp_type) const = 0; + virtual future, std::vector, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const = 0; }; } From 233d801f395e066ff6415b0f447ec60bc65bf439 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Patryk=20J=C4=99drzejczak?= Date: Thu, 27 Jul 2023 12:19:05 +0200 Subject: [PATCH 5/5] cql3: query_processor::execute_thrift_schema_command: remove an unused parameter After changing the prepare_ methods of migration_manager to functions, the migration_manager& parameter of query_processor::execute_thrift_schema_command and thrift::handler::execute_schema_command (that calls query_processor::execute_thrift_schema_command) has been unused. --- cql3/query_processor.cc | 4 ++-- cql3/query_processor.hh | 2 +- thrift/handler.cc | 14 +++++++------- 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/cql3/query_processor.cc b/cql3/query_processor.cc index 6de4c19d00..a2dd4309a9 100644 --- a/cql3/query_processor.cc +++ b/cql3/query_processor.cc @@ -929,7 +929,7 @@ query_processor::execute_schema_statement(const statements::schema_altering_stat future query_processor::execute_thrift_schema_command( std::function>( - service::migration_manager&, data_dictionary::database, api::timestamp_type) + data_dictionary::database, api::timestamp_type) > prepare_schema_mutations) { assert(this_shard_id() == 0); @@ -938,7 +938,7 @@ query_processor::execute_thrift_schema_command( auto group0_guard = co_await mm.start_group0_operation(); auto ts = group0_guard.write_timestamp(); - co_await mm.announce(co_await prepare_schema_mutations(mm, db(), ts), std::move(group0_guard)); + co_await mm.announce(co_await prepare_schema_mutations(db(), ts), std::move(group0_guard)); co_return std::string(db().get_version().to_sstring()); } diff --git a/cql3/query_processor.hh b/cql3/query_processor.hh index 2c510a4622..c956f02eb2 100644 --- a/cql3/query_processor.hh +++ b/cql3/query_processor.hh @@ -408,7 +408,7 @@ public: future execute_thrift_schema_command( std::function>( - service::migration_manager&, data_dictionary::database, api::timestamp_type) + data_dictionary::database, api::timestamp_type) > prepare_schema_mutations); std::unique_ptr get_statement( diff --git a/thrift/handler.cc b/thrift/handler.cc index 333f4d1cbc..d1ea66fe6e 100644 --- a/thrift/handler.cc +++ b/thrift/handler.cc @@ -871,7 +871,7 @@ public: }); } - future execute_schema_command(std::function>(service::migration_manager&, data_dictionary::database, api::timestamp_type)> ddl) { + future execute_schema_command(std::function>(data_dictionary::database, api::timestamp_type)> ddl) { return _query_processor.invoke_on(0, [ddl = std::move(ddl)] (cql3::query_processor& qp) mutable { return qp.execute_thrift_schema_command(std::move(ddl)); }); @@ -885,7 +885,7 @@ public: co_await t._query_state.get_client_state().has_keyspace_access(cf_def.keyspace, auth::permission::CREATE); - co_return co_await t.execute_schema_command([&p = t._proxy.local(), &cf_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + co_return co_await t.execute_schema_command([&p = t._proxy.local(), &cf_def] (data_dictionary::database db, api::timestamp_type ts) -> future> { if (!db.has_keyspace(cf_def.keyspace)) { throw NotFoundException(); } @@ -906,7 +906,7 @@ public: co_await t._query_state.get_client_state().has_column_family_access(t.current_keyspace(), column_family, auth::permission::DROP); co_return co_await t.execute_schema_command( - [&p = t._proxy.local(), &column_family, ¤t_keyspace = t.current_keyspace()] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + [&p = t._proxy.local(), &column_family, ¤t_keyspace = t.current_keyspace()] (data_dictionary::database db, api::timestamp_type ts) -> future> { auto cf = db.find_table(current_keyspace, column_family); if (cf.schema()->is_view()) { throw make_exception("Cannot drop Materialized Views from Thrift"); @@ -928,7 +928,7 @@ public: co_await t._query_state.get_client_state().has_all_keyspaces_access(auth::permission::CREATE); - co_return co_await t.execute_schema_command([&ks_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + co_return co_await t.execute_schema_command([&ks_def] (data_dictionary::database db, api::timestamp_type ts) -> future> { co_return service::prepare_new_keyspace_announcement(db.real_database(), keyspace_from_thrift(ks_def), ts); }); }); @@ -942,7 +942,7 @@ public: co_await t._query_state.get_client_state().has_keyspace_access(keyspace, auth::permission::DROP); - co_return co_await t.execute_schema_command([&keyspace] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + co_return co_await t.execute_schema_command([&keyspace] (data_dictionary::database db, api::timestamp_type ts) -> future> { thrift_validation::validate_keyspace_not_system(keyspace); if (!db.has_keyspace(keyspace)) { throw NotFoundException(); @@ -962,7 +962,7 @@ public: co_await t._query_state.get_client_state().has_keyspace_access(ks_def.name, auth::permission::ALTER); - co_return co_await t.execute_schema_command([&ks_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + co_return co_await t.execute_schema_command([&ks_def] (data_dictionary::database db, api::timestamp_type ts) -> future> { if (!db.has_keyspace(ks_def.name)) { throw NotFoundException(); } @@ -984,7 +984,7 @@ public: co_await t._query_state.get_client_state().has_schema_access(cf_def.keyspace, cf_def.name, auth::permission::ALTER); - co_return co_await t.execute_schema_command([&p = t._proxy.local(), &cf_def] (service::migration_manager& mm, data_dictionary::database db, api::timestamp_type ts) -> future> { + co_return co_await t.execute_schema_command([&p = t._proxy.local(), &cf_def] (data_dictionary::database db, api::timestamp_type ts) -> future> { auto cf = db.find_table(cf_def.keyspace, cf_def.name); auto schema = cf.schema();