db: fix migration of sstables with level greater than 0

Refresh will rewrite statistics of any migrated sstable with level
> 0. However, this operation is currently not working because O_EXCL
flag is used, meaning that create will fail.

It turns out that we don't actually need to change on-disk level of
a sstable by overwriting statistics file.
We can only set in-memory level of a sstable to 0. If Scylla reboots
before all migrated sstables are compacted, leveled strategy is smart
enough to detect sstables that overlap, and set their in-memory level
to 0.

Fixes #1124.

Signed-off-by: Raphael S. Carvalho <raphaelsc@scylladb.com>

database.cc

@@ -984,7 +984,13 @@ column_family::load_new_sstables(std::vector<sstables::entry_descriptor> new_tab
     return parallel_for_each(new_tables, [this] (auto comps) {
         auto sst = make_lw_shared<sstables::sstable>(_schema->ks_name(), _schema->cf_name(), _config.datadir, comps.generation, comps.version, comps.format);
         return sst->load().then([this, sst] {
-            return sst->mutate_sstable_level(0);
+            // This sets in-memory level of sstable to 0.
+            // When loading a migrated sstable, it's important to set it to level 0 because
+            // leveled compaction relies on a level > 0 having no overlapping sstables.
+            // If Scylla reboots before migrated sstable gets compacted, leveled strategy
+            // is smart enough to detect a sstable that overlaps and set its in-memory
+            // level to 0.
+            return sst->set_sstable_level(0);
         }).then([this, sst] {
             auto first = sst->get_first_partition_key(*_schema);
             auto last = sst->get_last_partition_key(*_schema);

sstables/leveled_manifest.hh

@@ -232,20 +232,8 @@ public:
 
     void send_back_to_L0(sstables::shared_sstable& sstable) {
         remove(sstable);
-#if 0
-        try
-        {
-            sstable.descriptor.getMetadataSerializer().mutateLevel(sstable.descriptor, 0);
-            sstable.reloadSSTableMetadata();
-            add(sstable);
-        }
-        catch (IOException e)
-        {
-            throw new RuntimeException("Could not reload sstable meta data", e);
-        }
-#else
         _generations[0].push_back(sstable);
-#endif
+        sstable->set_sstable_level(0);
     }
 
 #if 0

sstables/sstables.cc

@@ -1795,6 +1795,20 @@ double sstable::get_compression_ratio() const {
     }
 }
 
+void sstable::set_sstable_level(uint32_t new_level) {
+    auto entry = _statistics.contents.find(metadata_type::Stats);
+    if (entry == _statistics.contents.end()) {
+        return;
+    }
+    auto& p = entry->second;
+    if (!p) {
+        throw std::runtime_error("Statistics is malformed");
+    }
+    stats_metadata& s = *static_cast<stats_metadata *>(p.get());
+    sstlog.debug("set level of {} with generation {} from {} to {}", get_filename(), _generation, s.sstable_level, new_level);
+    s.sstable_level = new_level;
+}
+
 future<> sstable::mutate_sstable_level(uint32_t new_level) {
     if (!has_component(component_type::Statistics)) {
         return make_ready_future<>();

sstables/sstables.hh

@@ -535,6 +535,9 @@ public:
         return get_stats_metadata().sstable_level;
     }
 
+    // This will change sstable level only in memory.
+    void set_sstable_level(uint32_t);
+
     double get_compression_ratio() const;
 
     future<> mutate_sstable_level(uint32_t);