From 6f8d0947bca6d70bf658f15bbbf03aadbb104eba Mon Sep 17 00:00:00 2001 From: tfenne Date: Wed, 22 Jul 2026 13:59:01 -0600 Subject: [PATCH 1/2] feat(sort): add --max-temp-files to tune the spill-file consolidation limit The external sort consolidates spilled runs once their count reaches a hardcoded limit (64, matching samtools) by merging the oldest ~half into a single run. On large inputs that spill many runs this fires repeated consolidation passes whose merge loop is single-threaded, adding significant wall-time even though the final k-way merge is unaffected. Expose the limit as `--max-temp-files ` on `fgumi sort`, wired through `build_sorter` to the existing `RawExternalSorter::max_temp_files` builder. The flag is optional with a minimum of 2; when unset the engine default (64) applies, so behavior is byte-identical unless the user opts in. Raising it lets large sorts skip the extra consolidation passes, at the cost of more open file descriptors during the final merge. The flag deliberately declares no clap default and states no number in its help text, so tools that embed this `Sort` via `#[command(flatten)]` can apply their own default without inheriting a wrong one. Adds a `temp_file_limit()` accessor so the option-to-builder wiring can be unit tested. --- crates/fgumi-sort/src/external.rs | 8 ++ src/lib/commands/sort.rs | 167 ++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+) diff --git a/crates/fgumi-sort/src/external.rs b/crates/fgumi-sort/src/external.rs index 3f6005bec..25f004b5a 100644 --- a/crates/fgumi-sort/src/external.rs +++ b/crates/fgumi-sort/src/external.rs @@ -1660,6 +1660,14 @@ impl RawExternalSorter { self.merge_threads.unwrap_or(self.threads).max(1) } + /// Configured maximum number of temporary spill files kept before the + /// oldest runs are consolidated into one (`0` means unlimited). Exposed so + /// callers can assert how their `max_temp_files` setting resolved. + #[must_use] + pub fn temp_file_limit(&self) -> usize { + self.max_temp_files + } + /// Set the output compression level. #[must_use] pub fn output_compression(mut self, level: u32) -> Self { diff --git a/src/lib/commands/sort.rs b/src/lib/commands/sort.rs index 5095c3942..069bb2d57 100644 --- a/src/lib/commands/sort.rs +++ b/src/lib/commands/sort.rs @@ -154,6 +154,9 @@ PERFORMANCE: - Configurable temp file compression (--temp-compression) - Default 768M per-thread memory limit (samtools-compatible); pass `--max-memory auto` to detect system memory (opt-in) + - Spilled runs are consolidated once they exceed `--max-temp-files` + (default 64, matching samtools); raise it to avoid repeated + consolidation passes on very large inputs EXAMPLES: @@ -178,6 +181,9 @@ EXAMPLES: # Reserve extra memory for bwa mem running in a pipeline fgumi sort -i input.bam -o sorted.bam --memory-reserve 12GiB --threads 4 + # Allow more spilled runs before consolidating (fewer consolidation passes) + fgumi sort -i input.bam -o sorted.bam --order coordinate --max-temp-files 512 + # Verify a BAM file is correctly sorted fgumi sort -i sorted.bam --verify --order template-coordinate @@ -328,6 +334,22 @@ pub struct Sort { #[arg(long = "temp-codec", default_value = "zstd")] pub temp_codec: fgumi_sort::SpillCodec, + /// Maximum number of temporary spill files kept before the oldest are + /// consolidated into a single run. + /// + /// Large inputs spill many sorted runs to disk. When the number of runs + /// reaches this limit, the oldest are merged together in a single pass so + /// the final k-way merge opens fewer files at once. Raising it avoids + /// repeated consolidation passes on very large inputs (at the cost of more + /// open file descriptors during the final merge); lowering it keeps fewer + /// files open. Must be at least 2; for effectively unlimited, pass a large + /// value. + /// + /// When unset, a built-in default is used (see this command's help + /// overview for the default value). + #[arg(long = "max-temp-files", value_parser = parse_max_temp_files)] + pub max_temp_files: Option, + /// Write BAM index (.bai) alongside output. /// /// Only valid for coordinate sort. The index file will be written to @@ -408,6 +430,19 @@ pub(crate) fn parse_key_types(s: &str) -> Result { Ok(KeyTypesSpec::Explicit { cb, tertiary }) } +/// Parse `--max-temp-files`: a spill-file consolidation limit of at least 2. +/// +/// Values below 2 are meaningless (a merge needs at least two inputs) and are +/// rejected rather than silently clamped. For an effectively unlimited limit, +/// pass a large value. +pub(crate) fn parse_max_temp_files(s: &str) -> Result { + let n: usize = s.parse().map_err(|_| format!("`{s}` is not a non-negative integer"))?; + if n < 2 { + return Err(format!("must be at least 2 (got {n})")); + } + Ok(n) +} + impl Command for Sort { fn execute(&self, command_line: &str) -> Result<()> { if self.verify && self.output.is_some() { @@ -474,6 +509,10 @@ impl Sort { if let Some(n) = self.merge_threads { sorter = sorter.merge_threads(n); } + // Optional; falls back to the engine default when unset. + if let Some(n) = self.max_temp_files { + sorter = sorter.max_temp_files(n); + } sorter } @@ -833,6 +872,133 @@ mod tests { assert_eq!(sorter.phase2_threads(), expected_phase2); } + /// `--max-temp-files` parses onto the struct and defaults to `None` (the + /// builder then applies the engine default). + #[rstest] + #[case::unset(&[], None)] + #[case::explicit(&["--max-temp-files", "256"], Some(256))] + #[case::large(&["--max-temp-files", "100000"], Some(100_000))] + fn test_parse_max_temp_files(#[case] extra: &[&str], #[case] expected: Option) { + let base = ["sort", "-i", "in.bam", "-o", "out.bam", "--order", "coordinate"]; + let args: Vec<&str> = base.iter().copied().chain(extra.iter().copied()).collect(); + let sort = Sort::try_parse_from(args).expect("parse should succeed"); + assert_eq!(sort.max_temp_files, expected); + } + + /// Values below 2 are rejected at parse time rather than silently clamped. + #[rstest] + #[case::zero("0")] + #[case::one("1")] + fn test_parse_max_temp_files_rejects_below_two(#[case] value: &str) { + let args = [ + "sort", + "-i", + "in.bam", + "-o", + "out.bam", + "--order", + "coordinate", + "--max-temp-files", + value, + ]; + assert!(Sort::try_parse_from(args).is_err(), "--max-temp-files {value} must be rejected"); + } + + /// The flag reaches the sorter: `Some(n)` sets the limit; `None` leaves the + /// engine default (read from a fresh sorter rather than hardcoded, so this + /// test does not pin fgumi's default number). + #[rstest] + #[case::set(Some(256))] + #[case::unset(None)] + fn test_build_sorter_wires_max_temp_files(#[case] max_temp_files: Option) { + let engine_default = + RawExternalSorter::new(SortOrderArg::Coordinate.into()).temp_file_limit(); + let mut sort = make_sort(SortOrderArg::Coordinate); + sort.max_temp_files = max_temp_files; + + let sorter = sort.build_sorter(512 * 1024 * 1024, "fgumi sort (test)"); + assert_eq!(sorter.temp_file_limit(), max_temp_files.unwrap_or(engine_default)); + } + + /// The accessor test above only proves the flag reaches the builder. This + /// one proves `--max-temp-files` is purely a *how-we-consolidate* knob and + /// never changes *what* is written: with a 1 KiB memory budget (no floor in + /// `Fixed` mode) and enough records to spill many runs, `Some(2)` forces the + /// consolidation merge to fire repeatedly (oldest runs are merged once their + /// count reaches the limit), while the default engine limit (64) never + /// consolidates and merges every run in the final k-way pass. Both paths + /// must emit the same records in the same order — this guards the + /// consolidation path against a stable-order or record-loss regression the + /// accessor test cannot catch. + /// + /// We compare decoded records rather than the raw BGZF bytes on purpose: + /// the two write paths flush BGZF blocks at different points, so the + /// compressed framing legitimately differs even when the record stream is + /// identical. The contract consolidation must uphold is record identity, + /// not byte identity of the container. + #[test] + fn test_max_temp_files_consolidation_is_record_identical() { + use fgumi_sam::SamBuilder; + + // Reverse-ordered, all-distinct starts so the coordinate sort is a total + // order (no ties whose order could depend on merge grouping); 60 pairs + // at a 1 KiB budget spill well over a dozen runs. + let mut builder = SamBuilder::new(); + for i in 0..60 { + let _ = builder + .add_pair() + .name(&format!("read{i:04}")) + .start1((60 - i) * 100 + 1) + .start2((60 - i) * 100 + 51) + .build(); + } + let dir = tempfile::tempdir().expect("tempdir"); + let input = dir.path().join("in.bam"); + builder.write_bam(&input).expect("write bam"); + + // 1 KiB memory forces spilling; `build_sorter` is the exact option -> + // builder path `execute` uses, so this exercises the real wiring. + let run = |max_temp_files: Option, out: PathBuf| { + let mut sort = make_sort(SortOrderArg::Coordinate); + sort.input = input.clone(); + sort.output = Some(out.clone()); + sort.threads = 1; + sort.max_temp_files = max_temp_files; + + let sorter = sort.build_sorter(1024, "fgumi sort (test)"); + let stats = sorter.sort(&input, &out).expect("sort should succeed"); + + let mut reader = + noodles::bam::io::reader::Builder.build_from_path(&out).expect("open output"); + let header = reader.read_header().expect("read header"); + let records = reader + .record_bufs(&header) + .collect::>>() + .expect("read records"); + (stats, records) + }; + + let (default_stats, default_records) = run(None, dir.path().join("default.bam")); + let (limited_stats, limited_records) = run(Some(2), dir.path().join("limited.bam")); + + // Precondition: the tiny budget really did spill multiple runs, so the + // `Some(2)` run exercised consolidation rather than a trivial in-memory + // sort. Without this the identity assertion below could pass vacuously. + assert!( + default_stats.chunks_written >= 2, + "test must spill multiple runs to exercise consolidation, got {} chunk(s)", + default_stats.chunks_written, + ); + assert_eq!( + limited_stats.output_records, default_stats.output_records, + "consolidation must not drop or duplicate records", + ); + assert_eq!( + limited_records, default_records, + "--max-temp-files is a consolidation knob and must not change the emitted records", + ); + } + /// Separately, the knob must not change what is written. Covers the same /// asymmetric splits end-to-end through `execute`. #[rstest] @@ -896,6 +1062,7 @@ mod tests { compression: CompressionOptions::default(), temp_compression: 1, temp_codec: fgumi_sort::SpillCodec::default(), + max_temp_files: None, write_index: false, async_reader: false, } From da65744944324356a982d6c52c14cd25e66cafb2 Mon Sep 17 00:00:00 2001 From: tfenne Date: Wed, 22 Jul 2026 20:38:55 -0600 Subject: [PATCH 2/2] refactor(sort): validate --max-temp-files with clap instead of a hand-rolled parser Addresses review feedback on #643. Replace the custom `parse_max_temp_files` with `clap::builder::RangedU64ValueParser::`, which parses as u64, enforces the `>= 2` floor, and converts back to usize via `TryFrom`. This keeps the field `Option` (so tools that flatten `Sort` are unaffected) while letting clap own parsing, range-checking, and overflow. clap provides no ranged value parser for `usize` itself, which is why the parser was hand-rolled in the first place; going through `u64` is the idiomatic way to range-check a `usize`. The rejection messages are now accurate: an overflowing value reports "number too large to fit in target type" rather than the previous, misleading "not a non-negative integer", and a value below the floor reports "N is not in 2..". Also correct the command-overview wording (runs are consolidated once they "reach" `--max-temp-files`, not "exceed", matching the engine's `len >= max_temp_files` guard), document the flag in the performance-tuning guide's Sort section, and bump docs/LAST_SYNCED. Expand the parser tests to cover the minimum accepted value (2) and the rejected cases (0, 1, negative, non-numeric, overflow). --- docs/LAST_SYNCED | 2 +- docs/src/guide/performance-tuning.md | 5 +++++ src/lib/commands/sort.rs | 31 +++++++++++----------------- 3 files changed, 18 insertions(+), 20 deletions(-) diff --git a/docs/LAST_SYNCED b/docs/LAST_SYNCED index 3c3e5a04f..885ce3291 100644 --- a/docs/LAST_SYNCED +++ b/docs/LAST_SYNCED @@ -1,4 +1,4 @@ # Last commit reviewed for guide accuracy. # Run: git log ..HEAD -- src/lib/commands/ src/commands/ # to see what changed since the last guide update. -2f14de1cc487c4d0fe2f3f8cf28abfc9fc12ae28 +fb814f2a9f1a26d77fa29d878527579d3b10bd4c diff --git a/docs/src/guide/performance-tuning.md b/docs/src/guide/performance-tuning.md index 1ddb46fd0..31473e695 100644 --- a/docs/src/guide/performance-tuning.md +++ b/docs/src/guide/performance-tuning.md @@ -362,6 +362,11 @@ Requested memory 16GB exceeds 90% of system memory (14.4GB) better than a simple heap merge when the number of sorted runs is large - `--max-memory` controls how much RAM is used for sort buffers; increase for large files to reduce the number of intermediate merge passes +- `--max-temp-files` sets how many spilled runs may accumulate before the oldest are + consolidated into a single run (default 64, matching samtools). Consolidation is a + single-threaded merge pass, so on very large inputs raising this limit avoids repeated + consolidation passes at the cost of more open file descriptors during the final merge; the + output is unchanged. Must be at least 2 - For template-coordinate sort with single-cell data, the `CB` tag is included automatically - `--async-reader` is supported and can improve Phase 1 (input reading) throughput when disk latency is high or the OS page cache readahead is small diff --git a/src/lib/commands/sort.rs b/src/lib/commands/sort.rs index 069bb2d57..91d52d28f 100644 --- a/src/lib/commands/sort.rs +++ b/src/lib/commands/sort.rs @@ -154,7 +154,7 @@ PERFORMANCE: - Configurable temp file compression (--temp-compression) - Default 768M per-thread memory limit (samtools-compatible); pass `--max-memory auto` to detect system memory (opt-in) - - Spilled runs are consolidated once they exceed `--max-temp-files` + - Spilled runs are consolidated once they reach `--max-temp-files` (default 64, matching samtools); raise it to avoid repeated consolidation passes on very large inputs @@ -342,12 +342,12 @@ pub struct Sort { /// the final k-way merge opens fewer files at once. Raising it avoids /// repeated consolidation passes on very large inputs (at the cost of more /// open file descriptors during the final merge); lowering it keeps fewer - /// files open. Must be at least 2; for effectively unlimited, pass a large - /// value. + /// files open. Must be at least 2; to effectively disable consolidation, + /// pass a value larger than the number of runs you expect to spill. /// /// When unset, a built-in default is used (see this command's help /// overview for the default value). - #[arg(long = "max-temp-files", value_parser = parse_max_temp_files)] + #[arg(long = "max-temp-files", value_parser = clap::builder::RangedU64ValueParser::::new().range(2..))] pub max_temp_files: Option, /// Write BAM index (.bai) alongside output. @@ -430,19 +430,6 @@ pub(crate) fn parse_key_types(s: &str) -> Result { Ok(KeyTypesSpec::Explicit { cb, tertiary }) } -/// Parse `--max-temp-files`: a spill-file consolidation limit of at least 2. -/// -/// Values below 2 are meaningless (a merge needs at least two inputs) and are -/// rejected rather than silently clamped. For an effectively unlimited limit, -/// pass a large value. -pub(crate) fn parse_max_temp_files(s: &str) -> Result { - let n: usize = s.parse().map_err(|_| format!("`{s}` is not a non-negative integer"))?; - if n < 2 { - return Err(format!("must be at least 2 (got {n})")); - } - Ok(n) -} - impl Command for Sort { fn execute(&self, command_line: &str) -> Result<()> { if self.verify && self.output.is_some() { @@ -876,6 +863,7 @@ mod tests { /// builder then applies the engine default). #[rstest] #[case::unset(&[], None)] + #[case::minimum(&["--max-temp-files", "2"], Some(2))] #[case::explicit(&["--max-temp-files", "256"], Some(256))] #[case::large(&["--max-temp-files", "100000"], Some(100_000))] fn test_parse_max_temp_files(#[case] extra: &[&str], #[case] expected: Option) { @@ -885,11 +873,16 @@ mod tests { assert_eq!(sort.max_temp_files, expected); } - /// Values below 2 are rejected at parse time rather than silently clamped. + /// Invalid values are rejected at parse time rather than silently clamped or + /// truncated: values below the floor of 2 (a merge needs at least two + /// inputs), and non-numeric, negative, or overflowing input. #[rstest] #[case::zero("0")] #[case::one("1")] - fn test_parse_max_temp_files_rejects_below_two(#[case] value: &str) { + #[case::non_numeric("abc")] + #[case::negative("-1")] + #[case::overflow("99999999999999999999999999")] + fn test_parse_max_temp_files_rejects_invalid(#[case] value: &str) { let args = [ "sort", "-i",