diff --git a/readers/mutation_readers.cc b/readers/mutation_readers.cc
index d93bb334a1..026d8eb382 100644
--- a/readers/mutation_readers.cc
+++ b/readers/mutation_readers.cc
@@ -537,9 +537,9 @@ flat_mutation_reader_v2 make_reversing_reader(flat_mutation_reader_v2 original,
     return make_flat_mutation_reader_v2<partition_reversing_mutation_reader>(std::move(original), max_size, std::move(slice));
 }
 
-flat_mutation_reader make_nonforwardable(flat_mutation_reader r, bool single_partition) {
-    class reader : public flat_mutation_reader::impl {
-        flat_mutation_reader _underlying;
+flat_mutation_reader_v2 make_nonforwardable(flat_mutation_reader_v2 r, bool single_partition) {
+    class reader : public flat_mutation_reader_v2::impl {
+        flat_mutation_reader_v2 _underlying;
         bool _single_partition;
         bool _static_row_done = false;
         bool is_end_end_of_underlying_stream() const {
@@ -563,7 +563,7 @@ flat_mutation_reader make_nonforwardable(flat_mutation_reader r, bool single_par
           });
         }
     public:
-        reader(flat_mutation_reader r, bool single_partition)
+        reader(flat_mutation_reader_v2 r, bool single_partition)
             : impl(r.schema(), r.permit())
             , _underlying(std::move(r))
             , _single_partition(single_partition)
@@ -600,7 +600,7 @@ flat_mutation_reader make_nonforwardable(flat_mutation_reader r, bool single_par
             return _underlying.close();
         }
     };
-    return make_flat_mutation_reader<reader>(std::move(r), single_partition);
+    return make_flat_mutation_reader_v2<reader>(std::move(r), single_partition);
 }
 
 template<typename Generator>
diff --git a/readers/nonforwardable.hh b/readers/nonforwardable.hh
index 39cdba5b13..239a821d8e 100644
--- a/readers/nonforwardable.hh
+++ b/readers/nonforwardable.hh
@@ -8,7 +8,6 @@
 
 #pragma once
 
-class flat_mutation_reader;
-
-flat_mutation_reader make_nonforwardable(flat_mutation_reader, bool);
+class flat_mutation_reader_v2;
 
+flat_mutation_reader_v2 make_nonforwardable(flat_mutation_reader_v2, bool);
diff --git a/row_cache.cc b/row_cache.cc
index cbbfe688d3..7e8636d14f 100644
--- a/row_cache.cc
+++ b/row_cache.cc
@@ -23,6 +23,7 @@
 #include "readers/empty.hh"
 #include "readers/forwardable_v2.hh"
 #include "readers/nonforwardable.hh"
+#include "readers/conversion.hh"
 
 namespace cache {
 
@@ -351,7 +352,7 @@ future<> read_context::create_underlying() {
 static flat_mutation_reader read_directly_from_underlying(read_context& reader) {
     flat_mutation_reader res = make_delegating_reader(reader.underlying().underlying());
     res.upgrade_schema(reader.schema());
-    return make_nonforwardable(std::move(res), true);
+    return downgrade_to_v1(make_nonforwardable(upgrade_to_v2(std::move(res)), true));
 }
 
 // Reader which populates the cache using data from the delegate.