sstables: pass partition_range to create_single_key_sstable_reader()

We want to unify the various sstable reader creation methods and this
method taking a ring position instead of a partition range like
everybody else stands in the way of that.

This is effect reverts 68663d0de.

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

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.

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,

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)] (