Redesigning the manifest deduplication mechanism.

[zealchen]

Describe This Problem

The current manifest snapshot archiving process requires deduplication during every merge operation involving a single snapshot file and multiple SST files. Let's first look at the snapshot archiving procedure:

The duplication scenario can only exist when the manifest merger reads an old metafile that has already gone through the merging procedure once. This inconsistent problem is due to the failure to delete the meta files after the persistence of the snapshot file. The failure itself can lie in many reasons like network problem or crashing of the running process.

This issue is an idempotency problem, or in other words, the merge, write, and delete operations must be performed within a single transaction.

Proposal

We could design a batch operation transaction mechanism to ensure that:

1. If the last merge procedure was successfully completed, there is no need to perform deduplication.
2. If the last merge procedure was partially completed, deduplication is required.
3. If it is unclear whether the last merge procedure was successful or not, deduplication should be performed.

To determine which scenario applies, we need to record the states of the merge operations and store them locally. This allows us to handle the scenarios as follows:
a. If the state file does not exist, proceed to scenario 3.
b. If the state file indicates that the delete operation is complete, proceed to scenario 1.
c. If the state file indicates that the snapshot was successfully stored but the delete operation is incomplete, proceed to scenario 2.

The pseudo code is like below:

fn do_merge() {
      let sstfiles = ...;
      let snapshot = ...;
      let ops = MergeOps::new(snapshot, sstfiles);
     {
          let ops_guard = BatchOpsGuard::transaction(ops);
          ops_guard.execute(|| batch_ops.merge_sst_meta());
          ops_guard.execute(|| batch_ops.store_snapshot());
          ops_guard.execute(|| batch_ops.delete_sst_meta());
      }
}

trait BatchOps {
    type State;
    fn need_redo(state: State) -> bool;
    fn redo(state: State) -> Result<()>;
    fn start() -> State;
    fn finish() -> State;
}

impl BatchOps for MergeOps {
    type State = MergeState;

    fn need_redo(state: State) -> bool
    {
         // check the states to determine if to redo
    }
    fn redo(state: State) -> Result<()>
    {
        // just do dedup
    }
    fn start() -> State;
    fn finish() -> State;

     fn merge_sst_meta(..) -> State;
     fn store_snapshot(..) -> State;
     fn delete_sst_meta(..) -> State;
}

impl BatchOpsGuard {
   fn transaction(ops) -> Self
   where ops: BatchOps;
   {
       let state = read_file(..);
        if ops.need_redo(state) {
            ops.redo(state);
        }
        let state = ops.start();
        save_file(state);
        Self { ops }
   }

    fn execute(&self, fn: F) 
    where F: Fn -> BatchOps::State
    {
        let state = fn();
        save_file(state);
    }
}

impl Drop for BatchOpsGuard {
    fn drop(&mut self) {
        let state = self.ops.finish();
        save_file(state);
    }
}

Additional Context

No response