| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| /* | ||
| * Copyright (C) 2023-present ScyllaDB | ||
| */ | ||
|
|
||
| /* | ||
| * SPDX-License-Identifier: AGPL-3.0-or-later | ||
| */ | ||
|
|
||
| #include "locator/tablets.hh" | ||
| #include "replica/tablets.hh" | ||
| #include "locator/tablet_replication_strategy.hh" | ||
| #include "replica/database.hh" | ||
| #include "service/migration_manager.hh" | ||
| #include "service/tablet_allocator.hh" | ||
|
|
||
| using namespace locator; | ||
| using namespace replica; | ||
|
|
||
| namespace service { | ||
|
|
||
| class tablet_allocator_impl : public tablet_allocator::impl | ||
| , public service::migration_listener::empty_listener { | ||
| service::migration_notifier& _migration_notifier; | ||
| replica::database& _db; | ||
| bool _stopped = false; | ||
| public: | ||
| tablet_allocator_impl(service::migration_notifier& mn, replica::database& db) | ||
| : _migration_notifier(mn) | ||
| , _db(db) { | ||
| _migration_notifier.register_listener(this); | ||
| } | ||
|
|
||
| tablet_allocator_impl(tablet_allocator_impl&&) = delete; // "this" captured. | ||
|
|
||
| ~tablet_allocator_impl() { | ||
| assert(_stopped); | ||
| } | ||
|
|
||
| future<> stop() { | ||
| co_await _migration_notifier.unregister_listener(this); | ||
| _stopped = true; | ||
| } | ||
|
|
||
| void on_before_create_column_family(const schema& s, std::vector<mutation>& muts, api::timestamp_type ts) override { | ||
| keyspace& ks = _db.find_keyspace(s.ks_name()); | ||
| auto&& rs = ks.get_replication_strategy(); | ||
| if (auto&& tablet_rs = rs.maybe_as_tablet_aware()) { | ||
| auto tm = _db.get_shared_token_metadata().get(); | ||
| auto map = tablet_rs->allocate_tablets_for_new_table(s.shared_from_this(), tm).get0(); | ||
| muts.emplace_back(tablet_map_to_mutation(map, s.id(), s.keypace_name(), s.cf_name(), ts).get0()); | ||
| } | ||
| } | ||
|
|
||
| void on_before_drop_column_family(const schema& s, std::vector<mutation>& muts, api::timestamp_type ts) override { | ||
| keyspace& ks = _db.find_keyspace(s.ks_name()); | ||
| auto&& rs = ks.get_replication_strategy(); | ||
| std::vector<mutation> result; | ||
| if (rs.uses_tablets()) { | ||
| auto tm = _db.get_shared_token_metadata().get(); | ||
| muts.emplace_back(make_drop_tablet_map_mutation(s.keypace_name(), s.id(), ts)); | ||
| } | ||
| } | ||
|
|
||
| void on_before_drop_keyspace(const sstring& keyspace_name, std::vector<mutation>& muts, api::timestamp_type ts) override { | ||
| keyspace& ks = _db.find_keyspace(keyspace_name); | ||
| auto&& rs = ks.get_replication_strategy(); | ||
| if (rs.uses_tablets()) { | ||
| auto tm = _db.get_shared_token_metadata().get(); | ||
| for (auto&& [name, s] : ks.metadata()->cf_meta_data()) { | ||
| muts.emplace_back(make_drop_tablet_map_mutation(keyspace_name, s->id(), ts)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // FIXME: Handle materialized views. | ||
| }; | ||
|
|
||
| tablet_allocator::tablet_allocator(service::migration_notifier& mn, replica::database& db) | ||
| : _impl(std::make_unique<tablet_allocator_impl>(mn, db)) { | ||
| } | ||
|
|
||
| future<> tablet_allocator::stop() { | ||
| return impl().stop(); | ||
| } | ||
|
|
||
| tablet_allocator_impl& tablet_allocator::impl() { | ||
| return static_cast<tablet_allocator_impl&>(*_impl); | ||
| } | ||
|
|
||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| /* | ||
| * Copyright (C) 2023-present ScyllaDB | ||
| */ | ||
|
|
||
| /* | ||
| * SPDX-License-Identifier: AGPL-3.0-or-later | ||
| */ | ||
|
|
||
| #pragma once | ||
|
|
||
| #include "replica/database.hh" | ||
| #include "service/migration_manager.hh" | ||
| #include <any> | ||
|
|
||
| namespace service { | ||
|
|
||
| class tablet_allocator_impl; | ||
|
|
||
| class tablet_allocator { | ||
| public: | ||
| class impl { | ||
| public: | ||
| virtual ~impl() = default; | ||
| }; | ||
| private: | ||
| std::unique_ptr<impl> _impl; | ||
| tablet_allocator_impl& impl(); | ||
| public: | ||
| tablet_allocator(service::migration_notifier& mn, replica::database& db); | ||
| public: | ||
| future<> stop(); | ||
| }; | ||
|
|
||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,270 @@ | ||
| /* | ||
| * Copyright (C) 2023-present-2020 ScyllaDB | ||
| */ | ||
|
|
||
| /* | ||
| * SPDX-License-Identifier: AGPL-3.0-or-later | ||
| */ | ||
|
|
||
|
|
||
|
|
||
| #include "test/lib/scylla_test_case.hh" | ||
| #include <seastar/testing/thread_test_case.hh> | ||
| #include "test/lib/cql_test_env.hh" | ||
| #include "test/lib/log.hh" | ||
| #include "db/config.hh" | ||
| #include "schema/schema_builder.hh" | ||
|
|
||
| #include "replica/tablets.hh" | ||
| #include "locator/tablets.hh" | ||
| #include "locator/tablet_replication_strategy.hh" | ||
|
|
||
| using namespace locator; | ||
| using namespace replica; | ||
|
|
||
| static api::timestamp_type next_timestamp = api::new_timestamp(); | ||
|
|
||
| static | ||
| void verify_tablet_metadata_persistence(cql_test_env& env, const tablet_metadata& tm) { | ||
| save_tablet_metadata(env.local_db(), tm, next_timestamp++).get(); | ||
| auto tm2 = read_tablet_metadata(env.local_qp()).get0(); | ||
| BOOST_REQUIRE_EQUAL(tm, tm2); | ||
| } | ||
|
|
||
| static | ||
| cql_test_config tablet_cql_test_config() { | ||
| cql_test_config c; | ||
| c.db_config->experimental_features({ | ||
| db::experimental_features_t::feature::TABLETS, | ||
| }, db::config::config_source::CommandLine); | ||
| c.db_config->consistent_cluster_management(true); | ||
| return c; | ||
| } | ||
|
|
||
| static | ||
| future<table_id> add_table(cql_test_env& e) { | ||
| auto id = table_id(utils::UUID_gen::get_time_UUID()); | ||
| co_await e.create_table([id] (std::string_view ks_name) { | ||
| return *schema_builder(ks_name, id.to_sstring(), id) | ||
| .with_column("p1", utf8_type, column_kind::partition_key) | ||
| .with_column("r1", int32_type) | ||
| .build(); | ||
| }); | ||
| co_return id; | ||
| } | ||
|
|
||
| SEASTAR_TEST_CASE(test_tablet_metadata_persistence) { | ||
| return do_with_cql_env_thread([] (cql_test_env& e) { | ||
| auto h1 = host_id(utils::UUID_gen::get_time_UUID()); | ||
| auto h2 = host_id(utils::UUID_gen::get_time_UUID()); | ||
| auto h3 = host_id(utils::UUID_gen::get_time_UUID()); | ||
|
|
||
| auto table1 = add_table(e).get0(); | ||
| auto table2 = add_table(e).get0(); | ||
|
|
||
| { | ||
| tablet_metadata tm; | ||
|
|
||
| // Empty | ||
| verify_tablet_metadata_persistence(e, tm); | ||
|
|
||
| // Add table1 | ||
| { | ||
| tablet_map tmap(1); | ||
| tmap.set_tablet(tmap.first_tablet(), tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h1, 0}, | ||
| tablet_replica {h2, 3}, | ||
| tablet_replica {h3, 1}, | ||
| } | ||
| }); | ||
| tm.set_tablet_map(table1, std::move(tmap)); | ||
| } | ||
|
|
||
| verify_tablet_metadata_persistence(e, tm); | ||
|
|
||
| // Add table2 | ||
| { | ||
| tablet_map tmap(4); | ||
| auto tb = tmap.first_tablet(); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h1, 0}, | ||
| } | ||
| }); | ||
| tb = *tmap.next_tablet(tb); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h3, 3}, | ||
| } | ||
| }); | ||
| tb = *tmap.next_tablet(tb); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h2, 2}, | ||
| } | ||
| }); | ||
| tb = *tmap.next_tablet(tb); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h1, 1}, | ||
| } | ||
| }); | ||
| tm.set_tablet_map(table2, std::move(tmap)); | ||
| } | ||
|
|
||
| verify_tablet_metadata_persistence(e, tm); | ||
|
|
||
| // Increase RF of table2 | ||
| { | ||
| auto&& tmap = tm.get_tablet_map(table2); | ||
| auto tb = tmap.first_tablet(); | ||
| tb = *tmap.next_tablet(tb); | ||
|
|
||
| tmap.set_tablet_transition_info(tb, tablet_transition_info{ | ||
| tablet_replica_set { | ||
| tablet_replica {h3, 3}, | ||
| tablet_replica {h1, 7}, | ||
| }, | ||
| tablet_replica {h1, 7} | ||
| }); | ||
|
|
||
| tb = *tmap.next_tablet(tb); | ||
| tmap.set_tablet_transition_info(tb, tablet_transition_info{ | ||
| tablet_replica_set { | ||
| tablet_replica {h1, 4}, | ||
| tablet_replica {h2, 2}, | ||
| }, | ||
| tablet_replica {h1, 4} | ||
| }); | ||
| } | ||
|
|
||
| verify_tablet_metadata_persistence(e, tm); | ||
|
|
||
| // Reduce tablet count in table2 | ||
| { | ||
| tablet_map tmap(2); | ||
| auto tb = tmap.first_tablet(); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h1, 0}, | ||
| } | ||
| }); | ||
| tb = *tmap.next_tablet(tb); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h3, 3}, | ||
| } | ||
| }); | ||
| tm.set_tablet_map(table2, std::move(tmap)); | ||
| } | ||
|
|
||
| verify_tablet_metadata_persistence(e, tm); | ||
|
|
||
| // Reduce RF for table1, increasing tablet count | ||
| { | ||
| tablet_map tmap(2); | ||
| auto tb = tmap.first_tablet(); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h3, 7}, | ||
| } | ||
| }); | ||
| tb = *tmap.next_tablet(tb); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h1, 3}, | ||
| } | ||
| }); | ||
| tm.set_tablet_map(table1, std::move(tmap)); | ||
| } | ||
|
|
||
| verify_tablet_metadata_persistence(e, tm); | ||
|
|
||
| // Reduce tablet count for table1 | ||
| { | ||
| tablet_map tmap(1); | ||
| auto tb = tmap.first_tablet(); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h1, 3}, | ||
| } | ||
| }); | ||
| tm.set_tablet_map(table1, std::move(tmap)); | ||
| } | ||
|
|
||
| verify_tablet_metadata_persistence(e, tm); | ||
|
|
||
| // Change replica of table1 | ||
| { | ||
| tablet_map tmap(1); | ||
| auto tb = tmap.first_tablet(); | ||
| tmap.set_tablet(tb, tablet_info { | ||
| tablet_replica_set { | ||
| tablet_replica {h3, 7}, | ||
| } | ||
| }); | ||
| tm.set_tablet_map(table1, std::move(tmap)); | ||
| } | ||
|
|
||
| verify_tablet_metadata_persistence(e, tm); | ||
| } | ||
| }, tablet_cql_test_config()); | ||
| } | ||
|
|
||
| SEASTAR_TEST_CASE(test_large_tablet_metadata) { | ||
| return do_with_cql_env_thread([] (cql_test_env& e) { | ||
| tablet_metadata tm; | ||
|
|
||
| auto h1 = host_id(utils::UUID_gen::get_time_UUID()); | ||
| auto h2 = host_id(utils::UUID_gen::get_time_UUID()); | ||
| auto h3 = host_id(utils::UUID_gen::get_time_UUID()); | ||
|
|
||
| const int nr_tables = 1'00; | ||
| const int tablets_per_table = 1024; | ||
|
|
||
| for (int i = 0; i < nr_tables; ++i) { | ||
| tablet_map tmap(tablets_per_table); | ||
|
|
||
| for (tablet_id j : tmap.tablet_ids()) { | ||
| tmap.set_tablet(j, tablet_info { | ||
| tablet_replica_set {{h1, 0}, {h2, 1}, {h3, 2},} | ||
| }); | ||
| } | ||
|
|
||
| auto id = add_table(e).get0(); | ||
| tm.set_tablet_map(id, std::move(tmap)); | ||
| } | ||
|
|
||
| verify_tablet_metadata_persistence(e, tm); | ||
| }, tablet_cql_test_config()); | ||
| } | ||
|
|
||
| SEASTAR_THREAD_TEST_CASE(test_token_ownership_splitting) { | ||
| const auto real_min_token = dht::token(dht::token_kind::key, std::numeric_limits<int64_t>::min() + 1); | ||
| const auto real_max_token = dht::token(dht::token_kind::key, std::numeric_limits<int64_t>::max()); | ||
|
|
||
| for (auto&& tmap : { | ||
| tablet_map(1), | ||
| tablet_map(2), | ||
| tablet_map(4), | ||
| tablet_map(16), | ||
| tablet_map(1024), | ||
| }) { | ||
| testlog.debug("tmap: {}", tmap); | ||
|
|
||
| BOOST_REQUIRE_EQUAL(real_min_token, tmap.get_first_token(tmap.first_tablet())); | ||
| BOOST_REQUIRE_EQUAL(real_max_token, tmap.get_last_token(tmap.last_tablet())); | ||
|
|
||
| std::optional<tablet_id> prev_tb; | ||
| for (tablet_id tb : tmap.tablet_ids()) { | ||
| testlog.debug("first: {}, last: {}", tmap.get_first_token(tb), tmap.get_last_token(tb)); | ||
| BOOST_REQUIRE_EQUAL(tb, tmap.get_tablet_id(tmap.get_first_token(tb))); | ||
| BOOST_REQUIRE_EQUAL(tb, tmap.get_tablet_id(tmap.get_last_token(tb))); | ||
| if (prev_tb) { | ||
| BOOST_REQUIRE_EQUAL(dht::next_token(tmap.get_last_token(*prev_tb)), tmap.get_first_token(tb)); | ||
| } | ||
| prev_tb = tb; | ||
| } | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,176 @@ | ||
| /* | ||
| * Copyright (C) 2023-present ScyllaDB | ||
| */ | ||
|
|
||
| /* | ||
| * SPDX-License-Identifier: AGPL-3.0-or-later | ||
| */ | ||
|
|
||
| #include <seastar/core/distributed.hh> | ||
| #include <seastar/core/app-template.hh> | ||
| #include <seastar/core/sstring.hh> | ||
| #include <seastar/core/thread.hh> | ||
| #include <seastar/core/reactor.hh> | ||
|
|
||
| #include "locator/tablets.hh" | ||
| #include "replica/tablets.hh" | ||
| #include "locator/tablet_replication_strategy.hh" | ||
| #include "db/config.hh" | ||
| #include "schema/schema_builder.hh" | ||
| #include "service/storage_proxy.hh" | ||
| #include "db/system_keyspace.hh" | ||
|
|
||
| #include "test/perf/perf.hh" | ||
| #include "test/lib/log.hh" | ||
| #include "test/lib/cql_test_env.hh" | ||
|
|
||
| using namespace locator; | ||
| using namespace replica; | ||
|
|
||
| seastar::abort_source aborted; | ||
|
|
||
| static const size_t MiB = 1 << 20; | ||
|
|
||
| static | ||
| cql_test_config tablet_cql_test_config() { | ||
| cql_test_config c; | ||
| c.db_config->experimental_features({ | ||
| db::experimental_features_t::feature::TABLETS, | ||
| db::experimental_features_t::feature::RAFT | ||
| }, db::config::config_source::CommandLine); | ||
| return c; | ||
| } | ||
|
|
||
| static | ||
| future<table_id> add_table(cql_test_env& e) { | ||
| auto id = table_id(utils::UUID_gen::get_time_UUID()); | ||
| co_await e.create_table([id] (std::string_view ks_name) { | ||
| return *schema_builder(ks_name, id.to_sstring(), id) | ||
| .with_column("p1", utf8_type, column_kind::partition_key) | ||
| .with_column("r1", int32_type) | ||
| .build(); | ||
| }); | ||
| co_return id; | ||
| } | ||
|
|
||
| static future<> test_basic_operations(app_template& app) { | ||
| return do_with_cql_env_thread([&] (cql_test_env& e) { | ||
| tablet_metadata tm; | ||
|
|
||
| auto h1 = host_id(utils::UUID_gen::get_time_UUID()); | ||
|
|
||
| int nr_tables = app.configuration()["tables"].as<int>(); | ||
| int tablets_per_table = app.configuration()["tablets-per-table"].as<int>(); | ||
| int rf = app.configuration()["rf"].as<int>(); | ||
|
|
||
| size_t total_tablets = 0; | ||
|
|
||
| std::vector<table_id> ids; | ||
| ids.resize(nr_tables); | ||
| for (int i = 0; i < nr_tables; ++i) { | ||
| ids[i] = add_table(e).get0(); | ||
| } | ||
|
|
||
| testlog.info("Generating tablet metadata"); | ||
|
|
||
| for (int i = 0; i < nr_tables; ++i) { | ||
| tablet_map tmap(tablets_per_table); | ||
|
|
||
| for (tablet_id j : tmap.tablet_ids()) { | ||
| aborted.check(); | ||
| thread::maybe_yield(); | ||
| tablet_replica_set replicas; | ||
| for (int k = 0; k < rf; ++k) { | ||
| replicas.push_back({h1, 0}); | ||
| } | ||
| assert(replicas.size() == rf); | ||
| tmap.set_tablet(j, tablet_info{std::move(replicas)}); | ||
| ++total_tablets; | ||
| } | ||
|
|
||
| tm.set_tablet_map(ids[i], std::move(tmap)); | ||
| } | ||
|
|
||
| testlog.info("Total tablet count: {}", total_tablets); | ||
|
|
||
| testlog.info("Size of tablet_metadata in memory: {} KiB", | ||
| (tm.external_memory_usage() + sizeof(tablet_metadata)) / 1024); | ||
|
|
||
| tablet_metadata tm2; | ||
| auto time_to_copy = duration_in_seconds([&] { | ||
| tm2 = tm; | ||
| }); | ||
|
|
||
| testlog.info("Copied in {:.6f} [ms]", time_to_copy.count() * 1000); | ||
|
|
||
| auto time_to_clear = duration_in_seconds([&] { | ||
| tm2.clear_gently().get(); | ||
| }); | ||
|
|
||
| testlog.info("Cleared in {:.6f} [ms]", time_to_clear.count() * 1000); | ||
|
|
||
| auto time_to_save = duration_in_seconds([&] { | ||
| save_tablet_metadata(e.local_db(), tm, api::new_timestamp()).get(); | ||
| }); | ||
|
|
||
| testlog.info("Saved in {:.6f} [ms]", time_to_save.count() * 1000); | ||
|
|
||
| auto time_to_read = duration_in_seconds([&] { | ||
| tm2 = read_tablet_metadata(e.local_qp()).get0(); | ||
| }); | ||
| assert(tm == tm2); | ||
|
|
||
| testlog.info("Read in {:.6f} [ms]", time_to_read.count() * 1000); | ||
|
|
||
| std::vector<canonical_mutation> muts; | ||
| auto time_to_read_muts = duration_in_seconds([&] { | ||
| muts = replica::read_tablet_mutations(e.local_qp().proxy().container()).get0(); | ||
| }); | ||
|
|
||
| testlog.info("Read mutations in {:.6f} [ms]", time_to_read_muts.count() * 1000); | ||
|
|
||
| auto cm_size = 0; | ||
| for (auto&& cm : muts) { | ||
| cm_size += cm.representation().size(); | ||
| } | ||
|
|
||
| testlog.info("Size of canonical mutations: {:.6f} [MiB]", double(cm_size) / MiB); | ||
|
|
||
| auto&& tablets_table = e.local_db().find_column_family(db::system_keyspace::tablets()); | ||
| testlog.info("Disk space used by system.tablets: {:.6f} [MiB]", double(tablets_table.get_stats().live_disk_space_used) / MiB); | ||
| }, tablet_cql_test_config()); | ||
| } | ||
|
|
||
| namespace perf { | ||
|
|
||
| int scylla_tablets_main(int argc, char** argv) { | ||
| namespace bpo = boost::program_options; | ||
| app_template app; | ||
| app.add_options() | ||
| ("tables", bpo::value<int>()->default_value(100), "Number of tables to create.") | ||
| ("tablets-per-table", bpo::value<int>()->default_value(10000), "Number of tablets per table.") | ||
| ("rf", bpo::value<int>()->default_value(3), "Number of replicas per tablet.") | ||
| ("verbose", "Enables standard logging") | ||
| ; | ||
| return app.run(argc, argv, [&] { | ||
| return seastar::async([&] { | ||
| if (!app.configuration().contains("verbose")) { | ||
| auto testlog_level = logging::logger_registry().get_logger_level("testlog"); | ||
| logging::logger_registry().set_all_loggers_level(seastar::log_level::warn); | ||
| logging::logger_registry().set_logger_level("testlog", testlog_level); | ||
| } | ||
| engine().at_exit([] { | ||
| aborted.request_abort(); | ||
| return make_ready_future(); | ||
| }); | ||
| logalloc::prime_segment_pool(memory::stats().total_memory(), memory::min_free_memory()).get(); | ||
| try { | ||
| test_basic_operations(app).get(); | ||
| } catch (seastar::abort_requested_exception&) { | ||
| // Ignore | ||
| } | ||
| }); | ||
| }); | ||
| } | ||
|
|
||
| } // namespace perf |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| # | ||
| # Copyright (C) 2023-present ScyllaDB | ||
| # | ||
| # SPDX-License-Identifier: AGPL-3.0-or-later | ||
| # | ||
|
|
||
| from test.pylib.manager_client import ManagerClient | ||
| from test.pylib.rest_client import inject_error_one_shot | ||
| import pytest | ||
| import asyncio | ||
| import logging | ||
|
|
||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| async def inject_error_one_shot_on(manager, error_name, servers): | ||
| errs = [inject_error_one_shot(manager.api, s.ip_addr, error_name) for s in servers] | ||
| await asyncio.gather(*errs) | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_tablet_metadata_propagates_with_schema_changes_in_snapshot_mode(manager: ManagerClient): | ||
| """Test that you can create a table and insert and query data""" | ||
|
|
||
| servers = await manager.running_servers() | ||
|
|
||
| s0 = servers[0].server_id | ||
| not_s0 = servers[1:] | ||
|
|
||
| # s0 should miss schema and tablet changes | ||
| await manager.server_stop_gracefully(s0) | ||
|
|
||
| manager.cql.execute("CREATE KEYSPACE test WITH replication = {'class': 'NetworkTopologyStrategy', " | ||
| "'replication_factor': 3, 'initial_tablets': 100};") | ||
|
|
||
| # force s0 to catch up later from the snapshot and not the raft log | ||
| await inject_error_one_shot_on(manager, 'raft_server_force_snapshot', not_s0) | ||
| manager.cql.execute("CREATE TABLE test.test (pk int PRIMARY KEY, c int);") | ||
|
|
||
| keys = range(10) | ||
| await asyncio.gather(*[manager.cql.run_async(f"INSERT INTO test.test (pk, c) VALUES ({k}, 1);") for k in keys]) | ||
|
|
||
| rows = manager.cql.execute("SELECT * FROM test.test;") | ||
| assert len(list(rows)) == len(keys) | ||
| for r in rows: | ||
| assert r.c == 1 | ||
|
|
||
| await manager.server_start(s0, wait_others=2) | ||
|
|
||
| manager.driver_close() | ||
| await manager.driver_connect(server=servers[0]) | ||
|
|
||
| # Trigger a schema change to invoke schema agreement waiting to make sure that s0 has the latest schema | ||
| manager.cql.execute("CREATE KEYSPACE test_dummy WITH replication = {'class': 'NetworkTopologyStrategy', " | ||
| "'replication_factor': 1, 'initial_tablets': 1};") | ||
|
|
||
| await asyncio.gather(*[manager.cql.run_async(f"INSERT INTO test.test (pk, c) VALUES ({k}, 2);", execution_profile='whitelist') | ||
| for k in keys]) | ||
|
|
||
| rows = manager.cql.execute("SELECT * FROM test.test;") | ||
| assert len(list(rows)) == len(keys) | ||
| for r in rows: | ||
| assert r.c == 2 | ||
|
|
||
| # Check that after rolling restart the tablet metadata is still there | ||
| for s in servers: | ||
| manager.server_restart(s, wait_others=2) | ||
|
|
||
| await asyncio.gather(*[manager.cql.run_async(f"INSERT INTO test.test (pk, c) VALUES ({k}, 3);", execution_profile='whitelist') | ||
| for k in keys]) | ||
|
|
||
| rows = manager.cql.execute("SELECT * FROM test.test;") | ||
| assert len(list(rows)) == len(keys) | ||
| for r in rows: | ||
| assert r.c == 3 | ||
|
|
||
| manager.cql.execute("DROP KEYSPACE test;") | ||
| manager.cql.execute("DROP KEYSPACE test_dummy;") | ||
|
|
||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_scans(manager: ManagerClient): | ||
| manager.cql.execute("CREATE KEYSPACE test WITH replication = {'class': 'NetworkTopologyStrategy', " | ||
| "'replication_factor': 1, 'initial_tablets': 8};") | ||
| manager.cql.execute("CREATE TABLE test.test (pk int PRIMARY KEY, c int);") | ||
|
|
||
| keys = range(100) | ||
| await asyncio.gather(*[manager.cql.run_async(f"INSERT INTO test.test (pk, c) VALUES ({k}, {k});") for k in keys]) | ||
|
|
||
| rows = manager.cql.execute("SELECT count(*) FROM test.test;") | ||
| assert rows.one().count == len(keys) | ||
|
|
||
| rows = manager.cql.execute("SELECT * FROM test.test;") | ||
| assert len(list(rows)) == len(keys) | ||
| for r in rows: | ||
| assert r.c == r.pk | ||
|
|
||
| manager.cql.execute("DROP KEYSPACE test;") |