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

fix(mito2): keep deletion markers when compacting part of a window (#8872)

Browse Source 
`TwcsPicker::find_inputs` decides `filter_deleted` from the shape of the whole
time window, but the compaction inputs are only a subset of it: `reduce_runs`
and `merge_seq_files` narrow the selection down and the max input file num limit
narrows it further. When a deletion marker lands in the compacted set while the
file holding the row it masks stays behind, the marker is dropped from the
output and the old row becomes visible again.

Re-check the final selection against the rest of the window and stop filtering
deleted rows whenever something left behind still overlaps the inputs. The check
compares ranges inclusively, so it also covers files that share only a boundary
timestamp: run detection treats those as non-overlapping, which is how a single
timestamp delete file ends up in the same run as the file it deletes rows from.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
This commit is contained in:
jeremyhi
2026-08-13 14:17:09 +00:00
committed by GitHub
parent 7539e60139
commit 3c80df043a
2 changed files with 265 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/compaction/twcs.rs
+116 -1
View File
@@ -169,7 +169,7 @@ impl TwcsPicker {
let found_runs = sorted_runs.len();
// We only remove deletion markers if we found less than 2 runs and not in append mode.
// because after compaction there will be no overlapping files.
let filter_deleted =
let mut filter_deleted =
found_runs <= 2 && !self.append_mode && !window_has_overlap(files, windows);
if found_runs == 0 {
return (vec![], filter_deleted);
@@ -210,6 +210,14 @@ impl TwcsPicker {
);
}
// The inputs are only a subset of the window: `reduce_runs` and `merge_seq_files` both
// narrow the selection down and the file num limit above narrows it further. A deletion
// marker may only be dropped when every file that can hold a row it masks is compacted
// along with it, so re-check the final selection against what stays in the window.
if filter_deleted && overlaps_files_left_behind(&inputs, &files_to_merge) {
filter_deleted = false;
}
if inputs.len() > 1 {
// If we have more than one group to compact.
log_pick_result(
@@ -227,6 +235,20 @@ impl TwcsPicker {
}
}
/// Returns whether any file group of `window_files` that is not part of `inputs` overlaps
/// `inputs`. Such a group keeps rows a deletion marker in `inputs` masks, so the compaction
/// must not filter deleted rows out.
///
/// Overlapping is inclusive on both boundaries here: two files that only share a boundary
/// timestamp can still hold the same row.
fn overlaps_files_left_behind(inputs: &[FileGroup], window_files: &[FileGroup]) -> bool {
let picked: HashSet<_> = inputs.iter().flat_map(|fg| fg.file_ids()).collect();
window_files
.iter()
.filter(|fg| !fg.file_ids().iter().any(|id| picked.contains(id)))
.any(|fg| inputs.iter().any(|input| input.overlap_inclusive(fg)))
}
#[allow(clippy::too_many_arguments)]
fn log_pick_result(
region_id: RegionId,
@@ -1456,6 +1478,99 @@ mod tests {
assert_eq!(output[0].inputs.len(), 32);
}
#[tokio::test]
async fn test_limit_max_input_files_keeps_deletion_markers() {
common_telemetry::init_default_ut_logging();
// One large file group spanning the whole window plus 32 small ones nested inside
// it. That is exactly 2 runs, so the window on its own allows filtering deletions.
let mut files = vec![new_file_handle_with_size_and_sequence(
FileId::random(),
0,
3_000_000,
0,
1,
1024 * 1024 * 1024,
)];
files.extend((0..32).map(|idx: i64| {
new_file_handle_with_size_and_sequence(
FileId::random(),
(idx + 1) * 10_000,
(idx + 1) * 10_000 + 1_000,
0,
(idx + 2) as u64,
1024,
)
}));
let windows = assign_to_windows(files.iter(), 3600);
let picker = TwcsPicker {
trigger_file_num: 4,
time_window_seconds: Some(3600),
max_output_file_size: None,
append_mode: false,
max_background_tasks: None,
time_range: None,
};
let active_window = find_latest_window_in_seconds(files.iter(), 3600);
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
.await
.unwrap();
assert_eq!(1, output.len());
// The input file num limit picks the smallest groups first, so the large group is
// left behind while the small groups that overlap it are compacted.
assert_eq!(32, output[0].inputs.len());
assert!(
!output[0].filter_deleted,
"deletion markers must be kept once the file num limit drops files they may mask"
);
}
#[tokio::test]
async fn test_limit_max_input_files_still_filters_without_overlap() {
common_telemetry::init_default_ut_logging();
// 40 file groups with disjoint time ranges, i.e. a single run. Nothing the file num
// limit leaves behind can hold a row masked by a deletion marker we compact.
let files: Vec<_> = (0..40i64)
.map(|idx| {
new_file_handle_with_size_and_sequence(
FileId::random(),
(idx + 1) * 10_000,
(idx + 1) * 10_000 + 1_000,
0,
(idx + 1) as u64,
1024,
)
})
.collect();
let windows = assign_to_windows(files.iter(), 3600);
let picker = TwcsPicker {
trigger_file_num: 4,
time_window_seconds: Some(3600),
max_output_file_size: Some(1024 * 1024 * 1024),
append_mode: false,
max_background_tasks: None,
time_range: None,
};
let active_window = find_latest_window_in_seconds(files.iter(), 3600);
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
.await
.unwrap();
assert_eq!(1, output.len());
assert_eq!(32, output[0].inputs.len());
assert!(output[0].filter_deleted);
}
#[tokio::test]
async fn test_newer_windows_have_priority() {
let older_file_ids = [FileId::random(), FileId::random()];
src/mito2/src/engine/compaction_test.rs
+149
View File
@@ -1342,6 +1342,155 @@ async fn test_compaction_region_with_overlapping_delete_all_with_format(flat_for
assert!(vec.is_empty());
}
#[tokio::test]
async fn test_compaction_input_limit_keeps_rows_deleted() {
test_compaction_input_limit_keeps_rows_deleted_with_format(false).await;
test_compaction_input_limit_keeps_rows_deleted_with_format(true).await;
}
/// Creates a region that only compacts when asked, so that a test can build an exact file
/// layout with `put_and_flush` and `delete_and_flush`.
async fn env_for_manual_compaction(
env: &mut TestEnv,
region_id: RegionId,
flat_format: bool,
) -> (MitoEngine, Vec<ColumnSchema>) {
let engine = env
.create_engine(MitoConfig {
default_flat_format: flat_format,
min_compaction_interval: Duration::from_secs(3600),
..Default::default()
})
.await;
env.get_schema_metadata_manager()
.register_region_table_info(
region_id.table_id(),
"test_table",
"test_catalog",
"test_schema",
None,
env.get_kv_backend(),
)
.await;
let request = CreateRequestBuilder::new()
.insert_option("compaction.type", "twcs")
.insert_option("compaction.twcs.time_window", "1h")
.build();
let column_schemas = request
.column_metadatas
.iter()
.map(column_metadata_to_column_schema)
.collect::<Vec<_>>();
engine
.handle_request(region_id, RegionRequest::Create(request))
.await
.unwrap();
(engine, column_schemas)
}
/// The picker caps a compaction at 32 input files and drops the largest file groups to get
/// there. A deletion marker among the picked files must not be filtered out while the file
/// holding the rows it masks stays behind, otherwise those rows become visible again.
async fn test_compaction_input_limit_keeps_rows_deleted_with_format(flat_format: bool) {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
let region_id = RegionId::new(1, 1);
let (engine, column_schemas) =
env_for_manual_compaction(&mut env, region_id, flat_format).await;
// One large file spanning the whole time window.
put_and_flush(&engine, region_id, &column_schemas, 0..3000).await;
// Deletes 6 rows of that file. The markers land in a tiny file that overlaps it.
delete_and_flush(&engine, region_id, &column_schemas, 10..16).await;
// 31 more tiny files that overlap the large one but not each other, so the window holds
// 33 file groups forming 2 runs.
for i in 2..33 {
put_and_flush(&engine, region_id, &column_schemas, i * 10..i * 10 + 6).await;
}
let scanner = engine
.scanner(region_id, ScanRequest::default())
.await
.unwrap();
assert_eq!(
33,
scanner.num_files(),
"unexpected files: {:?}",
scanner.file_ids()
);
compact(&engine, region_id).await;
let scanner = engine
.scanner(region_id, ScanRequest::default())
.await
.unwrap();
// The 32 tiny files are merged into one; the large file exceeds the input file num limit
// and is left behind.
assert_eq!(
2,
scanner.num_files(),
"unexpected files: {:?}",
scanner.file_ids()
);
let stream = scanner.scan().await.unwrap();
let vec = collect_stream_ts(stream).await;
assert!(
!(10..16).any(|ts| vec.contains(&(ts * 1000))),
"deleted rows are visible again after compaction"
);
assert_eq!(2994, vec.len());
}
#[tokio::test]
async fn test_compaction_of_part_of_a_run_keeps_rows_deleted() {
test_compaction_of_part_of_a_run_keeps_rows_deleted_with_format(false).await;
test_compaction_of_part_of_a_run_keeps_rows_deleted_with_format(true).await;
}
/// A run is supposed to hold no overlapping files, which is what lets `merge_seq_files`
/// compact part of a run and still filter deleted rows. Run detection compares time ranges
/// exclusively though, so a file covering a single timestamp lands in the same run as the
/// file it deletes rows from. The rows must stay deleted when only one of the two is picked.
async fn test_compaction_of_part_of_a_run_keeps_rows_deleted_with_format(flat_format: bool) {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
let region_id = RegionId::new(1, 1);
let (engine, column_schemas) =
env_for_manual_compaction(&mut env, region_id, flat_format).await;
// One large file spanning the whole time window.
put_and_flush(&engine, region_id, &column_schemas, 0..3000).await;
// Deletes a single row of that file, so the marker lands in a file covering one timestamp.
delete_and_flush(&engine, region_id, &column_schemas, 1..2).await;
// 31 more single row files, each holding another key at its own timestamp.
for ts in 1..32 {
let rows = Rows {
schema: column_schemas.clone(),
rows: build_rows_for_key("b", ts * 10, ts * 10 + 1, 0),
};
put_rows(&engine, region_id, rows).await;
flush(&engine, region_id).await;
}
compact(&engine, region_id).await;
let scanner = engine
.scanner(region_id, ScanRequest::default())
.await
.unwrap();
let stream = scanner.scan().await.unwrap();
let vec = collect_stream_ts(stream).await;
assert!(
!vec.contains(&1000),
"deleted row is visible again after compaction"
);
assert_eq!(3030, vec.len());
}
// For issue https://github.com/GreptimeTeam/greptimedb/issues/3633
#[tokio::test]
async fn test_readonly_during_compaction() {