diff --git a/src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs b/src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs
index d615c4e6c4192..bbaa65bf16e35 100644
--- a/src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs
+++ b/src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs
@@ -73,6 +73,7 @@ impl CompactionPicker for LevelCompactionPicker {
         } else {
             vec![]
         };
+        println!("============partition_sub_levels============");
         if let Some(ret) =
             self.pick_multi_level_to_base(levels, level_handlers, partitions.clone(), stats)
         {
@@ -251,6 +252,7 @@ impl LevelCompactionPicker {
             stats.skip_by_pending_files += 1;
             return None;
         }
+        println!("============pick to base level============");
         let mut input_levels = vec![];
         let mut total_file_size = 0;
         let mut total_file_count = 0;
@@ -258,19 +260,18 @@ impl LevelCompactionPicker {
             self.config.max_compaction_bytes,
             target_level.total_file_size,
         );
-        let mut wait_enough = false;
+        let mut wait_enough = true;
         for level in &l0.sub_levels {
             if level.level_type() != LevelType::Nonoverlapping
                 || level_handlers[0].is_level_pending_compact(level)
             {
                 stats.skip_by_pending_files += 1;
+                wait_enough = false;
                 break;
             }
 
-            if total_file_size > max_compaction_size
-                || total_file_count > self.config.level0_max_compact_file_number
+            if total_file_size > max_compaction_size || total_file_count > self.config.level0_max_compact_file_number
             {
-                wait_enough = true;
                 break;
             }
             input_levels.push(InputLevel {
@@ -404,7 +405,7 @@ pub mod tests {
         let config = Arc::new(
             CompactionConfigBuilder::new()
                 .level0_tier_compact_file_number(2)
-                .level0_sub_level_compact_level_count(1)
+                .level0_sub_level_compact_level_count(2)
                 .build(),
         );
         LevelCompactionPicker::new(1, config)
@@ -443,22 +444,14 @@ pub mod tests {
         let ret = picker
             .pick_compaction(&levels, &levels_handler, &mut local_stats)
             .unwrap();
-        assert_eq!(ret.input_levels[0].table_infos.len(), 1);
-        assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 4);
-        assert_eq!(ret.input_levels[1].table_infos[0].get_sst_id(), 1);
+        println!("target level: {}. ret.input_levels: {}",ret.target_level, ret.input_levels.len());
+        assert_eq!(ret.input_levels[0].table_infos.len(), 2);
+        assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 5);
+        assert_eq!(ret.input_levels[1].table_infos[0].get_sst_id(), 3);
 
         ret.add_pending_task(0, &mut levels_handler);
-        {
-            push_table_level0_nonoverlapping(&mut levels, generate_table(6, 1, 100, 200, 2));
-            push_table_level0_nonoverlapping(&mut levels, generate_table(7, 1, 301, 333, 4));
-            let ret2 = picker
-                .pick_compaction(&levels, &levels_handler, &mut local_stats)
-                .unwrap();
-
-            assert_eq!(ret2.input_levels[0].table_infos.len(), 1);
-            assert_eq!(ret2.input_levels[0].table_infos[0].get_sst_id(), 6);
-            assert_eq!(ret2.input_levels[1].table_infos[0].get_sst_id(), 5);
-        }
+        push_table_level0_nonoverlapping(&mut levels, generate_table(6, 1, 100, 200, 2));
+        push_table_level0_nonoverlapping(&mut levels, generate_table(7, 1, 301, 333, 4));
 
         levels.l0.as_mut().unwrap().sub_levels[0]
             .table_infos
@@ -471,12 +464,12 @@ pub mod tests {
         let ret = picker
             .pick_compaction(&levels, &levels_handler, &mut local_stats)
             .unwrap();
-        assert_eq!(ret.input_levels.len(), 3);
-        assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 6);
-        assert_eq!(ret.input_levels[1].table_infos[0].get_sst_id(), 5);
-        assert_eq!(ret.input_levels[2].table_infos.len(), 2);
-        assert_eq!(ret.input_levels[2].table_infos[0].get_sst_id(), 3);
-        assert_eq!(ret.input_levels[2].table_infos[1].get_sst_id(), 2);
+        println!("target level: {}", ret.target_level);
+        assert_eq!(ret.input_levels.len(), 2);
+        assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 5);
+        assert_eq!(ret.input_levels[1].table_infos.len(), 2);
+        assert_eq!(ret.input_levels[1].table_infos[0].get_sst_id(), 3);
+        assert_eq!(ret.input_levels[1].table_infos[1].get_sst_id(), 2);
         ret.add_pending_task(1, &mut levels_handler);
 
         let mut local_stats = LocalPickerStatistic::default();
diff --git a/src/meta/src/hummock/compaction/picker/intral_sub_level_picker.rs b/src/meta/src/hummock/compaction/picker/intral_sub_level_picker.rs
index 3286a2d641ef4..0329c4ebda918 100644
--- a/src/meta/src/hummock/compaction/picker/intral_sub_level_picker.rs
+++ b/src/meta/src/hummock/compaction/picker/intral_sub_level_picker.rs
@@ -69,10 +69,11 @@ impl CompactionPicker for IntraSubLevelPicker {
             })
             .min()
             .unwrap_or(0);
+        println!("max_sub_level_id: {}: ", max_sub_level_id);
 
         for (idx, level) in l0.sub_levels.iter().enumerate() {
             if level.level_type() != LevelType::Nonoverlapping
-                || level_handlers[0].is_level_all_pending_compact(level)
+                || level_handlers[0].is_level_pending_compact(level)
             {
                 continue;
             }
@@ -115,6 +116,7 @@ impl CompactionPicker for IntraSubLevelPicker {
 
                 compaction_bytes += next_level.total_file_size;
                 compaction_file_count += next_level.table_infos.len() as u64;
+
                 input_levels.push(InputLevel {
                     level_idx: 0,
                     level_type: next_level.level_type,
@@ -128,6 +130,8 @@ impl CompactionPicker for IntraSubLevelPicker {
             {
                 continue;
             }
+            println!("level0_sub_level_compact_level_count: {}. compaction_bytes: {}, max_compaction_bytes: {}",
+                     self.config.level0_sub_level_compact_level_count, compaction_bytes, max_compaction_bytes);
             input_levels.reverse();
 
             return Some(CompactionInput {