refactor: add Compactor trait to abstract the compaction

[zyy17]

I hereby agree to the terms of the GreptimeDB CLA.

Refer to a related PR or issue link (optional)

What's changed and what's your intention?

The main purpose of the PR is to create a small module to encapsulate the logic of compaction without initializing the whole Mito engine.

It's very useful for creating the cli compact tool and making compaction a service.

For the compact tool that I'm working based on the PR, we can use the Compactor as the following code(greptime cli compact --region-id 4398046511104) :

...
#[async_trait]
impl Tool for Compact {
    async fn do_work(&self) -> Result<()> {
        let object_store_manager = self.create_object_store_manager();
        let compactor = DefaultCompactor {};
        let compaction_region =
            open_compaction_region(&self.request, self.data_home.as_str(), object_store_manager)
                .await
                .context(OpenCompactRegionSnafu)?;

        compactor
            .compact(compaction_region, self.request.compaction_options.clone())
            .await
            .context(CompactRegionSnafu)?;

        Ok(())
    }
}
...

Add the Compactor trait

The new file src/mito2/src/compaction/compactor.rs creates the new trait Compactor to encapsulate the compaction. The trait will split the compaction into multiple steps:

• merge_ssts()
• update_manifest()
• compact(): The default implementation will combine the merge_ssts() and update_manifest().

The implementation of merge_ssts() and update_manifests() are from the original handle_compaction().

The PR also provides the DefaultCompactor, which is the default implementation of Compactor. The original compaction will be replaced by DefaultCompactor.

Refactor the Picker trait

To better collaborate with Compactor, the Picker output PickerOutput instead of creating the internal compaction task:

pub trait Picker: Debug + Send + Sync + 'static {
    fn pick(&self, current_version: VersionRef) -> Option<PickerOutput>;
}

Add open_compaction_region()

It's heavy to use RegionOpener::open() to open the region, so I decided to create a lightweight version: open_compaction_region(). The function will accept CompactionRequest and other necessary arguments and return CompactionRegion, which is a subset of MitoRegion.

Checklist

• I have written the necessary rustdoc comments.
• I have added the necessary unit tests and integration tests.
• This PR requires documentation updates.

[zyy17]

@evenyag I'm unsure if it's appropriate to place the open_compaction_region()(maybe use another name) in RegionOpener? I try it before but finally still use current version in compactor.rs.

[codecov]

Codecov Report

Attention: Patch coverage is 60.00000% with 158 lines in your changes missing coverage. Please review.

Project coverage is 84.60%. Comparing base (2faa6d6) to head (f8d88a2).
Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4097      +/-   ##
==========================================
- Coverage   84.94%   84.60%   -0.34%     
==========================================
  Files        1018     1020       +2     
  Lines      178688   178846     +158     
==========================================
- Hits       151780   151310     -470     
- Misses      26908    27536     +628     

[killme2008]

@v0y4g3r Please take a look

[zyy17]

Note: I add some new important refactors beyond the code review comments:

1. Move build_compaction_task() in CompactionScheduler;
2. Use CompactionRegion as an argument of Picker;
3. Put CompactionRegion into CompactionTaskImpl and make it simple;

[killme2008]

Looks GTM. It's better that let @v0y4g3r take a look.

[v0y4g3r]

Regarding the interaction between database instance and compactor, it's better to put the pick and manifest update work in database instance, and the compactor focuses on merging SST files.

[zyy17]

@v0y4g3r @evenyag the latest version syncs the main branch, PTAL.

[evenyag]

LGTM

[zyy17]

Regarding the interaction between database instance and compactor, it's better to put the pick and manifest update work in database instance, and the compactor focuses on merging SST files.

I add the PickerOutput in CompactorRequest in the latest version.

For the real compactor service, the logic can be:

for rece_compactor_request().await {
    let compaction_region = open_compaction_region(compactor_request, ...);
    let merge_output = DefaultCompactor{}.merge_ssts(compaction_region, compactor_request);
    send_to_dn(merge_output);
}

When the datanode receives the response, it can use the merge_output and call update_manifest().

@v0y4g3r PTAL

[v0y4g3r]

LGTM. Maybe there's no need to specifically add a send_to_dn method. The compactor just finish merging sst files and report the results and datanode can poll this result in its RemoteJobScheduler.