diff --git a/Cargo.lock b/Cargo.lock index 9716d6b3..71f4782f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8395,6 +8395,7 @@ dependencies = [ "serde_json", "smallvec", "tempfile", + "tims_stage", "timscentroid", "timsquery", "timsrust", diff --git a/docs/development.md b/docs/development.md index f8441048..b8f8c61b 100644 --- a/docs/development.md +++ b/docs/development.md @@ -60,6 +60,17 @@ shared CLI/viewer plan report and Parquet scoring-plan metadata. Other elements are ignored by this approximation; isotope-labelled C/S and unspecified formula bases cannot supply its composition counts. +## Sample identity and output paths + +Search outputs use `//`, not the basename alone. +IDs combine a parent-location hash with the sample name, distinguishing same-named +inputs in different directories without hashing file contents. +`sample_id` and `sample_name` are recorded in run/performance reports and Parquet +file metadata, not repeated per row. Read these fields rather than parsing paths. + +Duplicate IDs within one invocation are rejected; rerunning with `--overwrite` +still replaces prior outputs. Existing basename-only output directories are not migrated. + ## Cargo features | Feature | Crate | Effect | Use case | Enable | diff --git a/rust/tims_stage/src/load.rs b/rust/tims_stage/src/load.rs index 07864436..11241d47 100644 --- a/rust/tims_stage/src/load.rs +++ b/rust/tims_stage/src/load.rs @@ -1,9 +1,9 @@ //! The single raw-index composition: URI → `IndexedTimstofPeaks`. //! -//! `load_raw` is the one place raw dispatch happens. It sniffs the URI via the -//! reader registry, asks the chosen reader for its [`Manifest`] (the files it -//! needs), materializes those -- in place for local inputs, or by fetching -//! exactly the declared files for remote ones -- and calls `read`. Transport +//! `PreparedSource` retains reader selection from preflight; `load_raw` selects +//! on demand for callers without preflight. Both ask that reader for its +//! [`Manifest`], materialize it -- in place for local inputs, or by fetching +//! exactly the declared files for remote ones -- and call `read`. Transport //! never guesses vendor shape: it fetches what the reader declared, by name. //! //! Lives in `tims_stage` (one crate above `timscentroid`, where the registry @@ -12,10 +12,13 @@ use std::ffi::OsString; use std::path::Path; +use std::sync::Arc; use http::Uri; use timscentroid::reader::{ + BrukerTdfReader, Manifest, + RawReader, ReadError, ReaderRegistry, ResolvedSource, @@ -54,6 +57,112 @@ pub enum LoadRawError { Stage(#[from] StageError), } +/// Reader selection retained from preflight through naming and index loading. +/// Standalone indexes may have no raw reader; tar transport contains Bruker .d. +#[derive(Clone)] +pub struct PreparedSource { + uri: String, + sample_name: String, + reader: Option>, +} + +impl std::fmt::Debug for PreparedSource { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PreparedSource") + .field("uri", &self.uri) + .field("sample_name", &self.sample_name) + .field("reader", &self.reader.as_ref().map(|r| r.name())) + .finish() + } +} + +impl PreparedSource { + pub fn new(uri: &str) -> Result { + let uri = crate::canonical_uri(uri); + let location = uri.trim_end_matches('/'); + let mut entry = location; + let mut wrapped = false; + let tar = location.to_ascii_lowercase().ends_with(".tar"); + let idx = location.to_ascii_lowercase().ends_with(".idx"); + loop { + let lower = entry.to_ascii_lowercase(); + if lower.ends_with(".idx") || lower.ends_with(".tar") { + entry = &entry[..entry.len() - 4]; + wrapped = true; + } else { + break; + } + } + let parsed = raw_uri(entry)?; + let reader = if tar { + // The existing tar loader is specifically a .d transport, not a + // second vendor-dispatch mechanism inferred from the archive name. + Some(Arc::new(BrukerTdfReader) as Arc) + } else { + match ReaderRegistry::with_builtins().pick(&parsed, || None) { + Ok(reader) => Some(reader), + Err(ReadError::UnknownFormat(_)) if idx => None, + Err(error) => return Err(error), + } + }; + Self::with_reader(&uri, entry, wrapped, reader) + } + + fn with_reader( + uri: &str, + entry: &str, + wrapped: bool, + reader: Option>, + ) -> Result { + let name = entry.rsplit('/').next().unwrap_or(entry); + let sample_name = reader + .as_ref() + .and_then(|r| r.sample_name(name)) + .or_else(|| wrapped.then_some(name)) + .ok_or_else(|| ReadError::Build(format!("selected reader cannot name {uri:?}")))? + .to_owned(); + Ok(Self { + uri: uri.to_owned(), + sample_name, + reader, + }) + } + + pub fn uri(&self) -> &str { + &self.uri + } + + pub fn sample_name(&self) -> &str { + &self.sample_name + } + + /// Read the raw artifact (or the .d extracted by tar staging) using the + /// same reader which supplied its name. No registry dispatch here. + pub fn read_raw( + &self, + uri: &str, + backend: &dyn StagingBackend, + cfg: &IndexingCentroidingConfig, + ) -> Result { + let reader = self + .reader + .as_ref() + .ok_or_else(|| ReadError::Build("cached index has no raw reader".into()))?; + load_with_reader(uri, reader.as_ref(), backend, cfg) + } +} + +fn raw_uri(uri: &str) -> Result { + if is_remote_uri(uri) { + uri.parse().map_err(|source| ReadError::UriParse { + uri: uri.to_owned(), + source, + }) + } else { + local_uri(&std::path::absolute(uri)?) + } +} + /// Build an index from a raw URI (local path or `s3://…`), dispatching through /// the reader registry. Remote inputs stage exactly the reader's declared /// manifest into a tempdir (kept alive across the read); local inputs are read @@ -63,8 +172,17 @@ pub fn load_raw( backend: &dyn StagingBackend, cfg: &IndexingCentroidingConfig, ) -> Result { - let registry = ReaderRegistry::with_builtins(); + let parsed = raw_uri(&crate::canonical_uri(uri))?; + let reader = ReaderRegistry::with_builtins().pick(&parsed, || None)?; + load_with_reader(uri, reader.as_ref(), backend, cfg) +} +fn load_with_reader( + uri: &str, + reader: &dyn RawReader, + backend: &dyn StagingBackend, + cfg: &IndexingCentroidingConfig, +) -> Result { if is_remote_uri(uri) { let parsed = uri.parse::().map_err(|source| ReadError::UriParse { uri: uri.to_string(), @@ -72,7 +190,6 @@ pub fn load_raw( })?; // No reader returns `Maybe` yet, so the magic-byte peek is never needed; // remote byte-sniffing (a range-GET here) is future work. - let reader = registry.pick(&parsed, || None)?; let manifest = reader.manifest(&parsed); // `staged` owns the tempdir and must outlive the read below. let staged = stage_manifest(backend, &manifest)?; @@ -84,14 +201,9 @@ pub fn load_raw( }); } - // Local: canonicalize first so RELATIVE paths resolve -- `local_uri` - // (sniff) and `local_in_place` (read) both require an absolute path. This - // is the single place all entry points funnel through, so relative inputs - // work uniformly (`read_index`, `load_index_auto`, the pyo3 binding, …). + // Resolve the actual local artifact without reselecting its reader. let abs = std::fs::canonicalize(uri) .map_err(|e| ReadError::Build(format!("cannot resolve local raw path {uri:?}: {e}")))?; - let parsed = local_uri(&abs)?; - let reader = registry.pick(&parsed, || None)?; let src = ResolvedSource::local_in_place(&abs)?; let index = reader.read(&src, cfg)?; Ok(RawRead { @@ -214,6 +326,68 @@ mod tests { }; use timscentroid::reader::Manifest; + #[test] + fn prepared_names_unwrap_transport() { + for name in [ + "My-Run.D", + "My-Run.d.IDX", + "My-Run.D.TAR", + "My-Run.idx", + "My-Run.tar", + ] { + assert_eq!(PreparedSource::new(name).unwrap().sample_name(), "My-Run"); + } + assert_eq!( + PreparedSource::new("run.raw.idx").unwrap().sample_name(), + "run.raw" + ); + assert!(PreparedSource::new("run.raw").is_err()); + assert!(PreparedSource::new("run.mzML.gz").is_err()); + } + + #[test] + fn prepared_source_names_and_reads_with_retained_reader() { + struct SelectedReader; + impl RawReader for SelectedReader { + fn name(&self) -> &'static str { + "selected" + } + + fn sniff(&self, _: &Uri) -> timscentroid::reader::Sniff { + panic!("must not redispatch after selection") + } + + fn sample_name<'a>(&self, name: &'a str) -> Option<&'a str> { + name.strip_suffix(".custom") + } + + fn manifest(&self, _: &Uri) -> Manifest { + unreachable!("local input") + } + + fn read( + &self, + src: &ResolvedSource, + _: &IndexingCentroidingConfig, + ) -> Result { + assert_eq!(src.entry_path().file_name().unwrap(), "run.custom"); + Err(ReadError::Build("retained reader called".into())) + } + } + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("run.custom"); + std::fs::write(&path, []).unwrap(); + let uri = path.to_str().unwrap(); + let source = + PreparedSource::with_reader(uri, uri, false, Some(Arc::new(SelectedReader))).unwrap(); + assert_eq!(source.sample_name(), "run"); + let backend = PerRunTempdir::new(StagingConfig::default()).unwrap(); + assert!( + matches!(source.read_raw(uri, &backend, &IndexingCentroidingConfig::default()), + Err(LoadRawError::Read(ReadError::Build(message))) if message == "retained reader called") + ); + } + fn u(p: &std::path::Path) -> Uri { // Raw path URI (tempdir paths are space-free, so no percent-encoding). p.to_str().unwrap().parse::().unwrap() diff --git a/rust/tims_stage/src/uri.rs b/rust/tims_stage/src/uri.rs index 14176b8a..a91b045f 100644 --- a/rust/tims_stage/src/uri.rs +++ b/rust/tims_stage/src/uri.rs @@ -39,9 +39,10 @@ pub(crate) fn parse_uri_shape(uri: &str) -> Result { } else { LocKind::Local }; - let name = if trimmed.ends_with(".idx") { + let lower = trimmed.to_ascii_lowercase(); + let name = if lower.ends_with(".idx") { NameKind::Idx - } else if trimmed.ends_with(".tar") { + } else if lower.ends_with(".tar") { NameKind::Tar } else { // Any other suffix is a raw vendor artifact; the reader registry @@ -144,6 +145,18 @@ pub fn canonical_uri(uri: &str) -> String { mod tests { use super::*; + #[test] + fn transport_suffixes_are_case_insensitive() { + assert_eq!( + parse_uri_shape("/data/run.D.IDX").unwrap().name, + NameKind::Idx + ); + assert_eq!( + parse_uri_shape("/data/run.D.TAR").unwrap().name, + NameKind::Tar + ); + } + #[test] fn recognizes_local_dotd() { let s = parse_uri_shape("/tmp/sample.d").unwrap(); diff --git a/rust/timscentroid/src/reader/mod.rs b/rust/timscentroid/src/reader/mod.rs index c09eac82..576bb708 100644 --- a/rust/timscentroid/src/reader/mod.rs +++ b/rust/timscentroid/src/reader/mod.rs @@ -1,7 +1,8 @@ //! Vendor-neutral raw-format dispatch. //! //! A [`RawReader`] is a self-describing backend: it declares which URIs it -//! claims ([`RawReader::sniff`]), which artifacts belong together +//! claims ([`RawReader::sniff`]), their canonical sample names +//! ([`RawReader::sample_name`]), which artifacts belong together //! ([`RawReader::manifest`]), and how to build the in-memory index //! ([`RawReader::read`]). The [`ReaderRegistry`] picks a backend for a URI by //! sniffing -- vendor suffix/scheme knowledge lives ONLY in each reader, never @@ -17,6 +18,7 @@ use std::path::{ Path, PathBuf, }; +use std::sync::Arc; use http::Uri; use url::Url; @@ -125,6 +127,10 @@ pub trait RawReader: Send + Sync { /// a magic-byte peek. fn sniff(&self, uri: &Uri) -> Sniff; + /// Canonical sample stem from an entry filename, without opening data. + /// Return None when this reader does not recognize the name. + fn sample_name<'a>(&self, name: &'a str) -> Option<&'a str>; + /// Confirm a `Maybe` by inspecting leading bytes. Only called when /// `sniff` == `Maybe`. fn sniff_bytes(&self, _head: &[u8]) -> bool { @@ -158,17 +164,17 @@ pub trait RawReader: Send + Sync { ) -> Result; } -/// Ordered set of backends. First `Yes` wins. -pub struct ReaderRegistry(Vec>); +/// Set of backends. Multiple definite claims are rejected as ambiguous. +pub struct ReaderRegistry(Vec>); impl ReaderRegistry { /// The built-in readers. `BrukerTdfReader` is always present; format /// backends behind features are pushed under their `#[cfg]`. pub fn with_builtins() -> Self { #[allow(unused_mut)] - let mut v: Vec> = vec![Box::new(BrukerTdfReader)]; + let mut v: Vec> = vec![Arc::new(BrukerTdfReader)]; #[cfg(feature = "mzdata")] - v.push(Box::new(crate::reader::mzdata::MzdataReader)); + v.push(Arc::new(crate::reader::mzdata::MzdataReader)); Self(v) } @@ -189,13 +195,13 @@ impl ReaderRegistry { &self, uri: &Uri, head: impl FnOnce() -> Option>, - ) -> Result<&dyn RawReader, ReadError> { - let mut winner: Option<&dyn RawReader> = None; - let mut maybes: Vec<&dyn RawReader> = Vec::new(); + ) -> Result, ReadError> { + let mut winner: Option> = None; + let mut maybes = Vec::new(); for r in &self.0 { match r.sniff(uri) { Sniff::Yes => match winner { - None => winner = Some(r.as_ref()), + None => winner = Some(Arc::clone(r)), Some(w) => { return Err(ReadError::AmbiguousFormat { first: w.name(), @@ -203,7 +209,7 @@ impl ReaderRegistry { }); } }, - Sniff::Maybe => maybes.push(r.as_ref()), + Sniff::Maybe => maybes.push(Arc::clone(r)), Sniff::No => {} } } @@ -223,12 +229,12 @@ impl ReaderRegistry { } } -/// Lowercased path suffix test for a URI (scheme-agnostic; keys on the path). -fn path_ends_with(uri: &Uri, suffix_lower: &str) -> bool { - uri.path() - .trim_end_matches('/') - .to_ascii_lowercase() - .ends_with(suffix_lower) +/// Case-insensitive format suffix matching shared by sniffing and naming. +pub(crate) fn strip_format_suffix<'a>(name: &'a str, suffix: &str) -> Option<&'a str> { + let start = name.len().checked_sub(suffix.len())?; + name.get(start..)? + .eq_ignore_ascii_case(suffix) + .then(|| &name[..start]) } /// Build a `file://` URI from an absolute local path, percent-encoding as @@ -262,13 +268,17 @@ impl RawReader for BrukerTdfReader { } fn sniff(&self, uri: &Uri) -> Sniff { - if path_ends_with(uri, ".d") { + if self.sample_name(uri.path().trim_end_matches('/')).is_some() { Sniff::Yes } else { Sniff::No } } + fn sample_name<'a>(&self, name: &'a str) -> Option<&'a str> { + strip_format_suffix(name, ".d") + } + fn manifest(&self, uri: &Uri) -> Manifest { Manifest { entry: uri.clone(), @@ -300,6 +310,27 @@ impl RawReader for BrukerTdfReader { mod tests { use super::*; + #[test] + fn selected_reader_supplies_name() { + let registry = ReaderRegistry::with_builtins(); + for (name, expected) in [ + ("My-Run.d", "My-Run"), + ("My-Run.D", "My-Run"), + ("run.raw.D", "run.raw"), + ("Échantillon.D", "Échantillon"), + ] { + let uri = local_uri(&std::path::absolute(name).unwrap()).unwrap(); + let reader = registry.pick(&uri, || None).unwrap(); + assert_eq!(reader.sample_name(name), Some(expected)); + } + #[cfg(feature = "mzdata")] + for name in ["My-Run.mzML", "My-Run.MZML", "My-Run.MzMl"] { + let uri = local_uri(&std::path::absolute(name).unwrap()).unwrap(); + let reader = registry.pick(&uri, || None).unwrap(); + assert_eq!(reader.sample_name(name), Some("My-Run")); + } + } + #[test] fn bruker_sniffs_dotd_yes_else_no() { let b = BrukerTdfReader; diff --git a/rust/timscentroid/src/reader/mzdata.rs b/rust/timscentroid/src/reader/mzdata.rs index 5f91a966..508d1b80 100644 --- a/rust/timscentroid/src/reader/mzdata.rs +++ b/rust/timscentroid/src/reader/mzdata.rs @@ -25,7 +25,6 @@ use super::{ ReadError, ResolvedSource, Sniff, - path_ends_with, }; use crate::centroiding::IndexingCentroidingConfig; use crate::dimension::MobilityKind; @@ -68,13 +67,17 @@ impl RawReader for MzdataReader { fn sniff(&self, uri: &Uri) -> Sniff { // `.mzML.gz` is out of scope: it ends with `.gz`, not `.mzml`, so it is // correctly NOT claimed here (registry → UnknownFormat, loud). - if path_ends_with(uri, ".mzml") { + if self.sample_name(uri.path().trim_end_matches('/')).is_some() { Sniff::Yes } else { Sniff::No } } + fn sample_name<'a>(&self, name: &'a str) -> Option<&'a str> { + super::strip_format_suffix(name, ".mzml") + } + fn manifest(&self, uri: &Uri) -> Manifest { Manifest { entry: uri.clone(), diff --git a/rust/timsquery/src/serde/index_serde.rs b/rust/timsquery/src/serde/index_serde.rs index 35950eab..97f8b292 100644 --- a/rust/timsquery/src/serde/index_serde.rs +++ b/rust/timsquery/src/serde/index_serde.rs @@ -865,6 +865,32 @@ pub fn load_index( backend: &dyn StagingBackend, save_sidecar: bool, centroid_cfg: IndexingCentroidingConfig, +) -> Result<(IndexedTimstofPeaks, IndexSource), LoadIndexError> { + load_index_inner(uri, backend, save_sidecar, centroid_cfg, None) +} + +/// Load using the reader already selected during sample preflight. +pub fn load_index_for_source( + source: &tims_stage::load::PreparedSource, + backend: &dyn StagingBackend, + save_sidecar: bool, + centroid_cfg: IndexingCentroidingConfig, +) -> Result<(IndexedTimstofPeaks, IndexSource), LoadIndexError> { + load_index_inner( + source.uri(), + backend, + save_sidecar, + centroid_cfg, + Some(source), + ) +} + +fn load_index_inner( + uri: &str, + backend: &dyn StagingBackend, + save_sidecar: bool, + centroid_cfg: IndexingCentroidingConfig, + source: Option<&tims_stage::load::PreparedSource>, ) -> Result<(IndexedTimstofPeaks, IndexSource), LoadIndexError> { let canon = canonical_uri(uri); match resolve(&canon)? { @@ -879,7 +905,10 @@ pub fn load_index( index, reader_name, caches_to_idx, - } = load_raw(&uri, backend, ¢roid_cfg)?; + } = match source { + Some(source) => source.read_raw(&uri, backend, ¢roid_cfg), + None => load_raw(&uri, backend, ¢roid_cfg), + }?; if save_sidecar && caches_to_idx { write_sidecar(&canon, &index)?; } @@ -901,7 +930,10 @@ pub fn load_index( index, reader_name, caches_to_idx, - } = load_raw(staged_path, backend, ¢roid_cfg)?; + } = match source { + Some(source) => source.read_raw(staged_path, backend, ¢roid_cfg), + None => load_raw(staged_path, backend, ¢roid_cfg), + }?; if save_sidecar && caches_to_idx { write_sidecar(&canon, &index)?; } diff --git a/rust/timsquery/src/serde/psims_origin_type.rs b/rust/timsquery/src/serde/psims_origin_type.rs index 02d26ee8..d6610258 100644 --- a/rust/timsquery/src/serde/psims_origin_type.rs +++ b/rust/timsquery/src/serde/psims_origin_type.rs @@ -14,7 +14,7 @@ /// `--check` compares this against the published ontology, so most /// releases will report a difference that leaves the table unchanged. /// Regenerating then edits only this line. -pub const DATA_VERSION: &str = "4.1.261"; +pub const DATA_VERSION: &str = "4.2.2"; /// One `spectrum origin type` term, and what it implies for a reader. #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/rust/timsseek/Cargo.toml b/rust/timsseek/Cargo.toml index fe50a50e..a57b1569 100644 --- a/rust/timsseek/Cargo.toml +++ b/rust/timsseek/Cargo.toml @@ -17,6 +17,7 @@ timsquery = { path = "../timsquery" } calibrt = { path = "../calibrt" } micromzpaf = { path = "../micromzpaf" } timscentroid = { path = "../timscentroid" } +tims_stage = { path = "../tims_stage" } array2d = { path = "../array2d" } timsseek_macros = { path = "../timsseek_macros" } diff --git a/rust/timsseek/src/lib.rs b/rust/timsseek/src/lib.rs index dff5992e..0db9f100 100644 --- a/rust/timsseek/src/lib.rs +++ b/rust/timsseek/src/lib.rs @@ -7,6 +7,7 @@ pub mod ml; pub mod models; pub mod protein; pub mod rt_calibration; +pub mod sample_identity; pub mod scoring; pub mod traits; pub mod utils; diff --git a/rust/timsseek/src/sample_identity.rs b/rust/timsseek/src/sample_identity.rs new file mode 100644 index 00000000..d0fb5a1d --- /dev/null +++ b/rust/timsseek/src/sample_identity.rs @@ -0,0 +1,139 @@ +//! Location-derived identity for a single sample. No content access or hashing. + +/// Construct only from a location; neither the ID nor its paired display name +/// can be assigned or deserialized independently. +/// +/// ```compile_fail +/// use timsseek::sample_identity::SampleIdentity; +/// let forged = SampleIdentity { sample_id: "handmade".into(), sample_name: "run".into() }; +/// ``` +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +pub struct SampleIdentity { + sample_id: String, + sample_name: String, +} + +impl SampleIdentity { + /// `location` is an absolute local path (forward slashes on Windows), or a + /// supported remote URI. The CLI expands local paths before calling this. + /// No symlink resolution, URI decoding, or content equivalence is attempted. + pub fn from_location(location: &str) -> std::io::Result { + let source = tims_stage::load::PreparedSource::new(location) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?; + Self::from_source(&source) + } + + /// Use the name from the reader selected for loading this source. + pub fn from_source(source: &tims_stage::load::PreparedSource) -> std::io::Result { + let location = source.uri(); + let invalid = || { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("Unable to derive sample identity from input {location:?}"), + ) + }; + let location = location.trim_end_matches('/'); + if let Some((scheme, rest)) = location.split_once("://") { + if !matches!(scheme, "s3" | "gs" | "az") { + return Err(invalid()); + } + let (bucket, key) = rest.split_once('/').ok_or_else(invalid)?; + if bucket.is_empty() || key.is_empty() { + return Err(invalid()); + } + } else if !std::path::Path::new(location).is_absolute() { + return Err(invalid()); + } + let (parent, _) = location.rsplit_once('/').ok_or_else(invalid)?; + let sample_name = source.sample_name().to_owned(); + if matches!(sample_name.as_str(), "" | "." | "..") + || sample_name.contains('\\') + || sample_name.chars().any(char::is_control) + { + return Err(invalid()); + } + let sample_id = format!( + "{:016x}-{sample_name}", + fnv1a64(format!("{parent}/").as_bytes()) + ); + Ok(Self { + sample_id, + sample_name, + }) + } + + pub fn sample_id(&self) -> &str { + &self.sample_id + } + + pub fn sample_name(&self) -> &str { + &self.sample_name + } +} + +// Fixed FNV-1a 64-bit over UTF-8 bytes. Not DefaultHasher: its implementation +// is not a persistence contract. Parent text includes its trailing slash. +fn fnv1a64(bytes: &[u8]) -> u64 { + bytes.iter().fold(0xcbf29ce484222325, |hash, byte| { + (hash ^ u64::from(*byte)).wrapping_mul(0x100000001b3) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fixed_hash_vectors() { + assert_eq!(fnv1a64(b""), 0xcbf29ce484222325); + assert_eq!(fnv1a64(b"hello"), 0xa430d84680aabd0b); + let sample = SampleIdentity::from_location("s3://bucket/rerun_1/my-run.d").unwrap(); + assert_eq!(sample.sample_id(), "0965dff92bab1aaf-my-run"); + let json = serde_json::to_value(&sample).unwrap(); + assert_eq!(json["sample_id"], sample.sample_id()); + assert_eq!(json["sample_name"], "my-run"); + } + + #[test] + fn supported_storage_suffixes() { + for name in ["run.d", "run.d.tar", "run.d.idx", "run.idx", "run.tar"] { + let identity = SampleIdentity::from_location(&format!("s3://bucket/{name}")).unwrap(); + assert_eq!(identity.sample_name(), "run"); + } + for name in [ + "run\\..\\other.d", + "run\0.d", + ".d", + ".idx", + "run.raw", + "run.mzML.gz", + ] { + assert!(SampleIdentity::from_location(&format!("s3://bucket/{name}")).is_err()); + } + } + + #[test] + fn suffix_case_is_ignored_without_changing_stem_or_parent_case() { + let expected = SampleIdentity::from_location("s3://bucket/Batch/My-Run.d").unwrap(); + for suffix in [".D", ".D.TaR", ".D.IdX/"] { + let location = format!("s3://bucket/Batch/My-Run{suffix}"); + assert_eq!(SampleIdentity::from_location(&location).unwrap(), expected); + } + assert_eq!(expected.sample_name(), "My-Run"); + for location in ["s3://bucket/batch/My-Run.d", "s3://bucket/Batch/my-run.d"] { + assert_ne!(SampleIdentity::from_location(location).unwrap(), expected); + } + assert_eq!( + SampleIdentity::from_location( + std::env::current_dir() + .unwrap() + .join("Échantillon.D") + .to_str() + .unwrap() + ) + .unwrap() + .sample_name(), + "Échantillon" + ); + } +} diff --git a/rust/timsseek/src/scoring/parquet_writer.rs b/rust/timsseek/src/scoring/parquet_writer.rs index 7c8914db..50bf5c53 100644 --- a/rust/timsseek/src/scoring/parquet_writer.rs +++ b/rust/timsseek/src/scoring/parquet_writer.rs @@ -26,6 +26,7 @@ use super::blocks::{ ScoreBlock, }; use super::results::FinalResult; +use crate::sample_identity::SampleIdentity; /// Bumped when a column's meaning or type changes, so a reader can tell a new /// file from an old one rather than silently misreading it. @@ -223,12 +224,16 @@ pub struct ResultParquetWriter<'a> { } impl<'a> ResultParquetWriter<'a> { - pub fn new( + /// Rescored results, including competition, discriminant score and q-value columns. + /// Writes already-computed values; this constructor does not perform rescoring. + /// Identity is required at construction, including for zero-row artifacts. + pub fn rescored( path: impl AsRef, row_group_size: usize, library: &'a crate::data_sources::reference_library::ReferenceLibrary, + sample: &SampleIdentity, ) -> std::io::Result { - Self::with_mode(path, row_group_size, library, false) + Self::with_mode(path, row_group_size, library, sample, false) } /// Common scores only: no competition, discriminant score or q-value columns. @@ -236,14 +241,16 @@ impl<'a> ResultParquetWriter<'a> { path: impl AsRef, row_group_size: usize, library: &'a crate::ReferenceLibrary, + sample: &SampleIdentity, ) -> std::io::Result { - Self::with_mode(path, row_group_size, library, true) + Self::with_mode(path, row_group_size, library, sample, true) } fn with_mode( path: impl AsRef, row_group_size: usize, library: &'a crate::ReferenceLibrary, + sample: &SampleIdentity, raw: bool, ) -> std::io::Result { let geom = library.geometry(); @@ -260,6 +267,14 @@ impl<'a> ResultParquetWriter<'a> { let schema = empty_batch.schema(); let kv = vec![ + KeyValue { + key: "sample_id".into(), + value: Some(sample.sample_id().to_owned()), + }, + KeyValue { + key: "sample_name".into(), + value: Some(sample.sample_name().to_owned()), + }, KeyValue { key: "result_mode".into(), value: Some(if raw { "raw" } else { "rescored" }.into()), @@ -347,6 +362,50 @@ mod tests { TargetColumnsBuilder, }; + #[test] + fn sample_metadata_survives_empty_and_nonempty_files_in_both_modes() { + let geom = one_row_arena(); + let library = crate::ReferenceLibrary::try_from(timsquery::serde::TargetTable::Mzpaf { + frag_intens: Some(vec![1.0; geom.n_fragments()]), + geom, + }) + .unwrap(); + let sample = + crate::sample_identity::SampleIdentity::from_location("s3://bucket/run.d").unwrap(); + let dir = tempfile::tempdir().unwrap(); + for raw in [false, true] { + for nrows in [0, 1] { + let path = dir.path().join(format!("{raw}-{nrows}.parquet")); + let mut writer = if raw { + ResultParquetWriter::raw(&path, 1, &library, &sample) + } else { + ResultParquetWriter::rescored(&path, 1, &library, &sample) + } + .unwrap(); + if nrows == 1 { + writer.add(sample_in(library.geometry())).unwrap(); + } + writer.close().unwrap(); + let reader = SerializedFileReader::new(File::open(path).unwrap()).unwrap(); + let meta = reader.metadata().file_metadata(); + assert_eq!(meta.num_rows(), nrows); + for (key, value) in [ + ("sample_id", "07cff6d98863b0e4-run"), + ("sample_name", "run"), + ] { + let entries: Vec<_> = meta + .key_value_metadata() + .unwrap() + .iter() + .filter(|kv| kv.key == key) + .collect(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].value.as_deref(), Some(value)); + } + } + } + } + /// A sealed arena with one row per `(sequence, id)`, for the writer to /// resolve ids against. `None` for an id leaves the row to be minted. #[test] @@ -586,7 +645,9 @@ mod tests { }, ) .unwrap(); - let writer = ResultParquetWriter::new(&path, 1024, &library).expect("create writer"); + let sample = SampleIdentity::from_location("s3://bucket/run.d").unwrap(); + let writer = ResultParquetWriter::rescored(&path, 1024, &library, &sample) + .expect("create writer"); writer.close().expect("close"); } let file = File::open(&path).expect("open"); diff --git a/rust/timsseek/src/scoring/timings.rs b/rust/timsseek/src/scoring/timings.rs index 7d13487c..cf38c1e1 100644 --- a/rust/timsseek/src/scoring/timings.rs +++ b/rust/timsseek/src/scoring/timings.rs @@ -218,6 +218,8 @@ pub struct RunReport<'a> { /// them with no extra discovery step. #[derive(Debug, Serialize)] pub struct FileReport { + #[serde(flatten)] + pub sample: crate::sample_identity::SampleIdentity, pub file_name: String, pub pipeline: PipelineReport, #[serde(default)] diff --git a/rust/timsseek_cli/src/artifacts.rs b/rust/timsseek_cli/src/artifacts.rs index 68935671..5a0fb3cd 100644 --- a/rust/timsseek_cli/src/artifacts.rs +++ b/rust/timsseek_cli/src/artifacts.rs @@ -13,8 +13,8 @@ use crate::errors::CliError; use crate::output_sink::{ join_output_uri, probe_uri_exists, - sample_name_from_uri, }; +use crate::sample_identity::SampleInput; pub(crate) const RESULTS_PARQUET: &str = "results.parquet"; pub(crate) const PERFORMANCE_REPORT: &str = "performance_report.json"; @@ -50,7 +50,7 @@ pub(crate) const BUILT_LIBRARY_SUFFIX: &str = ".mzspeclib.txt.gz"; /// /// The stem is the file name with one extension off, and a `.gz` before it, so /// `proteome.fasta` and `proteome.fasta.gz` name the same library. Not -/// [`sample_name_from_uri`], which strips the `.d`/`.tar`/`.idx` a raw input +/// the sample-name helper, which strips the `.d`/`.tar`/`.idx` a raw input /// carries and would leave a FASTA's own suffix in front of `.mzspeclib`. pub(crate) fn built_library_path(output_uri: &str, fasta: &Path) -> PathBuf { let name = fasta @@ -63,16 +63,14 @@ pub(crate) fn built_library_path(output_uri: &str, fasta: &Path) -> PathBuf { } /// Probe every artifact a run can write, returning the URIs of the ones already -/// there. Raw inputs rather than sample names, because the sample name is -/// derived from the URI and a URI with no derivable name is an error worth -/// reporting here. +/// there. Uses the same resolved identities as the writers and upload paths. /// /// `built_library` is the library the run is about to predict, which is an /// output of the run like any other: one already sitting there cost minutes to /// predict and describes whatever FASTA and settings produced it. pub(crate) fn probe_collisions( output_uri: &str, - raw_inputs: &[String], + samples: &[SampleInput], built_library: Option<&Path>, ) -> Result, CliError> { let mut collisions: Vec = Vec::new(); @@ -87,11 +85,8 @@ pub(crate) fn probe_collisions( } } } - for raw_uri in raw_inputs { - let sample = sample_name_from_uri(raw_uri).ok_or_else(|| CliError::Io { - source: "Unable to extract file stem".to_string(), - path: Some(raw_uri.clone()), - })?; + for sample in samples { + let sample = sample.sample_id(); for artifact in PER_SAMPLE_ARTIFACTS { let uri = join_output_uri(output_uri, &format!("{sample}/{artifact}")); if probe_uri_exists(&uri)? { @@ -118,14 +113,15 @@ mod tests { #[test] fn the_probe_reports_a_feature_stats_sidecar_from_an_earlier_run() { let dir = tempfile::tempdir().unwrap(); - let sample_dir = dir.path().join("run"); + let samples = + crate::sample_identity::resolve_samples(&["/data/run.d".to_string()]).unwrap(); + let sample_dir = dir.path().join(samples[0].sample_id()); std::fs::create_dir_all(&sample_dir).unwrap(); std::fs::write(sample_dir.join(FEATURE_STATS_TSV), "name\tmean\n").unwrap(); - let raw_inputs = vec!["/data/run.d".to_string()]; let output_uri = dir.path().to_string_lossy().to_string(); - let collisions = probe_collisions(&output_uri, &raw_inputs, None).unwrap(); + let collisions = probe_collisions(&output_uri, &samples, None).unwrap(); assert_eq!( collisions, vec![ @@ -147,7 +143,7 @@ mod tests { let collisions = probe_collisions( &dir.path().to_string_lossy(), - &["/data/run.d".to_string()], + &crate::sample_identity::resolve_samples(&["/data/run.d".to_string()]).unwrap(), Some(&library), ) .unwrap(); diff --git a/rust/timsseek_cli/src/main.rs b/rust/timsseek_cli/src/main.rs index ef5ef117..4ae1828c 100644 --- a/rust/timsseek_cli/src/main.rs +++ b/rust/timsseek_cli/src/main.rs @@ -11,6 +11,7 @@ mod output_sink; mod predicted_library; mod processing; mod run_inputs; +mod sample_identity; mod search; use clap::Parser; diff --git a/rust/timsseek_cli/src/output_sink.rs b/rust/timsseek_cli/src/output_sink.rs index 338c3aa1..d2edd514 100644 --- a/rust/timsseek_cli/src/output_sink.rs +++ b/rust/timsseek_cli/src/output_sink.rs @@ -192,57 +192,6 @@ pub(crate) fn probe_uri_exists(uri: &str) -> Result { }) } -/// `sample.d.tar`, `sample.d/`, `sample.d.idx/` all collapse to `sample`. -pub(crate) fn sample_name_from_uri(uri: &str) -> Option { - let trimmed = uri.trim_end_matches('/'); - let mut stem = trimmed.rsplit('/').next()?; - // Loop so chained suffixes collapse fully. Order matters: `.idx`/`.tar` - // come off before `.d` so they can't leave a bare `.d` behind. - loop { - let before = stem; - for ext in [".idx", ".tar", ".d"] { - if let Some(s) = stem.strip_suffix(ext) { - stem = s; - } - } - if stem == before { - break; - } - } - if stem.is_empty() { - None - } else { - Some(stem.to_string()) - } -} - -#[cfg(test)] -mod sample_name_tests { - use super::sample_name_from_uri; - #[test] - fn local_dotd_plain() { - assert_eq!(sample_name_from_uri("/data/run.d").as_deref(), Some("run")); - } - #[test] - fn local_dotd_trailing_slash() { - assert_eq!(sample_name_from_uri("/data/run.d/").as_deref(), Some("run")); - } - #[test] - fn s3_tar_collapses_both_suffixes() { - assert_eq!( - sample_name_from_uri("s3://bkt/run.d.tar").as_deref(), - Some("run") - ); - } - #[test] - fn s3_idx_directory() { - assert_eq!( - sample_name_from_uri("s3://bkt/run.d.idx/").as_deref(), - Some("run") - ); - } -} - #[cfg(test)] mod destination_tests { use super::OutputSink; diff --git a/rust/timsseek_cli/src/processing.rs b/rust/timsseek_cli/src/processing.rs index 762b2675..2398a6db 100644 --- a/rust/timsseek_cli/src/processing.rs +++ b/rust/timsseek_cli/src/processing.rs @@ -119,9 +119,13 @@ fn execute_raw_pipeline( source, path: Some(path.clone()), }; - let mut writer = - timsseek::scoring::parquet_writer::ResultParquetWriter::raw(&path, 20_000, library) - .map_err(io_error)?; + let mut writer = timsseek::scoring::parquet_writer::ResultParquetWriter::raw( + &path, + 20_000, + library, + options.sample.identity(), + ) + .map_err(io_error)?; let mut report = PipelineReport { raw_scores: true, ..Default::default() @@ -153,6 +157,7 @@ fn execute_raw_pipeline( tracing::instrument(skip_all, level = "trace") )] pub struct PipelineOptions<'a> { + pub sample: &'a crate::sample_identity::SampleInput, pub chunk_size: usize, pub output: &'a OutputConfig, pub max_qvalue: f32, @@ -175,6 +180,7 @@ pub fn execute_pipeline( return execute_raw_pipeline(speclib, pipeline, options); } let PipelineOptions { + sample, chunk_size, output: out_path, max_qvalue, @@ -374,12 +380,16 @@ pub fn execute_pipeline( // === PHASE 6: Write Parquet output === let step = TimedStep::begin("Phase 6: Write output"); let out_path_pq = std::path::Path::new(&out_path.uri).join(RESULTS_PARQUET); - let mut pq_writer = - timsseek::scoring::parquet_writer::ResultParquetWriter::new(&out_path_pq, 20_000, speclib) - .map_err(|e| TimsSeekError::Io { - path: out_path_pq.clone().into(), - source: e, - })?; + let mut pq_writer = timsseek::scoring::parquet_writer::ResultParquetWriter::rescored( + &out_path_pq, + 20_000, + speclib, + sample.identity(), + ) + .map_err(|e| TimsSeekError::Io { + path: out_path_pq.clone().into(), + source: e, + })?; for res in data.into_iter() { if res.qvalue <= max_qvalue { pq_writer.add(res).map_err(|e| TimsSeekError::Io { @@ -704,10 +714,20 @@ pub fn run_pipeline( let mut timings = execute_pipeline(speclib, calib_lib, pipeline, options)?; timings.load_index_ms = load_index_ms; // Write per-file report - let perf_report = - serde_json::to_string_pretty(&timings).map_err(|e| TimsSeekError::ParseError { - msg: format!("Error serializing performance report to JSON: {}", e), - })?; + #[derive(serde::Serialize)] + struct SamplePerformanceReport<'a> { + #[serde(flatten)] + sample: &'a timsseek::sample_identity::SampleIdentity, + #[serde(flatten)] + timings: &'a PipelineReport, + } + let perf_report = serde_json::to_string_pretty(&SamplePerformanceReport { + sample: options.sample.identity(), + timings: &timings, + }) + .map_err(|e| TimsSeekError::ParseError { + msg: format!("Error serializing performance report to JSON: {}", e), + })?; std::fs::write(&performance_report_path, perf_report).map_err(|e| TimsSeekError::Io { path: performance_report_path.into(), source: e, diff --git a/rust/timsseek_cli/src/sample_identity.rs b/rust/timsseek_cli/src/sample_identity.rs new file mode 100644 index 00000000..b110f1e1 --- /dev/null +++ b/rust/timsseek_cli/src/sample_identity.rs @@ -0,0 +1,166 @@ +//! Stable sample identity from the input location, never from its contents. + +use std::collections::HashMap; + +use crate::errors::CliError; +use timsseek::sample_identity::SampleIdentity; + +#[derive(Debug, Clone)] +pub(crate) struct SampleInput { + uri: String, + identity: SampleIdentity, + source: tims_stage::load::PreparedSource, +} + +impl SampleInput { + pub(crate) fn source(&self) -> &tims_stage::load::PreparedSource { + &self.source + } + + pub(crate) fn uri(&self) -> &str { + &self.uri + } + + pub(crate) fn identity(&self) -> &SampleIdentity { + &self.identity + } + + pub(crate) fn sample_id(&self) -> &str { + self.identity.sample_id() + } + + pub(crate) fn sample_name(&self) -> &str { + self.identity.sample_name() + } +} + +/// Resolve once, before staging changes an input's location. Local paths become +/// absolute without resolving symlinks; remote URI text is preserved. The parent +/// includes its trailing slash. Supported storage suffixes are not identity. +pub(crate) fn resolve_samples(uris: &[String]) -> Result, CliError> { + let mut seen = HashMap::new(); + let mut samples = Vec::with_capacity(uris.len()); + for uri in uris { + let expanded = tims_stage::expand_local_uri(uri); + let invalid = || CliError::Config { + source: format!("Unable to derive sample identity from input {uri:?}"), + }; + if expanded + .trim_end_matches(['/', '\\']) + .rsplit(['/', '\\']) + .next() + .is_none_or(|name| matches!(name, "" | "." | "..")) + { + return Err(invalid()); + } + let location = if tims_stage::is_remote_uri(&expanded) { + expanded + } else { + let absolute = std::path::absolute(&expanded).map_err(|e| CliError::Io { + source: format!("Resolving sample input: {e}"), + path: Some(uri.clone()), + })?; + let text = absolute.to_str().ok_or_else(invalid)?; + if cfg!(windows) { + text.replace('\\', "/") + } else { + text.to_owned() + } + }; + let source = + tims_stage::load::PreparedSource::new(&location).map_err(|e| CliError::Config { + source: e.to_string(), + })?; + let identity = SampleIdentity::from_source(&source).map_err(|e| CliError::Config { + source: e.to_string(), + })?; + let sample_id = identity.sample_id(); + if let Some(previous) = seen.insert(sample_id.to_owned(), uri) { + return Err(CliError::Config { + source: format!( + "Duplicate sample_id {sample_id:?} for inputs {previous:?} and {uri:?}; \ + duplicate identities are unsupported, including with --overwrite" + ), + }); + } + samples.push(SampleInput { + uri: uri.clone(), + identity, + source, + }); + } + Ok(samples) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn resolve(uris: &[&str]) -> Result, CliError> { + resolve_samples(&uris.iter().map(|s| (*s).to_owned()).collect::>()) + } + + #[test] + fn independent_runs_and_reordering_keep_identity() { + let uris = [ + "s3://bucket/rerun_1/my-run.d", + "s3://bucket/rerun_2/my-run.d", + ]; + let samples = resolve(&uris).unwrap(); + assert_ne!(samples[0].sample_id(), samples[1].sample_id()); + assert_eq!(samples[0].sample_name(), "my-run"); + assert_eq!( + samples[0].identity(), + resolve(&uris[..1]).unwrap()[0].identity() + ); + assert_eq!( + samples[0].identity(), + resolve(&[uris[1], uris[0]]).unwrap()[1].identity() + ); + } + + #[test] + fn storage_forms_share_identity_and_collide() { + let reference = resolve(&["s3://bucket/run.d"]).unwrap().remove(0); + for suffix in [".d/", ".d.tar", ".d.idx/", ".D.IDX/"] { + let uri = format!("s3://bucket/run{suffix}"); + assert_eq!( + resolve(&[&uri]).unwrap()[0].sample_id(), + reference.sample_id() + ); + let error = resolve(&[&reference.uri, &uri]).unwrap_err().to_string(); + assert!(error.contains(&reference.uri)); + assert!(error.contains(&uri)); + assert!(error.contains("Duplicate sample_id")); + } + } + + #[test] + fn relative_and_absolute_local_paths_agree_without_opening_input() { + let relative = "a1-nonexistent/run.d"; + let absolute = std::env::current_dir().unwrap().join(relative); + assert_eq!( + resolve(&[relative]).unwrap()[0].sample_id(), + resolve(&[absolute.to_str().unwrap()]).unwrap()[0].sample_id() + ); + assert!(resolve(&[relative, absolute.to_str().unwrap()]).is_err()); + let others = resolve(&["a1-nonexistent/run.d", "another-nonexistent/run.d"]).unwrap(); + assert_ne!(others[0].sample_id(), others[1].sample_id()); + } + + #[test] + fn missing_or_unsafe_names_are_rejected() { + for uri in [ + "", + "/", + ".", + "..", + "s3://bucket/", + "s3:///run.d", + "s3://bucket/.d", + "s3://bucket/..", + ] { + assert!(resolve(&[uri]).is_err(), "{uri}"); + } + } +} diff --git a/rust/timsseek_cli/src/search.rs b/rust/timsseek_cli/src/search.rs index b15663ae..06a1b35d 100644 --- a/rust/timsseek_cli/src/search.rs +++ b/rust/timsseek_cli/src/search.rs @@ -5,11 +5,8 @@ use tims_stage::{ expand_local_uri, is_remote_uri, }; +use timsquery::IndexedTimstofPeaks; use timsquery::utils::TupleRange; -use timsquery::{ - IndexedTimstofPeaks, - load_index, -}; use timsseek::scoring::Scorer; use timsseek::scoring::timings::TimedStep; use tracing::{ @@ -26,15 +23,16 @@ use crate::config::{ load_config, }; use crate::logging::init_tracing; -use crate::output_sink::{ - OutputSink, - sample_name_from_uri, -}; +use crate::output_sink::OutputSink; use crate::run_inputs::{ LibrarySource, ResolvedInputs, resolve_run_inputs, }; +use crate::sample_identity::{ + SampleInput, + resolve_samples, +}; use crate::{ artifacts, build_library, @@ -72,7 +70,12 @@ fn validate_fasta(fasta: &std::path::Path) -> std::result::Result<(), errors::Cl /// Probe the filesystem for everything the run is about to touch, so a missing /// input or a colliding artifact fails before the heavy analysis rather than /// after it. Every value it reads was resolved by [`resolve_run_inputs`]. -fn validate_inputs(resolved: &ResolvedInputs) -> std::result::Result<(), errors::CliError> { +fn validate_inputs( + resolved: &ResolvedInputs, +) -> std::result::Result, errors::CliError> { + // Unconditional: --overwrite permits replacing old artifacts, never two + // inputs in this invocation writing to the same destination. + let samples = resolve_samples(&resolved.raw_inputs)?; info!("Validating inputs and outputs before processing..."); let ResolvedInputs { @@ -165,7 +168,7 @@ fn validate_inputs(resolved: &ResolvedInputs) -> std::result::Result<(), errors: LibrarySource::Build { out, .. } => Some(out.as_path()), LibrarySource::File(_) | LibrarySource::Fasta(_) => None, }; - let collisions = artifacts::probe_collisions(output_uri, raw_inputs, built_library)?; + let collisions = artifacts::probe_collisions(output_uri, &samples, built_library)?; if !collisions.is_empty() { let list = collisions .iter() @@ -192,7 +195,7 @@ fn validate_inputs(resolved: &ResolvedInputs) -> std::result::Result<(), errors: info!("All validations passed! Starting processing..."); - Ok(()) + Ok(samples) } /// Record what the run resolved to, at the path the collision probe reserved. @@ -241,8 +244,9 @@ struct SearchRun<'a> { impl SearchRun<'_> { fn process_file( &self, - raw_uri: &str, + sample: &SampleInput, ) -> std::result::Result { + let raw_uri = sample.uri(); let file_name = std::path::Path::new(raw_uri) .file_name() .and_then(|s| s.to_str()) @@ -251,8 +255,8 @@ impl SearchRun<'_> { info!("Processing raw input: {}", raw_uri); let step = TimedStep::begin("Loading index"); - let (index, index_source) = load_index( - raw_uri, + let (index, index_source) = timsquery::serde::index_serde::load_index_for_source( + sample.source(), self.backend, self.save_sidecar, self.config @@ -293,11 +297,7 @@ impl SearchRun<'_> { fragmented_range, }; - let file_stem = sample_name_from_uri(raw_uri).ok_or_else(|| errors::CliError::Io { - source: "Unable to derive sample name from URI".to_string(), - path: Some(raw_uri.to_string()), - })?; - let file_output_dir = self.sink.sample_dir(&file_stem); + let file_output_dir = self.sink.sample_dir(sample.sample_id()); std::fs::create_dir_all(&file_output_dir).map_err(|e| errors::CliError::Io { source: format!("Failed to create output subdirectory: {}", e), @@ -305,7 +305,7 @@ impl SearchRun<'_> { })?; if self.overwrite { - self.sink.clear_existing(&file_stem)?; + self.sink.clear_existing(sample.sample_id())?; } let file_output_config = OutputConfig { @@ -313,6 +313,7 @@ impl SearchRun<'_> { }; let options = processing::PipelineOptions { + sample, chunk_size: self.config.analysis.chunk_size, output: &file_output_config, max_qvalue: self.max_qvalue, @@ -432,7 +433,7 @@ pub(crate) fn search(args: &SearchArgs) -> std::result::Result<(), errors::CliEr info!("Parsed configuration: {:#?}", config.clone()); alloc_track::snap!("start"); - validate_inputs(&validated)?; + let samples = validate_inputs(&validated)?; // The stale-tempdir sweep runs inside `PerRunTempdir::new`. let staging_cfg = config.staging.clone().unwrap_or_default(); @@ -576,7 +577,8 @@ pub(crate) fn search(args: &SearchArgs) -> std::result::Result<(), errors::CliEr no_feature_stats: args.no_feature_stats, }; - for (idx, raw_uri) in validated.raw_inputs.iter().enumerate() { + for (idx, sample) in samples.iter().enumerate() { + let raw_uri = sample.uri(); info!( "Processing input {} of {}: {}", idx + 1, @@ -584,28 +586,17 @@ pub(crate) fn search(args: &SearchArgs) -> std::result::Result<(), errors::CliEr raw_uri ); - let sample_name = match sample_name_from_uri(raw_uri) { - Some(s) => s, - None => { - let e = errors::CliError::Io { - source: "Unable to derive sample name from URI".to_string(), - path: Some(raw_uri.clone()), - }; - error!("Failed to process {}: {}", raw_uri, e); - failed_files.push((raw_uri.clone(), e)); - continue; - } - }; + let sample_name = sample.sample_name(); // Header/footer pair; `processing::run_pipeline` phase output lands // between them, so batched runs still show per-input wall time. println!("=== [{}/{}] {} ===", idx + 1, total_files, sample_name); let file_start = std::time::Instant::now(); - let sample_dest = sink.dest_uri_for(&sample_name); + let sample_dest = sink.dest_uri_for(sample.sample_id()); - match search_run.process_file(raw_uri) { + match search_run.process_file(sample) { Ok(report) => { - if let Err(e) = sink.finalize_sample(&sample_name) { + if let Err(e) = sink.finalize_sample(sample.sample_id()) { error!("Failed to finalize sample {}: {}", sample_name, e); println!( "=== [{}/{}] {} failed upload after {:?} ===", @@ -617,7 +608,7 @@ pub(crate) fn search(args: &SearchArgs) -> std::result::Result<(), errors::CliEr run_report.status = timsseek::scoring::timings::RunStatus::Aborted; run_report.abort_reason = Some(format!("upload failure on sample {sample_name}: {e}")); - failed_files.push((raw_uri.clone(), e)); + failed_files.push((raw_uri.to_owned(), e)); error!("Aborting batch due to upload failure"); break; } @@ -629,7 +620,7 @@ pub(crate) fn search(args: &SearchArgs) -> std::result::Result<(), errors::CliEr sample_name, file_start.elapsed() ); - successful_files.push(raw_uri.clone()); + successful_files.push(raw_uri.to_owned()); let mut outputs = vec![format!("{sample_dest}/{}", artifacts::RESULTS_PARQUET)]; if !args.no_feature_stats { outputs.push(format!("{sample_dest}/{}", artifacts::FEATURE_STATS_TSV)); @@ -639,7 +630,8 @@ pub(crate) fn search(args: &SearchArgs) -> std::result::Result<(), errors::CliEr )); } run_report.files.push(timsseek::scoring::FileReport { - file_name: raw_uri.clone(), + sample: sample.identity().clone(), + file_name: raw_uri.to_owned(), pipeline: report, outputs, }); @@ -659,11 +651,11 @@ pub(crate) fn search(args: &SearchArgs) -> std::result::Result<(), errors::CliEr if matches!(e, errors::CliError::Io { .. }) { run_report.status = timsseek::scoring::timings::RunStatus::Aborted; run_report.abort_reason = Some(format!("I/O error on {raw_uri}: {e}")); - failed_files.push((raw_uri.clone(), e)); + failed_files.push((raw_uri.to_owned(), e)); error!("Aborting batch due to I/O error"); break; } - failed_files.push((raw_uri.clone(), e)); + failed_files.push((raw_uri.to_owned(), e)); } } } @@ -702,6 +694,85 @@ pub(crate) fn search(args: &SearchArgs) -> std::result::Result<(), errors::CliEr mod tests { use super::*; + #[test] + fn duplicate_id_fails_before_io_even_with_overwrite() { + for overwrite in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("not-created"); + let resolved = ResolvedInputs { + raw_inputs: vec!["s3://bucket/run.d".into(), "s3://bucket/run.d.idx/".into()], + library: LibrarySource::Fasta(dir.path().join("absent.fasta")), + calib_lib_uri: None, + output_uri: output.to_string_lossy().into_owned(), + overwrite, + }; + let error = validate_inputs(&resolved).unwrap_err().to_string(); + assert!(error.contains("Duplicate sample_id"), "{error}"); + assert!(error.contains("s3://bucket/run.d.idx/"), "{error}"); + assert!(!output.exists()); + } + } + + #[test] + fn same_stem_inputs_have_separate_artifact_destinations_and_report_identity() { + let dir = tempfile::tempdir().unwrap(); + let library = dir.path().join("library"); + std::fs::write(&library, b"not opened by preflight").unwrap(); + let resolved = ResolvedInputs { + raw_inputs: vec![ + "s3://bucket/rerun_1/my-run.d".into(), + "s3://bucket/rerun_2/my-run.d".into(), + ], + library: LibrarySource::File(library.to_string_lossy().into_owned()), + calib_lib_uri: None, + output_uri: dir.path().join("out").to_string_lossy().into_owned(), + overwrite: false, + }; + let samples = validate_inputs(&resolved).unwrap(); + let sink = OutputSink::new(&resolved.output_uri).unwrap(); + for sample in &samples { + let sample_dir = sink.sample_dir(sample.sample_id()); + std::fs::create_dir_all(&sample_dir).unwrap(); + std::fs::write(sample_dir.join(artifacts::RESULTS_PARQUET), sample.uri()).unwrap(); + let report = timsseek::scoring::FileReport { + sample: sample.identity().clone(), + file_name: sample.uri().to_owned(), + pipeline: Default::default(), + outputs: vec![sink.dest_uri_for(sample.sample_id())], + }; + let json = serde_json::to_value(report).unwrap(); + assert_eq!(json["sample_id"], sample.sample_id()); + assert_eq!(json["sample_name"], "my-run"); + } + assert_eq!( + artifacts::probe_collisions(&resolved.output_uri, &samples, None) + .unwrap() + .len(), + 2 + ); + // A second invocation is different from duplicate inputs in one batch: + // pre-existing outputs are explicitly replaceable with --overwrite. + assert!(validate_inputs(&resolved).is_err()); + let rerun = ResolvedInputs { + overwrite: true, + ..resolved.clone() + }; + let rerun_samples = validate_inputs(&rerun).unwrap(); + assert_eq!( + rerun_samples + .iter() + .map(|s| s.identity()) + .collect::>(), + samples.iter().map(|s| s.identity()).collect::>() + ); + sink.clear_existing(samples[0].sample_id()).unwrap(); + assert!( + sink.sample_dir(samples[1].sample_id()) + .join(artifacts::RESULTS_PARQUET) + .exists() + ); + } + /// The first run of a script that names a library it has not built yet, and /// what it is told: the file is missing, and here is what would produce it. #[test] diff --git a/rust/timsseek_cli/tests/sample_identity.rs b/rust/timsseek_cli/tests/sample_identity.rs new file mode 100644 index 00000000..6fec94e6 --- /dev/null +++ b/rust/timsseek_cli/tests/sample_identity.rs @@ -0,0 +1,33 @@ +use std::process::Command; + +#[test] +fn cli_rejects_duplicate_samples_before_opening_inputs_with_or_without_overwrite() { + for overwrite in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("output"); + let mut command = Command::new(env!("CARGO_BIN_EXE_timsseek")); + command.args([ + "--raw-inputs", + "s3://bucket/run.d", + "--raw-inputs", + "s3://bucket/run.d.idx/", + "--speclib-uri", + "absent-library.mzspeclib.txt", + "--output-uri", + output.to_str().unwrap(), + "--log-path", + "-", + ]); + if overwrite { + command.arg("--overwrite"); + } + let result = command.output().unwrap(); + assert_eq!(result.status.code(), Some(1)); + let stderr = String::from_utf8_lossy(&result.stderr); + assert!(stderr.contains("Duplicate sample_id"), "{stderr}"); + assert!(stderr.contains("s3://bucket/run.d"), "{stderr}"); + assert!(stderr.contains("s3://bucket/run.d.idx/"), "{stderr}"); + assert!(!stderr.contains("panicked"), "{stderr}"); + assert!(!output.exists()); + } +}