table: Make sure major compaction doesn't miss data in memtable

Make sure that major will compact data in all sstables and memtable,
as tombstones sitting in memtable could shadow data in sstables.
For example, a tombstone in memtable deleting a large partition could
be missed in major, so space wouldn't be saved as expected.
Additionally, write amplification is reduced as data in memtable
won't have to travel through tiers once flushed.

Fixes scylladb#9514.

Signed-off-by: Raphael S. Carvalho <raphaelsc@scylladb.com>
Message-Id: <20211217160055.96693-2-raphaelsc@scylladb.com>

database.hh

@@ -820,10 +820,9 @@ public:
 
     bool can_flush() const;
 
-    // FIXME: this is just an example, should be changed to something more
-    // general. compact_all_sstables() starts a compaction of all sstables.
-    // It doesn't flush the current memtable first. It's just a ad-hoc method,
-    // not a real compaction policy.
+    // Start a compaction of all sstables in a process known as major compaction
+    // Active memtable is flushed first to guarantee that data like tombstone,
+    // sitting in the memtable, will be compacted with shadowed data.
     future<> compact_all_sstables();
     // Compact all sstables provided in the vector.
     future<> compact_sstables(sstables::compaction_descriptor descriptor, sstables::compaction_data& cdata);

table.cc

@@ -989,10 +989,10 @@ table::compact_sstables(sstables::compaction_descriptor descriptor, sstables::co
     });
 }
 
-// Note: We assume that the column_family does not get destroyed during compaction.
 future<>
 table::compact_all_sstables() {
-    return _compaction_manager.perform_major_compaction(this);
+    co_await flush();
+    co_await _compaction_manager.perform_major_compaction(this);
 }
 
 void table::start_compaction() {

test/boost/sstable_compaction_test.cc

@@ -4724,3 +4724,51 @@ SEASTAR_TEST_CASE(twcs_single_key_reader_through_compound_set_test) {
         BOOST_REQUIRE(cf_stats.clustering_filter_count > 0);
     });
 }
+
+SEASTAR_TEST_CASE(test_major_does_not_miss_data_in_memtable) {
+    return test_env::do_with_async([] (test_env& env) {
+        auto builder = schema_builder("tests", "test_major_does_not_miss_data_in_memtable")
+                .with_column("id", utf8_type, column_kind::partition_key)
+                .with_column("cl", int32_type, column_kind::clustering_key)
+                .with_column("value", int32_type);
+        auto s = builder.build();
+
+        auto tmp = tmpdir();
+        auto tokens = token_generation_for_shard(1, this_shard_id(), test_db_config.murmur3_partitioner_ignore_msb_bits(), smp::count);
+        auto pkey = partition_key::from_exploded(*s, {to_bytes(tokens[0].first)});
+
+        column_family_for_tests cf(env.manager(), s, tmp.path().string());
+        auto close_cf = deferred_stop(cf);
+        auto sst_gen = [&env, &cf, s, &tmp] () mutable {
+            return env.make_sstable(s, tmp.path().string(), column_family_test::calculate_generation_for_new_table(*cf),
+                sstables::get_highest_sstable_version(), big);
+        };
+
+        auto row_mut = [&] () {
+            static thread_local int32_t value = 1;
+            mutation m(s, pkey);
+            auto c_key = clustering_key::from_exploded(*s, {int32_type->decompose(value++)});
+            m.set_clustered_cell(c_key, bytes("value"), data_value(int32_t(value)), gc_clock::now().time_since_epoch().count());
+            return m;
+        }();
+        auto sst = make_sstable_containing(sst_gen, {std::move(row_mut)});
+        cf->add_sstable_and_update_cache(sst).get();
+        BOOST_REQUIRE(cf->get_sstables()->size() == 1);
+
+        auto deletion_mut = [&] () {
+            mutation m(s, pkey);
+            tombstone tomb(gc_clock::now().time_since_epoch().count(), gc_clock::now());
+            m.partition().apply(tomb);
+            return m;
+        }();
+        cf->apply(deletion_mut);
+
+        cf->compact_all_sstables().get();
+        BOOST_REQUIRE(cf->get_sstables()->size() == 1);
+        auto new_sst = *(cf->get_sstables()->begin());
+        BOOST_REQUIRE(new_sst->generation() != sst->generation());
+        assert_that(sstable_reader(new_sst, s, env.make_reader_permit()))
+                .produces(deletion_mut)
+                .produces_end_of_stream();
+    });
+}