fix(compaction): fix compaction task do not schedule even if the numb…

…er of L0 files is very large (#3117)

fix

Signed-off-by: Little-Wallace <[email protected]>

src/meta/src/hummock/compaction/tier_compaction_picker.rs

@@ -69,7 +69,7 @@ impl CompactionPicker for TierCompactionPicker {
                     break;
                 }
                 // no need to trigger a bigger compaction
-                if compaction_bytes >= self.config.min_compaction_bytes {
+                if compaction_bytes >= self.config.max_compaction_bytes {
                     break;
                 }
                 compaction_bytes += other.file_size;
@@ -321,7 +321,7 @@ impl LevelCompactionPicker {
                         // we only extend L0 files when write-amplification does not increase.
                         let inc_compaction_size =
                             target_level_ssts.calc_inc_compaction_size(&tables);
-                        if select_compaction_bytes > 0
+                        if select_compaction_bytes > self.config.min_compaction_bytes
                             && (target_level_ssts.compaction_bytes + inc_compaction_size)
                                 / (select_compaction_bytes + other.file_size)
                                 > target_level_ssts.compaction_bytes / select_compaction_bytes
@@ -732,4 +732,46 @@ pub mod tests {
         assert_eq!(ret.target_level.table_infos[0].id, 4);
         assert_eq!(ret.target_level.table_infos[1].id, 5);
     }
+
+    #[test]
+    fn test_compact_with_write_amplification_limit_2() {
+        let config = Arc::new(
+            CompactionConfigBuilder::new()
+                .level0_tier_compact_file_number(2)
+                .min_compaction_bytes(32)
+                .build(),
+        );
+        let picker =
+            LevelCompactionPicker::new(0, 1, config, Arc::new(RangeOverlapStrategy::default()));
+
+        let levels = vec![
+            Level {
+                level_idx: 0,
+                level_type: LevelType::Overlapping as i32,
+                table_infos: vec![
+                    generate_table(1, 1, 10, 16, 2),
+                    generate_table(2, 1, 48, 64, 2),
+                    generate_table(3, 1, 320, 330, 2),
+                ],
+                total_file_size: 0,
+            },
+            Level {
+                level_idx: 1,
+                level_type: LevelType::Nonoverlapping as i32,
+                table_infos: vec![
+                    generate_table(4, 1, 10, 50, 2),
+                    generate_table(5, 1, 64, 256, 2),
+                    generate_table(6, 1, 320, 900, 2),
+                ],
+                total_file_size: 0,
+            },
+        ];
+
+        let mut levels_handler = vec![LevelHandler::new(0), LevelHandler::new(1)];
+        let ret = picker
+            .pick_compaction(&levels, &mut levels_handler)
+            .unwrap();
+        assert_eq!(ret.select_level.table_infos.len(), 3);
+        assert_eq!(ret.select_level.table_infos.len(), 3);
+    }
 }