Fix Bug of L0L1 timer starting after compaction started

bug was that compaction started because the size of L0 exceeded
max_bytes_for_level_base and we only considered files above trigger.

db/column_family.cc

@@ -1129,14 +1129,27 @@ void ColumnFamilyData::SetL0BaseCompactionSpeed(uint64_t size) {
     started_l0_timer_ = false;
     l0_start_clearance_time_ = 0;
     ROCKS_LOG_INFO(ioptions_.logger,
-                   "L0L1 compaction ended. duration : %f"
+                   "L0L1 compaction ended. duration: %f"
                    ", amount cleared: %" PRIu64 ", current rate: %" PRIu64
                    ", avg rate: %" PRIu64,
                    l0_clearance_dur, size, cur_speed,
                    lo_base_compaction_speed_);
   }
 }
 
+void ColumnFamilyData::MaybeStartL0L1Timer() {
+  auto* vstorage = current_->storage_info();
+  // start timer for L0 clearance when trigger passed.
+  if (!started_l0_timer_ && vstorage->IsL0ScoreAbove1()) {
+    started_l0_timer_ = true;
+    l0_start_clearance_time_ = ioptions_.clock->NowMicros();
+    ROCKS_LOG_INFO(
+        ioptions_.logger,
+        "Auto tune: Started timer. time: %" PRIu64 " Num L0 files: %d",
+        l0_start_clearance_time_, vstorage->l0_delay_trigger_count());
+  }
+}
+
 WriteStallCondition ColumnFamilyData::RecalculateWriteStallConditions(
     const MutableCFOptions& mutable_cf_options) {
   auto write_stall_condition = WriteStallCondition::kNormal;
@@ -1158,17 +1171,8 @@ WriteStallCondition ColumnFamilyData::RecalculateWriteStallConditions(
     bool dynamic_delay = write_controller->is_dynamic_delay();
 
     if (dynamic_delay) {
-      // start timer for L0 clearance when trigger passed.
-      if (!started_l0_timer_ &&
-          vstorage->l0_delay_trigger_count() >=
-              mutable_cf_options.level0_file_num_compaction_trigger) {
-        started_l0_timer_ = true;
-        l0_start_clearance_time_ = ioptions_.clock->NowMicros();
-        ROCKS_LOG_INFO(
-            ioptions_.logger,
-            "Auto tune: Started timer. time: %" PRIu64 " Num L0 files: %d",
-            l0_start_clearance_time_, vstorage->l0_delay_trigger_count());
-      }
+      MaybeStartL0L1Timer();
+
       // TODO: treat stop pending bytes as delay until we have a better
       // way of handling it.
       if (write_stall_condition == WriteStallCondition::kStopped &&

db/column_family.h

@@ -522,6 +522,8 @@ class ColumnFamilyData {
   void SetL0BaseCompactionSpeed(uint64_t size);
 
  private:
+  void MaybeStartL0L1Timer();
+
   // In bytes per sec. same as delayed_write_rate
   uint64_t l0_base_compaction_speed() const {
     return lo_base_compaction_speed_;

db/compaction/compaction_job.cc

@@ -888,10 +888,12 @@ Status CompactionJob::Install(const MutableCFOptions& mutable_cf_options) {
   auto vstorage = cfd->current()->storage_info();
   const auto& stats = compaction_stats_.stats;
 
+  auto const compaction = compact_->compaction;
   // Set the speed of L0L1 compaction for auto tuning delayed write rate.
-  if (output_level == vstorage->base_level() &&
-      compact_->compaction->start_level() == 0) {
-    if (compact_->compaction->immutable_options()->compaction_style ==
+  if ((output_level == vstorage->base_level() &&
+       compaction->start_level() == 0) &&
+      (compaction->compaction_reason() == CompactionReason::kLevelL0FilesNum)) {
+    if (compaction->immutable_options()->compaction_style ==
         kCompactionStyleLevel) {
       cfd->SetL0BaseCompactionSpeed(stats.bytes_read_non_output_levels);
     }

db/version_set.cc

@@ -3631,6 +3631,15 @@ void VersionStorageInfo::ComputeCompactionScore(
   EstimateCompactionBytesNeeded(mutable_cf_options);
 }
 
+bool VersionStorageInfo::IsL0ScoreAbove1() {
+  for (size_t i = 0; i < compaction_level_.size(); i++) {
+    if (compaction_level_[i] == 0) {
+      return compaction_score_[i] > 1;
+    }
+  }
+  return false;
+}
+
 void VersionStorageInfo::ComputeFilesMarkedForCompaction(int last_level) {
   files_marked_for_compaction_.clear();
   int last_qualify_level = 0;

db/version_set.h

@@ -609,6 +609,8 @@ class VersionStorageInfo {
     return bottommost_files_mark_threshold_;
   }
 
+  bool IsL0ScoreAbove1();
+
   // Returns whether any key in [`smallest_key`, `largest_key`] could appear in
   // an older L0 file than `last_l0_idx` or in a greater level than `last_level`
   //