diff --git a/DevDocMD/implementation/partition_layer_cache.md b/DevDocMD/implementation/partition_layer_cache.md index c4d5deee..ee42564a 100644 --- a/DevDocMD/implementation/partition_layer_cache.md +++ b/DevDocMD/implementation/partition_layer_cache.md @@ -56,8 +56,12 @@ algorithm, extracted from `PartitionRouter` the same way `Dereplicator` was in (4). (7) done — `LayerBuilder`, the fourth and last pipeline algorithm; the indexing pipeline is now fully decomposed into `obikindexer::algorithms::{partitionner, dereplicator, counter, -layer_builder}`. (5) itself still not implemented, still first on the -"order of remaining work" list — see "(7) done" below. Earlier mix-up, for +layer_builder}`. (8) design agreed, item 1 (`obikindexer::extensions:: +IndexBuilder`) done in (9) — private extension trait, six construction-only +`KmerIndex` methods moved out; item 2 (`obikalgorithm::Algorithm`) still +not started. Note: `Layer` renamed `KmerLayer` (2026-08-21, outside this +conversation). (5) itself still not implemented, still first on the +"order of remaining work" list. Earlier mix-up, for context: an earlier version of this doc used the name `KmerPartition` (singular) for what was actually the *collection* type (later renamed `KmerPartitions`, later @@ -952,6 +956,198 @@ variants observed, worth revisiting whether a single trait can express all three or whether that's itself the answer: it can't, and the trait should not force it). +## (8) design agreed, not yet implemented (2026-08-21): `obikalgorithm::Algorithm` trait + `obikindexer::extensions` — private/public extension-trait split, `KmerLayer` rename + +Session note: `Layer` was renamed `KmerLayer` (user, outside this +conversation, alongside other naming homogenisation with `KmerIndex`/ +`KmerPartition`) — every reference to `Layer` in this doc from before +2026-08-21 means today's `obikindex::layer::KmerLayer`. + +### Why this came up + +Verifying "does `cmd/index` now rest entirely on the algorithm structs" +(it doesn't quite — see below) led to sorting `KmerIndex`'s own methods by +a criterion the user was explicit is **semantic, not mechanical**: "les +méthodes qui, sémantiquement, n'ont pas d'intérêt hors de la construction +de l'index" (methods that have no semantic interest outside index +construction) — not "methods only called from `cmd/index` today," which +a grep could answer but would miss methods construction-adjacent code +elsewhere (`merge`/`select`/`rebuild`/`reindex`) also depends on for the +same reason. + +**Checked, not assumed** (grepped every call site before classifying): + +- **Construction-only, real candidates for a private extension trait**: + `KmerIndex::{mark_scattered, mark_counted, mark_indexed, write_spectrum, + build_index_layer, remove_build_artifacts, clear_output_for_create, + create_skeleton, finalize_indexed, state}`. The last four are called + from `merge.rs`/`select.rs`/`rebuild.rs`/`reindex.rs` too (as + precondition checks — "is my source `Indexed`?" — or shared + skeleton/finalize machinery), not just from the 4-stage pipeline — so + this extension trait's scope is "construction of any kind," not + narrowly "the initial build pipeline." +- **Looked construction-only by name, checked, and kept on `KmerIndex`**: + `layer_unitigs_path` (unitigs are the only way to recover a built + index's kmer sequences — read by `rebuild_layer.rs` and others, well + beyond construction — see [[project_unitigs_always_kept]]), + `pack_matrices` (re-runnable maintenance on an already-finished index + via `obikmer pack`, not just a pipeline step), `upgrade_layer_meta` + (migration, runnable on any existing index at any time). + +### The general pattern (not obikindexer-specific) + +`KmerIndex`/`KmerPartition`/`KmerLayer` stay generic, in `obikindex` — +every domain-specific consumer crate gets to attach its own extension +trait(s), of two kinds: + +- **Private** (`pub(crate)`, invisible outside the defining crate) — for + plumbing only that crate's own algorithms need. `obikindexer` gets + exactly one of these (see below); no public counterpart makes sense for + it — "l'index est tellement central que le second trait n'a pas + vraiment d'intérêt" for construction specifically: nothing external + should ever want to call `mark_scattered` or `build_index_layer`. +- **Public** — for a genuinely reusable domain extension. The user's own + example, found while discussing this, not hypothetical: `obikindex/src/ + index/distance.rs` (phylogenetic distance metrics) is currently an + `impl KmerIndex` block **inside `obikindex` itself** — under this + principle it should be a public extension trait owned by `obikphylo` + instead (distance metrics are a phylo concept, `obikindex` has no more + business defining them than `obikindex::layer` has defining + "family"/"minorant", the reasoning `SiblingLayerExt` already followed + for `KmerLayer` — see `obikphylo/src/siblings/iter.rs`). **Explicitly + deferred** — noted here so it isn't lost, not part of this round. +- The future cache-manager crate (still blocked on (5), see above) will + add its own **public** extension trait mirroring part of `KmerIndex`'s/ + `KmerPartition`'s own read API in cached form (e.g. a cached + `.partition(i)` that doesn't re-touch disk) — same pattern, third data + point once built. + +### Concretely, next to implement (two items, in order) + +1. **`obikindexer::extensions`** — a private (`pub(crate)`) extension + trait, most likely named something like `IndexBuildExt` (final name + not yet chosen), implemented for `KmerIndex`, carrying the ten methods + listed above, moved out of `obikindex::index::{kmer_index, + index_layer}`. Every algorithm in `obikindexer::algorithms::*` that + currently calls `idx.mark_scattered()`/etc. keeps the same call syntax + (extension trait methods are called the same way as inherent ones, + just need the trait in scope) — `cmd/index/mod.rs` itself would need + `use obikindexer::extensions::IndexBuildExt;` (or the module re-exports + it) to keep compiling, since it's the one place outside `obikindexer`'s + own algorithms that currently calls `mark_scattered`/`write_spectrum`/ + `mark_counted`/`mark_indexed` directly. **Not yet decided**: exact + trait name, whether it's one trait or split further (e.g. sentinel + marking vs. skeleton/finalize machinery), and whether `merge`/`select`/ + `rebuild`/`reindex` (not yet extracted into algorithms themselves) move + onto it now too or keep calling the soon-to-be-inherent-no-longer + methods some other way in the meantime — **ask before implementing**, + this changes the blast radius significantly (4 more `obikindex` + internal files depend on `clear_output_for_create`/`create_skeleton`/ + `finalize_indexed`/`state`). +2. **`obikalgorithm::Algorithm` trait** — new crate, the shared trait + `obikpartitionner`→`obikindexer` merge (session start of 2026-08-21) + and (6)/(7) were deliberately building toward, now with four real + `new`/(setters)/`run` examples and three distinct callback-bound + shapes to reconcile (plain `FnMut` for `PartitionRouter`, `FnMut + + Send` for `LayerBuilder`, `Fn + Sync` for `Dereplicator`/`Counter` — + see (7)). Exact shape not yet drafted in this doc — do that as its own + design pass before coding, same discipline as everything above. + +Both items: **design only, nothing implemented yet** — this section is +the record to resume from, not a plan already executed. + +## (9) done (2026-08-21): `obikindexer::extensions::IndexBuilder` — item 1 above, implemented + +Scoped down from (8)'s six-method list to the concrete set that's +genuinely movable without further ripple — checked, not assumed, before +writing anything: + +```rust +pub(crate) trait IndexBuilder { + fn mark_scattered(&mut self) -> OKIResult<()>; + fn mark_counted(&self) -> OKIResult<()>; + fn mark_indexed(&self) -> OKIResult<()>; + fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap) -> OKIResult<()>; + fn build_index_layer(&self, i: usize, min_ab: u32, max_ab: Option, with_counts: bool, mode: &IndexMode, block_bits: u8) -> Result; + fn remove_build_artifacts(&self, i: usize); +} +impl IndexBuilder for KmerIndex { ... } +``` + +All six moved bodily out of `obikindex::index::{kmer_index, index_layer}` +into `obikindexer::extensions` (new module, `pub(crate)`) — `index_layer.rs` +is now empty and deleted outright. +`clear_output_for_create`/`create_skeleton`/`finalize_indexed`/`state` +stayed inherent on `KmerIndex`, per (8)'s reasoning: `merge`/`select`/ +`rebuild`/`reindex` — living *inside* `obikindex` itself — call them too, +and `obikindex` can never depend on `obikindexer` to reach a trait defined +there. Moving those four is real future work (extract +merge/select/rebuild/reindex into algorithms first), not part of this +step. + +**One new, small, deliberate API widening in `obikindex`**: `build_index_layer` +depends on three helpers that were `pub(crate)` to `obikindex` +(`graph_pipeline::{write_graph_as_unitigs, materialize_layer}`, +`common::olm_to_sk`) — widened to `pub` (re-exported from `obikindex`'s +crate root) so `obikindexer` could reach them. This is exactly the +"enrich shared/lower-level APIs instead of ad hoc local code" call the +project's own rules ask for, made explicitly rather than routed around: +three functions, already generically written (no rewrite needed), now +serve a second caller instead of being duplicated. + +**Why the trait had to be defined in `obikindexer`, not `obikindex`**: +Rust's orphan rule — implementing a trait for a foreign type requires +either the trait or the type to be local to the current crate. `KmerIndex` +is foreign to `obikindexer`, so the trait must be the local half; if it +were defined in `obikindex` instead, `pub(crate)` there would make it +invisible to `obikindexer` too (crate-private means private to *that* +crate, not "private except to one named dependent") — the opposite of +what was wanted. + +**A real design decision made while wiring callers up, not a mechanical +rename**: `IndexBuilder` being genuinely `pub(crate)` to `obikindexer` +means `obikmer::cmd::index` (a different crate) can no longer call +`mark_scattered`/`mark_counted`/`mark_indexed`/`write_spectrum` directly — +it never could have, once privacy was real rather than aspirational. Each +algorithm now marks its own completion as part of `run()`/`close()` +instead of leaving it to the caller: +- `PartitionRouter::close()` (not `run()`) calls `mark_scattered()` — + `close()`, not `run()`, is the actual shared completion point between + the file-driven `run()` path and the manual `write`/`write_batch`+ + `close()` path low-level callers (tests) use; putting it in `run()` + alone would have silently skipped marking for every caller that never + calls `run()`. `run()` already calls `self.close()` at its own end, so + this covers both paths through one line, not two. +- `Counter::run` calls `write_spectrum` then `mark_counted` before + returning. +- `LayerBuilder::run` calls `mark_indexed` before returning. + +`cmd/index/mod.rs` lost all four direct calls (`mark_scattered`/ +`write_spectrum`/`mark_counted`/`mark_indexed`) — each stage's `if +idx.state() < IndexState::X { ... }` block is now purely "run the +algorithm," no separate bookkeeping call after it. Confirms, precisely +this time (checked by re-reading the whole file, not assumed): `cmd/index` +now rests on the four algorithms for every read/write of pipeline state +except `KmerIndex::{exists, create, state, n_partitions}`, which are +genuinely index-identity concerns, not construction bookkeeping — the +original question this whole design pass started from. + +Same fix applied to `obikphylo`'s test harness (its four explicit +`mark_*`/`write_spectrum` calls removed, relying on the algorithms now +doing it themselves) — `obikindexer::algorithms::partitionner`'s own +`pipeline_counts` test needed no change (never called `mark_*` directly). + +Full workspace suite green (`cargo check --workspace --all-targets` + +`cargo test --workspace`, exit code 0), plus the CLI smoke test — 870 +kmers, same as (6)/(7). + +Still not done: item 2 from (8) (`obikalgorithm::Algorithm`), (5), the +future cache crate, the `distance.rs` → `obikphylo` relocation (noted in +(8), explicitly deferred), and extracting +`merge`/`select`/`rebuild`/`reindex` into algorithms (which would unblock +moving `clear_output_for_create`/`create_skeleton`/`finalize_indexed`/ +`state` the same way). + ## The problem Reading a layer's data (MPHF + matrix) is not free: `MphfLayer::open` mmaps diff --git a/src/Cargo.lock b/src/Cargo.lock index f9ac1471..878a7162 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -1563,6 +1563,7 @@ dependencies = [ "memmap2", "niffler", "obicompactvec", + "obidebruinj", "obikindex", "obikrope", "obikseq", diff --git a/src/obikindex/src/index/common.rs b/src/obikindex/src/index/common.rs index 1a8a3914..cfa20609 100644 --- a/src/obikindex/src/index/common.rs +++ b/src/obikindex/src/index/common.rs @@ -7,7 +7,7 @@ use obiskio::{SKError, SKResult}; // ── olm_to_sk ──────────────────────────────────────────────────────────────── -pub(crate) fn olm_to_sk(e: OLMError, context: &'static str) -> SKError { +pub fn olm_to_sk(e: OLMError, context: &'static str) -> SKError { match e { OLMError::Io(e) => SKError::Io(e), other => SKError::InvalidData { diff --git a/src/obikindex/src/index/graph_pipeline.rs b/src/obikindex/src/index/graph_pipeline.rs index 52ffed73..307436d8 100644 --- a/src/obikindex/src/index/graph_pipeline.rs +++ b/src/obikindex/src/index/graph_pipeline.rs @@ -121,7 +121,7 @@ where /// Phase 2 (write unitigs only): compute degrees, write unitigs to `layer_dir`, drop graph. /// /// Returns n_kmers. Does NOT build the MPHF — caller does it. -pub(crate) fn write_graph_as_unitigs(g: GraphDeBruijn, layer_dir: &Path) -> SKResult { +pub fn write_graph_as_unitigs(g: GraphDeBruijn, layer_dir: &Path) -> SKResult { let n_kmers = g.len(); g.compute_degrees_and_mark_starts(); std::fs::create_dir_all(layer_dir)?; @@ -137,7 +137,7 @@ pub(crate) fn write_graph_as_unitigs(g: GraphDeBruijn, layer_dir: &Path) -> SKRe /// Phase 2 (full): write_graph_as_unitigs + `TypedLayer::<()>::build`. /// /// Returns n_kmers. -pub(crate) fn materialize_layer( +pub fn materialize_layer( g: GraphDeBruijn, layer_dir: &Path, block_bits: u8, diff --git a/src/obikindex/src/index/index_layer.rs b/src/obikindex/src/index/index_layer.rs deleted file mode 100644 index bf234c9e..00000000 --- a/src/obikindex/src/index/index_layer.rs +++ /dev/null @@ -1,132 +0,0 @@ -use std::fs; -use std::io; - -use cacheline_ef::{CachelineEf, CachelineEfVec}; -use epserde::prelude::*; -use obicompactvec::{PersistentCompactIntMatrix, PersistentCompactIntVec}; -use obidebruinj::GraphDeBruijn; -use crate::layer::meta::PartitionMeta; -use crate::layer::{IndexMode, TypedLayer}; -use obiskio::{SKError, SKFileMeta, SKFileReader}; -use ptr_hash::{PtrHash, bucket_fn::CubicEps, hash::Xx64}; - -use crate::index::common::olm_to_sk; -use crate::index::graph_pipeline::{materialize_layer, write_graph_as_unitigs}; -use crate::index::kmer_index::KmerIndex; - -type Mphf = PtrHash>, Xx64, Vec>; - -fn remove_if_exists(path: &std::path::Path) { - if let Err(e) = fs::remove_file(path) { - if e.kind() != io::ErrorKind::NotFound { - eprintln!("warning: could not remove {}: {e}", path.display()); - } - } -} - -impl KmerIndex { - /// Build the layered MPHF index for partition `i`. - /// - /// Returns the number of canonical k-mers indexed, or 0 if the partition - /// has no data or its layer was already built (resume-safe). - /// - /// Abundance filtering is applied when `min_ab > 1` or `max_ab.is_some()`, - /// using `mphf1.bin` + `counts1.bin` if they exist. - /// Count payload is stored iff `with_counts` is true. - pub fn build_index_layer( - &self, - i: usize, - min_ab: u32, - max_ab: Option, - with_counts: bool, - mode: &IndexMode, - block_bits: u8, - ) -> Result { - let layer0_dir = self.layer_dir(i, 0); - let dedup_path = crate::layer::dereplicated_superkmers_path(&layer0_dir); - if !dedup_path.exists() { - return Ok(0); - } - - if layer0_dir.join("mphf.bin").exists() { - return Ok(0); - } - - let filter_active = min_ab > 1 || max_ab.is_some(); - let need_counts = filter_active || with_counts; - - let mphf1_opt: Option = if need_counts { - let p = layer0_dir.join("mphf1.bin"); - p.exists().then(|| Mphf::load_full(&p).ok()).flatten() - } else { - None - }; - - let counts1_opt: Option = if need_counts { - let p = layer0_dir.join("counts1.bin"); - p.exists() - .then(|| PersistentCompactIntVec::open(&p).ok()) - .flatten() - } else { - None - }; - - let mut g = GraphDeBruijn::new(); - let mut reader = SKFileReader::open(&dedup_path)?; - for sk in reader.iter() { - for kmer in sk.iter_canonical_kmers() { - let accept = if filter_active { - match (&mphf1_opt, &counts1_opt) { - (Some(mphf), Some(counts)) => { - let ab = counts.get(mphf.index(&kmer.raw())); - ab >= min_ab && max_ab.map_or(true, |max| ab <= max) - } - _ => true, - } - } else { - true - }; - if accept { - g.push(kmer); - } - } - } - - let n_kmers = - if with_counts { - let n = write_graph_as_unitigs(g, &layer0_dir)?; - TypedLayer::::build(&layer0_dir, block_bits, mode, |kmer| { - match (&mphf1_opt, &counts1_opt) { - (Some(mphf), Some(counts)) => counts.get(mphf.index(&kmer.raw())), - _ => 1, - } - }) - .map_err(|e| olm_to_sk(e, "layer build"))?; - n - } else { - materialize_layer(g, &layer0_dir, block_bits, mode)? - }; - - let index_dir = layer0_dir.parent().expect("layer_dir has a parent"); - PartitionMeta { - n_layers: 1, - mode: mode.clone(), - } - .save(index_dir) - .map_err(|e| olm_to_sk(e, "layer build"))?; - - Ok(n_kmers) - } - - /// Remove intermediate build artifacts for partition `i`. - /// - /// Deletes `dereplicated.skmer.zst` (+ sidecar), `mphf1.bin`, `counts1.bin`. - pub fn remove_build_artifacts(&self, i: usize) { - let layer0_dir = self.layer_dir(i, 0); - let dedup = crate::layer::dereplicated_superkmers_path(&layer0_dir); - remove_if_exists(&SKFileMeta::sidecar_path(&dedup)); - remove_if_exists(&dedup); - remove_if_exists(&layer0_dir.join("mphf1.bin")); - remove_if_exists(&layer0_dir.join("counts1.bin")); - } -} diff --git a/src/obikindex/src/index/kmer_index.rs b/src/obikindex/src/index/kmer_index.rs index f6a4f747..8153f034 100644 --- a/src/obikindex/src/index/kmer_index.rs +++ b/src/obikindex/src/index/kmer_index.rs @@ -1,4 +1,3 @@ -use std::collections::BTreeMap; use std::fs; use std::path::{Path, PathBuf}; @@ -11,7 +10,7 @@ use obikseq::{set_k, set_m}; use crate::index::common::load_meta; use crate::index::error::{OKIError, OKIResult}; use crate::index::meta::{GenomeInfo, IndexConfig, IndexMeta}; -use crate::index::state::{IndexState, SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED}; +use crate::index::state::{IndexState, SENTINEL_INDEXED}; pub struct KmerIndex { pub(crate) root_path: PathBuf, @@ -209,62 +208,6 @@ impl KmerIndex { Ok(self.n_layers(0)?) } - /// Mark scatter as complete and write `scatter.done`. - /// - /// If no genome label was set at creation time, one is derived from - /// the index root directory name (stripped of all extensions). - pub fn mark_scattered(&mut self) -> OKIResult<()> { - if self.meta.genomes.is_empty() { - let label = label_from_path(&self.root_path); - self.meta.genomes.push(GenomeInfo::new(label)); - self.meta.write(&self.root_path)?; - } - touch(&self.root_path.join(SENTINEL_SCATTERED))?; - Ok(()) - } - - /// Mark dereplicate+count as complete and write `count.done`. - pub fn mark_counted(&self) -> OKIResult<()> { - touch(&self.root_path.join(SENTINEL_COUNTED))?; - Ok(()) - } - - /// Mark layer construction as complete and write `index.done`. - pub fn mark_indexed(&self) -> OKIResult<()> { - touch(&self.root_path.join(SENTINEL_INDEXED))?; - Ok(()) - } - - /// Write `spectrums/{label}.json` from an already-computed kmer - /// spectrum (`f0`/`f1`/abundance histogram). Takes plain values rather - /// than `obikindexer::algorithms::partitionner::KmerSpectrum` — `KmerIndex` - /// is the data model, `PartitionRouter` the algorithm that depends on - /// it (see `DevDocMD/implementation/partition_layer_cache.md`), not - /// the other way around, and this is the only field of that type it - /// actually uses. - pub fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap) -> OKIResult<()> { - let label = self - .meta - .genomes - .first() - .map(|g| g.label.as_str()) - .unwrap_or("unknown"); - let spectrums_dir = self.root_path.join("spectrums"); - fs::create_dir_all(&spectrums_dir)?; - let path = spectrums_dir.join(format!("{label}.json")); - let spectrum_map: BTreeMap = counts - .iter() - .map(|(&c, &f)| (format!("{c:010}"), f)) - .collect(); - let f = fs::File::create(&path)?; - serde_json::to_writer_pretty( - f, - &serde_json::json!({ "f0": f0, "f1": f1, "spectrum": spectrum_map }), - ) - .map_err(OKIError::Json)?; - Ok(()) - } - /// Path to the unitigs file for partition `part`, layer `layer`. pub fn layer_unitigs_path(&self, part: usize, layer: usize) -> PathBuf { self.layer_dir(part, layer).join("unitigs.bin") @@ -370,23 +313,4 @@ impl KmerIndex { } } -fn label_from_path(path: &Path) -> String { - let name = path - .file_name() - .unwrap_or(path.as_os_str()) - .to_string_lossy() - .into_owned(); - let mut s = name; - while let Some(pos) = s.rfind('.') { - s.truncate(pos); - } - if s.is_empty() { - "unknown".to_string() - } else { - s - } -} -fn touch(path: &Path) -> Result<(), std::io::Error> { - fs::File::create(path).map(|_| ()) -} diff --git a/src/obikindex/src/index/mod.rs b/src/obikindex/src/index/mod.rs index 3fe91c5a..6803cb4f 100644 --- a/src/obikindex/src/index/mod.rs +++ b/src/obikindex/src/index/mod.rs @@ -9,7 +9,6 @@ mod dump_layer; pub mod filter; mod graph_pipeline; mod kmer_index; -mod index_layer; mod matrix_store; mod merge; mod merge_layer; @@ -23,6 +22,8 @@ mod select_layer; mod stats; pub use error::{OKIError, OKIResult}; +pub use common::olm_to_sk; +pub use graph_pipeline::{materialize_layer, write_graph_as_unitigs}; pub use distance::{DistanceMetric, DistanceOutput}; pub use filter::{GroupQuorumFilter, KmerFilter, passes_all}; pub use kmer_index::KmerIndex; diff --git a/src/obikindex/src/lib.rs b/src/obikindex/src/lib.rs index e63e54dd..7309fdc0 100644 --- a/src/obikindex/src/lib.rs +++ b/src/obikindex/src/lib.rs @@ -17,6 +17,6 @@ pub use index::{ GroupQuorumFilter, IndexBitsPerKmer, IndexConfig, IndexMeta, IndexState, KmerDesc, KmerFilter, KmerIndex, MergeMode, MetaPred, OKIError, OKIResult, OutputCol, PartitionRunner, QueryHit, QueryStats, META_FILENAME, SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED, - passes_all, + materialize_layer, olm_to_sk, write_graph_as_unitigs, passes_all, }; pub use index::{filter, meta}; diff --git a/src/obikindexer/Cargo.toml b/src/obikindexer/Cargo.toml index 9a91f209..2ad8111a 100644 --- a/src/obikindexer/Cargo.toml +++ b/src/obikindexer/Cargo.toml @@ -6,6 +6,7 @@ edition = "2024" [dependencies] obikindex = { path = "../obikindex" } obikseq = { path = "../obikseq" } +obidebruinj = { path = "../obidebruinj" } obiskio = { path = "../obiskio" } obisys = { path = "../obisys" } obicompactvec = { path = "../obicompactvec" } diff --git a/src/obikindexer/src/algorithms/counter/mod.rs b/src/obikindexer/src/algorithms/counter/mod.rs index 7319dc20..eb3b737b 100644 --- a/src/obikindexer/src/algorithms/counter/mod.rs +++ b/src/obikindexer/src/algorithms/counter/mod.rs @@ -22,6 +22,8 @@ use sysinfo::System; use count::count_partition; use kmer_sort::chunk_size_from_ram; +use crate::extensions::PrivateBuilder; + pub struct KmerSpectrum { pub f0: u64, pub f1: u64, @@ -67,7 +69,9 @@ impl<'a> Counter<'a> { /// `FnMut`, same reason as `Dereplicator::run`: the counting pass itself /// is a parallel `par_iter`, so the callback must tolerate concurrent /// calls. `total: Some(n_partitions)` — known up front. This algorithm - /// never renders anything itself. + /// never renders anything itself. Writes `spectrums/{label}.json` and + /// marks the index as counted (`count.done`) once every partition + /// succeeds. pub fn run(&self, on_progress: Option) -> SKResult { let sys = System::new_all(); // available_memory() can return 0 on macOS when the compressor page count exceeds @@ -92,7 +96,10 @@ impl<'a> Counter<'a> { }; if let Some(cb) = &on_progress { let pos = done.fetch_add(1, Ordering::Relaxed) + 1; - cb(Progress { position: pos, total: Some(self.n_partitions as u64) }); + cb(Progress { + position: pos, + total: Some(self.n_partitions as u64), + }); } result }) @@ -128,6 +135,13 @@ impl<'a> Counter<'a> { } } + self.index + .write_spectrum(f0, f1, &counts) + .map_err(|e| io::Error::other(e.to_string()))?; + self.index + .mark_counted() + .map_err(|e| io::Error::other(e.to_string()))?; + Ok(KmerSpectrum { f0, f1, counts }) } } diff --git a/src/obikindexer/src/algorithms/layer_builder/mod.rs b/src/obikindexer/src/algorithms/layer_builder/mod.rs index 4ffc3218..9f745e5c 100644 --- a/src/obikindexer/src/algorithms/layer_builder/mod.rs +++ b/src/obikindexer/src/algorithms/layer_builder/mod.rs @@ -8,6 +8,8 @@ use obikindex::{KmerIndex, PartitionRunner}; use obiskio::SKResult; use obisys::Progress; +use crate::extensions::PrivateBuilder; + /// Builds every partition's real layer 0 in parallel. /// /// Two-phase construction, same shape as the other pipeline algorithms: @@ -65,8 +67,9 @@ impl<'a> LayerBuilder<'a> { /// Build every partition's layer 0 in parallel via `PartitionRunner` /// (NUMA-aware scheduling — this stage is more CPU/memory-intensive /// per partition than scatter/dereplicate/count, unlike those it - /// doesn't use plain rayon). Returns the total number of kmers built - /// across all partitions. + /// doesn't use plain rayon). Marks the index as fully built + /// (`index.done`) once every partition succeeds. Returns the total + /// number of kmers built across all partitions. /// /// `on_progress`, when set, is called once per completed partition — /// `FnMut + Send`, not `Fn + Sync`: `PartitionRunner::run` calls @@ -88,12 +91,18 @@ impl<'a> LayerBuilder<'a> { let runner = PartitionRunner::new(); runner.run( &order, - |i| self.index.build_index_layer(i, min_ab, max_ab, with_counts, &evidence, block_bits), + |i| { + self.index + .build_index_layer(i, min_ab, max_ab, with_counts, &evidence, block_bits) + }, |_i, n_kmers, _elapsed| { total_kmers += n_kmers; done += 1; if let Some(cb) = on_progress.as_mut() { - cb(Progress { position: done, total: Some(self.n_partitions as u64) }); + cb(Progress { + position: done, + total: Some(self.n_partitions as u64), + }); } }, )?; @@ -104,6 +113,10 @@ impl<'a> LayerBuilder<'a> { } } + self.index + .mark_indexed() + .map_err(|e| std::io::Error::other(e.to_string()))?; + Ok(total_kmers) } } diff --git a/src/obikindexer/src/algorithms/partitionner/router.rs b/src/obikindexer/src/algorithms/partitionner/router.rs index 22e3a1cf..0b22fb50 100644 --- a/src/obikindexer/src/algorithms/partitionner/router.rs +++ b/src/obikindexer/src/algorithms/partitionner/router.rs @@ -18,6 +18,8 @@ use obiskio::SKFileWriter; use obipipeline::{ThrottleGuard, Throttled, throttle}; use obiread::NucPage; +use crate::extensions::PrivateBuilder; + // ── Pipeline plumbing, private to `run` ───────────────────────────────────── /// Carrier enum for `obipipeline::make_pipe!`'s two-stage transform — local @@ -160,6 +162,12 @@ impl<'a> PartitionRouter<'a> { Ok(()) } + /// Closes every open writer and marks scatter as complete + /// (`scatter.done`) — the single completion point shared by `run` + /// (the file-driven pipeline) and manual `write`/`write_batch` callers + /// (e.g. tests feeding in-memory superkmers directly): whichever path + /// was used, `close` is what both agree means "scatter is done". + /// Idempotent — a second call is a no-op, sentinel included. pub fn close(&mut self) -> SKResult<()> { if self.closed { return Ok(()); @@ -168,6 +176,9 @@ impl<'a> PartitionRouter<'a> { for writer in self.writers.iter_mut().flatten() { writer.close()?; } + self.index + .mark_scattered() + .map_err(|e| std::io::Error::other(e.to_string()))?; Ok(()) } diff --git a/src/obikindexer/src/extensions/mod.rs b/src/obikindexer/src/extensions/mod.rs new file mode 100644 index 00000000..148ada26 --- /dev/null +++ b/src/obikindexer/src/extensions/mod.rs @@ -0,0 +1,261 @@ +//! Construction-only `KmerIndex` operations — semantically meaningless +//! once an index is built and being used for reading (query, dump, phylo, +//! or as a `merge`/`select`/`rebuild` *source*): sentinel marking, the +//! per-partition layer-build primitive, and cleanup of build-time scratch +//! files. Private to this crate — no consumer outside the +//! indexing-pipeline algorithms in `obikindexer::algorithms` should ever +//! need them. See `DevDocMD/implementation/partition_layer_cache.md`, +//! "(8)". +//! +//! Defined here, not in `obikindex`, so it can be genuinely `pub(crate)` +//! while extending a foreign type (`KmerIndex`) — Rust's orphan rule +//! requires the trait itself to be local to the crate that implements it +//! for a foreign type; `obikindex` has no way to make a trait "private +//! except to `obikindexer`" from its own side. +//! +//! Scoped to the six methods exclusive to this crate's own pipeline +//! (`partitionner`/`dereplicator`/`counter`/`layer_builder`). +//! `clear_output_for_create`/`create_skeleton`/`finalize_indexed`/`state` +//! stay inherent on `KmerIndex`, deliberately not moved here: they're also +//! called from `obikindex`'s own `merge`/`select`/`rebuild`/`reindex` +//! modules, which — living inside `obikindex` itself — could never reach a +//! trait defined in `obikindexer` (the dependency only runs +//! `obikindexer → obikindex`, never the other way). Moving those four +//! would require extracting `merge`/`select`/`rebuild`/`reindex` into +//! algorithms first — separate, larger future work, not this one. + +use std::collections::BTreeMap; +use std::fs; +use std::io; +use std::path::Path; + +use cacheline_ef::{CachelineEf, CachelineEfVec}; +use epserde::prelude::*; +use obicompactvec::{PersistentCompactIntMatrix, PersistentCompactIntVec}; +use obidebruinj::GraphDeBruijn; +use obikindex::layer::IndexMode; +use obikindex::layer::{TypedLayer, meta::PartitionMeta}; +use obikindex::{GenomeInfo, KmerIndex, OKIError, OKIResult}; +use obikindex::{SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED}; +use obikindex::{materialize_layer, olm_to_sk, write_graph_as_unitigs}; +use obiskio::{SKError, SKFileMeta, SKFileReader}; +use ptr_hash::{PtrHash, bucket_fn::CubicEps, hash::Xx64}; + +type Mphf = PtrHash>, Xx64, Vec>; + +pub(crate) trait PrivateBuilder { + /// Mark scatter as complete and write `scatter.done`. + /// + /// If no genome label was set at creation time, one is derived from + /// the index root directory name (stripped of all extensions). + fn mark_scattered(&mut self) -> OKIResult<()>; + + /// Mark dereplicate+count as complete and write `count.done`. + fn mark_counted(&self) -> OKIResult<()>; + + /// Mark layer construction as complete and write `index.done`. + fn mark_indexed(&self) -> OKIResult<()>; + + /// Write `spectrums/{label}.json` from an already-computed kmer + /// spectrum (`f0`/`f1`/abundance histogram). Takes plain values rather + /// than `algorithms::counter::KmerSpectrum` — `KmerIndex` is the data + /// model, `Counter` the algorithm that depends on it, not the other + /// way around, and this is the only field of that type it actually + /// uses. + fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap) -> OKIResult<()>; + + /// Build the layered MPHF index for partition `i`. + /// + /// Returns the number of canonical k-mers indexed, or 0 if the + /// partition has no data or its layer was already built + /// (resume-safe). + /// + /// Abundance filtering is applied when `min_ab > 1` or + /// `max_ab.is_some()`, using `mphf1.bin` + `counts1.bin` if they + /// exist. Count payload is stored iff `with_counts` is true. + fn build_index_layer( + &self, + i: usize, + min_ab: u32, + max_ab: Option, + with_counts: bool, + mode: &IndexMode, + block_bits: u8, + ) -> Result; + + /// Remove intermediate build artifacts for partition `i`: dereplicated + /// superkmers (+ sidecar), `mphf1.bin`, `counts1.bin`. + fn remove_build_artifacts(&self, i: usize); +} + +impl PrivateBuilder for KmerIndex { + fn mark_scattered(&mut self) -> OKIResult<()> { + if self.meta().genomes.is_empty() { + let label = label_from_path(self.root_path()); + self.meta_mut().genomes.push(GenomeInfo::new(label)); + self.meta().write(self.root_path())?; + } + touch(&self.root_path().join(SENTINEL_SCATTERED))?; + Ok(()) + } + + fn mark_counted(&self) -> OKIResult<()> { + touch(&self.root_path().join(SENTINEL_COUNTED))?; + Ok(()) + } + + fn mark_indexed(&self) -> OKIResult<()> { + touch(&self.root_path().join(SENTINEL_INDEXED))?; + Ok(()) + } + + fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap) -> OKIResult<()> { + let label = self + .meta() + .genomes + .first() + .map(|g| g.label.as_str()) + .unwrap_or("unknown"); + let spectrums_dir = self.root_path().join("spectrums"); + fs::create_dir_all(&spectrums_dir)?; + let path = spectrums_dir.join(format!("{label}.json")); + let spectrum_map: BTreeMap = counts + .iter() + .map(|(&c, &f)| (format!("{c:010}"), f)) + .collect(); + let f = fs::File::create(&path)?; + serde_json::to_writer_pretty( + f, + &serde_json::json!({ "f0": f0, "f1": f1, "spectrum": spectrum_map }), + ) + .map_err(OKIError::Json)?; + Ok(()) + } + + fn build_index_layer( + &self, + i: usize, + min_ab: u32, + max_ab: Option, + with_counts: bool, + mode: &IndexMode, + block_bits: u8, + ) -> Result { + let layer0_dir = self.layer_dir(i, 0); + let dedup_path = obikindex::layer::dereplicated_superkmers_path(&layer0_dir); + if !dedup_path.exists() { + return Ok(0); + } + + if layer0_dir.join("mphf.bin").exists() { + return Ok(0); + } + + let filter_active = min_ab > 1 || max_ab.is_some(); + let need_counts = filter_active || with_counts; + + let mphf1_opt: Option = if need_counts { + let p = layer0_dir.join("mphf1.bin"); + p.exists().then(|| Mphf::load_full(&p).ok()).flatten() + } else { + None + }; + + let counts1_opt: Option = if need_counts { + let p = layer0_dir.join("counts1.bin"); + p.exists() + .then(|| PersistentCompactIntVec::open(&p).ok()) + .flatten() + } else { + None + }; + + let mut g = GraphDeBruijn::new(); + let mut reader = SKFileReader::open(&dedup_path)?; + for sk in reader.iter() { + for kmer in sk.iter_canonical_kmers() { + let accept = if filter_active { + match (&mphf1_opt, &counts1_opt) { + (Some(mphf), Some(counts)) => { + let ab = counts.get(mphf.index(&kmer.raw())); + ab >= min_ab && max_ab.map_or(true, |max| ab <= max) + } + _ => true, + } + } else { + true + }; + if accept { + g.push(kmer); + } + } + } + + let n_kmers = if with_counts { + let n = write_graph_as_unitigs(g, &layer0_dir)?; + TypedLayer::::build( + &layer0_dir, + block_bits, + mode, + |kmer| match (&mphf1_opt, &counts1_opt) { + (Some(mphf), Some(counts)) => counts.get(mphf.index(&kmer.raw())), + _ => 1, + }, + ) + .map_err(|e| olm_to_sk(e, "layer build"))?; + n + } else { + materialize_layer(g, &layer0_dir, block_bits, mode)? + }; + + let index_dir = layer0_dir.parent().expect("layer_dir has a parent"); + PartitionMeta { + n_layers: 1, + mode: mode.clone(), + } + .save(index_dir) + .map_err(|e| olm_to_sk(e, "layer build"))?; + + Ok(n_kmers) + } + + fn remove_build_artifacts(&self, i: usize) { + let layer0_dir = self.layer_dir(i, 0); + let dedup = obikindex::layer::dereplicated_superkmers_path(&layer0_dir); + remove_if_exists(&SKFileMeta::sidecar_path(&dedup)); + remove_if_exists(&dedup); + remove_if_exists(&layer0_dir.join("mphf1.bin")); + remove_if_exists(&layer0_dir.join("counts1.bin")); + } +} + +// ── private helpers ────────────────────────────────────────────────────── + +fn label_from_path(path: &Path) -> String { + let name = path + .file_name() + .unwrap_or(path.as_os_str()) + .to_string_lossy() + .into_owned(); + let mut s = name; + while let Some(pos) = s.rfind('.') { + s.truncate(pos); + } + if s.is_empty() { + "unknown".to_string() + } else { + s + } +} + +fn touch(path: &Path) -> Result<(), io::Error> { + fs::File::create(path).map(|_| ()) +} + +fn remove_if_exists(path: &Path) { + if let Err(e) = fs::remove_file(path) { + if e.kind() != io::ErrorKind::NotFound { + eprintln!("warning: could not remove {}: {e}", path.display()); + } + } +} diff --git a/src/obikindexer/src/lib.rs b/src/obikindexer/src/lib.rs index ecb9789e..0bda5d6d 100644 --- a/src/obikindexer/src/lib.rs +++ b/src/obikindexer/src/lib.rs @@ -6,8 +6,11 @@ //! and `obikindex` doesn't grow every algorithm's own dependencies. //! //! [`algorithms`] holds each algorithm as its own submodule: `partitionner` -//! (routing raw super-kmers into partitions, then counting), `dereplicator` -//! (deduplicating a partition's raw super-kmers before counting). A future -//! `extensions` module will sit alongside it. +//! (routing raw super-kmers into partitions), `dereplicator` (deduplicating +//! a partition's raw super-kmers), `counter` (counting unique canonical +//! kmers), `layer_builder` (turning dereplicated super-kmers + counts into +//! the real layer 0). [`extensions`] (private — see its own module docs) +//! holds construction-only `KmerIndex` operations these algorithms share. pub mod algorithms; +pub(crate) mod extensions; diff --git a/src/obikmer/src/cmd/index/mod.rs b/src/obikmer/src/cmd/index/mod.rs index bf8ac1a2..7203e017 100644 --- a/src/obikmer/src/cmd/index/mod.rs +++ b/src/obikmer/src/cmd/index/mod.rs @@ -280,12 +280,7 @@ pub fn run(args: IndexArgs) { }); pb.finish_and_clear(); rep.push(t.stop()); - drop(router); // ends the borrow of `idx` early — `PartitionRouter`'s `Drop` impl would otherwise extend it to the end of scope - - idx.mark_scattered().unwrap_or_else(|e| { - eprintln!("error marking scatter done: {e}"); - std::process::exit(1); - }); + drop(router); // ends the borrow of `idx` early — `PartitionRouter`'s `Drop` impl would otherwise extend it to the end of scope (`run()` already called `close()`, which marks scatter done, internally) } else { info!("scatter already done, skipping"); } @@ -305,7 +300,9 @@ pub fn run(args: IndexArgs) { let t = Stage::start("count_kmer"); let pb = progress_bar("counting", idx.n_partitions() as u64, "partitions"); - let spectrum = Counter::new(&idx) + // `Counter::run` writes `spectrums/{label}.json` and marks count + // done (`count.done`) internally once every partition succeeds. + Counter::new(&idx) .keep_partial(args.keep_intermediate) .run(Some(|_: Progress| pb.inc(1))) .unwrap_or_else(|e| { @@ -314,15 +311,6 @@ pub fn run(args: IndexArgs) { }); pb.finish_and_clear(); rep.push(t.stop()); - - idx.write_spectrum(spectrum.f0, spectrum.f1, &spectrum.counts).unwrap_or_else(|e| { - eprintln!("error: {e}"); - std::process::exit(1); - }); - idx.mark_counted().unwrap_or_else(|e| { - eprintln!("error marking count done: {e}"); - std::process::exit(1); - }); } else { info!("dereplicate+count already done, skipping"); } @@ -343,11 +331,8 @@ pub fn run(args: IndexArgs) { pb.finish_and_clear(); info!("done — {total_kmers} total kmers indexed"); rep.push(t.stop()); - - idx.mark_indexed().unwrap_or_else(|e| { - eprintln!("error marking index done: {e}"); - std::process::exit(1); - }); + // `LayerBuilder::run` marks the index done (`index.done`) internally + // once every partition succeeds. } else { info!("index already built, skipping"); } diff --git a/src/obikphylo/src/siblings/tests.rs b/src/obikphylo/src/siblings/tests.rs index cb3fae36..06a0aeba 100644 --- a/src/obikphylo/src/siblings/tests.rs +++ b/src/obikphylo/src/siblings/tests.rs @@ -77,19 +77,14 @@ fn build_single_genome_index(dir: &Path, label: &str, seq: &[u8]) -> KmerIndex { let batch = obiskbuilder::build_superkmers_page(page, K, /* level_max */ 1, /* theta */ 0.0); router.write_batch(batch).expect("write_batch"); } - router.close().expect("close partition writers"); + router.close().expect("close partition writers"); // also marks scatter done drop(router); // ends the borrow of `idx` early — `PartitionRouter`'s `Drop` impl would otherwise extend it to the end of scope Dereplicator::new(&idx).run(None::).expect("dereplicate"); - let spectrum = Counter::new(&idx).run(None::).expect("count_kmer"); - - idx.mark_scattered().expect("mark_scattered"); - idx.write_spectrum(spectrum.f0, spectrum.f1, &spectrum.counts).expect("write_spectrum"); - idx.mark_counted().expect("mark_counted"); + Counter::new(&idx).run(None::).expect("count_kmer"); // also writes the spectrum + marks counted LayerBuilder::new(&idx) .run(None::) - .expect("build_layers"); - idx.mark_indexed().expect("mark_indexed"); + .expect("build_layers"); // also marks indexed idx }