fix(compaction): correct level0_file_num_compaction_trigger in simple leveled compaction

Signed-off-by: zztaki <zztaki@outlook.com>

Author: zztaki

mini-lsm/src/compact/simple_leveled.rs

@@ -37,19 +37,35 @@ impl SimpleLeveledCompactionController {
         &self,
         snapshot: &LsmStorageState,
     ) -> Option<SimpleLeveledCompactionTask> {
+        if self.options.max_levels == 0 {
+            return None;
+        }
+
         let mut level_sizes = Vec::new();
         level_sizes.push(snapshot.l0_sstables.len());
         for (_, files) in &snapshot.levels {
             level_sizes.push(files.len());
         }
 
-        for i in 0..self.options.max_levels {
-            if i == 0
-                && snapshot.l0_sstables.len() < self.options.level0_file_num_compaction_trigger
-            {
-                continue;
+        // check level0_file_num_compaction_trigger for compaction of L0 and L1
+        {
+            if snapshot.l0_sstables.len() >= self.options.level0_file_num_compaction_trigger {
+                println!(
+                    "compaction triggered at level 0 and 1 because L0 has {} SSTs",
+                    snapshot.l0_sstables.len()
+                );
+                return Some(SimpleLeveledCompactionTask {
+                    upper_level: None,
+                    upper_level_sst_ids: snapshot.l0_sstables.clone(),
+                    lower_level: 1,
+                    lower_level_sst_ids: snapshot.levels[0].1.clone(),
+                    is_lower_level_bottom_level: false,
+                });
             }
+        }
 
+        // check size_ratio_percent for compaction of other levels (>= L1)
+        for i in 1..self.options.max_levels {
             let lower_level = i + 1;
             let size_ratio = level_sizes[lower_level] as f64 / level_sizes[i] as f64;
             if size_ratio < self.options.size_ratio_percent as f64 / 100.0 {
@@ -58,12 +74,8 @@ impl SimpleLeveledCompactionController {
                     i, lower_level, size_ratio
                 );
                 return Some(SimpleLeveledCompactionTask {
-                    upper_level: if i == 0 { None } else { Some(i) },
-                    upper_level_sst_ids: if i == 0 {
-                        snapshot.l0_sstables.clone()
-                    } else {
-                        snapshot.levels[i - 1].1.clone()
-                    },
+                    upper_level: Some(i),
+                    upper_level_sst_ids: snapshot.levels[i - 1].1.clone(),
                     lower_level,
                     lower_level_sst_ids: snapshot.levels[lower_level - 1].1.clone(),
                     is_lower_level_bottom_level: lower_level == self.options.max_levels,