diff --git a/test/boost/sstable_3_x_test.cc b/test/boost/sstable_3_x_test.cc
index 79e5922a8a..5aaa8263ad 100644
--- a/test/boost/sstable_3_x_test.cc
+++ b/test/boost/sstable_3_x_test.cc
@@ -3244,9 +3244,8 @@ static sstables::shared_sstable write_and_compare_sstables(test_env& env, schema
 }
 
 static sstables::shared_sstable write_sstables(test_env& env, schema_ptr s, lw_shared_ptr<replica::memtable> mt, sstable_version_types version) {
-    auto sst = env.make_sstable(s, version);
+    auto sst = make_sstable_containing(env.make_sstable(s, version), mt);
     BOOST_TEST_MESSAGE(format("write_sstable from memtable: {}", sst->get_filename()));
-    write_memtable_to_sstable_for_test(*mt, sst).get();
     return sst;
 }
 
diff --git a/test/boost/sstable_compaction_test.cc b/test/boost/sstable_compaction_test.cc
index 7202561d96..2863220acf 100644
--- a/test/boost/sstable_compaction_test.cc
+++ b/test/boost/sstable_compaction_test.cc
@@ -148,6 +148,7 @@ SEASTAR_TEST_CASE(compaction_manager_basic_test) {
     auto& cm = cf.get_compaction_manager();
     auto close_cf = deferred_stop(cf);
     cf->set_compaction_strategy(sstables::compaction_strategy_type::size_tiered);
+    auto sst_gen = cf.make_sst_factory();
 
     auto idx = std::vector<unsigned long>({1, 2, 3, 4});
     for (auto i : idx) {
@@ -165,10 +166,7 @@ SEASTAR_TEST_CASE(compaction_manager_basic_test) {
         m.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
         mt->apply(std::move(m));
 
-        auto sst = cf.make_sstable();
-
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        sst->load().get();
+        auto sst = make_sstable_containing(sst_gen, mt);
         column_family_test(cf).add_sstable(sst).get();
     }
 
@@ -335,10 +333,7 @@ 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 = env.make_sstable(s, generation);
-
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst->load().get();
+                auto sst = make_sstable_containing(env.make_sstable(s, generation), mt);
                 sstables.push_back(sst);
             }
         }
@@ -1130,6 +1125,7 @@ SEASTAR_TEST_CASE(sstable_rewrite) {
     return test_env::do_with_async([] (test_env& env) {
         auto s = make_shared_schema({}, some_keyspace, some_column_family,
             {{"p1", utf8_type}}, {{"c1", utf8_type}}, {{"r1", utf8_type}}, {}, utf8_type);
+        auto sst_gen = env.make_sst_factory(s);
 
         auto mt = make_lw_shared<replica::memtable>(s);
 
@@ -1144,13 +1140,11 @@ SEASTAR_TEST_CASE(sstable_rewrite) {
         };
         apply_key(key_for_this_shard[0]);
 
-        auto sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        auto sstp = env.reusable_sst(sst).get();
+        auto sstp = make_sstable_containing(sst_gen, mt);
         auto key = key_for_this_shard[0];
         std::vector<sstables::shared_sstable> new_tables;
         auto creator = [&] {
-            auto sst = env.make_sstable(s);
+            auto sst = sst_gen();
             new_tables.emplace_back(sst);
             return sst;
         };
@@ -1188,6 +1182,7 @@ SEASTAR_TEST_CASE(test_sstable_max_local_deletion_time_2) {
                 schema_ptr s = builder.build(schema_builder::compact_storage::no);
                 auto cf = env.make_table_for_tests(s);
                 auto close_cf = deferred_stop(cf);
+                auto sst_gen = cf.make_sst_factory(version);
                 auto mt = make_lw_shared<replica::memtable>(s);
                 auto now = gc_clock::now();
                 int32_t last_expiry = 0;
@@ -1198,11 +1193,8 @@ 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 = [&env, s, version](replica::memtable &mt) -> future<sstable_ptr> {
-                    auto sst = env.make_sstable(s, version);
-                    return write_memtable_to_sstable_for_test(mt, sst).then([&env, sst] {
-                        return env.reusable_sst(sst);
-                    });
+                auto get_usable_sst = [&] (lw_shared_ptr<replica::memtable>) {
+                    return make_sstable_containing(sst_gen, mt);
                 };
 
                 mutation m(s, partition_key::from_exploded(*s, {to_bytes("deletetest")}));
@@ -1210,7 +1202,7 @@ SEASTAR_TEST_CASE(test_sstable_max_local_deletion_time_2) {
                     add_row(m, to_bytes("deletecolumn" + to_sstring(i)), 100);
                 }
                 add_row(m, to_bytes("todelete"), 1000);
-                auto sst1 = get_usable_sst(*mt).get0();
+                auto sst1 = get_usable_sst(mt);
                 BOOST_REQUIRE(last_expiry == sst1->get_stats_metadata().max_local_deletion_time);
 
                 mt = make_lw_shared<replica::memtable>(s);
@@ -1218,10 +1210,10 @@ SEASTAR_TEST_CASE(test_sstable_max_local_deletion_time_2) {
                 tombstone tomb(api::new_timestamp(), now);
                 m.partition().apply_delete(*s, clustering_key::from_exploded(*s, {to_bytes("todelete")}), tomb);
                 mt->apply(std::move(m));
-                auto sst2 = get_usable_sst(*mt).get0();
+                auto sst2 = get_usable_sst(mt);
                 BOOST_REQUIRE(now.time_since_epoch().count() == sst2->get_stats_metadata().max_local_deletion_time);
 
-                auto creator = [&env, s, version] { return env.make_sstable(s, version); };
+                auto creator = [&] { return cf.make_sstable(version); };
                 auto info = compact_sstables(sstables::compaction_descriptor({sst1, sst2}, default_priority_class()), cf, creator).get0();
                 BOOST_REQUIRE(info.new_sstables.size() == 1);
                 BOOST_REQUIRE(((now + gc_clock::duration(100)).time_since_epoch().count()) ==
@@ -1297,9 +1289,7 @@ SEASTAR_TEST_CASE(compaction_with_fully_expired_table) {
         tombstone tomb(api::new_timestamp(), gc_clock::now() - std::chrono::seconds(3600));
         m.partition().apply_delete(*s, c_key, tomb);
         mt->apply(std::move(m));
-        auto sst = sst_gen();
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        sst = env.reusable_sst(sst).get0();
+        auto sst = make_sstable_containing(sst_gen, mt);
 
         auto cf = env.make_table_for_tests(s);
         auto close_cf = deferred_stop(cf);
@@ -1677,6 +1667,7 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test_2) {
                       .build();
             auto cf = env.make_table_for_tests(s);
             auto close_cf = deferred_stop(cf);
+            auto sst_gen = cf.make_sst_factory(version);
             auto mt = make_lw_shared<replica::memtable>(s);
             const column_definition &r1_col = *s->get_column_definition("r1");
 
@@ -1689,9 +1680,7 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test_2) {
                 }
                 mt->apply(std::move(m));
             }
-            auto sst = env.make_sstable(s, version);
-            write_memtable_to_sstable_for_test(*mt, sst).get();
-            sst = env.reusable_sst(sst).get0();
+            auto sst = make_sstable_containing(sst_gen, mt);
             check_min_max_column_names(sst, {"0ck100"}, {"7ck149"});
 
             mt = make_lw_shared<replica::memtable>(s);
@@ -1702,12 +1691,10 @@ SEASTAR_TEST_CASE(min_max_clustering_key_test_2) {
                 m.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
             }
             mt->apply(std::move(m));
-            auto sst2 = env.make_sstable(s, version);
-            write_memtable_to_sstable_for_test(*mt, sst2).get();
-            sst2 = env.reusable_sst(sst2).get0();
+            auto sst2 = make_sstable_containing(sst_gen, mt);
             check_min_max_column_names(sst2, {"9ck101"}, {"9ck298"});
 
-            auto creator = [&env, s, version] { return env.make_sstable(s, version); };
+            auto creator = [&] { return cf.make_sstable(version); };
             auto info = compact_sstables(sstables::compaction_descriptor({sst, sst2}, default_priority_class()), cf, creator).get0();
             BOOST_REQUIRE(info.new_sstables.size() == 1);
             check_min_max_column_names(info.new_sstables.front(), {"0ck100"}, {"9ck298"});
@@ -1739,6 +1726,9 @@ SEASTAR_TEST_CASE(sstable_expired_data_ratio) {
     return test_env::do_with_async([] (test_env& env) {
         auto s = make_shared_schema({}, some_keyspace, some_column_family,
             {{"p1", utf8_type}}, {{"c1", utf8_type}}, {{"r1", utf8_type}}, {}, utf8_type);
+        auto cf = env.make_table_for_tests(s);
+        auto close_cf = deferred_stop(cf);
+        auto sst_gen = cf.make_sst_factory();
 
         auto mt = make_lw_shared<replica::memtable>(s);
 
@@ -1767,9 +1757,7 @@ SEASTAR_TEST_CASE(sstable_expired_data_ratio) {
         for (auto i = 0; i < remaining; i++) {
             insert_key(to_bytes("key" + to_sstring(i)), 3600, expiration_time);
         }
-        auto sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        sst = env.reusable_sst(sst).get0();
+        auto sst = make_sstable_containing(sst_gen, mt);
         const auto& stats = sst->get_stats_metadata();
         BOOST_REQUIRE(stats.estimated_tombstone_drop_time.bin.size() == sstables::TOMBSTONE_HISTOGRAM_BIN_SIZE);
         auto gc_before = gc_clock::now() - s->gc_grace_seconds();
@@ -1780,12 +1768,7 @@ SEASTAR_TEST_CASE(sstable_expired_data_ratio) {
         run.insert(sst);
         BOOST_REQUIRE(std::fabs(run.estimate_droppable_tombstone_ratio(gc_before) - expired) <= 0.1);
 
-        auto cf = env.make_table_for_tests(s);
-        auto close_cf = deferred_stop(cf);
-        auto creator = [&] {
-            auto sst = env.make_sstable(s);
-            return sst;
-        };
+        auto creator = sst_gen;
         auto info = compact_sstables(sstables::compaction_descriptor({ sst }, default_priority_class()), cf, creator).get0();
         BOOST_REQUIRE(info.new_sstables.size() == 1);
         BOOST_REQUIRE(info.new_sstables.front()->estimate_droppable_tombstone_ratio(gc_before) == 0.0f);
diff --git a/test/boost/sstable_datafile_test.cc b/test/boost/sstable_datafile_test.cc
index 03aa8de09c..938018e0cd 100644
--- a/test/boost/sstable_datafile_test.cc
+++ b/test/boost/sstable_datafile_test.cc
@@ -106,10 +106,8 @@ SEASTAR_TEST_CASE(datafile_generation_09) {
         m.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
         mt->apply(std::move(m));
 
-        auto sst = env.make_sstable(s);
-
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        auto sst2 = env.make_sstable(s, sst->generation().value());
+        auto sst = make_sstable_containing(env.make_sstable(s), mt);
+        auto sst2 = env.reusable_sst(sst).get();
 
         sstables::test(sst2).read_summary().get();
         summary& sst1_s = sstables::test(sst).get_summary();
@@ -177,9 +175,7 @@ SEASTAR_TEST_CASE(datafile_generation_11) {
             });
         };
 
-        auto sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        auto sstp = env.reusable_sst(s, sst).get();
+        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());
@@ -238,9 +234,7 @@ SEASTAR_TEST_CASE(datafile_generation_12) {
         m.partition().apply_delete(*s, cp, tomb);
         mt->apply(std::move(m));
 
-        auto sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        auto sstp = env.reusable_sst(s, sst).get();
+        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);
@@ -253,8 +247,8 @@ SEASTAR_TEST_CASE(datafile_generation_12) {
     });
 }
 
-static future<> sstable_compression_test(compressor_ptr c, unsigned generation) {
-    return test_env::do_with_async([c, generation] (test_env& env) {
+static future<> sstable_compression_test(compressor_ptr c) {
+    return test_env::do_with_async([c] (test_env& env) {
         // NOTE: set a given compressor algorithm to schema.
         schema_builder builder(complex_schema());
         builder.set_compressor_params(c);
@@ -271,9 +265,7 @@ static future<> sstable_compression_test(compressor_ptr c, unsigned generation)
         m.partition().apply_delete(*s, cp, tomb);
         mtp->apply(std::move(m));
 
-        auto sst = env.make_sstable(s, generation);
-        write_memtable_to_sstable_for_test(*mtp, sst).get();
-        auto sstp = env.reusable_sst(s, sst).get();
+        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);
@@ -287,15 +279,15 @@ static future<> sstable_compression_test(compressor_ptr c, unsigned generation)
 }
 
 SEASTAR_TEST_CASE(datafile_generation_13) {
-    return sstable_compression_test(compressor::lz4, 13);
+    return sstable_compression_test(compressor::lz4);
 }
 
 SEASTAR_TEST_CASE(datafile_generation_14) {
-    return sstable_compression_test(compressor::snappy, 14);
+    return sstable_compression_test(compressor::snappy);
 }
 
 SEASTAR_TEST_CASE(datafile_generation_15) {
-    return sstable_compression_test(compressor::deflate, 15);
+    return sstable_compression_test(compressor::deflate);
 }
 
 SEASTAR_TEST_CASE(datafile_generation_16) {
@@ -314,10 +306,9 @@ SEASTAR_TEST_CASE(datafile_generation_16) {
             mtp->apply(std::move(m));
         }
 
-        auto sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mtp, sst).get();
-        env.reusable_sst(s, sst).get();
+        auto sst = make_sstable_containing(env.make_sstable(s), mtp);
         // Not crashing is enough
+        BOOST_REQUIRE(sst);
     });
 }
 
@@ -346,9 +337,7 @@ 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 sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mtp, sst).get();
-        auto sstp = env.reusable_sst(s, sst).get();
+        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);
@@ -377,9 +366,7 @@ 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 sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mtp, sst).get();
-        auto sstp = env.reusable_sst(s, sst).get();
+        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);
@@ -409,9 +396,7 @@ 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 sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mtp, sst).get();
-        auto sstp = env.reusable_sst(s, sst).get();
+        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);
@@ -438,9 +423,7 @@ SEASTAR_TEST_CASE(datafile_generation_41) {
         m.partition().apply_delete(*s, std::move(c_key), tomb);
         mt->apply(std::move(m));
 
-        auto sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        auto sstp = env.reusable_sst(s, sst).get();
+        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);
@@ -468,9 +451,7 @@ SEASTAR_TEST_CASE(datafile_generation_47) {
         m.set_clustered_cell(c_key, r1_col, make_atomic_cell(utf8_type, bytes(512*1024, 'a')));
         mt->apply(std::move(m));
 
-        auto sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        auto sstp = env.reusable_sst(s, sst).get();
+        auto sstp = make_sstable_containing(env.make_sstable(s), mt);
         auto reader = sstable_reader_v2(sstp, s, env.make_reader_permit());
         auto close_reader = deferred_close(reader);
         while (reader().get()) {
@@ -517,10 +498,7 @@ SEASTAR_TEST_CASE(test_counter_write) {
 
         mt->apply(m);
 
-        auto sst = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-
-        auto sstp = env.reusable_sst(s, sst).get0();
+        auto sstp = make_sstable_containing(env.make_sstable(s), mt);
         assert_that(sstable_reader_v2(sstp, s, env.make_reader_permit()))
             .produces(m)
             .produces_end_of_stream();
@@ -915,9 +893,7 @@ SEASTAR_TEST_CASE(test_sstable_max_local_deletion_time) {
                                          make_atomic_cell(utf8_type, bytes("a"), 3600 + i, last_expiry));
                     mt->apply(std::move(m));
                 }
-                auto sst = env.make_sstable(s, 53, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                auto sstp = env.reusable_sst(sst).get0();
+                auto sstp = make_sstable_containing(env.make_sstable(s, version), mt);
                 BOOST_REQUIRE(last_expiry == sstp->get_stats_metadata().max_local_deletion_time);
             }
     });
@@ -1029,10 +1005,7 @@ static void test_min_max_clustering_key(test_env& env, schema_ptr s, std::vector
             }
         }
     }
-    auto tmp = env.tempdir().make_sweeper();
-    auto sst = env.make_sstable(s, 1, version);
-    write_memtable_to_sstable_for_test(*mt, sst).get();
-    sst = env.reusable_sst(sst).get0();
+    auto sst = make_sstable_containing(env.make_sstable(s, version), mt);
     check_min_max_column_names(sst, std::move(min_components), std::move(max_components));
     sst->unlink().get();
 }
@@ -1152,6 +1125,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                     .with_column("ck1", utf8_type, column_kind::clustering_key)
                     .with_column("r1", int32_type)
                     .build();
+            auto sst_gen = env.make_sst_factory(s, version);
             auto key = partition_key::from_exploded(*s, {to_bytes("key1")});
             auto c_key = clustering_key_prefix::from_exploded(*s, {to_bytes("c1")});
             const column_definition& r1_col = *s->get_column_definition("r1");
@@ -1164,9 +1138,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                 tombstone tomb(api::new_timestamp(), gc_clock::now());
                 m.partition().apply_delete(*s, c_key, tomb);
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 1, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1"}, {"c1"});
             }
@@ -1176,9 +1148,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                 mutation m(s, key);
                 m.set_clustered_cell(c_key, r1_col, make_dead_atomic_cell(3600));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 2, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1"}, {"c1"});
             }
@@ -1188,9 +1158,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                 mutation m(s, key);
                 m.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 3, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(!sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1"}, {"c1"});
             }
@@ -1208,9 +1176,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                 m2.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
                 mt->apply(std::move(m2));
 
-                auto sst = env.make_sstable(s, 4, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1"}, {"c1"});
             }
@@ -1221,9 +1187,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                 tombstone tomb(api::new_timestamp(), gc_clock::now());
                 m.partition().apply(tomb);
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 5, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {}, {});
             }
@@ -1236,9 +1200,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                 "a")), clustering_key_prefix::from_single_value(*s, bytes("a")), tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 6, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"a"}, {"a"});
@@ -1256,9 +1218,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 7, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"a"}, {"c1"});
@@ -1276,9 +1236,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 8, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"c"}, {"d"});
@@ -1296,9 +1254,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 9, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"c1"}, {"z"});
@@ -1317,9 +1273,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                             tomb);
                     m.partition().apply_delete(*s, std::move(rt));
                     mt->apply(std::move(m));
-                    auto sst = env.make_sstable(s, 10, version);
-                    write_memtable_to_sstable_for_test(*mt, sst).get();
-                    sst = env.reusable_sst(sst).get0();
+                    auto sst = make_sstable_containing(sst_gen, mt);
                     BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                     check_min_max_column_names(sst, {}, {"z"});
                 }
@@ -1335,9 +1289,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                             tomb);
                     m.partition().apply_delete(*s, std::move(rt));
                     mt->apply(std::move(m));
-                    auto sst = env.make_sstable(s, 11, version);
-                    write_memtable_to_sstable_for_test(*mt, sst).get();
-                    sst = env.reusable_sst(sst).get0();
+                    auto sst = make_sstable_containing(sst_gen, mt);
                     BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                     check_min_max_column_names(sst, {"a"}, {});
                 }
@@ -1349,9 +1301,7 @@ SEASTAR_TEST_CASE(sstable_tombstone_metadata_check) {
                     tombstone tomb(api::new_timestamp(), gc_clock::now());
                     m.partition().apply_delete(*s, clustering_key_prefix::make_empty(), tomb);
                     mt->apply(std::move(m));
-                    auto sst = env.make_sstable(s, 12, version);
-                    write_memtable_to_sstable_for_test(*mt, sst).get();
-                    sst = env.reusable_sst(sst).get0();
+                    auto sst = make_sstable_containing(sst_gen, mt);
                     BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                     check_min_max_column_names(sst, {}, {});
                 }
@@ -1369,6 +1319,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                     .with_column("ck2", utf8_type, column_kind::clustering_key)
                     .with_column("r1", int32_type)
                     .build();
+            auto sst_gen = env.make_sst_factory(s, version);
             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");
@@ -1381,9 +1332,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                 tombstone tomb(api::new_timestamp(), gc_clock::now());
                 m.partition().apply_delete(*s, c_key, tomb);
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 1, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1", "c2"}, {"c1", "c2"});
             }
@@ -1393,9 +1342,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                 mutation m(s, key);
                 m.set_clustered_cell(c_key, r1_col, make_dead_atomic_cell(3600));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 2, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1", "c2"}, {"c1", "c2"});
             }
@@ -1405,9 +1352,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                 mutation m(s, key);
                 m.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 3, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(!sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1", "c2"}, {"c1", "c2"});
             }
@@ -1425,9 +1370,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                 m2.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
                 mt->apply(std::move(m2));
 
-                auto sst = env.make_sstable(s, 4, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1", "c2"}, {"c1", "c2"});
             }
@@ -1438,9 +1381,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                 tombstone tomb(api::new_timestamp(), gc_clock::now());
                 m.partition().apply(tomb);
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 5, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {}, {});
             }
@@ -1455,9 +1396,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 6, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"a", "aa"}, {"z", "zz"});
@@ -1475,9 +1414,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 7, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"a"}, {"c1", "c2"});
@@ -1495,9 +1432,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 8, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"c1", "aa"}, {"c1", "zz"});
@@ -1515,9 +1450,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 9, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"c1", "c2"}, {"z", "zz"});
@@ -1536,9 +1469,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                             tomb);
                     m.partition().apply_delete(*s, std::move(rt));
                     mt->apply(std::move(m));
-                    auto sst = env.make_sstable(s, 10, version);
-                    write_memtable_to_sstable_for_test(*mt, sst).get();
-                    sst = env.reusable_sst(sst).get0();
+                    auto sst = make_sstable_containing(sst_gen, mt);
                     BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                     check_min_max_column_names(sst, {}, {"z"});
                 }
@@ -1554,9 +1485,7 @@ SEASTAR_TEST_CASE(sstable_composite_tombstone_metadata_check) {
                             tomb);
                     m.partition().apply_delete(*s, std::move(rt));
                     mt->apply(std::move(m));
-                    auto sst = env.make_sstable(s, 11, version);
-                    write_memtable_to_sstable_for_test(*mt, sst).get();
-                    sst = env.reusable_sst(sst).get0();
+                    auto sst = make_sstable_containing(sst_gen, mt);
                     BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                     check_min_max_column_names(sst, {"a"}, {});
                 }
@@ -1574,6 +1503,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                     .with_column("ck2", reversed_type_impl::get_instance(utf8_type), column_kind::clustering_key)
                     .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")});
@@ -1587,9 +1517,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                 tombstone tomb(api::new_timestamp(), gc_clock::now());
                 m.partition().apply_delete(*s, c_key, tomb);
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 1, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1", "c2"}, {"c1", "c2"});
             }
@@ -1599,9 +1527,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                 mutation m(s, key);
                 m.set_clustered_cell(c_key, r1_col, make_dead_atomic_cell(3600));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 2, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1", "c2"}, {"c1", "c2"});
             }
@@ -1611,9 +1537,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                 mutation m(s, key);
                 m.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 3, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(!sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1", "c2"}, {"c1", "c2"});
             }
@@ -1631,9 +1555,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                 m2.set_clustered_cell(c_key, r1_col, make_atomic_cell(int32_type, int32_type->decompose(1)));
                 mt->apply(std::move(m2));
 
-                auto sst = env.make_sstable(s, 4, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {"c1", "c2"}, {"c1", "c2"});
             }
@@ -1644,9 +1566,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                 tombstone tomb(api::new_timestamp(), gc_clock::now());
                 m.partition().apply(tomb);
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 5, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 check_min_max_column_names(sst, {}, {});
             }
@@ -1661,9 +1581,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 6, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"a", "zz"}, {"a", "aa"});
@@ -1681,9 +1599,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 7, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"a", "zz"}, {"c1", "c2"});
@@ -1701,9 +1617,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 8, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"c1", "zz"}, {"c1"});
@@ -1721,9 +1635,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                         tomb);
                 m.partition().apply_delete(*s, std::move(rt));
                 mt->apply(std::move(m));
-                auto sst = env.make_sstable(s, 9, version);
-                write_memtable_to_sstable_for_test(*mt, sst).get();
-                sst = env.reusable_sst(sst).get0();
+                auto sst = make_sstable_containing(sst_gen, mt);
                 BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                 if (version >= sstable_version_types::mc) {
                     check_min_max_column_names(sst, {"c1", "zz"}, {"c1", "c2"});
@@ -1742,9 +1654,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                             tomb);
                     m.partition().apply_delete(*s, std::move(rt));
                     mt->apply(std::move(m));
-                    auto sst = env.make_sstable(s, 10, version);
-                    write_memtable_to_sstable_for_test(*mt, sst).get();
-                    sst = env.reusable_sst(sst).get0();
+                    auto sst = make_sstable_containing(sst_gen, mt);
                     BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                     check_min_max_column_names(sst, {}, {"z"});
                 }
@@ -1760,9 +1670,7 @@ SEASTAR_TEST_CASE(sstable_composite_reverse_tombstone_metadata_check) {
                             tomb);
                     m.partition().apply_delete(*s, std::move(rt));
                     mt->apply(std::move(m));
-                    auto sst = env.make_sstable(s, 11, version);
-                    write_memtable_to_sstable_for_test(*mt, sst).get();
-                    sst = env.reusable_sst(sst).get0();
+                    auto sst = make_sstable_containing(sst_gen, mt);
                     BOOST_REQUIRE(sst->get_stats_metadata().estimated_tombstone_drop_time.bin.size());
                     check_min_max_column_names(sst, {"a"}, {});
                 }
@@ -2955,9 +2863,7 @@ SEASTAR_TEST_CASE(sstable_reader_with_timeout) {
             m.partition().apply_delete(*s, cp, tomb);
             mt->apply(std::move(m));
 
-            auto sst = env.make_sstable(s, 12);
-            write_memtable_to_sstable_for_test(*mt, sst).get();
-            auto sstp = env.reusable_sst(sst).get0();
+            auto sstp = make_sstable_containing(env.make_sstable(s), mt);
             auto pr = dht::partition_range::make_singular(make_dkey(s, "key1"));
             auto timeout = db::timeout_clock::now();
             auto rd = sstp->make_reader(s, env.make_reader_permit(timeout), pr, s->full_slice());
diff --git a/test/boost/sstable_mutation_test.cc b/test/boost/sstable_mutation_test.cc
index 6d9e28c723..b4e8d520c2 100644
--- a/test/boost/sstable_mutation_test.cc
+++ b/test/boost/sstable_mutation_test.cc
@@ -419,9 +419,7 @@ 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 = env.make_sstable(s);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        sst->load().get();
+        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();
@@ -858,9 +856,7 @@ SEASTAR_TEST_CASE(test_non_compound_table_row_is_not_marked_as_static) {
         auto mt = make_lw_shared<replica::memtable>(s);
         mt->apply(std::move(m));
 
-        auto sst = env.make_sstable(s, version);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        sst->load().get();
+        auto sst = make_sstable_containing(env.make_sstable(s, version), 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();
@@ -888,10 +884,8 @@ SEASTAR_TEST_CASE(test_has_partition_key) {
             auto mt = make_lw_shared<replica::memtable>(s);
             mt->apply(std::move(m));
 
-            auto sst = env.make_sstable(s, version);
-            write_memtable_to_sstable_for_test(*mt, sst).get();
+            auto sst = make_sstable_containing(env.make_sstable(s, version), mt);
             auto hk = sstables::sstable::make_hashed_key(*s, dk.key());
-            sst->load().get();
             auto mr = sst->make_reader(s, env.make_reader_permit(), query::full_partition_range, s->full_slice());
             auto close_mr = deferred_close(mr);
             auto res =  sst->has_partition_key(hk, dk).get0();
diff --git a/test/boost/sstable_resharding_test.cc b/test/boost/sstable_resharding_test.cc
index a743faee16..8c262f5e47 100644
--- a/test/boost/sstable_resharding_test.cc
+++ b/test/boost/sstable_resharding_test.cc
@@ -75,9 +75,7 @@ void run_sstable_resharding_test() {
                 mt->apply(std::move(m));
             }
         }
-        auto sst = cf.make_sstable(version);
-        write_memtable_to_sstable_for_test(*mt, sst).get();
-        return env.reusable_sst(s, sst->generation().value(), version).get();
+        return make_sstable_containing(cf.make_sstable(version), mt);
     });
 
     // FIXME: sstable write has a limitation in which it will generate sharding metadata only
diff --git a/test/lib/sstable_utils.cc b/test/lib/sstable_utils.cc
index fb51ef7dd4..7d6555c896 100644
--- a/test/lib/sstable_utils.cc
+++ b/test/lib/sstable_utils.cc
@@ -17,6 +17,7 @@
 #include "sstables/version.hh"
 #include "test/lib/flat_mutation_reader_assertions.hh"
 #include "test/lib/reader_concurrency_semaphore.hh"
+#include "test/boost/sstable_test.hh"
 #include <seastar/core/reactor.hh>
 #include <seastar/core/seastar.hh>
 #include <seastar/core/coroutine.hh>
@@ -24,6 +25,17 @@
 using namespace sstables;
 using namespace std::chrono_literals;
 
+sstables::shared_sstable make_sstable_containing(std::function<sstables::shared_sstable()> sst_factory, lw_shared_ptr<replica::memtable> mt) {
+    return make_sstable_containing(sst_factory(), std::move(mt));
+}
+
+sstables::shared_sstable make_sstable_containing(sstables::shared_sstable sst, lw_shared_ptr<replica::memtable> mt) {
+    write_memtable_to_sstable_for_test(*mt, sst).get();
+    sstable_open_config cfg { .load_first_and_last_position_metadata = true };
+    sst->open_data(cfg).get();
+    return sst;
+}
+
 sstables::shared_sstable make_sstable_containing(std::function<sstables::shared_sstable()> sst_factory, std::vector<mutation> muts) {
     return make_sstable_containing(sst_factory(), std::move(muts));
 }
diff --git a/test/lib/sstable_utils.hh b/test/lib/sstable_utils.hh
index 4f848ad6b5..e2adac86a6 100644
--- a/test/lib/sstable_utils.hh
+++ b/test/lib/sstable_utils.hh
@@ -25,6 +25,9 @@
 using namespace sstables;
 using namespace std::chrono_literals;
 
+// Must be called in a seastar thread.
+sstables::shared_sstable make_sstable_containing(std::function<sstables::shared_sstable()> sst_factory, lw_shared_ptr<replica::memtable> mt);
+sstables::shared_sstable make_sstable_containing(sstables::shared_sstable sst, lw_shared_ptr<replica::memtable> mt);
 sstables::shared_sstable make_sstable_containing(std::function<sstables::shared_sstable()> sst_factory, std::vector<mutation> muts);
 sstables::shared_sstable make_sstable_containing(sstables::shared_sstable sst, std::vector<mutation> muts);
 
diff --git a/test/lib/test_services.hh b/test/lib/test_services.hh
index b6986463ea..f1b6febcaf 100644
--- a/test/lib/test_services.hh
+++ b/test/lib/test_services.hh
@@ -84,4 +84,16 @@ struct table_for_tests {
         auto& sstables_manager = table.get_sstables_manager();
         return sstables_manager.make_sstable(_data->s, _data->cfg.datadir, table.calculate_generation_for_new_table(), version);
     }
+
+    std::function<sstables::shared_sstable()> make_sst_factory() {
+        return [this] {
+            return make_sstable();
+        };
+    }
+
+    std::function<sstables::shared_sstable()> make_sst_factory(sstables::sstable_version_types version) {
+        return [this, version] {
+            return make_sstable(version);
+        };
+    }
 };