diff --git a/test/boost/database_test.cc b/test/boost/database_test.cc
index 113297fbde..15a78fb3a4 100644
--- a/test/boost/database_test.cc
+++ b/test/boost/database_test.cc
@@ -146,9 +146,9 @@ SEASTAR_TEST_CASE(test_querying_with_limits) {
     });
 }
 
-SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source) {
-    do_with_cql_env_thread([] (cql_test_env& e) {
-        run_mutation_source_tests([&] (schema_ptr s, const std::vector<mutation>& partitions) -> mutation_source {
+static void test_database(void (*run_tests)(populate_fn_ex, bool)) {
+    do_with_cql_env_thread([run_tests] (cql_test_env& e) {
+        run_tests([&] (schema_ptr s, const std::vector<mutation>& partitions, gc_clock::time_point) -> mutation_source {
             try {
                 e.local_db().find_column_family(s->ks_name(), s->cf_name());
                 e.migration_manager().local().announce_column_family_drop(s->ks_name(), s->cf_name()).get();
@@ -172,10 +172,26 @@ SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_sourc
                     mutation_reader::forwarding fwd_mr) {
                 return cf.make_reader(s, std::move(permit), range, slice, pc, std::move(trace_state), fwd, fwd_mr);
             });
-        });
+        }, true);
     }).get();
 }
 
+SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain) {
+    test_database(run_mutation_source_tests_plain);
+}
+
+SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_downgrade) {
+    test_database(run_mutation_source_tests_downgrade);
+}
+
+SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_upgrade) {
+    test_database(run_mutation_source_tests_upgrade);
+}
+
+SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse) {
+    test_database(run_mutation_source_tests_reverse);
+}
+
 SEASTAR_THREAD_TEST_CASE(test_distributed_loader_with_incomplete_sstables) {
     using sst = sstables::sstable;
 
diff --git a/test/boost/multishard_combining_reader_as_mutation_source_test.cc b/test/boost/multishard_combining_reader_as_mutation_source_test.cc
index b16cb6c9d7..39ecc389a1 100644
--- a/test/boost/multishard_combining_reader_as_mutation_source_test.cc
+++ b/test/boost/multishard_combining_reader_as_mutation_source_test.cc
@@ -45,7 +45,7 @@
 static std::list<dummy_sharder> keep_alive_sharder;
 
 static auto make_populate(bool evict_paused_readers, bool single_fragment_buffer) {
-    return [evict_paused_readers, single_fragment_buffer] (schema_ptr s, const std::vector<mutation>& mutations) mutable {
+    return [evict_paused_readers, single_fragment_buffer] (schema_ptr s, const std::vector<mutation>& mutations, gc_clock::time_point) mutable {
         // We need to group mutations that have the same token so they land on the same shard.
         std::map<dht::token, std::vector<frozen_mutation>> mutations_by_token;
 
@@ -135,6 +135,9 @@ SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_evict_paused) {
     }).get();
 }
 
+// Single fragment buffer tests are extremely slow, so the
+// run_mutation_source_tests execution is split
+
 SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer) {
     if (smp::count < 2) {
         std::cerr << "Cannot run test " << get_name() << " with smp::count < 2" << std::endl;
@@ -142,7 +145,43 @@ SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer) {
     }
 
     do_with_cql_env_thread([&] (cql_test_env& env) -> future<> {
-        run_mutation_source_tests(make_populate(true, true));
+        run_mutation_source_tests_plain(make_populate(true, true), true);
+        return make_ready_future<>();
+    }).get();
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer_dowgrade) {
+    if (smp::count < 2) {
+        std::cerr << "Cannot run test " << get_name() << " with smp::count < 2" << std::endl;
+        return;
+    }
+
+    do_with_cql_env_thread([&] (cql_test_env& env) -> future<> {
+        run_mutation_source_tests_downgrade(make_populate(true, true), true);
+        return make_ready_future<>();
+    }).get();
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer_upgrade) {
+    if (smp::count < 2) {
+        std::cerr << "Cannot run test " << get_name() << " with smp::count < 2" << std::endl;
+        return;
+    }
+
+    do_with_cql_env_thread([&] (cql_test_env& env) -> future<> {
+        run_mutation_source_tests_upgrade(make_populate(true, true), true);
+        return make_ready_future<>();
+    }).get();
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer_reverse) {
+    if (smp::count < 2) {
+        std::cerr << "Cannot run test " << get_name() << " with smp::count < 2" << std::endl;
+        return;
+    }
+
+    do_with_cql_env_thread([&] (cql_test_env& env) -> future<> {
+        run_mutation_source_tests_reverse(make_populate(true, true), true);
         return make_ready_future<>();
     }).get();
 }
diff --git a/test/lib/mutation_source_test.cc b/test/lib/mutation_source_test.cc
index 9d44b0f292..e6093ea649 100644
--- a/test/lib/mutation_source_test.cc
+++ b/test/lib/mutation_source_test.cc
@@ -1742,8 +1742,19 @@ void run_mutation_source_tests(populate_fn populate, bool with_partition_range_f
 }
 
 void run_mutation_source_tests(populate_fn_ex populate, bool with_partition_range_forwarding) {
-    run_mutation_reader_tests(populate, with_partition_range_forwarding);
+    run_mutation_source_tests_plain(populate, with_partition_range_forwarding);
+    run_mutation_source_tests_downgrade(populate, with_partition_range_forwarding);
+    run_mutation_source_tests_upgrade(populate, with_partition_range_forwarding);
+    run_mutation_source_tests_reverse(populate, with_partition_range_forwarding);
+    // Some tests call the sub-types individually, mind checking them
+    // if adding new stuff here
+}
 
+void run_mutation_source_tests_plain(populate_fn_ex populate, bool with_partition_range_forwarding) {
+    run_mutation_reader_tests(populate, with_partition_range_forwarding);
+}
+
+void run_mutation_source_tests_downgrade(populate_fn_ex populate, bool with_partition_range_forwarding) {
     // ? -> v2 -> v1 -> *
     run_mutation_reader_tests([populate] (schema_ptr s, const std::vector<mutation>& m, gc_clock::time_point t) -> mutation_source {
         return mutation_source([ms = populate(s, m, t)] (schema_ptr s,
@@ -1758,7 +1769,9 @@ void run_mutation_source_tests(populate_fn_ex populate, bool with_partition_rang
                     ms.make_reader_v2(s, std::move(permit), pr, slice, pc, std::move(tr), fwd, mr_fwd));
         });
     }, with_partition_range_forwarding);
+}
 
+void run_mutation_source_tests_upgrade(populate_fn_ex populate, bool with_partition_range_forwarding) {
     // ? -> v1 -> v2 -> *
     run_mutation_reader_tests([populate] (schema_ptr s, const std::vector<mutation>& m, gc_clock::time_point t) -> mutation_source {
         return mutation_source([ms = populate(s, m, t)] (schema_ptr s,
@@ -1773,7 +1786,9 @@ void run_mutation_source_tests(populate_fn_ex populate, bool with_partition_rang
                     ms.make_reader(s, std::move(permit), pr, slice, pc, std::move(tr), fwd, mr_fwd));
         });
     }, with_partition_range_forwarding);
+}
 
+void run_mutation_source_tests_reverse(populate_fn_ex populate, bool with_partition_range_forwarding) {
     // read in reverse
     run_mutation_reader_tests([populate] (schema_ptr s, const std::vector<mutation>& m, gc_clock::time_point t) -> mutation_source {
         auto table_schema = s->make_reversed();
diff --git a/test/lib/mutation_source_test.hh b/test/lib/mutation_source_test.hh
index 650a6633be..da527fc22a 100644
--- a/test/lib/mutation_source_test.hh
+++ b/test/lib/mutation_source_test.hh
@@ -30,6 +30,10 @@ using populate_fn_ex = std::function<mutation_source(schema_ptr s, const std::ve
 // Must be run in a seastar thread
 void run_mutation_source_tests(populate_fn populate, bool with_partition_range_forwarding = true);
 void run_mutation_source_tests(populate_fn_ex populate, bool with_partition_range_forwarding = true);
+void run_mutation_source_tests_plain(populate_fn_ex populate, bool with_partition_range_forwarding = true);
+void run_mutation_source_tests_downgrade(populate_fn_ex populate, bool with_partition_range_forwarding = true);
+void run_mutation_source_tests_upgrade(populate_fn_ex populate, bool with_partition_range_forwarding = true);
+void run_mutation_source_tests_reverse(populate_fn_ex populate, bool with_partition_range_forwarding = true);
 
 enum are_equal { no, yes };