compaction: Enable filtering reader only on behalf of cleanup compaction

After 13fa2be, every compaction will be performed through a filtering
reader because consumers cannot do the filtering if interposer consumer is
enabled.

It turns out that filtering_reader is adding significant overhead when regular
compactions are running. As no other compaction type need to actually do
any filtering, let's limit filtering_reader to cleanup compaction.
Alternatively, we could disable interposer consumer on behalf of cleanup,
or add support for the consumers to do the filtering themselves but that
would add lots of complexity.

Fixes #7748.

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

sstables/compaction.cc

@@ -614,9 +614,7 @@ class compaction {
                 reader.consume_in_thread(std::move(cfc), db::no_timeout);
             });
         });
-        // producer will filter out a partition before it reaches the consumer(s)
-        auto producer = make_filtering_reader(make_sstable_reader(), make_partition_filter());
-        return consumer(std::move(producer));
+        return consumer(make_sstable_reader());
     }
 
     virtual reader_consumer make_interposer_consumer(reader_consumer end_consumer) {
@@ -673,12 +671,6 @@ class compaction {
         };
     }
 
-    virtual flat_mutation_reader::filter make_partition_filter() const {
-        return [] (const dht::decorated_key&) {
-            return true;
-        };
-    }
-
     virtual void on_new_partition() {}
 
     virtual void on_end_of_compaction() {};
@@ -1070,6 +1062,10 @@ class cleanup_compaction final : public regular_compaction {
     cleanup_compaction(column_family& cf, compaction_descriptor descriptor, compaction_options::upgrade opts)
         : cleanup_compaction(opts.db, cf, std::move(descriptor)) {}
 
+    flat_mutation_reader make_sstable_reader() const override {
+        return make_filtering_reader(regular_compaction::make_sstable_reader(), make_partition_filter());
+    }
+
     std::string_view report_start_desc() const override {
         return "Cleaning";
     }
@@ -1078,7 +1074,7 @@ class cleanup_compaction final : public regular_compaction {
         return "Cleaned";
     }
 
-    flat_mutation_reader::filter make_partition_filter() const override {
+    flat_mutation_reader::filter make_partition_filter() const {
         return [this] (const dht::decorated_key& dk) {
             if (dht::shard_of(*_schema, dk.token()) != this_shard_id()) {
                 log_trace("Token {} does not belong to CPU {}, skipping", dk.token(), this_shard_id());