sstables: make sstable_set constructor less error-prone

Adding an non-empty set of sstables as the set of all sstables in
an sstable_set could cause inconsistencies with the values returned
by select_sstable_runs because the _all_runs map would still be
initialized empty. For similar reasons, the provided sstable_set_impl
should also be empty.

Dispel doubts by removing the unordered_set from the constructor, and
adding a check of emptiness of the sstable_set_impl.

Signed-off-by: Wojciech Mitros <wojciech.mitros@scylladb.com>

db/view/view_update_generator.cc

@@ -67,7 +67,7 @@ future<> view_update_generator::start() {
                     // Exploit the fact that sstables in the staging directory
                     // are usually non-overlapping and use a partitioned set for
                     // the read.
-                    auto ssts = make_lw_shared<sstables::sstable_set>(sstables::make_partitioned_sstable_set(s, make_lw_shared<sstable_list>(sstable_list{}), false));
+                    auto ssts = make_lw_shared<sstables::sstable_set>(sstables::make_partitioned_sstable_set(s, false));
                     for (auto& sst : sstables) {
                         ssts->insert(sst);
                     }

service/storage_service.cc

@@ -2872,8 +2872,7 @@ future<> storage_service::load_and_stream(sstring ks_name, sstring cf_name,
     size_t nr_sst_current = 0;
     while (!sstables.empty()) {
         auto ops_uuid = utils::make_random_uuid();
-        auto sst_set = make_lw_shared<sstables::sstable_set>(sstables::make_partitioned_sstable_set(s,
-                make_lw_shared<sstable_list>(sstable_list{}), false));
+        auto sst_set = make_lw_shared<sstables::sstable_set>(sstables::make_partitioned_sstable_set(s, false));
         size_t batch_sst_nr = 16;
         std::vector<sstring> sst_names;
         std::vector<sstables::shared_sstable> sst_processed;

sstables/sstable_set.cc

@@ -63,10 +63,13 @@ std::ostream& operator<<(std::ostream& os, const sstables::sstable_run& run) {
     return os;
 }
 
-sstable_set::sstable_set(std::unique_ptr<sstable_set_impl> impl, schema_ptr s, lw_shared_ptr<sstable_list> all)
+sstable_set::sstable_set(std::unique_ptr<sstable_set_impl> impl, schema_ptr s)
         : _impl(std::move(impl))
         , _schema(std::move(s))
-        , _all(std::move(all)) {
+        , _all(make_lw_shared<sstable_list>(sstable_list())) {
+    if (!_impl->empty()) {
+        throw std::logic_error("Can't create an sstable_set using a non-empty sstable_set_impl");
+    }
 }
 
 sstable_set::sstable_set(const sstable_set& x)
@@ -171,6 +174,10 @@ void bag_sstable_set::erase(shared_sstable sst) {
     }
 }
 
+bool bag_sstable_set::empty() const {
+    return _sstables.empty();
+}
+
 class bag_sstable_set::incremental_selector : public incremental_selector_impl {
     const std::vector<shared_sstable>& _sstables;
 public:
@@ -299,6 +306,10 @@ void partitioned_sstable_set::erase(shared_sstable sst) {
     }
 }
 
+bool partitioned_sstable_set::empty() const {
+    return _unleveled_sstables.empty() && _leveled_sstables.empty();
+}
+
 class partitioned_sstable_set::incremental_selector : public incremental_selector_impl {
     schema_ptr _schema;
     const std::vector<shared_sstable>& _unleveled_sstables;
@@ -396,6 +407,10 @@ void time_series_sstable_set::erase(shared_sstable sst) {
     }
 }
 
+bool time_series_sstable_set::empty() const {
+    return _sstables->empty();
+}
+
 std::unique_ptr<incremental_selector_impl> time_series_sstable_set::make_incremental_selector() const {
     struct selector : public incremental_selector_impl {
         const time_series_sstable_set& _set;
@@ -535,16 +550,15 @@ std::unique_ptr<sstable_set_impl> time_window_compaction_strategy::make_sstable_
     return std::make_unique<time_series_sstable_set>(std::move(schema));
 }
 
-sstable_set make_partitioned_sstable_set(schema_ptr schema, lw_shared_ptr<sstable_list> all, bool use_level_metadata) {
-    return sstable_set(std::make_unique<partitioned_sstable_set>(schema, use_level_metadata), schema, std::move(all));
+sstable_set make_partitioned_sstable_set(schema_ptr schema, bool use_level_metadata) {
+    return sstable_set(std::make_unique<partitioned_sstable_set>(schema, use_level_metadata), schema);
 }
 
 sstable_set
 compaction_strategy::make_sstable_set(schema_ptr schema) const {
     return sstable_set(
             _compaction_strategy_impl->make_sstable_set(schema),
-            schema,
-            make_lw_shared<sstable_list>());
+            schema);
 }
 
 using sstable_reader_factory_type = std::function<flat_mutation_reader(shared_sstable&, const dht::partition_range& pr)>;

sstables/sstable_set.hh

@@ -58,7 +58,7 @@ class sstable_set : public enable_lw_shared_from_this<sstable_set> {
     std::unordered_map<utils::UUID, sstable_run> _all_runs;
 public:
     ~sstable_set();
-    sstable_set(std::unique_ptr<sstable_set_impl> impl, schema_ptr s, lw_shared_ptr<sstable_list> all);
+    sstable_set(std::unique_ptr<sstable_set_impl> impl, schema_ptr s);
     sstable_set(const sstable_set&);
     sstable_set(sstable_set&&) noexcept;
     sstable_set& operator=(const sstable_set&);
@@ -173,7 +173,7 @@ flat_mutation_reader make_restricted_range_sstable_reader(
     mutation_reader::forwarding,
     read_monitor_generator& rmg = default_read_monitor_generator());
 
-sstable_set make_partitioned_sstable_set(schema_ptr schema, lw_shared_ptr<sstable_list> all, bool use_level_metadata = true);
+sstable_set make_partitioned_sstable_set(schema_ptr schema, bool use_level_metadata = true);
 
 std::ostream& operator<<(std::ostream& os, const sstables::sstable_run& run);
 

sstables/sstable_set_impl.hh

@@ -41,6 +41,7 @@ public:
     virtual std::vector<shared_sstable> select(const dht::partition_range& range) const = 0;
     virtual void insert(shared_sstable sst) = 0;
     virtual void erase(shared_sstable sst) = 0;
+    virtual bool empty() const = 0;
     virtual std::unique_ptr<incremental_selector_impl> make_incremental_selector() const = 0;
 
     virtual flat_mutation_reader create_single_key_sstable_reader(
@@ -65,6 +66,7 @@ public:
     virtual std::vector<shared_sstable> select(const dht::partition_range& range = query::full_partition_range) const override;
     virtual void insert(shared_sstable sst) override;
     virtual void erase(shared_sstable sst) override;
+    virtual bool empty() const override;
     virtual std::unique_ptr<incremental_selector_impl> make_incremental_selector() const override;
     class incremental_selector;
 };
@@ -103,6 +105,7 @@ public:
     virtual std::vector<shared_sstable> select(const dht::partition_range& range) const override;
     virtual void insert(shared_sstable sst) override;
     virtual void erase(shared_sstable sst) override;
+    virtual bool empty() const override;
     virtual std::unique_ptr<incremental_selector_impl> make_incremental_selector() const override;
     class incremental_selector;
 };
@@ -124,6 +127,7 @@ public:
     virtual std::vector<shared_sstable> select(const dht::partition_range& range = query::full_partition_range) const override;
     virtual void insert(shared_sstable sst) override;
     virtual void erase(shared_sstable sst) override;
+    virtual bool empty() const override;
     virtual std::unique_ptr<incremental_selector_impl> make_incremental_selector() const override;
 
     std::unique_ptr<position_reader_queue> make_min_position_reader_queue(