Strategized Compaction Plan

[zhangyue19921010]

Background

Currently, Lance allows users to manually trigger a standalone job to perform Compaction, to achieve the following:

1. Merging small files.
2. Materializing deletions.

However, in practice, there are some limitations:

1. The current Compaction plan phase only supports simple grouping strategies(as shown below), which are not sufficiently comprehensive. Adding new grouping logic is tightly coupled with existing logic.
   • Grouping based on target_rows_per_fragment.
   • Grouping based on whether a fragment contains deleted rows.
2. The current Compaction is a full compaction, with no limit on the total target I/O. This leads to high pressure on the Compaction Job and makes it difficult to guarantee efficiency.

Design

New CompactionPlanner Trait

#[async_trait::async_trait]
pub trait CompactionPlanner: Send + Sync {
    // get all fragments by default
    fn get_fragments(&self, dataset: &Dataset, _options: &CompactionOptions) -> Vec<FileFragment> {
        // get_fragments should be returning fragments in sorted order (by id)
        // and fragment ids should be unique
        dataset.get_fragments()
    }

    // no filter by default
    async fn filter_fragments(
        &self,
        _dataset: &Dataset,
        fragments: Vec<FileFragment>,
        _options: &CompactionOptions,
    ) -> Result<Vec<FileFragment>> {
        Ok(fragments)
    }

    async fn plan(&self, dataset: &Dataset, options: &CompactionOptions) -> Result<CompactionPlan>;
}

A new CompactionPlanner trait is introduced with a core plan method. This allows users to define custom compaction plan strategies, offering greater flexibility and decoupling the logic of different strategies.

Default Implementation

#[derive(Debug, Clone, Default)]
pub struct DefaultCompactionPlanner;

#[async_trait::async_trait]
impl CompactionPlanner for DefaultCompactionPlanner {
    async fn plan(
        &self,
        dataset: &Dataset,
        options: &CompactionOptions,
    ) -> Result<CompactionPlan> {
        // ..... Same as the current implementation
    }
}

Provide a default implementation to ensure that the current behavior remains identical to the historical behavior.

Landing Steps

1. Complete the foundational work for the Strategized Compaction Plan and provide a default implementation to ensure that the current behavior remains identical to the historical behavior.
2. Introduce new compaction plan strategies from different perspectives, for example:
   • IO-bounded Compaction Plan: Limits the total I/O of each compaction to control the load of a single merge job.
   • Deletion-Only Compaction Plan: Only merges fragments with deleted rows.
   • Incremental Compaction Plan: Only performs compaction on newly changed fragments.
   • etc.

Core API Compatibility

Two core APIs have been retained, and default implementations are provided within them to ensure API compatibility.

compact_files

pub async fn compact_files(
    dataset: &mut Dataset,
    options: CompactionOptions,
    remap_options: Option<Arc<dyn IndexRemapperOptions>>, // These will be deprecated later
) -> Result<CompactionMetrics> {
    let planner = DefaultCompactionPlanner;
    compact_files_with_planner(dataset, options, remap_options, &planner).await
}

plan_compaction

pub async fn plan_compaction(
    dataset: &Dataset,
    options: &CompactionOptions,
) -> Result<CompactionPlan> {
    let planner = DefaultCompactionPlanner;
    planner.plan(dataset, options).await
}

[wjones127]

I had some ideas earlier on compaction planning I wrote here: #4485

We should come up with a planner that is able to calculate these tradeoffs:

• What is the write amplification cost, including index remapping?
• What is the benefit to scan performance (high if many tiny files, low if files are sizeable)?
• What is the benefit to metadata overhead?

I think a good planner needs to balance those costs / benefits.

IO-bounded Compaction Plan: Limits the total I/O of each compaction to control the load of a single merge job.

I don't understand this part. When you say IO, are you talking about the rate of read and write? Or the total volume?

If you are talking about the rate, that seems like an execution concern, not a planning one. The executor could just tell the scheduler to rate limit.

If you are talking about total volume, I think what you are saying the same thing I was when I talked about write amplification in the other issue I linked.

Introduce new compaction plan strategies from different perspectives, for example:

• IO-bounded Compaction Plan: Limits the total I/O of each compaction to control the load of a single merge job.
• Deletion-Only Compaction Plan: Only merges fragments with deleted rows.
• Incremental Compaction Plan: Only performs compaction on newly changed fragments.
• etc.

Why do we need different planners rather than just parameters in CompactionOptions? I'm open to the idea, just want to make sure we have good reason to turn this into a trait.

[zhangyue19921010]

Thanks for your attention! @wjones127 . First, let me explain a few of your questions:

I had some ideas earlier on compaction planning I wrote here: #4485

Nice catch! I completely agree that a good compaction plan strategy needs to balance write amplification, read amplification, and metadata amplification.

Also the two are related, but they may not solve the same issue. Currently, the main goal is to address the issue of the flexibility of the Compaction plan through strategization. #4485 addresses the issue of the rationality of the general planner.

I don't understand this part. When you say IO, are you talking about the rate of read and write? Or the total volume?

Here, IO-bounded refers to the total volume that needs to be processed by the current compaction plan, such as 100GB, 1TB, etc.

Why do we need different planners rather than just parameters in CompactionOptions? I'm open to the idea, just want to make sure we have a good reason to turn this into a trait.

The CompactionOptions are necessary in all scenarios, even when we design different planners, because the behavior of different planners still needs to be controlled by reasonable parameters.

As for why we intend to design a planner trait to support defining different types of planners, the reason lies in the fact that the generation logic of the Compaction plan can vary drastically depending on scenario requirements. Relying solely on parameters is insufficient to cover all strategies.

For instance, in some scenarios, only the fragments containing deletes need to be compacted; in other cases, focus is solely on the most recently written fragments. If all such logics were integrated into a single planner, it would lead to severe code coupling and mutual interference between different functions.

In my view, a reasonable design would proceed in two Independent steps:

Step 1: Provide a Compaction planner Trait that includes three abstract methods: get_fragments, filter_fragments, and plan. Different strategies can implement their own methods at different stages as needed. For example:

• get_fragments:
  • By default, it iterates through all fragments.
  • For the incremental Compaction plan strategy, it only retrieves newly added or modified fragments during the get_fragments process.
• filter_fragments:
  • By default, it filters fragments based on their size.
  • However, under the Deletion-Only Compaction Plan strategy, it only focuses on fragments containing deleted rows.
• plan:
  • We can include all eligible fragments into the same compaction job.
  • For the IO-Boundary compaction plan strategy, we can limit the maximum value of a data volume, such as 100GB, thereby making the workload of a single compaction job more controllable.

PR #5187 is an attempt to implement Step 1 while ensuring full historical compatibility(After we reach a consensus on the design, there will be some changes.).

Step 2: Of course, we need to provide a universal default compaction plan strategy, which fully addresses issues related to amplification cost, benefit to scan performance, and benefit to metadata overhead.

Mainly consider three aspects: write amplification, read amplification, and metadata amplification. Perhaps https://github.com/facebook/rocksdb/wiki/Universal-Compaction is a good start, although this compaction plan is based on LSM design.

[zhangyue19921010]

Regarding compaction strategy optimization, a feasible optimization solution is hierarchical merging. In general, first plan Compaction at the Column level (Horizontal perspective), then plan Compaction at the Row level (Vertical perspective).

Horizontal (Column Perspective)

At the Fragment granularity, determine whether the current Fragment participates in compaction. If
$$(total\_file\_size - dropped\_column\_size)/ total\_file\_size < 70\%$$
the write amplification of the current fragment is considered less than 70%, and it can participate in Compaction.

Vertical (Row Perspective)

Similar to LSM-Tree, fragments are divided into different groups based on size, and different groups adopt different merging thresholds.
First, define the fragment write amplification rate to describe the write amplification degree of a single fragment: $$(total\_rows - deleted\_rows)/total\_rows$$

1. Bronze Tier: num_rows < 500,000. Fast merging. Merge when $$files\_number &gt; crystal\_file\_threshold$$. The fragment write amplification rate is not considered.
2. Silver Tier: num_rows between 500,000 - 1,000,000. Moderate merging. Merge when $$files\_number &gt; crystal\_file\_threshold * 10$$ OR the write amplification rate of a single fragment is less than 60%.
3. Gold Tier: num_rows > 1,000,000. Conservative merging. Merge when $$files\_number &gt; crystal\_file\_threshold * 100$$ AND the write amplification rate of a single fragment is less than 40%.

Expected Effects:

• Rapid compaction of small files.
• Avoid unnecessary rewriting of large files.

Reasonable control of small files also brings benefits in scan performance and metadata overhead.

How do you think about it @wjones127 ? Looking forward to your reply ：）

[wjones127]

I think both of those compaction strategy ideas (the horizontal and the tiered vertical compaction) sound good to me. I had started thinking about the tiered compaction a long time ago but never finished implementing it.

[zhangyue19921010]

Thanks for your reply @wjones127 :) I will split this task into 2 steps.

• Firstly, introduce the planner trait to make the planner pluggable, enabling customized optimization for different scenarios
• Secondly, optimize the default implementation based on the horizontal and tiered vertical strategies

Please let me know if you have anything you'd like to discuss ~