From bb3f07b4825e074c207ae29db28de1e3354c11ae Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com> Date: Thu, 22 Oct 2026 13:21:39 +0000 Subject: [PATCH] fix(mito2): pick TWCS L1 merges by closed units so compaction terminates (#9741) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(mito2): pick TWCS L1 merges by closed units TWCS could keep rewriting L1 files without new writes. Overlap-only picks could split their inputs into one more file than they read, and L1 file reduction then rewrote up to 16 files to remove that single file. The outputs, split by key, created new overlaps and the cycle repeated. Replace the three global pick phases with one pick per window, scheduled by class across windows. Every pick now decreases a termination measure when the window receives no writes. - Active windows only take merges that remove files. L1 files are grouped by conflicts and merged in key order, so size-split tails are merged together. A reduction rewrites at most 4 output files' worth of bytes per removed file. - Inactive windows absorb L0 files, then take closed merges: no other file intrudes on the inputs' bounding box, so outputs cannot add overlaps. Each merge resolves conflicts or removes files, decreasing (conflicts, files). Files that do not take part in an overlap are not rewritten. - Overlaps that only share a boundary key no longer count as conflicts, so a size-split sorted run does not look overlapping. - Closed merges may take 4x the input limit to resolve a whole window once. - Cache primary key ranges per pick instead of resolving them per comparison. - Log the pick branch of each compaction output. Signed-off-by: Lei, HUANG * docs(mito2): state the compactor assumptions behind TWCS termination Signed-off-by: Lei, HUANG * feat(mito2): resolve active window overlaps on manual compaction A window that stops receiving writes keeps being the active window, whose picks only remove files, so its overlaps were never resolved. Manual compactions now pick every window as inactive, so a closed merge can resolve them. Automatic compactions are unchanged. Signed-off-by: Lei, HUANG * fix(mito2): defer overlap resolution in the window before the active one Late data keeps landing in the window right before the active window. A closed merge that resolves its overlaps early rewrites the whole window, then the late data brings the overlaps back, and the balanced run check keeps them there. Overlap-only merges now skip that window until it is older. File-reducing merges and manual compactions are unchanged. On the 12h/1GB out-of-order bench this reaches overlap depth 1.0 instead of 2.08 at the same write amplification (3.01 vs 3.02). Signed-off-by: Lei, HUANG * refactor(mito2): only check earlier-starting files when closing units Closing units only checks files that start before a unit ends, and the overlap checks compare key bounds without cloning them. Signed-off-by: Lei, HUANG * fix(mito2): keep settling windows from rewriting L1 files The window before the active one still receives late data. Skipping only its overlap-only merges was not enough: file-reducing merges of conflicting files and mixed L0/L1 merges still rewrote its L1 files, and a mixed merge could rewrite a nearly full output for each small late file. A settling window now only merges L0 files among themselves and conflict-free files; the window resolves its overlaps once it is older. On the out-of-order bench, write amplification drops from 3.88 to 3.30 at 2GB/6h, 3.05 to 2.94 at 1GB/6h and 3.01 to 2.96 at 1GB/12h, with overlap depth 1.0 in all three. Signed-off-by: Lei, HUANG * fix(mito2): resolve late overlaps once a window is quiet Late data that lands in a window older than the settling one is merged into a small file covering the whole key range. Resolving it rewrites the window's dominant sorted run, which the balanced run check refuses, so the overlap stayed. Removing the check instead rewrites the window for every late flush. A window is quiet once the region has written at least as many rows as the window holds since its newest file. Overlap merges in a quiet window may rewrite a dominant run, so each quiet period pays for at most one such rewrite. Sequences are per row and kept by compaction, so this needs no state. On the bench with late data up to 2 hours, mean overlap depth drops from 1.67 to 1.17 at 2GB/6h and from 2.33 to 1.83 at 1GB/6h, at write amplification 3.70 to 4.25 and 3.26 to 3.74; all inactive windows of 2GB/6h realtime reach depth 1 (write amplification 2.80 to 3.30). The remaining depth is in the final active window. Signed-off-by: Lei, HUANG * test(mito2): check closed merges of quiet and settling windows terminate Signed-off-by: Lei, HUANG * fix(mito2): split TWCS closures over the input limit A file spanning the keys of a large sorted run joins every run file into one closure. Once a closure held more files than one compaction accepts, closed merges skipped it, so the window never merged those files again and late files kept piling up. Split such closures into single files, like active-window picks do, and replace the closed check with a conflict bound: outputs only conflict with files intruding on the inputs' box, so a merge is taken if `outputs · I` does not exceed the conflicts its inputs had with other files. Intervals of whole closures have no such conflicts and still need to be closed, so other windows pick as before, and `(conflicts, files)` still strictly decreases. Small run files and late files over a large run can be merged again. A late file over a run that cannot shrink stays: resolving it would rewrite the whole run. Signed-off-by: Lei, HUANG * fix(mito2): address TWCS picker review findings - Manual compactions may rewrite a dominant sorted run to resolve overlaps. The active window is never quiet, so manual compactions could not resolve its late-data overlaps before. - Single-key intervals predict one output: the writer never splits a series, so large files of one series were never merged. - The active window falls back to single files when no group of conflicting files qualifies, so a large file no longer keeps its small peers unmerged. - Compute the settling window distance with checked_sub, which overflowed for windows at the timestamp limits. Signed-off-by: Lei, HUANG * fix(mito2): count inner keys as TWCS conflicts and check balance per group - A single key strictly inside another file's key range now conflicts, so a late file of one series inside a wide file is resolved like other overlaps. Files sharing only a boundary key still do not conflict. Intrusion covers the new pairs, which a randomized test checks. - Check the dominant-run guard per group of conflicting files. Runs of separate groups never overlap, so pairing one group's run with another group's peer made an interval of two dominant runs look balanced. - A window with unknown row counts, written before they were recorded, is not quiet: its threshold would drop to a single row. Signed-off-by: Lei, HUANG * perf(mito2): index TWCS files by key to avoid quadratic grouping Grouping a window's files compared every file with all files sharing its time range, and closing groups and checking intrusions scanned the whole window. A sorted run split by key shares one time range, so a window of N such files took O(N²) checks without any conflict. The 1024-file fallback only covers the sorted-run path, not these groups. Index the files by lower primary key in a segment tree that keeps each subtree's largest upper key and its time bounds. Grouping, closing, and the conflict bound only check the files whose key and time ranges meet, so a file of a key-sliced run only meets its neighbors. Conflicts among split files are precomputed per file and counted per interval with a binary search instead of pairwise. On a run of 20,000 key-sliced files with 5 late files spanning all keys, a closed merge pick takes ~62 ms instead of ~3.9 s and an active pick ~19 ms instead of ~3.5 s. Signed-off-by: Lei, HUANG * test(object-store): retry transient S3 transport failures Signed-off-by: Lei, HUANG * fix(mito2): bound TWCS closed-merge conflict graph work Budget the candidate files visited while building conflict units and split-unit adjacency. When the budget is exceeded, treat every file as a split unit and pick a bounded file reduction instead of materializing a quadratic graph. The fallback candidate is admitted only through adds_no_conflicts, so outputs do not add conflict pairs and the (P, N) termination invariant still holds. Dense picks are logged with their own branch. Signed-off-by: Lei, HUANG * fix(mito2): charge TWCS index nodes to the conflict-graph budget The conflict-graph work budget only counted files that reached a callback, so a traversal with poor pruning could still visit a quadratic number of internal nodes without triggering the dense fallback. Charge every ExtentIndex node, including pruned nodes, and stop the traversal as soon as the shared budget is exhausted. Add a regression test where the root straddles the query time but neither leaf matches, so a callback-only budget would never fire. Signed-off-by: Lei, HUANG * fix(mito2): retry smaller TWCS dense compaction candidates If a dense fallback candidate fails admission, try one single-output subinterval within it before giving up. Reuse the reduction guards and check both candidates against the full window's external conflicts, preserving bounded work and termination.\n\nCover the chain regression through both fallback unit constructions and reject one-output retries that introduce conflicts across time gaps. Signed-off-by: Lei, HUANG * fix(mito2): compact shared primary-key boundaries Count shared primary-key endpoints as conflicts and apply the same overlap rule to bounding-box intrusions. Existing compaction writers keep series intact, so fresh output ranges remain disjoint. Align simulated output splits with the writer and cover boundary-only cleanup in inactive and manual windows, plus stable real output files after a subsequent TWCS request. Signed-off-by: Lei, HUANG * fix(mito2): compact small L1 files in inactive windows Allow bounded small-L1 reductions in settling and older inactive windows when normal picks cannot proceed. Preserve full-window conflict checks, large-file protection, and LastNonNull closure semantics.\n\nCover selection limits, termination, and real SST compaction with tombstones and reopen checks. Signed-off-by: Lei, HUANG * docs(mito2): clarify TWCS unit grouping guarantees Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG --- src/mito2/src/compaction/last_non_null.rs | 7 +- src/mito2/src/compaction/picker.rs | 3 + src/mito2/src/compaction/scheduler.rs | 11 +- .../src/compaction/scheduler/planning.rs | 8 +- src/mito2/src/compaction/twcs.rs | 1484 +++++++++--- src/mito2/src/compaction/twcs/units.rs | 2073 +++++++++++++++++ src/mito2/src/engine/compaction_test.rs | 206 +- src/mito2/src/gc/worker_test.rs | 14 +- src/object-store/tests/object_store_test.rs | 3 + 9 files changed, 3426 insertions(+), 383 deletions(-) create mode 100644 src/mito2/src/compaction/twcs/units.rs 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, "/");