mirrors/scylladb
1
0
Fork 0
mirror of https://github.com/scylladb/scylladb.git synced 2026-06-01 12:36:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge "tests: introduce test environment and cleanup sstable tests" from Benny

Browse Source 
"
As part of implementing sstables manager and fixing issue related
to updating large_data_handler on all delete paths, we want to funnel
all sstable creations, loading, and deletions through a manager.

The patchset lays out test infrastructure to funnel these opeations
through class sstables::test_env.

In the process, it cleans up many numerous call sites in the existing
unit tests that evolved over time.

Refs #4198
Refs #4149

Tests: unit (dev)
"

* 'projects/test_env/v3' of https://github.com/bhalevy/scylla:
  tests: introduce sstables::test_env
  tests: perf_sstable: rename test_env
  tests: sstable_datafile_test: use useable_sst
  tests: sstable_test: add write_and_validate_sst helper
  tests: sstable_test: add test_using_reusable_sst helper
  tests: sstable_test: use reusable_sst where possible
  tests: sstable_test: add test_using_working_sst helper
  tests: sstable_3_x_test: make_test_sstable
  tests: run_sstable_resharding_test: use default parameters to make_sstable
  tests: sstables::test::make_test_sstable: reorder params
  tests: test_setup: do_with_test_directory is unused
  tests: move sstable_resharding_strategy_tests to sstable_reharding_test
  tests: move create_token_from_key helpers to test_services
  tests: move column_family_for_tests to test_services
  dht: move declaration of default_partitioner from sstable_datafile_test to i_partitioner.hh
This commit is contained in:
Avi Kivity
2019-02-19 11:26:42 +02:00
parent 9c6f897731 c8f239ff2b
commit 934ba7ccb2
16 changed files with 861 additions and 678 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
configure.py
View File

@@ -791,6 +791,7 @@ scylla_tests_dependencies = scylla_core + idls + [
'tests/result_set_assertions.cc',
'tests/mutation_source_test.cc',
'tests/data_model.cc',
'tests/test_services.cc',
]
deps = {

2
dht/i_partitioner.hh
View File

@@ -808,6 +808,8 @@ public:
std::unique_ptr<dht::i_partitioner> make_partitioner(sstring name, unsigned shard_count, unsigned sharding_ignore_msb_bits);
extern std::unique_ptr<i_partitioner> default_partitioner;
} // dht
namespace std {

6
tests/broken_sstable_test.cc
View File

@@ -40,7 +40,8 @@ struct my_consumer {
static void broken_sst(sstring dir, unsigned long generation, schema_ptr s, sstring msg,
sstable_version_types version = la) {
try {
sstable_ptr sstp = std::get<0>(reusable_sst(s, dir, generation, version).get());
sstables::test_env env;
sstable_ptr sstp = std::get<0>(env.reusable_sst(s, dir, generation, version).get());
auto r = sstp->read_rows_flat(s);
r.consume(my_consumer{}, db::no_timeout).get();
BOOST_FAIL("expecting exception");
@@ -63,7 +64,8 @@ SEASTAR_THREAD_TEST_CASE(test_empty_index) {
.with_column("val", int32_type)
.set_compressor_params(compression_parameters::no_compression())
.build();
sstable_ptr sstp = std::get<0>(reusable_sst(s, "tests/sstables/empty_index", 36, sstable_version_types::mc).get());
sstables::test_env env;
sstable_ptr sstp = std::get<0>(env.reusable_sst(s, "tests/sstables/empty_index", 36, sstable_version_types::mc).get());
sstp->load().get();
auto fut = sstables::test(sstp).read_indexes();
BOOST_REQUIRE_EXCEPTION(fut.get(), malformed_sstable_exception, [](auto&& e) {

4
tests/memory_footprint.cc
View File

@@ -35,6 +35,7 @@
#include "canonical_mutation.hh"
#include "sstable_utils.hh"
#include "test_services.hh"
#include "sstable_test_env.hh"
class size_calculator {
class nest {
@@ -187,7 +188,8 @@ static sizes calculate_sizes(const mutation& m) {
result.query_result = m.query(partition_slice_builder(*s).build(), query::result_options::only_result()).buf().size();
tmpdir sstable_dir;
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
sstable_dir.path().string(),
1 /* generation */,
sstables::sstable::version_types::la,

38
tests/mutation_reader_test.cc
View File

@@ -35,6 +35,7 @@
#include "tests/flat_mutation_reader_assertions.hh"
#include "tests/tmpdir.hh"
#include "tests/sstable_utils.hh"
#include "tests/sstable_test.hh"
#include "tests/simple_schema.hh"
#include "tests/test_services.hh"
#include "tests/mutation_source_test.hh"
@@ -384,6 +385,7 @@ SEASTAR_TEST_CASE(test_sm_fast_forwarding_combining_reader) {
}
struct sst_factory {
sstables::test_env env;
schema_ptr s;
sstring path;
unsigned gen;
@@ -397,7 +399,7 @@ struct sst_factory {
{}
sstables::shared_sstable operator()() {
auto sst = sstables::make_sstable(s, path, gen, sstables::sstable::version_types::la, sstables::sstable::format_types::big);
auto sst = env.make_sstable(s, path, gen, sstables::sstable::version_types::la, sstables::sstable::format_types::big);
sst->set_unshared();
//sst->set_sstable_level(level);
sst->get_metadata_collector().sstable_level(level);
@@ -707,7 +709,7 @@ SEASTAR_TEST_CASE(reader_selector_fast_forwarding_test) {
static const std::size_t new_reader_base_cost{16 * 1024};
sstables::shared_sstable create_sstable(simple_schema& sschema, const sstring& path) {
sstables::shared_sstable create_sstable(sstables::test_env& env, simple_schema& sschema, const sstring& path) {
std::vector<mutation> mutations;
mutations.reserve(1 << 14);
@@ -724,17 +726,17 @@ sstables::shared_sstable create_sstable(simple_schema& sschema, const sstring& p
}
return make_sstable_containing([&] {
return make_lw_shared<sstables::sstable>(sschema.schema(), path, 0, sstables::sstable::version_types::la, sstables::sstable::format_types::big);
return env.make_sstable(sschema.schema(), path, 0, sstables::sstable::version_types::la, sstables::sstable::format_types::big);
}
, mutations);
}
static
sstables::shared_sstable create_sstable(schema_ptr s, std::vector<mutation> mutations) {
sstables::shared_sstable create_sstable(sstables::test_env& env, schema_ptr s, std::vector<mutation> mutations) {
static thread_local auto tmp = tmpdir();
static int gen = 0;
return make_sstable_containing([&] {
return make_lw_shared<sstables::sstable>(s, tmp.path().string(), gen++, sstables::sstable::version_types::la, sstables::sstable::format_types::big);
return env.make_sstable(s, tmp.path().string(), gen++, sstables::sstable::version_types::la, sstables::sstable::format_types::big);
}, mutations);
}
@@ -973,14 +975,14 @@ SEASTAR_TEST_CASE(reader_restriction_file_tracking) {
}
SEASTAR_TEST_CASE(restricted_reader_reading) {
return async([&] {
return sstables::test_env::do_with_async([&] (sstables::test_env& env) {
storage_service_for_tests ssft;
reader_concurrency_semaphore semaphore(2, new_reader_base_cost);
{
simple_schema s;
auto tmp = tmpdir();
auto sst = create_sstable(s, tmp.path().string());
auto sst = create_sstable(env, s, tmp.path().string());
auto reader1 = reader_wrapper(semaphore, s.schema(), sst);
@@ -1044,14 +1046,14 @@ SEASTAR_TEST_CASE(restricted_reader_reading) {
}
SEASTAR_TEST_CASE(restricted_reader_timeout) {
return async([&] {
return sstables::test_env::do_with_async([&] (sstables::test_env& env) {
storage_service_for_tests ssft;
reader_concurrency_semaphore semaphore(2, new_reader_base_cost);
{
simple_schema s;
auto tmp = tmpdir();
auto sst = create_sstable(s, tmp.path().string());
auto sst = create_sstable(env, s, tmp.path().string());
auto timeout = std::chrono::duration_cast<db::timeout_clock::time_point::duration>(std::chrono::milliseconds{1});
@@ -1089,7 +1091,7 @@ SEASTAR_TEST_CASE(restricted_reader_timeout) {
}
SEASTAR_TEST_CASE(restricted_reader_max_queue_length) {
return async([&] {
return sstables::test_env::do_with_async([&] (sstables::test_env& env) {
storage_service_for_tests ssft;
struct queue_overloaded_exception {};
@@ -1099,7 +1101,7 @@ SEASTAR_TEST_CASE(restricted_reader_max_queue_length) {
{
simple_schema s;
auto tmp = tmpdir();
auto sst = create_sstable(s, tmp.path().string());
auto sst = create_sstable(env, s, tmp.path().string());
auto reader1_ptr = std::make_unique<reader_wrapper>(semaphore, s.schema(), sst);
(*reader1_ptr)().get();
@@ -1128,14 +1130,14 @@ SEASTAR_TEST_CASE(restricted_reader_max_queue_length) {
}
SEASTAR_TEST_CASE(restricted_reader_create_reader) {
return async([&] {
return sstables::test_env::do_with_async([&] (sstables::test_env& env) {
storage_service_for_tests ssft;
reader_concurrency_semaphore semaphore(100, new_reader_base_cost);
{
simple_schema s;
auto tmp = tmpdir();
auto sst = create_sstable(s, tmp.path().string());
auto sst = create_sstable(env, s, tmp.path().string());
{
auto reader = reader_wrapper(semaphore, s.schema(), sst);
@@ -1196,7 +1198,7 @@ static mutation compacted(const mutation& m) {
}
SEASTAR_TEST_CASE(test_fast_forwarding_combined_reader_is_consistent_with_slicing) {
return async([&] {
return sstables::test_env::do_with_async([&] (sstables::test_env& env) {
storage_service_for_tests ssft;
random_mutation_generator gen(random_mutation_generator::generate_counters::no);
auto s = gen.schema();
@@ -1219,7 +1221,7 @@ SEASTAR_TEST_CASE(test_fast_forwarding_combined_reader_is_consistent_with_slicin
combined[j++].apply(m);
}
}
mutation_source ds = create_sstable(s, muts)->as_mutation_source();
mutation_source ds = create_sstable(env, s, muts)->as_mutation_source();
readers.push_back(ds.make_reader(s,
dht::partition_range::make({keys[0]}, {keys[0]}),
s->full_slice(), default_priority_class(), nullptr,
@@ -1271,7 +1273,7 @@ SEASTAR_TEST_CASE(test_fast_forwarding_combined_reader_is_consistent_with_slicin
}
SEASTAR_TEST_CASE(test_combined_reader_slicing_with_overlapping_range_tombstones) {
return async([&] {
return sstables::test_env::do_with_async([&] (sstables::test_env& env) {
storage_service_for_tests ssft;
simple_schema ss;
auto s = ss.schema();
@@ -1287,8 +1289,8 @@ SEASTAR_TEST_CASE(test_combined_reader_slicing_with_overlapping_range_tombstones
std::vector<flat_mutation_reader> readers;
mutation_source ds1 = create_sstable(s, {m1})->as_mutation_source();
mutation_source ds2 = create_sstable(s, {m2})->as_mutation_source();
mutation_source ds1 = create_sstable(env, s, {m1})->as_mutation_source();
mutation_source ds2 = create_sstable(env, s, {m2})->as_mutation_source();
// upper bound ends before the row in m2, so that the raw is fetched after next fast forward.
auto range = ss.make_ckey_range(0, 3);

26
tests/perf/perf_sstable.cc
View File

@@ -36,34 +36,34 @@ using namespace sstables;
static unsigned iterations = 30;
static unsigned parallelism = 1;
future<> test_write(distributed<test_env>& dt) {
future<> test_write(distributed<perf_sstable_test_env>& dt) {
return seastar::async([&dt] {
storage_service_for_tests ssft;
dt.invoke_on_all([] (test_env &t) {
dt.invoke_on_all([] (perf_sstable_test_env &t) {
return t.fill_memtable();
}).then([&dt] {
return time_runs(iterations, parallelism, dt, &test_env::flush_memtable);
return time_runs(iterations, parallelism, dt, &perf_sstable_test_env::flush_memtable);
}).get();
});
}
future<> test_compaction(distributed<test_env>& dt) {
future<> test_compaction(distributed<perf_sstable_test_env>& dt) {
return seastar::async([&dt] {
storage_service_for_tests ssft;
dt.invoke_on_all([] (test_env &t) {
dt.invoke_on_all([] (perf_sstable_test_env &t) {
return t.fill_memtable();
}).then([&dt] {
return time_runs(iterations, parallelism, dt, &test_env::compaction);
return time_runs(iterations, parallelism, dt, &perf_sstable_test_env::compaction);
}).get();
});
}
future<> test_index_read(distributed<test_env>& dt) {
return time_runs(iterations, parallelism, dt, &test_env::read_all_indexes);
future<> test_index_read(distributed<perf_sstable_test_env>& dt) {
return time_runs(iterations, parallelism, dt, &perf_sstable_test_env::read_all_indexes);
}
future<> test_sequential_read(distributed<test_env>& dt) {
return time_runs(iterations, parallelism, dt, &test_env::read_sequential_partitions);
future<> test_sequential_read(distributed<perf_sstable_test_env>& dt) {
return time_runs(iterations, parallelism, dt, &perf_sstable_test_env::read_sequential_partitions);
}
enum class test_modes {
@@ -98,9 +98,9 @@ int main(int argc, char** argv) {
("testdir", bpo::value<sstring>()->default_value("/var/lib/scylla/perf-tests"), "directory in which to store the sstables");
return app.run_deprecated(argc, argv, [&app] {
auto test = make_lw_shared<distributed<test_env>>();
auto test = make_lw_shared<distributed<perf_sstable_test_env>>();
auto cfg = test_env::conf();
auto cfg = perf_sstable_test_env::conf();
iterations = app.configuration()["iterations"].as<unsigned>();
parallelism = app.configuration()["parallelism"].as<unsigned>();
cfg.partitions = app.configuration()["partitions"].as<unsigned>();
@@ -121,7 +121,7 @@ int main(int argc, char** argv) {
engine().at_exit([test] { return test->stop(); });
if ((mode == test_modes::index_read) ||
(mode == test_modes::sequential_read)) {
return test->invoke_on_all([] (test_env &t) {
return test->invoke_on_all([] (perf_sstable_test_env &t) {
return t.load_sstables(iterations);
}).then_wrapped([] (future<> f) {
try {

30
tests/perf/perf_sstable.hh
View File

@@ -34,7 +34,9 @@ using namespace sstables;
static db::nop_large_data_handler nop_lp_handler;
class test_env {
class perf_sstable_test_env {
test_env _env;
public:
struct conf {
unsigned partitions;
@@ -99,7 +101,7 @@ private:
}
public:
test_env(conf cfg) : _cfg(std::move(cfg))
perf_sstable_test_env(conf cfg) : _cfg(std::move(cfg))
, s(create_schema())
, _distribution('@', '~')
, _mt(make_lw_shared<memtable>(s))
@@ -122,7 +124,7 @@ public:
}
future<> load_sstables(unsigned iterations) {
_sst.push_back(make_sstable(s, this->dir(), 0, sstable::version_types::ka, sstable::format_types::big));
_sst.push_back(_env.make_sstable(s, this->dir(), 0, sstable::version_types::ka, sstable::format_types::big));
return _sst.back()->load();
}
@@ -138,11 +140,11 @@ public:
size_t partitions = _mt->partition_count();
test_setup::create_empty_test_dir(dir()).get();
auto sst = sstables::test::make_test_sstable(_cfg.buffer_size, s, dir(), idx, sstable::version_types::ka, sstable::format_types::big);
auto sst = _env.make_sstable(s, dir(), idx, sstable::version_types::ka, sstable::format_types::big, _cfg.buffer_size);
auto start = test_env::now();
auto start = perf_sstable_test_env::now();
write_memtable_to_sstable_for_test(*_mt, sst).get();
auto end = test_env::now();
auto end = perf_sstable_test_env::now();
_mt->revert_flushed_memory();
@@ -155,7 +157,7 @@ public:
return test_setup::create_empty_test_dir(dir()).then([this, idx] {
return seastar::async([this, idx] {
auto sst_gen = [this, gen = make_lw_shared<unsigned>(idx)] () mutable {
return sstables::test::make_test_sstable(_cfg.buffer_size, s, dir(), (*gen)++, sstable::version_types::ka, sstable::format_types::big);
return _env.make_sstable(s, dir(), (*gen)++, sstable::version_types::ka, sstable::format_types::big, _cfg.buffer_size);
};
std::vector<shared_sstable> ssts;
@@ -174,9 +176,9 @@ public:
cfg.large_data_handler = &nop_lp_handler;
auto cf = make_lw_shared<column_family>(s, cfg, column_family::no_commitlog(), *cm, cl_stats, tracker);
auto start = test_env::now();
auto start = perf_sstable_test_env::now();
auto ret = sstables::compact_sstables(sstables::compaction_descriptor(std::move(ssts)), *cf, sst_gen, sstables::replacer_fn_no_op()).get0();
auto end = test_env::now();
auto end = perf_sstable_test_env::now();
auto partitions_per_sstable = _cfg.partitions / _cfg.sstables;
assert(ret.total_keys_written == partitions_per_sstable);
@@ -189,10 +191,10 @@ public:
future<double> read_all_indexes(int idx) {
return do_with(test(_sst[0]), [] (auto& sst) {
const auto start = test_env::now();
const auto start = perf_sstable_test_env::now();
return sst.read_indexes().then([start] (const auto& indexes) {
auto end = test_env::now();
auto end = perf_sstable_test_env::now();
auto duration = std::chrono::duration<double>(end - start).count();
return indexes.size() / duration;
});
@@ -201,7 +203,7 @@ public:
future<double> read_sequential_partitions(int idx) {
return do_with(_sst[0]->read_rows_flat(s), [this] (flat_mutation_reader& r) {
auto start = test_env::now();
auto start = perf_sstable_test_env::now();
auto total = make_lw_shared<size_t>(0);
auto done = make_lw_shared<bool>(false);
return do_until([done] { return *done; }, [this, done, total, &r] {
@@ -218,7 +220,7 @@ public:
}
});
}).then([total, start] {
auto end = test_env::now();
auto end = perf_sstable_test_env::now();
auto duration = std::chrono::duration<double>(end - start).count();
return *total / duration;
});
@@ -229,7 +231,7 @@ public:
// The function func should carry on with the test, and return the number of partitions processed.
// time_runs will then map reduce it, and return the aggregate partitions / sec for the whole system.
template <typename Func>
future<> time_runs(unsigned iterations, unsigned parallelism, distributed<test_env>& dt, Func func) {
future<> time_runs(unsigned iterations, unsigned parallelism, distributed<perf_sstable_test_env>& dt, Func func) {
using namespace boost::accumulators;
auto acc = make_lw_shared<accumulator_set<double, features<tag::mean, tag::error_of<tag::mean>>>>();
auto idx = boost::irange(0, int(iterations));

95
tests/sstable_3_x_test.cc
View File

@@ -37,7 +37,6 @@
#include "schema_builder.hh"
#include "sstable_test.hh"
#include "flat_mutation_reader_assertions.hh"
#include "sstable_test.hh"
#include "tests/test_services.hh"
#include "tests/tmpdir.hh"
#include "tests/sstable_utils.hh"
@@ -56,18 +55,22 @@
using namespace sstables;
class sstable_assertions final {
test_env _env;
shared_sstable _sst;
public:
sstable_assertions(schema_ptr schema, const sstring& path, int generation = 1)
: _sst(make_sstable(std::move(schema),
: _env()
, _sst(_env.make_sstable(std::move(schema),
path,
generation,
sstable_version_types::mc,
sstable_format_types::big,
gc_clock::now(),
default_io_error_handler_gen(),
1))
{ }
test_env& get_env() {
return _env;
}
void read_toc() {
_sst->read_toc().get();
}
@@ -2855,30 +2858,20 @@ SEASTAR_THREAD_TEST_CASE(test_uncompressed_collections_read) {
.produces_end_of_stream();
}
static sstables::shared_sstable open_sstable(schema_ptr schema, sstring dir, unsigned long generation) {
auto sst = sstables::make_sstable(std::move(schema), dir, generation,
sstables::sstable::version_types::mc,
sstables::sstable::format_types::big);
sst->load().get();
return sst;
static sstables::shared_sstable open_sstable(test_env& env, schema_ptr schema, sstring dir, unsigned long generation) {
return env.reusable_sst(std::move(schema), dir, generation, sstables::sstable::version_types::mc).get0();
}
static std::vector<sstables::shared_sstable> open_sstables(schema_ptr s, sstring dir, std::vector<unsigned long> generations) {
static std::vector<sstables::shared_sstable> open_sstables(test_env& env, schema_ptr s, sstring dir, std::vector<unsigned long> generations) {
std::vector<sstables::shared_sstable> result;
for(auto generation: generations) {
result.push_back(open_sstable(s, dir, generation));
result.push_back(open_sstable(env, s, dir, generation));
}
return result;
}
static flat_mutation_reader compacted_sstable_reader(schema_ptr s,
static flat_mutation_reader compacted_sstable_reader(test_env& env, schema_ptr s,
sstring table_name, std::vector<unsigned long> generations) {
auto column_family_test_config = [] {
static db::nop_large_data_handler nop_lp_handler;
column_family::config cfg;
cfg.large_data_handler = &nop_lp_handler;
return cfg;
};
storage_service_for_tests ssft;
auto cm = make_lw_shared<compaction_manager>();
@@ -2889,16 +2882,16 @@ static flat_mutation_reader compacted_sstable_reader(schema_ptr s,
lw_shared_ptr<memtable> mt = make_lw_shared<memtable>(s);
tmpdir tmp;
auto sstables = open_sstables(s, format("tests/sstables/3.x/uncompressed/{}", table_name), generations);
auto sstables = open_sstables(env, s, format("tests/sstables/3.x/uncompressed/{}", table_name), generations);
auto new_generation = generations.back() + 1;
auto new_sstable = [s, &tmp, new_generation] {
return sstables::test::make_test_sstable(4096, s, tmp.path().string(), new_generation,
sstables::sstable_version_types::mc, sstable::format_types::big);
auto new_sstable = [s, &tmp, &env, new_generation] {
return env.make_sstable(s, tmp.path().string(), new_generation,
sstables::sstable_version_types::mc, sstable::format_types::big, 4096);
};
sstables::compact_sstables(sstables::compaction_descriptor(std::move(sstables)), *cf, new_sstable, replacer_fn_no_op()).get();
auto compacted_sst = open_sstable(s, tmp.path().string(), new_generation);
auto compacted_sst = open_sstable(env, s, tmp.path().string(), new_generation);
return compacted_sst->as_mutation_source().make_reader(s, query::full_partition_range, s->full_slice());
}
@@ -2955,7 +2948,8 @@ SEASTAR_THREAD_TEST_CASE(compact_deleted_row) {
* }
* ]
*/
auto reader = compacted_sstable_reader(s, table_name, {1, 2});
test_env env;
auto reader = compacted_sstable_reader(env, s, table_name, {1, 2});
mutation_opt m = read_mutation_from_flat_mutation_reader(reader, db::no_timeout).get0();
BOOST_REQUIRE(m);
BOOST_REQUIRE(m->key().equal(*s, partition_key::from_singular(*s, data_value(sstring("key")))));
@@ -3025,7 +3019,8 @@ SEASTAR_THREAD_TEST_CASE(compact_deleted_cell) {
*]
*
*/
auto reader = compacted_sstable_reader(s, table_name, {1, 2});
test_env env;
auto reader = compacted_sstable_reader(env, s, table_name, {1, 2});
mutation_opt m = read_mutation_from_flat_mutation_reader(reader, db::no_timeout).get0();
BOOST_REQUIRE(m);
BOOST_REQUIRE(m->key().equal(*s, partition_key::from_singular(*s, data_value(sstring("key")))));
@@ -3071,11 +3066,11 @@ static void compare_sstables(const seastar::compat::filesystem::path& result_pat
}
}
static tmpdir write_sstables(schema_ptr s, lw_shared_ptr<memtable> mt1, lw_shared_ptr<memtable> mt2) {
static tmpdir write_sstables(test_env& env, schema_ptr s, lw_shared_ptr<memtable> mt1, lw_shared_ptr<memtable> mt2) {
static db::nop_large_data_handler nop_lp_handler;
storage_service_for_tests ssft;
tmpdir tmp;
auto sst = sstables::test::make_test_sstable(4096, s, tmp.path().string(), 1, sstables::sstable_version_types::mc, sstable::format_types::big);
auto sst = env.make_sstable(s, tmp.path().string(), 1, sstables::sstable_version_types::mc, sstable::format_types::big, 4096);
sstable_writer_config cfg;
cfg.large_data_handler = &nop_lp_handler;
sst->write_components(make_combined_reader(s,
@@ -3088,27 +3083,29 @@ static tmpdir write_sstables(schema_ptr s, lw_shared_ptr<memtable> mt1, lw_share
// that otherwise takes place for RTs put into one and the same memtable
static tmpdir write_and_compare_sstables(schema_ptr s, lw_shared_ptr<memtable> mt1, lw_shared_ptr<memtable> mt2,
sstring table_name) {
auto tmp = write_sstables(std::move(s), std::move(mt1), std::move(mt2));
test_env env;
auto tmp = write_sstables(env, std::move(s), std::move(mt1), std::move(mt2));
compare_sstables(tmp.path(), table_name);
return tmp;
}
static tmpdir write_sstables(schema_ptr s, lw_shared_ptr<memtable> mt) {
static tmpdir write_sstables(test_env& env, schema_ptr s, lw_shared_ptr<memtable> mt) {
storage_service_for_tests ssft;
tmpdir tmp;
auto sst = sstables::test::make_test_sstable(4096, s, tmp.path().string(), 1, sstables::sstable_version_types::mc, sstable::format_types::big);
auto sst = env.make_sstable(s, tmp.path().string(), 1, sstables::sstable_version_types::mc, sstable::format_types::big, 4096);
write_memtable_to_sstable_for_test(*mt, sst).get();
return tmp;
}
static tmpdir write_and_compare_sstables(schema_ptr s, lw_shared_ptr<memtable> mt, sstring table_name) {
auto tmp = write_sstables(std::move(s), std::move(mt));
test_env env;
auto tmp = write_sstables(env, std::move(s), std::move(mt));
compare_sstables(tmp.path(), table_name);
return tmp;
}
static sstable_assertions validate_read(schema_ptr s, const seastar::compat::filesystem::path& path, std::vector<mutation> mutations) {
sstable_assertions sst(s, path.string());
sstable_assertions sst(s, path.string(), 1);
sst.load();
auto assertions = assert_that(sst.read_rows_flat());
@@ -3124,8 +3121,7 @@ static constexpr api::timestamp_type write_timestamp = 1525385507816568;
static constexpr gc_clock::time_point write_time_point = gc_clock::time_point{} + gc_clock::duration{1525385507};
static void validate_stats_metadata(schema_ptr s, sstable_assertions written_sst, sstring table_name) {
auto orig_sst = sstables::make_sstable(s, get_write_test_path(table_name), 1, sstable_version_types::mc, big);
orig_sst->load().get();
auto orig_sst = written_sst.get_env().reusable_sst(s, get_write_test_path(table_name), 1, sstable_version_types::mc).get0();
const auto& orig_stats = orig_sst->get_stats_metadata();
const auto& written_stats = written_sst.get_stats_metadata();
@@ -3520,7 +3516,8 @@ static void test_write_many_partitions(sstring table_name, tombstone partition_t
}
bool compressed = cp.get_compressor() != nullptr;
tmpdir tmp = compressed ? write_sstables(s, mt) : write_and_compare_sstables(s, mt, table_name);
test_env env;
tmpdir tmp = compressed ? write_sstables(env, s, mt) : write_and_compare_sstables(s, mt, table_name);
boost::sort(muts, mutation_decorated_key_less_comparator());
validate_read(s, tmp.path(), muts);
}
@@ -4406,6 +4403,10 @@ static std::unique_ptr<index_reader> get_index_reader(shared_sstable sst) {
return std::make_unique<index_reader>(sst, default_priority_class());
}
shared_sstable make_test_sstable(test_env& env, schema_ptr schema, const sstring& table_name, int64_t gen = 1) {
return env.reusable_sst(schema, get_read_index_test_path(table_name), gen, sstable_version_types::mc).get0();
}
/*
* The SSTables read is generated using the following queries:
*
@@ -4421,8 +4422,8 @@ SEASTAR_THREAD_TEST_CASE(test_read_empty_index) {
builder.set_compressor_params(compression_parameters::no_compression());
schema_ptr s = builder.build(schema_builder::compact_storage::no);
auto sst = sstables::make_sstable(s, get_read_index_test_path(table_name), 1, sstable_version_types::mc , sstable_format_types::big);
sst->load().get0();
test_env env;
auto sst = make_test_sstable(env, s, table_name);
assert_that(get_index_reader(sst)).is_empty(*s);
}
@@ -4441,8 +4442,8 @@ SEASTAR_THREAD_TEST_CASE(test_read_rows_only_index) {
builder.set_compressor_params(compression_parameters::no_compression());
schema_ptr s = builder.build(schema_builder::compact_storage::no);
auto sst = sstables::make_sstable(s, get_read_index_test_path(table_name), 1, sstable_version_types::mc , sstable_format_types::big);
sst->load().get0();
test_env env;
auto sst = make_test_sstable(env, s, table_name);
assert_that(get_index_reader(sst)).has_monotonic_positions(*s);
}
@@ -4460,8 +4461,8 @@ SEASTAR_THREAD_TEST_CASE(test_read_range_tombstones_only_index) {
builder.set_compressor_params(compression_parameters::no_compression());
schema_ptr s = builder.build(schema_builder::compact_storage::no);
auto sst = sstables::make_sstable(s, get_read_index_test_path(table_name), 1, sstable_version_types::mc , sstable_format_types::big);
sst->load().get0();
test_env env;
auto sst = make_test_sstable(env, s, table_name);
assert_that(get_index_reader(sst)).has_monotonic_positions(*s);
}
@@ -4487,8 +4488,8 @@ SEASTAR_THREAD_TEST_CASE(test_read_range_tombstone_boundaries_index) {
builder.set_compressor_params(compression_parameters::no_compression());
schema_ptr s = builder.build(schema_builder::compact_storage::no);
auto sst = sstables::make_sstable(s, get_read_index_test_path(table_name), 1, sstable_version_types::mc , sstable_format_types::big);
sst->load().get0();
test_env env;
auto sst = make_test_sstable(env, s, table_name);
assert_that(get_index_reader(sst)).has_monotonic_positions(*s);
}
@@ -4945,7 +4946,8 @@ SEASTAR_THREAD_TEST_CASE(test_sstable_reader_on_unknown_column) {
sstable_writer_config cfg;
cfg.promoted_index_block_size = index_block_size;
cfg.large_data_handler = &nop_lp_handler;
auto sst = sstables::make_sstable(write_schema,
test_env env;
auto sst = env.make_sstable(write_schema,
dir.path().string(),
1 /* generation */,
sstable_version_types::mc,
@@ -4998,7 +5000,8 @@ struct large_row_handler : public db::large_data_handler {
static void test_sstable_write_large_row_f(schema_ptr s, memtable& mt, const partition_key& pk,
std::vector<clustering_key*> expected, uint64_t threshold) {
tmpdir dir;
auto sst = sstables::make_sstable(
test_env env;
auto sst = env.make_sstable(
s, dir.path().string(), 1 /* generation */, sstable_version_types::mc, sstables::sstable::format_types::big);
unsigned i = 0;

736
tests/sstable_datafile_test.cc
View File

File diff suppressed because it is too large Load Diff

142
tests/sstable_mutation_test.cc
View File

@@ -53,7 +53,8 @@ static db::nop_large_data_handler nop_lp_handler;
SEASTAR_THREAD_TEST_CASE(nonexistent_key) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([] (auto sstp) {
sstables::test_env env;
env.reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([] (auto sstp) {
return do_with(make_dkey(uncompressed_schema(), "invalid_key"), [sstp] (auto& key) {
auto s = uncompressed_schema();
auto rd = make_lw_shared<flat_mutation_reader>(sstp->read_row_flat(s, key));
@@ -65,8 +66,8 @@ SEASTAR_THREAD_TEST_CASE(nonexistent_key) {
}).get();
}
future<> test_no_clustered(bytes&& key, std::unordered_map<bytes, data_value> &&map) {
return reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([k = std::move(key), map = std::move(map)] (auto sstp) mutable {
future<> test_no_clustered(sstables::test_env& env, bytes&& key, std::unordered_map<bytes, data_value> &&map) {
return env.reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([k = std::move(key), map = std::move(map)] (auto sstp) mutable {
return do_with(make_dkey(uncompressed_schema(), std::move(k)), [sstp, map = std::move(map)] (auto& key) {
auto s = uncompressed_schema();
auto rd = make_lw_shared<flat_mutation_reader>(sstp->read_row_flat(s, key));
@@ -90,22 +91,26 @@ future<> test_no_clustered(bytes&& key, std::unordered_map<bytes, data_value> &&
SEASTAR_THREAD_TEST_CASE(uncompressed_1) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
test_no_clustered("vinna", {{ "col1", to_sstring("daughter") }, { "col2", 3 }}).get();
sstables::test_env env;
test_no_clustered(env, "vinna", {{ "col1", to_sstring("daughter") }, { "col2", 3 }}).get();
}
SEASTAR_THREAD_TEST_CASE(uncompressed_2) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
test_no_clustered("gustaf", {{ "col1", to_sstring("son") }, { "col2", 0 }}).get();
sstables::test_env env;
test_no_clustered(env, "gustaf", {{ "col1", to_sstring("son") }, { "col2", 0 }}).get();
}
SEASTAR_THREAD_TEST_CASE(uncompressed_3) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
test_no_clustered("isak", {{ "col1", to_sstring("son") }, { "col2", 1 }}).get();
sstables::test_env env;
test_no_clustered(env, "isak", {{ "col1", to_sstring("son") }, { "col2", 1 }}).get();
}
SEASTAR_THREAD_TEST_CASE(uncompressed_4) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
test_no_clustered("finna", {{ "col1", to_sstring("daughter") }, { "col2", 2 }}).get();
sstables::test_env env;
test_no_clustered(env, "finna", {{ "col1", to_sstring("daughter") }, { "col2", 2 }}).get();
}
/*
@@ -136,8 +141,8 @@ SEASTAR_THREAD_TEST_CASE(uncompressed_4) {
// FIXME: we are lacking a full deletion test
template <int Generation>
future<mutation> generate_clustered(bytes&& key) {
return reusable_sst(complex_schema(), "tests/sstables/complex", Generation).then([k = std::move(key)] (auto sstp) mutable {
future<mutation> generate_clustered(sstables::test_env& env, bytes&& key) {
return env.reusable_sst(complex_schema(), "tests/sstables/complex", Generation).then([k = std::move(key)] (auto sstp) mutable {
return do_with(make_dkey(complex_schema(), std::move(k)), [sstp] (auto& key) {
auto s = complex_schema();
auto rd = make_lw_shared<flat_mutation_reader>(sstp->read_row_flat(s, key));
@@ -157,7 +162,8 @@ inline auto clustered_row(mutation& mutation, const schema& s, std::vector<bytes
SEASTAR_THREAD_TEST_CASE(complex_sst1_k1) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
generate_clustered<1>("key1").then([] (auto&& mutation) {
sstables::test_env env;
generate_clustered<1>(env, "key1").then([] (auto&& mutation) {
auto s = complex_schema();
auto& sr = mutation.partition().static_row();
@@ -187,7 +193,8 @@ SEASTAR_THREAD_TEST_CASE(complex_sst1_k1) {
SEASTAR_THREAD_TEST_CASE(complex_sst1_k2) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
generate_clustered<1>("key2").then([] (auto&& mutation) {
sstables::test_env env;
generate_clustered<1>(env, "key2").then([] (auto&& mutation) {
auto s = complex_schema();
auto& sr = mutation.partition().static_row();
@@ -219,7 +226,8 @@ SEASTAR_THREAD_TEST_CASE(complex_sst1_k2) {
SEASTAR_THREAD_TEST_CASE(complex_sst2_k1) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
generate_clustered<2>("key1").then([] (auto&& mutation) {
sstables::test_env env;
generate_clustered<2>(env, "key1").then([] (auto&& mutation) {
auto s = complex_schema();
auto exploded = exploded_clustering_prefix({"cl1.1", "cl2.1"});
@@ -238,7 +246,8 @@ SEASTAR_THREAD_TEST_CASE(complex_sst2_k1) {
SEASTAR_THREAD_TEST_CASE(complex_sst2_k2) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
generate_clustered<2>("key2").then([] (auto&& mutation) {
sstables::test_env env;
generate_clustered<2>(env, "key2").then([] (auto&& mutation) {
auto s = complex_schema();
auto& sr = mutation.partition().static_row();
@@ -268,7 +277,8 @@ SEASTAR_THREAD_TEST_CASE(complex_sst2_k2) {
SEASTAR_THREAD_TEST_CASE(complex_sst2_k3) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
generate_clustered<2>("key3").then([] (auto&& mutation) {
sstables::test_env env;
generate_clustered<2>(env, "key3").then([] (auto&& mutation) {
auto s = complex_schema();
auto& sr = mutation.partition().static_row();
@@ -287,7 +297,8 @@ SEASTAR_THREAD_TEST_CASE(complex_sst2_k3) {
SEASTAR_THREAD_TEST_CASE(complex_sst3_k1) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
generate_clustered<3>("key1").then([] (auto&& mutation) {
sstables::test_env env;
generate_clustered<3>(env, "key1").then([] (auto&& mutation) {
auto s = complex_schema();
auto row = clustered_row(mutation, *s, {"cl1.2", "cl2.2"});
@@ -308,7 +319,8 @@ SEASTAR_THREAD_TEST_CASE(complex_sst3_k1) {
SEASTAR_THREAD_TEST_CASE(complex_sst3_k2) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
generate_clustered<3>("key2").then([] (auto&& mutation) {
sstables::test_env env;
generate_clustered<3>(env, "key2").then([] (auto&& mutation) {
auto s = complex_schema();
auto& sr = mutation.partition().static_row();
@@ -325,8 +337,8 @@ SEASTAR_THREAD_TEST_CASE(complex_sst3_k2) {
}).get();
}
future<> test_range_reads(const dht::token& min, const dht::token& max, std::vector<bytes>& expected) {
return reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([min, max, &expected] (auto sstp) mutable {
future<> test_range_reads(sstables::test_env& env, const dht::token& min, const dht::token& max, std::vector<bytes>& expected) {
return env.reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([min, max, &expected] (auto sstp) mutable {
auto s = uncompressed_schema();
auto count = make_lw_shared<size_t>(0);
auto expected_size = expected.size();
@@ -361,31 +373,31 @@ future<> test_range_reads(const dht::token& min, const dht::token& max, std::vec
SEASTAR_THREAD_TEST_CASE(read_range) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
std::vector<bytes> expected = { to_bytes("finna"), to_bytes("isak"), to_bytes("gustaf"), to_bytes("vinna") };
do_with(std::move(expected), [] (auto& expected) {
return test_range_reads(dht::minimum_token(), dht::maximum_token(), expected);
do_with(sstables::test_env(), std::move(expected), [] (auto& env, auto& expected) {
return test_range_reads(env, dht::minimum_token(), dht::maximum_token(), expected);
}).get();
}
SEASTAR_THREAD_TEST_CASE(read_partial_range) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
std::vector<bytes> expected = { to_bytes("finna"), to_bytes("isak") };
do_with(std::move(expected), [] (auto& expected) {
return test_range_reads(dht::global_partitioner().get_token(key_view(bytes_view(expected.back()))), dht::maximum_token(), expected);
do_with(sstables::test_env(), std::move(expected), [] (auto& env, auto& expected) {
return test_range_reads(env, dht::global_partitioner().get_token(key_view(bytes_view(expected.back()))), dht::maximum_token(), expected);
}).get();
}
SEASTAR_THREAD_TEST_CASE(read_partial_range_2) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
std::vector<bytes> expected = { to_bytes("gustaf"), to_bytes("vinna") };
do_with(std::move(expected), [] (auto& expected) {
return test_range_reads(dht::minimum_token(), dht::global_partitioner().get_token(key_view(bytes_view(expected.front()))), expected);
do_with(sstables::test_env(), std::move(expected), [] (auto& env, auto& expected) {
return test_range_reads(env, dht::minimum_token(), dht::global_partitioner().get_token(key_view(bytes_view(expected.front()))), expected);
}).get();
}
static
mutation_source make_sstable_mutation_source(schema_ptr s, sstring dir, std::vector<mutation> mutations,
mutation_source make_sstable_mutation_source(sstables::test_env& env, schema_ptr s, sstring dir, std::vector<mutation> mutations,
sstable_writer_config cfg, sstables::sstable::version_types version) {
auto sst = sstables::make_sstable(s,
auto sst = env.make_sstable(s,
dir,
1 /* generation */,
version,
@@ -405,11 +417,11 @@ mutation_source make_sstable_mutation_source(schema_ptr s, sstring dir, std::vec
// Must be run in a seastar thread
static
void test_mutation_source(sstable_writer_config cfg, sstables::sstable::version_types version) {
void test_mutation_source(sstables::test_env& env, sstable_writer_config cfg, sstables::sstable::version_types version) {
std::vector<tmpdir> dirs;
run_mutation_source_tests([&dirs, &cfg, version] (schema_ptr s, const std::vector<mutation>& partitions) -> mutation_source {
run_mutation_source_tests([&env, &dirs, &cfg, version] (schema_ptr s, const std::vector<mutation>& partitions) -> mutation_source {
dirs.emplace_back();
return make_sstable_mutation_source(s, dirs.back().path().string(), partitions, cfg, version);
return make_sstable_mutation_source(env, s, dirs.back().path().string(), partitions, cfg, version);
});
}
@@ -418,12 +430,13 @@ SEASTAR_TEST_CASE(test_sstable_conforms_to_mutation_source) {
return seastar::async([] {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
storage_service_for_tests ssft;
sstables::test_env env;
for (auto version : all_sstable_versions) {
for (auto index_block_size : {1, 128, 64*1024}) {
sstable_writer_config cfg;
cfg.promoted_index_block_size = index_block_size;
cfg.large_data_handler = &nop_lp_handler;
test_mutation_source(cfg, version);
test_mutation_source(env, cfg, version);
}
}
});
@@ -448,7 +461,8 @@ SEASTAR_TEST_CASE(test_sstable_can_write_and_read_range_tombstone) {
auto mt = make_lw_shared<memtable>(s);
mt->apply(std::move(m));
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
sstables::sstable::version_types::la,
@@ -472,7 +486,8 @@ SEASTAR_TEST_CASE(test_sstable_can_write_and_read_range_tombstone) {
SEASTAR_THREAD_TEST_CASE(compact_storage_sparse_read) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
reusable_sst(compact_sparse_schema(), "tests/sstables/compact_sparse", 1).then([] (auto sstp) {
sstables::test_env env;
env.reusable_sst(compact_sparse_schema(), "tests/sstables/compact_sparse", 1).then([] (auto sstp) {
return do_with(make_dkey(compact_sparse_schema(), "first_row"), [sstp] (auto& key) {
auto s = compact_sparse_schema();
auto rd = make_lw_shared<flat_mutation_reader>(sstp->read_row_flat(s, key));
@@ -490,7 +505,8 @@ SEASTAR_THREAD_TEST_CASE(compact_storage_sparse_read) {
SEASTAR_THREAD_TEST_CASE(compact_storage_simple_dense_read) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
reusable_sst(compact_simple_dense_schema(), "tests/sstables/compact_simple_dense", 1).then([] (auto sstp) {
sstables::test_env env;
env.reusable_sst(compact_simple_dense_schema(), "tests/sstables/compact_simple_dense", 1).then([] (auto sstp) {
return do_with(make_dkey(compact_simple_dense_schema(), "first_row"), [sstp] (auto& key) {
auto s = compact_simple_dense_schema();
auto rd = make_lw_shared<flat_mutation_reader>(sstp->read_row_flat(s, key));
@@ -510,7 +526,8 @@ SEASTAR_THREAD_TEST_CASE(compact_storage_simple_dense_read) {
SEASTAR_THREAD_TEST_CASE(compact_storage_dense_read) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
reusable_sst(compact_dense_schema(), "tests/sstables/compact_dense", 1).then([] (auto sstp) {
sstables::test_env env;
env.reusable_sst(compact_dense_schema(), "tests/sstables/compact_dense", 1).then([] (auto sstp) {
return do_with(make_dkey(compact_dense_schema(), "first_row"), [sstp] (auto& key) {
auto s = compact_dense_schema();
auto rd = make_lw_shared<flat_mutation_reader>(sstp->read_row_flat(s, key));
@@ -534,7 +551,8 @@ SEASTAR_THREAD_TEST_CASE(compact_storage_dense_read) {
// Make sure we don't regress on that.
SEASTAR_THREAD_TEST_CASE(broken_ranges_collection) {
auto wait_bg = seastar::defer([] { sstables::await_background_jobs().get(); });
reusable_sst(peers_schema(), "tests/sstables/broken_ranges", 2).then([] (auto sstp) {
sstables::test_env env;
env.reusable_sst(peers_schema(), "tests/sstables/broken_ranges", 2).then([] (auto sstp) {
auto s = peers_schema();
auto reader = make_lw_shared<flat_mutation_reader>(sstp->as_mutation_source().make_reader(s, query::full_partition_range));
return repeat([s, reader] {
@@ -586,7 +604,8 @@ static schema_ptr tombstone_overlap_schema() {
static future<sstable_ptr> ka_sst(schema_ptr schema, sstring dir, unsigned long generation) {
auto sst = make_sstable(std::move(schema), dir, generation, sstables::sstable::version_types::ka, big);
sstables::test_env env;
auto sst = env.make_sstable(std::move(schema), dir, generation, sstables::sstable::version_types::ka, big);
auto fut = sst->load();
return std::move(fut).then([sst = std::move(sst)] {
return make_ready_future<sstable_ptr>(std::move(sst));
@@ -823,7 +842,8 @@ SEASTAR_TEST_CASE(test_non_compound_table_row_is_not_marked_as_static) {
auto mt = make_lw_shared<memtable>(s);
mt->apply(std::move(m));
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
version,
@@ -859,7 +879,8 @@ SEASTAR_TEST_CASE(test_has_partition_key) {
auto mt = make_lw_shared<memtable>(s);
mt->apply(std::move(m));
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
version,
@@ -922,7 +943,8 @@ SEASTAR_TEST_CASE(test_promoted_index_blocks_are_monotonic) {
auto mt = make_lw_shared<memtable>(s);
mt->apply(std::move(m));
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
sstables::sstable::version_types::ka,
@@ -975,7 +997,8 @@ SEASTAR_TEST_CASE(test_promoted_index_blocks_are_monotonic_compound_dense) {
auto mt = make_lw_shared<memtable>(s);
mt->apply(std::move(m));
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
version,
@@ -1035,7 +1058,8 @@ SEASTAR_TEST_CASE(test_promoted_index_blocks_are_monotonic_non_compound_dense) {
auto mt = make_lw_shared<memtable>(s);
mt->apply(std::move(m));
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
version,
@@ -1092,7 +1116,8 @@ SEASTAR_TEST_CASE(test_promoted_index_repeats_open_tombstones) {
auto mt = make_lw_shared<memtable>(s);
mt->apply(m);
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
generation,
version,
@@ -1139,7 +1164,8 @@ SEASTAR_TEST_CASE(test_range_tombstones_are_correctly_seralized_for_non_compound
auto mt = make_lw_shared<memtable>(s);
mt->apply(m);
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
version,
@@ -1179,7 +1205,8 @@ SEASTAR_TEST_CASE(test_promoted_index_is_absent_for_schemas_without_clustering_k
auto mt = make_lw_shared<memtable>(s);
mt->apply(m);
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
version,
@@ -1220,7 +1247,8 @@ SEASTAR_TEST_CASE(test_can_write_and_read_non_compound_range_tombstone_as_compou
auto mt = make_lw_shared<memtable>(s);
mt->apply(m);
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
version,
@@ -1274,7 +1302,8 @@ SEASTAR_TEST_CASE(test_writing_combined_stream_with_tombstones_at_the_same_posit
auto mt2 = make_lw_shared<memtable>(s);
mt2->apply(m2);
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
version,
@@ -1319,7 +1348,8 @@ SEASTAR_TEST_CASE(test_no_index_reads_when_rows_fall_into_range_boundaries) {
cfg.large_data_handler = &nop_lp_handler;
tmpdir dir;
auto ms = make_sstable_mutation_source(s, dir.path().string(), {m1, m2}, cfg, version);
sstables::test_env env;
auto ms = make_sstable_mutation_source(env, s, dir.path().string(), {m1, m2}, cfg, version);
auto index_accesses = [] {
auto&& stats = sstables::shared_index_lists::shard_stats();
@@ -1396,7 +1426,8 @@ SEASTAR_THREAD_TEST_CASE(test_large_index_pages_do_not_cause_large_allocations)
mt->apply(m);
}
auto sst = sstables::make_sstable(s,
sstables::test_env env;
auto sst = env.make_sstable(s,
dir.path().string(),
1 /* generation */,
sstable_version_types::ka,
@@ -1438,18 +1469,19 @@ SEASTAR_THREAD_TEST_CASE(test_schema_changes) {
shared_sstable created_with_base_schema;
shared_sstable created_with_changed_schema;
sstables::test_env env;
if (it == cache.end()) {
auto mt = make_lw_shared<memtable>(base);
for (auto& m : base_mutations) {
mt->apply(m);
}
created_with_base_schema = sstables::make_sstable(base, dir.path().string(), gen, version, sstables::sstable::format_types::big);
created_with_base_schema = env.make_sstable(base, dir.path().string(), gen, version, sstables::sstable::format_types::big);
sstable_writer_config cfg;
cfg.large_data_handler = &nop_lp_handler;
created_with_base_schema->write_components(mt->make_flat_reader(base), base_mutations.size(), base, cfg, mt->get_encoding_stats()).get();
created_with_base_schema->load().get();
created_with_changed_schema = sstables::make_sstable(changed, dir.path().string(), gen, version, sstables::sstable::format_types::big);
created_with_changed_schema = env.make_sstable(changed, dir.path().string(), gen, version, sstables::sstable::format_types::big);
created_with_changed_schema->load().get();
cache.emplace(std::tuple { version, base }, std::tuple { created_with_base_schema, gen });
@@ -1457,7 +1489,7 @@ SEASTAR_THREAD_TEST_CASE(test_schema_changes) {
} else {
created_with_base_schema = std::get<shared_sstable>(it->second);
created_with_changed_schema = sstables::make_sstable(changed, dir.path().string(), std::get<int>(it->second), version, sstables::sstable::format_types::big);
created_with_changed_schema = env.make_sstable(changed, dir.path().string(), std::get<int>(it->second), version, sstables::sstable::format_types::big);
created_with_changed_schema->load().get();
}
@@ -1518,17 +1550,18 @@ SEASTAR_THREAD_TEST_CASE(test_reading_serialization_header) {
auto m1ow = md1_overwrite.build(s);
mt->apply(m1ow);
sstables::test_env env;
{
// SSTable class has way too many responsibilities. In particular, it mixes the reading and
// writting parts. Let's use a separate objects for writing and reading to ensure that nothing
// carries over that wouldn't normally be read from disk.
auto sst = sstables::make_sstable(s, dir.path().string(), 1, sstable::version_types::mc, sstables::sstable::format_types::big);
auto sst = env.make_sstable(s, dir.path().string(), 1, sstable::version_types::mc, sstables::sstable::format_types::big);
sstable_writer_config cfg;
cfg.large_data_handler = &nop_lp_handler;
sst->write_components(mt->make_flat_reader(s), 2, s, cfg, mt->get_encoding_stats()).get();
}
auto sst = sstables::make_sstable(s, dir.path().string(), 1, sstable::version_types::mc, sstables::sstable::format_types::big);
auto sst = env.make_sstable(s, dir.path().string(), 1, sstable::version_types::mc, sstables::sstable::format_types::big);
sst->load().get();
auto hdr = sst->get_serialization_header();
@@ -1617,8 +1650,9 @@ SEASTAR_THREAD_TEST_CASE(test_counter_header_size) {
auto mt = make_lw_shared<memtable>(s);
mt->apply(m);
sstables::test_env env;
for (const auto version : all_sstable_versions) {
auto sst = sstables::make_sstable(s, dir.path().string(), 1, version, sstables::sstable::format_types::big);
auto sst = env.make_sstable(s, dir.path().string(), 1, version, sstables::sstable::format_types::big);
sstable_writer_config cfg;
cfg.large_data_handler = &nop_lp_handler;
sst->write_components(mt->make_flat_reader(s), 1, s, cfg, mt->get_encoding_stats()).get();

69
tests/sstable_resharding_test.cc
View File

@@ -2,6 +2,7 @@
#include <memory>
#include <utility>
#include <seastar/testing/thread_test_case.hh>
#include <seastar/core/sstring.hh>
#include <seastar/core/future-util.hh>
#include <seastar/core/do_with.hh>
@@ -13,6 +14,7 @@
#include "sstables/compaction_manager.hh"
#include "mutation_reader.hh"
#include "sstable_test.hh"
#include "test_services.hh"
#include "tmpdir.hh"
#include "cell_locking.hh"
#include "flat_mutation_reader_assertions.hh"
@@ -23,13 +25,6 @@ using namespace sstables;
static db::nop_large_data_handler nop_lp_handler;
static inline dht::token create_token_from_key(sstring key) {
sstables::key_view key_view = sstables::key_view(bytes_view(reinterpret_cast<const signed char*>(key.c_str()), key.size()));
dht::token token = dht::global_partitioner().get_token(key_view);
assert(token == dht::global_partitioner().get_token(key_view));
return token;
}
static inline std::vector<std::pair<sstring, dht::token>> token_generation_for_shard(shard_id shard, unsigned tokens_to_generate) {
unsigned tokens = 0;
unsigned key_id = 0;
@@ -56,6 +51,10 @@ static inline std::vector<std::pair<sstring, dht::token>> token_generation_for_s
return key_and_token_pair;
}
static std::vector<std::pair<sstring, dht::token>> token_generation_for_current_shard(unsigned tokens_to_generate) {
return token_generation_for_shard(engine().cpu_id(), tokens_to_generate);
}
static schema_ptr get_schema() {
auto builder = schema_builder("tests", "sstable_resharding_test")
.with_column("id", utf8_type, column_kind::partition_key)
@@ -64,6 +63,7 @@ static schema_ptr get_schema() {
}
void run_sstable_resharding_test() {
test_env env;
cache_tracker tracker;
for (const auto version : all_sstable_versions) {
storage_service_for_tests ssft;
@@ -97,11 +97,10 @@ void run_sstable_resharding_test() {
mt->apply(std::move(m));
}
}
auto sst = sstables::make_sstable(s, tmp.path().string(), 0, version, sstables::sstable::format_types::big);
auto sst = env.make_sstable(s, tmp.path().string(), 0, version, sstables::sstable::format_types::big);
write_memtable_to_sstable_for_test(*mt, sst).get();
}
auto sst = sstables::make_sstable(s, tmp.path().string(), 0, version, sstables::sstable::format_types::big);
sst->load().get();
auto sst = env.reusable_sst(s, tmp.path().string(), 0, version, sstables::sstable::format_types::big).get0();
// FIXME: sstable write has a limitation in which it will generate sharding metadata only
// for a single shard. workaround that by setting shards manually. from this test perspective,
@@ -112,17 +111,15 @@ void run_sstable_resharding_test() {
auto filter_fname = sstables::test(sst).filename(component_type::Filter);
uint64_t bloom_filter_size_before = file_size(filter_fname).get0();
auto creator = [&cf, &tmp, version] (shard_id shard) mutable {
auto creator = [&env, &cf, &tmp, version] (shard_id shard) mutable {
// we need generation calculated by instance of cf at requested shard,
// or resource usage wouldn't be fairly distributed among shards.
auto gen = smp::submit_to(shard, [&cf] () {
return column_family_test::calculate_generation_for_new_table(*cf);
}).get0();
auto sst = sstables::make_sstable(cf->schema(), tmp.path().string(), gen,
version, sstables::sstable::format_types::big,
gc_clock::now(), default_io_error_handler_gen());
return sst;
return env.make_sstable(cf->schema(), tmp.path().string(), gen,
version, sstables::sstable::format_types::big);
};
auto new_sstables = sstables::reshard_sstables({ sst }, *cf, creator, std::numeric_limits<uint64_t>::max(), 0).get0();
BOOST_REQUIRE(new_sstables.size() == smp::count);
@@ -130,9 +127,8 @@ void run_sstable_resharding_test() {
uint64_t bloom_filter_size_after = 0;
for (auto& sstable : new_sstables) {
auto new_sst = sstables::make_sstable(s, tmp.path().string(), sstable->generation(),
version, sstables::sstable::format_types::big);
new_sst->load().get();
auto new_sst = env.reusable_sst(s, tmp.path().string(), sstable->generation(),
version, sstables::sstable::format_types::big).get0();
filter_fname = sstables::test(new_sst).filename(component_type::Filter);
bloom_filter_size_after += file_size(filter_fname).get0();
auto shards = new_sst->get_shards_for_this_sstable();
@@ -156,3 +152,40 @@ SEASTAR_TEST_CASE(sstable_resharding_test) {
run_sstable_resharding_test();
});
}
SEASTAR_THREAD_TEST_CASE(sstable_resharding_strategy_tests) {
test_env env;
for (const auto version : all_sstable_versions) {
auto s = make_lw_shared(schema({}, "ks", "cf", {{"p1", utf8_type}}, {}, {}, {}, utf8_type));
auto get_sstable = [&] (int64_t gen, sstring first_key, sstring last_key) mutable {
auto sst = env.make_sstable(s, "", gen, version, sstables::sstable::format_types::big);
stats_metadata stats = {};
stats.sstable_level = 1;
sstables::test(sst).set_values(std::move(first_key), std::move(last_key), std::move(stats));
return sst;
};
column_family_for_tests cf;
auto tokens = token_generation_for_current_shard(2);
auto stcs = sstables::make_compaction_strategy(sstables::compaction_strategy_type::size_tiered, s->compaction_strategy_options());
auto lcs = sstables::make_compaction_strategy(sstables::compaction_strategy_type::leveled, s->compaction_strategy_options());
auto sst1 = get_sstable(1, tokens[0].first, tokens[1].first);
auto sst2 = get_sstable(2, tokens[1].first, tokens[1].first);
{
// TODO: sstable_test runs with smp::count == 1, thus we will not be able to stress it more
// until we move this test case to sstable_resharding_test.
auto descriptors = stcs.get_resharding_jobs(*cf, { sst1, sst2 });
BOOST_REQUIRE(descriptors.size() == 2);
}
{
auto ssts = std::vector<sstables::shared_sstable>{ sst1, sst2 };
auto descriptors = lcs.get_resharding_jobs(*cf, ssts);
auto expected_jobs = ssts.size()/smp::count + ssts.size()%smp::count;
BOOST_REQUIRE(descriptors.size() == expected_jobs);
}
}
}

164
tests/sstable_test.cc
View File

@@ -48,21 +48,37 @@ bytes as_bytes(const sstring& s) {
return { reinterpret_cast<const int8_t*>(s.begin()), s.size() };
}
future<> test_using_working_sst(schema_ptr s, sstring dir, int64_t gen) {
return test_env::do_with([s = std::move(s), dir = std::move(dir), gen] (test_env& env) {
return env.working_sst(std::move(s), std::move(dir), gen);
});
}
SEASTAR_TEST_CASE(uncompressed_data) {
return working_sst(uncompressed_schema(), uncompressed_dir(), 1);
return test_using_working_sst(uncompressed_schema(), uncompressed_dir(), 1);
}
SEASTAR_TEST_CASE(compressed_data) {
auto s = make_lw_shared(schema({}, "ks", "cf", {}, {}, {}, {}, utf8_type));
return working_sst(std::move(s), "tests/sstables/compressed", 1);
return test_using_working_sst(std::move(s), "tests/sstables/compressed", 1);
}
SEASTAR_TEST_CASE(composite_index) {
return working_sst(composite_schema(), "tests/sstables/composite", 1);
return test_using_working_sst(composite_schema(), "tests/sstables/composite", 1);
}
template<typename Func>
inline auto
test_using_reusable_sst(schema_ptr s, sstring dir, unsigned long gen, Func&& func) {
return test_env::do_with([s = std::move(s), dir = std::move(dir), gen, func = std::move(func)] (test_env& env) {
return env.reusable_sst(std::move(s), std::move(dir), gen).then([func = std::move(func)] (sstable_ptr sst) {
return func(std::move(sst));
});
});
}
future<index_list> index_read(schema_ptr schema, sstring path) {
return reusable_sst(std::move(schema), path, 1).then([] (sstable_ptr ptr) {
return test_using_reusable_sst(std::move(schema), std::move(path), 1, [] (sstable_ptr ptr) {
return sstables::test(ptr).read_indexes();
});
}
@@ -81,7 +97,7 @@ SEASTAR_TEST_CASE(composite_index_read) {
template<uint64_t Position, uint64_t EntryPosition, uint64_t EntryKeySize>
future<> summary_query(schema_ptr schema, sstring path, int generation) {
return reusable_sst(std::move(schema), path, generation).then([] (sstable_ptr ptr) {
return test_using_reusable_sst(std::move(schema), path, generation, [] (sstable_ptr ptr) {
return sstables::test(ptr).read_summary_entry(Position).then([ptr] (auto entry) {
BOOST_REQUIRE(entry.position == EntryPosition);
BOOST_REQUIRE(entry.key.size() == EntryKeySize);
@@ -146,7 +162,7 @@ SEASTAR_TEST_CASE(missing_summary_interval_1_query_ok) {
#endif
SEASTAR_TEST_CASE(missing_summary_first_last_sane) {
return reusable_sst(uncompressed_schema(), uncompressed_dir(), 2).then([] (sstable_ptr ptr) {
return test_using_reusable_sst(uncompressed_schema(), uncompressed_dir(), 2, [] (shared_sstable ptr) {
auto& summary = sstables::test(ptr).get_summary();
BOOST_REQUIRE(summary.header.size == 1);
BOOST_REQUIRE(summary.positions.size() == 1);
@@ -157,9 +173,8 @@ SEASTAR_TEST_CASE(missing_summary_first_last_sane) {
});
}
static future<sstable_ptr> do_write_sst(schema_ptr schema, sstring load_dir, sstring write_dir, unsigned long generation) {
auto sst = sstables::make_sstable(std::move(schema), load_dir, generation, la, big);
return sst->load().then([sst, write_dir, generation] {
static future<sstable_ptr> do_write_sst(test_env& env, schema_ptr schema, sstring load_dir, sstring write_dir, unsigned long generation) {
return env.reusable_sst(std::move(schema), load_dir, generation, la).then([write_dir, generation] (sstable_ptr sst) {
sstables::test(sst).change_generation_number(generation + 1);
sstables::test(sst).change_dir(write_dir);
auto fut = sstables::test(sst).store();
@@ -169,6 +184,12 @@ static future<sstable_ptr> do_write_sst(schema_ptr schema, sstring load_dir, sst
});
}
static future<sstable_ptr> do_write_sst(schema_ptr schema, sstring load_dir, sstring write_dir, unsigned long generation) {
return test_env::do_with([schema = std::move(schema), load_dir = std::move(load_dir), write_dir = std::move(write_dir), generation] (test_env& env) {
return do_write_sst(env, std::move(schema), std::move(load_dir), std::move(write_dir), generation);
});
}
static future<> write_sst_info(schema_ptr schema, sstring load_dir, sstring write_dir, unsigned long generation) {
return do_write_sst(std::move(schema), load_dir, write_dir, generation).then([] (auto ptr) { return make_ready_future<>(); });
}
@@ -225,11 +246,20 @@ SEASTAR_TEST_CASE(check_compressed_info_func) {
return check_component_integrity(component_type::CompressionInfo);
}
template <typename Func>
inline auto
write_and_validate_sst(schema_ptr s, sstring dir, Func&& func) {
return test_env::do_with(tmpdir(), [s = std::move(s), dir = std::move(dir), func = std::move(func)] (test_env& env, tmpdir& tmp) {
return do_write_sst(env, s, dir, tmp.path().string(), 1).then([&env, &tmp, s = std::move(s), func = std::move(func)] (auto sst1) {
auto sst2 = env.make_sstable(s, tmp.path().string(), 2, la, big);
return func(std::move(sst1), std::move(sst2));
});
});
}
SEASTAR_TEST_CASE(check_summary_func) {
auto tmp = make_lw_shared<tmpdir>();
auto s = make_lw_shared(schema({}, "ks", "cf", {}, {}, {}, {}, utf8_type));
return do_write_sst(s, "tests/sstables/compressed", tmp->path().string(), 1).then([tmp, s] (auto sst1) {
auto sst2 = make_sstable(s, tmp->path().string(), 2, la, big);
return write_and_validate_sst(std::move(s), "tests/sstables/compressed", [] (shared_sstable sst1, shared_sstable sst2) {
return sstables::test(sst2).read_summary().then([sst1, sst2] {
summary& sst1_s = sstables::test(sst1).get_summary();
summary& sst2_s = sstables::test(sst2).get_summary();
@@ -239,8 +269,9 @@ SEASTAR_TEST_CASE(check_summary_func) {
BOOST_REQUIRE(sst1_s.entries == sst2_s.entries);
BOOST_REQUIRE(sst1_s.first_key.value == sst2_s.first_key.value);
BOOST_REQUIRE(sst1_s.last_key.value == sst2_s.last_key.value);
return make_ready_future<>();
});
}).then([tmp] {});
});
}
SEASTAR_TEST_CASE(check_filter_func) {
@@ -248,10 +279,8 @@ SEASTAR_TEST_CASE(check_filter_func) {
}
SEASTAR_TEST_CASE(check_statistics_func) {
auto tmp = make_lw_shared<tmpdir>();
auto s = make_lw_shared(schema({}, "ks", "cf", {}, {}, {}, {}, utf8_type));
return do_write_sst(s, "tests/sstables/compressed", tmp->path().string(), 1).then([tmp, s] (auto sst1) {
auto sst2 = make_sstable(s, tmp->path().string(), 2, la, big);
return write_and_validate_sst(std::move(s), "tests/sstables/compressed", [] (shared_sstable sst1, shared_sstable sst2) {
return sstables::test(sst2).read_statistics().then([sst1, sst2] {
statistics& sst1_s = sstables::test(sst1).get_statistics();
statistics& sst2_s = sstables::test(sst2).get_statistics();
@@ -263,26 +292,26 @@ SEASTAR_TEST_CASE(check_statistics_func) {
BOOST_REQUIRE(boost::get<0>(e).second == boost::get<1>(e).second);
}
// TODO: compare the field contents from both sstables.
return make_ready_future<>();
});
}).then([tmp] {});
});
}
SEASTAR_TEST_CASE(check_toc_func) {
auto tmp = make_lw_shared<tmpdir>();
auto s = make_lw_shared(schema({}, "ks", "cf", {}, {}, {}, {}, utf8_type));
return do_write_sst(s, "tests/sstables/compressed", tmp->path().string(), 1).then([tmp, s] (auto sst1) {
auto sst2 = sstables::make_sstable(s, tmp->path().string(), 2, la, big);
return write_and_validate_sst(std::move(s), "tests/sstables/compressed", [] (shared_sstable sst1, shared_sstable sst2) {
return sstables::test(sst2).read_toc().then([sst1, sst2] {
auto& sst1_c = sstables::test(sst1).get_components();
auto& sst2_c = sstables::test(sst2).get_components();
BOOST_REQUIRE(sst1_c == sst2_c);
return make_ready_future<>();
});
}).then([tmp] {});
});
}
SEASTAR_TEST_CASE(uncompressed_random_access_read) {
return reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), uncompressed_dir(), 1, [] (auto sstp) {
// note: it's important to pass on a shared copy of sstp to prevent its
// destruction until the continuation finishes reading!
return sstables::test(sstp).data_read(97, 6).then([sstp] (temporary_buffer<char> buf) {
@@ -294,7 +323,7 @@ SEASTAR_TEST_CASE(uncompressed_random_access_read) {
SEASTAR_TEST_CASE(compressed_random_access_read) {
auto s = make_lw_shared(schema({}, "ks", "cf", {}, {}, {}, {}, utf8_type));
return reusable_sst(std::move(s), "tests/sstables/compressed", 1).then([] (auto sstp) {
return test_using_reusable_sst(std::move(s), "tests/sstables/compressed", 1, [] (auto sstp) {
return sstables::test(sstp).data_read(97, 6).then([sstp] (temporary_buffer<char> buf) {
BOOST_REQUIRE(sstring(buf.get(), buf.size()) == "gustaf");
return make_ready_future<>();
@@ -388,7 +417,7 @@ public:
};
SEASTAR_TEST_CASE(uncompressed_rows_read_one) {
return reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), uncompressed_dir(), 1, [] (auto sstp) {
return do_with(test_row_consumer(1418656871665302), [sstp] (auto& c) {
auto context = data_consume_rows<data_consume_rows_context>(*uncompressed_schema(), sstp, c, {0, 95}, 95);
auto fut = context.read();
@@ -406,7 +435,7 @@ SEASTAR_TEST_CASE(uncompressed_rows_read_one) {
SEASTAR_TEST_CASE(compressed_rows_read_one) {
auto s = make_lw_shared(schema({}, "ks", "cf", {}, {}, {}, {}, utf8_type));
return reusable_sst(std::move(s), "tests/sstables/compressed", 1).then([] (auto sstp) {
return test_using_reusable_sst(std::move(s), "tests/sstables/compressed", 1, [] (auto sstp) {
return do_with(test_row_consumer(1418654707438005), [sstp] (auto& c) {
auto context = data_consume_rows<data_consume_rows_context>(*uncompressed_schema(), sstp, c, {0, 95}, 95);
auto fut = context.read();
@@ -472,7 +501,7 @@ public:
SEASTAR_TEST_CASE(uncompressed_rows_read_all) {
return reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), uncompressed_dir(), 1, [] (auto sstp) {
return do_with(count_row_consumer(), [sstp] (auto& c) {
auto context = data_consume_rows<data_consume_rows_context>(*uncompressed_schema(), sstp, c);
auto fut = context.read();
@@ -490,7 +519,7 @@ SEASTAR_TEST_CASE(uncompressed_rows_read_all) {
SEASTAR_TEST_CASE(compressed_rows_read_all) {
auto s = make_lw_shared(schema({}, "ks", "cf", {}, {}, {}, {}, utf8_type));
return reusable_sst(std::move(s), "tests/sstables/compressed", 1).then([] (auto sstp) {
return test_using_reusable_sst(std::move(s), "tests/sstables/compressed", 1, [] (auto sstp) {
return do_with(count_row_consumer(), [sstp] (auto& c) {
auto context = data_consume_rows<data_consume_rows_context>(*uncompressed_schema(), sstp, c);
auto fut = context.read();
@@ -517,7 +546,7 @@ public:
};
SEASTAR_TEST_CASE(pausable_uncompressed_rows_read_all) {
return reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), uncompressed_dir(), 1, [] (auto sstp) {
return do_with(pausable_count_row_consumer(), [sstp] (auto& c) {
auto context = data_consume_rows<data_consume_rows_context>(*uncompressed_schema(), sstp, c);
auto fut = context.read();
@@ -564,7 +593,7 @@ public:
};
SEASTAR_TEST_CASE(read_set) {
return reusable_sst(set_schema(), "tests/sstables/set", 1).then([] (auto sstp) {
return test_using_reusable_sst(set_schema(), "tests/sstables/set", 1, [] (auto sstp) {
return do_with(set_consumer(), [sstp] (auto& c) {
auto context = data_consume_rows<data_consume_rows_context>(*uncompressed_schema(), sstp, c);
auto fut = context.read();
@@ -621,7 +650,7 @@ public:
};
SEASTAR_TEST_CASE(ttl_read) {
return reusable_sst(uncompressed_schema(), "tests/sstables/ttl", 1).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), "tests/sstables/ttl", 1, [] (auto sstp) {
return do_with(ttl_row_consumer(1430151018675502), [sstp] (auto& c) {
auto context = data_consume_rows<data_consume_rows_context>(*uncompressed_schema(), sstp, c);
auto fut = context.read();
@@ -660,7 +689,7 @@ public:
};
SEASTAR_TEST_CASE(deleted_cell_read) {
return reusable_sst(uncompressed_schema(), "tests/sstables/deleted_cell", 2).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), "tests/sstables/deleted_cell", 2, [] (auto sstp) {
return do_with(deleted_cell_row_consumer(), [sstp] (auto& c) {
auto context = data_consume_rows<data_consume_rows_context>(*uncompressed_schema(), sstp, c);
auto fut = context.read();
@@ -678,7 +707,7 @@ SEASTAR_TEST_CASE(deleted_cell_read) {
SEASTAR_TEST_CASE(find_key_map) {
return reusable_sst(map_schema(), "tests/sstables/map_pk", 1).then([] (auto sstp) {
return test_using_reusable_sst(map_schema(), "tests/sstables/map_pk", 1, [] (auto sstp) {
schema_ptr s = map_schema();
auto& summary = sstables::test(sstp)._summary();
std::vector<data_value> kk;
@@ -695,11 +724,12 @@ SEASTAR_TEST_CASE(find_key_map) {
auto key = sstables::key::from_deeply_exploded(*s, kk);
BOOST_REQUIRE(sstables::test(sstp).binary_search(summary.entries, key) == 0);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(find_key_set) {
return reusable_sst(set_schema(), "tests/sstables/set_pk", 1).then([] (auto sstp) {
return test_using_reusable_sst(set_schema(), "tests/sstables/set_pk", 1, [] (auto sstp) {
schema_ptr s = set_schema();
auto& summary = sstables::test(sstp)._summary();
std::vector<data_value> kk;
@@ -716,11 +746,12 @@ SEASTAR_TEST_CASE(find_key_set) {
auto key = sstables::key::from_deeply_exploded(*s, kk);
BOOST_REQUIRE(sstables::test(sstp).binary_search(summary.entries, key) == 0);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(find_key_list) {
return reusable_sst(list_schema(), "tests/sstables/list_pk", 1).then([] (auto sstp) {
return test_using_reusable_sst(list_schema(), "tests/sstables/list_pk", 1, [] (auto sstp) {
schema_ptr s = set_schema();
auto& summary = sstables::test(sstp)._summary();
std::vector<data_value> kk;
@@ -737,12 +768,13 @@ SEASTAR_TEST_CASE(find_key_list) {
auto key = sstables::key::from_deeply_exploded(*s, kk);
BOOST_REQUIRE(sstables::test(sstp).binary_search(summary.entries, key) == 0);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(find_key_composite) {
return reusable_sst(composite_schema(), "tests/sstables/composite", 1).then([] (auto sstp) {
return test_using_reusable_sst(composite_schema(), "tests/sstables/composite", 1, [] (auto sstp) {
schema_ptr s = composite_schema();
auto& summary = sstables::test(sstp)._summary();
std::vector<data_value> kk;
@@ -755,11 +787,12 @@ SEASTAR_TEST_CASE(find_key_composite) {
auto key = sstables::key::from_deeply_exploded(*s, kk);
BOOST_REQUIRE(sstables::test(sstp).binary_search(summary.entries, key) == 0);
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(all_in_place) {
return reusable_sst(uncompressed_schema(), "tests/sstables/bigsummary", 76).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), "tests/sstables/bigsummary", 76, [] (auto sstp) {
auto& summary = sstables::test(sstp)._summary();
int idx = 0;
@@ -767,11 +800,12 @@ SEASTAR_TEST_CASE(all_in_place) {
auto key = sstables::key::from_bytes(bytes(e.key));
BOOST_REQUIRE(sstables::test(sstp).binary_search(summary.entries, key) == idx++);
}
return make_ready_future<>();
});
}
SEASTAR_TEST_CASE(full_index_search) {
return reusable_sst(uncompressed_schema(), uncompressed_dir(), 1).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), uncompressed_dir(), 1, [] (auto sstp) {
return sstables::test(sstp).read_indexes().then([sstp] (auto index_list) {
int idx = 0;
for (auto& ie: index_list) {
@@ -783,7 +817,7 @@ SEASTAR_TEST_CASE(full_index_search) {
}
SEASTAR_TEST_CASE(not_find_key_composite_bucket0) {
return reusable_sst(composite_schema(), "tests/sstables/composite", 1).then([] (auto sstp) {
return test_using_reusable_sst(composite_schema(), "tests/sstables/composite", 1, [] (auto sstp) {
schema_ptr s = composite_schema();
auto& summary = sstables::test(sstp)._summary();
std::vector<data_value> kk;
@@ -797,12 +831,13 @@ SEASTAR_TEST_CASE(not_find_key_composite_bucket0) {
auto key = sstables::key::from_deeply_exploded(*s, kk);
// (result + 1) * -1 -1 = 0
BOOST_REQUIRE(sstables::test(sstp).binary_search(summary.entries, key) == -2);
return make_ready_future<>();
});
}
// See CASSANDRA-7593. This sstable writes 0 in the range_start. We need to handle that case as well
SEASTAR_TEST_CASE(wrong_range) {
return reusable_sst(uncompressed_schema(), "tests/sstables/wrongrange", 114).then([] (auto sstp) {
return test_using_reusable_sst(uncompressed_schema(), "tests/sstables/wrongrange", 114, [] (auto sstp) {
return do_with(make_dkey(uncompressed_schema(), "todata"), [sstp] (auto& key) {
auto s = columns_schema();
auto rd = make_lw_shared<flat_mutation_reader>(sstp->read_row_flat(s, key));
@@ -830,11 +865,11 @@ test_sstable_exists(sstring dir, unsigned long generation, bool exists) {
// We will therefore run it in an empty directory, and first link a known SSTable from another
// directory to it.
SEASTAR_TEST_CASE(set_generation) {
return test_setup::do_with_cloned_tmp_directory(uncompressed_dir(), [] (sstring uncompressed_dir, sstring generation_dir) {
return reusable_sst(uncompressed_schema(), uncompressed_dir, 1).then([generation_dir] (auto sstp) {
return test_setup::do_with_cloned_tmp_directory(uncompressed_dir(), [] (test_env& env, sstring uncompressed_dir, sstring generation_dir) {
return env.reusable_sst(uncompressed_schema(), uncompressed_dir, 1).then([generation_dir] (auto sstp) {
return sstp->create_links(generation_dir).then([sstp] {});
}).then([generation_dir] {
return reusable_sst(uncompressed_schema(), generation_dir, 1).then([] (auto sstp) {
}).then([&env, generation_dir] {
return env.reusable_sst(uncompressed_schema(), generation_dir, 1).then([] (auto sstp) {
return sstp->set_generation(2).then([sstp] {});
});
}).then([generation_dir] {
@@ -848,8 +883,8 @@ SEASTAR_TEST_CASE(set_generation) {
}
SEASTAR_TEST_CASE(reshuffle) {
return test_setup::do_with_cloned_tmp_directory(uncompressed_dir(), [] (sstring uncompressed_dir, sstring generation_dir) {
return reusable_sst(uncompressed_schema(), uncompressed_dir, 1).then([generation_dir] (auto sstp) {
return test_setup::do_with_cloned_tmp_directory(uncompressed_dir(), [] (test_env& env, sstring uncompressed_dir, sstring generation_dir) {
return env.reusable_sst(uncompressed_schema(), uncompressed_dir, 1).then([generation_dir] (auto sstp) {
return sstp->create_links(generation_dir, 1).then([sstp, generation_dir] {
return sstp->create_links(generation_dir, 5).then([sstp, generation_dir] {
return sstp->create_links(generation_dir, 10);
@@ -892,19 +927,20 @@ SEASTAR_TEST_CASE(reshuffle) {
}
SEASTAR_TEST_CASE(statistics_rewrite) {
return test_setup::do_with_cloned_tmp_directory(uncompressed_dir(), [] (sstring uncompressed_dir, sstring generation_dir) {
return reusable_sst(uncompressed_schema(), uncompressed_dir, 1).then([generation_dir] (auto sstp) {
return test_setup::do_with_cloned_tmp_directory(uncompressed_dir(), [] (test_env& env, sstring uncompressed_dir, sstring generation_dir) {
return env.reusable_sst(uncompressed_schema(), uncompressed_dir, 1).then([generation_dir] (auto sstp) {
return sstp->create_links(generation_dir).then([sstp] {});
}).then([generation_dir] {
return test_sstable_exists(generation_dir, 1, true);
}).then([generation_dir] {
return reusable_sst(uncompressed_schema(), generation_dir, 1).then([] (auto sstp) {
}).then([&env, generation_dir] {
return env.reusable_sst(uncompressed_schema(), generation_dir, 1).then([] (auto sstp) {
// mutate_sstable_level results in statistics rewrite
return sstp->mutate_sstable_level(10).then([sstp] {});
});
}).then([generation_dir] {
return reusable_sst(uncompressed_schema(), generation_dir, 1).then([] (auto sstp) {
}).then([&env, generation_dir] {
return env.reusable_sst(uncompressed_schema(), generation_dir, 1).then([] (auto sstp) {
BOOST_REQUIRE(sstp->get_sstable_level() == 10);
return make_ready_future<>();
});
});
});
@@ -938,14 +974,10 @@ static schema_ptr large_partition_schema() {
return s;
}
static future<shared_sstable> load_large_partition_sst(const sstables::sstable::version_types version) {
static future<shared_sstable> load_large_partition_sst(test_env& env, const sstables::sstable::version_types version) {
auto s = large_partition_schema();
auto sst = make_sstable(s, get_test_dir("large_partition", s), 3,
version, big);
auto fut = sst->load();
return std::move(fut).then([sst = std::move(sst)] {
return std::move(sst);
});
auto dir = get_test_dir("large_partition", s);
return env.reusable_sst(std::move(s), std::move(dir), 3, version);
}
// This is a rudimentary test that reads an sstable exported from Cassandra
@@ -954,12 +986,14 @@ static future<shared_sstable> load_large_partition_sst(const sstables::sstable::
// search for anything.
SEASTAR_TEST_CASE(promoted_index_read) {
return for_each_sstable_version([] (const sstables::sstable::version_types version) {
return load_large_partition_sst(version).then([] (auto sstp) {
return test_env::do_with([version] (test_env& env) {
return load_large_partition_sst(env, version).then([] (auto sstp) {
schema_ptr s = large_partition_schema();
return sstables::test(sstp).read_indexes().then([sstp] (index_list vec) {
BOOST_REQUIRE(vec.size() == 1);
BOOST_REQUIRE(vec.front().get_promoted_index_size() > 0);
});
});
});
});
}
@@ -1061,7 +1095,8 @@ static future<int> count_rows(sstable_ptr sstp, schema_ptr s, sstring ck1, sstri
SEASTAR_TEST_CASE(sub_partition_read) {
schema_ptr s = large_partition_schema();
return for_each_sstable_version([s] (const sstables::sstable::version_types version) {
return load_large_partition_sst(version).then([s] (auto sstp) {
return test_env::do_with([s, version] (test_env& env) {
return load_large_partition_sst(env, version).then([s] (auto sstp) {
return count_rows(sstp, s, "v1", "18wX", "18xB").then([] (int nrows) {
// there should be 5 rows (out of 13520 = 20*26*26) in this range:
// 18wX, 18wY, 18wZ, 18xA, 18xB.
@@ -1119,6 +1154,7 @@ SEASTAR_TEST_CASE(sub_partition_read) {
});
});
});
});
});
}
@@ -1128,11 +1164,13 @@ SEASTAR_TEST_CASE(sub_partition_read) {
SEASTAR_TEST_CASE(sub_partitions_read) {
schema_ptr s = large_partition_schema();
return for_each_sstable_version([s] (const sstables::sstable::version_types version) {
return load_large_partition_sst(version).then([s] (auto sstp) {
return test_env::do_with([s, version] (test_env& env) {
return load_large_partition_sst(env, version).then([s] (auto sstp) {
return count_rows(sstp, s, "18wX", "18xB").then([] (int nrows) {
BOOST_REQUIRE(nrows == 5);
});
});
});
});
}
@@ -1158,7 +1196,7 @@ static future<> compare_files(sstring file1, sstring file2) {
// writing code, because the promoted index points to offsets in the data).
SEASTAR_TEST_CASE(promoted_index_write) {
auto s = large_partition_schema();
return test_setup::do_with_tmp_directory([s] (auto dirname) {
return test_setup::do_with_tmp_directory([s] (test_env& env, auto dirname) {
auto mtp = make_lw_shared<memtable>(s);
auto key = partition_key::from_exploded(*s, {to_bytes("v1")});
mutation m(s, key);
@@ -1178,7 +1216,7 @@ SEASTAR_TEST_CASE(promoted_index_write) {
}
}
mtp->apply(std::move(m));
auto sst = make_sstable(s, dirname, 100,
auto sst = env.make_sstable(s, dirname, 100,
sstables::sstable::version_types::la, big);
return write_memtable_to_sstable_for_test(*mtp, sst).then([s, dirname] {
auto large_partition_file = seastar::format("{}/la-3-big-Index.db", get_test_dir("large_partition", s));

44
tests/sstable_test.hh
View File

@@ -31,12 +31,13 @@
#include <seastar/core/thread.hh>
#include "sstables/index_reader.hh"
#include "tests/test_services.hh"
#include "tests/sstable_test_env.hh"
#include "tmpdir.hh"
#include <boost/filesystem.hpp>
#include <array>
static auto la = sstables::sstable::version_types::la;
static auto big = sstables::sstable::format_types::big;
constexpr auto la = sstables::sstable::version_types::la;
constexpr auto big = sstables::sstable::format_types::big;
class column_family_test {
lw_shared_ptr<column_family> _cf;
@@ -171,11 +172,6 @@ public:
});
}
static sstable_ptr make_test_sstable(size_t buffer_size, schema_ptr schema, sstring dir, unsigned long generation, sstable::version_types v,
sstable::format_types f, gc_clock::time_point now = gc_clock::now()) {
return sstables::make_sstable(std::move(schema), dir, generation, v, f, now, default_io_error_handler_gen(), buffer_size);
}
// Used to create synthetic sstables for testing leveled compaction strategy.
void set_values_for_leveled_strategy(uint64_t fake_data_size, uint32_t sstable_level, int64_t max_timestamp, sstring first_key, sstring last_key) {
_sst->_data_file_size = fake_data_size;
@@ -249,19 +245,6 @@ future<> for_each_sstable_version(AsyncAction action) {
return seastar::do_for_each(all_sstable_versions, std::move(action));
}
inline future<sstable_ptr> reusable_sst(schema_ptr schema, sstring dir,
unsigned long generation, sstable_version_types version = la) {
auto sst = sstables::make_sstable(std::move(schema), dir, generation, version, big);
auto fut = sst->load();
return std::move(fut).then([sst = std::move(sst)] {
return make_ready_future<sstable_ptr>(std::move(sst));
});
}
inline future<> working_sst(schema_ptr schema, sstring dir, unsigned long generation) {
return reusable_sst(std::move(schema), dir, generation).then([] (auto ptr) { return make_ready_future<>(); });
}
inline schema_ptr composite_schema() {
static thread_local auto s = [] {
schema_builder builder(make_lw_shared(schema({}, "tests", "composite",
@@ -692,26 +675,16 @@ public:
});
}
static future<> do_with_test_directory(std::function<future<> ()>&& fut, sstring p = path()) {
return seastar::async([p, fut = std::move(fut)] {
storage_service_for_tests ssft;
boost::filesystem::create_directories(std::string(p));
fut().get();
test_setup::empty_test_dir(p).get();
// FIXME: this removes only the last component in the path that often contains: `test-name/ks/cf-uuid'
engine().remove_file(p).get();
});
}
static future<> do_with_tmp_directory(std::function<future<> (sstring tmpdir_path)>&& fut) {
static future<> do_with_tmp_directory(std::function<future<> (test_env&, sstring tmpdir_path)>&& fut) {
return seastar::async([fut = std::move(fut)] {
storage_service_for_tests ssft;
auto tmp = tmpdir();
fut(tmp.path().string()).get();
test_env env;
fut(env, tmp.path().string()).get();
});
}
static future<> do_with_cloned_tmp_directory(sstring src, std::function<future<> (sstring srcdir_path, sstring destdir_path)>&& fut) {
static future<> do_with_cloned_tmp_directory(sstring src, std::function<future<> (test_env&, sstring srcdir_path, sstring destdir_path)>&& fut) {
return seastar::async([fut = std::move(fut), src = std::move(src)] {
storage_service_for_tests ssft;
auto src_dir = tmpdir();
@@ -721,7 +694,8 @@ public:
}
auto dest_path = dest_dir.path() / src.c_str();
seastar::compat::filesystem::create_directories(dest_path);
fut(src_dir.path().string(), dest_path.string()).get();
test_env env;
fut(env, src_dir.path().string(), dest_path.string()).get();
});
}
};

76
tests/sstable_test_env.hh Normal file
View File

@@ -0,0 +1,76 @@
/*
* Copyright (C) 2019 ScyllaDB
*/
/*
* This file is part of Scylla.
*
* Scylla is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* Scylla is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with Scylla. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include <seastar/core/do_with.hh>
#include "sstables/sstables.hh"
#include "tests/tmpdir.hh"
namespace sstables {
class test_env {
public:
test_env() = default;
shared_sstable make_sstable(schema_ptr schema, sstring dir, unsigned long generation,
sstable::version_types v, sstable::format_types f = sstable::format_types::big,
size_t buffer_size = default_sstable_buffer_size(), gc_clock::time_point now = gc_clock::now()) {
return sstables::make_sstable(std::move(schema), dir, generation, v, f, now, default_io_error_handler_gen(), buffer_size);
}
future<shared_sstable> reusable_sst(schema_ptr schema, sstring dir, unsigned long generation,
sstable::version_types version = sstable::version_types::la, sstable::format_types f = sstable::format_types::big) {
auto sst = make_sstable(std::move(schema), dir, generation, version, f);
return sst->load().then([sst = std::move(sst)] {
return make_ready_future<shared_sstable>(std::move(sst));
});
}
future<> working_sst(schema_ptr schema, sstring dir, unsigned long generation) {
return reusable_sst(std::move(schema), dir, generation).then([] (auto ptr) { return make_ready_future<>(); });
}
template <typename Func>
static inline auto do_with(Func&& func) {
return seastar::do_with(test_env(), [func = std::move(func)] (test_env& env) mutable {
return func(env);
});
}
template <typename T, typename Func>
static inline auto do_with(T&& rval, Func&& func) {
return seastar::do_with(test_env(), std::forward<T>(rval), [func = std::move(func)] (test_env& env, T& val) mutable {
return func(env, val);
});
}
template <typename Func>
static inline auto do_with_async(Func&& func) {
return seastar::async([func = std::move(func)] {
test_env env;
func(env);
});
}
};
} // namespace sstables

70
tests/test_services.cc Normal file
View File

@@ -0,0 +1,70 @@
/*
* Copyright (C) 2019 ScyllaDB
*/
/*
* This file is part of Scylla.
*
* Scylla is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* Scylla is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with Scylla. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tests/test_services.hh"
#include "dht/i_partitioner.hh"
#include "dht/murmur3_partitioner.hh"
dht::token create_token_from_key(sstring key) {
sstables::key_view key_view = sstables::key_view(bytes_view(reinterpret_cast<const signed char*>(key.c_str()), key.size()));
dht::token token = dht::global_partitioner().get_token(key_view);
assert(token == dht::global_partitioner().get_token(key_view));
return token;
}
range<dht::token> create_token_range_from_keys(sstring start_key, sstring end_key) {
dht::token start = create_token_from_key(start_key);
assert(engine().cpu_id() == dht::global_partitioner().shard_of(start));
dht::token end = create_token_from_key(end_key);
assert(engine().cpu_id() == dht::global_partitioner().shard_of(end));
assert(end >= start);
return range<dht::token>::make(start, end);
}
static db::nop_large_data_handler nop_lp_handler;
static const sstring some_keyspace("ks");
static const sstring some_column_family("cf");
column_family::config column_family_test_config() {
column_family::config cfg;
cfg.large_data_handler = &nop_lp_handler;
return cfg;
}
column_family_for_tests::column_family_for_tests()
: column_family_for_tests(
schema_builder(some_keyspace, some_column_family)
.with_column(utf8_type->decompose("p1"), utf8_type, column_kind::partition_key)
.build()
)
{ }
column_family_for_tests::column_family_for_tests(schema_ptr s)
: _data(make_lw_shared<data>())
{
_data->s = s;
_data->cfg.enable_disk_writes = false;
_data->cfg.enable_commitlog = false;
_data->cfg.large_data_handler = &nop_lp_handler;
_data->cf = make_lw_shared<column_family>(_data->s, _data->cfg, column_family::no_commitlog(), _data->cm, _data->cl_stats, _data->tracker);
_data->cf->mark_ready_for_writes();
}

36
tests/test_services.hh
View File

@@ -23,6 +23,14 @@
#pragma once
#include <memory>
#include <seastar/core/reactor.hh>
#include "schema.hh"
#include "schema_builder.hh"
#include "row_cache.hh"
#include "database.hh"
#include "cell_locking.hh"
#include "sstables/compaction_manager.hh"
// Includes: database, auth, storage_service
class storage_service_for_tests {
@@ -32,3 +40,31 @@ public:
storage_service_for_tests();
~storage_service_for_tests();
};
column_family::config column_family_test_config();
struct column_family_for_tests {
struct data {
schema_ptr s;
cache_tracker tracker;
column_family::config cfg;
cell_locker_stats cl_stats;
compaction_manager cm;
lw_shared_ptr<column_family> cf;
};
lw_shared_ptr<data> _data;
column_family_for_tests();
explicit column_family_for_tests(schema_ptr s);
schema_ptr schema() { return _data->s; }
operator lw_shared_ptr<column_family>() { return _data->cf; }
column_family& operator*() { return *_data->cf; }
column_family* operator->() { return _data->cf.get(); }
};
dht::token create_token_from_key(sstring key);
range<dht::token> create_token_range_from_keys(sstring start_key, sstring end_key);