From 44a6cef4e1ed7fd25cce547dbcccdeace2ff787b Mon Sep 17 00:00:00 2001
From: Avi Kivity <avi@scylladb.com>
Date: Thu, 30 Jun 2016 18:00:27 +0300
Subject: [PATCH] sstable mutation readers: use sstable_set::select()

Apply compaction strategy specific logic to narrow down the set of sstables
used for a query; can speed up reads using LeveledCompactionStrategy
significantly.

Fixes #1185.
---
 database.cc | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/database.cc b/database.cc
index 6448f6d04c..cdea59ea60 100644
--- a/database.cc
+++ b/database.cc
@@ -190,7 +190,7 @@ public:
         , _ck_filtering(ck_filtering)
     {
         std::vector<mutation_reader> readers;
-        for (const lw_shared_ptr<sstables::sstable>& sst : *_sstables->all()) {
+        for (const lw_shared_ptr<sstables::sstable>& sst : _sstables->select(pr)) {
             // FIXME: make sstable::read_range_rows() return ::mutation_reader so that we can drop this wrapper.
             mutation_reader reader =
                 make_mutation_reader<sstable_range_wrapping_reader>(sst, s, pr, _ck_filtering, _pc);
@@ -211,6 +211,7 @@ public:
 
 class single_key_sstable_reader final : public mutation_reader::impl {
     schema_ptr _schema;
+    dht::ring_position _rp;
     sstables::key _key;
     std::vector<streamed_mutation> _mutations;
     bool _done = false;
@@ -226,6 +227,7 @@ public:
                               query::clustering_key_filtering_context ck_filtering,
                               const io_priority_class& pc)
         : _schema(std::move(schema))
+        , _rp(dht::global_partitioner().decorate_key(*_schema, key))
         , _key(sstables::key::from_partition_key(*_schema, key))
         , _sstables(std::move(sstables))
         , _pc(pc)
@@ -236,7 +238,7 @@ public:
         if (_done) {
             return make_ready_future<streamed_mutation_opt>();
         }
-        return parallel_for_each(*_sstables->all(),
+        return parallel_for_each(_sstables->select(query::partition_range(_rp)),
             [this](const lw_shared_ptr<sstables::sstable>& sstable) {
                 return sstable->read_row(_schema, _key, _ck_filtering, _pc).then([this](auto smo) {
                     if (smo) {