diff --git a/src/mito2/src/compaction/last_non_null.rs b/src/mito2/src/compaction/last_non_null.rs index afcdc87e90f..906788658d3 100644 --- a/src/mito2/src/compaction/last_non_null.rs +++ b/src/mito2/src/compaction/last_non_null.rs @@ -293,7 +293,7 @@ mod tests { .unwrap() .set_compacting(busy); let request = compact_request::Options::Regular(Default::default()); - let picker = new_picker(&request, ®ion.region_options, limit, None); + let picker = new_picker(&request, ®ion.region_options, limit, None, false); let picked = picker.pick(®ion).await.unwrap().unwrap(); assert_eq!(expected_windows.len(), picked.outputs.len()); @@ -312,7 +312,7 @@ mod tests { // The LastRow picker must retain its original seed cap and priority. region.region_options.merge_mode = Some(MergeMode::LastRow); - let picker = new_picker(&request, ®ion.region_options, limit, None); + let picker = new_picker(&request, ®ion.region_options, limit, None, false); let picked = picker.pick(®ion).await.unwrap().unwrap(); assert_eq!(limit.unwrap_or(3).min(3), picked.outputs.len()); for (output, window) in picked.outputs.iter().rev().zip(windows.iter().rev()) { @@ -487,6 +487,7 @@ mod tests { ®ion.region_options, Some(1), TimestampRange::new(Timestamp::new_second(0), Timestamp::new_second(3600)), + false, ); let picked = picker.pick(®ion).await.unwrap(); @@ -543,6 +544,7 @@ mod tests { ®ion.region_options, Some(1), TimestampRange::new(Timestamp::new_second(10), Timestamp::new_second(20)), + false, ); let picked = picker.pick(®ion).await.unwrap().unwrap(); @@ -590,6 +592,7 @@ mod tests { ®ion.region_options, Some(1), None, + false, ); let picked = picker.pick(®ion).await.unwrap().unwrap(); assert!(picked.outputs.is_empty()); diff --git a/src/mito2/src/compaction/picker.rs b/src/mito2/src/compaction/picker.rs index 3628953cc31..539638e570f 100644 --- a/src/mito2/src/compaction/picker.rs +++ b/src/mito2/src/compaction/picker.rs @@ -128,6 +128,7 @@ pub fn new_picker( region_options: &RegionOptions, max_background_tasks: Option, time_range: Option, + manual: bool, ) -> Arc { if let compact_request::Options::StrictWindow(window) = compact_request_options { let window = if window.window_seconds == 0 { @@ -148,8 +149,10 @@ pub fn new_picker( time_window_seconds: twcs_opts.time_window_seconds(), max_output_file_size: twcs_opts.max_output_file_size.map(|r| r.as_bytes()), append_mode: region_options.append_mode, + merge_mode: region_options.merge_mode(), max_background_tasks, time_range, + all_windows_inactive: manual, }, }; if region_options.merge_mode() == MergeMode::LastNonNull { diff --git a/src/mito2/src/compaction/scheduler.rs b/src/mito2/src/compaction/scheduler.rs index 7406a639305..22df8c094ab 100644 --- a/src/mito2/src/compaction/scheduler.rs +++ b/src/mito2/src/compaction/scheduler.rs @@ -571,7 +571,7 @@ impl CompactionScheduler { max_parallelism, ); status.merge_waiter(waiter); - self.dispatch_compaction_planning(plan_id, request, compact_options, time_range); + self.dispatch_compaction_planning(plan_id, request, compact_options, time_range, trigger); self.listener.on_compaction_scheduled(region_id); Ok(true) } @@ -617,7 +617,13 @@ impl CompactionScheduler { // was fetched above. let plan_id = Self::next_plan_id(&mut self.next_plan_id); status.start_picking_with_trigger(plan_id, CompactionTrigger::Manual); - self.dispatch_compaction_planning(plan_id, request, options, time_range); + self.dispatch_compaction_planning( + plan_id, + request, + options, + time_range, + CompactionTrigger::Manual, + ); debug!( "Successfully scheduled manual compaction planning for region id: {}", region_id @@ -703,6 +709,7 @@ impl CompactionScheduler { request, compact_request::Options::Regular(Default::default()), None, + CompactionTrigger::Automatic, ); debug!( "Successfully scheduled next compaction planning for region id: {}", diff --git a/src/mito2/src/compaction/scheduler/planning.rs b/src/mito2/src/compaction/scheduler/planning.rs index d427044b955..2cc620ebbd1 100644 --- a/src/mito2/src/compaction/scheduler/planning.rs +++ b/src/mito2/src/compaction/scheduler/planning.rs @@ -32,7 +32,9 @@ use crate::access_layer::AccessLayerRef; use crate::cache::CacheManagerRef; use crate::compaction::compactor::{CompactionRegion, CompactionVersion}; use crate::compaction::picker::{PickerOutput, new_picker}; -use crate::compaction::scheduler::state::{CompactingFiles, CompactionExecution, CompactionPhase}; +use crate::compaction::scheduler::state::{ + CompactingFiles, CompactionExecution, CompactionPhase, CompactionTrigger, +}; use crate::compaction::scheduler::{CompactionScheduler, CompactionTransition}; use crate::compaction::unit::CompactionUnit; use crate::compaction::{CompactionOutput, find_dynamic_options, format_panic_reason}; @@ -115,6 +117,7 @@ impl CompactionScheduler { request: CompactionRequest, options: compact_request::Options, time_range: Option, + trigger: CompactionTrigger, ) { let plugins = self.plugins.clone(); let max_background_compactions = self.engine_config.max_background_compactions; @@ -127,6 +130,7 @@ impl CompactionScheduler { plugins, max_background_compactions, time_range, + trigger, ); Self::notify_planning_result(region_id, plan_id, request_sender, planning).await; }); @@ -180,6 +184,7 @@ impl CompactionScheduler { plugins: Plugins, max_background_compactions: usize, time_range: Option, + trigger: CompactionTrigger, ) -> CompactionPlanningResult { let region_id = request.region_id(); let (dynamic_compaction_opts, ttl) = find_dynamic_options( @@ -208,6 +213,7 @@ impl CompactionScheduler { ®ion_options, Some(max_picker_outputs), time_range, + trigger == CompactionTrigger::Manual, ); let region_id = request.region_id(); let CompactionRequest { diff --git a/src/mito2/src/compaction/twcs.rs b/src/mito2/src/compaction/twcs.rs index a1f51b64091..bcbb3314525 100644 --- a/src/mito2/src/compaction/twcs.rs +++ b/src/mito2/src/compaction/twcs.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +mod units; + use std::collections::hash_map::Entry; use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt::Debug; @@ -34,43 +36,36 @@ use crate::compaction::run::{ Ranged, SortedRun, files_overlap, files_overlap_inclusive, find_sorted_runs, find_sorted_runs_by_time_range, merge_primary_key_ranges, primary_key_ranges_overlap, }; +use crate::compaction::twcs::units::{ClosedMergeOptions, PickLimits}; use crate::error::{JoinSnafu, Result}; +use crate::region::options::MergeMode; use crate::sst::file::{FileHandle, Level, overlaps}; use crate::sst::primary_key::PrimaryKeyRangeMapper; use crate::sst::version::LevelMeta; const LEVEL_COMPACTED: Level = 1; -#[derive(Clone, Copy, Debug)] -enum PickPhase { - HasL0, - L1FileReduction, - L1OverlapOnly, +/// Scheduling priority of a window's pick. Earlier classes are scheduled first, +/// and newer windows first within a class. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +enum PickClass { + /// Merges L0 files, which also absorbs late data of inactive windows. + L0Merge, + /// Merges files to reduce the window's file count. + FileReduction, + /// Merges files of an inactive window to resolve their overlaps. + OverlapResolution, } -impl PickPhase { - fn applies_to(self, window: &Window) -> bool { - match self { - Self::HasL0 => window.files().any(|file| file.level() == 0), - Self::L1FileReduction | Self::L1OverlapOnly => { - window.files().any(|file| file.level() != 0) - } - } - } -} - -const PICK_PHASES: [PickPhase; 3] = [ - PickPhase::HasL0, - PickPhase::L1FileReduction, - PickPhase::L1OverlapOnly, -]; - -struct WindowPickContext<'a> { - active_window: Option, - files: &'a Window, - windows: &'a BTreeMap, - phase: PickPhase, - primary_key_mapper: &'a PrimaryKeyRangeMapper, +/// Compaction inputs picked for one window. +#[derive(Debug)] +struct WindowPick { + class: PickClass, + /// Name of the strategy that picked the inputs, for logs. + branch: &'static str, + inputs: Vec, + /// Number of sorted runs or units the strategy considered, for logs. + found_runs: usize, } struct WindowOutputContext { @@ -95,6 +90,11 @@ static MAX_INPUT_FILES: LazyLock = LazyLock::new(|| { parse_max_input_files(env_value.as_deref()) }); +/// A closed merge in an inactive window may take this many times +/// [`MAX_INPUT_FILES`]. Resolving overlaps across a whole window needs all of +/// its overlapping files at once, and each byte is rewritten once per window. +const CLOSED_MERGE_INPUT_FACTOR: usize = 4; + fn parse_max_input_files(env_value: Option<&str>) -> usize { env_value .and_then(|env_value| env_value.parse().ok()) @@ -121,13 +121,22 @@ pub struct TwcsPicker { pub max_output_file_size: Option, /// Whether the target region is in append mode. pub append_mode: bool, + /// LastNonNull picks expand to an overlap closure, so cannot use partial + /// small-file reductions that must leave large peers untouched. + pub(crate) merge_mode: MergeMode, /// Max background compaction tasks. pub max_background_tasks: Option, /// Optional time range that constrains candidate compaction windows. pub(crate) time_range: Option, + /// Picks the active window like an inactive one. Manual compactions set it, + /// so a window that stopped receiving writes still gets its overlaps + /// resolved. + pub(crate) all_windows_inactive: bool, } impl TwcsPicker { + /// Picks at most one compaction input per window and schedules them across + /// windows by [`PickClass`], newer windows first within a class. async fn build_output( &self, region_id: RegionId, @@ -140,7 +149,13 @@ impl TwcsPicker { max_outputs, primary_key_mapper, } = context; - let mut output = vec![]; + let region_sequence = time_windows + .values() + .flat_map(|window| window.files()) + .filter_map(|file| file.meta_ref().sequence) + .map(|sequence| sequence.get()) + .max() + .unwrap_or(0); let windows = time_windows .values() .rev() @@ -159,278 +174,330 @@ impl TwcsPicker { .map(|window| window.time_window) .collect::>(); let time_windows = Arc::new(time_windows); + let max_outputs = max_outputs.unwrap_or(usize::MAX); // Enumerating all LastNonNull seeds must not increase the number of // concurrently planned windows along with the output limit. let chunk_size = self.max_background_tasks.unwrap_or(windows.len()).max(1); - let mut selected_windows = HashSet::new(); - 'phases: for phase in PICK_PHASES { - for chunk in windows.chunks(chunk_size) { - let mut handles = Vec::with_capacity(chunk.len()); - for window in chunk { - if selected_windows.contains(window) { - continue; - } - if !time_windows - .get(window) - .is_some_and(|files| phase.applies_to(files)) - { - continue; - } - let picker = self.clone(); - let time_windows = time_windows.clone(); - let window = *window; - let primary_key_mapper = primary_key_mapper.clone(); - handles.push(common_runtime::spawn_blocking_compact(move || { - time_windows.get(&window).map(|files| { - ( - window, - picker.find_inputs( - region_id, - WindowPickContext { - active_window, - files, - windows: &time_windows, - phase, - primary_key_mapper: &primary_key_mapper, - }, - ), - ) - }) - })); - tokio::task::yield_now().await; - } - for result in futures::future::join_all(handles).await { - let Some((window, (inputs, filter_deleted))) = result.context(JoinSnafu)? - else { - continue; - }; - if inputs.is_empty() { - continue; - } - - selected_windows.insert(window); - output.push(CompactionOutput { - output_level: LEVEL_COMPACTED, // always compact to l1 - inputs, - filter_deleted, - output_time_range: None, // we do not enforce output time range in twcs compactions. + let mut picks = Vec::new(); + for chunk in windows.chunks(chunk_size) { + let mut handles = Vec::with_capacity(chunk.len()); + for &window in chunk { + let picker = self.clone(); + let time_windows = time_windows.clone(); + let primary_key_mapper = primary_key_mapper.clone(); + handles.push(common_runtime::spawn_blocking_compact(move || { + let pick = time_windows.get(&window).and_then(|files| { + picker.pick_window( + region_id, + files, + active_window, + time_window_size, + region_sequence, + &primary_key_mapper, + ) }); - - if let Some(max_outputs) = max_outputs - && output.len() >= max_outputs - { - debug!( - "Region ({:?}) compaction output limit ({}) reached, remaining candidates discarded", - region_id, max_outputs - ); - break 'phases; - } - } + (window, pick) + })); + tokio::task::yield_now().await; + } + for result in futures::future::join_all(handles).await { + let (window, pick) = result.context(JoinSnafu)?; + picks.extend(pick.map(|pick| (window, pick))); + } + // Windows are visited newest first, so older windows cannot displace + // a full set of picks of the first class. + if picks + .iter() + .filter(|(_, pick)| pick.class == PickClass::L0Merge) + .count() + >= max_outputs + { + break; } } - // The compactor pops outputs from the end, preserving phase priority and - // newer-window-first pop order within each phase. + + // The stable sort keeps newer windows first within a class. + picks.sort_by_key(|(_, pick)| pick.class); + if picks.len() > max_outputs { + debug!( + "Region ({:?}) compaction output limit ({}) reached, remaining candidates discarded", + region_id, max_outputs + ); + picks.truncate(max_outputs); + } + let mut output = picks + .into_iter() + .map(|(window, pick)| { + let files = &time_windows[&window]; + let filter_deleted = !self.append_mode + && !window_has_overlap(files, &time_windows) + && !selected_overlaps_unselected(&pick.inputs, files, &primary_key_mapper); + if pick.inputs.len() > 1 { + log_pick_result( + region_id, + window, + active_window, + &pick, + files.files.len(), + self.max_output_file_size, + filter_deleted, + ); + } + CompactionOutput { + output_level: LEVEL_COMPACTED, // always compact to l1 + inputs: pick.inputs, + filter_deleted, + output_time_range: None, // we do not enforce output time range in twcs compactions. + } + }) + .collect::>(); + // The compactor pops outputs from the end. output.reverse(); Ok(output) } - fn find_inputs( + /// Picks the compaction inputs of one window, or `None` if it needs none. + fn pick_window( &self, region_id: RegionId, - context: WindowPickContext<'_>, - ) -> (Vec, bool) { - let WindowPickContext { - active_window, - files, - windows, - phase, - primary_key_mapper, - } = context; - let is_active_window = active_window == Some(files.time_window); - let window = &files.time_window; - let mut files_to_merge: Vec<_> = files.files().cloned().collect(); + window: &Window, + active_window: Option, + time_window_size: Option, + region_sequence: u64, + mapper: &PrimaryKeyRangeMapper, + ) -> Option { + let mut files: Vec<_> = window.files().cloned().collect(); + let quiet = is_quiet_window(&files, region_sequence); // Filter out large files in append mode - they won't benefit from compaction if self.append_mode && let Some(max_size) = self.max_output_file_size { - let (kept_files, ignored_files) = files_to_merge - .into_iter() - .partition(|file| file.size() <= max_size); - files_to_merge = kept_files; + let (kept_files, ignored_files) = + files.into_iter().partition(|file| file.size() <= max_size); + files = kept_files; if !ignored_files.is_empty() { info!( "Skipped {} large files in append mode for region {}, window {}, max_size: {}", ignored_files.len(), region_id, - window, + window.time_window, max_size ); } } - let (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(); - if !is_active_window - && files.files.len() < self.inactive_window_trigger_file_num - && num_l1_files < self.inactive_window_l1_merge_trigger + let (l0_files, l1_files): (Vec<_>, Vec<_>) = + files.into_iter().partition(|file| file.level() == 0); + if !self.all_windows_inactive && active_window == Some(window.time_window) { + self.pick_active_window(l0_files, l1_files, mapper) + } else if window.files.len() < self.inactive_window_trigger_file_num + && l1_files.len() < self.inactive_window_l1_merge_trigger { - return (vec![], false); - } - let (inputs, found_runs) = if is_active_window { - match phase { - PickPhase::HasL0 if num_l0_files >= self.trigger_file_num => pick_candidate_files( - l0_files, - self.max_output_file_size, - pick_count_first, - primary_key_mapper, - ), - PickPhase::L1FileReduction | PickPhase::L1OverlapOnly - if num_l1_files >= self.active_window_l1_merge_trigger => - { - pick_l1_candidate_files( - l1_files, - self.max_output_file_size, - phase, - primary_key_mapper, - ) - } - _ => (vec![], 0), - } + None } else { - pick_inactive_window_files( + let settling = !self.all_windows_inactive + && is_settling_window(window.time_window, active_window, time_window_size); + let num_l1_files = l1_files.len(); + self.pick_inactive_window(l0_files, l1_files, mapper, settling, quiet) + .or_else(|| { + if self.merge_mode == MergeMode::LastNonNull + || num_l1_files < self.inactive_window_l1_merge_trigger + { + return None; + } + // Include even append-mode files excluded from normal picks: + // a small-file merge must not add conflicts with large peers. + let pick = units::pick_small_l1(&window.files, mapper, self.pick_limits())?; + Some(WindowPick { + class: PickClass::FileReduction, + branch: "inactive:pick_small_l1", + inputs: pick.inputs, + found_runs: pick.units, + }) + }) + } + } + + /// The active window keeps receiving writes, so every pick must reduce its + /// file count: + /// + /// 1. Merge L0 files once they reach the trigger. + /// 2. Otherwise, merge L1 files once they reach the L1 safety trigger. L1 + /// files are merged separately from L0 files to avoid chained rewrites. + /// + /// Overlaps are left for the window to resolve once it becomes inactive: + /// resolving them while data keeps arriving rewrites the window repeatedly, + /// and leaves a dominant run that blocks the inactive-window merge. + fn pick_active_window( + &self, + l0_files: Vec, + l1_files: Vec, + mapper: &PrimaryKeyRangeMapper, + ) -> Option { + if l0_files.len() >= self.trigger_file_num { + let (inputs, found_runs) = pick_candidate_files( l0_files, - l1_files, - InactiveWindowPick { - l0_file_num: self.inactive_window_trigger_file_num, - l1_file_num: self.inactive_window_l1_merge_trigger, - phase, - primary_key_mapper, - }, self.max_output_file_size, - ) - }; - if inputs.is_empty() { - return (inputs, false); - } - - let filter_deleted = !self.append_mode - && !window_has_overlap(files, windows) - && !selected_overlaps_unselected(&inputs, files, primary_key_mapper); - - if inputs.len() > 1 { - // If we have more than one file to compact. - log_pick_result( - region_id, - *window, - active_window, - found_runs, - files.files.len(), - self.max_output_file_size, - filter_deleted, - &inputs, + pick_count_first, + mapper, ); + if !inputs.is_empty() { + return Some(WindowPick { + class: PickClass::L0Merge, + branch: "active:pick_count_first", + inputs, + found_runs, + }); + } } - (inputs, filter_deleted) + if l1_files.len() < self.active_window_l1_merge_trigger { + return None; + } + let pick = units::pick_active_l1(&l1_files, mapper, self.pick_limits())?; + Some(WindowPick { + class: PickClass::FileReduction, + branch: "active:pick_l1_file_reduction", + inputs: pick.inputs, + found_runs: pick.units, + }) } -} -/// Picks compaction inputs for an inactive window. -/// -/// The window no longer receives fresh writes (late arrivals aside), so it -/// should converge, but merging must stay within a bounded rewrite cost: -/// -/// 1. L0 candidates are selected before pure L1 candidates across all windows. -/// 2. Balanced L0 picks below the trigger continue converging without pulling -/// L1 files into the rewrite. -/// 3. A mixed merge may bypass the balance checks, but only when the total -/// rewrite fits in the output file budget, bounding write amplification. -/// 4. Last resort: converge L0 files among themselves regardless of balance. -/// The rewrite is bounded by the L0 bytes and leaves large compacted files -/// untouched. If nothing qualifies, the window is left as-is. -#[derive(Debug, Clone, Copy)] -struct InactiveWindowPick<'a> { - l0_file_num: usize, - l1_file_num: usize, - phase: PickPhase, - primary_key_mapper: &'a PrimaryKeyRangeMapper, -} - -fn pick_inactive_window_files( - l0_files: Vec, - l1_files: Vec, - pick: InactiveWindowPick<'_>, - max_output_file_size: Option, -) -> (Vec, usize) { - if !matches!(pick.phase, PickPhase::HasL0) { - return if l1_files.len() >= pick.l1_file_num { - pick_l1_candidate_files( - l1_files, - max_output_file_size, - pick.phase, - pick.primary_key_mapper, + /// An inactive window should converge at a bounded rewrite cost: + /// + /// 1. Absorb L0 files (late data), see [`pick_inactive_l0`]. + /// 2. Otherwise, merge a closed group of files that resolves overlaps or + /// removes files, see [`units::pick_closed_merge`]. Pure file reduction + /// waits for the L1 trigger; overlap resolution does not. + /// + /// A `settling` window is the one right before the active window. Late data + /// still lands there, and any rewrite of its overlapping files would be + /// repeated once more late data arrives. It only merges L0 files among + /// themselves and conflict-free files, and leaves overlaps for when the + /// window is older. + /// + /// Late data can also land in older windows. Resolving such an overlap + /// rewrites a dominant sorted run for a small peer, which would repeat for + /// every late flush, so it waits until the window is `quiet`: the region + /// has written a window's worth of rows elsewhere since the window last + /// received data. Each quiet period then pays for at most one such rewrite. + /// A manual compaction is a one-off request to resolve overlaps, so it does + /// not wait. + /// + /// Without new writes each pick consumes an L0 file or makes progress on + /// the closed merge's termination measure, so the window stops compacting. + fn pick_inactive_window( + &self, + l0_files: Vec, + l1_files: Vec, + mapper: &PrimaryKeyRangeMapper, + settling: bool, + quiet: bool, + ) -> Option { + let num_l1_files = l1_files.len(); + if !l0_files.is_empty() + && let Some((inputs, found_runs, branch)) = pick_inactive_l0( + l0_files.clone(), + (!settling).then_some(l1_files.as_slice()), + self.max_output_file_size, + mapper, ) + { + return Some(WindowPick { + class: PickClass::L0Merge, + branch, + inputs, + found_runs, + }); + } + + let mut files = l1_files; + files.extend(l0_files); + let pick = units::pick_closed_merge( + &files, + mapper, + ClosedMergeOptions { + limits: self.closed_merge_limits(), + allow_file_reduction: num_l1_files >= self.inactive_window_l1_merge_trigger, + allow_conflicts: !settling, + allow_unbalanced: (quiet || self.all_windows_inactive) && !settling, + }, + )?; + let (class, branch) = if pick.dense_fallback { + (PickClass::FileReduction, "inactive:dense_file_reduction") + } else if pick.reduces_files { + (PickClass::FileReduction, "inactive:pick_l1_file_reduction") } else { - (vec![], 0) + (PickClass::OverlapResolution, "inactive:pick_l1_overlap") }; + Some(WindowPick { + class, + branch, + inputs: pick.inputs, + found_runs: pick.units, + }) } - if l0_files.len() >= pick.l0_file_num { - let candidate = pick_candidate_files( - l0_files.clone(), - max_output_file_size, - pick_count_first, - pick.primary_key_mapper, - ); - if !candidate.0.is_empty() { - return candidate; + fn pick_limits(&self) -> PickLimits { + PickLimits { + max_output_file_size: self.max_output_file_size, + max_input_files: *MAX_INPUT_FILES, } } - let candidate = pick_candidate_files( + fn closed_merge_limits(&self) -> PickLimits { + PickLimits { + max_input_files: MAX_INPUT_FILES.saturating_mul(CLOSED_MERGE_INPUT_FACTOR), + ..self.pick_limits() + } + } +} + +/// Absorbs L0 files of an inactive window, in order of preference: +/// +/// 1. Balanced merges of L0 files, which leave L1 files untouched. +/// 2. Mixed merges that bypass the balance checks, but only when the total +/// rewrite fits in the output file budget, bounding write amplification. +/// Skipped without `l1_files`. +/// 3. Merges of L0 files among themselves regardless of balance. The rewrite +/// is bounded by the L0 bytes and leaves large compacted files untouched. +/// +/// Every pick consumes L0 files. +fn pick_inactive_l0( + l0_files: Vec, + l1_files: Option<&[FileHandle]>, + max_output_file_size: Option, + mapper: &PrimaryKeyRangeMapper, +) -> Option<(Vec, usize, &'static str)> { + let (inputs, found_runs) = pick_candidate_files( l0_files.clone(), max_output_file_size, pick_count_first, - pick.primary_key_mapper, + mapper, ); - if !candidate.0.is_empty() { - return candidate; + if !inputs.is_empty() { + return Some((inputs, found_runs, "inactive:pick_count_first")); } - let mut all_files = l0_files.clone(); - all_files.extend(l1_files); - let candidate = pick_candidate_files( - all_files, - max_output_file_size, - pick_mixed_within_budget, - pick.primary_key_mapper, - ); - if !candidate.0.is_empty() { - return candidate; + if let Some(l1_files) = l1_files { + let mut all_files = l0_files.clone(); + all_files.extend_from_slice(l1_files); + let (inputs, found_runs) = pick_candidate_files( + all_files, + max_output_file_size, + pick_mixed_within_budget, + mapper, + ); + if !inputs.is_empty() { + return Some((inputs, found_runs, "inactive:pick_mixed_within_budget")); + } } - - pick_candidate_files( + let (inputs, found_runs) = pick_candidate_files( l0_files, max_output_file_size, pick_unbalanced_count_first, - pick.primary_key_mapper, - ) -} - -fn pick_l1_candidate_files( - l1_files: Vec, - max_output_file_size: Option, - phase: PickPhase, - mapper: &PrimaryKeyRangeMapper, -) -> (Vec, usize) { - let picker = match phase { - PickPhase::L1FileReduction => pick_l1_file_reduction, - PickPhase::L1OverlapOnly => pick_l1_overlap_only, - PickPhase::HasL0 => return (vec![], 0), - }; - pick_candidate_files(l1_files, max_output_file_size, picker, mapper) + mapper, + ); + (!inputs.is_empty()).then_some((inputs, found_runs, "inactive:pick_unbalanced_count_first")) } type CandidatePicker = @@ -459,17 +526,29 @@ struct OrderedFile<'a> { file: &'a FileHandle, run_id: usize, position_in_run: usize, + /// Primary key range in the pick's schema, resolved once per pick. + primary_key_range: Option<(bytes::Bytes, bytes::Bytes)>, } -/// Metrics of a candidate compaction input, accumulated incrementally as the -/// candidate interval expands one file at a time. +impl OrderedFile<'_> { + /// Same as [`files_overlap_inclusive`] with the cached key ranges. + fn overlaps_inclusive(&self, other: &Self) -> bool { + self.file.overlap_inclusive(other.file) + && match (&self.primary_key_range, &other.primary_key_range) { + (Some(lhs), Some(rhs)) => primary_key_ranges_overlap(lhs, rhs), + _ => true, + } + } +} + +/// Metrics of a candidate compaction input, accumulated one file at a time. #[derive(Debug, Default)] struct Candidate { - /// Number of files in the interval. + /// Number of input files. num_files: usize, - /// Total input bytes of the interval. + /// Total input bytes. total_size: usize, - /// Size of the largest single file in the interval. + /// Size of the largest single input file. largest_file_size: usize, /// Files in the interval that overlap another interval file from a different /// sorted run. Resolving these overlaps is what reduces sorted runs. @@ -493,18 +572,12 @@ impl Candidate { file: &OrderedFile, preceding: &[OrderedFile], participations: &mut Vec, - mapper: &PrimaryKeyRangeMapper, ) { - self.num_files += 1; - 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); + self.absorb_file(file.file); let mut participates = false; for (offset, other) in preceding.iter().enumerate() { - if file.run_id != other.run_id && files_overlap_inclusive(file.file, other.file, mapper) - { + if file.run_id != other.run_id && file.overlaps_inclusive(other) { if !participations[offset] { participations[offset] = true; self.overlap_participants += 1; @@ -518,6 +591,15 @@ impl Candidate { } } + /// Accumulates file statistics independently of overlap accounting. + fn absorb_file(&mut self, file: &FileHandle) { + self.num_files += 1; + let file_size = 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); + } + fn absorb_level_rows(&mut self, file: &FileHandle) { let num_rows = file.num_rows(); self.has_unknown_rows |= num_rows == 0; @@ -644,28 +726,6 @@ fn pick_count_first( ) } -fn pick_l1_file_reduction( - sorted_runs: Vec>, - max_output_file_size: Option, - mapper: &PrimaryKeyRangeMapper, -) -> Vec { - pick_count_first_where(sorted_runs, max_output_file_size, mapper, |candidate| { - is_balanced_candidate(candidate) && candidate.file_reduction(max_output_file_size) > 0 - }) -} - -fn pick_l1_overlap_only( - sorted_runs: Vec>, - max_output_file_size: Option, - mapper: &PrimaryKeyRangeMapper, -) -> Vec { - pick_count_first_where(sorted_runs, max_output_file_size, mapper, |candidate| { - is_balanced_candidate(candidate) - && candidate.file_reduction(max_output_file_size) == 0 - && candidate.overlap_participants > 0 - }) -} - #[cfg(test)] fn pick_mixed_count_first( sorted_runs: Vec>, @@ -710,7 +770,7 @@ fn pick_count_first_where( mapper: &PrimaryKeyRangeMapper, is_eligible: impl Fn(&Candidate) -> bool, ) -> Vec { - let files = ordered_files(&sorted_runs); + let files = ordered_files(&sorted_runs, mapper); let mut best = None; for left in 0..files.len() { @@ -718,12 +778,7 @@ fn pick_count_first_where( let right_bound = left.saturating_add(*MAX_INPUT_FILES).min(files.len()); let mut participations: Vec = Vec::with_capacity(right_bound - left); for right in left..right_bound { - candidate.absorb( - &files[right], - &files[left..right], - &mut participations, - mapper, - ); + candidate.absorb(&files[right], &files[left..right], &mut participations); if candidate.num_files < 2 || !is_eligible(&candidate) || !candidate.makes_progress(max_output_file_size) @@ -749,7 +804,10 @@ fn pick_count_first_where( /// Flattens sorted runs into files ordered by `(start asc, end desc)`, breaking ties /// by run and position to keep the order deterministic. -fn ordered_files(sorted_runs: &[SortedRun]) -> Vec> { +fn ordered_files<'a>( + sorted_runs: &'a [SortedRun], + mapper: &PrimaryKeyRangeMapper, +) -> Vec> { let mut files = sorted_runs .iter() .enumerate() @@ -761,6 +819,7 @@ fn ordered_files(sorted_runs: &[SortedRun]) -> Vec> file, run_id, position_in_run, + primary_key_range: file.primary_key_range(mapper), }) }) .collect::>(); @@ -808,18 +867,17 @@ fn selected_overlaps_unselected( }) } -#[allow(clippy::too_many_arguments)] fn log_pick_result( region_id: RegionId, window: i64, active_window: Option, - found_runs: usize, + pick: &WindowPick, file_num: usize, max_output_file_size: Option, filter_deleted: bool, - inputs: &[FileHandle], ) { - let input_file_str: Vec = inputs + let input_file_str: Vec = pick + .inputs .iter() .map(|f| { let range = f.range(); @@ -840,13 +898,14 @@ fn log_pick_result( let active_window_str = active_window.map(|s| Timestamp::new_second(s).to_iso8601_string()); let max_output_file_size = max_output_file_size.map(|size| ReadableSize(size).to_string()); info!( - "Region ({:?}) compaction pick result: current window: {}, active window: {:?}, \ + "Region ({:?}) compaction pick result: current window: {}, active window: {:?}, branch: {}, \ found runs: {}, file num: {}, max output file size: {:?}, filter deleted: {}, \ input files: {:?}", region_id, window_str, active_window_str, - found_runs, + pick.branch, + pick.found_runs, file_num, max_output_file_size, filter_deleted, @@ -1010,6 +1069,44 @@ impl Window { } } +/// Whether `window` is the one right before `active_window`, where late data +/// still lands. +fn is_settling_window( + window: i64, + active_window: Option, + time_window_size: Option, +) -> bool { + active_window + .zip(time_window_size) + .is_some_and(|(active, size)| { + // Windows at the timestamp limits may be too far apart to subtract. + active + .checked_sub(window) + .is_some_and(|distance| (1..=size).contains(&distance)) + }) +} + +/// Whether the region has written at least as many rows as `files` hold, and at +/// least one row, since the newest of them was written. Sequences are assigned +/// per row and compaction keeps the inputs' maximum, so this needs no state. +/// Files written before row counts were recorded have zero rows, so a window +/// holding them is not quiet. +fn is_quiet_window(files: &[FileHandle], region_sequence: u64) -> bool { + if files.iter().any(|file| file.num_rows() == 0) { + return false; + } + let Some(window_sequence) = files + .iter() + .filter_map(|file| file.meta_ref().sequence) + .map(|sequence| sequence.get()) + .max() + else { + return false; + }; + let rows: u64 = files.iter().map(|file| file.num_rows() as u64).sum(); + region_sequence.saturating_sub(window_sequence) >= rows.max(1) +} + /// Assigns files to windows with predefined window size (in seconds) by their max timestamps. fn assign_to_windows<'a>( files: impl Iterator, @@ -1330,8 +1427,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker.pick(&compaction_region).await.unwrap().unwrap(); @@ -1353,8 +1452,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let compaction_region = compaction_region_with_expired_sst().await; @@ -1409,8 +1510,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker.pick(&compaction_region).await.unwrap().unwrap(); @@ -1471,8 +1574,10 @@ mod tests { time_window_seconds: Some(3600), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; assert!(picker.pick(&compaction_region).await.unwrap().is_none()); @@ -1629,6 +1734,17 @@ mod tests { } } + #[test] + fn test_settling_window_at_timestamp_limits() { + assert!(is_settling_window(0, Some(60), Some(60))); + assert!(!is_settling_window(60, Some(60), Some(60))); + assert!(!is_settling_window(-60, Some(60), Some(60))); + assert!(!is_settling_window(0, None, Some(60))); + // The distance between these windows does not fit in an i64. + assert!(!is_settling_window(i64::MAX, Some(i64::MIN), Some(1))); + assert!(!is_settling_window(i64::MIN, Some(i64::MAX), Some(1))); + } + #[test] fn test_window_assignment_at_timestamp_limits() { for (seconds, expected_window) in [(i64::MIN, i64::MIN), (i64::MAX, i64::MAX - 1)] { @@ -2028,8 +2144,10 @@ mod tests { time_window_seconds: None, max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, } .build_output_with_time_range(RegionId::from_u64(0), windows, active_window, None) .await @@ -2179,8 +2297,10 @@ mod tests { time_window_seconds: None, max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, } .build_output_with_time_range(RegionId::from_u64(0), windows, active_window, None) .await @@ -2250,8 +2370,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let active_window = find_latest_window_in_seconds(files.iter(), 3); @@ -2298,8 +2420,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: Some(1000), append_mode: true, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let active_window = find_latest_window_in_seconds(files.iter(), 3); @@ -2332,8 +2456,10 @@ mod tests { time_window_seconds: Some(1), max_output_file_size: Some(1_000), append_mode: true, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -2377,8 +2503,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: Some(max_background_tasks), time_range: None, + all_windows_inactive: false, }; let active_window = find_latest_window_in_seconds(files.iter(), 3); @@ -2404,8 +2532,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let windows_no_limit = assign_to_windows(files.iter(), 3); @@ -2451,8 +2581,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: Some(max_background_tasks), time_range: None, + all_windows_inactive: false, }; let active_window = find_latest_window_in_seconds(files.iter(), 3); @@ -2503,8 +2635,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let active_window = find_latest_window_in_seconds(files.iter(), 3); @@ -2550,8 +2684,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let active_window = find_latest_window_in_seconds(files.iter(), 3); @@ -2599,8 +2735,10 @@ mod tests { time_window_seconds: Some(3600), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let active_window = find_latest_window_in_seconds(files.iter(), 3600); @@ -2648,8 +2786,10 @@ mod tests { time_window_seconds: Some(3600), max_output_file_size: Some(1024 * 1024 * 1024), append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let active_window = find_latest_window_in_seconds(files.iter(), 3600); @@ -2682,8 +2822,10 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: Some(1), time_range: None, + all_windows_inactive: false, }; let output = picker @@ -2779,11 +2921,13 @@ mod tests { time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: Some(1), time_range: TimestampRange::new( Timestamp::new_millisecond(7_200), Timestamp::new_millisecond(7_800), ), + all_windows_inactive: false, }; let output = picker @@ -2817,8 +2961,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -2830,79 +2976,6 @@ mod tests { assert_eq!(2, output[0].inputs.len()); } - #[test] - fn test_inactive_window_l1_trigger_is_independent_of_l0_trigger() { - let l0_files = (0..6) - .map(|idx| { - new_file_handle_with_size_and_sequence( - FileId::random(), - idx * 20, - idx * 20 + 10, - 0, - idx as u64 + 1, - 10, - ) - }) - .collect(); - let l1_files = (0..2) - .map(|idx| { - new_file_handle_with_size_and_sequence( - FileId::random(), - idx * 20, - idx * 20 + 10, - 1, - idx as u64 + 10, - 10, - ) - }) - .collect(); - - let (inputs, _) = pick_inactive_window_files( - l0_files, - l1_files, - InactiveWindowPick { - l0_file_num: 8, - l1_file_num: 2, - phase: PickPhase::L1FileReduction, - primary_key_mapper: &primary_key_mapper_for_test(), - }, - None, - ); - - assert_eq!(2, inputs.len()); - assert!(inputs.iter().all(|file| file.level() == 1)); - } - - #[test] - fn test_inactive_window_does_not_fallback_to_l1_below_trigger() { - let l1_files = (0..2) - .map(|idx| { - new_file_handle_with_size_and_sequence( - FileId::random(), - idx * 20, - idx * 20 + 10, - 1, - idx as u64 + 1, - 10, - ) - }) - .collect(); - - let (inputs, _) = pick_inactive_window_files( - vec![], - l1_files, - InactiveWindowPick { - l0_file_num: 2, - l1_file_num: 8, - phase: PickPhase::L1FileReduction, - primary_key_mapper: &primary_key_mapper_for_test(), - }, - None, - ); - - assert!(inputs.is_empty()); - } - #[tokio::test] async fn test_inactive_window_requires_a_level_trigger_before_mixed_fallback() { let files = [ @@ -2918,8 +2991,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: Some(1_000), append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -2947,8 +3022,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -2976,8 +3053,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3004,8 +3083,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3047,8 +3128,10 @@ mod tests { time_window_seconds: Some(1000), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3229,8 +3312,10 @@ mod tests { time_window_seconds: Some(1), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; for (case, files, expected_level, expected_len) in [ @@ -3280,8 +3365,10 @@ mod tests { time_window_seconds: Some(1), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3317,8 +3404,10 @@ mod tests { time_window_seconds: Some(1), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3367,8 +3456,10 @@ mod tests { time_window_seconds: Some(1), max_output_file_size: Some(512), append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3414,8 +3505,10 @@ mod tests { time_window_seconds: Some(1), max_output_file_size: Some(100), append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: Some(max_background_tasks), time_range: None, + all_windows_inactive: false, } } @@ -3428,6 +3521,209 @@ mod tests { .value() } + #[tokio::test] + async fn test_manual_compaction_resolves_active_window_overlaps() { + let files = priority_test_files(5, 1, 100, true); + for (all_windows_inactive, expected) in [(false, 0), (true, 1)] { + let picker = TwcsPicker { + all_windows_inactive, + ..priority_test_picker(1) + }; + let output = picker + .build_output_with_time_range( + RegionId::from_u64(1), + assign_to_windows(files.iter(), 1), + Some(5), + None, + ) + .await + .unwrap(); + assert_eq!(expected, output.len(), "manual: {all_windows_inactive}"); + } + } + + #[tokio::test] + async fn test_shared_boundary_l1_files_resolve_in_inactive_and_manual_windows() { + let files = [(b"a", b"m"), (b"m", b"z")].map(|(min, max)| { + new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + 0, + 100, + 1, + 1, + 100, + pk_range(min, max), + ) + }); + for (active_window, all_windows_inactive, expected) in + [(0, false, 0), (0, true, 1), (3, false, 1)] + { + let picker = TwcsPicker { + all_windows_inactive, + ..priority_test_picker(1) + }; + let output = picker + .build_output_with_time_range( + RegionId::from_u64(1), + assign_to_windows(files.iter(), 1), + Some(active_window), + Some(1), + ) + .await + .unwrap(); + assert_eq!( + expected, + output.len(), + "active: {active_window}, manual: {all_windows_inactive}" + ); + if let Some(output) = output.first() { + assert_eq!(2, output.inputs.len()); + } + } + } + + #[tokio::test] + async fn test_settling_window_defers_overlap_resolution() { + let files = priority_test_files(3, 1, 100, true); + for (active_window, all_windows_inactive, expected) in + [(4, false, 0), (5, false, 1), (2, false, 1), (4, true, 1)] + { + let picker = TwcsPicker { + all_windows_inactive, + ..priority_test_picker(1) + }; + let output = picker + .build_output_with_time_range( + RegionId::from_u64(1), + assign_to_windows(files.iter(), 1), + Some(active_window), + Some(1), + ) + .await + .unwrap(); + assert_eq!( + expected, + output.len(), + "active: {active_window}, manual: {all_windows_inactive}" + ); + } + } + + #[tokio::test] + async fn test_inactive_windows_merge_small_l1_without_large_peer() { + let large = + new_file_handle_with_size_and_sequence(FileId::random(), 3_000, 3_999, 1, 1, 1_000); + let small = priority_test_files(3, 1, 20, true); + let files = [vec![large.clone()], small.to_vec()].concat(); + // The immediate predecessor is settling; the older window is not quiet. + for (active_window, merge_mode, trigger, expected) in [ + (4, MergeMode::LastRow, 2, 1), + (5, MergeMode::LastRow, 2, 1), + (4, MergeMode::LastNonNull, 2, 0), + (5, MergeMode::LastNonNull, 2, 0), + (4, MergeMode::LastRow, 4, 0), + ] { + let picker = TwcsPicker { + merge_mode, + inactive_window_l1_merge_trigger: trigger, + ..priority_test_picker(1) + }; + let output = picker + .build_output_with_time_range( + RegionId::from_u64(1), + assign_to_windows(files.iter(), 1), + Some(active_window), + Some(1), + ) + .await + .unwrap(); + assert_eq!( + expected, + output.len(), + "active: {active_window}, mode: {merge_mode:?}, trigger: {trigger}" + ); + if expected == 0 { + continue; + } + let inputs: HashSet<_> = output[0].inputs.iter().map(FileHandle::file_id).collect(); + assert_eq!( + small + .iter() + .map(FileHandle::file_id) + .collect::>(), + inputs + ); + assert!( + !output[0].filter_deleted, + "the unselected large file still overlaps" + ); + } + } + + #[tokio::test] + async fn test_small_l1_fallback_checks_large_append_mode_peer() { + // Two conflicting small files form a non-rectangular area. Their union + // would introduce a conflict with the large file in the empty corner. + let files = [ + (0, 20, b"a", b"m", 20), + (10, 30, b"l", b"z", 20), + (0, 5, b"y", b"z", 1_000), + ] + .map(|(start, end, min, max, size)| { + new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + start, + end, + 1, + 1, + size, + pk_range(min, max), + ) + }); + for append_mode in [false, true] { + let picker = TwcsPicker { + append_mode, + ..priority_test_picker(1) + }; + let output = picker + .build_output_with_time_range( + RegionId::from_u64(1), + assign_to_windows(files.iter(), 1), + Some(1), + Some(1), + ) + .await + .unwrap(); + assert!(output.is_empty(), "append mode: {append_mode}"); + } + } + + #[tokio::test] + async fn test_settling_window_does_not_rewrite_l1_for_late_data() { + // A late L0 file overlapping both L1 files, which a mixed merge absorbs. + let mut files = priority_test_files(3, 1, 40, false).to_vec(); + files.push(new_file_handle_with_size_and_sequence( + FileId::random(), + 3_000, + 3_299, + 0, + 10, + 10, + )); + for (active_window, expected) in [(4, 0), (5, 1)] { + let output = priority_test_picker(1) + .build_output_with_time_range( + RegionId::from_u64(1), + assign_to_windows(files.iter(), 1), + Some(active_window), + Some(1), + ) + .await + .unwrap(); + assert_eq!(expected, output.len(), "active: {active_window}"); + } + } + #[tokio::test] async fn test_older_l0_candidate_has_priority_over_newer_pure_l1_candidate() { let files = priority_test_files(0, 0, 40, false) @@ -3447,7 +3743,7 @@ mod tests { } #[tokio::test] - async fn test_older_l1_file_reduction_has_priority_over_newer_overlap_only_l1() { + async fn test_older_l1_file_reduction_has_priority_over_newer_overlap_resolution() { let files = priority_test_files(0, 1, 40, false) .into_iter() .chain(priority_test_files(1, 1, 100, true)) @@ -3488,7 +3784,9 @@ mod tests { .map(|output| (output.inputs[0].level(), output_window(output))) .collect::>(); - assert_eq!(vec![(0, 1), (0, 0), (1, 3), (1, 2), (1, 5)], pop_priority); + // The active window 5 only reduces files, so the overlap slot goes to + // the inactive window 4. + assert_eq!(vec![(0, 1), (0, 0), (1, 3), (1, 2), (1, 4)], pop_priority); let units = crate::compaction::unit::CompactionUnit::from_picker(PickerOutput { outputs: output, time_window_size: 1, @@ -3525,8 +3823,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: Some(2_000), append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3556,8 +3856,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: Some(100_000), append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3587,8 +3889,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: Some(100_000), append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3618,8 +3922,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: Some(1_000_000), append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker @@ -3710,6 +4016,436 @@ mod tests { assert_eq!(vec![(0, 19), (10, 29)], picked_ranges(&picked)); } + #[tokio::test] + async fn test_l1_partial_pk_overlap_does_not_rewrite_unrelated_files() { + // Region 1025's 2026-09-23 R2: A1..A8 are R1's outputs, plus one L1 E. + // Keep the logged, rounded sizes (tenths of MiB) and normalize timestamps + // while preserving the observed order: A3, A7, A6, A1, A4, A5, A2, A8, E. + // The PK ranges are hypothetical: the logs only contain their lengths. + let a_files = [ + (4, 5249), // A1 + (7, 5218), // A2 + (1, 5255), // A3 + (5, 5263), // A4 + (6, 5131), // A5 + (3, 5143), // A6 + (2, 5197), // A7 + (8, 4716), // A8 + ] + .into_iter() + .enumerate() + .map(|(idx, (start, size))| { + let key = b'a' + idx as u8; + new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + start, + 10_000, + 1, + 100, + size * 1024 * 1024 / 10, + pk_range(&[key, b'0'], &[key, b'9']), + ) + }) + .collect::>(); + let mapper = primary_key_mapper_for_test(); + for (idx, file) in a_files.iter().enumerate() { + for other in &a_files[idx + 1..] { + assert!(file.overlap_inclusive(other)); // Their time ranges overlap. + assert!(!files_overlap_inclusive(file, other, &mapper)); + } + } + + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 4, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(60), + max_output_file_size: Some(512 * 1024 * 1024), + append_mode: false, + merge_mode: MergeMode::LastRow, + max_background_tasks: Some(1), + time_range: None, + all_windows_inactive: false, + }; + + for (case, key_range, extra_size, overlapping_a, expected_inputs) in [ + ( + "A3 and E skip the unrelated A files", + Some((b'c', b'c')), + 5155, + &[2][..], + &[2, 8][..], + ), + ( + "A8 and E must not pull in A2", + Some((b'h', b'h')), + 5155, + &[7][..], + &[7, 8][..], + ), + ( + "E overlaps no A: no progress", + Some((b'z', b'z')), + 5155, + &[][..], + &[] as &[usize], + ), + ( + "A3 and E may split into one extra tail file", + Some((b'c', b'c')), + 5255, + &[2][..], + &[2, 8][..], + ), + ( + "two A files exceed E's run budget", + Some((b'b', b'c')), + 5155, + &[1, 2][..], + &[][..], + ), + ( + "sufficient E bytes allow two A files separated in time order", + Some((b'b', b'c')), + 6000, + &[1, 2][..], + &[1, 2, 8][..], + ), + ( + "unknown PK bounds include the whole dominant run in the budget", + None, + 5155, + &[0, 1, 2, 3, 4, 5, 6, 7][..], + &[][..], + ), + ] { + let extra = new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + 1_000, + 9_000, + 1, + 100, + extra_size * 1024 * 1024 / 10, + key_range.and_then(|(min, max)| pk_range(&[min, b'0'], &[max, b'9'])), + ); + for (idx, file) in a_files.iter().enumerate() { + assert!(file.overlap_inclusive(&extra)); + assert_eq!( + overlapping_a.contains(&idx), + files_overlap_inclusive(file, &extra, &mapper), + "{case}: A{}", + idx + 1, + ); + } + + let mut files = a_files.clone(); + files.push(extra); + for active_window in [Some(0), None] { + // The active window leaves pure overlap resolution to the window + // becoming inactive. + let expected_inputs = if active_window.is_some() { + &[][..] + } else { + expected_inputs + }; + let expected = expected_inputs + .iter() + .map(|&idx| files[idx].file_id()) + .collect::>(); + let windows = assign_to_windows(files.iter(), 60); + let outputs = picker + .build_output_with_time_range( + RegionId::from_u64(0), + windows, + active_window, + Some(60), + ) + .await + .unwrap(); + + assert_eq!( + usize::from(!expected_inputs.is_empty()), + outputs.len(), + "{case}, active window: {active_window:?}" + ); + let actual = outputs + .iter() + .flat_map(|output| output.inputs.iter().map(FileHandle::file_id)) + .collect::>(); + assert_eq!(expected, actual, "{case}, active window: {active_window:?}"); + } + } + } + + #[rstest::rstest] + #[case::active_filtered(Some(0), false, 1000, &[0, 2])] + #[case::inactive_filtered(None, false, 1000, &[0, 2])] + #[case::older_reduction_before_newer_overlap(Some(3), true, 1000, &[0, 2])] + #[case::contiguous_has_more_reduction(Some(0), false, 90, &[0, 1, 2, 3])] + #[case::filtered_has_equal_reduction_with_less_io(Some(0), false, 600, &[0, 2])] + #[tokio::test] + async fn test_l1_file_reduction_considers_filtered_candidates( + #[case] active_window: Option, + #[case] newer_overlap_window: bool, + #[case] unrelated_size: u64, + #[case] expected_indices: &[usize], + ) { + // Only A/B overlap. With 1000-byte unrelated files between/after them, + // no contiguous interval reduces the file count, but A+B alone does. + let mut files = [ + (b'a', 100), + (b'b', unrelated_size), + (b'a', 100), + (b'c', unrelated_size), + ] + .into_iter() + .enumerate() + .map(|(idx, (key, size))| { + new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + idx as i64 * 100, + 749, + 1, + 1, + size, + pk_range(&[key], &[key]), + ) + }) + .collect::>(); + let expected = expected_indices + .iter() + .map(|&idx| files[idx].file_id()) + .collect::>(); + if newer_overlap_window { + // One slot: a filtered file-reducing pick must beat a newer pure + // rewrite, rather than being deferred behind overlap resolution. + files.extend(priority_test_files(1, 1, 512, true)); + } + let picker = TwcsPicker { + max_output_file_size: Some(512), + ..priority_test_picker(1) + }; + let outputs = picker + .build_output_with_time_range( + RegionId::from_u64(0), + assign_to_windows(files.iter(), 1), + active_window, + Some(1), + ) + .await + .unwrap(); + + assert_eq!(1, outputs.len()); + assert_eq!( + expected, + outputs[0] + .inputs + .iter() + .map(FileHandle::file_id) + .collect::>() + ); + } + + #[tokio::test] + async fn test_overlap_resolution_limits_dominant_run_rewrites() { + let file_size = 512 * 1024 * 1024; + // Peers are full-size files in one run, so a pure rewrite keeps the file count. + for (case, peer_sizes, max_output_size, should_pick) in [ + ( + "small peer cannot rewrite any slice of the large run", + vec![file_size], + file_size, + false, + ), + ( + "one byte below the run budget", + [vec![file_size; 6], vec![file_size - 1]].concat(), + file_size, + false, + ), + ( + "exact run budget boundary", + vec![file_size; 7], + file_size, + true, + ), + ("equal sized runs", vec![file_size; 8], file_size, true), + ( + "file reduction keeps its own admission", + vec![file_size], + 2 * file_size, + true, + ), + ] { + let mut files = (0..8) + .map(|idx| { + let key = b'a' + idx as u8; + new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + idx, + 10_000, + 1, + 100, + file_size, + pk_range(&[key, b'0'], &[key, b'9']), + ) + }) + .collect::>(); + files.extend(peer_sizes.into_iter().enumerate().map(|(idx, size)| { + let start = 1_000 * (idx as i64 + 1); + new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + start, + start + 999, + 1, + 100, + size, + pk_range(b"a0", b"h9"), + ) + })); + // Only the last case removes files; the active window leaves pure + // overlap resolution to the window becoming inactive. + let reduces_files = max_output_size > file_size; + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 4, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(60), + max_output_file_size: Some(max_output_size), + append_mode: false, + merge_mode: MergeMode::LastRow, + max_background_tasks: Some(1), + time_range: None, + all_windows_inactive: false, + }; + + for active_window in [Some(0), None] { + let expected = if should_pick && (reduces_files || active_window.is_none()) { + files + .iter() + .map(FileHandle::file_id) + .collect::>() + } else { + HashSet::new() + }; + let outputs = picker + .build_output_with_time_range( + RegionId::from_u64(0), + assign_to_windows(files.iter(), 60), + active_window, + Some(60), + ) + .await + .unwrap(); + let actual = outputs + .iter() + .flat_map(|output| output.inputs.iter().map(FileHandle::file_id)) + .collect::>(); + assert_eq!(expected, actual, "{case}, active window: {active_window:?}"); + } + } + } + + #[tokio::test] + async fn test_quiet_or_manual_window_resolves_dominant_run_overlap() { + let file_size = 512 * 1024 * 1024; + let l1_file = + |start, end, sequence, size, range: Option<(bytes::Bytes, bytes::Bytes)>, num_rows| { + let file = new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + start, + end, + 1, + sequence, + size, + range.clone(), + ); + FileHandle::new_with_primary_key_range( + FileMeta { + num_rows, + ..file.meta_ref().clone() + }, + crate::test_util::new_noop_file_purger(), + range, + ) + }; + + // The window holds 9 rows. It is quiet once the region writes 9 more + // rows after it, and only if its row counts are known. A manual + // compaction does not wait, even for the active window. + for (newer_sequence, num_rows, manual, resolves) in [ + (108, 1, false, false), + (109, 1, false, true), + (109, 0, false, false), + (100, 1, true, true), + ] { + // A run of eight key-sliced files and one small file overlapping all + // of them: resolving the overlap rewrites the run for its small peer. + let mut window_files = (0..8) + .map(|idx| { + let key = b'a' + idx as u8; + let range = pk_range(&[key, b'0'], &[key, b'9']); + l1_file(idx, 10_000, 100, file_size, range, num_rows) + }) + .collect::>(); + window_files.push(l1_file( + 1_000, + 1_999, + 100, + file_size / 8, + pk_range(b"a0", b"h9"), + num_rows, + )); + let expected = window_files + .iter() + .map(FileHandle::file_id) + .collect::>(); + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 4, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 16, + time_window_seconds: Some(60), + max_output_file_size: Some(file_size), + append_mode: false, + merge_mode: MergeMode::LastRow, + max_background_tasks: Some(1), + time_range: None, + all_windows_inactive: manual, + }; + let mut files = window_files; + files.push(l1_file( + 120_000, + 120_999, + newer_sequence, + file_size, + pk_range(b"a0", b"h9"), + 1, + )); + let outputs = picker + .build_output_with_time_range( + RegionId::from_u64(0), + assign_to_windows(files.iter(), 60), + manual.then_some(0), + Some(60), + ) + .await + .unwrap(); + let actual = outputs + .iter() + .flat_map(|output| output.inputs.iter().map(FileHandle::file_id)) + .collect::>(); + let case = format!("sequence {newer_sequence}, rows {num_rows}, manual {manual}"); + if resolves { + assert_eq!(expected, actual, "{case}"); + } else { + assert!(actual.is_empty(), "{case}: {actual:?}"); + } + } + } + #[test] fn test_count_first_prefers_guaranteed_reduction_over_pure_rewrite() { // A window mixing large and small files: the large triple on its own is a @@ -3820,8 +4556,10 @@ mod tests { time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, + merge_mode: MergeMode::LastRow, max_background_tasks: None, time_range: None, + all_windows_inactive: false, }; let output = picker diff --git a/src/mito2/src/compaction/twcs/units.rs b/src/mito2/src/compaction/twcs/units.rs new file mode 100644 index 00000000000..eeb60153614 --- /dev/null +++ b/src/mito2/src/compaction/twcs/units.rs @@ -0,0 +1,2073 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Unit-based compaction picks for one TWCS window. +//! +//! Picks select a contiguous interval of *units* in a fixed order. Grouping +//! prefers rewriting related files together, but input limits and fallback +//! paths may split groups or use single-file units. Separate admission checks +//! preserve each pick's progress guarantees under the assumptions below. +//! +//! - [`pick_active_l1`] reduces the L1 file count of the active window. +//! - [`pick_small_l1`] reduces small L1 files without rewriting large peers. +//! - [`pick_closed_merge`] only merges intervals of an inactive window whose +//! outputs cannot add conflicts. Compaction outputs stay inside the bounding +//! box of the inputs, so they can only overlap files intruding on that box. +//! +//! When a window's split-unit conflict graph is too dense to materialize, +//! [`pick_closed_merge`] falls back to a bounded file reduction and admits it +//! only when the same conflict bound holds. +//! +//! # Termination +//! +//! Let `P` be the number of conflicting file pairs ([`Extent::conflicts`]) in an +//! inactive window and `N` its file count. A merge removes every pair among its +//! inputs. Its outputs are one sorted run with no internal conflicts, +//! so they only conflict with the `I` files intruding on the inputs' box, at +//! most `outputs · I` pairs. An inactive pick only merges inputs that had at +//! least as many pairs with other files, so `P` does not grow; without +//! intruders the inputs are closed and add no pairs. Every inactive pick either +//! resolves a pair (`P` drops) or removes a file (`P` does not grow and `N` +//! drops), so `(P, N)` strictly decreases in lexicographic order until the +//! window receives new writes. Active-window picks may add conflicts, so each +//! of them must remove a file: `N` strictly decreases. +//! +//! The argument relies on how the compactor writes outputs: +//! +//! - Outputs hold a subset of the input rows, so each output's time and key +//! bounds lie inside the inputs' bounding box. +//! - Outputs are one stream in (key, time) order. For regions with a primary +//! key, the writer only splits between complete series, so output key ranges +//! are disjoint. Without a primary key, outputs have disjoint or touching +//! time ranges. Neither is an [`Extent::conflicts`] pair. +//! - A split only starts a new file once the current one reaches the size +//! threshold, and outputs are not larger than inputs in total, so +//! [`PickLimits::predicted_outputs`] bounds the output count. +//! +//! `test_closed_merges_terminate` and `test_active_l1_picks_terminate` replay +//! random windows with outputs written under these rules. + +use std::cmp::Reverse; +use std::ops::Range; + +use bytes::Bytes; +use common_time::Timestamp; + +use crate::sst::file::FileHandle; +use crate::sst::primary_key::PrimaryKeyRangeMapper; + +/// An L1 file reduction may rewrite at most this many output files' worth of +/// bytes per SST it removes. +const MAX_REWRITE_FILES_PER_REDUCED_FILE: u64 = 4; + +/// A small L1 file occupies at most one quarter of the output size budget. +const SMALL_L1_SIZE_FACTOR: u64 = 4; + +/// Maximum number of index nodes visited while building conflict structures for +/// a pick. Above this, the picker falls back to a bounded file reduction instead +/// of materializing an unbounded conflict graph. +const CONFLICT_GRAPH_BUDGET: usize = 1 << 20; + +/// Output size limits and input limits shared by both picks. +#[derive(Debug, Clone, Copy)] +pub(crate) struct PickLimits { + /// Split threshold of compaction outputs; `None` writes a single output. + pub(crate) max_output_file_size: Option, + /// Maximum number of input files of one compaction. + pub(crate) max_input_files: usize, +} + +impl PickLimits { + /// Upper bound of the outputs of compacting an interval: the writer only + /// starts a new file after the current one reaches the threshold, and never + /// splits a series, so files of a single primary key compact into one file. + fn predicted_outputs(&self, stats: &IntervalStats) -> usize { + match self.max_output_file_size { + Some(max) if max > 0 && stats.single_key.is_none() => { + stats.size.div_ceil(max).max(1) as usize + } + _ => 1, + } + } + + /// Whether rewriting `size` bytes to remove `reduction` files stays within + /// the rewrite budget. An unset output size imposes no bound. + fn within_reduction_cost(&self, size: u64, reduction: usize) -> bool { + match self.max_output_file_size { + Some(max) if max > 0 => { + size <= (reduction as u64) + .saturating_mul(MAX_REWRITE_FILES_PER_REDUCED_FILE) + .saturating_mul(max) + } + _ => true, + } + } + + /// Bytes a dominant sorted run may exceed its peers by in an overlap merge. + fn run_tolerance(&self) -> u64 { + self.max_output_file_size.unwrap_or(0) + } +} + +/// A selected compaction input. +#[derive(Debug)] +pub(crate) struct UnitPick { + pub(crate) inputs: Vec, + /// Whether the pick is predicted to reduce the window's file count. Otherwise + /// it only resolves overlaps. + pub(crate) reduces_files: bool, + /// Number of units the candidate files were grouped into. + pub(crate) units: usize, + /// Whether this pick came from the bounded dense-window fallback. + pub(crate) dense_fallback: bool, +} + +/// Picks L1 files of the active window whose compaction removes files. +/// +/// Units are groups of conflicting files, so a merge does not leave a file +/// behind while rewriting its overlapping peers. A group larger than the input +/// limit is split back into single files. If no interval of groups qualifies, +/// for example because a large file dominates its group, the groups are split +/// into single files, so the smaller files can still be merged. Intervals +/// follow [`sort_units`]; the pick removes the most files, then rewrites the +/// fewest bytes. +pub(crate) fn pick_active_l1( + l1_files: &[FileHandle], + mapper: &PrimaryKeyRangeMapper, + options: PickLimits, +) -> Option { + pick_active_l1_with_budget(l1_files, mapper, options, CONFLICT_GRAPH_BUDGET) +} + +/// [`pick_active_l1`] with an explicit work budget, so tests can exercise the +/// single-file fallback without a window-sized conflict graph. +fn pick_active_l1_with_budget( + l1_files: &[FileHandle], + mapper: &PrimaryKeyRangeMapper, + options: PickLimits, + work_budget: usize, +) -> Option { + let extents = extents_of(l1_files, mapper); + let index = ExtentIndex::new(&extents); + let mut units = match conflict_units(l1_files, &extents, &index, work_budget) { + Some(units) => units, + None => single_file_units(l1_files, &extents), + }; + split_oversized_units(&mut units, l1_files, &extents, options.max_input_files); + if let Some(pick) = pick_reducing_interval(&mut units, l1_files, options) { + return Some(pick); + } + if units.len() == l1_files.len() { + return None; + } + split_oversized_units(&mut units, l1_files, &extents, 1); + pick_reducing_interval(&mut units, l1_files, options) +} + +/// Picks the interval of `units` that removes the most files within the +/// rewrite budget, then rewrites the fewest bytes. +fn pick_reducing_interval( + units: &mut [Unit], + files: &[FileHandle], + options: PickLimits, +) -> Option { + sort_units(units); + let range = pick_reducing_interval_where(units, options, |_| true)?; + Some(UnitPick { + inputs: collect_inputs(&units[range], files), + reduces_files: true, + units: units.len(), + dense_fallback: false, + }) +} + +/// Selects a reducing interval of sorted units, with an additional filter so +/// the dense fallback can retry without duplicating the reduction guards. +fn pick_reducing_interval_where( + units: &[Unit], + options: PickLimits, + is_eligible: impl Fn(&IntervalStats) -> bool, +) -> Option> { + let mut best: Option<(usize, u64, Range)> = None; + // File reductions ignore conflicts, so split files are not counted. + for_each_interval(units, &[], options.max_input_files, |range, stats| { + let reduction = stats + .num_files + .saturating_sub(options.predicted_outputs(stats)); + if stats.num_files < 2 + || reduction == 0 + || !options.within_reduction_cost(stats.size, reduction) + || stats.largest_file_size > stats.size - stats.largest_file_size + || !is_eligible(stats) + { + return; + } + let better = best.as_ref().is_none_or(|(best_reduction, best_size, _)| { + (reduction, Reverse(stats.size)) > (*best_reduction, Reverse(*best_size)) + }); + if better { + best = Some((reduction, stats.size, range)); + } + }); + + best.map(|(_, _, range)| range) +} + +/// Reduces small L1 files left behind by a closed merge, including in settling +/// windows. The input bytes fit in one output and no input dominates its peers, +/// so each merge removes files without repeatedly rewriting an accumulator. +/// +/// Large and L0 files remain in the external conflict index. Check only the +/// best reducing interval to bound planning work; rejecting it can miss other +/// admissible intervals, but cannot increase the window's conflict count. +pub(crate) fn pick_small_l1( + files: &[FileHandle], + mapper: &PrimaryKeyRangeMapper, + limits: PickLimits, +) -> Option { + let max_output = limits.max_output_file_size.filter(|size| *size > 0)?; + let extents = extents_of(files, mapper); + let mut units = single_file_units(files, &extents); + units.retain(|unit| { + let file = &files[unit.files[0]]; + file.level() == 1 && file.size() <= max_output / SMALL_L1_SIZE_FACTOR + }); + sort_units(&mut units); + let range = pick_reducing_interval_where(&units, limits, |stats| stats.size <= max_output)?; + let index = ExtentIndex::new(&extents); + if !adds_no_conflicts(&units[range.clone()], &extents, &index, 1) { + return None; + } + Some(UnitPick { + inputs: collect_inputs(&units[range], files), + reduces_files: true, + units: units.len(), + dense_fallback: false, + }) +} + +/// Thresholds of the closed merge pick. +#[derive(Debug, Clone, Copy)] +pub(crate) struct ClosedMergeOptions { + pub(crate) limits: PickLimits, + /// Whether merges that only remove files are allowed. + pub(crate) allow_file_reduction: bool, + /// Whether merges may rewrite conflicting files. Otherwise only + /// conflict-free intervals are merged. + pub(crate) allow_conflicts: bool, + /// Whether overlap merges may rewrite a dominant sorted run for a small + /// peer. Otherwise such a merge would be repeated whenever late data adds + /// another small peer. + pub(crate) allow_unbalanced: bool, +} + +/// Picks an interval of units among all files of an inactive window. +/// +/// Units are the closures of conflicting files: no file outside a unit intrudes +/// on its bounding box. A closure larger than the input limit is split back into +/// single files, so files around a late file spanning a large sorted run can +/// still be merged. Intervals follow [`sort_units`]. An interval is eligible if +/// its outputs add no more conflicts than its files have with other files (see +/// [`adds_no_conflicts`]), and it +/// +/// - removes files within the rewrite budget, or +/// - resolves conflicts without rewriting a dominant sorted run (unless +/// [`ClosedMergeOptions::allow_unbalanced`]), and adds at most one tail file. +/// +/// The pick maximizes resolved conflicts plus removed files per rewritten byte. +/// A window whose split-unit conflict graph is too dense to materialize falls +/// back to a bounded file reduction; see [`pick_dense_closed_merge`]. +pub(crate) fn pick_closed_merge( + files: &[FileHandle], + mapper: &PrimaryKeyRangeMapper, + options: ClosedMergeOptions, +) -> Option { + pick_closed_merge_with_budget(files, mapper, options, CONFLICT_GRAPH_BUDGET) +} + +/// [`pick_closed_merge`] with an explicit work budget, so tests can exercise +/// the dense fallback without building a window-sized conflict graph. +fn pick_closed_merge_with_budget( + files: &[FileHandle], + mapper: &PrimaryKeyRangeMapper, + options: ClosedMergeOptions, + work_budget: usize, +) -> Option { + let limits = options.limits; + let extents = extents_of(files, mapper); + let index = ExtentIndex::new(&extents); + let mut dense = false; + let mut units = match conflict_units(files, &extents, &index, work_budget) { + Some(mut units) => { + close_units(&mut units, &extents, &index); + split_oversized_units(&mut units, files, &extents, limits.max_input_files); + units + } + None => { + dense = true; + single_file_units(files, &extents) + } + }; + sort_units(&mut units); + if dense { + return pick_dense_closed_merge(&units, files, &extents, &index, options); + } + let Some(pairs) = split_conflicts(&units, &extents, &index, work_budget) else { + return pick_dense_closed_merge(&units, files, &extents, &index, options); + }; + + let mut best: Option<(u64, u64, Range, bool)> = None; + for_each_interval(&units, &pairs, limits.max_input_files, |range, stats| { + if stats.num_files < 2 { + return; + } + let outputs = limits.predicted_outputs(stats); + if outputs > stats.num_files + 1 { + return; + } + let reduction = stats.num_files.saturating_sub(outputs); + let reduces_files = reduction > 0 + && (options.allow_file_reduction || stats.conflicts > 0) + && limits.within_reduction_cost(stats.size, reduction); + if (stats.conflicts > 0 && !options.allow_conflicts) + || (!reduces_files && stats.conflicts == 0) + { + return; + } + let benefit = (reduction + stats.conflicts) as u64; + let better = best.as_ref().is_none_or(|(best_benefit, best_size, _, _)| { + is_more_efficient(benefit, stats.size, *best_benefit, *best_size) + }); + if !better { + return; + } + let units = &units[range.clone()]; + // Overlap merges must not rewrite a dominant run for a small peer. A + // file-reducing merge is already bounded by its rewrite budget. + if !reduces_files + && !options.allow_unbalanced + && !has_balanced_components(units, files, &extents, limits.run_tolerance()) + { + return; + } + if units.len() > 1 && !adds_no_conflicts(units, &extents, &index, outputs) { + return; + } + best = Some((benefit, stats.size, range, reduces_files)); + }); + + best.map(|(_, _, range, reduces_files)| UnitPick { + inputs: collect_inputs(&units[range], files), + reduces_files, + units: units.len(), + dense_fallback: false, + }) +} + +/// Picks a bounded file reduction when the split-unit conflict graph is too +/// dense to materialize. +/// +/// The candidate comes from [`pick_reducing_interval_where`], which ignores +/// conflicts. If rejected, it is searched once more for a one-output interval. +/// Both candidates must pass [`adds_no_conflicts`], so the `(P, N)` termination +/// argument still holds. Searching only within the original candidate adds at +/// most O(K²) interval visits for K input files and one extra conflict check, +/// rather than checking external conflicts for every interval in the window. +/// This bounded retry can still miss other safely reducible intervals. +fn pick_dense_closed_merge( + units: &[Unit], + files: &[FileHandle], + extents: &[Extent], + index: &ExtentIndex, + options: ClosedMergeOptions, +) -> Option { + let limits = options.limits; + let range = pick_reducing_interval_where(units, limits, |_| true)?; + let mut selected = &units[range]; + if !is_admissible_dense_interval(selected, extents, index, options) { + let range = pick_reducing_interval_where(selected, limits, |stats| { + limits.predicted_outputs(stats) == 1 + })?; + selected = &selected[range]; + if !is_admissible_dense_interval(selected, extents, index, options) { + return None; + } + } + Some(UnitPick { + inputs: collect_inputs(selected, files), + reduces_files: true, + units: units.len(), + dense_fallback: true, + }) +} + +/// File reduction and rewrite limits are checked by the interval picker; +/// each dense candidate must also respect the closed-window conflict policy. +fn is_admissible_dense_interval( + units: &[Unit], + extents: &[Extent], + index: &ExtentIndex, + options: ClosedMergeOptions, +) -> bool { + let stats = interval_stats_exact(units, extents); + (options.allow_file_reduction || stats.conflicts > 0) + && (options.allow_conflicts || stats.conflicts == 0) + && adds_no_conflicts( + units, + extents, + index, + options.limits.predicted_outputs(&stats), + ) +} + +/// Totals of an interval without materializing [`split_conflicts`]. +/// +/// After [`close_units`], files of different units never conflict, so only +/// split units can conflict across units, and each split unit holds one file. +/// The dense fallback may instead treat every file as a split unit, and then +/// the same pairwise loop covers every cross-unit pair. Counting them takes +/// `O(K²)` for an interval of at most `K` files, instead of building a +/// window-wide adjacency. +fn interval_stats_exact<'a>(units: &'a [Unit], extents: &[Extent]) -> IntervalStats<'a> { + let mut stats = IntervalStats::default(); + let mut first = true; + for unit in units { + stats.num_files += unit.files.len(); + stats.size += unit.size; + stats.largest_file_size = stats.largest_file_size.max(unit.largest_file_size); + stats.conflicts += unit.conflicts; + stats.single_key = if first { + unit.single_key() + } else { + stats + .single_key + .filter(|&key| unit.single_key() == Some(key)) + }; + first = false; + } + + let split_files: Vec = units + .iter() + .filter(|unit| unit.split) + .map(|unit| unit.files[0]) + .collect(); + for (offset, &lhs) in split_files.iter().enumerate() { + for &rhs in &split_files[offset + 1..] { + if extents[lhs].conflicts(&extents[rhs]) { + stats.conflicts += 1; + } + } + } + stats +} + +/// Orders units by minimum primary key, then by time, so adjacent units are +/// key-adjacent. Compaction outputs and time-sliced flushes split a window by +/// key ranges, so merging neighbors in this order keeps outputs from covering +/// the key ranges of files left behind. +fn sort_units(units: &mut [Unit]) { + units.sort_by(|lhs, rhs| { + (lhs.first_key(), lhs.extent.start, Reverse(lhs.extent.end)) + .cmp(&(rhs.first_key(), rhs.extent.start, Reverse(rhs.extent.end))) + .then_with(|| lhs.files[0].cmp(&rhs.files[0])) + }); +} + +/// Whether `benefit` per `size` bytes beats the best so far; ties prefer fewer +/// bytes, then the earlier interval. +fn is_more_efficient(benefit: u64, size: u64, best_benefit: u64, best_size: u64) -> bool { + let lhs = u128::from(benefit) * u128::from(best_size.max(1)); + let rhs = u128::from(best_benefit) * u128::from(size.max(1)); + lhs > rhs || (lhs == rhs && size < best_size) +} + +/// Time and primary key bounds of a file or a group of files. +#[derive(Debug, Clone)] +pub(crate) struct Extent { + start: Timestamp, + end: Timestamp, + /// Inclusive primary key bounds. `None` is unknown and matches any key. + primary_key: Option<(Bytes, Bytes)>, +} + +impl Extent { + pub(crate) fn of_file(file: &FileHandle, mapper: &PrimaryKeyRangeMapper) -> Self { + let (start, end) = file.time_range(); + Self { + start, + end, + primary_key: file.primary_key_range(mapper), + } + } + + fn union(&self, other: &Self) -> Self { + Self { + start: self.start.min(other.start), + end: self.end.max(other.end), + primary_key: match (&self.primary_key, &other.primary_key) { + (Some((lhs_min, lhs_max)), Some((rhs_min, rhs_max))) => Some(( + lhs_min.clone().min(rhs_min.clone()), + lhs_max.clone().max(rhs_max.clone()), + )), + _ => None, + }, + } + } + + /// Time ranges overlap; touching boundaries do not. + fn time_overlaps(&self, other: &Self) -> bool { + self.start.max(other.start) < self.end.min(other.end) + } + + /// Whether two files overlap for the picker. + /// + /// Shared boundary keys count: independently written files can hold + /// different versions of the same row there. Compaction outputs do not + /// conflict with each other because the writer never splits a series. + pub(crate) fn conflicts(&self, other: &Self) -> bool { + self.time_overlaps(other) + && match (&self.primary_key, &other.primary_key) { + (Some(lhs), Some(rhs)) => { + let (low, high) = ((&lhs.0).max(&rhs.0), (&lhs.1).min(&rhs.1)); + low <= high + } + _ => true, + } + } + + /// Whether `self` may conflict with a file inside `area`, which is the + /// bounding box of a group of files. + /// + /// A shared boundary key may occur in an output, so the same inclusive + /// key overlap used for individual files also applies to the box. + fn intrudes(&self, area: &Self) -> bool { + self.conflicts(area) + } +} + +fn is_single_key(range: &(Bytes, Bytes)) -> bool { + range.0 == range.1 +} + +/// Upper primary key bound, ordered after every key when unknown. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +enum UpperKey<'a> { + Key(&'a Bytes), + Unbounded, +} + +/// Key and time bounds of the files in a subtree of [`ExtentIndex`]. +#[derive(Debug, Clone, Copy)] +struct SubtreeBounds<'a> { + max_upper_key: UpperKey<'a>, + min_start: Timestamp, + max_end: Timestamp, +} + +impl<'a> SubtreeBounds<'a> { + fn of(extent: &'a Extent) -> Self { + Self { + max_upper_key: match &extent.primary_key { + Some((_, max)) => UpperKey::Key(max), + None => UpperKey::Unbounded, + }, + min_start: extent.start, + max_end: extent.end, + } + } + + fn merge(self, other: Self) -> Self { + Self { + max_upper_key: self.max_upper_key.max(other.max_upper_key), + min_start: self.min_start.min(other.min_start), + max_end: self.max_end.max(other.max_end), + } + } +} + +/// Files of a window ordered by their lower primary key, in a segment tree +/// holding the bounds of each subtree. +/// +/// A query visits the files whose key and time ranges may overlap an area and +/// skips subtrees whose keys end before the area or whose times miss it. The +/// files of a sorted run split by key then only meet their key neighbors, +/// instead of every file sharing their time range. +struct ExtentIndex<'a> { + extents: &'a [Extent], + /// Indices into `extents`, sorted by lower key with unknown keys first. + /// Sorted offset `i` is leaf `leaf_base + i`. + by_lower_key: Vec, + /// Array-backed tree rooted at index 1; padding leaves are `None`. + bounds: Vec>>, + /// First leaf index, a power of two. + leaf_base: usize, +} + +impl<'a> ExtentIndex<'a> { + fn new(extents: &'a [Extent]) -> Self { + let lower_key = |i: usize| extents[i].primary_key.as_ref().map(|(min, _)| min); + let mut by_lower_key: Vec = (0..extents.len()).collect(); + by_lower_key.sort_by(|&lhs, &rhs| { + (lower_key(lhs), extents[lhs].start).cmp(&(lower_key(rhs), extents[rhs].start)) + }); + let leaf_base = extents.len().next_power_of_two(); + let mut bounds = vec![None; 2 * leaf_base]; + for (offset, &i) in by_lower_key.iter().enumerate() { + bounds[leaf_base + offset] = Some(SubtreeBounds::of(&extents[i])); + } + for node in (1..leaf_base).rev() { + bounds[node] = match (bounds[2 * node], bounds[2 * node + 1]) { + (Some(lhs), Some(rhs)) => Some(lhs.merge(rhs)), + (lhs, rhs) => lhs.or(rhs), + }; + } + Self { + extents, + by_lower_key, + bounds, + leaf_base, + } + } + + /// Visits the files whose inclusive key range meets `area`'s and whose time + /// range overlaps it, a superset of the files that conflict with or intrude + /// on `area`. Takes O(log N) per visited file. + fn for_each_near(&self, area: &Extent, visit: impl FnMut(usize)) { + let mut budget = usize::MAX; + let _ = self.for_each_near_with_budget(area, &mut budget, visit); + } + + /// [`for_each_near`] with a budget on visited tree nodes. + /// + /// Every node reached by the traversal consumes one unit, including nodes + /// pruned before any file callback. Returns `false` as soon as the budget is + /// exhausted, so a caller can stop a traversal whose pruning is poor even + /// when it finds few files. + fn for_each_near_with_budget( + &self, + area: &Extent, + budget: &mut usize, + mut visit: impl FnMut(usize), + ) -> bool { + let leaves_end = match &area.primary_key { + Some((_, max)) => self.by_lower_key.partition_point(|&i| { + self.extents[i] + .primary_key + .as_ref() + .is_none_or(|(min, _)| min <= max) + }), + None => self.by_lower_key.len(), + }; + self.visit_subtree(1, 0..self.leaf_base, leaves_end, area, budget, &mut visit) + } + + fn visit_subtree( + &self, + node: usize, + leaves: Range, + leaves_end: usize, + area: &Extent, + budget: &mut usize, + visit: &mut impl FnMut(usize), + ) -> bool { + if *budget == 0 { + return false; + } + *budget -= 1; + if leaves.start >= leaves_end { + return true; + } + let Some(bounds) = self.bounds[node] else { + return true; + }; + if bounds.min_start >= area.end + || bounds.max_end <= area.start + || area + .primary_key + .as_ref() + .is_some_and(|(min, _)| bounds.max_upper_key < UpperKey::Key(min)) + { + return true; + } + if leaves.len() == 1 { + let i = self.by_lower_key[leaves.start]; + if self.extents[i].time_overlaps(area) { + visit(i); + } + return true; + } + let mid = leaves.start + leaves.len() / 2; + if !self.visit_subtree(2 * node, leaves.start..mid, leaves_end, area, budget, visit) { + return false; + } + self.visit_subtree( + 2 * node + 1, + mid..leaves.end, + leaves_end, + area, + budget, + visit, + ) + } +} + +fn extents_of(files: &[FileHandle], mapper: &PrimaryKeyRangeMapper) -> Vec { + files + .iter() + .map(|file| Extent::of_file(file, mapper)) + .collect() +} + +/// Files that are compacted together. +#[derive(Debug)] +struct Unit { + /// Ascending indices into the candidate files. + files: Vec, + extent: Extent, + size: u64, + largest_file_size: u64, + /// Number of conflicting file pairs inside the unit. + conflicts: usize, + /// A single file split from a unit over the input limit. It may conflict + /// with other split files, see [`split_conflicts`]. + split: bool, +} + +impl Unit { + fn new( + members: Vec, + files: &[FileHandle], + extents: &[Extent], + conflicts: usize, + ) -> Self { + let extent = members[1..] + .iter() + .fold(extents[members[0]].clone(), |extent, &i| { + extent.union(&extents[i]) + }); + let sizes = members.iter().map(|&i| files[i].size()); + Self { + size: sizes.clone().sum(), + largest_file_size: sizes.max().unwrap_or(0), + files: members, + extent, + conflicts, + split: false, + } + } + + /// The minimum primary key, with unknown keys ordered first. + fn first_key(&self) -> Option<&Bytes> { + self.extent.primary_key.as_ref().map(|(min, _)| min) + } + + /// The primary key of a unit whose files all hold that single key. + fn single_key(&self) -> Option<&Bytes> { + self.extent + .primary_key + .as_ref() + .filter(|range| is_single_key(range)) + .map(|(min, _)| min) + } +} + +/// Groups files into connected components of [`Extent::conflicts`], each with +/// its number of conflicting pairs. Each file only checks the files the index +/// finds near it: O((N + M) log N) for M pairs of files whose key and time +/// ranges meet. +/// +/// Returns `None` once the index traversal has visited more than `work_budget` +/// nodes, so a dense window can fall back before building a quadratic graph. +fn conflict_units( + files: &[FileHandle], + extents: &[Extent], + index: &ExtentIndex, + work_budget: usize, +) -> Option> { + let mut sets = DisjointSet::new(extents.len()); + let mut conflicts = vec![0; extents.len()]; + let mut budget = work_budget; + for i in 0..extents.len() { + let completed = index.for_each_near_with_budget(&extents[i], &mut budget, |j| { + // Counts each pair once, from its smaller index. + if j > i && extents[i].conflicts(&extents[j]) { + sets.union(i, j); + conflicts[i] += 1; + } + }); + if !completed { + return None; + } + } + Some( + sets.groups() + .into_iter() + .map(|members| { + let pairs = members.iter().map(|&i| conflicts[i]).sum(); + Unit::new(members, files, extents, pairs) + }) + .collect(), + ) +} + +/// Treats every file as its own split unit, for the dense fallback that cannot +/// afford to build conflict components. +fn single_file_units(files: &[FileHandle], extents: &[Extent]) -> Vec { + (0..files.len()) + .map(|i| Unit { + split: true, + ..Unit::new(vec![i], files, extents, 0) + }) + .collect() +} + +/// Replaces units with more files than one compaction accepts by single files, +/// which are no longer closed. +fn split_oversized_units( + units: &mut Vec, + files: &[FileHandle], + extents: &[Extent], + max_input_files: usize, +) { + let (oversized, mut kept): (Vec<_>, Vec<_>) = std::mem::take(units) + .into_iter() + .partition(|unit| unit.files.len() > max_input_files); + kept.extend( + oversized + .into_iter() + .flat_map(|unit| unit.files) + .map(|i| Unit { + split: true, + ..Unit::new(vec![i], files, extents, 0) + }), + ); + *units = kept; +} + +/// Merges units until no file intrudes on a unit it does not belong to. +/// +/// Each check only visits the files the index finds near the unit, and each +/// merge re-checks only the merged unit. +fn close_units(units: &mut Vec, extents: &[Extent], index: &ExtentIndex) { + let mut owner = vec![0; extents.len()]; + for (u, unit) in units.iter().enumerate() { + for &i in &unit.files { + owner[i] = u; + } + } + let mut alive = vec![true; units.len()]; + let mut pending: Vec = (0..units.len()).collect(); + while let Some(u) = pending.pop() { + if !alive[u] { + continue; + } + let area = &units[u].extent; + let mut intruders = std::collections::BTreeSet::new(); + index.for_each_near(area, |i| { + if owner[i] != u && extents[i].intrudes(area) { + intruders.insert(owner[i]); + } + }); + if intruders.is_empty() { + continue; + } + for v in intruders { + alive[v] = false; + let absorbed = std::mem::take(&mut units[v].files); + for &i in &absorbed { + owner[i] = u; + } + let (extent, size, largest, conflicts) = { + let other = &units[v]; + ( + other.extent.clone(), + other.size, + other.largest_file_size, + other.conflicts, + ) + }; + let unit = &mut units[u]; + unit.files.extend(absorbed); + unit.extent = unit.extent.union(&extent); + unit.size += size; + unit.largest_file_size = unit.largest_file_size.max(largest); + unit.conflicts += conflicts; + } + pending.push(u); + } + let mut closed = Vec::with_capacity(units.len()); + for (u, mut unit) in std::mem::take(units).into_iter().enumerate() { + if alive[u] { + unit.files.sort_unstable(); + closed.push(unit); + } + } + *units = closed; +} + +/// Totals of a contiguous interval of units. +#[derive(Debug, Default)] +struct IntervalStats<'a> { + num_files: usize, + size: u64, + largest_file_size: u64, + conflicts: usize, + /// The primary key of an interval whose files all hold that single key. + single_key: Option<&'a Bytes>, +} + +/// Positions of the split units each unit conflicts with, sorted. Files of +/// different units never conflict, so only split files have any. +/// +/// Returns `None` once the index traversal has visited more than `work_budget` +/// nodes, so a dense window cannot materialize an unbounded adjacency. +fn split_conflicts( + units: &[Unit], + extents: &[Extent], + index: &ExtentIndex, + work_budget: usize, +) -> Option>> { + let mut position = vec![None; extents.len()]; + for (p, unit) in units.iter().enumerate() { + if unit.split { + position[unit.files[0]] = Some(p); + } + } + let mut budget = work_budget; + let mut result = Vec::with_capacity(units.len()); + for unit in units { + if !unit.split { + result.push(Vec::new()); + continue; + } + let i = unit.files[0]; + let mut neighbors = Vec::new(); + let completed = index.for_each_near_with_budget(&extents[i], &mut budget, |j| { + if let Some(p) = position[j] + && j != i + && extents[i].conflicts(&extents[j]) + { + neighbors.push(p); + } + }); + if !completed { + return None; + } + neighbors.sort_unstable(); + result.push(neighbors); + } + Some(result) +} + +/// Visits every contiguous interval of units with at most `max_input_files` +/// files: O(U · K) intervals for an input limit K. Each interval counts the +/// conflicts of its split files from `split_conflicts` (see +/// [`split_conflicts`]) with a binary search; an empty `split_conflicts` +/// skips them. +fn for_each_interval<'a>( + units: &'a [Unit], + split_conflicts: &[Vec], + max_input_files: usize, + mut visit: impl FnMut(Range, &IntervalStats<'a>), +) { + for left in 0..units.len() { + let mut stats = IntervalStats::default(); + for (right, unit) in units.iter().enumerate().skip(left) { + stats.num_files += unit.files.len(); + if stats.num_files > max_input_files { + break; + } + stats.size += unit.size; + stats.largest_file_size = stats.largest_file_size.max(unit.largest_file_size); + stats.conflicts += unit.conflicts; + stats.single_key = if right == left { + unit.single_key() + } else { + stats + .single_key + .filter(|&key| unit.single_key() == Some(key)) + }; + if let Some(neighbors) = split_conflicts.get(right) { + // Conflicts with the split files already in the interval. + stats.conflicts += neighbors.partition_point(|&p| p < right) + - neighbors.partition_point(|&p| p < left); + } + visit(left..right + 1, &stats); + } + } +} + +/// Whether merging `units` into `outputs` files adds no more conflicts than +/// their files have with other files. Outputs stay inside the bounding box of +/// the inputs, so each of them may conflict with every file intruding on that +/// box. An interval without intruders is closed and adds no conflicts. +fn adds_no_conflicts( + units: &[Unit], + extents: &[Extent], + index: &ExtentIndex, + outputs: usize, +) -> bool { + let area = units[1..] + .iter() + .fold(units[0].extent.clone(), |area, unit| { + area.union(&unit.extent) + }); + let selected: std::collections::HashSet = units + .iter() + .flat_map(|unit| unit.files.iter().copied()) + .collect(); + let mut intruders = Vec::new(); + index.for_each_near(&area, |i| { + if !selected.contains(&i) && extents[i].intrudes(&area) { + intruders.push(i); + } + }); + if intruders.is_empty() { + return true; + } + let existing: usize = intruders + .iter() + .map(|&other| { + units + .iter() + .flat_map(|unit| unit.files.iter()) + .filter(|&&i| extents[i].conflicts(&extents[other])) + .count() + }) + .sum(); + outputs.saturating_mul(intruders.len()) <= existing +} + +/// Whether every group of conflicting files among `units` is balanced, see +/// [`has_balanced_runs`]. Runs of separate groups never overlap, so pairing a +/// dominant run with the peer of another group would hide it. Takes O(K²) +/// conflict checks for K files. +fn has_balanced_components( + units: &[Unit], + files: &[FileHandle], + extents: &[Extent], + tolerance: u64, +) -> bool { + let members: Vec = units + .iter() + .flat_map(|unit| unit.files.iter().copied()) + .collect(); + let mut sets = DisjointSet::new(members.len()); + for (i, &lhs) in members.iter().enumerate() { + for (j, &rhs) in members.iter().enumerate().skip(i + 1) { + if extents[lhs].conflicts(&extents[rhs]) { + sets.union(i, j); + } + } + } + sets.groups() + .into_iter() + .filter(|group| group.len() > 1) + .all(|group| { + has_balanced_runs( + group.into_iter().map(|i| members[i]), + files, + extents, + tolerance, + ) + }) +} + +/// Whether the largest sorted run among `files` holds at most `tolerance` bytes +/// more than all other runs together. Runs are built greedily in time order, +/// which takes O(K²) conflict checks for K files. +fn has_balanced_runs( + members: impl Iterator, + files: &[FileHandle], + extents: &[Extent], + tolerance: u64, +) -> bool { + let mut members: Vec = members.collect(); + members.sort_by(|&lhs, &rhs| { + (extents[lhs].start, Reverse(extents[lhs].end), lhs).cmp(&( + extents[rhs].start, + Reverse(extents[rhs].end), + rhs, + )) + }); + let mut runs: Vec<(Vec, u64)> = Vec::new(); + for i in members { + let size = files[i].size(); + match runs.iter_mut().find(|(run, _)| { + run.iter() + .all(|&other| !extents[i].conflicts(&extents[other])) + }) { + Some((run, run_size)) => { + run.push(i); + *run_size += size; + } + None => runs.push((vec![i], size)), + } + } + let total: u64 = runs.iter().map(|(_, size)| size).sum(); + let largest = runs.iter().map(|(_, size)| *size).max().unwrap_or(0); + largest <= (total - largest).saturating_add(tolerance) +} + +fn collect_inputs(units: &[Unit], files: &[FileHandle]) -> Vec { + let mut indices: Vec = units + .iter() + .flat_map(|unit| unit.files.iter().copied()) + .collect(); + indices.sort_unstable(); + indices.into_iter().map(|i| files[i].clone()).collect() +} + +/// Union-find over file indices; each root is the smallest index of its set. +struct DisjointSet { + parent: Vec, +} + +impl DisjointSet { + fn new(len: usize) -> Self { + Self { + parent: (0..len).collect(), + } + } + + fn find(&mut self, mut i: usize) -> usize { + while self.parent[i] != i { + self.parent[i] = self.parent[self.parent[i]]; + i = self.parent[i]; + } + i + } + + fn union(&mut self, lhs: usize, rhs: usize) { + let (lhs, rhs) = (self.find(lhs), self.find(rhs)); + let (root, child) = (lhs.min(rhs), lhs.max(rhs)); + self.parent[child] = root; + } + + /// Sets in ascending order of their smallest index, each sorted. + fn groups(&mut self) -> Vec> { + let mut groups: Vec> = Vec::new(); + let mut slot = vec![usize::MAX; self.parent.len()]; + for i in 0..self.parent.len() { + let root = self.find(i); + if slot[root] == usize::MAX { + slot[root] = groups.len(); + groups.push(Vec::new()); + } + groups[slot[root]].push(i); + } + groups + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + + use store_api::storage::FileId; + + use super::*; + use crate::compaction::test_util::{ + new_file_handle_with_size_sequence_and_primary_key_range, pk_range, + primary_key_mapper_for_test, + }; + + const MAX_OUTPUT: u64 = 16; + + fn file(start: i64, end: i64, min: &[u8], max: &[u8], size: u64) -> FileHandle { + new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + start, + end, + 1, + 1, + size, + pk_range(min, max), + ) + } + + /// Files that all overlap each other in time and primary key. + fn dense_files(count: usize) -> Vec { + (0..count) + .map(|idx| file(0, 100, b"a", b"z", 1 + (idx % 3) as u64)) + .collect() + } + + /// A chain of files whose key ranges overlap their immediate neighbors. + fn chain_files(count: usize, size: u64) -> Vec { + (0..count) + .map(|idx| { + let min = format!("{idx:04}"); + let max = format!("{:04}", idx + 2); + file(0, 100, min.as_bytes(), max.as_bytes(), size) + }) + .collect() + } + + fn limits(max_output_file_size: u64) -> PickLimits { + PickLimits { + max_output_file_size: Some(max_output_file_size), + max_input_files: 16, + } + } + + fn closed_merge(files: &[FileHandle], max_output_file_size: u64) -> Option { + pick_closed_merge( + files, + &primary_key_mapper_for_test(), + closed_merge_options(max_output_file_size), + ) + } + + fn closed_merge_with_budget( + files: &[FileHandle], + max_output_file_size: u64, + work_budget: usize, + ) -> Option { + pick_closed_merge_with_budget( + files, + &primary_key_mapper_for_test(), + closed_merge_options(max_output_file_size), + work_budget, + ) + } + + fn closed_merge_options(max_output_file_size: u64) -> ClosedMergeOptions { + ClosedMergeOptions { + limits: limits(max_output_file_size), + allow_file_reduction: true, + allow_conflicts: true, + allow_unbalanced: false, + } + } + + #[test] + fn test_small_l1_reduction_limits() { + let mapper = primary_key_mapper_for_test(); + // Empty/single inputs, the small-file boundary, balance, byte and count + // caps, and an unlimited output size all affect eligibility. + for (sizes, max_output, max_inputs, expected) in [ + (vec![], Some(16), 16, 0), + (vec![4], Some(16), 16, 0), + (vec![4, 4], Some(16), 16, 2), + (vec![5, 5], Some(16), 16, 0), + (vec![4, 1], Some(16), 16, 0), + (vec![4, 2, 2], Some(16), 16, 3), + (vec![4; 8], Some(16), 16, 4), + (vec![4; 8], Some(16), 3, 3), + (vec![4, 4], Some(16), 1, 0), + (vec![4, 4], Some(0), 16, 0), + (vec![4, 4], None, 16, 0), + ] { + let files = sizes + .iter() + .map(|&size| file(0, 100, b"a", b"z", size)) + .collect::>(); + let pick = pick_small_l1( + &files, + &mapper, + PickLimits { + max_output_file_size: max_output, + max_input_files: max_inputs, + }, + ); + assert_eq!( + expected, + pick.map_or(0, |pick| pick.inputs.len()), + "sizes={sizes:?}, target={max_output:?}, cap={max_inputs}" + ); + } + } + + #[test] + fn test_small_l1_checks_unselected_large_and_l0_files() { + let mapper = primary_key_mapper_for_test(); + let small = [file(0, 10, b"a", b"b", 4), file(90, 100, b"y", b"z", 4)]; + for level in [0, 1] { + for (min, max, expected) in [(b"m", b"n", false), (b"a", b"z", true)] { + let peer = new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + 0, + 100, + level, + 1, + 100, + pk_range(min, max), + ); + let files = [small.to_vec(), vec![peer]].concat(); + let pick = pick_small_l1(&files, &mapper, limits(16)); + assert_eq!( + expected, + pick.is_some(), + "level={level}, range={min:?}..{max:?}" + ); + if let Some(pick) = pick { + assert_eq!(ids(&small), ids(&pick.inputs)); + } + } + } + // A small L0 must not be selected as the second member of an L1 merge. + let l0 = new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + 90, + 100, + 0, + 1, + 4, + pk_range(b"y", b"z"), + ); + assert!(pick_small_l1(&[small[0].clone(), l0], &mapper, limits(16)).is_none()); + } + + fn ids(files: &[FileHandle]) -> HashSet { + files.iter().map(|file| file.file_id().file_id()).collect() + } + + fn extent(start: i64, end: i64, min: &[u8], max: &[u8]) -> Extent { + Extent::of_file( + &file(start, end, min, max, 1), + &primary_key_mapper_for_test(), + ) + } + + #[test] + fn test_conflicts_include_shared_boundary_keys() { + // Independently written files may share rows at their boundary keys. + assert!(extent(0, 10, b"a", b"m").conflicts(&extent(0, 10, b"m", b"z"))); + assert!(extent(0, 10, b"m", b"m").conflicts(&extent(0, 10, b"m", b"z"))); + // A writer-produced run has strictly disjoint key ranges. + assert!(!extent(0, 10, b"a", b"l").conflicts(&extent(0, 10, b"m", b"z"))); + // A single key shared by two single-key files is a real overlap. + assert!(extent(0, 10, b"m", b"m").conflicts(&extent(5, 15, b"m", b"m"))); + // So is a single key inside another range. + assert!(extent(0, 10, b"m", b"m").conflicts(&extent(0, 10, b"a", b"z"))); + assert!(extent(0, 10, b"a", b"z").conflicts(&extent(0, 10, b"m", b"m"))); + assert!(extent(0, 10, b"a", b"n").conflicts(&extent(0, 10, b"m", b"z"))); + // Touching time ranges do not overlap. + assert!(!extent(0, 10, b"a", b"z").conflicts(&extent(10, 20, b"a", b"z"))); + } + + #[test] + fn test_intrusion_covers_every_conflict_inside_an_area() { + let area = extent(0, 10, b"m", b"z"); + // Any file sharing the area's boundary key may conflict with an output. + assert!(extent(0, 10, b"m", b"m").intrudes(&area)); + assert!(extent(0, 10, b"a", b"m").intrudes(&area)); + assert!(!extent(0, 10, b"a", b"l").intrudes(&area)); + assert!(extent(0, 10, b"a", b"n").intrudes(&area)); + assert!(!extent(10, 20, b"a", b"z").intrudes(&area)); + // A range may conflict with a single-key file inside a single-key area. + let point = extent(0, 10, b"m", b"m"); + assert!(extent(0, 10, b"a", b"z").intrudes(&point)); + assert!(extent(0, 10, b"a", b"m").intrudes(&point)); + } + + #[test] + fn test_intrusion_covers_conflicts_of_random_files() { + let mut rng = Rng(23); + let random_extent = |rng: &mut Rng| { + let start = rng.next(10) as i64; + let end = start + 1 + rng.next(10) as i64; + let min = rng.next(6) as u8; + let max = min + rng.next(3) as u8; + extent(start, end, &[b'0' + min], &[b'0' + max]) + }; + for _ in 0..2_000 { + let group: Vec<_> = (0..1 + rng.next(3)) + .map(|_| random_extent(&mut rng)) + .collect(); + let area = group[1..] + .iter() + .fold(group[0].clone(), |area, extent| area.union(extent)); + let other = random_extent(&mut rng); + if group.iter().any(|extent| other.conflicts(extent)) { + assert!(other.intrudes(&area), "{other:?} conflicts inside {area:?}"); + } + } + } + + #[test] + fn test_extent_index_budget_counts_nodes() { + // The root's time bounds straddle the query, but neither leaf overlaps + // it. A callback-only budget would never fire; the node budget must. + let files = vec![file(0, 10, b"a", b"z", 1), file(90, 100, b"a", b"z", 1)]; + let extents = extents_of(&files, &primary_key_mapper_for_test()); + let index = ExtentIndex::new(&extents); + let query = extent(50, 51, b"a", b"z"); + let mut matched = 0; + + let mut budget = 2; + assert!(!index.for_each_near_with_budget(&query, &mut budget, |_| matched += 1)); + assert_eq!(0, budget); + assert_eq!(0, matched); + + let mut budget = 3; + assert!(index.for_each_near_with_budget(&query, &mut budget, |_| matched += 1)); + assert_eq!(0, budget); + assert_eq!(0, matched); + } + + #[test] + fn test_extent_index_matches_linear_scan() { + let mut rng = Rng(29); + let mapper = primary_key_mapper_for_test(); + for count in [0, 1, 2, 7, 64, 200] { + let extents: Vec<_> = (0..count) + .map(|_| { + let start = rng.next(100) as i64; + let end = start + rng.next(30) as i64; + let min = rng.next(50) as u8; + let max = min + rng.next(10) as u8; + let range = if rng.next(10) > 0 { + pk_range(&[b'0' + min], &[b'0' + max]) + } else { + None + }; + let file = new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + start, + end, + 1, + 1, + 1, + range, + ); + Extent::of_file(&file, &mapper) + }) + .collect(); + let index = ExtentIndex::new(&extents); + for area in &extents { + let mut near = Vec::new(); + index.for_each_near(area, |i| near.push(i)); + near.sort_unstable(); + let expected: Vec<_> = (0..extents.len()) + .filter(|&i| { + let other = &extents[i]; + let keys_meet = match (&other.primary_key, &area.primary_key) { + (Some(lhs), Some(rhs)) => lhs.0 <= rhs.1 && rhs.0 <= lhs.1, + _ => true, + }; + keys_meet && other.time_overlaps(area) + }) + .collect(); + assert_eq!(expected, near, "{area:?}"); + for (i, other) in extents.iter().enumerate() { + if other.conflicts(area) || other.intrudes(area) { + assert!(near.binary_search(&i).is_ok(), "{other:?} near {area:?}"); + } + } + } + } + } + + #[test] + fn test_active_l1_packs_pk_disjoint_files() { + let files = vec![ + file(0, 1000, b"a", b"a", 100), + file(0, 1000, b"z", b"z", 100), + ]; + let pick = pick_active_l1(&files, &primary_key_mapper_for_test(), limits(512)).unwrap(); + assert_eq!(ids(&files), ids(&pick.inputs)); + } + + #[test] + fn test_active_l1_bounds_rewrite_cost() { + let overlapping = |sizes: &[u64]| { + sizes + .iter() + .enumerate() + .map(|(idx, &size)| file(idx as i64, 100, b"a", b"z", size)) + .collect::>() + }; + let mapper = primary_key_mapper_for_test(); + assert!(pick_active_l1(&overlapping(&[300, 300, 300]), &mapper, limits(512)).is_some()); + // Removing one file would rewrite 4600 bytes, beyond 4 * 512. + let costly = overlapping(&[500, 500, 500, 500, 500, 500, 500, 500, 500, 100]); + assert!(pick_active_l1(&costly, &mapper, limits(512)).is_none()); + } + + #[test] + fn test_active_l1_merges_small_files_beside_dominant_file() { + // All files overlap. Merging the group would rewrite the large file for + // its small peers, so only the small files are merged. + let mut files = vec![file(0, 1000, b"a", b"z", 1024)]; + files.extend((1..16).map(|start| file(start, 1000, b"a", b"z", 1))); + let pick = pick_active_l1(&files, &primary_key_mapper_for_test(), limits(512)).unwrap(); + assert_eq!(ids(&files[1..]), ids(&pick.inputs)); + } + + #[test] + fn test_merges_of_a_single_key_write_one_file() { + // The writer never splits a series, so the overlapping files of one key + // compact into a single file however large they are. + let files = vec![file(0, 100, b"m", b"m", 32), file(50, 150, b"m", b"m", 32)]; + let mapper = primary_key_mapper_for_test(); + let pick = closed_merge(&files, MAX_OUTPUT).unwrap(); + assert_eq!(ids(&files), ids(&pick.inputs)); + assert!(pick.reduces_files); + let pick = pick_active_l1(&files, &mapper, limits(MAX_OUTPUT)).unwrap(); + assert_eq!(ids(&files), ids(&pick.inputs)); + } + + #[test] + fn test_closed_merge_does_not_enclose_other_files() { + let a = file(0, 10, b"a", b"c", 10); + let b = file(0, 10, b"x", b"z", 10); + let c = file(0, 10, b"m", b"m", 1); + let files = vec![a.clone(), b, c.clone()]; + let pick = closed_merge(&files, MAX_OUTPUT).unwrap(); + // Merging A and B alone would produce a file covering C. + assert_eq!(ids(&[a, c]), ids(&pick.inputs)); + assert!(pick.reduces_files); + } + + #[test] + fn test_shared_boundary_files_are_picked_without_file_reduction() { + // Independently written files can hold different versions at m. Their + // total size predicts two outputs, so only overlap resolution helps. + let size = 512 * 1024 * 1024; + let files = [ + file(0, 100, b"a", b"m", size), + file(0, 100, b"m", b"z", size), + ]; + let pick = closed_merge(&files, size).unwrap(); + assert_eq!(ids(&files), ids(&pick.inputs)); + assert!(!pick.reduces_files); + } + + /// A sorted run of `sizes.len()` key-sliced files, and files spanning all + /// of its keys in the given time ranges, which overlap the whole run. + fn run_with_late_files( + sizes: &[u64], + late: &[(i64, i64)], + ) -> (Vec, Vec) { + let run = sizes + .iter() + .enumerate() + .map(|(idx, &size)| { + let idx = 2 * idx as u8; + file(0, 1000, &[b'a', idx], &[b'a', idx + 1], size) + }) + .collect::>(); + let max_key = 2 * sizes.len() as u8; + let late = late + .iter() + .map(|&(start, end)| file(start, end, &[b'a', 0], &[b'a', max_key], 1)) + .collect(); + (run, late) + } + + #[test] + fn test_closed_merge_splits_components_over_input_limit() { + // The late file joins all 20 run files into one component, over the + // input limit. Small run files are still merged. + let (run, late) = run_with_late_files(&[1; 20], &[(100, 900)]); + let files = [run, late].concat(); + let pick = closed_merge(&files, MAX_OUTPUT).unwrap(); + assert!(pick.reduces_files); + assert!(pick.inputs.len() <= 16); + } + + #[test] + fn test_closed_merge_leaves_late_file_over_large_run() { + // Resolving the overlap would rewrite the whole run, beyond the input + // limit, and merging part of it would only spread the overlap. + let (run, late) = run_with_late_files(&[MAX_OUTPUT; 20], &[(100, 900)]); + let files = [run, late].concat(); + assert!(closed_merge(&files, MAX_OUTPUT).is_none()); + } + + #[test] + fn test_closed_merge_compacts_late_files_over_large_run() { + let (run, late) = + run_with_late_files(&[MAX_OUTPUT; 20], &[(100, 200), (300, 400), (500, 600)]); + let files = [run, late.clone()].concat(); + let pick = closed_merge(&files, MAX_OUTPUT).unwrap(); + assert_eq!(ids(&late), ids(&pick.inputs)); + assert!(pick.reduces_files); + } + + #[test] + fn test_closed_merge_rejects_dominant_runs() { + // Each of these components has a 3 GiB run and a 512 MiB peer. + let size = 512 * 1024 * 1024; + // The peer spans the keys of its run, written before or after it. + let mut files = (0..6u8) + .map(|idx| file(idx.into(), 10_000, &[b'b' + idx], &[b'b' + idx], size)) + .collect::>(); + files.push(file(6, 10_000, b"a", b"h", size)); + files.extend((0..6u8).map(|idx| { + file( + 8 + i64::from(idx), + 10_000, + &[b'n' + idx], + &[b'n' + idx], + size, + ) + })); + files.push(file(7, 10_000, b"m", b"t", size)); + // A separate balanced pair is still resolved. + let pair = [ + file(14, 10_000, b"x", b"y", size), + file(15, 10_000, b"x", b"y", size), + ]; + files.extend(pair.clone()); + + let pick = closed_merge(&files, size).unwrap(); + assert_eq!(ids(&pair), ids(&pick.inputs)); + assert!(!pick.reduces_files); + } + + #[test] + fn test_closed_merge_ignores_touching_time_ranges() { + let first = file(0, 10, b"a", b"z", 1024); + let pair = [file(10, 20, b"a", b"z", 512), file(11, 19, b"a", b"z", 512)]; + let mut files = vec![first]; + files.extend(pair.clone()); + let pick = closed_merge(&files, 512).unwrap(); + assert_eq!(ids(&pair), ids(&pick.inputs)); + } + + #[test] + fn test_conflict_units_respects_work_budget() { + let files = dense_files(20); + let mapper = primary_key_mapper_for_test(); + let extents = extents_of(&files, &mapper); + let index = ExtentIndex::new(&extents); + + assert!(conflict_units(&files, &extents, &index, 0).is_none()); + assert!(conflict_units(&files, &extents, &index, usize::MAX).is_some()); + } + + #[test] + fn test_split_conflicts_respects_work_budget() { + let files = dense_files(20); + let mapper = primary_key_mapper_for_test(); + let extents = extents_of(&files, &mapper); + let index = ExtentIndex::new(&extents); + let mut units = conflict_units(&files, &extents, &index, usize::MAX).unwrap(); + close_units(&mut units, &extents, &index); + split_oversized_units(&mut units, &files, &extents, 16); + sort_units(&mut units); + + assert_eq!(20, units.iter().filter(|unit| unit.split).count()); + assert!(split_conflicts(&units, &extents, &index, 0).is_none()); + assert!(split_conflicts(&units, &extents, &index, usize::MAX).is_some()); + } + + #[test] + fn test_interval_stats_exact_matches_for_each_interval() { + let mapper = primary_key_mapper_for_test(); + // One closed unit with internal conflicts, and one oversized component + // split into single-file units with cross-unit conflicts. + let mut files = (0..10) + .map(|_| file(0, 100, b"a", b"c", 1)) + .collect::>(); + files.extend((0..20).map(|_| file(0, 100, b"x", b"z", 1))); + let extents = extents_of(&files, &mapper); + let index = ExtentIndex::new(&extents); + let mut units = conflict_units(&files, &extents, &index, usize::MAX).unwrap(); + close_units(&mut units, &extents, &index); + split_oversized_units(&mut units, &files, &extents, 16); + sort_units(&mut units); + + let pairs = split_conflicts(&units, &extents, &index, usize::MAX).unwrap(); + for_each_interval(&units, &pairs, 16, |range, stats| { + let exact = interval_stats_exact(&units[range], &extents); + assert_eq!(stats.num_files, exact.num_files); + assert_eq!(stats.size, exact.size); + assert_eq!(stats.largest_file_size, exact.largest_file_size); + assert_eq!(stats.conflicts, exact.conflicts); + assert_eq!(stats.single_key, exact.single_key); + }); + } + + #[test] + fn test_active_l1_falls_back_on_dense_conflict_graph() { + let files = dense_files(300); + let pick = pick_active_l1_with_budget( + &files, + &primary_key_mapper_for_test(), + limits(MAX_OUTPUT), + 1 << 10, + ) + .unwrap(); + assert!(!pick.dense_fallback); + assert!(pick.reduces_files); + assert!((2..=16).contains(&pick.inputs.len())); + } + + #[test] + fn test_dense_closed_merge_reduces_files() { + let files = dense_files(300); + let pick = closed_merge_with_budget(&files, MAX_OUTPUT, 1 << 10).unwrap(); + assert!(pick.dense_fallback); + assert!(pick.reduces_files); + assert!((2..=16).contains(&pick.inputs.len())); + } + + #[test] + fn test_dense_closed_merge_resolves_conflicts_without_file_reduction_setting() { + let files = dense_files(300); + let mapper = primary_key_mapper_for_test(); + let options = ClosedMergeOptions { + allow_file_reduction: false, + ..closed_merge_options(MAX_OUTPUT) + }; + let pick = pick_closed_merge_with_budget(&files, &mapper, options, 1 << 10).unwrap(); + assert!(pick.dense_fallback); + assert!(pick.reduces_files); + } + + #[test] + fn test_dense_closed_merge_defers_when_conflicts_are_not_allowed() { + let files = dense_files(300); + let mapper = primary_key_mapper_for_test(); + let options = ClosedMergeOptions { + allow_conflicts: false, + ..closed_merge_options(MAX_OUTPUT) + }; + assert!(pick_closed_merge_with_budget(&files, &mapper, options, 1 << 10).is_none()); + } + + #[test] + fn test_dense_closed_merge_retries_one_output_interval() { + // PR #9741: the first 16 files predict two outputs but have only one + // conflict with the rest of the chain. The first eight predict one + // output, so rejecting the larger interval must not prevent progress. + let files = chain_files(300, 2); + let mapper = primary_key_mapper_for_test(); + let extents = extents_of(&files, &mapper); + let index = ExtentIndex::new(&extents); + let mut units = conflict_units(&files, &extents, &index, usize::MAX).unwrap(); + close_units(&mut units, &extents, &index); + split_oversized_units(&mut units, &files, &extents, 16); + sort_units(&mut units); + + assert!(!adds_no_conflicts(&units[..16], &extents, &index, 2)); + assert!(adds_no_conflicts(&units[..8], &extents, &index, 1)); + // Cover both single-file units from an exhausted grouping budget and + // split units from an oversized closure. + for pick in [ + closed_merge_with_budget(&files, MAX_OUTPUT, 0).unwrap(), + pick_dense_closed_merge( + &units, + &files, + &extents, + &index, + closed_merge_options(MAX_OUTPUT), + ) + .unwrap(), + ] { + assert_eq!(ids(&files[..8]), ids(&pick.inputs)); + assert!(pick.dense_fallback); + assert!(pick.reduces_files); + assert_eq!(300, pick.units); + } + } + + #[test] + fn test_dense_closed_merge_rejects_unsafe_one_output_retry() { + // Intruders lie in the time gaps between inputs, so even a one-output + // merge can introduce conflicts that no input previously had. + let mut files: Vec<_> = (0..4) + .map(|i| file(30 * i, 30 * i + 10, b"a", b"z", 6)) + .collect(); + files.extend((0..3).map(|i| file(30 * i + 15, 30 * i + 25, b"b", b"y", 1_000))); + + assert!(closed_merge_with_budget(&files, MAX_OUTPUT, 0).is_none()); + } + + /// SplitMix64, so the simulation needs no random number dependency. + struct Rng(u64); + + impl Rng { + fn next(&mut self, bound: u64) -> u64 { + self.0 = self.0.wrapping_add(0x9e37_79b9_7f4a_7c15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + (z ^ (z >> 31)) % bound + } + } + + /// A simulated file with its key bounds as numbers. + #[derive(Clone)] + struct SimFile { + handle: FileHandle, + min: u64, + max: u64, + } + + fn sim_file(start: i64, end: i64, min: u64, max: u64, size: u64) -> SimFile { + let (min_key, max_key) = (format!("{min:04}"), format!("{max:04}")); + SimFile { + handle: file(start, end, min_key.as_bytes(), max_key.as_bytes(), size), + min, + max, + } + } + + /// Writes the outputs of merging `inputs` like the compactor: one stream in + /// (key, time) order, split by size into consecutive files. The data is + /// spread evenly over keys, and a series is never split between files. + fn merge(inputs: &[SimFile], rng: &mut Rng) -> Vec { + let min = inputs.iter().map(|file| file.min).min().unwrap(); + let max = inputs.iter().map(|file| file.max).max().unwrap(); + let start = inputs + .iter() + .map(|file| file.handle.time_range().0.value()) + .min() + .unwrap(); + let end = inputs + .iter() + .map(|file| file.handle.time_range().1.value()) + .max() + .unwrap(); + let size: u64 = inputs.iter().map(|file| file.handle.size()).sum(); + // Outputs may be fewer than predicted, never more. A single key is + // never split. + let predicted = if min == max { + 1 + } else { + size.div_ceil(MAX_OUTPUT).max(1) + }; + let keys = max - min + 1; + let outputs = (predicted - rng.next(2).min(predicted - 1)).min(keys); + (0..outputs) + .map(|i| { + let first_key = min + keys * i / outputs; + let last_key = min + keys * (i + 1) / outputs - 1; + let output_size = size * (i + 1) / outputs - size * i / outputs; + sim_file(start, end, first_key, last_key, output_size) + }) + .collect() + } + + fn conflict_pairs(files: &[FileHandle]) -> usize { + let extents = extents_of(files, &primary_key_mapper_for_test()); + (0..extents.len()) + .map(|i| { + (i + 1..extents.len()) + .filter(|&j| extents[i].conflicts(&extents[j])) + .count() + }) + .sum() + } + + fn random_window(rng: &mut Rng) -> Vec { + (0..2 + rng.next(30)) + .map(|_| { + let start = rng.next(3_000) as i64; + let end = start + 1 + rng.next(600) as i64; + let min = rng.next(200); + let span = if rng.next(4) == 0 { 200 } else { 20 }; + let max = min + rng.next(span); + sim_file(start, end, min, max, 1 + rng.next(2 * MAX_OUTPUT)) + }) + .collect() + } + + /// A sorted run of more files than one compaction accepts, and a few files + /// spanning many of its keys, which join them into oversized components. + fn oversized_window(rng: &mut Rng) -> Vec { + let run_files = 17 + rng.next(40); + let mut files: Vec<_> = (0..run_files) + .map(|idx| { + let start = rng.next(100) as i64; + let end = 2_900 + rng.next(100) as i64; + sim_file( + start, + end, + 5 * idx, + 5 * (idx + 1) - 1, + 1 + rng.next(2 * MAX_OUTPUT), + ) + }) + .collect(); + files.extend((0..1 + rng.next(6)).map(|_| { + let start = rng.next(2_800) as i64; + let end = start + 1 + rng.next(200) as i64; + let min = rng.next(5 * run_files / 2); + let max = min + 5 * run_files / 2 + rng.next(5 * run_files / 2); + sim_file(start, end, min, max, 1 + rng.next(MAX_OUTPUT)) + })); + files + } + + #[test] + fn test_dense_closed_merge_terminates() { + let mut rng = Rng(23); + let mut files: Vec<_> = (0..300) + .map(|_| sim_file(0, 100, 0, 8, 1 + rng.next(2 * MAX_OUTPUT))) + .collect(); + let handles = |files: &[SimFile]| { + files + .iter() + .map(|file| file.handle.clone()) + .collect::>() + }; + let mut current = (conflict_pairs(&handles(&files)), files.len()); + let mut steps = 0; + let mut dense_steps = 0; + while let Some(pick) = closed_merge_with_budget(&handles(&files), MAX_OUTPUT, 1 << 10) { + assert!(pick.inputs.len() >= 2); + if pick.dense_fallback { + dense_steps += 1; + } + let picked = ids(&pick.inputs); + let (inputs, kept): (Vec<_>, Vec<_>) = files + .into_iter() + .partition(|file| picked.contains(&file.handle.file_id().file_id())); + files = kept; + files.extend(merge(&inputs, &mut rng)); + let next = (conflict_pairs(&handles(&files)), files.len()); + assert!(next < current, "step {steps}: {current:?} -> {next:?}"); + current = next; + steps += 1; + assert!(steps < 1_000); + } + assert!(dense_steps > 0); + assert!(files.len() < 300); + } + + /// Applies `pick` until it picks nothing, asserting that `measure` strictly + /// decreases with every merge. Returns the number of merges. + fn run_to_fixpoint( + rng: &mut Rng, + pick: impl Fn(&[FileHandle]) -> Option, + measure: impl Fn(&[FileHandle]) -> (usize, usize), + ) -> usize { + run_windows_to_fixpoint(rng, random_window, pick, measure) + } + + fn run_windows_to_fixpoint( + rng: &mut Rng, + new_window: impl Fn(&mut Rng) -> Vec, + pick: impl Fn(&[FileHandle]) -> Option, + measure: impl Fn(&[FileHandle]) -> (usize, usize), + ) -> usize { + let mut total_steps = 0; + for window in 0..300 { + let mut files = new_window(rng); + let handles = + |files: &[SimFile]| files.iter().map(|f| f.handle.clone()).collect::>(); + let mut current = measure(&handles(&files)); + let mut steps = 0; + while let Some(pick) = pick(&handles(&files)) { + assert!(pick.inputs.len() >= 2); + let picked = ids(&pick.inputs); + let (inputs, kept): (Vec<_>, Vec<_>) = files + .into_iter() + .partition(|file| picked.contains(&file.handle.file_id().file_id())); + files = kept; + files.extend(merge(&inputs, rng)); + let next = measure(&handles(&files)); + assert!(next < current, "window {window}: {current:?} -> {next:?}"); + current = next; + steps += 1; + } + total_steps += steps; + } + total_steps + } + + #[test] + fn test_closed_merges_terminate() { + let steps = run_to_fixpoint( + &mut Rng(7), + |files| closed_merge(files, MAX_OUTPUT), + |files| (conflict_pairs(files), files.len()), + ); + assert!(steps > 300); + } + + #[test] + fn test_closed_merges_of_oversized_components_terminate() { + let steps = run_windows_to_fixpoint( + &mut Rng(19), + oversized_window, + |files| closed_merge(files, MAX_OUTPUT), + |files| (conflict_pairs(files), files.len()), + ); + assert!(steps > 300); + } + + #[test] + fn test_closed_merges_of_quiet_and_settling_windows_terminate() { + let mapper = primary_key_mapper_for_test(); + // Quiet windows take merges a default window refuses; settling windows + // skip the conflicting ones. + for (seed, options, more_merges) in [ + ( + 13, + ClosedMergeOptions { + allow_unbalanced: true, + ..closed_merge_options(MAX_OUTPUT) + }, + true, + ), + ( + 17, + ClosedMergeOptions { + allow_conflicts: false, + ..closed_merge_options(MAX_OUTPUT) + }, + false, + ), + ] { + let steps = run_to_fixpoint( + &mut Rng(seed), + |files| pick_closed_merge(files, &mapper, options), + |files| (conflict_pairs(files), files.len()), + ); + let base = run_to_fixpoint( + &mut Rng(seed), + |files| closed_merge(files, MAX_OUTPUT), + |files| (conflict_pairs(files), files.len()), + ); + assert_eq!(more_merges, steps > base, "seed {seed}: {steps} vs {base}"); + } + } + + #[test] + fn test_small_l1_merges_terminate() { + let mapper = primary_key_mapper_for_test(); + let steps = run_windows_to_fixpoint( + &mut Rng(31), + |rng| { + let mut files = random_window(rng); + // Include a reducible cluster inside a large peer as well as + // random external intruders to exercise the conflict guard. + files.push(sim_file(0, 3000, 0, 400, 64)); + files.extend((0..8).map(|_| sim_file(0, 100, 0, 8, 1 + rng.next(4)))); + files + }, + |files| pick_small_l1(files, &mapper, limits(MAX_OUTPUT)), + |files| (conflict_pairs(files), files.len()), + ); + assert!(steps >= 300); + } + + #[test] + fn test_active_l1_picks_terminate() { + let mapper = primary_key_mapper_for_test(); + let steps = run_to_fixpoint( + &mut Rng(11), + |files| pick_active_l1(files, &mapper, limits(MAX_OUTPUT)), + |files| (files.len(), 0), + ); + assert!(steps > 200); + } +} diff --git a/src/mito2/src/engine/compaction_test.rs b/src/mito2/src/engine/compaction_test.rs index 82f8bcaecc8..02f76af6f5f 100644 --- a/src/mito2/src/engine/compaction_test.rs +++ b/src/mito2/src/engine/compaction_test.rs @@ -212,6 +212,7 @@ async fn test_strict_window_output_file_size( assert!(version.ssts.levels()[0].files.is_empty()); let primary_key_mapper = version.ssts.primary_key_mapper(); let files = version.ssts.levels()[1].files().collect::>(); + let max_output_size = files.iter().map(|file| file.size()).max().unwrap(); assert_eq!( 6000, files @@ -229,6 +230,14 @@ async fn test_strict_window_output_file_size( .collect::>(); if expect_split { assert!(window_files.len() > 1, "each window must split by size"); + assert_eq!( + window_files.len() as u64, + window_files + .iter() + .map(|file| file.size()) + .sum::() + .div_ceil(max_output_size) + ); } else { assert_eq!(1, window_files.len(), "zero must disable size splitting"); } @@ -253,6 +262,35 @@ async fn test_strict_window_output_file_size( sort_batches_and_print(&before, &["tag_0", "ts"]), sort_batches_and_print(&after, &["tag_0", "ts"]) ); + + // Fresh size-split outputs have disjoint series and must not look like + // independently written files sharing a boundary key to the TWCS picker. + // Raise the tiny split threshold so an overlap merge would be eligible by + // size; otherwise the output-count guard alone would reject these files. + engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: SetRegionOptions { + options: vec![SetRegionOption::Twsc( + "compaction.twcs.max_output_file_size".into(), + format!("{max_output_size}B"), + )], + }, + }), + ) + .await + .unwrap(); + let output_ids: HashSet<_> = files.iter().map(|file| file.file_id()).collect(); + compact(&engine, region_id).await; + let version = engine.get_region(region_id).unwrap().version(); + assert_eq!( + output_ids, + version.ssts.levels()[1] + .files() + .map(|file| file.file_id()) + .collect() + ); } /// Flush may collapse versions within one file, but compaction must not promote @@ -544,6 +582,169 @@ async fn assert_partial_compaction_preserves_delete_order(flat_format: bool) { ); } +/// Exercise the small-L1 fallback with real SST sizes, preserved tombstones, +/// and both layouts. No fabricated file metadata or hand-picked merge inputs. +#[rstest::rstest] +#[tokio::test] +async fn test_inactive_small_l1_compaction_preserves_large_peer( + #[values(false, true)] flat_format: bool, + #[values(3600, 7200)] active_ts: usize, +) { + let mut env = TestEnv::new().await; + let region_id = RegionId::new(1, 1); + let (engine, columns) = env_for_manual_compaction(&mut env, region_id, flat_format).await; + let mut rng = ::seed_from_u64(41); + let mut old_rows = build_rows_for_key("m", 10, 16, 0); + for _ in 0..30_000 { + old_rows.extend(build_rows_for_key( + &format!("a{:032x}", rand::Rng::random::(&mut rng)), + 1, + 2, + 0, + )); + } + for _ in 0..2 { + put_rows( + &engine, + region_id, + Rows { + schema: columns.clone(), + rows: old_rows.clone(), + }, + ) + .await; + flush(&engine, region_id).await; + } + compact(&engine, region_id).await; + let region = engine.get_region(region_id).unwrap(); + let table_dir = region.table_dir().to_string(); + let large = region.version().ssts.levels()[1] + .files() + .next() + .unwrap() + .clone(); + const OUTPUT_SIZE: u64 = 64 * 1024; + assert!(large.size() > 8 * OUTPUT_SIZE); + engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: SetRegionOptions { + options: vec![ + SetRegionOption::Twsc( + "compaction.twcs.max_output_file_size".into(), + "64KiB".into(), + ), + SetRegionOption::Twsc( + "compaction.twcs.inactive_window.l1_merge_trigger".into(), + "2".into(), + ), + ], + }, + }), + ) + .await + .unwrap(); + + // Each pair of L0 files becomes one small L1. The first contains a Delete + // masking the large SST; later pairs update other rows in its key/time box. + for ts in 10..14 { + for _ in 0..2 { + let rows = Rows { + schema: columns.clone(), + rows: build_rows_for_key("m", ts, ts + 2, ts), + }; + if ts == 10 { + engine + .handle_request( + region_id, + RegionRequest::Delete(RegionDeleteRequest { + rows, + hint: None, + partition_expr_version: None, + }), + ) + .await + .unwrap(); + } else { + put_rows(&engine, region_id, rows).await; + } + flush(&engine, region_id).await; + } + compact(&engine, region_id).await; + } + put_and_flush(&engine, region_id, &columns, active_ts..active_ts + 1).await; + let version = region.version(); + let small = version.ssts.levels()[1] + .files() + .filter(|file| file.file_id() != large.file_id()) + .collect::>(); + assert_eq!(4, small.len()); + assert!(small.iter().all(|file| file.size() <= OUTPUT_SIZE / 4)); + let small_ids = small + .iter() + .map(|file| file.file_id()) + .collect::>(); + let pick = preview_regular_compaction(&engine, region_id).await; + assert_eq!(1, pick.outputs.len()); + assert_eq!( + small_ids, + pick.outputs[0] + .inputs + .iter() + .map(|file| file.file_id()) + .collect() + ); + assert!(!pick.outputs[0].filter_deleted); + let before = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap() + .pretty_print() + .unwrap(); + + // Even manual cleanup cannot rewrite the dominant peer within the closed + // merge's output bound, so it executes the same small-only fallback. + compact(&engine, region_id).await; + let version = region.version(); + let l1 = version.ssts.levels()[1].files().collect::>(); + assert_eq!(2, l1.len()); + assert!(l1.iter().any(|file| file.file_id() == large.file_id())); + let output = l1 + .iter() + .find(|file| file.file_id() != large.file_id()) + .unwrap(); + assert!(!small_ids.contains(&output.file_id())); + assert_eq!(1, count_sst_deletes(&engine, region_id, output).await); + let after = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert_eq!(before, after); + crate::test_util::reopen_region(&engine, region_id, table_dir, true, HashMap::new()).await; + let reopened = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert_eq!(before, reopened); +} + /// Region fixture containing only identifiers, so it cannot pin an old SST snapshot. struct CompactionTestRegion { region_id: RegionId, @@ -1806,6 +2007,7 @@ async fn preview_regular_compaction( &version.options, None, None, + false, ); let compaction_region = CompactionRegion { region_id, @@ -2777,9 +2979,9 @@ async fn test_compaction_region_with_format(flat_format: bool) { // -[15.........29]- (delete) // [15.....24] // Count-first compaction consumes the first 4 SSTs as soon as the trigger is reached. - // The compacted output and final flush leave 2 SSTs. + // The manual compaction then merges the compacted output with the final flush. assert_eq!( - 2, + 1, scanner.num_files(), "unexpected files: {:?}", scanner.file_ids() diff --git a/src/mito2/src/gc/worker_test.rs b/src/mito2/src/gc/worker_test.rs index 2a66294f59b..3d389e7e1d7 100644 --- a/src/mito2/src/gc/worker_test.rs +++ b/src/mito2/src/gc/worker_test.rs @@ -399,7 +399,13 @@ async fn test_gc_worker_basic_compact() { let region = engine.get_region(region_id).unwrap(); let manifest = region.manifest_ctx.manifest().await; - let removed_file_num = manifest.removed_files.removed_files[0].files.len(); + // A manual compaction may take several rounds, each recording its removals. + let removed_file_num: usize = manifest + .removed_files + .removed_files + .iter() + .map(|removed| removed.files.len()) + .sum(); assert!(removed_file_num > 0); let version = manifest.manifest_version; @@ -488,9 +494,11 @@ async fn test_gc_worker_compact_with_ref() { let file_ref_manifest = FileRefsManifest { file_refs: HashMap::from([( region_id, - manifest.removed_files.removed_files[0] - .files + manifest + .removed_files + .removed_files .iter() + .flat_map(|removed| removed.files.iter()) .map(|removed_file| match removed_file { RemovedFile::File(file_id, v) => FileRef::new(region_id, *file_id, *v), RemovedFile::Index(file_id, v) => FileRef::new(region_id, *file_id, Some(*v)), diff --git a/src/object-store/tests/object_store_test.rs b/src/object-store/tests/object_store_test.rs index de526834bad..d38480e6302 100644 --- a/src/object-store/tests/object_store_test.rs +++ b/src/object-store/tests/object_store_test.rs @@ -304,6 +304,9 @@ async fn test_s3_backend() -> Result<()> { } let store = ObjectStore::new(builder).unwrap(); + // Match production S3 clients: retry transient transport failures without + // retrying conditional-write conflicts or weakening their assertions. + let store = object_store::util::with_retry_layers(store); let store = object_store::util::with_instrument_layers(store, false); let guard = TempFolder::new(&store, "/");