bahman/scylla
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

test: add verify_mutation helpers.

Browse Source 
Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
This commit is contained in:
Benny Halevy
2023-03-10 16:49:04 +02:00
parent cf4eaa1fbc
commit d62df5cac6
7 changed files with 132 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
test/boost/sstable_compaction_test.cc
View File

@@ -307,6 +307,9 @@ static future<std::vector<unsigned long>> compact_sstables(test_env& env, std::v
builder.set_compressor_params(compression_parameters::no_compression());
builder.set_min_compaction_threshold(4);
auto s = builder.build(schema_builder::compact_storage::no);
auto sst_gen = [&, generation = make_lw_shared<unsigned long>(create_sstables ? generations[0] : new_generation)] () mutable {
return env.make_sstable(s, (*generation)++);
};
auto cf = env.make_table_for_tests(s);
auto stop_cf = deferred_stop(cf);
@@ -333,14 +336,15 @@ static future<std::vector<unsigned long>> compact_sstables(test_env& env, std::v
m.set_clustered_cell(c_key, r1_col, make_atomic_cell(utf8_type, bytes(min_sstable_size, 'a')));
mt->apply(std::move(m));
auto sst = make_sstable_containing(env.make_sstable(s, generation), mt);
auto sst = make_sstable_containing(sst_gen, mt);
sstables.push_back(sst);
}
}
auto new_sstable = [&] {
auto gen = new_generation++;
created.push_back(gen);
return env.make_sstable(s, gen);
auto sst = sst_gen();
created.push_back(sst->generation().value());
return sst;
};
// We must have opened at least all original candidates.
BOOST_REQUIRE(generations.size() == sstables.size());
@@ -1193,7 +1197,7 @@ SEASTAR_TEST_CASE(test_sstable_max_local_deletion_time_2) {
make_atomic_cell(utf8_type, bytes(""), ttl, last_expiry));
mt->apply(std::move(m));
};
auto get_usable_sst = [&] (lw_shared_ptr<replica::memtable>) {
auto get_usable_sst = [&] (lw_shared_ptr<replica::memtable> mt) {
return make_sstable_containing(sst_gen, mt);
};

141
test/boost/sstable_datafile_test.cc
View File

@@ -175,12 +175,7 @@ SEASTAR_TEST_CASE(datafile_generation_11) {
});
};
auto sstp = make_sstable_containing(env.make_sstable(s), mt);
std::invoke([&] {
auto pr = dht::partition_range::make_singular(make_dkey(s, "key1"));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mutation = read_mutation_from_flat_mutation_reader(rd).get();
auto sstp = verify_mutation(env, env.make_sstable(s), mt, "key1", [&] (mutation_opt& mutation) {
auto verify_set = [&tomb] (const collection_mutation_description& m) {
BOOST_REQUIRE(bool(m.tomb) == true);
BOOST_REQUIRE(m.tomb == tomb);
@@ -190,7 +185,6 @@ SEASTAR_TEST_CASE(datafile_generation_11) {
BOOST_REQUIRE(m.cells[2].first == to_bytes("3"));
};
auto& mp = mutation->partition();
auto& ssr = mp.static_row();
auto scol = ssr.find_cell(static_set_col.id);
@@ -206,11 +200,7 @@ SEASTAR_TEST_CASE(datafile_generation_11) {
verify_set(m);
});
std::invoke([&] {
auto pr = dht::partition_range::make_singular(make_dkey(s, "key2"));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mutation = read_mutation_from_flat_mutation_reader(rd).get();
verify_mutation(env, sstp, "key2", [&] (mutation_opt& mutation) {
auto m = verifier(mutation);
BOOST_REQUIRE(!m.tomb);
BOOST_REQUIRE(m.cells.size() == 1);
@@ -234,16 +224,13 @@ SEASTAR_TEST_CASE(datafile_generation_12) {
m.partition().apply_delete(*s, cp, tomb);
mt->apply(std::move(m));
auto sstp = make_sstable_containing(env.make_sstable(s), mt);
auto pr = dht::partition_range::make_singular(make_dkey(s, "key1"));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mutation = read_mutation_from_flat_mutation_reader(rd).get();
auto& mp = mutation->partition();
BOOST_REQUIRE(mp.row_tombstones().size() == 1);
for (auto& rt: mp.row_tombstones()) {
BOOST_REQUIRE(rt.tombstone().tomb == tomb);
}
verify_mutation(env, env.make_sstable(s), mt, "key1", [&] (mutation_opt& mutation) {
auto& mp = mutation->partition();
BOOST_REQUIRE(mp.row_tombstones().size() == 1);
for (auto& rt: mp.row_tombstones()) {
BOOST_REQUIRE(rt.tombstone().tomb == tomb);
}
});
});
}
@@ -265,16 +252,13 @@ static future<> sstable_compression_test(compressor_ptr c) {
m.partition().apply_delete(*s, cp, tomb);
mtp->apply(std::move(m));
auto sstp = make_sstable_containing(env.make_sstable(s), mtp);
auto pr = dht::partition_range::make_singular(make_dkey(s, "key1"));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mutation = read_mutation_from_flat_mutation_reader(rd).get();
auto& mp = mutation->partition();
BOOST_REQUIRE(mp.row_tombstones().size() == 1);
for (auto& rt: mp.row_tombstones()) {
BOOST_REQUIRE(rt.tombstone().tomb == tomb);
}
verify_mutation(env, env.make_sstable(s), mtp, "key1", [&] (mutation_opt& mutation) {
auto& mp = mutation->partition();
BOOST_REQUIRE(mp.row_tombstones().size() == 1);
for (auto& rt: mp.row_tombstones()) {
BOOST_REQUIRE(rt.tombstone().tomb == tomb);
}
});
});
}
@@ -337,17 +321,14 @@ SEASTAR_TEST_CASE(datafile_generation_37) {
m.set_clustered_cell(c_key, cl2, make_atomic_cell(bytes_type, bytes_type->decompose(data_value(to_bytes("cl2")))));
mtp->apply(std::move(m));
auto sstp = make_sstable_containing(env.make_sstable(s), mtp);
auto pr = dht::partition_range::make_singular(make_dkey(s, "key1"));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mutation = read_mutation_from_flat_mutation_reader(rd).get();
auto& mp = mutation->partition();
verify_mutation(env, env.make_sstable(s), mtp, "key1", [&] (mutation_opt& mutation) {
auto& mp = mutation->partition();
auto clustering = clustering_key_prefix::from_exploded(*s, {to_bytes("cl1")});
auto clustering = clustering_key_prefix::from_exploded(*s, {to_bytes("cl1")});
auto& row = mp.clustered_row(*s, clustering);
match_live_cell(row.cells(), *s, "cl2", data_value(to_bytes("cl2")));
auto& row = mp.clustered_row(*s, clustering);
match_live_cell(row.cells(), *s, "cl2", data_value(to_bytes("cl2")));
});
});
}
@@ -366,16 +347,13 @@ SEASTAR_TEST_CASE(datafile_generation_38) {
m.set_clustered_cell(c_key, cl3, make_atomic_cell(bytes_type, bytes_type->decompose(data_value(to_bytes("cl3")))));
mtp->apply(std::move(m));
auto sstp = make_sstable_containing(env.make_sstable(s), mtp);
auto pr = dht::partition_range::make_singular(make_dkey(s, "key1"));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mutation = read_mutation_from_flat_mutation_reader(rd).get();
auto& mp = mutation->partition();
auto clustering = clustering_key_prefix::from_exploded(*s, {to_bytes("cl1"), to_bytes("cl2")});
verify_mutation(env, env.make_sstable(s), mtp, "key1", [&] (mutation_opt& mutation) {
auto& mp = mutation->partition();
auto clustering = clustering_key_prefix::from_exploded(*s, {to_bytes("cl1"), to_bytes("cl2")});
auto& row = mp.clustered_row(*s, clustering);
match_live_cell(row.cells(), *s, "cl3", data_value(to_bytes("cl3")));
auto& row = mp.clustered_row(*s, clustering);
match_live_cell(row.cells(), *s, "cl3", data_value(to_bytes("cl3")));
});
});
}
@@ -396,15 +374,12 @@ SEASTAR_TEST_CASE(datafile_generation_39) {
m.set_clustered_cell(c_key, cl2, make_atomic_cell(bytes_type, bytes_type->decompose(data_value(to_bytes("cl2")))));
mtp->apply(std::move(m));
auto sstp = make_sstable_containing(env.make_sstable(s), mtp);
auto pr = dht::partition_range::make_singular(make_dkey(s, "key1"));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mutation = read_mutation_from_flat_mutation_reader(rd).get();
auto& mp = mutation->partition();
auto& row = mp.clustered_row(*s, clustering_key::make_empty());
match_live_cell(row.cells(), *s, "cl1", data_value(data_value(to_bytes("cl1"))));
match_live_cell(row.cells(), *s, "cl2", data_value(data_value(to_bytes("cl2"))));
verify_mutation(env, env.make_sstable(s), mtp, "key1", [&] (mutation_opt& mutation) {
auto& mp = mutation->partition();
auto& row = mp.clustered_row(*s, clustering_key::make_empty());
match_live_cell(row.cells(), *s, "cl1", data_value(data_value(to_bytes("cl1"))));
match_live_cell(row.cells(), *s, "cl2", data_value(data_value(to_bytes("cl2"))));
});
});
}
@@ -423,15 +398,12 @@ SEASTAR_TEST_CASE(datafile_generation_41) {
m.partition().apply_delete(*s, std::move(c_key), tomb);
mt->apply(std::move(m));
auto sstp = make_sstable_containing(env.make_sstable(s), mt);
auto pr = dht::partition_range::make_singular(make_dkey(s, "key1"));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mutation = read_mutation_from_flat_mutation_reader(rd).get();
auto& mp = mutation->partition();
BOOST_REQUIRE(mp.clustered_rows().calculate_size() == 1);
auto& c_row = *(mp.clustered_rows().begin());
BOOST_REQUIRE(c_row.row().deleted_at().tomb() == tomb);
verify_mutation(env, env.make_sstable(s), mt, "key1", [&] (mutation_opt& mutation) {
auto& mp = mutation->partition();
BOOST_REQUIRE(mp.clustered_rows().calculate_size() == 1);
auto& c_row = *(mp.clustered_rows().begin());
BOOST_REQUIRE(c_row.row().deleted_at().tomb() == tomb);
});
});
}
@@ -971,7 +943,7 @@ static void check_min_max_column_names(const sstable_ptr& sst, std::vector<bytes
}
}
static void test_min_max_clustering_key(test_env& env, schema_ptr s, std::vector<bytes> exploded_pk, std::vector<std::vector<bytes>> exploded_cks,
static void test_min_max_clustering_key(test_env& env, schema_ptr s, std::function<shared_sstable()> sst_gen, std::vector<bytes> exploded_pk, std::vector<std::vector<bytes>> exploded_cks,
std::vector<bytes> min_components, std::vector<bytes> max_components, sstable_version_types version, bool remove = false) {
auto mt = make_lw_shared<replica::memtable>(s);
auto insert_data = [&mt, &s] (std::vector<bytes>& exploded_pk, std::vector<bytes>&& exploded_ck) {
@@ -1020,7 +992,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("ck2", utf8_type, column_kind::clustering_key)
.with_column("r1", int32_type)
.build();
test_min_max_clustering_key(env, s, {"key1"}, {{"a", "b"},
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"a", "b"},
{"a", "c"}}, {"a", "b"}, {"a", "c"}, version);
}
{
@@ -1031,7 +1004,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("ck2", utf8_type, column_kind::clustering_key)
.with_column("r1", int32_type)
.build();
test_min_max_clustering_key(env, s, {"key1"}, {{"a", "b"},
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"a", "b"},
{"a", "c"}}, {"a", "b"}, {"a", "c"}, version);
}
{
@@ -1042,7 +1016,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("r1", int32_type)
.build();
BOOST_TEST_MESSAGE(fmt::format("min_max_clustering_key_test: min={{\"a\", \"c\"}} max={{\"b\", \"a\"}} version={}", to_string(version)));
test_min_max_clustering_key(env, s, {"key1"}, {{"b", "a"}, {"a", "c"}}, {"a", "c"}, {"b", "a"}, version);
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"b", "a"}, {"a", "c"}}, {"a", "c"}, {"b", "a"}, version);
}
{
auto s = schema_builder("ks", "cf")
@@ -1053,7 +1028,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("r1", int32_type)
.build();
BOOST_TEST_MESSAGE(fmt::format("min_max_clustering_key_test: min={{\"a\", \"c\"}} max={{\"b\", \"a\"}} with compact storage version={}", to_string(version)));
test_min_max_clustering_key(env, s, {"key1"}, {{"b", "a"}, {"a", "c"}}, {"a", "c"}, {"b", "a"}, version);
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"b", "a"}, {"a", "c"}}, {"a", "c"}, {"b", "a"}, version);
}
{
auto s = schema_builder("ks", "cf")
@@ -1063,7 +1039,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("r1", int32_type)
.build();
BOOST_TEST_MESSAGE(fmt::format("min_max_clustering_key_test: reversed order: min={{\"a\", \"z\"}} max={{\"a\", \"a\"}} version={}", to_string(version)));
test_min_max_clustering_key(env, s, {"key1"}, {{"a", "a"}, {"a", "z"}}, {"a", "z"}, {"a", "a"}, version);
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"a", "a"}, {"a", "z"}}, {"a", "z"}, {"a", "a"}, version);
}
{
auto s = schema_builder("ks", "cf")
@@ -1073,7 +1050,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("r1", int32_type)
.build();
BOOST_TEST_MESSAGE(fmt::format("min_max_clustering_key_test: reversed order: min={{\"a\", \"a\"}} max={{\"b\", \"z\"}} version={}", to_string(version)));
test_min_max_clustering_key(env, s, {"key1"}, {{"b", "z"}, {"a", "a"}}, {"a", "a"}, {"b", "z"}, version);
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"b", "z"}, {"a", "a"}}, {"a", "a"}, {"b", "z"}, version);
}
{
auto s = schema_builder("ks", "cf")
@@ -1081,7 +1059,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("ck1", utf8_type, column_kind::clustering_key)
.with_column("r1", int32_type)
.build();
test_min_max_clustering_key(env, s, {"key1"}, {{"a"},
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"a"},
{"z"}}, {"a"}, {"z"}, version);
}
{
@@ -1090,7 +1069,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("ck1", utf8_type, column_kind::clustering_key)
.with_column("r1", int32_type)
.build();
test_min_max_clustering_key(env, s, {"key1"}, {{"a"},
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"a"},
{"z"}}, {"a"}, {"z"}, version, true);
}
{
@@ -1098,7 +1078,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("pk", utf8_type, column_kind::partition_key)
.with_column("r1", int32_type)
.build();
test_min_max_clustering_key(env, s, {"key1"}, {}, {}, {}, version);
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {}, {}, {}, version);
}
if (version >= sstable_version_types::mc) {
{
@@ -1110,7 +1091,8 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test) {
.with_column("r1", int32_type)
.build();
BOOST_TEST_MESSAGE(fmt::format("min_max_clustering_key_test: reversed order: min={{\"a\"}} max={{\"a\"}} with compact storage version={}", to_string(version)));
test_min_max_clustering_key(env, s, {"key1"}, {{"a", "z"}, {"a"}}, {"a"}, {"a"}, version);
auto sst_gen = env.make_sst_factory(s, version);
test_min_max_clustering_key(env, s, sst_gen, {"key1"}, {{"a", "z"}, {"a"}}, {"a"}, {"a"}, version);
}
}
}
@@ -1504,7 +1486,6 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
.with_column("r1", int32_type)
.build();
auto sst_gen = env.make_sst_factory(s, version);
auto tmp = env.tempdir().make_sweeper();
auto key = partition_key::from_exploded(*s, {to_bytes("key1")});
auto c_key = clustering_key_prefix::from_exploded(*s, {to_bytes("c1"), to_bytes("c2")});
const column_definition& r1_col = *s->get_column_definition("r1");

28
test/boost/sstable_mutation_test.cc
View File

@@ -407,6 +407,7 @@ SEASTAR_TEST_CASE(test_sstable_can_write_and_read_range_tombstone) {
return test_env::do_with_async([] (test_env& env) {
auto s = make_shared_schema({}, "ks", "cf",
{{"p1", utf8_type}}, {{"c1", int32_type}}, {{"r1", int32_type}}, {}, utf8_type);
auto sst_gen = env.make_sst_factory(s);
auto key = tests::generate_partition_key(s);
auto c_key_start = clustering_key::from_exploded(*s, {int32_type->decompose(1)});
@@ -419,20 +420,19 @@ SEASTAR_TEST_CASE(test_sstable_can_write_and_read_range_tombstone) {
auto mt = make_lw_shared<replica::memtable>(s);
mt->apply(std::move(m));
auto sst = make_sstable_containing(env.make_sstable(s), mt);
auto mut = with_closeable(sst->make_reader(s, env.make_reader_permit(), query::full_partition_range, s->full_slice()), [] (auto& mr) {
return read_mutation_from_flat_mutation_reader(mr);
}).get0();
BOOST_REQUIRE(bool(mut));
auto rts = mut->partition().row_tombstones();
BOOST_REQUIRE(rts.size() == 1);
auto it = rts.begin();
BOOST_REQUIRE(it->tombstone().equal(*s, range_tombstone(
c_key_start,
bound_kind::excl_start,
c_key_end,
bound_kind::excl_end,
tombstone(9, ttl))));
verify_mutation(env, env.make_sstable(s), mt, query::full_partition_range, [&] (mutation_opt& mut) {
BOOST_REQUIRE(bool(mut));
auto rts = mut->partition().row_tombstones();
BOOST_REQUIRE(rts.size() == 1);
auto it = rts.begin();
BOOST_REQUIRE(it->tombstone().equal(*s, range_tombstone(
c_key_start,
bound_kind::excl_start,
c_key_end,
bound_kind::excl_end,
tombstone(9, ttl))));
return stop_iteration::yes;
}).get();
});
}

1
test/boost/sstable_resharding_test.cc
View File

@@ -53,6 +53,7 @@ void run_sstable_resharding_test() {
auto s = get_schema();
auto cf = env.make_table_for_tests(s);
auto close_cf = deferred_stop(cf);
auto sst_gen = cf.make_sst_factory(version);
std::unordered_map<shard_id, std::vector<mutation>> muts;
static constexpr auto keys_per_shard = 1000u;

13
test/boost/sstable_test.hh
View File

@@ -445,3 +445,16 @@ inline dht::decorated_key make_dkey(schema_ptr s, bytes b)
auto sst_key = sstables::key::from_bytes(b);
return dht::decorate_key(*s, sst_key.to_partition_key(*s));
}
// Must be called from a seastar thread.
shared_sstable verify_mutation(test_env& env, shared_sstable sst, lw_shared_ptr<replica::memtable> mt, bytes key, std::function<void(mutation_opt&)> verify);
inline shared_sstable verify_mutation(test_env& env, std::function<shared_sstable()> sst_gen, lw_shared_ptr<replica::memtable> mt, bytes key, std::function<void(mutation_opt&)> verify) {
return verify_mutation(env, sst_gen(), std::move(mt), std::move(key), std::move(verify));
}
shared_sstable verify_mutation(test_env& env, shared_sstable sstp, bytes key, std::function<void(mutation_opt&)> verify);
shared_sstable verify_mutation(test_env& env, shared_sstable sst, lw_shared_ptr<replica::memtable> mt, dht::partition_range pr, std::function<stop_iteration(mutation_opt&)> verify);
inline shared_sstable verify_mutation(test_env& env, std::function<shared_sstable()> sst_gen, lw_shared_ptr<replica::memtable> mt, dht::partition_range pr, std::function<stop_iteration(mutation_opt&)> verify) {
return verify_mutation(env, sst_gen(), std::move(mt), std::move(pr), std::move(verify));
}
shared_sstable verify_mutation(test_env& env, shared_sstable sstp, dht::partition_range pr, std::function<stop_iteration(mutation_opt&)> verify);

32
test/lib/sstable_utils.cc
View File

@@ -208,3 +208,35 @@ void compaction_manager_test::deregister_compaction(const sstables::compaction_d
testlog.error("compaction_manager_test: deregister_compaction uuid={}: task not found", c.compaction_uuid);
}
}
shared_sstable verify_mutation(test_env& env, shared_sstable sst, lw_shared_ptr<replica::memtable> mt, bytes key, std::function<void(mutation_opt&)> verify) {
auto sstp = make_sstable_containing(std::move(sst), mt);
return verify_mutation(env, std::move(sstp), std::move(key), std::move(verify));
}
shared_sstable verify_mutation(test_env& env, shared_sstable sstp, bytes key, std::function<void(mutation_opt&)> verify) {
auto s = sstp->get_schema();
auto pr = dht::partition_range::make_singular(make_dkey(s, key));
auto rd = sstp->make_reader(s, env.make_reader_permit(), pr, s->full_slice());
auto close_rd = deferred_close(rd);
auto mopt = read_mutation_from_flat_mutation_reader(rd).get();
verify(mopt);
return sstp;
}
shared_sstable verify_mutation(test_env& env, shared_sstable sst, lw_shared_ptr<replica::memtable> mt, dht::partition_range pr, std::function<stop_iteration(mutation_opt&)> verify) {
auto sstp = make_sstable_containing(std::move(sst), mt);
return verify_mutation(env, std::move(sstp), std::move(pr), std::move(verify));
}
shared_sstable verify_mutation(test_env& env, shared_sstable sstp, dht::partition_range pr, std::function<stop_iteration(mutation_opt&)> verify) {
auto s = sstp->get_schema();
auto rd = sstp->make_reader(s, env.make_reader_permit(), std::move(pr), s->full_slice());
auto close_rd = deferred_close(rd);
while (auto mopt = read_mutation_from_flat_mutation_reader(rd).get()) {
if (verify(mopt) == stop_iteration::yes) {
break;
}
}
return sstp;
}

2
test/lib/sstable_utils.hh
View File

@@ -8,6 +8,8 @@
#pragma once
#include <optional>
#include "sstables/sstables.hh"
#include "sstables/shared_sstable.hh"
#include "sstables/index_reader.hh"