From 17d61635481376cdadcfda40f98cb7fdd019fc71 Mon Sep 17 00:00:00 2001
From: Tomasz Grabiec <tgrabiec@scylladb.com>
Date: Thu, 11 May 2023 13:43:48 +0200
Subject: [PATCH] sstables: Generate sharding metadata using sharder from erm
 when writing

We need to keep sharding metadata consistent with tablet mapping to
shards in order for node restart to detect that those sstables belong
to a single shard and that resharding is not necessary. Resharding of
sstables based on tablet metadata is not implemented yet and will
abort after this series.

Keeping sharding metadata accurate for tablets is only necessary until
compaction group integration is finished. After that, we can use the
sstable token range to determine the owning tablet and thus the owning
shard. Before that, we can't, because a single sstable may contain
keys from different tablets, and the whole key range may overlap with
keys which belong to other shards.
---
 replica/table.cc      |  5 ++++-
 sstables/mx/writer.cc |  4 +++-
 sstables/sstables.cc  |  5 +++--
 sstables/sstables.hh  | 11 +++++++++--
 streaming/consumer.cc |  5 +++--
 5 files changed, 22 insertions(+), 8 deletions(-)

diff --git a/replica/table.cc b/replica/table.cc
index 6b3548315f..f3f85c9cdc 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -873,6 +873,7 @@ table::try_flush_memtable_to_sstable(compaction_group& cg, lw_shared_ptr<memtabl
           try {
             sstables::sstable_writer_config cfg = get_sstables_manager().configure_writer("memtable");
             cfg.backup = incremental_backups_enabled();
+            cfg.erm = _erm;
 
             auto newtab = make_sstable();
             newtabs.push_back(newtab);
@@ -2782,7 +2783,9 @@ public:
         return _t.make_sstable();
     }
     sstables::sstable_writer_config configure_writer(sstring origin) const override {
-        return _t.get_sstables_manager().configure_writer(std::move(origin));
+        auto cfg = _t.get_sstables_manager().configure_writer(std::move(origin));
+        cfg.erm = _t.get_effective_replication_map();
+        return cfg;
     }
     api::timestamp_type min_memtable_timestamp() const override {
         return _cg.min_memtable_timestamp();
diff --git a/sstables/mx/writer.cc b/sstables/mx/writer.cc
index aed4c74036..f9329b1c4f 100644
--- a/sstables/mx/writer.cc
+++ b/sstables/mx/writer.cc
@@ -1471,7 +1471,9 @@ void writer::consume_end_of_stream() {
             { large_data_type::elements_in_collection, std::move(_elements_in_collection_entry) },
         }
     });
-    _sst.write_scylla_metadata(_shard, std::move(features), std::move(identifier), std::move(ld_stats), _cfg.origin);
+    const dht::sharder& sharder = _cfg.erm ? _cfg.erm->get_sharder(_schema)
+                                           : _schema.get_sharder(); // Used in tests
+    _sst.write_scylla_metadata(_shard, sharder, std::move(features), std::move(identifier), std::move(ld_stats), _cfg.origin);
     _sst.seal_sstable(_cfg.backup).get();
 }
 
diff --git a/sstables/sstables.cc b/sstables/sstables.cc
index b5fcf4be2a..5370bd7695 100644
--- a/sstables/sstables.cc
+++ b/sstables/sstables.cc
@@ -1624,11 +1624,12 @@ sstable::read_scylla_metadata() noexcept {
 }
 
 void
-sstable::write_scylla_metadata(shard_id shard, sstable_enabled_features features, struct run_identifier identifier,
+sstable::write_scylla_metadata(shard_id shard, const dht::sharder& sharder, sstable_enabled_features features, struct run_identifier identifier,
         std::optional<scylla_metadata::large_data_stats> ld_stats, sstring origin) {
     auto&& first_key = get_first_decorated_key();
     auto&& last_key = get_last_decorated_key();
-    auto sm = create_sharding_metadata(_schema, _schema->get_sharder(), first_key, last_key, shard);
+
+    auto sm = create_sharding_metadata(_schema, sharder, first_key, last_key, shard);
 
     // sstable write may fail to generate empty metadata if mutation source has only data from other shard.
     // see https://github.com/scylladb/scylla/issues/2932 for details on how it can happen.
diff --git a/sstables/sstables.hh b/sstables/sstables.hh
index 5a9fa16dba..e735643112 100644
--- a/sstables/sstables.hh
+++ b/sstables/sstables.hh
@@ -39,6 +39,7 @@
 #include "readers/flat_mutation_reader_fwd.hh"
 #include "tracing/trace_state.hh"
 #include "utils/updateable_value.hh"
+#include "locator/abstract_replication_strategy.hh"
 
 #include <seastar/util/optimized_optional.hh>
 
@@ -110,6 +111,7 @@ struct sstable_writer_config {
     run_id run_identifier = run_id::create_random_id();
     size_t summary_byte_cost;
     sstring origin;
+    locator::effective_replication_map_ptr erm;
 
 private:
     explicit sstable_writer_config() {}
@@ -579,8 +581,13 @@ private:
     void write_compression();
 
     future<> read_scylla_metadata() noexcept;
-    void write_scylla_metadata(shard_id shard, sstable_enabled_features features, run_identifier identifier,
-            std::optional<scylla_metadata::large_data_stats> ld_stats, sstring origin);
+
+    void write_scylla_metadata(shard_id shard,
+                               const dht::sharder& sharder,
+                               sstable_enabled_features features,
+                               run_identifier identifier,
+                               std::optional<scylla_metadata::large_data_stats> ld_stats,
+                               sstring origin);
 
     future<> read_filter(sstable_open_config cfg = {});
 
diff --git a/streaming/consumer.cc b/streaming/consumer.cc
index 5b748d7cf4..a62d2bea08 100644
--- a/streaming/consumer.cc
+++ b/streaming/consumer.cc
@@ -54,9 +54,10 @@ std::function<future<> (flat_mutation_reader_v2)> make_streaming_consumer(sstrin
                 }
                 schema_ptr s = reader.schema();
 
+                auto cfg = cf->get_sstables_manager().configure_writer(origin);
+                cfg.erm = cf->get_effective_replication_map();
                 return sst->write_components(std::move(reader), adjusted_estimated_partitions, s,
-                                             cf->get_sstables_manager().configure_writer(origin),
-                                             encoding_stats{}).then([sst] {
+                                             cfg, encoding_stats{}).then([sst] {
                     return sst->open_data();
                 }).then([cf, sst, offstrategy, origin] {
                     if (offstrategy && sstables::repair_origin == origin) {