Merge 'sstables: generation_type tidy-up' from Michael Livshin

- Use `sstables::generation_type` in more places
- Enforce conceptual separation of `sstables::generation_type` and `int64_t`
- Fix `extremum_tracker` so that `sstables::generation_type` can be non-default-constructible

Fixes scylladb#10796.

Closes scylladb#10844

* github.com:scylladb/scylla:
  sstables: make generation_type an actual separate type
  sstables: use generation_type more soundly
  extremum_tracker: do not require default-constructible value types

Author: denesb

api/storage_service.cc

@@ -1183,7 +1183,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
                             ss::sstable info;
 
                             info.timestamp = t;
-                            info.generation = sstable->generation();
+                            info.generation = sstables::generation_value(sstable->generation());
                             info.level = sstable->get_sstable_level();
                             info.size = sstable->bytes_on_disk();
                             info.data_size = sstable->ondisk_data_size();

replica/database.hh

@@ -428,7 +428,7 @@ private:
     std::vector<sstables::shared_sstable> _sstables_compacted_but_not_deleted;
     // sstables that should not be compacted (e.g. because they need to be used
     // to generate view updates later)
-    std::unordered_map<uint64_t, sstables::shared_sstable> _sstables_staging;
+    std::unordered_map<sstables::generation_type, sstables::shared_sstable> _sstables_staging;
     // Control background fibers waiting for sstables to be deleted
     seastar::gate _sstable_deletion_gate;
     // This semaphore ensures that an operation like snapshot won't have its selected
@@ -574,24 +574,24 @@ private:
     struct merge_comparator;
 
     // update the sstable generation, making sure that new new sstables don't overwrite this one.
-    void update_sstables_known_generation(unsigned generation) {
+    void update_sstables_known_generation(sstables::generation_type generation) {
         if (!_sstable_generation) {
             _sstable_generation = 1;
         }
-        _sstable_generation = std::max<uint64_t>(*_sstable_generation, generation /  smp::count + 1);
+        _sstable_generation = std::max<uint64_t>(*_sstable_generation, sstables::generation_value(generation) / smp::count + 1);
     }
 
     sstables::generation_type calculate_generation_for_new_table() {
         assert(_sstable_generation);
         // FIXME: better way of ensuring we don't attempt to
         // overwrite an existing table.
-        return (*_sstable_generation)++ * smp::count + this_shard_id();
+        return sstables::generation_from_value((*_sstable_generation)++ * smp::count + this_shard_id());
     }
 
     // inverse of calculate_generation_for_new_table(), used to determine which
     // shard a sstable should be opened at.
-    static int64_t calculate_shard_from_sstable_generation(int64_t sstable_generation) {
-        return sstable_generation % smp::count;
+    static seastar::shard_id calculate_shard_from_sstable_generation(sstables::generation_type sstable_generation) {
+        return sstables::generation_value(sstable_generation) % smp::count;
     }
 public:
     // This will update sstable lists on behalf of off-strategy compaction, where
@@ -668,7 +668,7 @@ public:
     // likely already called. We need to call this explicitly when we are sure we're ready
     // to issue disk operations safely.
     void mark_ready_for_writes() {
-        update_sstables_known_generation(0);
+        update_sstables_known_generation(sstables::generation_from_value(0));
     }
 
     // Creates a mutation reader which covers all data sources for this column family.

replica/distributed_loader.cc

@@ -219,9 +219,9 @@ distributed_loader::reshard(sharded<sstables::sstable_directory>& dir, sharded<r
     co_await run_resharding_jobs(dir, std::move(destinations), db, ks_name, table_name, std::move(creator));
 }
 
-future<int64_t>
+future<sstables::generation_type>
 highest_generation_seen(sharded<sstables::sstable_directory>& directory) {
-    return directory.map_reduce0(std::mem_fn(&sstables::sstable_directory::highest_generation_seen), int64_t(0), [] (int64_t a, int64_t b) {
+    return directory.map_reduce0(std::mem_fn(&sstables::sstable_directory::highest_generation_seen), sstables::generation_from_value(0), [] (sstables::generation_type a, sstables::generation_type b) {
         return std::max(a, b);
     });
 }
@@ -316,7 +316,7 @@ distributed_loader::process_upload_dir(distributed<replica::database>& db, distr
         process_sstable_dir(directory).get();
 
         auto generation = highest_generation_seen(directory).get0();
-        auto shard_generation_base = generation / smp::count + 1;
+        auto shard_generation_base = sstables::generation_value(generation) / smp::count + 1;
 
         // We still want to do our best to keep the generation numbers shard-friendly.
         // Each destination shard will manage its own generation counter.
@@ -329,14 +329,14 @@ distributed_loader::process_upload_dir(distributed<replica::database>& db, distr
             // we need generation calculated by instance of cf at requested shard
             auto gen = shard_gen[shard].fetch_add(smp::count, std::memory_order_relaxed);
 
-            return global_table->make_sstable(upload.native(), gen,
+            return global_table->make_sstable(upload.native(), sstables::generation_from_value(gen),
                     global_table->get_sstables_manager().get_highest_supported_format(),
                     sstables::sstable::format_types::big, &error_handler_gen_for_upload_dir);
         }).get();
 
         reshape(directory, db, sstables::reshape_mode::strict, ks, cf, [global_table, upload, &shard_gen] (shard_id shard) {
             auto gen = shard_gen[shard].fetch_add(smp::count, std::memory_order_relaxed);
-            return global_table->make_sstable(upload.native(), gen,
+            return global_table->make_sstable(upload.native(), sstables::generation_from_value(gen),
                   global_table->get_sstables_manager().get_highest_supported_format(),
                   sstables::sstable::format_types::big,
                   &error_handler_gen_for_upload_dir);

sstables/generation_type.hh

@@ -6,8 +6,64 @@
  * SPDX-License-Identifier: AGPL-3.0-or-later
  */
 
+#pragma once
+
 #include <cstdint>
+#include <compare>
+#include <limits>
+#include <iostream>
+#include <seastar/core/sstring.hh>
 
 namespace sstables {
-using generation_type = int64_t;
+
+class generation_type {
+    int64_t _value;
+public:
+    generation_type() = delete;
+
+    explicit constexpr generation_type(int64_t value) noexcept: _value(value) {}
+    constexpr int64_t value() const noexcept { return _value; }
+
+    constexpr bool operator==(const generation_type& other) const noexcept { return _value == other._value; }
+    constexpr std::strong_ordering operator<=>(const generation_type& other) const noexcept { return _value <=> other._value; }
+};
+
+constexpr generation_type generation_from_value(int64_t value) {
+    return generation_type{value};
+}
+constexpr int64_t generation_value(generation_type generation) {
+    return generation.value();
+}
+
+} //namespace sstables
+
+namespace seastar {
+template <typename string_type = sstring>
+string_type to_sstring(sstables::generation_type generation) {
+    return to_sstring(sstables::generation_value(generation));
+}
+} //namespace seastar
+
+namespace std {
+template <>
+struct hash<sstables::generation_type> {
+    constexpr size_t operator()(const sstables::generation_type& generation) const noexcept {
+        return hash<int64_t>{}(generation.value());
+    }
+};
+
+// for min_max_tracker
+template <>
+struct numeric_limits<sstables::generation_type> : public numeric_limits<int64_t> {
+    static constexpr sstables::generation_type min() noexcept {
+        return sstables::generation_type{numeric_limits<int64_t>::min()};
+    }
+    static constexpr sstables::generation_type max() noexcept {
+        return sstables::generation_type{numeric_limits<int64_t>::max()};
+    }
+};
+
+[[maybe_unused]] static ostream& operator<<(ostream& s, const sstables::generation_type& generation) {
+    return s << generation.value();
 }
+} //namespace std

sstables/sstable_directory.cc

@@ -50,7 +50,7 @@ sstable_directory::sstable_directory(fs::path sstable_dir,
 
 void
 sstable_directory::handle_component(scan_state& state, sstables::entry_descriptor desc, fs::path filename) {
-    if ((desc.generation % smp::count) != this_shard_id()) {
+    if ((generation_value(desc.generation) % smp::count) != this_shard_id()) {
         return;
     }
 
@@ -142,7 +142,7 @@ sstable_directory::sort_sstable(sstables::shared_sstable sst) {
     });
 }
 
-int64_t
+generation_type
 sstable_directory::highest_generation_seen() const {
     return _max_generation_seen;
 }
@@ -202,16 +202,16 @@ sstable_directory::process_sstable_dir(const ::io_priority_class& iop, bool sort
         state.descriptors.erase(desc.generation);
     }
 
-    _max_generation_seen =  boost::accumulate(state.generations_found | boost::adaptors::map_keys, int64_t(0), [] (int64_t a, int64_t b) {
-        return std::max<int64_t>(a, b);
+    _max_generation_seen =  boost::accumulate(state.generations_found | boost::adaptors::map_keys, generation_from_value(0), [] (generation_type a, generation_type b) {
+        return std::max<generation_type>(a, b);
     });
 
     dirlog.debug("After {} scanned, seen generation {}. {} descriptors found, {} different files found ",
             _sstable_dir, _max_generation_seen, state.descriptors.size(), state.generations_found.size());
 
     // _descriptors is everything with a TOC. So after we remove this, what's left is
     // SSTables for which a TOC was not found.
-    co_await parallel_for_each_restricted(state.descriptors, [this, sort_sstables_according_to_owner, &state, &iop] (std::tuple<int64_t, sstables::entry_descriptor>&& t) {
+    co_await parallel_for_each_restricted(state.descriptors, [this, sort_sstables_according_to_owner, &state, &iop] (std::tuple<generation_type, sstables::entry_descriptor>&& t) {
         auto& desc = std::get<1>(t);
         state.generations_found.erase(desc.generation);
         // This will try to pre-load this file and throw an exception if it is invalid

sstables/sstable_directory.hh

@@ -85,7 +85,7 @@ private:
     // How to create an SSTable object from an existing SSTable file (respecting generation, etc)
     sstable_object_from_existing_fn _sstable_object_from_existing_sstable;
 
-    int64_t _max_generation_seen = 0;
+    generation_type _max_generation_seen = generation_from_value(0);
     sstables::sstable_version_types _max_version_seen = sstables::sstable_version_types::ka;
 
     // SSTables that are unshared and belong to this shard. They are already stored as an
@@ -137,7 +137,7 @@ public:
     future<> move_foreign_sstables(sharded<sstable_directory>& source_directory);
 
     // returns what is the highest generation seen in this directory.
-    int64_t highest_generation_seen() const;
+    generation_type highest_generation_seen() const;
 
     // returns what is the highest version seen in this directory.
     sstables::sstable_version_types highest_version_seen() const;

sstables/sstable_set.cc

@@ -628,7 +628,7 @@ class incremental_reader_selector : public reader_selector {
     lw_shared_ptr<const sstable_set> _sstables;
     tracing::trace_state_ptr _trace_state;
     std::optional<sstable_set::incremental_selector> _selector;
-    std::unordered_set<int64_t> _read_sstable_gens;
+    std::unordered_set<generation_type> _read_sstable_gens;
     sstable_reader_factory_type _fn;
 
     flat_mutation_reader_v2 create_reader(shared_sstable sst) {

sstables/sstables.cc

@@ -2256,7 +2256,7 @@ static entry_descriptor make_entry_descriptor(sstring sstdir, sstring fname, sst
     } else {
         throw malformed_sstable_exception(seastar::format("invalid version for file {}. Name doesn't match any known version.", fname));
     }
-    return entry_descriptor(sstdir, ks, cf, boost::lexical_cast<unsigned long>(generation), version, sstable::format_from_sstring(format), sstable::component_from_sstring(version, component));
+    return entry_descriptor(sstdir, ks, cf, generation_from_value(boost::lexical_cast<unsigned long>(generation)), version, sstable::format_from_sstring(format), sstable::component_from_sstring(version, component));
 }
 
 entry_descriptor entry_descriptor::make_descriptor(sstring sstdir, sstring fname) {

sstables/sstables.hh

@@ -504,7 +504,7 @@ private:
     schema_ptr _schema;
     sstring _dir;
     std::optional<sstring> _temp_dir; // Valid while the sstable is being created, until sealed
-    generation_type _generation = 0;
+    generation_type _generation{0};
 
     version_types _version;
     format_types _format;

test/boost/database_test.cc

@@ -253,25 +253,25 @@ SEASTAR_THREAD_TEST_CASE(test_distributed_loader_with_incomplete_sstables) {
         require_exist(file_name, true);
     };
 
-    auto temp_sst_dir = sst::temp_sst_dir(sst_dir, 2);
+    auto temp_sst_dir = sst::temp_sst_dir(sst_dir, generation_from_value(2));
     touch_dir(temp_sst_dir);
 
-    temp_sst_dir = sst::temp_sst_dir(sst_dir, 3);
+    temp_sst_dir = sst::temp_sst_dir(sst_dir, generation_from_value(3));
     touch_dir(temp_sst_dir);
-    auto temp_file_name = sst::filename(temp_sst_dir, ks, cf, sst::version_types::mc, 3, sst::format_types::big, component_type::TemporaryTOC);
+    auto temp_file_name = sst::filename(temp_sst_dir, ks, cf, sst::version_types::mc, generation_from_value(3), sst::format_types::big, component_type::TemporaryTOC);
     touch_file(temp_file_name);
 
-    temp_file_name = sst::filename(sst_dir, ks, cf, sst::version_types::mc, 4, sst::format_types::big, component_type::TemporaryTOC);
+    temp_file_name = sst::filename(sst_dir, ks, cf, sst::version_types::mc, generation_from_value(4), sst::format_types::big, component_type::TemporaryTOC);
     touch_file(temp_file_name);
-    temp_file_name = sst::filename(sst_dir, ks, cf, sst::version_types::mc, 4, sst::format_types::big, component_type::Data);
+    temp_file_name = sst::filename(sst_dir, ks, cf, sst::version_types::mc, generation_from_value(4), sst::format_types::big, component_type::Data);
     touch_file(temp_file_name);
 
     do_with_cql_env_thread([&sst_dir, &ks, &cf, &require_exist] (cql_test_env& e) {
-        require_exist(sst::temp_sst_dir(sst_dir, 2), false);
-        require_exist(sst::temp_sst_dir(sst_dir, 3), false);
+        require_exist(sst::temp_sst_dir(sst_dir, generation_from_value(2)), false);
+        require_exist(sst::temp_sst_dir(sst_dir, generation_from_value(3)), false);
 
-        require_exist(sst::filename(sst_dir, ks, cf, sst::version_types::mc, 4, sst::format_types::big, component_type::TemporaryTOC), false);
-        require_exist(sst::filename(sst_dir, ks, cf, sst::version_types::mc, 4, sst::format_types::big, component_type::Data), false);
+        require_exist(sst::filename(sst_dir, ks, cf, sst::version_types::mc, generation_from_value(4), sst::format_types::big, component_type::TemporaryTOC), false);
+        require_exist(sst::filename(sst_dir, ks, cf, sst::version_types::mc, generation_from_value(4), sst::format_types::big, component_type::Data), false);
     }, db_cfg_ptr).get();
 }
 
@@ -320,11 +320,11 @@ SEASTAR_THREAD_TEST_CASE(test_distributed_loader_with_pending_delete) {
     };
 
     auto component_basename = [&ks, &cf] (int64_t gen, component_type ctype) {
-        return sst::component_basename(ks, cf, sst::version_types::mc, gen, sst::format_types::big, ctype);
+        return sst::component_basename(ks, cf, sst::version_types::mc, generation_from_value(gen), sst::format_types::big, ctype);
     };
 
     auto gen_filename = [&sst_dir, &ks, &cf] (int64_t gen, component_type ctype) {
-        return sst::filename(sst_dir, ks, cf, sst::version_types::mc, gen, sst::format_types::big, ctype);
+        return sst::filename(sst_dir, ks, cf, sst::version_types::mc, generation_from_value(gen), sst::format_types::big, ctype);
     };
 
     touch_dir(pending_delete_dir);

test/boost/mutation_reader_test.cc

@@ -3876,7 +3876,7 @@ static future<> do_test_clustering_order_merger_sstable_set(bool reversed) {
                 return sst.make_reader(query_schema, permit, pr,
                                           query_slice, seastar::default_priority_class(), nullptr, fwd);
             },
-            [included_gens] (const sstable& sst) { return included_gens.contains(sst.generation()); },
+            [included_gens] (const sstable& sst) { return included_gens.contains(generation_value(sst.generation())); },
             pk, query_schema, permit, fwd, reversed);
         return make_clustering_combined_reader(query_schema, permit, fwd, std::move(q));
     };

test/boost/sstable_3_x_test.cc

@@ -3211,9 +3211,9 @@ static void compare_sstables(const std::filesystem::path& result_path, sstring t
                            component_type::Filter}) {
         auto orig_filename =
                 sstable::filename(get_write_test_path(table_name),
-                                  "ks", table_name, sstables::sstable_version_types::mc, 1, big, file_type);
+                                  "ks", table_name, sstables::sstable_version_types::mc, generation_from_value(1), big, file_type);
         auto result_filename =
-                sstable::filename(result_path.string(), "ks", table_name, version, 1, big, file_type);
+                sstable::filename(result_path.string(), "ks", table_name, version, generation_from_value(1), big, file_type);
         compare_files(orig_filename, result_filename);
     }
 }
@@ -5194,7 +5194,7 @@ static void test_sstable_write_large_row_f(schema_ptr s, reader_permit permit, r
     auto stop_manager = defer([&] { manager.close().get(); });
     tmpdir dir;
     auto sst = manager.make_sstable(
-            s, dir.path().string(), 1 /* generation */, version, sstables::sstable::format_types::big);
+            s, dir.path().string(), generation_from_value(1), version, sstables::sstable::format_types::big);
 
     // The test provides thresholds values for the large row handler. Whether the handler gets
     // trigger depends on the size of rows after they are written in the MC format and that size
@@ -5253,7 +5253,7 @@ static void test_sstable_log_too_many_rows_f(int rows, uint64_t threshold, bool
     sstables_manager manager(handler, test_db_config, test_feature_service, tracker);
     auto close_manager = defer([&] { manager.close().get(); });
     tmpdir dir;
-    auto sst = manager.make_sstable(sc, dir.path().string(), 1, version, sstables::sstable::format_types::big);
+    auto sst = manager.make_sstable(sc, dir.path().string(), generation_from_value(1), version, sstables::sstable::format_types::big);
     sst->write_components(mt->make_flat_reader(sc, semaphore.make_permit()), 1, sc, manager.configure_writer("test"), encoding_stats{}).get();
 
     BOOST_REQUIRE_EQUAL(logged, expected);