/*
 * Copyright (C) 2016-present ScyllaDB
 */

/*
 * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
 */


#include <boost/test/tools/old/interface.hpp>
#include <seastar/core/seastar.hh>
#include <seastar/core/shard_id.hh>
#include <seastar/core/smp.hh>
#include <seastar/core/thread.hh>
#include <seastar/core/coroutine.hh>
#include <seastar/core/bitops.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 "utils/rjson.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 "sstables/sstable_version.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<>();
        });
    });
}

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_thread([](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_thread([] (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_thread([] (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_thread([](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)) {
    do_with_cql_env_thread([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();
}

SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_basic) {
    test_database(run_mutation_source_tests_plain_basic);
}

SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_reader_conversion) {
    test_database(run_mutation_source_tests_plain_reader_conversion);
}

SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_fragments_monotonic) {
    test_database(run_mutation_source_tests_plain_fragments_monotonic);
}

SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain_read_back) {
    test_database(run_mutation_source_tests_plain_read_back);
}

SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_basic) {
    test_database(run_mutation_source_tests_reverse_basic);
}

SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_reader_conversion) {
    test_database(run_mutation_source_tests_reverse_reader_conversion);
}

SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_fragments_monotonic) {
    test_database(run_mutation_source_tests_reverse_fragments_monotonic);
}

SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse_read_back) {
    test_database(run_mutation_source_tests_reverse_read_back);
}

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_thread([&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_thread([&] (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
// \param func: function to be called back, in a seastar thread.
future<> do_with_some_data_in_thread(std::vector<sstring> cf_names, std::function<void (cql_test_env&)> func, bool create_mvs = false, shared_ptr<db::config> db_cfg_ptr = {}, size_t num_keys = 2) {
    return seastar::async([cf_names = std::move(cf_names), func = std::move(func), create_mvs,  db_cfg_ptr = std::move(db_cfg_ptr), num_keys] () 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_thread([cf_names = std::move(cf_names), func = std::move(func), create_mvs, num_keys] (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();
                auto stmt = e.prepare(fmt::format("insert into {} (p1, c1, c2, r1) values (?, ?, ?, ?)", cf_name)).get();
                auto make_key = [] (int64_t k) {
                    std::string s = fmt::format("key{}", k);
                    return cql3::raw_value::make_value(utf8_type->decompose(s));
                };
                auto make_val = [] (int64_t x) {
                    return cql3::raw_value::make_value(int32_type->decompose(int32_t{x}));
                };
                for (size_t i = 0; i < num_keys; ++i) {
                    auto key = tests::random::get_int<int32_t>(1, 1000000);
                    e.execute_prepared(stmt, {make_key(key), make_val(key), make_val(key + 1), make_val(key + 2)}).get();
                    e.execute_prepared(stmt, {make_key(key), make_val(key + 1), make_val(key + 1), make_val(key + 2)}).get();
                    e.execute_prepared(stmt, {make_key(key), make_val(key + 2), make_val(key + 1), make_val(key + 2)}).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);
        }, db_cfg_ptr).get();
    });
}

future<> do_with_some_data(std::vector<sstring> cf_names, std::function<future<> (cql_test_env&)> func, bool create_mvs = false, shared_ptr<db::config> db_cfg_ptr = {}) {
    co_await do_with_some_data_in_thread(cf_names, [&] (cql_test_env& e) {
        func(e).get();
    }, create_mvs, db_cfg_ptr);
}

future<> take_snapshot(cql_test_env& e, sstring ks_name = "ks", sstring cf_name = "cf", sstring snapshot_name = "test", db::snapshot_options opts = {}) {
    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, opts);
    } catch (...) {
        testlog.error("Could not take snapshot for {}.{} snapshot_name={} skip_flush={}: {}",
                ks_name, cf_name, snapshot_name, opts.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 bool is_component(const sstring& fname, const sstring& suffix) {
    return fname.ends_with(suffix);
}

static std::set<sstring> collect_sstables(const std::set<sstring>& all_files, const sstring& suffix) {
    // Verify manifest against the files in the snapshots dir
    auto pred = [&suffix] (const sstring& fname) {
        return is_component(fname, suffix);
    };
    return std::ranges::filter_view(all_files, pred) | std::ranges::to<std::set<sstring>>();
}

// Validate that the manifest.json lists exactly the SSTables present in the snapshot directory
static future<> validate_manifest(const locator::topology& topology, const fs::path& snapshot_dir, const std::set<sstring>& in_snapshot_dir, gc_clock::time_point min_time, bool tablets_enabled) {
    sstring suffix = "-TOC.txt";
    auto sstables_in_snapshot = collect_sstables(in_snapshot_dir, suffix);
    std::set<sstring> sstables_in_manifest;
    std::set<sstring> non_sstables_in_manifest;

    auto manifest_str = co_await util::read_entire_file_contiguous(snapshot_dir / "manifest.json");
    testlog.debug("manifest.json: {}", manifest_str);
    auto manifest_json = rjson::parse(manifest_str);
    BOOST_REQUIRE(manifest_json.IsObject());

    BOOST_REQUIRE(manifest_json.HasMember("manifest"));
    auto& manifest_info = manifest_json["manifest"];
    BOOST_REQUIRE(manifest_info.IsObject());
    BOOST_REQUIRE(manifest_info.HasMember("version"));
    auto& manifest_version = manifest_info["version"];
    BOOST_REQUIRE(manifest_version.IsString());
    BOOST_REQUIRE_EQUAL(manifest_version.GetString(), "1.0");
    BOOST_REQUIRE(manifest_info.HasMember("scope"));
    auto& manifest_scope = manifest_info["scope"];
    BOOST_REQUIRE(manifest_scope.IsString());
    BOOST_REQUIRE_EQUAL(manifest_scope.GetString(), "node");

    BOOST_REQUIRE(manifest_json.HasMember("node"));
    auto& node_info = manifest_json["node"];
    BOOST_REQUIRE(node_info.IsObject());
    BOOST_REQUIRE(node_info.HasMember("host_id"));
    auto& host_id_json = node_info["host_id"];
    BOOST_REQUIRE(host_id_json.IsString());
    auto id = utils::UUID(host_id_json.GetString());
    BOOST_REQUIRE_EQUAL(id, topology.my_host_id().uuid());
    BOOST_REQUIRE(node_info.HasMember("datacenter"));
    auto& datacenter = node_info["datacenter"];
    BOOST_REQUIRE(datacenter.IsString());
    BOOST_REQUIRE_EQUAL(datacenter.GetString(), topology.get_location().dc);
    BOOST_REQUIRE(node_info.HasMember("rack"));
    auto& rack = node_info["rack"];
    BOOST_REQUIRE(rack.IsString());
    BOOST_REQUIRE_EQUAL(rack.GetString(), topology.get_location().rack);

    BOOST_REQUIRE(manifest_json.HasMember("snapshot"));
    auto& manifest_snapshot = manifest_json["snapshot"];
    BOOST_REQUIRE(manifest_snapshot.IsObject());
    BOOST_REQUIRE(manifest_snapshot.HasMember("name"));
    auto& snapshot_name = manifest_snapshot["name"];
    BOOST_REQUIRE(snapshot_name.IsString());
    BOOST_REQUIRE_EQUAL(snapshot_dir.filename(), snapshot_name.GetString());
    BOOST_REQUIRE(manifest_snapshot.HasMember("created_at"));
    auto& created_at = manifest_snapshot["created_at"];
    BOOST_REQUIRE(created_at.IsNumber());
    time_t created_at_seconds = created_at.GetInt64();
    BOOST_REQUIRE_GE(created_at_seconds, min_time.time_since_epoch().count());
    BOOST_REQUIRE_LT(created_at_seconds, min_time.time_since_epoch().count() + 60);
    if (manifest_snapshot.HasMember("expires_at")) {
        BOOST_REQUIRE(created_at_seconds > 0);
        auto& expires_at = manifest_snapshot["expires_at"];
        BOOST_REQUIRE(expires_at.IsNumber());
        BOOST_REQUIRE_GE(expires_at.GetInt64(), created_at_seconds);
    }

    BOOST_REQUIRE(manifest_json.HasMember("table"));
    auto& manifest_table = manifest_json["table"];
    BOOST_REQUIRE(manifest_table.IsObject());
    BOOST_REQUIRE(manifest_table.HasMember("keyspace_name"));
    auto& manifest_table_ks_name = manifest_table["keyspace_name"];
    BOOST_REQUIRE(manifest_table_ks_name.IsString());
    BOOST_REQUIRE_EQUAL(snapshot_dir.parent_path().parent_path().parent_path().filename().native(), manifest_table_ks_name.GetString());
    auto& manifest_table_table_name = manifest_table["table_name"];
    BOOST_REQUIRE(manifest_table_table_name.IsString());
    BOOST_REQUIRE(snapshot_dir.parent_path().parent_path().filename().native().starts_with(manifest_table_table_name.GetString()));
    std::optional<sstring> tablets_type;
    if (manifest_table.HasMember("tablets_type")) {
        auto& tablets_type_json = manifest_table["tablets_type"];
        BOOST_REQUIRE(tablets_type_json.IsString());
        tablets_type = tablets_type_json.GetString();
    }
    if (tablets_enabled) {
        BOOST_REQUIRE(tablets_type.has_value());
        BOOST_REQUIRE_EQUAL(*tablets_type, "powof2");
        BOOST_REQUIRE(manifest_table.HasMember("tablet_count"));
        auto& tablet_count_json = manifest_table["tablet_count"];
        BOOST_REQUIRE(tablet_count_json.IsNumber());
        uint64_t tablet_count = tablet_count_json.GetInt64();
        BOOST_REQUIRE_EQUAL(tablet_count, 1 << log2ceil(tablet_count));
    } else {
        if (tablets_type) {
            BOOST_REQUIRE_EQUAL(*tablets_type, "none");
        }
        if (manifest_table.HasMember("tablet_count")) {
            auto& tablet_count = manifest_table["tablet_count"];
            if (!tablet_count.IsNull()) {
                BOOST_REQUIRE(tablet_count.IsNumber());
                BOOST_REQUIRE_EQUAL(tablet_count.GetInt64(), 0);
            }
        }
    }

    if (manifest_json.HasMember("sstables")) {
        auto& sstables = manifest_json["sstables"];
        BOOST_REQUIRE(sstables.IsArray());
        for (auto& sst_json : sstables.GetArray()) {
            BOOST_REQUIRE(sst_json.IsObject());

            auto& id = sst_json["id"];
            BOOST_REQUIRE(id.IsString());
            auto uuid = utils::UUID(id.GetString());
            BOOST_REQUIRE(!uuid.is_null());

            auto& toc_name = sst_json["toc_name"];
            BOOST_REQUIRE(toc_name.IsString());
            BOOST_REQUIRE(is_component(toc_name.GetString(), suffix));
            sstables_in_manifest.insert(toc_name.GetString());

            auto& data_size = sst_json["data_size"];
            BOOST_REQUIRE(data_size.IsNumber());
            auto& index_size = sst_json["index_size"];
            BOOST_REQUIRE(index_size.IsNumber());

            if (sst_json.HasMember("first_token")) {
                auto& first_token = sst_json["first_token"];
                BOOST_REQUIRE(first_token.IsNumber());

                BOOST_REQUIRE(sst_json.HasMember("last_token"));
                auto& last_token = sst_json["last_token"];
                BOOST_REQUIRE(last_token.IsNumber());
                BOOST_REQUIRE_LE(first_token.GetInt64(), last_token.GetInt64());
            } else {
                BOOST_REQUIRE(!sst_json.HasMember("last_token"));
            }
        }
    }

    if (manifest_json.HasMember("files")) {
        auto& manifest_files = manifest_json["files"];
        BOOST_REQUIRE(manifest_files.IsArray());
        for (auto& f : manifest_files.GetArray()) {
            non_sstables_in_manifest.insert(f.GetString());
        }
    }

    BOOST_REQUIRE_EQUAL(sstables_in_manifest, sstables_in_snapshot);
    BOOST_REQUIRE_EQUAL(non_sstables_in_manifest, std::set<sstring>{});
}

static future<> snapshot_works(const std::string& table_name, bool create_mvs, bool tablets_enabled = false) {
    auto db_cfg_ptr = make_shared<db::config>();
    db_cfg_ptr->tablets_mode_for_new_keyspaces(tablets_enabled ? db::tablets_mode_t::mode::enabled : db::tablets_mode_t::mode::disabled);
    return do_with_some_data_in_thread({"cf"}, [table_name] (cql_test_env& e) {
        auto min_time = gc_clock::now();
        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);

        const auto& topology = e.local_db().get_token_metadata().get_topology();
        validate_manifest(topology, snapshot_dir, in_snapshot_dir, min_time, cf.uses_tablets()).get();
    }, create_mvs, db_cfg_ptr, 100);
}

SEASTAR_TEST_CASE(table_snapshot_works) {
    return snapshot_works("cf", true, false);
}

SEASTAR_TEST_CASE(table_snapshot_works_with_tablets) {
    // FIXME: do_with_some_data does not work with views and tablets yet
    return snapshot_works("cf", false, true);
}

SEASTAR_TEST_CASE(view_snapshot_works) {
    return snapshot_works("view_cf", true, false);
}

SEASTAR_TEST_CASE(index_snapshot_works) {
    return snapshot_works(::secondary_index::index_table_name("index_cf"), true, false);
}

SEASTAR_TEST_CASE(snapshot_skip_flush_works) {
    return do_with_some_data_in_thread({"cf"}, [] (cql_test_env& e) {
        db::snapshot_options opts = {.skip_flush = true};
        take_snapshot(e, "ks", "cf", "test", opts).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"}));
    });
}

SEASTAR_TEST_CASE(snapshot_list_okay) {
    return do_with_some_data_in_thread({"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);
    });
}

SEASTAR_TEST_CASE(snapshot_list_contains_dropped_tables) {
    return do_with_some_data_in_thread({"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);
            }
        }
    });
}

SEASTAR_TEST_CASE(snapshot_list_inexistent) {
    return do_with_some_data_in_thread({"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);
    });
}


SEASTAR_TEST_CASE(clear_snapshot) {
    return do_with_some_data_in_thread({"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);
    });
}

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_in_thread({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);
    });
}

SEASTAR_TEST_CASE(clear_nonexistent_snapshot) {
    // no crashes, no exceptions
    return do_with_some_data_in_thread({"cf"}, [] (cql_test_env& e) {
        e.local_db().clear_snapshot("test", {"ks"}, "").get();
    });
}

SEASTAR_TEST_CASE(test_snapshot_ctl_details) {
    return do_with_some_data_in_thread({"cf"}, [] (cql_test_env& e) {
        sharded<db::snapshot_ctl> sc;
        sc.start(std::ref(e.db()), std::ref(e.get_storage_proxy()), 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);
    });
}

SEASTAR_TEST_CASE(test_snapshot_ctl_true_snapshots_size) {
    return do_with_some_data_in_thread({"cf"}, [] (cql_test_env& e) {
        sharded<db::snapshot_ctl> sc;
        sc.start(std::ref(e.db()), std::ref(e.get_storage_proxy()), 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);
    });
}

SEASTAR_TEST_CASE(test_snapshot_ctl_details_exception_handling) {
#ifndef SCYLLA_ENABLE_ERROR_INJECTION
    testlog.debug("Skipping test as it depends on error injection. Please run in mode where it's enabled (debug,dev).\n");
    return make_ready_future();
#endif
    return do_with_some_data_in_thread({"cf"}, [] (cql_test_env& e) {
        sharded<db::snapshot_ctl> sc;
        sc.start(std::ref(e.db()), std::ref(e.get_storage_proxy()), 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();

        utils::get_local_injector().enable("get_snapshot_details", true);
        BOOST_REQUIRE_THROW(cf.get_snapshot_details().get(), std::runtime_error);

        utils::get_local_injector().enable("per-snapshot-get_snapshot_details", true);
        BOOST_REQUIRE_THROW(cf.get_snapshot_details().get(), std::runtime_error);

        auto details = cf.get_snapshot_details().get();
        BOOST_REQUIRE_EQUAL(details.size(), 1);
    });
}

// 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_thread([] (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_thread([] (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_thread([] (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_thread([&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_thread([&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_thread([] (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", db::snapshot_options{.skip_flush = true});

        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_thread([] (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::no_gc());

        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_schema_initializer([] (schema_builder& builder) {
        if (builder.ks_name() == "test" && builder.cf_name() == "table_gc_mode_group0") {
            builder.set_is_group0_table();
        }
    });
    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_THREAD_TEST_CASE(test_tombstone_gc_state_snapshot_rf_one_tables) {
    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", "188"} }))
            .build();

    // 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_repair1, utf8_type->decompose(data_value("pk")));
    const auto dk = dht::decorate_key(*table_gc_mode_repair1, pk);

    shared_tombstone_gc_state shared_state;
    shared_state.set_table_rf_one(table_gc_mode_repair1->id());

    const auto now = gc_clock::now();
    const auto gc_state = tombstone_gc_state(shared_state).with_commitlog_check_disabled();

    BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair1, dk, now), now);
    BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair2, dk, now), gc_clock::time_point::min());

    auto snapshot = shared_state.snapshot();

    BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair1, dk, now), now);
    BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair2, dk, now), gc_clock::time_point::min());

    BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_repair1, dk, false), snapshot.query_time());
    BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_repair2, dk, false), gc_clock::time_point::min());

    shared_state.set_table_rf_n(table_gc_mode_repair1->id());
    shared_state.set_table_rf_one(table_gc_mode_repair2->id());

    BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair1, dk, now), gc_clock::time_point::min());
    BOOST_REQUIRE_EQUAL(gc_state.get_gc_before_for_key(table_gc_mode_repair2, dk, now), now);

    BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_repair1, dk, false), snapshot.query_time());
    BOOST_REQUIRE_EQUAL(snapshot.get_gc_before_for_key(table_gc_mode_repair2, dk, false), gc_clock::time_point::min());
}

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<>();
}

SEASTAR_TEST_CASE(test_tombstone_gc_rf_one) {
    return do_with_cql_env_thread([] (cql_test_env& env) {
        const std::string keyspace_name = get_name();
        const std::string table_name = "tbl";

        env.execute_cql(std::format("CREATE KEYSPACE {} WITH"
                " replication = {{'class': 'NetworkTopologyStrategy', 'datacenter1': 1}} AND"
                " tablets = {{'enabled': 'false'}}", keyspace_name)).get();
        env.execute_cql(std::format("CREATE TABLE {}.{} (pk int PRIMARY KEY)"
                " WITH compaction = {{'class': 'NullCompactionStrategy'}}", 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();

        auto& st = tbl.get_compaction_manager().get_shared_tombstone_gc_state();

        BOOST_REQUIRE(st.is_table_rf_one(schema->id()));

        auto tombstone_gc = tbl.get_tombstone_gc_state();

        const auto pk = partition_key::from_single_value(*schema, serialized(1));
        const auto dk = dht::decorate_key(*schema, pk);

        const auto now = gc_clock::now();

        // With RF=1, tombstone-gc should act as if configured in 'immediate' mode.
        BOOST_REQUIRE_EQUAL(tombstone_gc.get_gc_before_for_key(schema, dk, now), now);

        env.execute_cql(std::format("ALTER KEYSPACE {}"
                    " WITH replication = {{'class': 'NetworkTopologyStrategy', 'datacenter1': 2}}",
                    keyspace_name)).get();

        BOOST_REQUIRE(!st.is_table_rf_one(schema->id()));

        // After changing RF to > 1, tombstone-gc should revert to regular
        // repair-mode behaviour. Since there was no repair yet, this should
        // return min timepoint (no GC).
        BOOST_REQUIRE_EQUAL(tombstone_gc.get_gc_before_for_key(schema, dk, now), gc_clock::time_point::min());

        // Altering back to RF=1 should re-add the table to the rf=1 table list
        env.execute_cql(std::format("ALTER KEYSPACE {}"
                    " WITH replication = {{'class': 'NetworkTopologyStrategy', 'datacenter1': 1}}",
                    keyspace_name)).get();
        BOOST_REQUIRE(st.is_table_rf_one(schema->id()));

        // Dropping the table should clean up (removing the table from the rf=1 list).
        env.execute_cql(std::format("DROP TABLE {}.{}", keyspace_name, table_name)).get();
        BOOST_REQUIRE(!st.is_table_rf_one(schema->id()));
    });
}

SEASTAR_TEST_CASE(test_default_tombstone_gc_local_replication) {
    return do_with_cql_env_thread([] (cql_test_env& env) {
        const std::string keyspace_name = get_name();
        const std::string table_name = "tbl";

        env.execute_cql(std::format("CREATE KEYSPACE {} WITH replication = {{'class': 'LocalStrategy'}}", keyspace_name)).get();
        env.execute_cql(std::format("CREATE TABLE {}.{} (pk int PRIMARY KEY)"
                " WITH compaction = {{'class': 'NullCompactionStrategy'}}", 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();

        BOOST_REQUIRE_EQUAL(schema->tombstone_gc_options().mode(), tombstone_gc_mode::timeout);
    });
}

SEASTAR_TEST_CASE(test_flush_empty_table_waits_on_outstanding_flush) {
#ifndef SCYLLA_ENABLE_ERROR_INJECTION
    testlog.debug("Skipping test as it depends on error injection. Please run in mode where it's enabled (debug,dev).\n");
    return make_ready_future();
#endif
    return do_with_some_data_in_thread({"cf"}, [] (cql_test_env& e) {
        auto found = e.db().map_reduce0([&] (replica::database& db) -> future<bool> {
            auto& cf = db.find_column_family("ks", "cf");

            if (!cf.needs_flush()) {
                co_return false;
            }

            utils::get_local_injector().enable("flush_memtable_to_sstable_wait");

            auto flushed_0 = cf.flush();
            if (flushed_0.available()) {
                testlog.error("Table flush completed too early");
                BOOST_REQUIRE(!flushed_0.available());
            }

            if (!cf.needs_flush()) {
                testlog.error("Expecting needs_flush when waiting in flush_memtable_to_sstable_wait");
                BOOST_REQUIRE(cf.needs_flush());
            }

            // Now flush again when the active memtable is empty.
            // Expect that this waits on the ongoing flush
            auto flushed_1 = cf.flush();

            // While flush_0 is blocked, flush_1 should be blocked behind it
            if (flushed_0.available()) {
                testlog.error("First table flush expected to be blocked on injected wait");
                BOOST_REQUIRE(!flushed_0.available());
            }
            if (flushed_1.available()) {
                testlog.error("Second table flush expected to be blocked behind first flush");
                BOOST_REQUIRE(!flushed_1.available());
            }
            if (!cf.needs_flush()) {
                testlog.error("Expecting needs_flush when waiting in second flush");
                BOOST_REQUIRE(cf.needs_flush());
            }

            utils::get_local_injector().receive_message("flush_memtable_to_sstable_wait");
            co_await std::move(flushed_0);
            co_await std::move(flushed_1);
            
            if (cf.needs_flush()) {
                testlog.error("Table is not expected to need flush after flush completed");
                BOOST_REQUIRE(!cf.needs_flush());
            }

            co_return true;
        }, false, std::logical_or<bool>()).get();
        BOOST_REQUIRE(found);
    });
}

struct scoped_execption_log_level {
    scoped_execption_log_level() {
        smp::invoke_on_all([] {
            global_logger_registry().set_logger_level("exception", log_level::debug);
        }).get();
    }
    ~scoped_execption_log_level() {
        smp::invoke_on_all([] {
            global_logger_registry().set_logger_level("exception", log_level::info);
        }).get();
    }
};

SEASTAR_TEST_CASE(replica_read_timeout_no_exception) {
    cql_test_config cfg;
    cfg.db_config->reader_concurrency_semaphore_preemptive_abort_factor.set(0.0);
    const auto read_timeout = 10ms;
    const auto write_timeout = 10s;
    cfg.query_timeout.emplace(timeout_config{
            .read_timeout = read_timeout,
            .write_timeout = write_timeout,
            .range_read_timeout = read_timeout,
            .counter_write_timeout = write_timeout,
            .truncate_timeout = write_timeout,
            .cas_timeout = write_timeout,
            .other_timeout = read_timeout});

    return do_with_cql_env_thread([] (cql_test_env& e) {
        const sstring ks_name = get_name();
        const sstring tbl_name = "tbl";

        // Disable tablets because we want to exercise the legacy range-scan path too.
        // With tablets, only the table::query() path is used. Vnodes cover both.
        e.execute_cql(format("CREATE KEYSPACE {} WITH"
                    " replication = {{'class': 'NetworkTopologyStrategy', 'replication_factor': 1}} AND"
                    " tablets = {{'enabled': 'false'}}", ks_name)).get();

        e.execute_cql(format("CREATE TABLE {}.{} (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH compaction = {{'class': 'NullCompactionStrategy'}}", ks_name, tbl_name)).get();

        e.execute_cql(format("INSERT INTO {}.{} (pk, ck, v) VALUES (1, 0, 0)", ks_name, tbl_name)).get();
        replica::database::flush_table_on_all_shards(e.db(), ks_name, tbl_name).get();

        e.execute_cql(format("INSERT INTO {}.{} (pk, ck, v) VALUES (1, 1, 1)", ks_name, tbl_name)).get();
        replica::database::flush_table_on_all_shards(e.db(), ks_name, tbl_name).get();

        e.execute_cql(format("INSERT INTO {}.{} (pk, ck, v) VALUES (1, 2, 2)", ks_name, tbl_name)).get();

        // One successful read, so the auth cache is populated.
        e.execute_cql(format("SELECT * FROM {}.{} WHERE pk = 1", ks_name, tbl_name)).get();

        auto get_cxx_exceptions = [&e] {
            return e.db().map_reduce0([] (replica::database&) { return engine().cxx_exceptions(); }, 0, std::plus<size_t>()).get();
        };

        auto full_scan_id = e.prepare(format("SELECT * FROM {}.{} BYPASS CACHE", ks_name, tbl_name)).get();
        auto partition_scan_id = e.prepare(format("SELECT * FROM {}.{} WHERE pk = 1 BYPASS CACHE", ks_name, tbl_name)).get();

        auto execute_test = [&] (const char* test_name, bool scan) {
            testlog.info("Executing test: {} scan={}", test_name, scan);

            const auto cxx_exceptions_before = get_cxx_exceptions();
            const size_t num_reads = 1000;

            scoped_execption_log_level exception_log_level_scope;

            const auto query = scan ? full_scan_id  : partition_scan_id;

            testlog.info("Executing {} {}", num_reads, scan ? "full scans" : "partition scans");

            std::vector<future<>> futures;
            for (size_t i = 0; i < num_reads; ++i) {
                futures.emplace_back(e.execute_prepared(query, {}).discard_result());
            }

            // Wait for all futures and check that they all fail with a timeout exception -- without throwing any exception in the process.
            while (!futures.empty()) {
                auto f = std::move(futures.back());
                futures.pop_back();

                f.wait();
                BOOST_REQUIRE(f.failed());
                auto ex = f.get_exception();
                BOOST_REQUIRE(try_catch<exceptions::read_timeout_exception>(ex));
            }
            const auto cxx_exceptions_after = get_cxx_exceptions();

            testlog.info("Exceptions: before={}, after={}", cxx_exceptions_before, cxx_exceptions_after);

            BOOST_REQUIRE_EQUAL(cxx_exceptions_after, cxx_exceptions_before);
        };

        // Test 1: execute reads that reach the disk but time out while reading from the disk.
        // Ensure no exceptions are thrown, but the reads fail with a read timeout exception.
        if constexpr (std::is_same_v<utils::error_injection_type, utils::error_injection<true>>) {
            utils::get_local_injector().enable("sstables_mx_reader_fill_buffer_timeout", false, {{"table", format("{}.{}", ks_name, tbl_name)}});
            execute_test("disk reads", false);
            execute_test("disk reads", true);
            utils::get_local_injector().disable("sstables_mx_reader_fill_buffer_timeout");
        }

        // Test 2: execute reads that get queued on the semaphore and time out while waiting for the permit.
        // Ensure no exceptions are thrown, but the reads fail with a read timeout exception.
        {
            // Take all semaphore resources and add an active permit
            // Ensures that all new subsequent reads will be queued.
            std::vector<foreign_ptr<std::unique_ptr<reader_permit>>> dummy_permits;
            for (shard_id shard = 0; shard < smp::count; ++shard) {
                dummy_permits.emplace_back();
            }
            e.db().invoke_on_all([&] (replica::database& db) -> future<> {
                auto& sem = db.get_reader_concurrency_semaphore();
                sem.set_resources({1, 1});
                dummy_permits[this_shard_id()] = std::make_unique<reader_permit>(co_await sem.obtain_permit(
                    db.find_column_family(ks_name, tbl_name).schema(),
                    "dummy",
                    1,
                    db::no_timeout,
                    {}));
            }).get();

            execute_test("queued reads", false);
            execute_test("queued reads", true);
        }
    }, cfg);
}

BOOST_AUTO_TEST_SUITE_END()