sstable_set_impl: Return also schema from make_incremental_selector

Define sstable_set_impl::selector_and_schema_t type as a tuple that
contains both a newly created selector and a schema that the selector
is using.

This will allow removal of _schema field from sstable_set class as
the only place it was used was make_incremental_selector.

Signed-off-by: Piotr Jastrzebski <haaawk@gmail.com>

sstables/sstable_set.cc

@@ -218,7 +218,8 @@ sstable_set::incremental_selector::select(const dht::ring_position_view& pos) co
 
 sstable_set::incremental_selector
 sstable_set::make_incremental_selector() const {
-    return incremental_selector(_impl->make_incremental_selector(), *_schema);
+    auto selector = _impl->make_incremental_selector();
+    return incremental_selector(std::get<0>(std::move(selector)), std::get<1>(selector));
 }
 
 partitioned_sstable_set::interval_type partitioned_sstable_set::make_interval(const schema& s, const dht::partition_range& range) {
@@ -569,7 +570,7 @@ bool time_series_sstable_set::erase(shared_sstable sst) {
     return found;
 }
 
-std::unique_ptr<incremental_selector_impl> time_series_sstable_set::make_incremental_selector() const {
+sstable_set_impl::selector_and_schema_t time_series_sstable_set::make_incremental_selector() const {
     struct selector : public incremental_selector_impl {
         const time_series_sstable_set& _set;
 
@@ -581,7 +582,7 @@ std::unique_ptr<incremental_selector_impl> time_series_sstable_set::make_increme
         }
     };
 
-    return std::make_unique<selector>(*this);
+    return std::make_tuple(std::make_unique<selector>(*this), std::cref(*_schema));
 }
 
 // Queue of readers of sstables in a time_series_sstable_set,
@@ -737,8 +738,8 @@ std::unique_ptr<position_reader_queue> time_series_sstable_set::make_position_re
             std::move(pk), std::move(permit), fwd_sm, reversed);
 }
 
-std::unique_ptr<incremental_selector_impl> partitioned_sstable_set::make_incremental_selector() const {
-    return std::make_unique<incremental_selector>(_schema, _unleveled_sstables, _leveled_sstables, _leveled_sstables_change_cnt);
+sstable_set_impl::selector_and_schema_t partitioned_sstable_set::make_incremental_selector() const {
+    return std::make_tuple(std::make_unique<incremental_selector>(_schema, _unleveled_sstables, _leveled_sstables, _leveled_sstables_change_cnt), std::cref(*_schema));
 }
 
 std::unique_ptr<sstable_set_impl> compaction_strategy_impl::make_sstable_set(schema_ptr schema) const {
@@ -1192,7 +1193,7 @@ public:
     }
 };
 
-std::unique_ptr<incremental_selector_impl> compound_sstable_set::make_incremental_selector() const {
+sstable_set_impl::selector_and_schema_t compound_sstable_set::make_incremental_selector() const {
     if (_sets.empty()) {
         // compound_sstable_set must manage one sstable set at least.
         abort();
@@ -1207,7 +1208,7 @@ std::unique_ptr<incremental_selector_impl> compound_sstable_set::make_incrementa
         const auto& set = sets.front();
         return set->_impl->make_incremental_selector();
     }
-    return std::make_unique<incremental_selector>(*_schema, _sets);
+    return std::make_tuple(std::make_unique<incremental_selector>(*_schema, _sets), std::cref(*_schema));
 }
 
 sstable_set make_compound_sstable_set(schema_ptr schema, std::vector<lw_shared_ptr<sstable_set>> sets) {

sstables/sstable_set.hh

@@ -17,6 +17,7 @@
 #include <seastar/core/shared_ptr.hh>
 #include <type_traits>
 #include <vector>
+#include <tuple>
 
 namespace utils {
 class estimated_histogram;
@@ -97,7 +98,8 @@ public:
     uint64_t sub_bytes_on_disk(uint64_t delta) noexcept {
         return _bytes_on_disk -= delta;
     }
-    virtual std::unique_ptr<incremental_selector_impl> make_incremental_selector() const = 0;
+    using selector_and_schema_t = std::tuple<std::unique_ptr<incremental_selector_impl>, const schema&>;
+    virtual selector_and_schema_t make_incremental_selector() const = 0;
 
     virtual flat_mutation_reader_v2 create_single_key_sstable_reader(
         replica::column_family*,

sstables/sstable_set_impl.hh

@@ -69,7 +69,7 @@ public:
     virtual bool insert(shared_sstable sst) override;
     virtual bool erase(shared_sstable sst) override;
     virtual size_t size() const noexcept override;
-    virtual std::unique_ptr<incremental_selector_impl> make_incremental_selector() const override;
+    virtual sstable_set_impl::selector_and_schema_t make_incremental_selector() const override;
     class incremental_selector;
 };
 
@@ -97,7 +97,7 @@ public:
     virtual bool insert(shared_sstable sst) override;
     virtual bool erase(shared_sstable sst) override;
     virtual size_t size() const noexcept override;
-    virtual std::unique_ptr<incremental_selector_impl> make_incremental_selector() const override;
+    virtual sstable_set_impl::selector_and_schema_t make_incremental_selector() const override;
 
     std::unique_ptr<position_reader_queue> make_position_reader_queue(
         std::function<flat_mutation_reader_v2(sstable&)> create_reader,
@@ -139,7 +139,7 @@ public:
     virtual bool erase(shared_sstable sst) override;
     virtual size_t size() const noexcept override;
     virtual uint64_t bytes_on_disk() const noexcept override;
-    virtual std::unique_ptr<incremental_selector_impl> make_incremental_selector() const override;
+    virtual sstable_set_impl::selector_and_schema_t make_incremental_selector() const override;
 
     virtual flat_mutation_reader_v2 create_single_key_sstable_reader(
             replica::column_family*,