fix(storage): fix condition of compaction task validator

src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs

@@ -137,7 +137,7 @@ impl LevelCompactionPicker {
             min_compaction_bytes,
             // divide by 2 because we need to select files of base level and it need use the other
             // half quota.
-            std::cmp::min(
+            std::cmp::max(
                 self.config.max_bytes_for_level_base,
                 self.config.max_compaction_bytes / 2,
             ),
@@ -573,6 +573,7 @@ pub mod tests {
         let config = Arc::new(
             CompactionConfigBuilder::new()
                 .max_compaction_bytes(100010)
+                .max_bytes_for_level_base(512)
                 .level0_sub_level_compact_level_count(1)
                 .build(),
         );

src/meta/src/hummock/compaction/picker/compaction_task_validator.rs

@@ -89,32 +89,32 @@ struct TierCompactionTaskValidationRule {
 
 impl CompactionTaskValidationRule for TierCompactionTaskValidationRule {
     fn validate(&self, input: &CompactionInput, stats: &mut LocalPickerStatistic) -> bool {
-        // so the design here wants to merge multiple overlapping-levels in one compaction
-        let max_compaction_bytes = std::cmp::min(
-            self.config.max_compaction_bytes,
-            self.config.sub_level_max_compaction_bytes
-                * self.config.level0_overlapping_sub_level_compact_level_count as u64,
-        );
-
         // Limit sstable file count to avoid using too much memory.
         let overlapping_max_compact_file_numer = std::cmp::min(
             self.config.level0_max_compact_file_number,
             MAX_COMPACT_LEVEL_COUNT as u64,
         );
 
-        let waiting_enough_files = {
-            if input.select_input_size > max_compaction_bytes {
-                false
-            } else {
-                input.total_file_count <= overlapping_max_compact_file_numer
-            }
-        };
+        if input.total_file_count >= overlapping_max_compact_file_numer
+            || input.input_levels.len() >= MAX_COMPACT_LEVEL_COUNT
+        {
+            return true;
+        }
+
+        // so the design here wants to merge multiple overlapping-levels in one compaction
+        let max_compaction_bytes = std::cmp::min(
+            self.config.max_compaction_bytes,
+            self.config.sub_level_max_compaction_bytes
+                * self.config.level0_overlapping_sub_level_compact_level_count as u64,
+        );
 
         // If waiting_enough_files is not satisfied, we will raise the priority of the number of
         // levels to ensure that we can merge as many sub_levels as possible
         let tier_sub_level_compact_level_count =
             self.config.level0_overlapping_sub_level_compact_level_count as usize;
-        if input.input_levels.len() < tier_sub_level_compact_level_count && waiting_enough_files {
+        if input.input_levels.len() < tier_sub_level_compact_level_count
+            && input.select_input_size < max_compaction_bytes
+        {
             stats.skip_by_count_limit += 1;
             return false;
         }
@@ -129,7 +129,9 @@ struct IntraCompactionTaskValidationRule {
 
 impl CompactionTaskValidationRule for IntraCompactionTaskValidationRule {
     fn validate(&self, input: &CompactionInput, stats: &mut LocalPickerStatistic) -> bool {
-        if input.total_file_count >= self.config.level0_max_compact_file_number {
+        if input.total_file_count >= self.config.level0_max_compact_file_number
+            || input.input_levels.len() >= MAX_COMPACT_LEVEL_COUNT
+        {
             return true;
         }
 
@@ -175,6 +177,12 @@ struct BaseCompactionTaskValidationRule {
 
 impl CompactionTaskValidationRule for BaseCompactionTaskValidationRule {
     fn validate(&self, input: &CompactionInput, stats: &mut LocalPickerStatistic) -> bool {
+        if input.total_file_count >= self.config.level0_max_compact_file_number
+            || input.input_levels.len() >= MAX_COMPACT_LEVEL_COUNT
+        {
+            return true;
+        }
+
         // The size of target level may be too large, we shall skip this compact task and wait
         //  the data in base level compact to lower level.
         if input.target_input_size > self.config.max_compaction_bytes {