Use Vec::resize and Vec::truncate to update levels in week 2

[ppdogg]

In compact.rs of week2 day1 commit. You use HashSet to update state.l0_sstables in function force_full_compaction:

let mut l0_sstables_map = l0_sstables.iter().copied().collect::<HashSet<_>>();
state.l0_sstables = state
                    .l0_sstables
                    .iter()
                    .filter(|x| !l0_sstables_map.remove(x))
                    .copied()
                    .collect::<Vec<_>>();

Actually, Vec::resize can be used to update state.l0_sstables. Because we always choose consecutive sst_ids from l0_sstables array:

snapshot
      .l0_sstables
      .resize(snapshot.l0_sstables.len() - l0_sstables.len(), 0);

[ppdogg]

The situation in tiered.rs of week2 day3 is similar. In function apply_compaction_result:

let mut tier_to_remove = task
     .tiers
     .iter()
     .map(|(x, y)| (*x, y))
     .collect::<HashMap<_, _>>();

We always choose consecutive tiers. So we can retain the front, truncate the middle and retain the tail:

        // levels(tiers):
        // [0 ... i ... i + tiers.len() ..]
        //        |          |
        // task_tier_start task_tier_end
        
        let mut i = 0;
        for (idx, (level, _)) in snapshot.levels.iter().enumerate() {
            if *level == task.tiers[0].0 {
                i = idx;
                break;
            }
        }
        // remove previous tiers
        new_snapshot.levels.truncate(i);
        // add new tiers
        new_snapshot.levels.push((output[0], Vec::from(output)));
        // add remain tiers
        if !task.bottom_tier_included {
            new_snapshot
                .levels
                .extend_from_slice(&snapshot.levels[i + task.tiers.len()..]);
        }

[ppdogg]

Also, in simple_leveled.rs from week2 day2. You directly clear l0_sstables in apply_compaction_result:

snapshot.l0_sstables.clear()

I doubt this. Because the snapshot from when we generate task and the snapshot from when we apply result is different. The user might flush new sstables to l0_sstables in the period. Maybe use this:

new_snapshot.l0_sstables.resize(
      new_snapshot.l0_sstables.len() - task.upper_level_sst_ids.len(),
      0,
);

[ppdogg]

Also, in simple_leveled.rs from week2 day2. You directly clear l0_sstables in apply_compaction_result:

snapshot.l0_sstables.clear()

I doubt this. Because the snapshot from when we generate task and the snapshot from when we apply result is different. The user might flush new sstables to l0_sstables in the period. Maybe use this:

new_snapshot.l0_sstables.resize(
      new_snapshot.l0_sstables.len() - task.upper_level_sst_ids.len(),
      0,
);

Oh, you have fixed this.

[ppdogg]

And in level.rs:

    pub fn apply_compaction_result(
        &self,
        snapshot: &LsmStorageState,
        task: &LeveledCompactionTask,
        output: &[usize],
        in_recovery: bool,
    ) -> (LsmStorageState, Vec<usize>) {
        let mut new_snapshot = snapshot.clone();

        if let Some(upper_level) = task.upper_level {
            let mut i = 0;
            for (idx, sst_id) in snapshot.levels[upper_level].1.iter().enumerate() {
                if *sst_id == task.upper_level_sst_ids[0] {
                    i = idx;
                    break;
                }
            }
            // remove the sstable from upper level
            let _ = new_snapshot.levels[upper_level].1.remove(i);
        } else {
            // truncate from l0_sstables
            new_snapshot.l0_sstables.resize(
                snapshot.l0_sstables.len() - task.upper_level_sst_ids.len(),
                0,
            );
        }

        // remove old sstables from lower level
        if !task.lower_level_sst_ids.is_empty() {
            let mut i = 0;
            for (idx, sst_id) in snapshot.levels[task.lower_level].1.iter().enumerate() {
                if *sst_id == task.lower_level_sst_ids[0] {
                    i = idx;
                    break;
                }
            }
            // levels:
            // [0 ... i ... i + sst_ids.len() ..]
            //        |            |
            //    task_start    task_end
            new_snapshot.levels[task.lower_level].1.truncate(i);
            new_snapshot.levels[task.lower_level]
                .1
                .extend(&snapshot.levels[task.lower_level].1[i + task.lower_level_sst_ids.len()..]);
        }
        // add new sstables to lower level
        new_snapshot.levels[task.lower_level].1.extend(output);
        if !in_recovery {
            new_snapshot.levels[task.lower_level].1.sort_by(|x, y| {
                snapshot
                    .sstables
                    .get(x)
                    .unwrap()
                    .first_key()
                    .cmp(snapshot.sstables.get(y).unwrap().first_key())
            });
        }

        let mut old_sst_ids = task.upper_level_sst_ids.clone();
        old_sst_ids.extend_from_slice(task.lower_level_sst_ids.as_slice());

        (new_snapshot, old_sst_ids)
    }