diff --git a/db/view/view.cc b/db/view/view.cc
index adfbcc8498..1709618f38 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -487,7 +487,7 @@ mutation_partition& view_updates::partition_for(partition_key&& key) {
     if (it != _updates.end()) {
         return it->second;
     }
-    return _updates.emplace(std::move(key), mutation_partition(_view)).first->second;
+    return _updates.emplace(std::move(key), mutation_partition(*_view)).first->second;
 }
 
 size_t view_updates::op_count() const {
diff --git a/mutation/mutation.cc b/mutation/mutation.cc
index 242eee1960..5145e5d121 100644
--- a/mutation/mutation.cc
+++ b/mutation/mutation.cc
@@ -13,13 +13,13 @@
 mutation::data::data(dht::decorated_key&& key, schema_ptr&& schema)
     : _schema(std::move(schema))
     , _dk(std::move(key))
-    , _p(_schema)
+    , _p(*_schema)
 { }
 
 mutation::data::data(partition_key&& key_, schema_ptr&& schema)
     : _schema(std::move(schema))
     , _dk(dht::decorate_key(*_schema, std::move(key_)))
-    , _p(_schema)
+    , _p(*_schema)
 { }
 
 mutation::data::data(schema_ptr&& schema, dht::decorated_key&& key, const mutation_partition& mp)
diff --git a/mutation/mutation_partition.cc b/mutation/mutation_partition.cc
index 1bff13b980..fd7f6f3173 100644
--- a/mutation/mutation_partition.cc
+++ b/mutation/mutation_partition.cc
@@ -1919,7 +1919,7 @@ bool row_marker::compact_and_expire(tombstone tomb, gc_clock::time_point now,
 mutation_partition mutation_partition::difference(schema_ptr s, const mutation_partition& other) const
 {
     check_schema(*s);
-    mutation_partition mp(s);
+    mutation_partition mp(*s);
     if (_tombstone > other._tombstone) {
         mp.apply(_tombstone);
     }
@@ -1979,7 +1979,7 @@ void mutation_partition::accept(const schema& s, mutation_partition_visitor& v)
 void
 mutation_partition::upgrade(const schema& old_schema, const schema& new_schema) {
     // We need to copy to provide strong exception guarantees.
-    mutation_partition tmp(new_schema.shared_from_this());
+    mutation_partition tmp(new_schema);
     tmp.set_static_row_continuous(_static_row_continuous);
     converting_mutation_partition_applier v(old_schema.get_column_mapping(), new_schema, tmp);
     accept(old_schema, v);
diff --git a/mutation/mutation_partition.hh b/mutation/mutation_partition.hh
index 3b3e9f8092..0811e73880 100644
--- a/mutation/mutation_partition.hh
+++ b/mutation/mutation_partition.hh
@@ -1203,11 +1203,11 @@ public:
     static mutation_partition make_incomplete(const schema& s, tombstone t = {}) {
         return mutation_partition(incomplete_tag(), s, t);
     }
-    mutation_partition(schema_ptr s)
+    mutation_partition(const schema& s)
         : _rows()
-        , _row_tombstones(*s)
+        , _row_tombstones(s)
 #ifdef SEASTAR_DEBUG
-        , _schema_version(s->version())
+        , _schema_version(s.version())
 #endif
     { }
     mutation_partition(mutation_partition& other, copy_comparators_only)
diff --git a/mutation/mutation_partition_v2.cc b/mutation/mutation_partition_v2.cc
index efac0fa8a7..ecd1292e5d 100644
--- a/mutation/mutation_partition_v2.cc
+++ b/mutation/mutation_partition_v2.cc
@@ -510,7 +510,7 @@ mutation_partition_v2::apply_row_tombstone(const schema& schema, clustering_key_
 void
 mutation_partition_v2::apply_row_tombstone(const schema& schema, range_tombstone rt) {
     check_schema(schema);
-    mutation_partition mp(schema.shared_from_this());
+    mutation_partition mp(schema);
     mp.apply_row_tombstone(schema, std::move(rt));
     apply(schema, std::move(mp));
 }
@@ -919,7 +919,7 @@ void mutation_partition_v2::accept(const schema& s, mutation_partition_visitor&
 void
 mutation_partition_v2::upgrade(const schema& old_schema, const schema& new_schema) {
     // We need to copy to provide strong exception guarantees.
-    mutation_partition tmp(new_schema.shared_from_this());
+    mutation_partition tmp(new_schema);
     tmp.set_static_row_continuous(_static_row_continuous);
     converting_mutation_partition_applier v(old_schema.get_column_mapping(), new_schema, tmp);
     accept(old_schema, v);
@@ -927,7 +927,7 @@ mutation_partition_v2::upgrade(const schema& old_schema, const schema& new_schem
 }
 
 mutation_partition mutation_partition_v2::as_mutation_partition(const schema& s) const {
-    mutation_partition tmp(s.shared_from_this());
+    mutation_partition tmp(s);
     tmp.set_static_row_continuous(_static_row_continuous);
     partition_builder v(s, tmp);
     accept(s, v);
diff --git a/replica/memtable.cc b/replica/memtable.cc
index 0f0c51628c..ebcdfcd7f3 100644
--- a/replica/memtable.cc
+++ b/replica/memtable.cc
@@ -225,7 +225,7 @@ memtable::find_or_create_partition(const dht::decorated_key& key) {
     if (i == partitions.end() || !hint.match) {
         partitions_type::iterator entry = partitions.emplace_before(i,
                 key.token().raw(), hint,
-                _schema, dht::decorated_key(key), mutation_partition(_schema));
+                _schema, dht::decorated_key(key), mutation_partition(*_schema));
         ++nr_partitions;
         ++_table_stats.memtable_partition_insertions;
         if (!hint.emplace_keeps_iterators()) {
@@ -793,7 +793,7 @@ memtable::apply(const frozen_mutation& m, const schema_ptr& m_schema, db::rp_han
     with_allocator(allocator(), [this, &m, &m_schema] {
         _allocating_section(*this, [&, this] {
             auto& p = find_or_create_partition_slow(m.key());
-            mutation_partition mp(m_schema);
+            mutation_partition mp(*m_schema);
             partition_builder pb(*m_schema, mp);
             m.partition().accept(*m_schema, pb);
             _stats_collector.update(*m_schema, mp);
diff --git a/row_cache.cc b/row_cache.cc
index bd9675138a..35f0e30d60 100644
--- a/row_cache.cc
+++ b/row_cache.cc
@@ -846,7 +846,7 @@ cache_entry& row_cache::find_or_create_incomplete(const partition_start& ps, row
 
 cache_entry& row_cache::find_or_create_missing(const dht::decorated_key& key) {
     return do_find_or_create_entry(key, nullptr, [&] (auto i, const partitions_type::bound_hint& hint) {
-        mutation_partition mp(_schema);
+        mutation_partition mp(*_schema);
         bool cont = i->continuous();
         partitions_type::iterator entry = _partitions.emplace_before(i, key.token().raw(), hint,
                 _schema, key, std::move(mp));
@@ -1035,7 +1035,7 @@ future<> row_cache::update(external_updater eu, replica::memtable& m) {
             // Partition is absent in underlying. First, insert a neutral partition entry.
             partitions_type::iterator entry = _partitions.emplace_before(cache_i, mem_e.key().token().raw(), hint,
                 cache_entry::evictable_tag(), _schema, dht::decorated_key(mem_e.key()),
-                partition_entry::make_evictable(*_schema, mutation_partition(_schema)));
+                partition_entry::make_evictable(*_schema, mutation_partition(*_schema)));
             entry->set_continuous(cache_i->continuous());
             _tracker.insert(*entry);
             mem_e.upgrade_schema(_schema, _tracker.memtable_cleaner());
diff --git a/test/boost/mutation_test.cc b/test/boost/mutation_test.cc
index 0a91946e48..b626b121be 100644
--- a/test/boost/mutation_test.cc
+++ b/test/boost/mutation_test.cc
@@ -1921,9 +1921,9 @@ SEASTAR_TEST_CASE(test_mutation_diff_with_random_generator) {
             auto m12_with_diff = m1;
             m12_with_diff.partition().apply(*s, m2.partition().difference(s, m1.partition()), app_stats);
             check_partitions_match(m12.partition(), m12_with_diff.partition(), *s);
-            check_partitions_match(mutation_partition{s}, m1.partition().difference(s, m1.partition()), *s);
-            check_partitions_match(m1.partition(), m1.partition().difference(s, mutation_partition{s}), *s);
-            check_partitions_match(mutation_partition{s}, mutation_partition{s}.difference(s, m1.partition()), *s);
+            check_partitions_match(mutation_partition{*s}, m1.partition().difference(s, m1.partition()), *s);
+            check_partitions_match(m1.partition(), m1.partition().difference(s, mutation_partition{*s}), *s);
+            check_partitions_match(mutation_partition{*s}, mutation_partition{*s}.difference(s, m1.partition()), *s);
         });
     });
 }
diff --git a/test/boost/mvcc_test.cc b/test/boost/mvcc_test.cc
index b414fc36f0..7a8cd97509 100644
--- a/test/boost/mvcc_test.cc
+++ b/test/boost/mvcc_test.cc
@@ -41,7 +41,7 @@ static thread_local mutation_application_stats app_stats_for_tests;
 // The cursor must be pointing at a row and valid.
 // The cursor will not be pointing at a row after this.
 static mutation_partition read_partition_from(const schema& schema, partition_snapshot_row_cursor& cur) {
-    mutation_partition p(schema.shared_from_this());
+    mutation_partition p(schema);
     position_in_partition prev = position_in_partition::before_all_clustered_rows();
     do {
         testlog.trace("cur: {}", cur);
@@ -304,7 +304,7 @@ SEASTAR_TEST_CASE(test_apply_to_incomplete) {
             assert_that(table.schema(), e.squashed()).is_equal_to((m2 + m3).partition());
 
             // Check that snapshot data is not stolen when its entry is applied
-            auto e2 = ms.make_evictable(mutation_partition(table.schema()));
+            auto e2 = ms.make_evictable(mutation_partition(s));
             e2 += std::move(e);
             assert_that(table.schema(), ms.squashed(snap1)).is_equal_to(m1.partition());
             assert_that(table.schema(), e2.squashed()).is_equal_to((m2 + m3).partition());
@@ -381,7 +381,7 @@ SEASTAR_TEST_CASE(test_eviction_with_active_reader) {
             auto ck1 = table.make_ckey(1);
             auto ck2 = table.make_ckey(2);
 
-            auto e = ms.make_evictable(mutation_partition(table.schema()));
+            auto e = ms.make_evictable(mutation_partition(s));
 
             mutation m1(table.schema(), pk);
             m1.partition().clustered_row(s, ck2);
@@ -670,7 +670,7 @@ SEASTAR_TEST_CASE(test_partition_snapshot_row_cursor) {
             simple_schema table;
             auto&& s = *table.schema();
 
-            auto e = partition_entry::make_evictable(s, mutation_partition(table.schema()));
+            auto e = partition_entry::make_evictable(s, mutation_partition(s));
             auto snap1 = e.read(r, tracker.cleaner(), table.schema(), &tracker);
 
             {
@@ -841,7 +841,7 @@ SEASTAR_TEST_CASE(test_partition_snapshot_row_cursor_reversed) {
             simple_schema table;
             auto&& s = *table.schema();
 
-            auto e = partition_entry::make_evictable(s, mutation_partition(table.schema()));
+            auto e = partition_entry::make_evictable(s, mutation_partition(s));
             auto snap1 = e.read(r, tracker.cleaner(), table.schema(), &tracker);
 
             int ck_0 = 10;
@@ -1032,7 +1032,7 @@ SEASTAR_TEST_CASE(test_cursor_tracks_continuity_in_reversed_mode) {
             simple_schema table;
             auto&& s = *table.schema();
 
-            auto e = partition_entry::make_evictable(s, mutation_partition(table.schema()));
+            auto e = partition_entry::make_evictable(s, mutation_partition(s));
             tracker.insert(e);
             auto snap1 = e.read(r, tracker.cleaner(), table.schema(), &tracker);
 
@@ -1537,7 +1537,7 @@ SEASTAR_TEST_CASE(test_ensure_entry_in_latest_in_reversed_mode) {
             simple_schema table;
             auto&& s = *table.schema();
 
-            auto e = partition_entry::make_evictable(s, mutation_partition(table.schema()));
+            auto e = partition_entry::make_evictable(s, mutation_partition(s));
             auto snap1 = e.read(r, tracker.cleaner(), table.schema(), &tracker);
 
             {
@@ -1593,7 +1593,7 @@ SEASTAR_TEST_CASE(test_ensure_entry_in_latest_does_not_set_continuity_in_reverse
             simple_schema table;
             auto&& s = *table.schema();
 
-            auto e = partition_entry::make_evictable(s, mutation_partition(table.schema()));
+            auto e = partition_entry::make_evictable(s, mutation_partition(s));
             auto snap1 = e.read(r, tracker.cleaner(), table.schema(), &tracker);
 
             {