mirrors/scylladb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
copilot/fix-host-variable-error
scylladb/test/boost/database_test.cc
copilot-swe-agent[bot] 77ee7f3417 Revert "Merge 'Add option to use sstable identifier in snapshot' from Benny Halevy" 
This reverts commit 8192f45e84.

The merge exposed a bug where truncate (via drop) fails and causes Raft
errors, leading to schema inconsistencies across nodes. This results in
test_table_drop_with_auto_snapshot failures with 'Keyspace test does not exist'
errors.

The specific problematic change was in commit 19b6207f which modified
truncate_table_on_all_shards to set use_sstable_identifier = true. This
causes exceptions during truncate that are not properly handled, leading
to Raft applier fiber stopping and nodes losing schema synchronization.
2025-12-12 03:55:13 +00:00

1962 lines
95 KiB
C++
Raw Permalink Blame History

/*
* Copyright (C) 2016-present ScyllaDB
*/
/*
* SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
*/
#include <seastar/core/seastar.hh>
#include <seastar/core/smp.hh>
#include <seastar/core/thread.hh>
#include <seastar/core/coroutine.hh>
#include <seastar/util/file.hh>
#undef SEASTAR_TESTING_MAIN
#include <seastar/testing/test_case.hh>
#include <seastar/testing/thread_test_case.hh>
#include <utility>
#include <fmt/ranges.h>
#include <fmt/std.h>
#include "test/lib/cql_test_env.hh"
#include "test/lib/result_set_assertions.hh"
#include "test/lib/log.hh"
#include "test/lib/random_utils.hh"
#include "test/lib/simple_schema.hh"
#include "test/lib/test_utils.hh"
#include "test/lib/key_utils.hh"
#include "replica/database.hh"
#include "utils/assert.hh"
#include "utils/lister.hh"
#include "partition_slice_builder.hh"
#include "mutation/frozen_mutation.hh"
#include "test/lib/mutation_source_test.hh"
#include "schema/schema_builder.hh"
#include "service/migration_manager.hh"
#include "sstables/sstables.hh"
#include "sstables/generation_type.hh"
#include "db/config.hh"
#include "db/commitlog/commitlog_replayer.hh"
#include "db/commitlog/commitlog.hh"
#include "test/lib/tmpdir.hh"
#include "db/data_listeners.hh"
#include "replica/multishard_query.hh"
#include "mutation_query.hh"
#include "transport/messages/result_message.hh"
#include "compaction/compaction_manager.hh"
#include "db/snapshot-ctl.hh"
#include "db/system_keyspace.hh"
#include "db/view/view_builder.hh"
#include "replica/mutation_dump.hh"
using namespace std::chrono_literals;
using namespace sstables;
using namespace tests;
class database_test_wrapper {
replica::database& _db;
public:
explicit database_test_wrapper(replica::database& db) : _db(db) { }
reader_concurrency_semaphore& get_user_read_concurrency_semaphore() {
return _db.read_concurrency_sem();
}
reader_concurrency_semaphore& get_streaming_read_concurrency_semaphore() {
return _db._streaming_concurrency_sem;
}
reader_concurrency_semaphore& get_system_read_concurrency_semaphore() {
return _db._system_read_concurrency_sem;
}
size_t get_total_user_reader_concurrency_semaphore_memory() {
return _db._reader_concurrency_semaphores_group._total_memory;
}
size_t get_total_user_reader_concurrency_semaphore_weight() {
return _db._reader_concurrency_semaphores_group._total_weight;
}
};
static future<> apply_mutation(sharded<replica::database>& sharded_db, table_id uuid, const mutation& m, bool do_flush = false,
db::commitlog::force_sync fs = db::commitlog::force_sync::no, db::timeout_clock::time_point timeout = db::no_timeout) {
auto& t = sharded_db.local().find_column_family(uuid);
auto shard = t.shard_for_reads(m.token());
return sharded_db.invoke_on(shard, [uuid, fm = freeze(m), do_flush, fs, timeout] (replica::database& db) {
auto& t = db.find_column_family(uuid);
return db.apply(t.schema(), fm, tracing::trace_state_ptr(), fs, timeout).then([do_flush, &t] {
return do_flush ? t.flush() : make_ready_future<>();
});
});
}
future<> do_with_cql_env_and_compaction_groups_cgs(unsigned cgs, std::function<void(cql_test_env&)> func, cql_test_config cfg = {}, thread_attributes thread_attr = {}) {
// clean the dir before running
if (cfg.db_config->data_file_directories.is_set() && cfg.clean_data_dir_before_test) {
co_await recursive_remove_directory(fs::path(cfg.db_config->data_file_directories()[0]));
co_await recursive_touch_directory(cfg.db_config->data_file_directories()[0]);
}
// TODO: perhaps map log2_compaction_groups into initial_tablets when creating the testing keyspace.
co_await do_with_cql_env_thread(func, cfg, thread_attr);
}
future<> do_with_cql_env_and_compaction_groups(std::function<void(cql_test_env&)> func, cql_test_config cfg = {}, thread_attributes thread_attr = {}) {
std::vector<unsigned> x_log2_compaction_group_values = { 0 /* 1 CG */ };
for (auto x_log2_compaction_groups : x_log2_compaction_group_values) {
co_await do_with_cql_env_and_compaction_groups_cgs(x_log2_compaction_groups, func, cfg, thread_attr);
}
}
BOOST_AUTO_TEST_SUITE(database_test)
SEASTAR_TEST_CASE(test_safety_after_truncate) {
auto cfg = make_shared<db::config>();
cfg->auto_snapshot.set(false);
return do_with_cql_env_and_compaction_groups([](cql_test_env& e) {
e.execute_cql("create table ks.cf (k text, v int, primary key (k));").get();
auto& db = e.local_db();
sstring ks_name = "ks";
sstring cf_name = "cf";
auto s = db.find_schema(ks_name, cf_name);
auto&& table = db.find_column_family(s);
auto uuid = s->id();
std::vector<size_t> keys_per_shard;
std::vector<dht::partition_range_vector> pranges_per_shard;
keys_per_shard.resize(smp::count);
pranges_per_shard.resize(smp::count);
for (uint32_t i = 1; i <= 1000; ++i) {
auto pkey = partition_key::from_single_value(*s, to_bytes(fmt::format("key{}", i)));
mutation m(s, pkey);
m.set_clustered_cell(clustering_key_prefix::make_empty(), "v", int32_t(42), {});
auto shard = table.shard_for_reads(m.token());
keys_per_shard[shard]++;
pranges_per_shard[shard].emplace_back(dht::partition_range::make_singular(dht::decorate_key(*s, std::move(pkey))));
apply_mutation(e.db(), uuid, m).get();
}
auto assert_query_result = [&] (const std::vector<size_t>& expected_sizes) {
auto max_size = std::numeric_limits<size_t>::max();
auto cmd = query::read_command(s->id(), s->version(), partition_slice_builder(*s).build(), query::max_result_size(max_size),
query::tombstone_limit::max, query::row_limit(1000));
e.db().invoke_on_all([&] (replica::database& db) -> future<> {
auto shard = this_shard_id();
auto s = db.find_schema(uuid);
auto&& [result, cache_tempature] = co_await db.query(s, cmd, query::result_options::only_result(), pranges_per_shard[shard], nullptr, db::no_timeout);
assert_that(query::result_set::from_raw_result(s, cmd.slice, *result)).has_size(expected_sizes[shard]);
}).get();
};
assert_query_result(keys_per_shard);
replica::database::truncate_table_on_all_shards(e.db(), e.get_system_keyspace(), "ks", "cf").get();
for (auto it = keys_per_shard.begin(); it < keys_per_shard.end(); ++it) {
*it = 0;
}
assert_query_result(keys_per_shard);
e.db().invoke_on_all([&] (replica::database& db) -> future<> {
auto cl = db.commitlog();
auto rp = co_await db::commitlog_replayer::create_replayer(e.db(), e.get_system_keyspace());
auto paths = co_await cl->list_existing_segments();
co_await rp.recover(paths, db::commitlog::descriptor::FILENAME_PREFIX);
}).get();
assert_query_result(keys_per_shard);
return make_ready_future<>();
}, cfg);
}
// Reproducer for:
// https://github.com/scylladb/scylla/issues/10421
// https://github.com/scylladb/scylla/issues/10423
SEASTAR_TEST_CASE(test_truncate_without_snapshot_during_writes) {
auto cfg = make_shared<db::config>();
cfg->auto_snapshot.set(false);
return do_with_cql_env_and_compaction_groups([] (cql_test_env& e) {
sstring ks_name = "ks";
sstring cf_name = "cf";
e.execute_cql(fmt::format("create table {}.{} (k text, v int, primary key (k));", ks_name, cf_name)).get();
auto& db = e.local_db();
auto uuid = db.find_uuid(ks_name, cf_name);
auto s = db.find_column_family(uuid).schema();
int count = 0;
auto insert_data = [&] (uint32_t begin, uint32_t end) {
return parallel_for_each(std::views::iota(begin, end), [&] (auto i) {
auto pkey = partition_key::from_single_value(*s, to_bytes(fmt::format("key-{}", tests::random::get_int<uint64_t>())));
mutation m(s, pkey);
m.set_clustered_cell(clustering_key_prefix::make_empty(), "v", int32_t(42), {});
return apply_mutation(e.db(), uuid, m, true /* do_flush */).finally([&] {
++count;
});
});
};
uint32_t num_keys = 1000;
auto f0 = insert_data(0, num_keys);
auto f1 = do_until([&] { return std::cmp_greater_equal(count, num_keys); }, [&, ts = db_clock::now()] {
return replica::database::truncate_table_on_all_shards(e.db(), e.get_system_keyspace(), "ks", "cf", ts, false /* with_snapshot */).then([] {
return yield();
});
});
f0.get();
f1.get();
}, cfg);
}
// Reproducer for:
// https://github.com/scylladb/scylla/issues/21719
SEASTAR_TEST_CASE(test_truncate_saves_replay_position) {
auto cfg = make_shared<db::config>();
cfg->auto_snapshot.set(false);
return do_with_cql_env_and_compaction_groups([] (cql_test_env& e) {
BOOST_REQUIRE_GT(smp::count, 1);
const sstring ks_name = "ks";
const sstring cf_name = "cf";
e.execute_cql(fmt::format("CREATE TABLE {}.{} (k TEXT PRIMARY KEY, v INT);", ks_name, cf_name)).get();
const table_id uuid = e.local_db().find_uuid(ks_name, cf_name);
replica::database::truncate_table_on_all_shards(e.db(), e.get_system_keyspace(), ks_name, cf_name, db_clock::now(), false /* with_snapshot */).get();
auto res = e.execute_cql(fmt::format("SELECT * FROM system.truncated WHERE table_uuid = {}", uuid)).get();
auto rows = dynamic_pointer_cast<cql_transport::messages::result_message::rows>(res);
BOOST_REQUIRE(rows);
auto row_count = rows->rs().result_set().size();
BOOST_REQUIRE_EQUAL(row_count, smp::count);
}, cfg);
}
SEASTAR_TEST_CASE(test_querying_with_limits) {
return do_with_cql_env_and_compaction_groups([](cql_test_env& e) {
// FIXME: restore indent.
e.execute_cql("create table ks.cf (k text, v int, primary key (k));").get();
auto& db = e.local_db();
auto s = db.find_schema("ks", "cf");
auto&& table = db.find_column_family(s);
auto uuid = s->id();
std::vector<size_t> keys_per_shard;
std::vector<dht::partition_range_vector> pranges_per_shard;
keys_per_shard.resize(smp::count);
pranges_per_shard.resize(smp::count);
for (uint32_t i = 1; i <= 3 * smp::count; ++i) {
auto pkey = partition_key::from_single_value(*s, to_bytes(format("key{:d}", i)));
mutation m(s, pkey);
m.partition().apply(tombstone(api::timestamp_type(1), gc_clock::now()));
apply_mutation(e.db(), uuid, m).get();
auto shard = table.shard_for_reads(m.token());
pranges_per_shard[shard].emplace_back(dht::partition_range::make_singular(dht::decorate_key(*s, std::move(pkey))));
}
for (uint32_t i = 3 * smp::count; i <= 8 * smp::count; ++i) {
auto pkey = partition_key::from_single_value(*s, to_bytes(format("key{:d}", i)));
mutation m(s, pkey);
m.set_clustered_cell(clustering_key_prefix::make_empty(), "v", int32_t(42), 1);
apply_mutation(e.db(), uuid, m).get();
auto shard = table.shard_for_reads(m.token());
keys_per_shard[shard]++;
pranges_per_shard[shard].emplace_back(dht::partition_range::make_singular(dht::decorate_key(*s, std::move(pkey))));
}
auto max_size = std::numeric_limits<size_t>::max();
{
auto cmd = query::read_command(s->id(), s->version(), partition_slice_builder(*s).build(), query::max_result_size(max_size),
query::tombstone_limit::max, query::row_limit(3));
e.db().invoke_on_all([&] (replica::database& db) -> future<> {
auto shard = this_shard_id();
auto s = db.find_schema(uuid);
auto result = std::get<0>(co_await db.query(s, cmd, query::result_options::only_result(), pranges_per_shard[shard], nullptr, db::no_timeout));
auto expected_size = std::min<size_t>(keys_per_shard[shard], 3);
assert_that(query::result_set::from_raw_result(s, cmd.slice, *result)).has_size(expected_size);
}).get();
}
{
auto cmd = query::read_command(s->id(), s->version(), partition_slice_builder(*s).build(), query::max_result_size(max_size),
query::tombstone_limit::max, query::row_limit(query::max_rows), query::partition_limit(5));
e.db().invoke_on_all([&] (replica::database& db) -> future<> {
auto shard = this_shard_id();
auto s = db.find_schema(uuid);
auto result = std::get<0>(co_await db.query(s, cmd, query::result_options::only_result(), pranges_per_shard[shard], nullptr, db::no_timeout));
auto expected_size = std::min<size_t>(keys_per_shard[shard], 5);
assert_that(query::result_set::from_raw_result(s, cmd.slice, *result)).has_size(expected_size);
}).get();
}
{
auto cmd = query::read_command(s->id(), s->version(), partition_slice_builder(*s).build(), query::max_result_size(max_size),
query::tombstone_limit::max, query::row_limit(query::max_rows), query::partition_limit(3));
e.db().invoke_on_all([&] (replica::database& db) -> future<> {
auto shard = this_shard_id();
auto s = db.find_schema(uuid);
auto result = std::get<0>(co_await db.query(s, cmd, query::result_options::only_result(), pranges_per_shard[shard], nullptr, db::no_timeout));
auto expected_size = std::min<size_t>(keys_per_shard[shard], 3);
assert_that(query::result_set::from_raw_result(s, cmd.slice, *result)).has_size(expected_size);
}).get();
}
});
}
static void test_database(void (*run_tests)(populate_fn_ex, bool), unsigned cgs) {
do_with_cql_env_and_compaction_groups_cgs(cgs, [run_tests] (cql_test_env& e) {
run_tests([&] (schema_ptr s, const utils::chunked_vector<mutation>& partitions, gc_clock::time_point) -> mutation_source {
auto& mm = e.migration_manager().local();
try {
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(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(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) {
apply_mutation(e.db(), uuid, m).get();
}
cf.flush().get();
cf.get_row_cache().invalidate(row_cache::external_updater([] {})).get();
return mutation_source([&] (schema_ptr s,
reader_permit permit,
const dht::partition_range& range,
const query::partition_slice& slice,
tracing::trace_state_ptr trace_state,
streamed_mutation::forwarding fwd,
mutation_reader::forwarding fwd_mr) {
return cf.make_mutation_reader(s, std::move(permit), range, slice, std::move(trace_state), fwd, fwd_mr);
});
}, true);
}).get();
}
// plain cg0
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_basic_cg0) {
test_database(run_mutation_source_tests_plain_basic, 0);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_reader_conversion_cg0) {
test_database(run_mutation_source_tests_plain_reader_conversion, 0);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_fragments_monotonic_cg0) {
test_database(run_mutation_source_tests_plain_fragments_monotonic, 0);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_read_back_cg0) {
test_database(run_mutation_source_tests_plain_read_back, 0);
}
// plain cg1
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_basic_cg1) {
test_database(run_mutation_source_tests_plain_basic, 1);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_reader_conversion_cg1) {
test_database(run_mutation_source_tests_plain_reader_conversion, 1);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_fragments_monotonic_cg1) {
test_database(run_mutation_source_tests_plain_fragments_monotonic, 1);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_read_back_cg1) {
test_database(run_mutation_source_tests_plain_read_back, 1);
}
// reverse cg0
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_basic_cg0) {
test_database(run_mutation_source_tests_reverse_basic, 0);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_reader_conversion_cg0) {
test_database(run_mutation_source_tests_reverse_reader_conversion, 0);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_fragments_monotonic_cg0) {
test_database(run_mutation_source_tests_reverse_fragments_monotonic, 0);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_read_back_cg0) {
test_database(run_mutation_source_tests_reverse_read_back, 0);
}
// reverse cg1
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_basic_cg1) {
test_database(run_mutation_source_tests_reverse_basic, 1);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_reader_conversion_cg1) {
test_database(run_mutation_source_tests_reverse_reader_conversion, 1);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_fragments_monotonic_cg1) {
test_database(run_mutation_source_tests_reverse_fragments_monotonic, 1);
}
SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_read_back_cg1) {
test_database(run_mutation_source_tests_reverse_read_back, 1);
}
static void require_exist(const sstring& filename, bool should) {
auto exists = file_exists(filename).get();
BOOST_REQUIRE_EQUAL(exists, should);
}
static void touch_dir(const sstring& dirname) {
recursive_touch_directory(dirname).get();
require_exist(dirname, true);
}
static void touch_file(const sstring& filename) {
tests::touch_file(filename).get();
require_exist(filename, true);
}
SEASTAR_THREAD_TEST_CASE(test_distributed_loader_with_incomplete_sstables) {
using sst = sstables::sstable;
tmpdir data_dir;
auto db_cfg_ptr = make_shared<db::config>();
auto& db_cfg = *db_cfg_ptr;
db_cfg.data_file_directories({data_dir.path().string()}, db::config::config_source::CommandLine);
// Create incomplete sstables in test data directory
sstring ks = "system";
sstring cf = "peers-37f71aca7dc2383ba70672528af04d4f";
sstring sst_dir = (data_dir.path() / std::string_view(ks) / std::string_view(cf)).string();
auto temp_sst_dir_2 = fmt::format("{}/{}{}", sst_dir, generation_from_value(2), tempdir_extension);
touch_dir(temp_sst_dir_2);
auto temp_sst_dir_3 = fmt::format("{}/{}{}", sst_dir, generation_from_value(3), tempdir_extension);
touch_dir(temp_sst_dir_3);
auto temp_file_name = sst::filename(temp_sst_dir_3, ks, cf, sstables::get_highest_sstable_version(), generation_from_value(3), sst::format_types::big, component_type::TemporaryTOC);
touch_file(temp_file_name);
temp_file_name = sst::filename(sst_dir, ks, cf, sstables::get_highest_sstable_version(), generation_from_value(4), sst::format_types::big, component_type::TemporaryTOC);
touch_file(temp_file_name);
// Reproducer for #scylladb/scylladb#26393
temp_file_name = sst::filename(sst_dir, ks, cf, sstables::get_highest_sstable_version(), generation_from_value(4), sst::format_types::big, component_type::TemporaryHashes);
touch_file(temp_file_name);
temp_file_name = sst::filename(sst_dir, ks, cf, sstables::get_highest_sstable_version(), generation_from_value(4), sst::format_types::big, component_type::Data);
touch_file(temp_file_name);
auto test_config = cql_test_config(db_cfg_ptr);
test_config.clean_data_dir_before_test = false;
do_with_cql_env_and_compaction_groups([&sst_dir, &ks, &cf, &temp_sst_dir_2, &temp_sst_dir_3] (cql_test_env& e) {
require_exist(temp_sst_dir_2, false);
require_exist(temp_sst_dir_3, false);
require_exist(sst::filename(sst_dir, ks, cf, sstables::get_highest_sstable_version(), generation_from_value(4), sst::format_types::big, component_type::TemporaryTOC), false);
require_exist(sst::filename(sst_dir, ks, cf, sstables::get_highest_sstable_version(), generation_from_value(4), sst::format_types::big, component_type::TemporaryHashes), false);
require_exist(sst::filename(sst_dir, ks, cf, sstables::get_highest_sstable_version(), generation_from_value(4), sst::format_types::big, component_type::Data), false);
}, test_config).get();
}
SEASTAR_THREAD_TEST_CASE(test_distributed_loader_with_pending_delete) {
using sst = sstables::sstable;
tmpdir data_dir;
auto db_cfg_ptr = make_shared<db::config>();
auto& db_cfg = *db_cfg_ptr;
db_cfg.data_file_directories({data_dir.path().string()}, db::config::config_source::CommandLine);
// Create incomplete sstables in test data directory
sstring ks = "system";
sstring cf = "peers-37f71aca7dc2383ba70672528af04d4f";
sstring sst_dir = (data_dir.path() / std::string_view(ks) / std::string_view(cf)).string();
sstring pending_delete_dir = sst_dir + "/" + sstables::pending_delete_dir;
auto write_file = [] (const sstring& file_name, const sstring& text) {
auto f = open_file_dma(file_name, open_flags::wo | open_flags::create | open_flags::truncate).get();
auto os = make_file_output_stream(f, file_output_stream_options{}).get();
os.write(text).get();
os.flush().get();
os.close().get();
require_exist(file_name, true);
};
auto component_basename = [&ks, &cf] (sstables::generation_type gen, component_type ctype) {
return sst::component_basename(ks, cf, sstables::get_highest_sstable_version(), gen, sst::format_types::big, ctype);
};
auto gen_filename = [&sst_dir, &ks, &cf] (sstables::generation_type gen, component_type ctype) {
return sst::filename(sst_dir, ks, cf, sstables::get_highest_sstable_version(), gen, sst::format_types::big, ctype);
};
touch_dir(pending_delete_dir);
// Empty log file
touch_file(pending_delete_dir + "/sstables-0-0.log");
// Empty temporary log file
touch_file(pending_delete_dir + "/sstables-1-1.log.tmp");
const sstring toc_text = "TOC.txt\nData.db\n";
sstables::sstable_generation_generator gen_generator;
std::vector<sstables::generation_type> gen;
constexpr size_t num_gens = 9;
std::generate_n(std::back_inserter(gen), num_gens, [&] {
return gen_generator();
});
// Regular log file with single entry
write_file(gen_filename(gen[2], component_type::TOC), toc_text);
touch_file(gen_filename(gen[2], component_type::Data));
write_file(pending_delete_dir + "/sstables-2-2.log",
component_basename(gen[2], component_type::TOC) + "\n");
// Temporary log file with single entry
write_file(pending_delete_dir + "/sstables-3-3.log.tmp",
component_basename(gen[3], component_type::TOC) + "\n");
// Regular log file with multiple entries
write_file(gen_filename(gen[4], component_type::TOC), toc_text);
touch_file(gen_filename(gen[4], component_type::Data));
write_file(gen_filename(gen[5], component_type::TOC), toc_text);
touch_file(gen_filename(gen[5], component_type::Data));
write_file(pending_delete_dir + "/sstables-4-5.log",
component_basename(gen[4], component_type::TOC) + "\n" +
component_basename(gen[5], component_type::TOC) + "\n");
// Regular log file with multiple entries and some deleted sstables
write_file(gen_filename(gen[6], component_type::TemporaryTOC), toc_text);
touch_file(gen_filename(gen[6], component_type::Data));
write_file(gen_filename(gen[7], component_type::TemporaryTOC), toc_text);
write_file(pending_delete_dir + "/sstables-6-8.log",
component_basename(gen[6], component_type::TOC) + "\n" +
component_basename(gen[7], component_type::TOC) + "\n" +
component_basename(gen[8], component_type::TOC) + "\n");
auto test_config = cql_test_config(db_cfg_ptr);
test_config.clean_data_dir_before_test = false;
do_with_cql_env_and_compaction_groups([&] (cql_test_env& e) {
// Empty log filesst_dir
// Empty temporary log file
require_exist(pending_delete_dir + "/sstables-1-1.log.tmp", false);
// Regular log file with single entry
require_exist(gen_filename(gen[2], component_type::TOC), false);
require_exist(gen_filename(gen[2], component_type::Data), false);
require_exist(pending_delete_dir + "/sstables-2-2.log", false);
// Temporary log file with single entry
require_exist(pending_delete_dir + "/sstables-3-3.log.tmp", false);
// Regular log file with multiple entries
require_exist(gen_filename(gen[4], component_type::TOC), false);
require_exist(gen_filename(gen[4], component_type::Data), false);
require_exist(gen_filename(gen[5], component_type::TOC), false);
require_exist(gen_filename(gen[5], component_type::Data), false);
require_exist(pending_delete_dir + "/sstables-4-5.log", false);
// Regular log file with multiple entries and some deleted sstables
require_exist(gen_filename(gen[6], component_type::TemporaryTOC), false);
require_exist(gen_filename(gen[6], component_type::Data), false);
require_exist(gen_filename(gen[7], component_type::TemporaryTOC), false);
require_exist(pending_delete_dir + "/sstables-6-8.log", false);
}, test_config).get();
}
// Snapshot tests and their helpers
future<> do_with_some_data(std::vector<sstring> cf_names, std::function<future<> (cql_test_env& env)> func, bool create_mvs = false, shared_ptr<db::config> db_cfg_ptr = {}) {
return seastar::async([cf_names = std::move(cf_names), func = std::move(func), create_mvs, db_cfg_ptr = std::move(db_cfg_ptr)] () mutable {
lw_shared_ptr<tmpdir> tmpdir_for_data;
if (!db_cfg_ptr) {
tmpdir_for_data = make_lw_shared<tmpdir>();
db_cfg_ptr = make_shared<db::config>();
db_cfg_ptr->data_file_directories(std::vector<sstring>({ tmpdir_for_data->path().string() }));
}
do_with_cql_env_and_compaction_groups([cf_names = std::move(cf_names), func = std::move(func), create_mvs] (cql_test_env& e) {
for (const auto& cf_name : cf_names) {
e.create_table([&cf_name] (std::string_view ks_name) {
return *schema_builder(ks_name, cf_name)
.with_column("p1", utf8_type, column_kind::partition_key)
.with_column("c1", int32_type, column_kind::clustering_key)
.with_column("c2", int32_type, column_kind::clustering_key)
.with_column("r1", int32_type)
.build();
}).get();
e.execute_cql(fmt::format("insert into {} (p1, c1, c2, r1) values ('key1', 1, 2, 3);", cf_name)).get();
e.execute_cql(fmt::format("insert into {} (p1, c1, c2, r1) values ('key1', 2, 2, 3);", cf_name)).get();
e.execute_cql(fmt::format("insert into {} (p1, c1, c2, r1) values ('key1', 3, 2, 3);", cf_name)).get();
e.execute_cql(fmt::format("insert into {} (p1, c1, c2, r1) values ('key2', 4, 5, 6);", cf_name)).get();
e.execute_cql(fmt::format("insert into {} (p1, c1, c2, r1) values ('key2', 5, 5, 6);", cf_name)).get();
e.execute_cql(fmt::format("insert into {} (p1, c1, c2, r1) values ('key2', 6, 5, 6);", cf_name)).get();
if (create_mvs) {
auto f1 = e.local_view_builder().wait_until_built("ks", seastar::format("view_{}", cf_name));
e.execute_cql(seastar::format("create materialized view view_{0} as select * from {0} where p1 is not null and c1 is not null and c2 is "
"not null primary key (p1, c1, c2)",
cf_name))
.get();
f1.get();
auto f2 = e.local_view_builder().wait_until_built("ks", "index_cf_index");
e.execute_cql(seastar::format("CREATE INDEX index_{0} ON {0} (r1);", cf_name)).get();
f2.get();
}
}
func(e).get();
}, db_cfg_ptr).get();
});
}
future<> take_snapshot(cql_test_env& e, sstring ks_name = "ks", sstring cf_name = "cf", sstring snapshot_name = "test", bool skip_flush = false) {
try {
auto uuid = e.db().local().find_uuid(ks_name, cf_name);
co_await replica::database::snapshot_table_on_all_shards(e.db(), uuid, snapshot_name, skip_flush);
} catch (...) {
testlog.error("Could not take snapshot for {}.{} snapshot_name={} skip_flush={}: {}",
ks_name, cf_name, snapshot_name, skip_flush, std::current_exception());
throw;
}
}
future<std::set<sstring>> collect_files(fs::path path) {
std::set<sstring> ret;
directory_lister lister(path, lister::dir_entry_types::of<directory_entry_type::regular>());
while (auto de = co_await lister.get()) {
ret.insert(de->name);
}
co_return ret;
}
static future<> snapshot_works(const std::string& table_name) {
return do_with_some_data({"cf"}, [table_name] (cql_test_env& e) {
take_snapshot(e, "ks", table_name).get();
auto& cf = e.local_db().find_column_family("ks", table_name);
auto table_directory = table_dir(cf);
auto snapshot_dir = table_directory / sstables::snapshots_dir / "test";
auto in_table_dir = collect_files(table_directory).get();
// snapshot triggered a flush and wrote the data down.
BOOST_REQUIRE_GE(in_table_dir.size(), 9);
auto in_snapshot_dir = collect_files(snapshot_dir).get();
in_table_dir.insert("manifest.json");
in_table_dir.insert("schema.cql");
// all files were copied and manifest was generated
BOOST_REQUIRE_EQUAL(in_table_dir, in_snapshot_dir);
return make_ready_future<>();
}, true);
}
SEASTAR_TEST_CASE(table_snapshot_works) {
return snapshot_works("cf");
}
SEASTAR_TEST_CASE(view_snapshot_works) {
return snapshot_works("view_cf");
}
SEASTAR_TEST_CASE(index_snapshot_works) {
return snapshot_works(::secondary_index::index_table_name("index_cf"));
}
SEASTAR_TEST_CASE(snapshot_skip_flush_works) {
return do_with_some_data({"cf"}, [] (cql_test_env& e) {
take_snapshot(e, "ks", "cf", "test", true /* skip_flush */).get();
auto& cf = e.local_db().find_column_family("ks", "cf");
auto in_table_dir = collect_files(table_dir(cf)).get();
// Snapshot did not trigger a flush.
BOOST_REQUIRE(in_table_dir.empty());
auto in_snapshot_dir = collect_files(table_dir(cf) / sstables::snapshots_dir / "test").get();
BOOST_REQUIRE_EQUAL(in_snapshot_dir, std::set<sstring>({"manifest.json", "schema.cql"}));
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(snapshot_list_okay) {
return do_with_some_data({"cf"}, [] (cql_test_env& e) {
auto& cf = e.local_db().find_column_family("ks", "cf");
take_snapshot(e).get();
auto details = cf.get_snapshot_details().get();
BOOST_REQUIRE_EQUAL(details.size(), 1);
auto sd = details["test"];
BOOST_REQUIRE_EQUAL(sd.live, 0);
BOOST_REQUIRE_GT(sd.total, 0);
auto table_directory = table_dir(cf);
for (auto& f : collect_files(table_directory).get()) {
fs::remove(table_directory / f);
}
auto sd_post_deletion = cf.get_snapshot_details().get().at("test");
BOOST_REQUIRE_EQUAL(sd_post_deletion.total, sd_post_deletion.live);
BOOST_REQUIRE_EQUAL(sd.total, sd_post_deletion.live);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(snapshot_list_contains_dropped_tables) {
return do_with_some_data({"cf1", "cf2", "cf3", "cf4"}, [] (cql_test_env& e) {
e.execute_cql("DROP TABLE ks.cf1;").get();
auto details = e.local_db().get_snapshot_details().get();
BOOST_REQUIRE_EQUAL(details.size(), 1);
BOOST_REQUIRE_EQUAL(details.begin()->second.size(), 1);
const auto& sd = details.begin()->second.front().details;
BOOST_REQUIRE_GT(sd.live, 0);
BOOST_REQUIRE_EQUAL(sd.total, sd.live);
take_snapshot(e, "ks", "cf2", "test2").get();
take_snapshot(e, "ks", "cf3", "test3").get();
details = e.local_db().get_snapshot_details().get();
BOOST_REQUIRE_EQUAL(details.size(), 3);
e.execute_cql("DROP TABLE ks.cf4;").get();
details = e.local_db().get_snapshot_details().get();
BOOST_REQUIRE_EQUAL(details.size(), 4);
for (const auto& [name, r] : details) {
BOOST_REQUIRE_EQUAL(r.size(), 1);
const auto& result = r.front();
const auto& sd = result.details;
if (name == "test2" || name == "test3") {
BOOST_REQUIRE_EQUAL(sd.live, 0);
BOOST_REQUIRE_GT(sd.total, 0);
} else {
BOOST_REQUIRE_GT(sd.live, 0);
BOOST_REQUIRE_EQUAL(sd.total, sd.live);
}
}
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(snapshot_list_inexistent) {
return do_with_some_data({"cf"}, [] (cql_test_env& e) {
auto& cf = e.local_db().find_column_family("ks", "cf");
auto details = cf.get_snapshot_details().get();
BOOST_REQUIRE_EQUAL(details.size(), 0);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(clear_snapshot) {
return do_with_some_data({"cf"}, [] (cql_test_env& e) {
take_snapshot(e).get();
auto& cf = e.local_db().find_column_family("ks", "cf");
unsigned count = collect_files(table_dir(cf) / sstables::snapshots_dir / "test").get().size();
BOOST_REQUIRE_GT(count, 1); // expect more than the manifest alone
e.local_db().clear_snapshot("test", {"ks"}, "").get();
count = 0;
BOOST_REQUIRE_EQUAL(fs::exists(table_dir(cf) / sstables::snapshots_dir / "test"), false);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(clear_multiple_snapshots) {
sstring ks_name = "ks";
sstring table_name = "cf";
auto num_snapshots = 2;
auto snapshot_name = [] (int idx) {
return format("test-snapshot-{}", idx);
};
co_await do_with_some_data({table_name}, [&] (cql_test_env& e) {
auto& t = e.local_db().find_column_family(ks_name, table_name);
auto tdir = table_dir(t);
auto snapshots_dir = tdir / sstables::snapshots_dir;
for (auto i = 0; i < num_snapshots; i++) {
testlog.debug("Taking snapshot {} on {}.{}", snapshot_name(i), ks_name, table_name);
take_snapshot(e, ks_name, table_name, snapshot_name(i)).get();
}
for (auto i = 0; i < num_snapshots; i++) {
testlog.debug("Verifying {}", snapshots_dir / snapshot_name(i));
unsigned count = collect_files(snapshots_dir / snapshot_name(i)).get().size();
BOOST_REQUIRE_GT(count, 1); // expect more than the manifest alone
}
// non-existent tag
testlog.debug("Clearing bogus tag");
e.local_db().clear_snapshot("bogus", {ks_name}, table_name).get();
for (auto i = 0; i < num_snapshots; i++) {
BOOST_REQUIRE_EQUAL(fs::exists(snapshots_dir / snapshot_name(i)), true);
}
// clear single tag
testlog.debug("Clearing snapshot={} of {}.{}", snapshot_name(0), ks_name, table_name);
e.local_db().clear_snapshot(snapshot_name(0), {ks_name}, table_name).get();
BOOST_REQUIRE_EQUAL(fs::exists(snapshots_dir / snapshot_name(0)), false);
for (auto i = 1; i < num_snapshots; i++) {
BOOST_REQUIRE_EQUAL(fs::exists(snapshots_dir / snapshot_name(i)), true);
}
// clear all tags (all tables)
testlog.debug("Clearing all snapshots in {}", ks_name);
e.local_db().clear_snapshot("", {ks_name}, "").get();
for (auto i = 0; i < num_snapshots; i++) {
BOOST_REQUIRE_EQUAL(fs::exists(snapshots_dir / snapshot_name(i)), false);
}
testlog.debug("Taking an extra {} of {}.{}", snapshot_name(num_snapshots), ks_name, table_name);
take_snapshot(e, ks_name, table_name, snapshot_name(num_snapshots)).get();
// existing snapshots expected to remain after dropping the table
testlog.debug("Dropping table {}.{}", ks_name, table_name);
replica::database::legacy_drop_table_on_all_shards(e.db(), e.get_system_keyspace(), ks_name, table_name).get();
BOOST_REQUIRE_EQUAL(fs::exists(snapshots_dir / snapshot_name(num_snapshots)), true);
// clear all tags
testlog.debug("Clearing all snapshots in {}.{} after it had been dropped", ks_name, table_name);
e.local_db().clear_snapshot("", {ks_name}, table_name).get();
SCYLLA_ASSERT(!fs::exists(tdir));
// after all snapshots had been cleared,
// the dropped table directory is expected to be removed.
BOOST_REQUIRE_EQUAL(fs::exists(tdir), false);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(clear_nonexistent_snapshot) {
// no crashes, no exceptions
return do_with_some_data({"cf"}, [] (cql_test_env& e) {
e.local_db().clear_snapshot("test", {"ks"}, "").get();
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(test_snapshot_ctl_details) {
return do_with_some_data({"cf"}, [] (cql_test_env& e) {
sharded<db::snapshot_ctl> sc;
sc.start(std::ref(e.db()), std::ref(e.get_task_manager()), std::ref(e.get_sstorage_manager()), db::snapshot_ctl::config{}).get();
auto stop_sc = deferred_stop(sc);
auto& cf = e.local_db().find_column_family("ks", "cf");
take_snapshot(e).get();
auto details = cf.get_snapshot_details().get();
BOOST_REQUIRE_EQUAL(details.size(), 1);
auto sd = details["test"];
BOOST_REQUIRE_EQUAL(sd.live, 0);
BOOST_REQUIRE_GT(sd.total, 0);
auto sc_details = sc.local().get_snapshot_details().get();
BOOST_REQUIRE_EQUAL(sc_details.size(), 1);
auto sc_sd_vec = sc_details["test"];
BOOST_REQUIRE_EQUAL(sc_sd_vec.size(), 1);
const auto &sc_sd = sc_sd_vec[0];
BOOST_REQUIRE_EQUAL(sc_sd.ks, "ks");
BOOST_REQUIRE_EQUAL(sc_sd.cf, "cf");
BOOST_REQUIRE_EQUAL(sc_sd.details.live, sd.live);
BOOST_REQUIRE_EQUAL(sc_sd.details.total, sd.total);
auto table_directory = table_dir(cf);
for (auto& f : collect_files(table_directory).get()) {
fs::remove(table_directory / f);
}
auto sd_post_deletion = cf.get_snapshot_details().get().at("test");
BOOST_REQUIRE_EQUAL(sd_post_deletion.total, sd_post_deletion.live);
BOOST_REQUIRE_EQUAL(sd.total, sd_post_deletion.live);
sc_details = sc.local().get_snapshot_details().get();
auto sc_sd_post_deletion_vec = sc_details["test"];
BOOST_REQUIRE_EQUAL(sc_sd_post_deletion_vec.size(), 1);
const auto &sc_sd_post_deletion = sc_sd_post_deletion_vec[0];
BOOST_REQUIRE_EQUAL(sc_sd_post_deletion.ks, "ks");
BOOST_REQUIRE_EQUAL(sc_sd_post_deletion.cf, "cf");
BOOST_REQUIRE_EQUAL(sc_sd_post_deletion.details.live, sd_post_deletion.live);
BOOST_REQUIRE_EQUAL(sc_sd_post_deletion.details.total, sd_post_deletion.total);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(test_snapshot_ctl_true_snapshots_size) {
return do_with_some_data({"cf"}, [] (cql_test_env& e) {
sharded<db::snapshot_ctl> sc;
sc.start(std::ref(e.db()), std::ref(e.get_task_manager()), std::ref(e.get_sstorage_manager()), db::snapshot_ctl::config{}).get();
auto stop_sc = deferred_stop(sc);
auto& cf = e.local_db().find_column_family("ks", "cf");
take_snapshot(e).get();
auto details = cf.get_snapshot_details().get();
BOOST_REQUIRE_EQUAL(details.size(), 1);
auto sd = details["test"];
BOOST_REQUIRE_EQUAL(sd.live, 0);
BOOST_REQUIRE_GT(sd.total, 0);
auto sc_live_size = sc.local().true_snapshots_size().get();
BOOST_REQUIRE_EQUAL(sc_live_size, sd.live);
auto table_directory = table_dir(cf);
for (auto& f : collect_files(table_directory).get()) {
fs::remove(table_directory / f);
}
auto sd_post_deletion = cf.get_snapshot_details().get().at("test");
BOOST_REQUIRE_EQUAL(sd_post_deletion.total, sd_post_deletion.live);
BOOST_REQUIRE_EQUAL(sd.total, sd_post_deletion.live);
sc_live_size = sc.local().true_snapshots_size().get();
BOOST_REQUIRE_EQUAL(sc_live_size, sd_post_deletion.live);
return make_ready_future<>();
});
}
// toppartitions_query caused a lw_shared_ptr to cross shards when moving results, #5104
SEASTAR_TEST_CASE(toppartitions_cross_shard_schema_ptr) {
return do_with_cql_env_and_compaction_groups([] (cql_test_env& e) {
e.execute_cql("CREATE TABLE ks.tab (id int PRIMARY KEY)").get();
db::toppartitions_query tq(e.db(), {{"ks", "tab"}}, {}, 1s, 100, 100);
tq.scatter().get();
auto q = e.prepare("INSERT INTO ks.tab(id) VALUES(?)").get();
// Generate many values to ensure crossing shards
for (auto i = 0; i != 100; ++i) {
e.execute_prepared(q, {cql3::raw_value::make_value(int32_type->decompose(i))}).get();
}
// This should trigger the bug in debug mode
tq.gather().get();
});
}
SEASTAR_THREAD_TEST_CASE(read_max_size) {
do_with_cql_env_and_compaction_groups([] (cql_test_env& e) {
e.execute_cql("CREATE TABLE test (pk text, ck int, v text, PRIMARY KEY (pk, ck));").get();
auto id = e.prepare("INSERT INTO test (pk, ck, v) VALUES (?, ?, ?);").get();
auto& db = e.local_db();
auto& tab = db.find_column_family("ks", "test");
auto s = tab.schema();
auto pk = tests::generate_partition_key(s);
const auto cql3_pk = cql3::raw_value::make_value(pk.key().explode().front());
const auto value = sstring(1024, 'a');
const auto raw_value = utf8_type->decompose(data_value(value));
const auto cql3_value = cql3::raw_value::make_value(raw_value);
const int num_rows = 1024;
for (int i = 0; i != num_rows; ++i) {
const auto cql3_ck = cql3::raw_value::make_value(int32_type->decompose(data_value(i)));
e.execute_prepared(id, {cql3_pk, cql3_ck, cql3_value}).get();
}
const auto partition_ranges = std::vector<dht::partition_range>{query::full_partition_range};
const std::vector<std::pair<sstring, std::function<future<size_t>(schema_ptr, const query::read_command&)>>> query_methods{
{"query_mutations()", [&db, &partition_ranges] (schema_ptr s, const query::read_command& cmd) -> future<size_t> {
return db.query_mutations(s, cmd, partition_ranges.front(), {}, db::no_timeout).then(
[] (const std::tuple<reconcilable_result, cache_temperature>& res) {
return std::get<0>(res).memory_usage();
});
}},
{"query()", [&db, &partition_ranges] (schema_ptr s, const query::read_command& cmd) -> future<size_t> {
return db.query(s, cmd, query::result_options::only_result(), partition_ranges, {}, db::no_timeout).then(
[] (const std::tuple<lw_shared_ptr<query::result>, cache_temperature>& res) {
return size_t(std::get<0>(res)->buf().size());
});
}},
{"query_mutations_on_all_shards()", [&e, &partition_ranges] (schema_ptr s, const query::read_command& cmd) -> future<size_t> {
return query_mutations_on_all_shards(e.db(), s, cmd, partition_ranges, {}, db::no_timeout).then(
[] (const std::tuple<foreign_ptr<lw_shared_ptr<reconcilable_result>>, cache_temperature>& res) {
return std::get<0>(res)->memory_usage();
});
}}
};
for (auto [query_method_name, query_method] : query_methods) {
for (auto allow_short_read : {true, false}) {
for (auto max_size : {1024u, 1024u * 1024u, 1024u * 1024u * 1024u}) {
const auto should_throw = max_size < (num_rows * value.size() * 2) && !allow_short_read;
testlog.info("checking: query_method={}, allow_short_read={}, max_size={}, should_throw={}", query_method_name, allow_short_read, max_size, should_throw);
auto slice = s->full_slice();
if (allow_short_read) {
slice.options.set<query::partition_slice::option::allow_short_read>();
} else {
slice.options.remove<query::partition_slice::option::allow_short_read>();
}
query::read_command cmd(s->id(), s->version(), slice, query::max_result_size(max_size), query::tombstone_limit::max);
try {
auto size = query_method(s, cmd).get();
// Just to ensure we are not interpreting empty results as success.
BOOST_REQUIRE(size != 0);
if (should_throw) {
BOOST_FAIL("Expected exception, but none was thrown.");
} else {
testlog.trace("No exception thrown, as expected.");
}
} catch (std::runtime_error& e) {
if (should_throw) {
testlog.trace("Exception thrown, as expected: {}", e.what());
} else {
BOOST_FAIL(fmt::format("Expected no exception, but caught: {}", e.what()));
}
}
}
}
}
}).get();
}
// Check that mutation queries, those that are stopped when the memory
// consumed by their results reach the local/global limit, are aborted
// instead of silently terminated when this happens.
SEASTAR_THREAD_TEST_CASE(unpaged_mutation_read_global_limit) {
auto cfg = cql_test_config{};
cfg.dbcfg.emplace();
// The memory available to the result memory limiter (global limit) is
// configured based on the available memory, so give a small amount to
// the "node", so we don't have to work with large amount of data.
cfg.dbcfg->available_memory = 2 * 1024 * 1024;
do_with_cql_env_and_compaction_groups([] (cql_test_env& e) {
e.execute_cql("CREATE TABLE test (pk text, ck int, v text, PRIMARY KEY (pk, ck));").get();
auto id = e.prepare("INSERT INTO test (pk, ck, v) VALUES (?, ?, ?);").get();
auto& db = e.local_db();
auto& tab = db.find_column_family("ks", "test");
auto s = tab.schema();
auto pk = tests::generate_partition_key(s);
const auto cql3_pk = cql3::raw_value::make_value(pk.key().explode().front());
const auto value = sstring(1024, 'a');
const auto raw_value = utf8_type->decompose(data_value(value));
const auto cql3_value = cql3::raw_value::make_value(raw_value);
const int num_rows = 1024;
const auto max_size = 1024u * 1024u * 1024u;
for (int i = 0; i != num_rows; ++i) {
const auto cql3_ck = cql3::raw_value::make_value(int32_type->decompose(data_value(i)));
e.execute_prepared(id, {cql3_pk, cql3_ck, cql3_value}).get();
}
const auto partition_ranges = std::vector<dht::partition_range>{query::full_partition_range};
const std::vector<std::pair<sstring, std::function<future<size_t>(schema_ptr, const query::read_command&)>>> query_methods{
{"query_mutations()", [&db, &partition_ranges] (schema_ptr s, const query::read_command& cmd) -> future<size_t> {
return db.query_mutations(s, cmd, partition_ranges.front(), {}, db::no_timeout).then(
[] (const std::tuple<reconcilable_result, cache_temperature>& res) {
return std::get<0>(res).memory_usage();
});
}},
{"query_mutations_on_all_shards()", [&e, &partition_ranges] (schema_ptr s, const query::read_command& cmd) -> future<size_t> {
return query_mutations_on_all_shards(e.db(), s, cmd, partition_ranges, {}, db::no_timeout).then(
[] (const std::tuple<foreign_ptr<lw_shared_ptr<reconcilable_result>>, cache_temperature>& res) {
return std::get<0>(res)->memory_usage();
});
}}
};
for (auto [query_method_name, query_method] : query_methods) {
testlog.info("checking: query_method={}", query_method_name);
auto slice = s->full_slice();
slice.options.remove<query::partition_slice::option::allow_short_read>();
query::read_command cmd(s->id(), s->version(), slice, query::max_result_size(max_size), query::tombstone_limit::max);
try {
auto size = query_method(s, cmd).get();
// Just to ensure we are not interpreting empty results as success.
BOOST_REQUIRE(size != 0);
BOOST_FAIL("Expected exception, but none was thrown.");
} catch (std::runtime_error& e) {
testlog.trace("Exception thrown, as expected: {}", e.what());
}
}
}, std::move(cfg)).get();
}
SEASTAR_THREAD_TEST_CASE(reader_concurrency_semaphore_selection_test) {
cql_test_config cfg;
scheduling_group unknown_scheduling_group = create_scheduling_group("unknown", 800).get();
auto cleanup_unknown_scheduling_group = defer([&unknown_scheduling_group] {
destroy_scheduling_group(unknown_scheduling_group).get();
});
const auto user_semaphore = std::mem_fn(&database_test_wrapper::get_user_read_concurrency_semaphore);
const auto system_semaphore = std::mem_fn(&database_test_wrapper::get_system_read_concurrency_semaphore);
const auto streaming_semaphore = std::mem_fn(&database_test_wrapper::get_streaming_read_concurrency_semaphore);
std::vector<std::pair<scheduling_group, std::function<reader_concurrency_semaphore&(database_test_wrapper&)>>> scheduling_group_and_expected_semaphore{
{default_scheduling_group(), system_semaphore}
};
auto sched_groups = get_scheduling_groups().get();
scheduling_group_and_expected_semaphore.emplace_back(sched_groups.compaction_scheduling_group, system_semaphore);
scheduling_group_and_expected_semaphore.emplace_back(sched_groups.memory_compaction_scheduling_group, system_semaphore);
scheduling_group_and_expected_semaphore.emplace_back(sched_groups.streaming_scheduling_group, streaming_semaphore);
scheduling_group_and_expected_semaphore.emplace_back(sched_groups.statement_scheduling_group, user_semaphore);
scheduling_group_and_expected_semaphore.emplace_back(sched_groups.memtable_scheduling_group, system_semaphore);
scheduling_group_and_expected_semaphore.emplace_back(sched_groups.memtable_to_cache_scheduling_group, system_semaphore);
scheduling_group_and_expected_semaphore.emplace_back(sched_groups.gossip_scheduling_group, system_semaphore);
scheduling_group_and_expected_semaphore.emplace_back(unknown_scheduling_group, user_semaphore);
do_with_cql_env_and_compaction_groups([&scheduling_group_and_expected_semaphore] (cql_test_env& e) {
auto& db = e.local_db();
database_test_wrapper tdb(db);
for (const auto& [sched_group, expected_sem_getter] : scheduling_group_and_expected_semaphore) {
with_scheduling_group(sched_group, [&db, sched_group = sched_group, &tdb, &expected_sem_getter = expected_sem_getter] {
auto expected_sem_ptr = &expected_sem_getter(tdb);
auto& sem = db.get_reader_concurrency_semaphore();
if (&sem != expected_sem_ptr) {
BOOST_FAIL(fmt::format("Unexpected semaphore for scheduling group {}, expected {}, got {}", sched_group.name(), expected_sem_ptr->name(), sem.name()));
}
}).get();
}
}, std::move(cfg)).get();
}
SEASTAR_THREAD_TEST_CASE(max_result_size_for_query_selection_test) {
cql_test_config cfg;
cfg.db_config->max_memory_for_unlimited_query_soft_limit(1 * 1024 * 1024, utils::config_file::config_source::CommandLine);
cfg.db_config->max_memory_for_unlimited_query_hard_limit(2 * 1024 * 1024, utils::config_file::config_source::CommandLine);
scheduling_group unknown_scheduling_group = create_scheduling_group("unknown", 800).get();
auto cleanup_unknown_scheduling_group = defer([&unknown_scheduling_group] {
destroy_scheduling_group(unknown_scheduling_group).get();
});
const auto user_max_result_size = query::max_result_size(
cfg.db_config->max_memory_for_unlimited_query_soft_limit(),
cfg.db_config->max_memory_for_unlimited_query_hard_limit(),
query::result_memory_limiter::maximum_result_size);
const auto system_max_result_size = query::max_result_size(
query::result_memory_limiter::unlimited_result_size,
query::result_memory_limiter::unlimited_result_size,
query::result_memory_limiter::maximum_result_size);
const auto maintenance_max_result_size = system_max_result_size;
std::vector<std::pair<scheduling_group, query::max_result_size>> scheduling_group_and_expected_max_result_size{
{default_scheduling_group(), system_max_result_size}
};
auto sched_groups = get_scheduling_groups().get();
scheduling_group_and_expected_max_result_size.emplace_back(sched_groups.compaction_scheduling_group, system_max_result_size);
scheduling_group_and_expected_max_result_size.emplace_back(sched_groups.memory_compaction_scheduling_group, system_max_result_size);
scheduling_group_and_expected_max_result_size.emplace_back(sched_groups.streaming_scheduling_group, maintenance_max_result_size);
scheduling_group_and_expected_max_result_size.emplace_back(sched_groups.statement_scheduling_group, user_max_result_size);
scheduling_group_and_expected_max_result_size.emplace_back(sched_groups.memtable_scheduling_group, system_max_result_size);
scheduling_group_and_expected_max_result_size.emplace_back(sched_groups.memtable_to_cache_scheduling_group, system_max_result_size);
scheduling_group_and_expected_max_result_size.emplace_back(sched_groups.gossip_scheduling_group, system_max_result_size);
scheduling_group_and_expected_max_result_size.emplace_back(unknown_scheduling_group, user_max_result_size);
do_with_cql_env_and_compaction_groups([&scheduling_group_and_expected_max_result_size] (cql_test_env& e) {
auto& db = e.local_db();
database_test_wrapper tdb(db);
for (const auto& [sched_group, expected_max_size] : scheduling_group_and_expected_max_result_size) {
with_scheduling_group(sched_group, [&db, sched_group = sched_group, expected_max_size = expected_max_size] {
const auto max_size = db.get_query_max_result_size();
if (max_size != expected_max_size) {
BOOST_FAIL(fmt::format("Unexpected max_size for scheduling group {}, expected {{{}, {}}}, got {{{}, {}}}",
sched_group.name(),
expected_max_size.soft_limit,
expected_max_size.hard_limit,
max_size.soft_limit,
max_size.hard_limit));
}
}).get();
}
}, std::move(cfg)).get();
}
// Check that during a multi-page range scan:
// * semaphore mismatch is detected
// * code is exception safe w.r.t. to the mismatch exception, e.g. readers are closed properly
SEASTAR_TEST_CASE(multipage_range_scan_semaphore_mismatch) {
return do_with_cql_env_thread([] (cql_test_env& e) {
const auto do_abort = set_abort_on_internal_error(false);
auto reset_abort = defer([do_abort] {
set_abort_on_internal_error(do_abort);
});
e.execute_cql("CREATE TABLE ks.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));").get();
auto insert_id = e.prepare("INSERT INTO ks.tbl(pk, ck, v) VALUES(?, ?, ?)").get();
auto& db = e.local_db();
auto& tbl = db.find_column_family("ks", "tbl");
auto s = tbl.schema();
auto dk = tests::generate_partition_key(tbl.schema());
const auto pk = cql3::raw_value::make_value(managed_bytes(*dk.key().begin(*s)));
const auto v = cql3::raw_value::make_value(int32_type->decompose(0));
for (int32_t ck = 0; ck < 100; ++ck) {
e.execute_prepared(insert_id, {pk, cql3::raw_value::make_value(int32_type->decompose(ck)), v}).get();
}
auto sched_groups = get_scheduling_groups().get();
query::read_command cmd1(
s->id(),
s->version(),
s->full_slice(),
query::max_result_size(std::numeric_limits<uint64_t>::max(), std::numeric_limits<uint64_t>::max(), std::numeric_limits<uint64_t>::max()),
query::tombstone_limit::max,
query::row_limit(4),
query::partition_limit::max,
gc_clock::now(),
std::nullopt,
query_id::create_random_id(),
query::is_first_page::yes);
auto cmd2 = cmd1;
auto cr = query::clustering_range::make_starting_with({clustering_key::from_single_value(*s, int32_type->decompose(3)), false});
cmd2.slice = partition_slice_builder(*s).set_specific_ranges(query::specific_ranges(dk.key(), {cr})).build();
cmd2.is_first_page = query::is_first_page::no;
auto pr = dht::partition_range::make_starting_with({dk, true});
auto prs = dht::partition_range_vector{pr};
auto read_page = [&] (scheduling_group sg, const query::read_command& cmd) {
with_scheduling_group(sg, [&] {
return query_data_on_all_shards(e.db(), s, cmd, prs, query::result_options::only_result(), {}, db::no_timeout);
}).get();
};
read_page(default_scheduling_group(), cmd1);
BOOST_REQUIRE_EXCEPTION(read_page(sched_groups.statement_scheduling_group, cmd2), std::runtime_error,
testing::exception_predicate::message_contains("semaphore mismatch detected, dropping reader"));
});
}
// Test `upgrade_sstables` on all keyspaces (including the system keyspace).
// Refs: #9494 (https://github.com/scylladb/scylla/issues/9494)
SEASTAR_TEST_CASE(upgrade_sstables) {
return do_with_cql_env_and_compaction_groups([] (cql_test_env& e) {
e.db().invoke_on_all([] (replica::database& db) -> future<> {
auto& cm = db.get_compaction_manager();
for (auto& [ks_name, ks] : db.get_keyspaces()) {
const auto& erm = ks.get_static_effective_replication_map();
auto owned_ranges_ptr = compaction::make_owned_ranges_ptr(co_await db.get_keyspace_local_ranges(erm));
for (auto& [cf_name, schema] : ks.metadata()->cf_meta_data()) {
auto& t = db.find_column_family(schema->id());
constexpr bool exclude_current_version = false;
co_await t.parallel_foreach_compaction_group_view([&] (compaction::compaction_group_view& ts) {
return cm.perform_sstable_upgrade(owned_ranges_ptr, ts, exclude_current_version, tasks::task_info{});
});
}
}
}).get();
});
}
SEASTAR_THREAD_TEST_CASE(per_service_level_reader_concurrency_semaphore_test) {
cql_test_config cfg;
do_with_cql_env_thread([] (cql_test_env& e) {
const size_t num_service_levels = 3;
const size_t num_keys_to_insert = 10;
const size_t num_individual_reads_to_test = 50;
auto& db = e.local_db();
database_test_wrapper dbt(db);
size_t total_memory = dbt.get_total_user_reader_concurrency_semaphore_memory();
sharded<qos::service_level_controller>& sl_controller = e.service_level_controller_service();
std::array<sstring, num_service_levels> sl_names;
qos::service_level_options slo;
size_t expected_total_weight = 200; // 200 from `sl:driver`
auto index_to_weight = [] (size_t i) -> size_t {
return (i + 1)*100;
};
// make the default service level take as little memory as possible
slo.shares.emplace<int32_t>(1);
expected_total_weight += 1;
sl_controller.local().add_service_level(qos::service_level_controller::default_service_level_name, slo).get();
// Just to make the code more readable.
auto get_reader_concurrency_semaphore_for_sl = [&] (sstring sl_name) -> reader_concurrency_semaphore& {
return *sl_controller.local().with_service_level(sl_name, noncopyable_function<reader_concurrency_semaphore*()>([&] {
return &db.get_reader_concurrency_semaphore();
})).get();
};
for (unsigned i = 0; i < num_service_levels; i++) {
sstring sl_name = format("sl{}", i);
slo.shares.emplace<int32_t>(index_to_weight(i));
sl_controller.local().add_service_level(sl_name, slo).get();
expected_total_weight += index_to_weight(i);
// Make sure that the total weight is tracked correctly in the semaphore group
BOOST_REQUIRE_EQUAL(expected_total_weight, dbt.get_total_user_reader_concurrency_semaphore_weight());
sl_names[i] = sl_name;
size_t total_distributed_memory = 0;
// Include `sl:driver` in computations
total_distributed_memory += get_reader_concurrency_semaphore_for_sl("driver").available_resources().memory;
for (unsigned j = 0 ; j <= i ; j++) {
reader_concurrency_semaphore& sem = get_reader_concurrency_semaphore_for_sl(sl_names[j]);
// Make sure that all semaphores that has been created until now - have the right amount of available memory
// after the operation has ended.
// We allow for a small delta of up to num_service_levels. This allows an off-by-one for each semaphore,
// the remainder being added to one of the semaphores.
// We make sure this didn't leak/create memory by checking the total below.
const auto delta = std::abs(ssize_t((index_to_weight(j) * total_memory) / expected_total_weight) - sem.available_resources().memory);
BOOST_REQUIRE_LE(delta, num_service_levels);
total_distributed_memory += sem.available_resources().memory;
}
total_distributed_memory += get_reader_concurrency_semaphore_for_sl(qos::service_level_controller::default_service_level_name).available_resources().memory;
BOOST_REQUIRE_EQUAL(total_distributed_memory, total_memory);
}
auto get_semaphores_stats_snapshot = [&] () {
std::unordered_map<sstring, reader_concurrency_semaphore::stats> snapshot;
for (auto&& sl_name : sl_names) {
snapshot[sl_name] = get_reader_concurrency_semaphore_for_sl(sl_name).get_stats();
}
return snapshot;
};
e.execute_cql("CREATE TABLE tbl (a int, b int, PRIMARY KEY (a));").get();
for (unsigned i = 0; i < num_keys_to_insert; i++) {
for (unsigned j = 0; j < num_keys_to_insert; j++) {
e.execute_cql(format("INSERT INTO tbl(a, b) VALUES ({}, {});", i, j)).get();
}
}
for (unsigned i = 0; i < num_individual_reads_to_test; i++) {
int random_service_level = tests::random::get_int(num_service_levels - 1);
auto snapshot_before = get_semaphores_stats_snapshot();
sl_controller.local().with_service_level(sl_names[random_service_level], noncopyable_function<future<>()> ([&] {
return e.execute_cql("SELECT * FROM tbl;").discard_result();
})).get();
auto snapshot_after = get_semaphores_stats_snapshot();
for (auto& [sl_name, stats] : snapshot_before) {
// Make sure that the only semaphore that experienced any activity (at least measured activity) is
// the semaphore that belongs to the current service level.
BOOST_REQUIRE((stats == snapshot_after[sl_name] && sl_name != sl_names[random_service_level]) ||
(stats != snapshot_after[sl_name] && sl_name == sl_names[random_service_level]));
}
}
}, std::move(cfg)).get();
}
SEASTAR_TEST_CASE(populate_from_quarantine_works) {
auto tmpdir_for_data = make_lw_shared<tmpdir>();
auto db_cfg_ptr = make_shared<db::config>();
db_cfg_ptr->data_file_directories(std::vector<sstring>({ tmpdir_for_data->path().string() }));
locator::host_id host_id;
// populate tmpdir_for_data and
// move a random sstable to quarantine
co_await do_with_some_data({"cf"}, [&host_id] (cql_test_env& e) -> future<> {
host_id = e.local_db().get_token_metadata().get_my_id();
auto& db = e.db();
co_await db.invoke_on_all([] (replica::database& db) {
auto& cf = db.find_column_family("ks", "cf");
return cf.flush();
});
auto shard = tests::random::get_int<unsigned>(0, smp::count);
auto found = false;
for (unsigned i = 0; i < smp::count && !found; i++) {
found = co_await db.invoke_on((shard + i) % smp::count, [] (replica::database& db) -> future<bool> {
auto& cf = db.find_column_family("ks", "cf");
bool found = false;
co_await cf.parallel_foreach_compaction_group_view([&] (compaction::compaction_group_view& ts) -> future<> {
auto sstables = co_await in_strategy_sstables(ts);
if (sstables.empty()) {
co_return;
}
auto idx = tests::random::get_int<size_t>(0, sstables.size() - 1);
testlog.debug("Moving sstable #{} out of {} to quarantine", idx, sstables.size());
auto sst = sstables[idx];
co_await sst->change_state(sstables::sstable_state::quarantine);
found |= true;
});
co_return found;
});
}
BOOST_REQUIRE(found);
}, false, db_cfg_ptr);
// reload the table from tmpdir_for_data and
// verify that all rows are still there
size_t row_count = 0;
cql_test_config test_config(db_cfg_ptr);
test_config.host_id = host_id;
co_await do_with_cql_env([&row_count] (cql_test_env& e) -> future<> {
auto res = co_await e.execute_cql("select * from ks.cf;");
auto rows = dynamic_pointer_cast<cql_transport::messages::result_message::rows>(res);
BOOST_REQUIRE(rows);
row_count = rows->rs().result_set().size();
}, std::move(test_config));
BOOST_REQUIRE_EQUAL(row_count, 6);
}
SEASTAR_TEST_CASE(snapshot_with_quarantine_works) {
return do_with_some_data({"cf"}, [] (cql_test_env& e) -> future<> {
auto& db = e.db();
co_await db.invoke_on_all([] (replica::database& db) {
auto& cf = db.find_column_family("ks", "cf");
return cf.flush();
});
std::set<sstring> expected = {
"manifest.json",
};
// move a random sstable to quarantine
auto shard = tests::random::get_int<unsigned>(0, smp::count);
auto found = false;
for (unsigned i = 0; i < smp::count; i++) {
co_await db.invoke_on((shard + i) % smp::count, [&] (replica::database& db) -> future<> {
auto& cf = db.find_column_family("ks", "cf");
co_await cf.parallel_foreach_compaction_group_view([&] (compaction::compaction_group_view& ts) -> future<> {
auto sstables = co_await in_strategy_sstables(ts);
if (sstables.empty()) {
co_return;
}
// collect all expected sstable data files
for (auto sst : sstables) {
expected.insert(sst->component_basename(sstables::component_type::Data));
}
if (std::exchange(found, true)) {
co_return;
}
auto idx = tests::random::get_int<size_t>(0, sstables.size() - 1);
auto sst = sstables[idx];
co_await sst->change_state(sstables::sstable_state::quarantine);
});
});
}
BOOST_REQUIRE(found);
co_await take_snapshot(e, "ks", "cf", "test", true /* skip_flush */);
testlog.debug("Expected: {}", expected);
// snapshot triggered a flush and wrote the data down.
BOOST_REQUIRE_GT(expected.size(), 1);
auto& cf = db.local().find_column_family("ks", "cf");
auto in_snap_dir = co_await collect_files(table_dir(cf) / sstables::snapshots_dir / "test");
// all files were copied and manifest was generated
BOOST_REQUIRE(std::includes(in_snap_dir.begin(), in_snap_dir.end(), expected.begin(), expected.end()));
});
}
SEASTAR_TEST_CASE(database_drop_column_family_clears_querier_cache) {
return do_with_cql_env_and_compaction_groups([] (cql_test_env& e) {
e.execute_cql("create table ks.cf (k text, v int, primary key (k));").get();
auto& db = e.local_db();
auto& tbl = db.find_column_family("ks", "cf");
auto op = std::optional(tbl.read_in_progress());
auto s = tbl.schema();
auto q = replica::querier(
tbl.as_mutation_source(),
tbl.schema(),
database_test_wrapper(db).get_user_read_concurrency_semaphore().make_tracking_only_permit(s, "test", db::no_timeout, {}),
query::full_partition_range,
s->full_slice(),
nullptr,
tombstone_gc_state(nullptr));
auto f = replica::database::legacy_drop_table_on_all_shards(e.db(), e.get_system_keyspace(), "ks", "cf");
// we add a querier to the querier cache while the drop is ongoing
auto& qc = db.get_querier_cache();
qc.insert_data_querier(query_id::create_random_id(), std::move(q), nullptr);
BOOST_REQUIRE_EQUAL(qc.get_stats().population, 1);
op.reset(); // this should allow the drop to finish
f.get();
// the drop should have cleaned up all entries belonging to that table
BOOST_REQUIRE_EQUAL(qc.get_stats().population, 0);
});
}
static future<> test_drop_table_with_auto_snapshot(bool auto_snapshot) {
sstring ks_name = "ks";
sstring table_name = format("table_with_auto_snapshot_{}", auto_snapshot ? "enabled" : "disabled");
auto tmpdir_for_data = make_lw_shared<tmpdir>();
auto db_cfg_ptr = make_shared<db::config>();
db_cfg_ptr->data_file_directories(std::vector<sstring>({ tmpdir_for_data->path().string() }));
db_cfg_ptr->auto_snapshot(auto_snapshot);
co_await do_with_some_data({table_name}, [&] (cql_test_env& e) -> future<> {
auto cf_dir = table_dir(e.local_db().find_column_family(ks_name, table_name)).native();
// Pass `with_snapshot=true` to drop_table_on_all
// to allow auto_snapshot (based on the configuration above).
// The table directory should therefore exist after the table is dropped if auto_snapshot is disabled in the configuration.
co_await replica::database::legacy_drop_table_on_all_shards(e.db(), e.get_system_keyspace(), ks_name, table_name, true);
auto cf_dir_exists = co_await file_exists(cf_dir);
BOOST_REQUIRE_EQUAL(cf_dir_exists, auto_snapshot);
co_return;
}, false, db_cfg_ptr);
}
SEASTAR_TEST_CASE(drop_table_with_auto_snapshot_enabled) {
return test_drop_table_with_auto_snapshot(true);
}
SEASTAR_TEST_CASE(drop_table_with_auto_snapshot_disabled) {
return test_drop_table_with_auto_snapshot(false);
}
SEASTAR_TEST_CASE(drop_table_with_no_snapshot) {
sstring ks_name = "ks";
sstring table_name = "table_with_no_snapshot";
co_await do_with_some_data({table_name}, [&] (cql_test_env& e) -> future<> {
auto cf_dir = table_dir(e.local_db().find_column_family(ks_name, table_name)).native();
// Pass `with_snapshot=false` to drop_table_on_all
// to disallow auto_snapshot.
// The table directory should therefore not exist after the table is dropped.
co_await replica::database::legacy_drop_table_on_all_shards(e.db(), e.get_system_keyspace(), ks_name, table_name, false);
auto cf_dir_exists = co_await file_exists(cf_dir);
BOOST_REQUIRE_EQUAL(cf_dir_exists, false);
co_return;
});
}
SEASTAR_TEST_CASE(drop_table_with_explicit_snapshot) {
sstring ks_name = "ks";
sstring table_name = "table_with_explicit_snapshot";
co_await do_with_some_data({table_name}, [&] (cql_test_env& e) -> future<> {
auto snapshot_tag = format("test-{}", db_clock::now().time_since_epoch().count());
co_await take_snapshot(e, ks_name, table_name, snapshot_tag);
auto cf_dir = table_dir(e.local_db().find_column_family(ks_name, table_name)).native();
// With explicit snapshot and with_snapshot=false
// dir should still be kept, regardless of the
// with_snapshot parameter and auto_snapshot config.
co_await replica::database::legacy_drop_table_on_all_shards(e.db(), e.get_system_keyspace(), ks_name, table_name, false);
auto cf_dir_exists = co_await file_exists(cf_dir);
BOOST_REQUIRE_EQUAL(cf_dir_exists, true);
co_return;
});
}
SEASTAR_TEST_CASE(mutation_dump_generated_schema_deterministic_id_version) {
simple_schema s;
auto os1 = replica::mutation_dump::generate_output_schema_from_underlying_schema(s.schema());
auto os2 = replica::mutation_dump::generate_output_schema_from_underlying_schema(s.schema());
BOOST_REQUIRE_EQUAL(os1->id(), os2->id());
BOOST_REQUIRE_EQUAL(os1->version(), os2->version());
return make_ready_future<>();
}
SEASTAR_TEST_CASE(enable_drained_compaction_manager) {
return do_with_cql_env_thread([] (cql_test_env& e) {
e.db().invoke_on_all([] (replica::database& db) -> future<> {
auto& cm = db.get_compaction_manager();
co_await cm.drain();
cm.enable();
}).get();
});
}
SEASTAR_TEST_CASE(test_drop_quarantined_sstables) {
return do_with_cql_env_thread([] (cql_test_env& e) {
e.execute_cql("create table cf (p text PRIMARY KEY, c int)").get();
for (int i = 0; i < 100; i++) {
e.execute_cql(format("insert into cf (p, c) values ('key{}', {})", i * i, i)).get();
e.db().invoke_on_all([] (replica::database& db) {
auto& cf = db.find_column_family("ks", "cf");
return cf.flush();
}).get();
}
auto initial_sstable_count = e.db().map_reduce0(
[] (replica::database& db) -> future<size_t> {
auto& cf = db.find_column_family("ks", "cf");
co_return cf.sstables_count();
},
0,
std::plus<size_t>{}
).get();
BOOST_REQUIRE_GT(initial_sstable_count, 0);
auto quarantined_count = e.db().map_reduce0(
[] (replica::database& _db) -> future<size_t> {
auto& cf = _db.find_column_family("ks", "cf");
size_t quarantined_on_shard = 0;
auto& cm = cf.get_compaction_manager();
co_await cf.parallel_foreach_compaction_group_view([&] (compaction::compaction_group_view& ts) -> future<> {
return cm.run_with_compaction_disabled(ts, [&] () -> future<> {
auto sstables = co_await in_strategy_sstables(ts);
if (sstables.empty()) {
co_return;
}
auto quarantine_n = 1 + tests::random::get_int<size_t>(sstables.size() / 5);
quarantined_on_shard += quarantine_n;
for (size_t i = 0; i < quarantine_n; i++) {
co_await sstables[i]->change_state(sstables::sstable_state::quarantine);
}
});
});
co_return quarantined_on_shard;
},
size_t(0),
std::plus<size_t>{}
).get();
BOOST_REQUIRE_GT(quarantined_count, 0);
e.db().invoke_on_all([] (replica::database& db) {
auto& cf = db.find_column_family("ks", "cf");
return cf.drop_quarantined_sstables();
}).get();
auto remaining_quarantined = e.db().map_reduce0(
[] (replica::database& db) -> future<size_t> {
auto& cf = db.find_column_family("ks", "cf");
auto& sstables = *cf.get_sstables();
co_return std::count_if(sstables.begin(), sstables.end(), [] (shared_sstable sst) {
return sst->is_quarantined();
});
},
size_t(0),
std::plus<size_t>{}
).get();
BOOST_REQUIRE_EQUAL(remaining_quarantined, 0);
});
}
SEASTAR_THREAD_TEST_CASE(test_tombstone_gc_state_snapshot) {
auto table_gc_mode_timeout = schema_builder("test", "table_gc_mode_timeout")
.with_column("pk", utf8_type, column_kind::partition_key)
.with_tombstone_gc_options(tombstone_gc_options({ {"mode", "timeout"} }))
.set_gc_grace_seconds(10)
.build();
auto table_gc_mode_disabled = schema_builder("test", "table_gc_mode_disabled")
.with_column("pk", utf8_type, column_kind::partition_key)
.with_tombstone_gc_options(tombstone_gc_options({ {"mode", "disabled"} }))
.build();
auto table_gc_mode_immediate = schema_builder("test", "table_gc_mode_immediate")
.with_column("pk", utf8_type, column_kind::partition_key)
.with_tombstone_gc_options(tombstone_gc_options({ {"mode", "immediate"} }))
.build();
auto table_gc_mode_repair1 = schema_builder("test", "table_gc_mode_repair1")
.with_column("pk", utf8_type, column_kind::partition_key)
.with_tombstone_gc_options(tombstone_gc_options({ {"mode", "repair"}, {"propagation_delay_in_seconds", "188"} }))
.build();
auto table_gc_mode_repair2 = schema_builder("test", "table_gc_mode_repair2")
.with_column("pk", utf8_type, column_kind::partition_key)
.with_tombstone_gc_options(tombstone_gc_options({ {"mode", "repair"}, {"propagation_delay_in_seconds", "288"} }))
.build();
auto table_gc_mode_repair3 = schema_builder("test", "table_gc_mode_repair3")
.with_column("pk", utf8_type, column_kind::partition_key)
.with_tombstone_gc_options(tombstone_gc_options({ {"mode", "repair"}, {"propagation_delay_in_seconds", "388"} }))
.build();
schema_builder::register_static_configurator([] (const sstring& ks_name, const sstring& cf_name, schema_static_props& props) {
if (ks_name == "test" && cf_name == "table_gc_mode_group0") {
props.is_group0_table = true;
}
});
auto table_gc_mode_group0 = schema_builder("test", "table_gc_mode_group0")
.with_column("pk", utf8_type, column_kind::partition_key)
.build();
BOOST_REQUIRE(table_gc_mode_group0->static_props().is_group0_table);
// One pk to rule them all, all schemas have the same partition key, so we
// can reuse a single key for this test.
const auto pk = partition_key::from_single_value(*table_gc_mode_timeout, utf8_type->decompose(data_value("pk")));
const auto dk = dht::decorate_key(*table_gc_mode_timeout, pk);
const auto repair_range = dht::token_range::make(dht::first_token(), dk.token());
shared_tombstone_gc_state shared_state;
const auto first_repair_time = gc_clock::now() - gc_clock::duration(std::chrono::hours(6));
shared_state.update_repair_time(table_gc_mode_repair1->id(), repair_range, first_repair_time);
shared_state.update_repair_time(table_gc_mode_repair2->id(), repair_range, first_repair_time);
shared_state.update_group0_refresh_time(first_repair_time);
auto snapshot = shared_state.snapshot();
BOOST_REQUIRE_LE(gc_clock::now() - snapshot.query_time(), gc_clock::duration(std::chrono::seconds(1)));
// Advance gc clock and change the gc state to simulate a later point in time.
// Then check that gc-before against the shared-state yields the current
// state, while gc-before against the snapshot yields the before state.
const auto now = gc_clock::now() + gc_clock::duration(std::chrono::hours(6));
const auto gc_state = tombstone_gc_state(shared_state).with_commitlog_check_disabled();
const auto second_repair_time = gc_clock::now() + gc_clock::duration(std::chrono::hours(3));
shared_state.update_repair_time(table_gc_mode_repair1->id(), repair_range, second_repair_time);
shared_state.drop_repair_history_for_table(table_gc_mode_repair2->id());
shared_state.update_repair_time(table_gc_mode_repair3->id(), repair_range, second_repair_time);
shared_state.update_group0_refresh_time(second_repair_time);
BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_timeout, dk, now), now - table_gc_mode_timeout->gc_grace_seconds());
BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_timeout, dk, false), snapshot.query_time() - table_gc_mode_timeout->gc_grace_seconds());
BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_disabled, dk, now), gc_clock::time_point::min());
BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_disabled, dk, false), gc_clock::time_point::min());
BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_immediate, dk, now), now);
BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_immediate, dk, false), snapshot.query_time());
BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair1, dk, now), second_repair_time - table_gc_mode_repair1->tombstone_gc_options().propagation_delay_in_seconds());
BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair2, dk, now), gc_clock::time_point::min());
BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair3, dk, now), second_repair_time - table_gc_mode_repair3->tombstone_gc_options().propagation_delay_in_seconds());
BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_repair1, dk, false), first_repair_time - table_gc_mode_repair1->tombstone_gc_options().propagation_delay_in_seconds());
BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_repair2, dk, false), first_repair_time - table_gc_mode_repair2->tombstone_gc_options().propagation_delay_in_seconds());
BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_repair3, dk, false), gc_clock::time_point::min());
BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_group0, dk, now), second_repair_time);
BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_group0, dk, false), first_repair_time);
}
SEASTAR_TEST_CASE(test_max_purgeable_combine) {
const gc_clock::time_point t_pre_treshold = gc_clock::now();
const gc_clock::time_point t1 = t_pre_treshold + std::chrono::seconds(10);
const gc_clock::time_point t2 = t1 + std::chrono::seconds(10);
const gc_clock::time_point t_post_treshold = t2 + std::chrono::seconds(10);
auto check_tombstone = [] (const max_purgeable& mp, const max_purgeable& combined, tombstone t, bool expected_can_purge, std::source_location loc = std::source_location::current()) {
testlog.trace("check_tombstone({}, {}, {}, {}) @ {}:{}", mp, combined, t, expected_can_purge, loc.file_name(), loc.line());
BOOST_REQUIRE_EQUAL(expected_can_purge, mp.can_purge(t).can_purge);
if (!expected_can_purge) {
// The combined max_purgeable can be weaker than this input, if
// the other input had no expiry threshold.
BOOST_REQUIRE(!combined.can_purge(t).can_purge);
}
};
auto check_tombstones = [&] (const max_purgeable& mp, const max_purgeable& combined) {
if (mp) {
check_tombstone(mp, combined, {mp.timestamp() - 1, t_post_treshold}, true);
check_tombstone(mp, combined, {mp.timestamp() + 1, t_post_treshold}, false);
if (mp.expiry_threshold()) {
check_tombstone(mp, combined, {mp.timestamp() - 1, t_pre_treshold}, true);
check_tombstone(mp, combined, {mp.timestamp() + 1, t_pre_treshold}, true);
}
} else {
check_tombstone(mp, combined, {1, t_post_treshold}, true);
}
};
auto check = [&] (const max_purgeable& a, const max_purgeable& b, const max_purgeable& expected, std::source_location loc = std::source_location::current()) {
auto combined = a;
combined.combine(b);
testlog.debug("combine({}, {}) => {} == {} @ {}:{}", a, b, combined, expected, loc.file_name(), loc.line());
BOOST_REQUIRE(a || b);
BOOST_REQUIRE_EQUAL(combined, expected);
check_tombstones(a, combined);
check_tombstones(b, combined);
};
check({}, max_purgeable{100}, max_purgeable{100});
check(max_purgeable{100}, {}, max_purgeable{100});
check(max_purgeable{10}, max_purgeable{100}, max_purgeable{10});
const auto ts_mt = max_purgeable::timestamp_source::memtable_possibly_shadowing_data;
const auto ts_sst = max_purgeable::timestamp_source::other_sstables_possibly_shadowing_data;
check({}, max_purgeable{100, ts_mt}, max_purgeable{100, ts_mt});
check(max_purgeable{100, ts_mt}, {}, max_purgeable{100, ts_mt});
check(max_purgeable{10, ts_sst}, max_purgeable{100, ts_mt}, max_purgeable{10, ts_sst});
check({}, max_purgeable{100, t1, ts_mt}, max_purgeable{100, t1, ts_mt});
check(max_purgeable{10, ts_mt}, max_purgeable{100, t1, ts_sst}, max_purgeable{10, ts_mt});
check(max_purgeable{100, ts_mt}, max_purgeable{10, t1, ts_sst}, max_purgeable{10, ts_sst});
check(max_purgeable{10, t1, ts_mt}, max_purgeable{100, t2, ts_sst}, max_purgeable{10, t1, ts_mt});
check(max_purgeable{100, t1, ts_mt}, max_purgeable{10, t2, ts_sst}, max_purgeable{10, t1, ts_sst});
return make_ready_future<>();
}
SEASTAR_TEST_CASE(test_max_purgeable_can_purge) {
const gc_clock::time_point t_pre_treshold = gc_clock::now();
const gc_clock::time_point t1 = t_pre_treshold + std::chrono::seconds(10);
const gc_clock::time_point t_post_treshold = t1 + std::chrono::seconds(10);
const auto ts_mt = max_purgeable::timestamp_source::memtable_possibly_shadowing_data;
const auto ts_sst = max_purgeable::timestamp_source::other_sstables_possibly_shadowing_data;
auto check = [] (const max_purgeable& mp, tombstone t, bool expected_can_gc) {
const auto res = mp.can_purge(t);
BOOST_REQUIRE_EQUAL(res.can_purge, expected_can_gc);
BOOST_REQUIRE_EQUAL(res.timestamp_source, mp.source());
};
check({}, {100, t1}, true);
check(max_purgeable{10, ts_mt}, tombstone{100, t_post_treshold}, false);
check(max_purgeable{100, ts_sst}, tombstone{100, t_post_treshold}, false);
check(max_purgeable{200, ts_sst}, tombstone{100, t_post_treshold}, true);
check(max_purgeable{10, t1, ts_sst}, tombstone{100, t_post_treshold}, false);
check(max_purgeable{100, t1, ts_sst}, tombstone{100, t_post_treshold}, false);
check(max_purgeable{200, t1, ts_sst}, tombstone{100, t_post_treshold}, true);
check(max_purgeable{10, t1, ts_sst}, tombstone{100, t_pre_treshold}, true);
check(max_purgeable{100, t1, ts_sst}, tombstone{100, t_pre_treshold}, true);
check(max_purgeable{200, t1, ts_sst}, tombstone{100, t_pre_treshold}, true);
return make_ready_future<>();
}
SEASTAR_TEST_CASE(test_query_tombstone_gc) {
return do_with_cql_env_thread([] (cql_test_env& env) {
const auto keyspace_name = get_name();
const auto table_name = "tbl";
// Can use tablets and RF=1 after #21623 is fixed.
env.execute_cql(std::format("CREATE KEYSPACE {} WITH"
" replication = {{'class': 'NetworkTopologyStrategy', 'replication_factor': 3}} AND"
" tablets = {{'enabled': 'false'}}", keyspace_name)).get();
env.execute_cql(std::format("CREATE TABLE {}.{} (pk int, ck int, v int, PRIMARY KEY (pk, ck))"
" WITH compaction = {{'class': 'NullCompactionStrategy'}}"
" AND tombstone_gc = {{'mode': 'repair', 'propagation_delay_in_seconds': 0}}", keyspace_name, table_name)).get();
auto& db = env.local_db();
auto& tbl = db.find_column_family(keyspace_name, table_name);
const auto schema = tbl.schema();
const auto tid = schema->id();
const auto pk_value = 1;
const auto pk = partition_key::from_exploded(*schema, {data_value(pk_value).serialize_nonnull()});
const auto dk = dht::decorate_key(*schema, pk);
const auto key_shard = tbl.shard_for_reads(dk.token());
env.execute_cql(format("DELETE FROM {}.{} WHERE pk = 1 AND ck = 1", keyspace_name, table_name, pk_value)).get();
env.db().invoke_on(key_shard, [] (replica::database& db) {
return db.flush_commitlog();
}).get();
const auto repair_range = dht::token_range::make(dht::first_token(), dht::last_token());
const auto repair_time = gc_clock::now() + gc_clock::duration(std::chrono::hours(1));
env.db().invoke_on_all([tid, &repair_range, &repair_time] (replica::database& db) {
auto& tbl = db.find_column_family(tid);
tbl.get_compaction_manager().get_shared_tombstone_gc_state().update_repair_time(tid, repair_range, repair_time);
}).get();
testlog.info("repair_time: {}", repair_time);
auto slice = partition_slice_builder(*schema, schema->full_slice())
.with_option<query::partition_slice::option::bypass_cache>()
.build();
const auto cmd = query::read_command(schema->id(), schema->version(), slice, db.get_query_max_result_size(), query::tombstone_limit::max);
const auto pr = dht::partition_range::make_singular(dk);
env.db().invoke_on(key_shard, [tid, &cmd, &pr] (replica::database& db) -> future<> {
auto& tbl = db.find_column_family(tid);
const auto schema = tbl.schema();
auto permit = co_await db.obtain_reader_permit(tbl, "read", db::no_timeout, {});
auto accounter = co_await db.get_result_memory_limiter().new_mutation_read(*cmd.max_result_size, query::short_read::no);
const auto res = co_await tbl.mutation_query(schema, std::move(permit), cmd, pr, {}, std::move(accounter), db::no_timeout);
BOOST_CHECK_EQUAL(res.partitions().size(), 0);
}).get();
{
const auto res = replica::query_mutations_on_all_shards(env.db(), schema, cmd, {pr}, {}, db::no_timeout).get();
BOOST_CHECK_EQUAL(std::get<0>(res)->partitions().size(), 0);
}
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(test_tombstone_gc_state_gc_mode) {
const auto compact_range = dht::token_range::make(dht::token(100), dht::token(200));
auto check = [&] (tombstone_gc_state tombstone_gc, schema_ptr schema, dht::decorated_key dk, gc_clock::time_point now,
gc_clock::time_point expected_gc_before, std::source_location loc = std::source_location::current()) {
testlog.info("check() @ {}:{}", loc.file_name(), loc.line());
auto gc_before = tombstone_gc.get_gc_before_for_key(schema, dk, now);
BOOST_REQUIRE_EQUAL(gc_before, expected_gc_before);
auto gc_res = tombstone_gc.get_gc_before_for_range(schema, compact_range, now);
BOOST_REQUIRE_EQUAL(gc_res.min_gc_before, expected_gc_before);
BOOST_REQUIRE_EQUAL(gc_res.max_gc_before, expected_gc_before);
BOOST_REQUIRE_EQUAL(gc_res.knows_entire_range, true);
};
shared_tombstone_gc_state shared_state;
for (auto gc_mode : {tombstone_gc_mode::timeout, tombstone_gc_mode::disabled, tombstone_gc_mode::immediate, tombstone_gc_mode::repair}) {
auto schema = schema_builder("ks", "tbl")
.with_column("pk", int32_type, column_kind::partition_key)
.with_tombstone_gc_options(tombstone_gc_options({{"mode", fmt::to_string(gc_mode)}}))
.build();
const auto repair_time = gc_clock::now() - gc_clock::duration(std::chrono::hours(6));
const auto repair_range = dht::token_range::make(dht::first_token(), dht::last_token());
shared_state.update_repair_time(schema->id(), repair_range, repair_time);
const auto now = gc_clock::now();
const auto pk = partition_key::from_single_value(*schema, serialized(1));
const auto dk = dht::decorate_key(*schema, pk);
// These constructors overrides gc_mode
check(tombstone_gc_state::no_gc(), schema, dk, now, gc_clock::time_point::min());
check(tombstone_gc_state::gc_all(), schema, dk, now, gc_clock::time_point::max());
switch (gc_mode) {
case tombstone_gc_mode::timeout:
check(tombstone_gc_state(shared_state, false), schema, dk, now, now - std::chrono::seconds(schema->gc_grace_seconds()));
break;
case tombstone_gc_mode::disabled:
check(tombstone_gc_state(shared_state, false), schema, dk, now, gc_clock::time_point::min());
break;
case tombstone_gc_mode::immediate:
check(tombstone_gc_state(shared_state, false), schema, dk, now, now);
break;
case tombstone_gc_mode::repair:
check(tombstone_gc_state(shared_state, false), schema, dk, now, repair_time - schema->tombstone_gc_options().propagation_delay_in_seconds());
break;
}
}
return make_ready_future<>();
}
BOOST_AUTO_TEST_SUITE_END()
Reference in New Issue View Git Blame Copy Permalink