rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-05 21:18:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mito2): avoid chained L1 rewrites in TWCS (#8981)

Browse Source 
* fix(mito2): avoid chained L1 rewrites in TWCS

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): fall back after ineligible L0 pick

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
2026-08-31 09:57:28 +00:00
committed by GitHub
parent fb86f6573e
commit 6ce5747e02
1 changed files with 291 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/compaction/twcs.rs
+291 -6
View File
@@ -40,6 +40,9 @@ use crate::sst::version::LevelMeta;
const LEVEL_COMPACTED: Level = 1;
/// A mixed L0/L1 compaction may rewrite at most this many L1 rows per L0 row.
const MAX_L1_L0_ROW_RATIO: usize = 2;
/// Default maximum number of input SST files in one compaction input.
const DEFAULT_MAX_INPUT_FILES: usize = 32;
@@ -181,13 +184,33 @@ impl TwcsPicker {
}
}
let sorted_runs = if files_to_merge.len() < 1024 {
find_sorted_runs(&mut files_to_merge)
let (mut l0_files, l1_files): (Vec<_>, Vec<_>) = files_to_merge
.into_iter()
.partition(|file| file.level() == 0);
let num_l0_files = l0_files.len();
let num_l1_files = l1_files.len();
// Keep fresh L0 data and compacted L1 data in separate tasks whenever either
// level can trigger compaction on its own. This prevents each L0 batch from
// pulling the previous L1 output into another rewrite.
let (inputs, found_runs) = if num_l0_files >= self.trigger_file_num {
let l0_pick =
pick_candidate_files(l0_files, self.max_output_file_size, pick_count_first);
if l0_pick.0.is_empty() && num_l1_files >= self.trigger_file_num {
pick_candidate_files(l1_files, self.max_output_file_size, pick_count_first)
} else {
l0_pick
}
} else if num_l1_files >= self.trigger_file_num {
pick_candidate_files(l1_files, self.max_output_file_size, pick_count_first)
} else {
find_sorted_runs_by_time_range(&mut files_to_merge)
l0_files.extend(l1_files);
let picker = if num_l0_files > 0 && num_l1_files > 0 {
pick_mixed_count_first
} else {
pick_count_first
};
pick_candidate_files(l0_files, self.max_output_file_size, picker)
};
let found_runs = sorted_runs.len();
let inputs = pick_count_first(sorted_runs, self.max_output_file_size);
let filter_deleted = !self.append_mode
&& !window_has_overlap(files, windows)
&& !selected_overlaps_unselected(&inputs, files);
@@ -209,6 +232,20 @@ impl TwcsPicker {
}
}
fn pick_candidate_files(
mut files: Vec<FileHandle>,
max_output_file_size: Option<u64>,
picker: fn(Vec<SortedRun<FileHandle>>, Option<u64>) -> Vec<FileHandle>,
) -> (Vec<FileHandle>, usize) {
let sorted_runs = if files.len() < 1024 {
find_sorted_runs(&mut files)
} else {
find_sorted_runs_by_time_range(&mut files)
};
let found_runs = sorted_runs.len();
(picker(sorted_runs, max_output_file_size), found_runs)
}
#[derive(Debug)]
struct OrderedFile<'a> {
file: &'a FileHandle,
@@ -229,6 +266,14 @@ struct Candidate {
/// Files in the interval that overlap another interval file from a different
/// sorted run. Resolving these overlaps is what reduces sorted runs.
overlap_participants: usize,
/// Number of rows contributed by uncompacted files.
l0_rows: usize,
/// Number of rows contributed by compacted files.
l1_rows: usize,
has_l0: bool,
has_l1: bool,
/// Whether at least one file uses 0 to represent an unknown row count.
has_unknown_rows: bool,
}
impl Candidate {
@@ -245,6 +290,7 @@ impl Candidate {
let file_size = file.file.size() as usize;
self.total_size += file_size;
self.largest_file_size = self.largest_file_size.max(file_size);
self.absorb_level_rows(file.file);
let mut participates = false;
for (offset, other) in preceding.iter().enumerate() {
@@ -262,6 +308,18 @@ impl Candidate {
}
}
fn absorb_level_rows(&mut self, file: &FileHandle) {
let num_rows = file.num_rows();
self.has_unknown_rows |= num_rows == 0;
if file.level() == 0 {
self.has_l0 = true;
self.l0_rows = self.l0_rows.saturating_add(num_rows);
} else {
self.has_l1 = true;
self.l1_rows = self.l1_rows.saturating_add(num_rows);
}
}
/// Predicted number of output files after compacting this candidate. Compaction
/// output holds at most the input bytes, so a split threshold of
/// `max_output_file_size` yields at most `ceil(total_size / threshold)` files.
@@ -287,6 +345,19 @@ impl Candidate {
self.largest_file_size <= self.total_size - self.largest_file_size
}
/// Bounds rewrite amplification for mixed L0/L1 candidates. Legacy files with
/// unknown row counts retain the byte-balance behavior above.
fn has_balanced_level_rows(&self) -> bool {
!self.has_l0
|| !self.has_l1
|| self.has_unknown_rows
|| self.l1_rows <= self.l0_rows.saturating_mul(MAX_L1_L0_ROW_RATIO)
}
fn has_mixed_levels(&self) -> bool {
self.has_l0 && self.has_l1
}
/// A candidate is worth compacting only if it makes progress on at least one
/// axis: it reduces the physical file count, or it resolves at least one
/// overlap (merging sorted runs and reducing read amplification). A pure
@@ -332,6 +403,7 @@ impl CandidateScore {
///
/// - holds at least 2 files,
/// - is balanced: no single file dominates it (`largest <= sum of the others`),
/// - when mixing levels with known row counts, has at most twice as many L1 rows as L0 rows,
/// - makes progress on at least one axis: it reduces the physical file count given
/// the output split threshold `max_output_file_size`, or it resolves at least one
/// overlap between sorted runs.
@@ -340,6 +412,25 @@ impl CandidateScore {
fn pick_count_first(
sorted_runs: Vec<SortedRun<FileHandle>>,
max_output_file_size: Option<u64>,
) -> Vec<FileHandle> {
pick_count_first_where(sorted_runs, max_output_file_size, |_| true)
}
fn pick_mixed_count_first(
sorted_runs: Vec<SortedRun<FileHandle>>,
max_output_file_size: Option<u64>,
) -> Vec<FileHandle> {
pick_count_first_where(
sorted_runs,
max_output_file_size,
Candidate::has_mixed_levels,
)
}
fn pick_count_first_where(
sorted_runs: Vec<SortedRun<FileHandle>>,
max_output_file_size: Option<u64>,
is_eligible: impl Fn(&Candidate) -> bool,
) -> Vec<FileHandle> {
let files = ordered_files(&sorted_runs);
@@ -352,6 +443,8 @@ fn pick_count_first(
candidate.absorb(&files[right], &files[left..right], &mut participations);
if candidate.num_files < 2
|| !candidate.is_balanced()
|| !is_eligible(&candidate)
|| !candidate.has_balanced_level_rows()
|| !candidate.makes_progress(max_output_file_size)
{
continue;
@@ -1290,7 +1383,9 @@ mod tests {
output_level: 1,
},
ExpectedOutput {
input_files: vec![0, 1, 4],
// L1 reaches the trigger on its own, so compact it without
// rewriting the L0 tail. A chained pick can handle the tail.
input_files: vec![0, 1],
output_level: 1,
},
],
@@ -1458,6 +1553,33 @@ mod tests {
);
}
#[tokio::test]
async fn test_append_mode_can_pick_remaining_single_level_files() {
let files = [
new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 100),
new_file_handle_with_size_and_sequence(FileId::random(), 20, 29, 0, 2, 100),
new_file_handle_with_size_and_sequence(FileId::random(), 40, 49, 0, 3, 100),
new_file_handle_with_size_and_sequence(FileId::random(), 60, 69, 0, 4, 2_000),
];
let windows = assign_to_windows(files.iter(), 1);
let picker = TwcsPicker {
trigger_file_num: 4,
time_window_seconds: Some(1),
max_output_file_size: Some(1_000),
append_mode: true,
max_background_tasks: None,
time_range: None,
};
let output = picker
.build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None)
.await
.unwrap();
assert_eq!(1, output.len());
assert_eq!(3, output[0].inputs.len());
}
#[tokio::test]
async fn test_max_background_tasks_truncation() {
let file_ids = (0..10).map(|_| FileId::random()).collect::<Vec<_>>();
@@ -1978,6 +2100,27 @@ mod tests {
new_file_handle_with_size_and_sequence(FileId::random(), start, end, 0, sequence, file_size)
}
fn new_file_with_level_and_rows(
start: i64,
end: i64,
level: Level,
sequence: u64,
file_size: u64,
num_rows: u64,
) -> FileHandle {
let file = new_file_handle_with_size_and_sequence(
FileId::random(),
start,
end,
level,
sequence,
file_size,
);
let mut meta = file.meta_ref().clone();
meta.num_rows = num_rows;
FileHandle::new(meta, crate::test_util::new_noop_file_purger())
}
fn picked_ranges(files: &[FileHandle]) -> Vec<(i64, i64)> {
files
.iter()
@@ -2068,6 +2211,148 @@ mod tests {
assert!(ranges.contains(&(690, 710)));
}
#[tokio::test]
async fn test_picker_avoids_chained_l1_rewrites_by_compacting_levels_separately() {
let mut enough_l0 = (0..DEFAULT_MAX_INPUT_FILES)
.map(|idx| {
let start = idx as i64 * 20;
new_file_handle_with_size_and_sequence(
FileId::random(),
start,
start + 9,
0,
idx as u64 + 1,
10,
)
})
.collect::<Vec<_>>();
enough_l0.push(new_file_handle_with_size_and_sequence(
FileId::random(),
0,
700,
1,
100,
100,
));
let mut enough_l1 = (0..4)
.map(|idx| {
new_file_handle_with_size_and_sequence(
FileId::random(),
idx * 100,
idx * 100 + 99,
1,
idx as u64 + 1,
100,
)
})
.collect::<Vec<_>>();
enough_l1.extend((0..3).map(|idx| {
new_file_handle_with_size_and_sequence(
FileId::random(),
idx * 20,
idx * 20 + 9,
0,
idx as u64 + 10,
10,
)
}));
let picker = TwcsPicker {
trigger_file_num: 4,
time_window_seconds: Some(1),
max_output_file_size: None,
append_mode: false,
max_background_tasks: None,
time_range: None,
};
for (case, files, expected_level, expected_len) in [
("L0 reaches trigger", enough_l0, 0, DEFAULT_MAX_INPUT_FILES),
("L1 reaches trigger", enough_l1, 1, 4),
] {
let windows = assign_to_windows(files.iter(), 1);
let output = picker
.build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None)
.await
.unwrap();
assert_eq!(1, output.len(), "{case}");
assert_eq!(expected_len, output[0].inputs.len(), "{case}");
assert!(
output[0]
.inputs
.iter()
.all(|file| file.level() == expected_level),
"{case}"
);
}
}
#[tokio::test]
async fn test_picker_falls_back_to_l1_when_triggered_l0_cannot_make_progress() {
let mut files = (0..4)
.map(|idx| {
let start = idx * 20;
new_file_handle_with_size_and_sequence(
FileId::random(),
start,
start + 9,
0,
idx as u64 + 1,
600,
)
})
.collect::<Vec<_>>();
files.extend((0..4).map(|idx| {
let start = idx * 20 + 100;
new_file_handle_with_size_and_sequence(
FileId::random(),
start,
start + 9,
1,
idx as u64 + 10,
100,
)
}));
let windows = assign_to_windows(files.iter(), 1);
let picker = TwcsPicker {
trigger_file_num: 4,
time_window_seconds: Some(1),
max_output_file_size: Some(512),
append_mode: false,
max_background_tasks: None,
time_range: None,
};
let output = picker
.build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None)
.await
.unwrap();
assert_eq!(1, output.len());
assert_eq!(4, output[0].inputs.len());
assert!(output[0].inputs.iter().all(|file| file.level() == 1));
}
#[test]
fn test_mixed_candidate_row_balance() {
for (case, l1_rows, l0_rows, expected_len) in [
("L1 rows dominate", 1_000_000, 10_000, 0),
("L1 rows meet ratio", 60_000, 10_000, 4),
("L0 rows are unknown", 1_000_000, 0, 4),
] {
let files = vec![
new_file_with_level_and_rows(0, 99, 1, 1, 100, l1_rows),
new_file_with_level_and_rows(0, 9, 0, 2, 100, l0_rows),
new_file_with_level_and_rows(20, 29, 0, 3, 100, l0_rows),
new_file_with_level_and_rows(40, 49, 0, 4, 100, l0_rows),
];
let picked = pick_mixed_count_first(vec![SortedRun::from(files)], None);
assert_eq!(expected_len, picked.len(), "{case}");
}
}
#[test]
fn test_count_first_prefers_smaller_bytes_when_file_counts_match() {
let files = (0..=DEFAULT_MAX_INPUT_FILES)