diff --git a/sstables/sstable_set.cc b/sstables/sstable_set.cc
index 8eefb0d2c1..d55f39a421 100644
--- a/sstables/sstable_set.cc
+++ b/sstables/sstable_set.cc
@@ -696,14 +696,15 @@ sstable_set_impl::create_single_key_sstable_reader(
         schema_ptr schema,
         reader_permit permit,
         utils::estimated_histogram& sstable_histogram,
-        const dht::ring_position& pos,
+        const dht::partition_range& pr,
         const query::partition_slice& slice,
         const io_priority_class& pc,
         tracing::trace_state_ptr trace_state,
         streamed_mutation::forwarding fwd,
         mutation_reader::forwarding fwd_mr) const
 {
-    auto selected_sstables = filter_sstable_for_reader_by_pk(select({pos}), *schema, pos);
+    const auto& pos = pr.start()->value();
+    auto selected_sstables = filter_sstable_for_reader_by_pk(select(pr), *schema, pos);
     auto num_sstables = selected_sstables.size();
     if (!num_sstables) {
         return make_empty_flat_reader(schema, permit);
@@ -738,12 +739,13 @@ time_series_sstable_set::create_single_key_sstable_reader(
         schema_ptr schema,
         reader_permit permit,
         utils::estimated_histogram& sstable_histogram,
-        const dht::ring_position& pos,
+        const dht::partition_range& pr,
         const query::partition_slice& slice,
         const io_priority_class& pc,
         tracing::trace_state_ptr trace_state,
         streamed_mutation::forwarding fwd_sm,
         mutation_reader::forwarding fwd_mr) const {
+    const auto& pos = pr.start()->value();
     // First check if the optimized algorithm for TWCS single partition queries can be applied.
     // Multiple conditions must be satisfied:
     // 1. The sstables must be sufficiently modern so they contain the min/max column metadata.
@@ -762,7 +764,7 @@ time_series_sstable_set::create_single_key_sstable_reader(
         // Some of the conditions were not satisfied so we use the standard query path.
         return sstable_set_impl::create_single_key_sstable_reader(
                 cf, std::move(schema), std::move(permit), sstable_histogram,
-                pos, slice, pc, std::move(trace_state), fwd_sm, fwd_mr);
+                pr, slice, pc, std::move(trace_state), fwd_sm, fwd_mr);
     }
 
     auto pk_filter = make_pk_filter(pos, *schema);
@@ -821,14 +823,15 @@ sstable_set::create_single_key_sstable_reader(
         schema_ptr schema,
         reader_permit permit,
         utils::estimated_histogram& sstable_histogram,
-        const dht::ring_position& pos,
+        const dht::partition_range& pr,
         const query::partition_slice& slice,
         const io_priority_class& pc,
         tracing::trace_state_ptr trace_state,
         streamed_mutation::forwarding fwd,
         mutation_reader::forwarding fwd_mr) const {
+    assert(pr.is_singular() && pr.start()->value().has_key());
     return _impl->create_single_key_sstable_reader(cf, std::move(schema),
-            std::move(permit), sstable_histogram, pos, slice, pc, std::move(trace_state), fwd, fwd_mr);
+            std::move(permit), sstable_histogram, pr, slice, pc, std::move(trace_state), fwd, fwd_mr);
 }
 
 flat_mutation_reader
diff --git a/sstables/sstable_set.hh b/sstables/sstable_set.hh
index f0f5a86c73..8fbcf4c13a 100644
--- a/sstables/sstable_set.hh
+++ b/sstables/sstable_set.hh
@@ -111,7 +111,7 @@ public:
         schema_ptr,
         reader_permit,
         utils::estimated_histogram&,
-        const dht::ring_position&, // must contain a key
+        const dht::partition_range&, // must be singular and contain a key
         const query::partition_slice&,
         const io_priority_class&,
         tracing::trace_state_ptr,
@@ -144,6 +144,17 @@ public:
         streamed_mutation::forwarding,
         mutation_reader::forwarding,
         read_monitor_generator& rmg = default_read_monitor_generator()) const;
+
+    flat_mutation_reader make_reader(
+            schema_ptr,
+            reader_permit,
+            const dht::partition_range&,
+            const query::partition_slice&,
+            const io_priority_class&,
+            tracing::trace_state_ptr,
+            streamed_mutation::forwarding,
+            mutation_reader::forwarding,
+            read_monitor_generator& rmg = default_read_monitor_generator()) const;
 };
 
 /// Read a range from the passed-in sstables.
diff --git a/sstables/sstable_set_impl.hh b/sstables/sstable_set_impl.hh
index 40c2519ed3..28e3cf77fd 100644
--- a/sstables/sstable_set_impl.hh
+++ b/sstables/sstable_set_impl.hh
@@ -48,7 +48,7 @@ public:
         schema_ptr,
         reader_permit,
         utils::estimated_histogram&,
-        const dht::ring_position&,
+        const dht::partition_range&,
         const query::partition_slice&,
         const io_priority_class&,
         tracing::trace_state_ptr,
@@ -135,7 +135,7 @@ public:
         schema_ptr,
         reader_permit,
         utils::estimated_histogram&,
-        const dht::ring_position&,
+        const dht::partition_range&,
         const query::partition_slice&,
         const io_priority_class&,
         tracing::trace_state_ptr,
diff --git a/table.cc b/table.cc
index d6651c9186..51c86109bf 100644
--- a/table.cc
+++ b/table.cc
@@ -91,9 +91,8 @@ table::make_sstable_reader(schema_ptr s,
                     tracing::trace_state_ptr trace_state,
                     streamed_mutation::forwarding fwd,
                     mutation_reader::forwarding fwd_mr) {
-                assert(pr.is_singular() && pr.start()->value().has_key());
                 return sstables->create_single_key_sstable_reader(const_cast<column_family*>(this), std::move(s), std::move(permit),
-                        _stats.estimated_sstable_per_read, pr.start()->value(), slice, pc, std::move(trace_state), fwd, fwd_mr);
+                        _stats.estimated_sstable_per_read, pr, slice, pc, std::move(trace_state), fwd, fwd_mr);
             });
         } else {
             return mutation_source([sstables=std::move(sstables)] (