Push zunrplorkwkt #70

Merged
coissac merged 93 commits from push-zunrplorkwkt into main 2026-08-28 23:15:38 +00:00
37 changed files with 573 additions and 251 deletions
Showing only changes of commit 8d6ba6546b - Show all commits
@@ -1224,6 +1224,127 @@ correctly inherent, not construction-only by the semantic criterion) —
`create`/`open`/`exists` (identity, can't be trait methods needing `Self` `create`/`open`/`exists` (identity, can't be trait methods needing `Self`
before one exists) round that out. before one exists) round that out.
## (11) done (2026-08-21): `KmerIndex`/`IndexMeta` made fully stateless, `IndexState` moved off sentinel files
Triggered mid-discussion of `obikalgorithm::Algorithm` (still not started —
see "Still not done" below): user asked why `PartitionRouter::new` still
took `&mut KmerIndex` at all, and questioned whether `mark_scattered`
belonged in the algorithm or in `cmd/index`. Investigation found `&mut`
had become *newly* necessary since (9) — `mark_scattered` was mutating
`self.meta.genomes` in memory so `Counter`'s later `write_spectrum` call
(same `idx` instance) would see the derived label. User's resolution: the
"disk is truth, stateless" principle already agreed for `KmerPartition`/
`Layer` in (5) (still unimplemented for those two) should extend to
`KmerIndex` itself — move `IndexState` (`Empty`/`Scattered`/`Counted`/
`Indexed`) off the three sentinel files (`scatter.done`/`count.done`/
`index.done`, detected by existence) and into a field of `index.meta`'s
own JSON, so the `mark_*` calls become plain disk writes an algorithm can
legitimately make on `&self` — no in-memory mutation left to protect.
**Shape of `IndexMeta`, per the user's explicit spec**: one JSON file per
index (`index.meta`), one `IndexMeta` instance per index, held and
returned as `Arc<IndexMeta>` (not `&IndexMeta`) by `KmerIndex::meta()`.
`config` (`kmer_size`/`minimizer_size`/`n_bits`/`with_counts`/`evidence`/
`block_bits`) is fixed at construction, cached as a `pub` field (getter
kept alongside, for symmetry) — "les champs constants restent des champs
de la structure", read once, never re-read from disk. `genomes` and
`state` are the opposite: no in-memory cache at all, every accessor
(`genomes()`, `state()`) re-reads `index.meta` from disk, every mutator
(`push_genome`/`rename_genome`/`set_genomes`/`set_state`/`mark_scattered`/
`mark_counted`/`mark_indexed`) does a full read-modify-write of the same
file. An internal `std::sync::RwLock<()>` is held across each
read-modify-write sequence (not just the write) so two callers sharing
the same `Arc<IndexMeta>` can't lose an update to each other — this is
*not* a cross-process lock (that's `obisys::DirLock`, already held by
`cmd/index` for the whole build); it only serialises access through one
shared in-process instance.
**Construction, chain-of-responsibility style, matching (5)'s pattern**:
`IndexMeta::create(&KmerIndex, config, genomes)` / `IndexMeta::open(&KmerIndex)`
ask the index for its own root path rather than taking one directly. Since
`KmerIndex::create` doesn't have a complete `KmerIndex` yet to hand in
(it's what's being built), added lower-level `pub(crate)` path-based
primitives `create_at(&Path, ...)` / `open_at(&Path)` that `KmerIndex::create`/
`open` and `builder.rs`'s `create_skeleton` call directly, bypassing the
convenience wrappers for that one bootstrap case.
**The one deliberate exception**: `select_in_place` and `reindex`
genuinely rewrite `config` after an index already exists (output
type/evidence mode changes in place) — contradicting "config never
changes" for the general case. Resolved with a separate, explicitly
rare-labelled `IndexMeta::rewrite_config(config, genomes)` (preserves
`state`, overwrites everything else); callers refresh their own cached
`Arc<IndexMeta>` afterward (`self.meta = Arc::new(IndexMeta::open(self)?)`)
since `IndexMeta` has no way to reach back into whichever `KmerIndex`
holds it.
**Consequence confirmed, not just hoped for**: with `mark_scattered` no
longer touching anything in memory, `PartitionRouter` genuinely never
needs `&mut KmerIndex` — `PartitionRouter<'a> { index: &'a KmerIndex }`,
`new(&'a KmerIndex)`. This is effectively the `PartitionRouter` half of
(5)'s "order of remaining work" item done as a side effect; `KmerPartition`/
`Layer` themselves are still unimplemented for (5).
**Blast radius — much larger than (9)/(10), touched nearly every crate**:
every `.meta().genomes`/`.meta.genomes` field access became a fallible
`.genomes()?` method call (`genomes` reads `io::Result<Vec<GenomeInfo>>`
now, not a field), and `.meta_mut()` was removed outright (no more direct
field mutation from outside `IndexMeta`). Fixed across:
- `obikindex` internals: `meta.rs`/`state.rs`/`kmer_index.rs`/`builder.rs`
(full rewrites), `reindex.rs`/`select.rs` (switched to `rewrite_config`),
`merge.rs` (heaviest single file — genome counts precomputed once per
source into a `Vec<Vec<GenomeInfo>>` up front rather than re-reading
`index.meta` from disk repeatedly through the function, sentinel write
replaced with `dst2.meta.mark_indexed()`), `stats.rs`, `distance.rs`,
`dump.rs`, `predicate.rs` (its `IndexMeta`-inherent `matching_genome_indices`/
`build_group_filter` now read genomes fresh internally), `mod.rs`/`lib.rs`
(sentinel constant re-exports removed — `IndexState` no longer has
`SENTINEL_*`/`detect()` at all).
- `obikindexer::extensions::PrivateBuilder`: `mark_scattered` signature
dropped `&mut self` → `&self`; the four `mark_*`/`write_spectrum` bodies
became one-line delegations to `self.meta().mark_*()`.
- `obikphylo::siblings`: `alignment.rs`/`cardinality.rs`/`distance.rs`/
`entropy.rs`/`sankoff_bundle.rs`/`stats.rs`/`tests.rs` — all had
`self.meta().genomes.len()`-shaped reads, mechanically fixed to
`.genomes().map_err(OKIError::Io)?.len()` (tests: `.unwrap()`).
- `obikmer::cmd::*`: `annotate` (rewrote its rename path to load genomes
once, mutate the in-memory `Vec`, then `idx.meta().set_genomes(...)`
instead of `meta_mut()`), `filter`/`pack`/`dump`/`unitig`/`merge`/`select`/
`phylo` (fetch-once-then-use pattern for genome counts/labels),
`utils/maintenance.rs` (`run_rename` now calls the pre-existing
`IndexMeta::rename_genome`, dropping its own hand-rolled field mutation
entirely), `index/mod.rs` (three `idx.state() < IndexState::X`
resumability checks needed a fallible read — factored into a small
`current_state(&KmerIndex) -> IndexState` helper rather than repeating
the same `unwrap_or_else` three times), `query/*` (`emit_batch`'s
signature changed from `&IndexMeta` to `&[GenomeInfo]`, and `genomes` is
now fetched once in `run()` and threaded down through `process_chunk`
as `Arc<Vec<GenomeInfo>>` rather than re-reading `index.meta` from disk
on every chunk — a deliberate deviation from the "always re-read"
default, justified because this is a genuine per-chunk hot path, unlike
every other call site touched in this pass).
- One `&IndexMeta`-vs-`Arc<IndexMeta>` argument-type mismatch pattern
recurred at several CLI call sites (`build_filters`/`build_specs`/
`emit_batch`'s original signature) — resolved via `Arc`'s deref
coercion (`&idx.meta()` coerces to `&IndexMeta`) rather than changing
every downstream signature to accept `Arc<IndexMeta>`.
**Verification**: `cargo check --workspace --all-targets` and
`cargo test --workspace` both green (0 failures) after the full
propagation, `scripts/smoke_test_index.sh` green (870 kmers, same as every
prior round), plus a manual CLI run of `index` (×2) → `merge` → `select`
→ `reindex` → `utils --new-label` (rename) → `utils --stats`, all exit 0,
confirming the four most-affected commands (the ones (10)'s verification
already flagged as under-covered by the automated test suite) still work
end to end against the new `Arc<IndexMeta>`/on-disk-`IndexState` shape.
Still not done: (5)'s `KmerPartition`/`Layer` self-naming redesign itself
(only the `PartitionRouter`-`&mut`-removal piece landed, as a side
effect); (8)'s `obikalgorithm::Algorithm` trait (paused, not abandoned,
for this detour — points 2–4 from that discussion are still open); the
`distance.rs` → `obikphylo` relocation ((9), explicitly deferred); the
future cache-manager crate.
## The problem ## The problem
Reading a layer's data (MPHF + matrix) is not free: `MphfLayer::open` mmaps Reading a layer's data (MPHF + matrix) is not free: `MphfLayer::open` mmaps
+17 -13
View File
@@ -15,18 +15,19 @@
//! there and covers a different set of methods — the ones exclusive to //! there and covers a different set of methods — the ones exclusive to
//! the four-stage build pipeline, never needed by `merge`/`select`/ //! the four-stage build pipeline, never needed by `merge`/`select`/
//! `rebuild`/`reindex`. See //! `rebuild`/`reindex`. See
//! `DevDocMD/implementation/partition_layer_cache.md`, "(8)"/"(9)". //! `DevDocMD/implementation/partition_layer_cache.md`, "(8)"/"(9)"/"(11)".
use std::fs; use std::fs;
use std::path::Path; use std::path::Path;
use std::sync::Arc;
use obisys::Reporter; use obisys::Reporter;
use obisys::Stage; use obisys::Stage;
use crate::index::error::{OKIError, OKIResult}; use crate::index::error::{OKIError, OKIResult};
use crate::index::kmer_index::KmerIndex; use crate::index::kmer_index::KmerIndex;
use crate::index::meta::IndexMeta; use crate::index::meta::{GenomeInfo, IndexConfig, IndexMeta};
use crate::index::state::{IndexState, SENTINEL_INDEXED}; use crate::index::state::IndexState;
pub trait IndexBuilder: Sized { pub trait IndexBuilder: Sized {
/// Make `output` ready to receive a newly created index. /// Make `output` ready to receive a newly created index.
@@ -37,12 +38,13 @@ pub trait IndexBuilder: Sized {
fn clear_output_for_create<P: AsRef<Path>>(output: P, force: bool) -> OKIResult<()>; fn clear_output_for_create<P: AsRef<Path>>(output: P, force: bool) -> OKIResult<()>;
/// Lay out a fresh index skeleton at `output`: the root directory, /// Lay out a fresh index skeleton at `output`: the root directory,
/// `index.meta` (from `meta`), and an empty partition layout. /// `index.meta` (from `config` + an initial `genomes` list — often
/// empty), and an empty partition layout.
/// ///
/// For construction paths that build partitions from scratch (`select`, /// For construction paths that build partitions from scratch (`select`,
/// `rebuild`). `merge` bootstraps by copying a source index instead, so /// `rebuild`). `merge` bootstraps by copying a source index instead, so
/// it does not use this. /// it does not use this.
fn create_skeleton<P: AsRef<Path>>(output: P, meta: &IndexMeta) -> OKIResult<Self>; fn create_skeleton<P: AsRef<Path>>(output: P, config: IndexConfig, genomes: Vec<GenomeInfo>) -> OKIResult<Self>;
/// Mark `output` as fully indexed, pack its column matrices, and reopen it. /// Mark `output` as fully indexed, pack its column matrices, and reopen it.
/// ///
@@ -50,8 +52,10 @@ pub trait IndexBuilder: Sized {
/// partitions have been written. /// partitions have been written.
fn finalize_indexed<P: AsRef<Path>>(output: P, rep: &mut Reporter) -> OKIResult<Self>; fn finalize_indexed<P: AsRef<Path>>(output: P, rep: &mut Reporter) -> OKIResult<Self>;
/// Current construction state, as reported by sentinel files on disk. /// Current construction state, read fresh from `index.meta` (see
fn state(&self) -> IndexState; /// `DevDocMD/implementation/partition_layer_cache.md`, "(11)" — no
/// longer detected from sentinel files, so this is now fallible).
fn state(&self) -> OKIResult<IndexState>;
} }
impl IndexBuilder for KmerIndex { impl IndexBuilder for KmerIndex {
@@ -70,24 +74,24 @@ impl IndexBuilder for KmerIndex {
Ok(()) Ok(())
} }
fn create_skeleton<P: AsRef<Path>>(output: P, meta: &IndexMeta) -> OKIResult<Self> { fn create_skeleton<P: AsRef<Path>>(output: P, config: IndexConfig, genomes: Vec<GenomeInfo>) -> OKIResult<Self> {
let output = output.as_ref(); let output = output.as_ref();
fs::create_dir_all(output).map_err(OKIError::Io)?; fs::create_dir_all(output).map_err(OKIError::Io)?;
meta.write(output).map_err(OKIError::Io)?; let meta = IndexMeta::create_at(output, config, genomes).map_err(OKIError::Io)?;
Ok(KmerIndex { root_path: output.to_owned(), meta: meta.clone() }) Ok(KmerIndex { root_path: output.to_owned(), meta: Arc::new(meta) })
} }
fn finalize_indexed<P: AsRef<Path>>(output: P, rep: &mut Reporter) -> OKIResult<Self> { fn finalize_indexed<P: AsRef<Path>>(output: P, rep: &mut Reporter) -> OKIResult<Self> {
let output = output.as_ref(); let output = output.as_ref();
fs::File::create(output.join(SENTINEL_INDEXED)).map_err(OKIError::Io)?;
let idx = KmerIndex::open(output)?; let idx = KmerIndex::open(output)?;
idx.meta.mark_indexed().map_err(OKIError::Io)?;
let t_pack = Stage::start("pack"); let t_pack = Stage::start("pack");
idx.pack_matrices(false)?; idx.pack_matrices(false)?;
rep.push(t_pack.stop()); rep.push(t_pack.stop());
Ok(idx) Ok(idx)
} }
fn state(&self) -> IndexState { fn state(&self) -> OKIResult<IndexState> {
IndexState::detect(&self.root_path).unwrap_or(IndexState::Empty) Ok(self.meta.state().map_err(OKIError::Io)?)
} }
} }
+1 -1
View File
@@ -55,7 +55,7 @@ impl DistanceMetric {
impl KmerIndex { impl KmerIndex {
pub fn distance(&self, metric: DistanceMetric, shared_kmers: bool, presence_threshold: u32) -> OKIResult<DistanceOutput> { pub fn distance(&self, metric: DistanceMetric, shared_kmers: bool, presence_threshold: u32) -> OKIResult<DistanceOutput> {
let n_genomes = self.meta.genomes.len(); let n_genomes = self.meta.genomes().map_err(OKIError::Io)?.len();
if n_genomes < 2 { if n_genomes < 2 {
return Err(OKIError::InvalidInput( return Err(OKIError::InvalidInput(
"distance requires at least 2 genomes in the index".into(), "distance requires at least 2 genomes in the index".into(),
+2 -2
View File
@@ -30,7 +30,7 @@ impl KmerIndex {
filters: &[Box<dyn KmerFilter>], filters: &[Box<dyn KmerFilter>],
on_partition: F, on_partition: F,
) -> OKIResult<()> { ) -> OKIResult<()> {
let genomes = &self.meta.genomes; let genomes = self.meta.genomes().map_err(OKIError::Io)?;
let use_counts = self.meta.config.with_counts && !force_presence; let use_counts = self.meta.config.with_counts && !force_presence;
let n_genomes = genomes.len().max(1); let n_genomes = genomes.len().max(1);
let kmer_size = self.kmer_size(); let kmer_size = self.kmer_size();
@@ -40,7 +40,7 @@ impl KmerIndex {
write!(out, "partition,layer,")?; write!(out, "partition,layer,")?;
} }
write!(out, "kmer")?; write!(out, "kmer")?;
for g in genomes { for g in &genomes {
write!(out, ",{}", g.label)?; write!(out, ",{}", g.label)?;
} }
writeln!(out)?; writeln!(out)?;
+13 -18
View File
@@ -1,5 +1,6 @@
use std::fs; use std::fs;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::Arc;
use crate::layer::meta::PartitionMeta; use crate::layer::meta::PartitionMeta;
use obisys::progress_bar; use obisys::progress_bar;
@@ -13,7 +14,7 @@ use crate::index::meta::{GenomeInfo, IndexConfig, IndexMeta};
pub struct KmerIndex { pub struct KmerIndex {
pub(crate) root_path: PathBuf, pub(crate) root_path: PathBuf,
pub(crate) meta: IndexMeta, pub(crate) meta: Arc<IndexMeta>,
} }
impl KmerIndex { impl KmerIndex {
@@ -30,20 +31,17 @@ impl KmerIndex {
fs::create_dir_all(&root_path).map_err(OKIError::Io)?; fs::create_dir_all(&root_path).map_err(OKIError::Io)?;
set_k(config.kmer_size); set_k(config.kmer_size);
set_m(config.minimizer_size); set_m(config.minimizer_size);
let mut meta = IndexMeta::new(config); let genomes = genome_info.into_iter().collect();
if let Some(info) = genome_info { let meta = IndexMeta::create_at(&root_path, config, genomes).map_err(OKIError::Io)?;
meta.genomes.push(info); Ok(Self { root_path, meta: Arc::new(meta) })
}
meta.write(&root_path)?;
Ok(Self { root_path, meta })
} }
pub fn open<P: AsRef<Path>>(path: P) -> OKIResult<Self> { pub fn open<P: AsRef<Path>>(path: P) -> OKIResult<Self> {
let root_path = path.as_ref().to_owned(); let root_path = path.as_ref().to_owned();
let meta = IndexMeta::read(&root_path).map_err(OKIError::Io)?; let meta = IndexMeta::open_at(&root_path).map_err(OKIError::Io)?;
set_k(meta.config.kmer_size); set_k(meta.config.kmer_size);
set_m(meta.config.minimizer_size); set_m(meta.config.minimizer_size);
Ok(Self { root_path, meta }) Ok(Self { root_path, meta: Arc::new(meta) })
} }
/// Return `true` if `path` contains an `index.meta` file. /// Return `true` if `path` contains an `index.meta` file.
@@ -55,11 +53,8 @@ impl KmerIndex {
pub fn root_path(&self) -> &Path { pub fn root_path(&self) -> &Path {
&self.root_path &self.root_path
} }
pub fn meta(&self) -> &IndexMeta { pub fn meta(&self) -> Arc<IndexMeta> {
&self.meta Arc::clone(&self.meta)
}
pub fn meta_mut(&mut self) -> &mut IndexMeta {
&mut self.meta
} }
pub fn kmer_size(&self) -> usize { pub fn kmer_size(&self) -> usize {
self.meta.config.kmer_size self.meta.config.kmer_size
@@ -91,8 +86,10 @@ impl KmerIndex {
pub fn block_bits(&self) -> u8 { pub fn block_bits(&self) -> u8 {
self.meta.config.block_bits self.meta.config.block_bits
} }
pub fn genomes(&self) -> &[GenomeInfo] { /// Genome list — reads `index.meta` fresh (see `IndexMeta::genomes`,
&self.meta.genomes /// this can legitimately change during a run, unlike `config`).
pub fn genomes(&self) -> OKIResult<Vec<GenomeInfo>> {
Ok(self.meta.genomes().map_err(OKIError::Io)?)
} }
pub fn n_partitions(&self) -> usize { pub fn n_partitions(&self) -> usize {
1usize << self.meta.config.n_bits 1usize << self.meta.config.n_bits
@@ -256,5 +253,3 @@ impl KmerIndex {
Ok(()) Ok(())
} }
} }
+42 -33
View File
@@ -13,7 +13,7 @@ use crate::layer::IndexMode;
use crate::index::error::{OKIError, OKIResult}; use crate::index::error::{OKIError, OKIResult};
use crate::index::kmer_index::KmerIndex; use crate::index::kmer_index::KmerIndex;
use crate::index::meta::{GenomeInfo, IndexMeta}; use crate::index::meta::{GenomeInfo, IndexMeta};
use crate::index::state::{IndexState, SENTINEL_INDEXED}; use crate::index::state::IndexState;
pub use crate::index::merge_layer::MergeMode; pub use crate::index::merge_layer::MergeMode;
@@ -50,7 +50,7 @@ impl KmerIndex {
// ── Validate config compatibility ───────────────────────────────────── // ── Validate config compatibility ─────────────────────────────────────
let ref0 = sources[0]; let ref0 = sources[0];
for src in sources { for src in sources {
if src.state() != IndexState::Indexed { if src.state()? != IndexState::Indexed {
return Err(OKIError::NotIndexed(src.root_path.clone())); return Err(OKIError::NotIndexed(src.root_path.clone()));
} }
if src.kmer_size() != ref0.kmer_size() if src.kmer_size() != ref0.kmer_size()
@@ -64,6 +64,12 @@ impl KmerIndex {
} }
} }
// Read each source's genome list once — `IndexMeta::genomes` is a
// fresh disk read every call, so cache it rather than re-reading it
// repeatedly through the rest of this function.
let src_genomes: Vec<Vec<GenomeInfo>> =
sources.iter().map(|s| s.genomes()).collect::<OKIResult<_>>()?;
// ── Log source characteristics and choose base ──────────────────────── // ── Log source characteristics and choose base ────────────────────────
let mode_str = if mode == MergeMode::Presence { let mode_str = if mode == MergeMode::Presence {
"presence" "presence"
@@ -77,12 +83,12 @@ impl KmerIndex {
mode_str, mode_str,
); );
for (i, src) in sources.iter().enumerate() { for (i, src) in sources.iter().enumerate() {
let genome_str = if src.meta.genomes.len() == 1 { let genome_str = if src_genomes[i].len() == 1 {
"mono-genome".to_string() "mono-genome".to_string()
} else { } else {
format!("{} genomes", src.meta.genomes.len()) format!("{} genomes", src_genomes[i].len())
}; };
let trivial_str = if is_trivial(src, mode) { let trivial_str = if is_trivial(&src_genomes[i], mode) {
" [trivial: no data approximation]" " [trivial: no data approximation]"
} else { } else {
"" ""
@@ -98,9 +104,9 @@ impl KmerIndex {
); );
} }
let base_idx = choose_base(sources, mode); let base_idx = choose_base(sources, &src_genomes, mode);
let needs_approx = sources.iter().any(|src| { let needs_approx = sources.iter().enumerate().any(|(i, src)| {
!is_trivial(src, mode) !is_trivial(&src_genomes[i], mode)
&& matches!( && matches!(
src.meta.config.evidence, src.meta.config.evidence,
IndexMode::Approx { .. } | IndexMode::Hybrid { .. } IndexMode::Approx { .. } | IndexMode::Hybrid { .. }
@@ -119,17 +125,21 @@ impl KmerIndex {
); );
let mut ordered: Vec<&KmerIndex> = Vec::with_capacity(sources.len()); let mut ordered: Vec<&KmerIndex> = Vec::with_capacity(sources.len());
let mut ordered_genomes: Vec<Vec<GenomeInfo>> = Vec::with_capacity(sources.len());
ordered.push(sources[base_idx]); ordered.push(sources[base_idx]);
ordered_genomes.push(src_genomes[base_idx].clone());
for (i, &src) in sources.iter().enumerate() { for (i, &src) in sources.iter().enumerate() {
if i != base_idx { if i != base_idx {
ordered.push(src); ordered.push(src);
ordered_genomes.push(src_genomes[i].clone());
} }
} }
let sources: &[&KmerIndex] = &ordered; let sources: &[&KmerIndex] = &ordered;
let src_genomes = ordered_genomes;
let evidence = sources[0].meta.config.evidence.clone(); let evidence = sources[0].meta.config.evidence.clone();
// ── Compute final genome labels ──────────────────────────────────────── // ── Compute final genome labels ────────────────────────────────────────
let (source_labels, all_genomes) = compute_labels(sources, rename_duplicates)?; let (source_labels, all_genomes) = compute_labels(&src_genomes, rename_duplicates)?;
// ── Prepare output directory ────────────────────────────────────────── // ── Prepare output directory ──────────────────────────────────────────
KmerIndex::clear_output_for_create(output, force)?; KmerIndex::clear_output_for_create(output, force)?;
@@ -139,18 +149,18 @@ impl KmerIndex {
"bootstrap: copying {} → {} ({} genome(s))", "bootstrap: copying {} → {} ({} genome(s))",
sources[0].root_path.display(), sources[0].root_path.display(),
output.display(), output.display(),
sources[0].meta.genomes.len(), src_genomes[0].len(),
); );
let t = Stage::start("bootstrap"); let t = Stage::start("bootstrap");
let pb = spinner("bootstrap"); let pb = spinner("bootstrap");
pb.set_message("copying index …"); pb.set_message("copying index …");
copy_dir_all(&sources[0].root_path, output)?; copy_dir_all(&sources[0].root_path, output)?;
let mut meta = IndexMeta::read(output).map_err(OKIError::Io)?; let dst_meta = IndexMeta::open_at(output).map_err(OKIError::Io)?;
meta.genomes = all_genomes; let mut config = dst_meta.config.clone();
meta.config.with_counts = mode == MergeMode::Count; config.with_counts = mode == MergeMode::Count;
meta.config.evidence = evidence.clone(); config.evidence = evidence.clone();
meta.write(output)?; dst_meta.rewrite_config(config, all_genomes).map_err(OKIError::Io)?;
if mode == MergeMode::Presence { if mode == MergeMode::Presence {
remove_dirs_named(output, "counts")?; remove_dirs_named(output, "counts")?;
@@ -167,9 +177,8 @@ impl KmerIndex {
if spectrums_dir.exists() { if spectrums_dir.exists() {
fs::remove_dir_all(&spectrums_dir)?; fs::remove_dir_all(&spectrums_dir)?;
} }
for (src, new_labels) in sources.iter().zip(&source_labels) { for ((src, new_labels), genomes) in sources.iter().zip(&source_labels).zip(&src_genomes) {
let old_labels: Vec<String> = let old_labels: Vec<String> = genomes.iter().map(|g| g.label.clone()).collect();
src.meta.genomes.iter().map(|g| g.label.clone()).collect();
copy_spectrums(&src.root_path, output, &old_labels, new_labels)?; copy_spectrums(&src.root_path, output, &old_labels, new_labels)?;
} }
pb.finish_and_clear(); pb.finish_and_clear();
@@ -178,12 +187,12 @@ impl KmerIndex {
// ── Open destination ────────────────────────────────────────────────── // ── Open destination ──────────────────────────────────────────────────
let dst = KmerIndex::open(output)?; let dst = KmerIndex::open(output)?;
let n_partitions = dst.n_partitions(); let n_partitions = dst.n_partitions();
let n_dst_genomes = sources[0].meta.genomes.len(); let n_dst_genomes = src_genomes[0].len();
// ── Merge partitions ────────────────────────────────────────────────── // ── Merge partitions ──────────────────────────────────────────────────
let remaining_sources: Vec<&KmerIndex> = sources[1..].to_vec(); let remaining_sources: Vec<&KmerIndex> = sources[1..].to_vec();
if !remaining_sources.is_empty() { if !remaining_sources.is_empty() {
let n_src_genomes: usize = remaining_sources.iter().map(|s| s.meta.genomes.len()).sum(); let n_src_genomes: usize = src_genomes[1..].iter().map(|g| g.len()).sum();
info!( info!(
"merging {} partition(s) × {} additional source genome(s) into {} destination genome(s)", "merging {} partition(s) × {} additional source genome(s) into {} destination genome(s)",
n_partitions, n_src_genomes, n_dst_genomes, n_partitions, n_src_genomes, n_dst_genomes,
@@ -196,7 +205,8 @@ impl KmerIndex {
// Pre-build source list once (avoid rebuilding per partition) // Pre-build source list once (avoid rebuilding per partition)
let srcs: Vec<(&KmerIndex, usize)> = remaining_sources let srcs: Vec<(&KmerIndex, usize)> = remaining_sources
.iter() .iter()
.map(|s| (*s, s.meta.genomes.len())) .zip(&src_genomes[1..])
.map(|(s, g)| (*s, g.len()))
.collect(); .collect();
// Per-partition unitig byte sizes across remaining sources (stat() only) // Per-partition unitig byte sizes across remaining sources (stat() only)
@@ -266,12 +276,11 @@ impl KmerIndex {
pb.set_message("consolidating column files …"); pb.set_message("consolidating column files …");
let dst2 = KmerIndex::open(output)?; let dst2 = KmerIndex::open(output)?;
dst2.pack_matrices(false)?; dst2.pack_matrices(false)?;
dst2.meta.mark_indexed().map_err(OKIError::Io)?;
pb.finish_and_clear(); pb.finish_and_clear();
rep.push(t.stop()); rep.push(t.stop());
} }
fs::File::create(output.join(SENTINEL_INDEXED)).map_err(OKIError::Io)?;
KmerIndex::open(output) KmerIndex::open(output)
} }
} }
@@ -341,16 +350,16 @@ fn partition_unitig_bytes(src: &KmerIndex, i: usize) -> u64 {
} }
fn compute_labels( fn compute_labels(
sources: &[&KmerIndex], src_genomes: &[Vec<GenomeInfo>],
rename_duplicates: bool, rename_duplicates: bool,
) -> OKIResult<(Vec<Vec<String>>, Vec<GenomeInfo>)> { ) -> OKIResult<(Vec<Vec<String>>, Vec<GenomeInfo>)> {
let mut seen: HashMap<String, usize> = HashMap::new(); let mut seen: HashMap<String, usize> = HashMap::new();
let mut source_labels: Vec<Vec<String>> = Vec::with_capacity(sources.len()); let mut source_labels: Vec<Vec<String>> = Vec::with_capacity(src_genomes.len());
let mut all_genomes: Vec<GenomeInfo> = Vec::new(); let mut all_genomes: Vec<GenomeInfo> = Vec::new();
for src in sources { for genomes in src_genomes {
let mut labels = Vec::with_capacity(src.meta.genomes.len()); let mut labels = Vec::with_capacity(genomes.len());
for genome in &src.meta.genomes { for genome in genomes {
let label = &genome.label; let label = &genome.label;
let count = seen.entry(label.clone()).or_insert(0); let count = seen.entry(label.clone()).or_insert(0);
let new_label = if *count == 0 { let new_label = if *count == 0 {
@@ -414,8 +423,8 @@ fn format_evidence(ev: &IndexMode) -> String {
} }
} }
fn is_trivial(src: &KmerIndex, mode: MergeMode) -> bool { fn is_trivial(genomes: &[GenomeInfo], mode: MergeMode) -> bool {
src.meta.genomes.len() == 1 && mode == MergeMode::Presence genomes.len() == 1 && mode == MergeMode::Presence
} }
fn index_unitig_size(src: &KmerIndex) -> u64 { fn index_unitig_size(src: &KmerIndex) -> u64 {
@@ -423,9 +432,9 @@ fn index_unitig_size(src: &KmerIndex) -> u64 {
(0..n).map(|i| partition_unitig_bytes(src, i)).sum() (0..n).map(|i| partition_unitig_bytes(src, i)).sum()
} }
fn choose_base(sources: &[&KmerIndex], mode: MergeMode) -> usize { fn choose_base(sources: &[&KmerIndex], src_genomes: &[Vec<GenomeInfo>], mode: MergeMode) -> usize {
let needs_approx = sources.iter().any(|src| { let needs_approx = sources.iter().enumerate().any(|(i, src)| {
!is_trivial(src, mode) !is_trivial(&src_genomes[i], mode)
&& matches!( && matches!(
src.meta.config.evidence, src.meta.config.evidence,
IndexMode::Approx { .. } | IndexMode::Hybrid { .. } IndexMode::Approx { .. } | IndexMode::Hybrid { .. }
+193 -15
View File
@@ -1,8 +1,11 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::fs; use std::fs;
use std::io; use std::io;
use std::path::Path; use std::path::{Path, PathBuf};
use std::sync::RwLock;
use crate::index::kmer_index::KmerIndex;
use crate::index::state::IndexState;
use crate::layer::IndexMode; use crate::layer::IndexMode;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -23,6 +26,9 @@ impl GenomeInfo {
} }
} }
/// Fixed at index creation, never changes afterward — freely mutable while
/// being assembled (before any `IndexMeta::create` call), immutable once an
/// `IndexMeta` wraps it.
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IndexConfig { pub struct IndexConfig {
pub kmer_size: usize, pub kmer_size: usize,
@@ -38,36 +44,208 @@ pub struct IndexConfig {
pub block_bits: u8, pub block_bits: u8,
} }
/// On-disk shape of `index.meta` — the whole file, read/written atomically
/// as one JSON blob by every `IndexMeta` operation below.
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
struct OnDiskMeta {
version: u32,
config: IndexConfig,
#[serde(default)]
genomes: Vec<GenomeInfo>,
#[serde(default)]
state: IndexState,
}
/// Stateless accessor for one index's `index.meta` — `config` is the only
/// part cached in memory (fixed forever once written; re-reading it would
/// only ever return the same value), everything else (`genomes`, `state`)
/// is read fresh from disk on every access and written fresh on every
/// change, since it can legitimately change during a run — see
/// `DevDocMD/implementation/partition_layer_cache.md`.
///
/// One instance per index, held as `Arc<IndexMeta>` by `KmerIndex` and
/// shared by every borrower — the internal `lock` only serialises access
/// *through that one shared instance*; it is not a cross-process file lock
/// (that's `obisys::DirLock`, held by `cmd/index` for the whole build).
/// Mutating methods hold the write lock across the full read-modify-write
/// sequence, not just the write, so two concurrent callers can't silently
/// clobber each other's change.
pub struct IndexMeta { pub struct IndexMeta {
pub version: u32, root_path: PathBuf,
version: u32,
pub config: IndexConfig, pub config: IndexConfig,
/// Ordered list of genomes indexed here (label + optional categorical metadata). lock: RwLock<()>,
pub genomes: Vec<GenomeInfo>,
} }
impl IndexMeta { impl IndexMeta {
pub fn new(config: IndexConfig) -> Self { /// Create a brand-new `index.meta` for `index`, from fixed config and
Self { version: META_VERSION, config, genomes: Vec::new() } /// an initial genome list (often empty — genomes are usually added
/// later via [`push_genome`](Self::push_genome)).
pub fn create(index: &KmerIndex, config: IndexConfig, genomes: Vec<GenomeInfo>) -> io::Result<Self> {
Self::create_at(index.root_path(), config, genomes)
} }
pub fn write(&self, root: &Path) -> io::Result<()> { /// Same as [`create`](Self::create), taking a raw path — for
let file = fs::File::create(root.join(META_FILENAME))?; /// `KmerIndex::create` itself, which doesn't have a complete
serde_json::to_writer_pretty(file, self).map_err(io::Error::other) /// `KmerIndex` yet to hand in (it's what's being built).
pub(crate) fn create_at(root_path: &Path, config: IndexConfig, genomes: Vec<GenomeInfo>) -> io::Result<Self> {
let meta = Self {
root_path: root_path.to_owned(),
version: META_VERSION,
config,
lock: RwLock::new(()),
};
meta.write_full(&OnDiskMeta {
version: meta.version,
config: meta.config.clone(),
genomes,
state: IndexState::Empty,
})?;
Ok(meta)
} }
pub fn read(root: &Path) -> io::Result<Self> { /// Reopen the `index.meta` already on disk for `index`.
let file = fs::File::open(root.join(META_FILENAME))?; pub fn open(index: &KmerIndex) -> io::Result<Self> {
serde_json::from_reader(file).map_err(io::Error::other) Self::open_at(index.root_path())
}
pub(crate) fn open_at(root_path: &Path) -> io::Result<Self> {
let on_disk = Self::read_full(root_path)?;
Ok(Self {
root_path: root_path.to_owned(),
version: on_disk.version,
config: on_disk.config,
lock: RwLock::new(()),
})
} }
pub fn exists(root: &Path) -> bool { pub fn exists(root: &Path) -> bool {
root.join(META_FILENAME).exists() root.join(META_FILENAME).exists()
} }
/// Iterate over genome labels only. // ── immutable, cached — no disk access ──────────────────────────────────
pub fn genome_labels(&self) -> impl Iterator<Item = &str> {
self.genomes.iter().map(|g| g.label.as_str()) pub fn version(&self) -> u32 {
self.version
}
pub fn config(&self) -> &IndexConfig {
&self.config
}
// ── variable, stateless — every call reads or writes `index.meta` fresh ─
pub fn genomes(&self) -> io::Result<Vec<GenomeInfo>> {
let _guard = self.lock.read().unwrap();
Ok(Self::read_full(&self.root_path)?.genomes)
}
pub fn state(&self) -> io::Result<IndexState> {
let _guard = self.lock.read().unwrap();
Ok(Self::read_full(&self.root_path)?.state)
}
pub fn push_genome(&self, info: GenomeInfo) -> io::Result<()> {
let _guard = self.lock.write().unwrap();
let mut on_disk = Self::read_full(&self.root_path)?;
on_disk.genomes.push(info);
self.write_full(&on_disk)
}
pub fn rename_genome(&self, pos: usize, new_label: impl Into<String>) -> io::Result<()> {
let _guard = self.lock.write().unwrap();
let mut on_disk = Self::read_full(&self.root_path)?;
on_disk.genomes[pos].label = new_label.into();
self.write_full(&on_disk)
}
/// Replace the whole genome list at once — for construction paths
/// (`select`/`merge`) that compute the full output list up front
/// rather than pushing one at a time.
pub fn set_genomes(&self, genomes: Vec<GenomeInfo>) -> io::Result<()> {
let _guard = self.lock.write().unwrap();
let mut on_disk = Self::read_full(&self.root_path)?;
on_disk.genomes = genomes;
self.write_full(&on_disk)
}
/// Overwrite `config` and the whole `genomes` list at once, preserving
/// `state`. `config` otherwise never changes once an index exists —
/// this is the deliberate, rare exception for construction paths that
/// legitimately rewrite it in place (`select_in_place`, `reindex`).
/// The caller must refresh its own cached `Arc<IndexMeta>` afterward
/// (e.g. `self.meta = Arc::new(IndexMeta::open(self)?)`) — this method
/// only updates the file, it has no way to reach back into whatever
/// `KmerIndex` holds it.
pub fn rewrite_config(&self, config: IndexConfig, genomes: Vec<GenomeInfo>) -> io::Result<()> {
let _guard = self.lock.write().unwrap();
let state = Self::read_full(&self.root_path)?.state;
self.write_full(&OnDiskMeta { version: self.version, config, genomes, state })
}
pub fn set_state(&self, state: IndexState) -> io::Result<()> {
let _guard = self.lock.write().unwrap();
let mut on_disk = Self::read_full(&self.root_path)?;
on_disk.state = state;
self.write_full(&on_disk)
}
/// Mark scatter as complete (`IndexState::Scattered`).
///
/// If no genome label was set yet, one is derived from the index root
/// directory name (stripped of all extensions).
pub fn mark_scattered(&self) -> io::Result<()> {
let _guard = self.lock.write().unwrap();
let mut on_disk = Self::read_full(&self.root_path)?;
if on_disk.genomes.is_empty() {
let label = label_from_path(&self.root_path);
on_disk.genomes.push(GenomeInfo::new(label));
}
on_disk.state = IndexState::Scattered;
self.write_full(&on_disk)
}
pub fn mark_counted(&self) -> io::Result<()> {
let _guard = self.lock.write().unwrap();
let mut on_disk = Self::read_full(&self.root_path)?;
on_disk.state = IndexState::Counted;
self.write_full(&on_disk)
}
pub fn mark_indexed(&self) -> io::Result<()> {
let _guard = self.lock.write().unwrap();
let mut on_disk = Self::read_full(&self.root_path)?;
on_disk.state = IndexState::Indexed;
self.write_full(&on_disk)
}
// ── private ──────────────────────────────────────────────────────────────
fn read_full(root: &Path) -> io::Result<OnDiskMeta> {
let file = fs::File::open(root.join(META_FILENAME))?;
serde_json::from_reader(file).map_err(io::Error::other)
}
fn write_full(&self, on_disk: &OnDiskMeta) -> io::Result<()> {
let file = fs::File::create(self.root_path.join(META_FILENAME))?;
serde_json::to_writer_pretty(file, on_disk).map_err(io::Error::other)
}
}
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
} }
} }
+1 -1
View File
@@ -34,6 +34,6 @@ pub use meta::{validate_label, GenomeInfo, IndexConfig, IndexMeta, META_FILENAME
pub use predicate::{GroupFilterParams, MetaPred}; pub use predicate::{GroupFilterParams, MetaPred};
pub use query_layer::{KmerDesc, QueryHit, QueryStats}; pub use query_layer::{KmerDesc, QueryHit, QueryStats};
pub use select_layer::{AggOp, OutputCol}; pub use select_layer::{AggOp, OutputCol};
pub use state::{IndexState, SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED}; pub use state::IndexState;
pub use stats::IndexBitsPerKmer; pub use stats::IndexBitsPerKmer;
pub use numa::PartitionRunner; pub use numa::PartitionRunner;
+5 -3
View File
@@ -157,7 +157,8 @@ impl IndexMeta {
/// Returns indices of genomes matching `pred_str` (single predicate). /// Returns indices of genomes matching `pred_str` (single predicate).
pub fn matching_genome_indices(&self, pred_str: &str) -> Result<Vec<usize>, String> { pub fn matching_genome_indices(&self, pred_str: &str) -> Result<Vec<usize>, String> {
let pred = MetaPred::parse(pred_str)?; let pred = MetaPred::parse(pred_str)?;
Ok(self.genomes.iter().enumerate() let genomes = self.genomes().map_err(|e| e.to_string())?;
Ok(genomes.iter().enumerate()
.filter_map(|(i, g)| { .filter_map(|(i, g)| {
if g.matches(&pred) == Some(true) { Some(i) } else { std::option::Option::None } if g.matches(&pred) == Some(true) { Some(i) } else { std::option::Option::None }
}) })
@@ -176,10 +177,11 @@ impl IndexMeta {
outgroup_preds: &[MetaPred], outgroup_preds: &[MetaPred],
p: GroupFilterParams, p: GroupFilterParams,
) -> Result<GroupQuorumFilter, String> { ) -> Result<GroupQuorumFilter, String> {
let genomes = self.genomes().map_err(|e| e.to_string())?;
let (ingroup_idx, outgroup_idx) = if ingroup_preds.is_empty() && outgroup_preds.is_empty() { let (ingroup_idx, outgroup_idx) = if ingroup_preds.is_empty() && outgroup_preds.is_empty() {
((0..self.genomes.len()).collect(), vec![]) ((0..genomes.len()).collect(), vec![])
} else { } else {
let members = classify(&self.genomes, ingroup_preds, outgroup_preds); let members = classify(&genomes, ingroup_preds, outgroup_preds);
let in_idx: Vec<usize> = members.iter().enumerate() let in_idx: Vec<usize> = members.iter().enumerate()
.filter(|(_, m)| matches!(m, Membership::Ingroup)) .filter(|(_, m)| matches!(m, Membership::Ingroup))
.map(|(i, _)| i).collect(); .map(|(i, _)| i).collect();
+7 -9
View File
@@ -7,7 +7,6 @@ use tracing::info;
use crate::index::error::{OKIError, OKIResult}; use crate::index::error::{OKIError, OKIResult};
use crate::index::kmer_index::KmerIndex; use crate::index::kmer_index::KmerIndex;
use crate::index::meta::IndexMeta;
use crate::index::state::IndexState; use crate::index::state::IndexState;
impl KmerIndex { impl KmerIndex {
@@ -29,7 +28,7 @@ impl KmerIndex {
) -> OKIResult<Self> { ) -> OKIResult<Self> {
let output = output.as_ref(); let output = output.as_ref();
if src.state() != IndexState::Indexed { if src.state()? != IndexState::Indexed {
return Err(OKIError::NotIndexed(src.root_path.clone())); return Err(OKIError::NotIndexed(src.root_path.clone()));
} }
@@ -42,15 +41,16 @@ impl KmerIndex {
KmerIndex::clear_output_for_create(output, force)?; KmerIndex::clear_output_for_create(output, force)?;
// ── Create output directory + metadata ──────────────────────────────── // ── Create output directory + metadata ────────────────────────────────
let mut meta = IndexMeta::new(src.meta.config.clone()); let mut config = src.meta.config.clone();
meta.config.with_counts = mode == MergeMode::Count; config.with_counts = mode == MergeMode::Count;
meta.genomes = src.meta.genomes.clone(); let genomes = src.genomes()?;
let n_genomes = src.meta.genomes.len(); let n_genomes = genomes.len();
let n_partitions = src.n_partitions(); let n_partitions = src.n_partitions();
let block_bits = config.block_bits;
// ── Create an empty destination KmerPartition ───────────────────────── // ── Create an empty destination KmerPartition ─────────────────────────
let dst_partition = KmerIndex::create_skeleton(output, &meta)?; let dst_partition = KmerIndex::create_skeleton(output, config, genomes)?;
info!( info!(
"rebuild: {} partition(s), {} genome(s), mode={:?}", "rebuild: {} partition(s), {} genome(s), mode={:?}",
@@ -60,8 +60,6 @@ impl KmerIndex {
let t = Stage::start("rebuild"); let t = Stage::start("rebuild");
let pb = progress_bar("rebuild", n_partitions as u64, "partitions"); let pb = progress_bar("rebuild", n_partitions as u64, "partitions");
let block_bits = meta.config.block_bits;
let order: Vec<usize> = (0..n_partitions).collect(); let order: Vec<usize> = (0..n_partitions).collect();
let runner = crate::index::numa::PartitionRunner::new(); let runner = crate::index::numa::PartitionRunner::new();
runner.run( runner.run(
+10 -5
View File
@@ -1,8 +1,10 @@
use crate::index::builder::IndexBuilder; use crate::index::builder::IndexBuilder;
use crate::index::meta::IndexMeta;
use crate::layer::{IndexMode, TypedLayer}; use crate::layer::{IndexMode, TypedLayer};
use obisys::{Reporter, Stage, progress_bar}; use obisys::{Reporter, Stage, progress_bar};
use std::fs; use std::fs;
use std::path::Path; use std::path::Path;
use std::sync::Arc;
use tracing::info; use tracing::info;
use crate::index::error::{OKIError, OKIResult}; use crate::index::error::{OKIError, OKIResult};
@@ -31,7 +33,7 @@ impl KmerIndex {
block_bits: u8, block_bits: u8,
rep: &mut Reporter, rep: &mut Reporter,
) -> OKIResult<()> { ) -> OKIResult<()> {
if self.state() != IndexState::Indexed { if self.state()? != IndexState::Indexed {
return Err(OKIError::NotIndexed(self.root_path.clone())); return Err(OKIError::NotIndexed(self.root_path.clone()));
} }
@@ -59,11 +61,14 @@ impl KmerIndex {
pb.finish_and_clear(); pb.finish_and_clear();
self.meta.config.evidence = target; let mut config = self.meta.config.clone();
if matches!(self.meta.config.evidence, IndexMode::Exact) { config.evidence = target;
self.meta.config.block_bits = block_bits; if matches!(config.evidence, IndexMode::Exact) {
config.block_bits = block_bits;
} }
self.meta.write(&self.root_path)?; let genomes = self.meta.genomes().map_err(OKIError::Io)?;
self.meta.rewrite_config(config, genomes).map_err(OKIError::Io)?;
self.meta = Arc::new(IndexMeta::open(self).map_err(OKIError::Io)?);
rep.push(t.stop()); rep.push(t.stop());
Ok(()) Ok(())
} }
+14 -11
View File
@@ -1,4 +1,5 @@
use std::path::Path; use std::path::Path;
use std::sync::Arc;
use crate::index::builder::IndexBuilder; use crate::index::builder::IndexBuilder;
use crate::index::OutputCol; use crate::index::OutputCol;
@@ -28,23 +29,23 @@ impl KmerIndex {
) -> OKIResult<Self> { ) -> OKIResult<Self> {
let output = output.as_ref(); let output = output.as_ref();
if src.state() != IndexState::Indexed { if src.state()? != IndexState::Indexed {
return Err(OKIError::NotIndexed(src.root_path.clone())); return Err(OKIError::NotIndexed(src.root_path.clone()));
} }
KmerIndex::clear_output_for_create(output, force)?; KmerIndex::clear_output_for_create(output, force)?;
let mut meta = IndexMeta::new(src.meta.config.clone()); let mut config = src.meta.config.clone();
meta.config.with_counts = !output_presence; config.with_counts = !output_presence;
meta.genomes = specs let genomes: Vec<GenomeInfo> = specs
.iter() .iter()
.map(|s| GenomeInfo::new(s.label.clone())) .map(|s| GenomeInfo::new(s.label.clone()))
.collect(); .collect();
let n_src_genomes = src.meta.genomes.len(); let n_src_genomes = src.meta.genomes().map_err(OKIError::Io)?.len();
let n_partitions = src.n_partitions(); let n_partitions = src.n_partitions();
let dst_partition = KmerIndex::create_skeleton(output, &meta)?; let dst_partition = KmerIndex::create_skeleton(output, config, genomes)?;
info!( info!(
"select: {} partition(s), {} source genome(s) → {} output column(s)", "select: {} partition(s), {} source genome(s) → {} output column(s)",
@@ -94,11 +95,11 @@ impl KmerIndex {
output_presence: bool, output_presence: bool,
rep: &mut Reporter, rep: &mut Reporter,
) -> OKIResult<()> { ) -> OKIResult<()> {
if self.state() != IndexState::Indexed { if self.state()? != IndexState::Indexed {
return Err(OKIError::NotIndexed(self.root_path.clone())); return Err(OKIError::NotIndexed(self.root_path.clone()));
} }
let n_src_genomes = self.meta.genomes.len(); let n_src_genomes = self.meta.genomes().map_err(OKIError::Io)?.len();
let n_partitions = self.n_partitions(); let n_partitions = self.n_partitions();
info!( info!(
@@ -136,12 +137,14 @@ impl KmerIndex {
pb.finish_and_clear(); pb.finish_and_clear();
rep.push(t.stop()); rep.push(t.stop());
self.meta.config.with_counts = !output_presence; let mut config = self.meta.config.clone();
self.meta.genomes = specs config.with_counts = !output_presence;
let genomes: Vec<GenomeInfo> = specs
.iter() .iter()
.map(|s| GenomeInfo::new(s.label.clone())) .map(|s| GenomeInfo::new(s.label.clone()))
.collect(); .collect();
self.meta.write(&self.root_path)?; self.meta.rewrite_config(config, genomes).map_err(OKIError::Io)?;
self.meta = Arc::new(IndexMeta::open(self).map_err(OKIError::Io)?);
let t_pack = Stage::start("pack"); let t_pack = Stage::start("pack");
self.pack_matrices(false)?; self.pack_matrices(false)?;
+9 -35
View File
@@ -1,45 +1,19 @@
use std::path::Path; use serde::{Deserialize, Serialize};
use crate::index::meta::META_FILENAME; /// Progression state of a `KmerIndex` — a field of `index.meta`
/// (`IndexMeta`'s `state`), not detected from sentinel files anymore (see
pub const SENTINEL_SCATTERED: &str = "scatter.done"; /// `DevDocMD/implementation/partition_layer_cache.md`).
pub const SENTINEL_COUNTED: &str = "count.done";
pub const SENTINEL_INDEXED: &str = "index.done";
/// Progression state of a `KmerIndex`.
/// ///
/// Variants are ordered: `Empty < Scattered < Counted < Indexed`. /// Variants are ordered: `Empty < Scattered < Counted < Indexed`.
/// A state is reported only when its sentinel file is fully present — #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default, Serialize, Deserialize)]
/// partial states (e.g. scatter interrupted mid-way) are not accepted.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum IndexState { pub enum IndexState {
/// `index.meta` present; scatter not yet completed. /// `index.meta` present; scatter not yet completed.
#[default]
Empty, Empty,
/// `scatter.done` sentinel present — all super-kmers have been routed. /// All super-kmers have been routed.
Scattered, Scattered,
/// `count.done` sentinel present — dereplicate + count complete. /// Dereplicate + count complete.
Counted, Counted,
/// `index.done` sentinel present — layered MPHF index fully built. /// Layered MPHF index fully built.
Indexed, Indexed,
} }
impl IndexState {
/// Detect the state of the index at `root`.
///
/// Returns `None` if `index.meta` is absent (not an obikindex directory).
pub fn detect(root: &Path) -> Option<Self> {
if !root.join(META_FILENAME).exists() {
return None;
}
if root.join(SENTINEL_INDEXED).exists() {
return Some(Self::Indexed);
}
if root.join(SENTINEL_COUNTED).exists() {
return Some(Self::Counted);
}
if root.join(SENTINEL_SCATTERED).exists() {
return Some(Self::Scattered);
}
Some(Self::Empty)
}
}
+3 -3
View File
@@ -5,7 +5,7 @@ use obicompactvec::{LayerMeta, PersistentBitMatrix, PersistentCompactIntMatrix};
use obicompactvec::traits::ColumnWeights; use obicompactvec::traits::ColumnWeights;
use rayon::prelude::*; use rayon::prelude::*;
use crate::index::error::OKIResult; use crate::index::error::{OKIError, OKIResult};
use crate::index::kmer_index::KmerIndex; use crate::index::kmer_index::KmerIndex;
/// Bits per kmer broken down by index component. /// Bits per kmer broken down by index component.
@@ -84,7 +84,7 @@ impl KmerIndex {
/// Computation is parallelised across partitions. /// Computation is parallelised across partitions.
pub fn bits_per_kmer(&self) -> OKIResult<IndexBitsPerKmer> { pub fn bits_per_kmer(&self) -> OKIResult<IndexBitsPerKmer> {
let n = self.n_partitions(); let n = self.n_partitions();
let n_genomes = self.meta().genomes.len().max(1); let n_genomes = self.meta().genomes().map_err(OKIError::Io)?.len().max(1);
let (n_kmers, mphf_b, evidence_b, matrix_b) = (0..n) let (n_kmers, mphf_b, evidence_b, matrix_b) = (0..n)
.into_par_iter() .into_par_iter()
@@ -130,7 +130,7 @@ impl KmerIndex {
/// Partitions are scanned in parallel; results are summed across partitions. /// Partitions are scanned in parallel; results are summed across partitions.
pub fn genome_kmer_counts(&self) -> OKIResult<(usize, Vec<u64>)> { pub fn genome_kmer_counts(&self) -> OKIResult<(usize, Vec<u64>)> {
let n = self.n_partitions(); let n = self.n_partitions();
let n_genomes = self.meta.genomes.len(); let n_genomes = self.meta.genomes().map_err(OKIError::Io)?.len();
let partials: Vec<(usize, Vec<u64>)> = (0..n) let partials: Vec<(usize, Vec<u64>)> = (0..n)
.into_par_iter() .into_par_iter()
+2 -2
View File
@@ -16,7 +16,7 @@ pub use index::{
validate_label, AggOp, DistanceMetric, DistanceOutput, GenomeInfo, GroupFilterParams, validate_label, AggOp, DistanceMetric, DistanceOutput, GenomeInfo, GroupFilterParams,
GroupQuorumFilter, IndexBitsPerKmer, IndexBuilder, IndexConfig, IndexMeta, IndexState, GroupQuorumFilter, IndexBitsPerKmer, IndexBuilder, IndexConfig, IndexMeta, IndexState,
KmerDesc, KmerFilter, KmerIndex, MergeMode, MetaPred, OKIError, OKIResult, OutputCol, KmerDesc, KmerFilter, KmerIndex, MergeMode, MetaPred, OKIError, OKIResult, OutputCol,
PartitionRunner, QueryHit, QueryStats, META_FILENAME, SENTINEL_COUNTED, SENTINEL_INDEXED, PartitionRunner, QueryHit, QueryStats, META_FILENAME,
SENTINEL_SCATTERED, materialize_layer, olm_to_sk, write_graph_as_unitigs, passes_all, materialize_layer, olm_to_sk, write_graph_as_unitigs, passes_all,
}; };
pub use index::{filter, meta}; pub use index::{filter, meta};
@@ -79,7 +79,7 @@ impl Drop for GuardedIter {
/// whether) the caller displays them; rendering a spinner/progress bar is /// whether) the caller displays them; rendering a spinner/progress bar is
/// the caller's decision, not this crate's. /// the caller's decision, not this crate's.
pub struct PartitionRouter<'a> { pub struct PartitionRouter<'a> {
index: &'a mut KmerIndex, index: &'a KmerIndex,
partitions_mask: u64, partitions_mask: u64,
writers: Vec<Option<SKFileWriter>>, writers: Vec<Option<SKFileWriter>>,
level: Level, level: Level,
@@ -94,7 +94,7 @@ impl<'a> PartitionRouter<'a> {
/// Configure a router for `index`'s partition layout. Doesn't touch /// Configure a router for `index`'s partition layout. Doesn't touch
/// disk by itself — partitions and their layer-0 shells are created /// disk by itself — partitions and their layer-0 shells are created
/// lazily, on the first super-kmer routed to each of them. /// lazily, on the first super-kmer routed to each of them.
pub fn new(index: &'a mut KmerIndex) -> Self { pub fn new(index: &'a KmerIndex) -> Self {
let n_bits = index.n_bits(); let n_bits = index.n_bits();
let n_partitions = 1usize << n_bits; let n_partitions = 1usize << n_bits;
let workers = obisys::effective_parallelism(); let workers = obisys::effective_parallelism();
+16 -44
View File
@@ -35,8 +35,7 @@ use obicompactvec::{PersistentCompactIntMatrix, PersistentCompactIntVec};
use obidebruinj::GraphDeBruijn; use obidebruinj::GraphDeBruijn;
use obikindex::layer::IndexMode; use obikindex::layer::IndexMode;
use obikindex::layer::{TypedLayer, meta::PartitionMeta}; use obikindex::layer::{TypedLayer, meta::PartitionMeta};
use obikindex::{GenomeInfo, KmerIndex, OKIError, OKIResult}; use obikindex::{KmerIndex, OKIError, OKIResult};
use obikindex::{SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED};
use obikindex::{materialize_layer, olm_to_sk, write_graph_as_unitigs}; use obikindex::{materialize_layer, olm_to_sk, write_graph_as_unitigs};
use obiskio::{SKError, SKFileMeta, SKFileReader}; use obiskio::{SKError, SKFileMeta, SKFileReader};
use ptr_hash::{PtrHash, bucket_fn::CubicEps, hash::Xx64}; use ptr_hash::{PtrHash, bucket_fn::CubicEps, hash::Xx64};
@@ -44,16 +43,18 @@ use ptr_hash::{PtrHash, bucket_fn::CubicEps, hash::Xx64};
type Mphf = PtrHash<u64, CubicEps, CachelineEfVec<Vec<CachelineEf>>, Xx64, Vec<u8>>; type Mphf = PtrHash<u64, CubicEps, CachelineEfVec<Vec<CachelineEf>>, Xx64, Vec<u8>>;
pub(crate) trait PrivateBuilder { pub(crate) trait PrivateBuilder {
/// Mark scatter as complete and write `scatter.done`. /// Mark scatter as complete (`IndexState::Scattered`).
/// ///
/// If no genome label was set at creation time, one is derived from /// If no genome label was set at creation time, one is derived from
/// the index root directory name (stripped of all extensions). /// the index root directory name (stripped of all extensions). `&self`
fn mark_scattered(&mut self) -> OKIResult<()>; /// — not `&mut self` anymore: nothing is cached in memory to mutate,
/// see `DevDocMD/implementation/partition_layer_cache.md`, "(11)".
fn mark_scattered(&self) -> OKIResult<()>;
/// Mark dereplicate+count as complete and write `count.done`. /// Mark dereplicate+count as complete (`IndexState::Counted`).
fn mark_counted(&self) -> OKIResult<()>; fn mark_counted(&self) -> OKIResult<()>;
/// Mark layer construction as complete and write `index.done`. /// Mark layer construction as complete (`IndexState::Indexed`).
fn mark_indexed(&self) -> OKIResult<()>; fn mark_indexed(&self) -> OKIResult<()>;
/// Write `spectrums/{label}.json` from an already-computed kmer /// Write `spectrums/{label}.json` from an already-computed kmer
@@ -89,33 +90,25 @@ pub(crate) trait PrivateBuilder {
} }
impl PrivateBuilder for KmerIndex { impl PrivateBuilder for KmerIndex {
fn mark_scattered(&mut self) -> OKIResult<()> { fn mark_scattered(&self) -> OKIResult<()> {
if self.meta().genomes.is_empty() { self.meta().mark_scattered().map_err(OKIError::Io)
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<()> { fn mark_counted(&self) -> OKIResult<()> {
touch(&self.root_path().join(SENTINEL_COUNTED))?; self.meta().mark_counted().map_err(OKIError::Io)
Ok(())
} }
fn mark_indexed(&self) -> OKIResult<()> { fn mark_indexed(&self) -> OKIResult<()> {
touch(&self.root_path().join(SENTINEL_INDEXED))?; self.meta().mark_indexed().map_err(OKIError::Io)
Ok(())
} }
fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap<u32, u64>) -> OKIResult<()> { fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap<u32, u64>) -> OKIResult<()> {
let label = self let genomes = self.meta().genomes().map_err(OKIError::Io)?;
.meta() let label = genomes
.genomes
.first() .first()
.map(|g| g.label.as_str()) .map(|g| g.label.as_str())
.unwrap_or("unknown"); .unwrap_or("unknown")
.to_owned();
let spectrums_dir = self.root_path().join("spectrums"); let spectrums_dir = self.root_path().join("spectrums");
fs::create_dir_all(&spectrums_dir)?; fs::create_dir_all(&spectrums_dir)?;
let path = spectrums_dir.join(format!("{label}.json")); let path = spectrums_dir.join(format!("{label}.json"));
@@ -231,27 +224,6 @@ impl PrivateBuilder for KmerIndex {
// ── private helpers ────────────────────────────────────────────────────── // ── 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) { fn remove_if_exists(path: &Path) {
if let Err(e) = fs::remove_file(path) { if let Err(e) = fs::remove_file(path) {
if e.kind() != io::ErrorKind::NotFound { if e.kind() != io::ErrorKind::NotFound {
+14 -7
View File
@@ -47,7 +47,11 @@ pub fn run(args: AnnotateArgs) {
fn run_dump(args: &AnnotateArgs) { fn run_dump(args: &AnnotateArgs) {
let idx = open_index(&args.index); let idx = open_index(&args.index);
let genomes = &idx.meta().genomes; let genomes = idx.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
});
let genomes = &genomes;
// Collect all keys in stable order (sorted for determinism) // Collect all keys in stable order (sorted for determinism)
let mut key_set: HashSet<String> = HashSet::new(); let mut key_set: HashSet<String> = HashSet::new();
@@ -89,12 +93,15 @@ fn run_annotate(args: &AnnotateArgs) {
} }
}; };
let mut idx = open_index(&args.index); let idx = open_index(&args.index);
let mut genomes = idx.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
});
// Build a label → genome index position map // Build a label → genome index position map
let label_to_pos: std::collections::HashMap<String, usize> = idx let label_to_pos: std::collections::HashMap<String, usize> = genomes
.meta()
.genomes
.iter() .iter()
.enumerate() .enumerate()
.map(|(i, g)| (g.label.clone(), i)) .map(|(i, g)| (g.label.clone(), i))
@@ -144,7 +151,7 @@ fn run_annotate(args: &AnnotateArgs) {
} }
}; };
let genome = &mut idx.meta_mut().genomes[pos]; let genome = &mut genomes[pos];
for (col_idx, key) in &meta_cols { for (col_idx, key) in &meta_cols {
let val = record.get(*col_idx).unwrap_or(""); let val = record.get(*col_idx).unwrap_or("");
if val == args.na_value { if val == args.na_value {
@@ -158,7 +165,7 @@ fn run_annotate(args: &AnnotateArgs) {
updated += 1; updated += 1;
} }
idx.meta_mut().write(&args.index).unwrap_or_else(|e| { idx.meta().set_genomes(genomes).unwrap_or_else(|e| {
eprintln!("error writing index metadata: {e}"); eprintln!("error writing index metadata: {e}");
std::process::exit(1); std::process::exit(1);
}); });
+6 -2
View File
@@ -35,13 +35,17 @@ pub fn run(args: DumpArgs) {
std::process::exit(1); std::process::exit(1);
}); });
let n_genomes = idx.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
}).len();
info!( info!(
"dumping {} partitions, {} genome(s)", "dumping {} partitions, {} genome(s)",
idx.n_partitions(), idx.n_partitions(),
&idx.meta().genomes.len() n_genomes
); );
let filters = args.filter.build_filters(idx.meta()); let filters = args.filter.build_filters(&idx.meta());
let pb = progress_bar("dump", idx.n_partitions() as u64, "partitions"); let pb = progress_bar("dump", idx.n_partitions() as u64, "partitions");
let stdout = io::stdout(); let stdout = io::stdout();
+6 -2
View File
@@ -61,12 +61,16 @@ pub fn run(args: FilterCmdArgs) {
MergeMode::Count MergeMode::Count
}; };
let n_genomes = src.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
}).len();
info!( info!(
"filter: {} genome(s), mode={:?}, source={}", "filter: {} genome(s), mode={:?}, source={}",
&src.meta().genomes.len(), mode, args.source.display() n_genomes, mode, args.source.display()
); );
let mut filters = args.filter.build_filters(src.meta()); let mut filters = args.filter.build_filters(&src.meta());
if let Some(v) = args.min_total_count { if let Some(v) = args.min_total_count {
filters.push(Box::new(MinTotalCount { total: v })); filters.push(Box::new(MinTotalCount { total: v }));
+10 -3
View File
@@ -9,6 +9,13 @@ use obikindexer::algorithms::partitionner::PartitionRouter;
use obikindex::{validate_label, GenomeInfo, IndexBuilder, IndexConfig, IndexState, KmerIndex}; use obikindex::{validate_label, GenomeInfo, IndexBuilder, IndexConfig, IndexState, KmerIndex};
use obikindex::layer::IndexMode; use obikindex::layer::IndexMode;
fn current_state(idx: &KmerIndex) -> IndexState {
idx.state().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
})
}
fn parse_key_value(s: &str) -> Result<(String, String), String> { fn parse_key_value(s: &str) -> Result<(String, String), String> {
let pos = s.find('=').ok_or_else(|| format!("invalid key=value: no '=' in '{s}'"))?; let pos = s.find('=').ok_or_else(|| format!("invalid key=value: no '=' in '{s}'"))?;
Ok((s[..pos].to_string(), s[pos + 1..].to_string())) Ok((s[..pos].to_string(), s[pos + 1..].to_string()))
@@ -236,7 +243,7 @@ pub fn run(args: IndexArgs) {
// ── Stage 1: scatter ───────────────────────────────────────────────────── // ── Stage 1: scatter ─────────────────────────────────────────────────────
if idx.state() < IndexState::Scattered { if current_state(&idx) < IndexState::Scattered {
let n_workers = args.common.threads.max(1); let n_workers = args.common.threads.max(1);
let max_open = args.common.effective_max_open(); let max_open = args.common.effective_max_open();
@@ -286,7 +293,7 @@ pub fn run(args: IndexArgs) {
} }
// ── Stage 2: dereplicate + count ───────────────────────────────────────── // ── Stage 2: dereplicate + count ─────────────────────────────────────────
if idx.state() < IndexState::Counted { if current_state(&idx) < IndexState::Counted {
let t = Stage::start("dereplicate"); let t = Stage::start("dereplicate");
let pb = progress_bar("dereplication", idx.n_partitions() as u64, "partitions"); let pb = progress_bar("dereplication", idx.n_partitions() as u64, "partitions");
Dereplicator::new(&idx) Dereplicator::new(&idx)
@@ -316,7 +323,7 @@ pub fn run(args: IndexArgs) {
} }
// ── Stage 3: build layered index ───────────────────────────────────────── // ── Stage 3: build layered index ─────────────────────────────────────────
if idx.state() < IndexState::Indexed { if current_state(&idx) < IndexState::Indexed {
let t = Stage::start("index"); let t = Stage::start("index");
let pb = progress_bar("index", idx.n_partitions() as u64, "partitions"); let pb = progress_bar("index", idx.n_partitions() as u64, "partitions");
let total_kmers = LayerBuilder::new(&idx) let total_kmers = LayerBuilder::new(&idx)
+6 -1
View File
@@ -58,7 +58,12 @@ pub fn run(args: MergeArgs) {
let source_refs: Vec<&KmerIndex> = sources.iter().collect(); let source_refs: Vec<&KmerIndex> = sources.iter().collect();
let n_genomes: usize = sources.iter().map(|s| s.meta().genomes.len()).sum(); let n_genomes: usize = sources.iter().map(|s| {
s.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
}).len()
}).sum();
info!( info!(
"merging {} index(es), {} genome(s) total → {}", "merging {} index(es), {} genome(s) total → {}",
sources.len(), n_genomes, args.output.display() sources.len(), n_genomes, args.output.display()
+5 -1
View File
@@ -32,10 +32,14 @@ pub fn run(args: PackArgs) {
std::process::exit(1); std::process::exit(1);
}); });
let n_genomes = idx.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
}).len();
info!( info!(
"pack: {} partition(s), {} genome(s)", "pack: {} partition(s), {} genome(s)",
idx.n_partitions(), idx.n_partitions(),
idx.meta().genomes.len(), n_genomes,
); );
let mut rep = Reporter::new(); let mut rep = Reporter::new();
+4 -1
View File
@@ -36,7 +36,10 @@ pub fn run(args: PhyloArgs) {
}); });
let labels: Vec<String> = idx.meta().genomes.iter().map(|g| g.label.clone()).collect(); let labels: Vec<String> = idx.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
}).iter().map(|g| g.label.clone()).collect();
let n = labels.len(); let n = labels.len();
let mut rep = Reporter::new(); let mut rep = Reporter::new();
+3 -2
View File
@@ -1,7 +1,7 @@
use std::time::Instant; use std::time::Instant;
use obikindex::KmerIndex; use obikindex::KmerIndex;
use obikindex::{KmerDesc, QueryHit, QueryStats}; use obikindex::{GenomeInfo, KmerDesc, QueryHit, QueryStats};
use obikrope::Rope; use obikrope::Rope;
use obikseq::CanonicalKmer; use obikseq::CanonicalKmer;
use obiread::record::parse_chunk; use obiread::record::parse_chunk;
@@ -40,6 +40,7 @@ pub(super) fn process_chunk(
count_missing: bool, count_missing: bool,
force_presence: bool, force_presence: bool,
presence_threshold: u32, presence_threshold: u32,
genomes: &[GenomeInfo],
) -> Vec<u8> { ) -> Vec<u8> {
let chunk_start = Instant::now(); let chunk_start = Instant::now();
let chunk_bytes = rope.len(); let chunk_bytes = rope.len();
@@ -269,7 +270,7 @@ pub(super) fn process_chunk(
emit_batch( emit_batch(
&batch, &batch,
&accs, &accs,
idx.meta(), genomes,
count_missing, count_missing,
detail, detail,
&cov, &cov,
+8 -1
View File
@@ -129,7 +129,12 @@ pub fn run(args: QueryArgs) {
})); }));
let k = idx.kmer_size(); let k = idx.kmer_size();
let n_genomes = idx.meta().genomes.len(); let genomes = idx.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
});
let n_genomes = genomes.len();
let genomes = Arc::new(genomes);
let n_partitions = idx.n_partitions(); let n_partitions = idx.n_partitions();
let with_counts = idx.meta().config.with_counts; let with_counts = idx.meta().config.with_counts;
let n_workers = args.threads.max(1); let n_workers = args.threads.max(1);
@@ -278,6 +283,7 @@ pub fn run(args: QueryArgs) {
} : Path => Chunk, } : Path => Chunk,
| { | {
let idx = Arc::clone(&idx); let idx = Arc::clone(&idx);
let genomes = Arc::clone(&genomes);
let total_bytes = Arc::clone(&total_bytes); let total_bytes = Arc::clone(&total_bytes);
let chunks_active = Arc::clone(&chunks_active); let chunks_active = Arc::clone(&chunks_active);
move |rope: Rope| { move |rope: Rope| {
@@ -286,6 +292,7 @@ pub fn run(args: QueryArgs) {
let out = process_chunk( let out = process_chunk(
&idx, rope, k, n_genomes, n_partitions, with_counts, &idx, rope, k, n_genomes, n_partitions, with_counts,
effective_z, detail, count_missing, force_presence, presence_threshold, effective_z, detail, count_missing, force_presence, presence_threshold,
&genomes,
); );
total_bytes.fetch_add(bytes, Ordering::Relaxed); total_bytes.fetch_add(bytes, Ordering::Relaxed);
chunks_active.fetch_sub(1, Ordering::Relaxed); chunks_active.fetch_sub(1, Ordering::Relaxed);
+5 -3
View File
@@ -1,12 +1,14 @@
use std::io::Write; use std::io::Write;
use obikindex::GenomeInfo;
use super::batch::QueryBatch; use super::batch::QueryBatch;
use super::chunk::SeqAcc; use super::chunk::SeqAcc;
pub(super) fn emit_batch( pub(super) fn emit_batch(
batch: &QueryBatch, batch: &QueryBatch,
accs: &[SeqAcc], accs: &[SeqAcc],
meta: &obikindex::meta::IndexMeta, genomes: &[GenomeInfo],
count_missing: bool, count_missing: bool,
detail: bool, detail: bool,
cov: &[Vec<Vec<u32>>], cov: &[Vec<Vec<u32>>],
@@ -22,7 +24,7 @@ pub(super) fn emit_batch(
} }
let mut match_map = serde_json::Map::new(); let mut match_map = serde_json::Map::new();
for (g, genome) in meta.genomes.iter().enumerate() { for (g, genome) in genomes.iter().enumerate() {
if acc.genome_totals[g] != 0 { if acc.genome_totals[g] != 0 {
match_map.insert(genome.label.clone(), acc.genome_totals[g].into()); match_map.insert(genome.label.clone(), acc.genome_totals[g].into());
} }
@@ -31,7 +33,7 @@ pub(super) fn emit_batch(
if detail && !cov.is_empty() { if detail && !cov.is_empty() {
let mut cov_map = serde_json::Map::new(); let mut cov_map = serde_json::Map::new();
for (g, genome) in meta.genomes.iter().enumerate() { for (g, genome) in genomes.iter().enumerate() {
let v: Vec<serde_json::Value> = cov[seq_idx][g].iter().map(|&x| x.into()).collect(); let v: Vec<serde_json::Value> = cov[seq_idx][g].iter().map(|&x| x.into()).collect();
cov_map.insert(genome.label.clone(), v.into()); cov_map.insert(genome.label.clone(), v.into());
} }
+11 -3
View File
@@ -115,7 +115,11 @@ fn build_specs(
meta: &IndexMeta, meta: &IndexMeta,
src_is_count: bool, src_is_count: bool,
) -> (Vec<OutputCol>, bool) { ) -> (Vec<OutputCol>, bool) {
let genomes = &meta.genomes; let genomes = meta.genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
});
let genomes = &genomes;
// ── 1. Build group_indices: name → Vec<usize> ──────────────────────────── // ── 1. Build group_indices: name → Vec<usize> ────────────────────────────
// Also keep insertion order for the default `--select *` case. // Also keep insertion order for the default `--select *` case.
@@ -231,11 +235,15 @@ pub fn run(args: SelectArgs) {
}); });
let src_is_count = src.meta().config.with_counts; let src_is_count = src.meta().config.with_counts;
let (specs, output_presence) = build_specs(&args, src.meta(), src_is_count); let (specs, output_presence) = build_specs(&args, &src.meta(), src_is_count);
let n_genomes = src.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
}).len();
info!( info!(
"select: {} genome(s) → {} output column(s), output={}", "select: {} genome(s) → {} output column(s), output={}",
src.meta().genomes.len(), n_genomes,
specs.len(), specs.len(),
if output_presence { "presence" } else { "count" }, if output_presence { "presence" } else { "count" },
); );
+5 -2
View File
@@ -30,12 +30,15 @@ pub fn run(args: UnitigArgs) {
let k = idx.kmer_size(); let k = idx.kmer_size();
let n = idx.n_partitions(); let n = idx.n_partitions();
let n_genomes = idx.meta().genomes.len().max(1); let n_genomes = idx.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
}).len().max(1);
let use_counts = idx.meta().config.with_counts; let use_counts = idx.meta().config.with_counts;
info!("unitig: building de Bruijn graph from {n} partition(s) (k={k})"); info!("unitig: building de Bruijn graph from {n} partition(s) (k={k})");
let filters = args.filter.build_filters(idx.meta()); let filters = args.filter.build_filters(&idx.meta());
let mut rep = Reporter::new(); let mut rep = Reporter::new();
// ── Phase 1 : collect filtered kmers in parallel ────────────────────────── // ── Phase 1 : collect filtered kmers in parallel ──────────────────────────
+14 -8
View File
@@ -12,8 +12,12 @@ pub(super) fn run_stats(index_path: &PathBuf) {
eprintln!("error computing stats: {e}"); eprintln!("error computing stats: {e}");
std::process::exit(1); std::process::exit(1);
}); });
let genomes = idx.meta().genomes().unwrap_or_else(|e| {
eprintln!("error reading index metadata: {e}");
std::process::exit(1);
});
println!("genome,n_kmers"); println!("genome,n_kmers");
for (g, &n) in idx.meta().genomes.iter().zip(per_genome.iter()) { for (g, &n) in genomes.iter().zip(per_genome.iter()) {
println!("{},{}", g.label, n); println!("{},{}", g.label, n);
} }
println!("total,{total}"); println!("total,{total}");
@@ -54,14 +58,17 @@ pub(super) fn run_upgrade_index(index_path: &PathBuf) {
pub(super) fn run_rename(index_path: &PathBuf, spec: &str) { pub(super) fn run_rename(index_path: &PathBuf, spec: &str) {
let (old_label, new_label) = parse_rename_spec(spec); let (old_label, new_label) = parse_rename_spec(spec);
let mut idx = KmerIndex::open(index_path).unwrap_or_else(|e| { let idx = KmerIndex::open(index_path).unwrap_or_else(|e| {
eprintln!("error opening index: {e}"); eprintln!("error opening index: {e}");
std::process::exit(1); std::process::exit(1);
}); });
let pos = idx let genomes = idx.meta().genomes().unwrap_or_else(|e| {
.meta() eprintln!("error reading index metadata: {e}");
.genomes std::process::exit(1);
});
let pos = genomes
.iter() .iter()
.position(|g| g.label == old_label) .position(|g| g.label == old_label)
.unwrap_or_else(|| { .unwrap_or_else(|| {
@@ -74,13 +81,12 @@ pub(super) fn run_rename(index_path: &PathBuf, spec: &str) {
std::process::exit(1); std::process::exit(1);
}); });
if idx.meta().genomes.iter().any(|g| g.label == new_label) { if genomes.iter().any(|g| g.label == new_label) {
eprintln!("error: label '{new_label}' already exists in index"); eprintln!("error: label '{new_label}' already exists in index");
std::process::exit(1); std::process::exit(1);
} }
idx.meta_mut().genomes[pos].label = new_label.clone(); idx.meta().rename_genome(pos, new_label.clone()).unwrap_or_else(|e| {
idx.meta_mut().write(index_path).unwrap_or_else(|e| {
eprintln!("error writing index metadata: {e}"); eprintln!("error writing index metadata: {e}");
std::process::exit(1); std::process::exit(1);
}); });
+2 -2
View File
@@ -3,7 +3,7 @@ use std::sync::Arc;
use obisys::progress_bar; use obisys::progress_bar;
use obikindex::KmerIndex; use obikindex::KmerIndex;
use obikindex::OKIResult; use obikindex::{OKIError, OKIResult};
use super::cache::PartitionCache; use super::cache::PartitionCache;
use super::family_scan::{Selection, scan_layer_families}; use super::family_scan::{Selection, scan_layer_families};
@@ -83,7 +83,7 @@ impl SnpAlignmentExt for KmerIndex {
entropy_bias: Option<EntropyBias>, entropy_bias: Option<EntropyBias>,
) -> OKIResult<SnpAlignment> { ) -> OKIResult<SnpAlignment> {
let n_parts = self.n_partitions(); let n_parts = self.n_partitions();
let n_genomes = self.meta().genomes.len(); let n_genomes = self.meta().genomes().map_err(OKIError::Io)?.len();
let with_counts = self.meta().config.with_counts; let with_counts = self.meta().config.with_counts;
let k = self.kmer_size(); let k = self.kmer_size();
let cache = Arc::new(PartitionCache::build(self, n_parts, with_counts)?); let cache = Arc::new(PartitionCache::build(self, n_parts, with_counts)?);
+2 -2
View File
@@ -5,7 +5,7 @@ use ndarray::Array2;
use obisys::progress_bar; use obisys::progress_bar;
use obikindex::KmerIndex; use obikindex::KmerIndex;
use obikindex::OKIResult; use obikindex::{OKIError, OKIResult};
use super::cache::PartitionCache; use super::cache::PartitionCache;
use super::distance::RawSnpDistanceOutput; use super::distance::RawSnpDistanceOutput;
@@ -65,7 +65,7 @@ impl CardinalityExt for KmerIndex {
ratio_ceiling: f64, ratio_ceiling: f64,
) -> OKIResult<CardinalityTally> { ) -> OKIResult<CardinalityTally> {
let n_parts = self.n_partitions(); let n_parts = self.n_partitions();
let n_genomes = self.meta().genomes.len(); let n_genomes = self.meta().genomes().map_err(OKIError::Io)?.len();
let with_counts = self.meta().config.with_counts; let with_counts = self.meta().config.with_counts;
let k = self.kmer_size(); let k = self.kmer_size();
let included = Array2::from_shape_fn((n_genomes, n_genomes), |(i, j)| { let included = Array2::from_shape_fn((n_genomes, n_genomes), |(i, j)| {
+4 -4
View File
@@ -5,7 +5,7 @@ use ndarray::Array2;
use obisys::progress_bar; use obisys::progress_bar;
use obikindex::KmerIndex; use obikindex::KmerIndex;
use obikindex::OKIResult; use obikindex::{OKIError, OKIResult};
use super::cache::PartitionCache; use super::cache::PartitionCache;
use super::family_scan::{Selection, scan_layer_families}; use super::family_scan::{Selection, scan_layer_families};
@@ -66,7 +66,7 @@ where
C: Fn(Acc, Acc) -> Acc, C: Fn(Acc, Acc) -> Acc,
{ {
let n_parts = index.n_partitions(); let n_parts = index.n_partitions();
let n_genomes = index.meta().genomes.len(); let n_genomes = index.meta().genomes().map_err(OKIError::Io)?.len();
let with_counts = index.meta().config.with_counts; let with_counts = index.meta().config.with_counts;
let k = index.kmer_size(); let k = index.kmer_size();
let cache = Arc::new(PartitionCache::build(index, n_parts, with_counts)?); let cache = Arc::new(PartitionCache::build(index, n_parts, with_counts)?);
@@ -149,7 +149,7 @@ pub trait DistanceExt {
impl DistanceExt for KmerIndex { impl DistanceExt for KmerIndex {
fn raw_snp_distance(&self) -> OKIResult<RawSnpDistanceOutput> { fn raw_snp_distance(&self) -> OKIResult<RawSnpDistanceOutput> {
let n_genomes = self.meta().genomes.len(); let n_genomes = self.meta().genomes().map_err(OKIError::Io)?.len();
let (snp, shared) = scan_family_pairs( let (snp, shared) = scan_family_pairs(
self, self,
"raw_snp_distance", "raw_snp_distance",
@@ -182,7 +182,7 @@ impl DistanceExt for KmerIndex {
raw: &RawSnpDistanceOutput, raw: &RawSnpDistanceOutput,
ratio_ceiling: f64, ratio_ceiling: f64,
) -> OKIResult<BasePairTally> { ) -> OKIResult<BasePairTally> {
let n_genomes = self.meta().genomes.len(); let n_genomes = self.meta().genomes().map_err(OKIError::Io)?.len();
let included = Array2::from_shape_fn((n_genomes, n_genomes), |(i, j)| { let included = Array2::from_shape_fn((n_genomes, n_genomes), |(i, j)| {
if i == j { if i == j {
return false; return false;
+2 -2
View File
@@ -135,7 +135,7 @@ impl ShannonEntropyExt for KmerIndex {
entropy_bias: Option<EntropyBias>, entropy_bias: Option<EntropyBias>,
) -> OKIResult<()> { ) -> OKIResult<()> {
let n_parts = self.n_partitions(); let n_parts = self.n_partitions();
let n_genomes = self.meta().genomes.len(); let n_genomes = self.meta().genomes().map_err(OKIError::Io)?.len();
let with_counts = self.meta().config.with_counts; let with_counts = self.meta().config.with_counts;
let k = self.kmer_size(); let k = self.kmer_size();
let cache = Arc::new(PartitionCache::build(self, n_parts, with_counts)?); let cache = Arc::new(PartitionCache::build(self, n_parts, with_counts)?);
@@ -227,7 +227,7 @@ pub(super) fn ensure_entropy_annexes(index: &KmerIndex, layer_dirs: &[PathBuf])
} }
let n_parts = index.n_partitions(); let n_parts = index.n_partitions();
let n_genomes = index.meta().genomes.len(); let n_genomes = index.meta().genomes().map_err(OKIError::Io)?.len();
let with_counts = index.meta().config.with_counts; let with_counts = index.meta().config.with_counts;
let k = index.kmer_size(); let k = index.kmer_size();
let cache = Arc::new(PartitionCache::build(index, n_parts, with_counts)?); let cache = Arc::new(PartitionCache::build(index, n_parts, with_counts)?);
+2 -2
View File
@@ -32,7 +32,7 @@ use ndarray::Array2;
use obisys::progress_bar; use obisys::progress_bar;
use obikindex::KmerIndex; use obikindex::KmerIndex;
use obikindex::OKIResult; use obikindex::{OKIError, OKIResult};
use super::alignment::{SnpAlignment, iupac_code}; use super::alignment::{SnpAlignment, iupac_code};
use super::cache::PartitionCache; use super::cache::PartitionCache;
@@ -79,7 +79,7 @@ impl SankoffBundleExt for KmerIndex {
exclude_mask: &[bool], exclude_mask: &[bool],
) -> OKIResult<SankoffBundle> { ) -> OKIResult<SankoffBundle> {
let n_parts = self.n_partitions(); let n_parts = self.n_partitions();
let n_genomes = self.meta().genomes.len(); let n_genomes = self.meta().genomes().map_err(OKIError::Io)?.len();
let with_counts = self.meta().config.with_counts; let with_counts = self.meta().config.with_counts;
let k = self.kmer_size(); let k = self.kmer_size();
let cache = Arc::new(PartitionCache::build(self, n_parts, with_counts)?); let cache = Arc::new(PartitionCache::build(self, n_parts, with_counts)?);
+2 -2
View File
@@ -5,7 +5,7 @@ use rayon::prelude::*;
use obisys::progress_bar; use obisys::progress_bar;
use obikindex::KmerIndex; use obikindex::KmerIndex;
use obikindex::OKIResult; use obikindex::{OKIError, OKIResult};
use super::ANNEX_FILE_NAME; use super::ANNEX_FILE_NAME;
use super::SiblingAnnex; use super::SiblingAnnex;
@@ -105,7 +105,7 @@ impl SiblingStatsExt for KmerIndex {
fn sibling_annex_stats(&self) -> OKIResult<SiblingAnnexStats> { fn sibling_annex_stats(&self) -> OKIResult<SiblingAnnexStats> {
let n_parts = self.n_partitions(); let n_parts = self.n_partitions();
let n_genomes = self.meta().genomes.len(); let n_genomes = self.meta().genomes().map_err(OKIError::Io)?.len();
let with_counts = self.meta().config.with_counts; let with_counts = self.meta().config.with_counts;
let k = self.kmer_size(); let k = self.kmer_size();
// Same whole-run cache as `build_sibling_annex` — see its docs for // Same whole-run cache as `build_sibling_annex` — see its docs for
+4 -4
View File
@@ -275,7 +275,7 @@ fn family_scan_consumers_agree_on_one_sibling_each() {
// Merge doesn't promise to preserve source order, so resolve each // Merge doesn't promise to preserve source order, so resolve each
// genome's index by label rather than assuming g1 -> 0, g2 -> 1. // genome's index by label rather than assuming g1 -> 0, g2 -> 1.
let idx_of = |label: &str| merged.meta().genomes.iter().position(|g| g.label == label).unwrap(); let idx_of = |label: &str| merged.meta().genomes().unwrap().iter().position(|g| g.label == label).unwrap();
let (i1, i2) = (idx_of("g1"), idx_of("g2")); let (i1, i2) = (idx_of("g1"), idx_of("g2"));
// snp_pseudo_alignment: one variable family, one column — g1's row // snp_pseudo_alignment: one variable family, one column — g1's row
@@ -389,7 +389,7 @@ fn subsample_and_shannon_on_one_variable_family() {
let merged = merge_two(dir.path(), &g1, &g2); let merged = merge_two(dir.path(), &g1, &g2);
merged.build_sibling_annex().expect("build_sibling_annex"); merged.build_sibling_annex().expect("build_sibling_annex");
let idx_of = |label: &str| merged.meta().genomes.iter().position(|g| g.label == label).unwrap(); let idx_of = |label: &str| merged.meta().genomes().unwrap().iter().position(|g| g.label == label).unwrap();
let (i1, i2) = (idx_of("g1"), idx_of("g2")); let (i1, i2) = (idx_of("g1"), idx_of("g2"));
// This fixture has exactly one non-monomorphic family (see // This fixture has exactly one non-monomorphic family (see
@@ -442,7 +442,7 @@ fn sankoff_bundle_matches_old_separate_calls() {
let merged = merge_two(dir.path(), &g1, &g2); let merged = merge_two(dir.path(), &g1, &g2);
merged.build_sibling_annex().expect("build_sibling_annex"); merged.build_sibling_annex().expect("build_sibling_annex");
let n_genomes = merged.meta().genomes.len(); let n_genomes = merged.meta().genomes().unwrap().len();
let exclude_mask = vec![false; n_genomes]; let exclude_mask = vec![false; n_genomes];
let ratio_ceiling = 0.5; let ratio_ceiling = 0.5;
@@ -487,7 +487,7 @@ fn base_pair_tally_accumulates_base_a_diagnostic() {
).expect("merge"); ).expect("merge");
merged.build_sibling_annex().expect("build_sibling_annex"); merged.build_sibling_annex().expect("build_sibling_annex");
let n_genomes = merged.meta().genomes.len(); let n_genomes = merged.meta().genomes().unwrap().len();
let exclude_mask = vec![false; n_genomes]; let exclude_mask = vec![false; n_genomes];
let bundle = merged.sankoff_bundle(None, None, 0.5, &exclude_mask).expect("sankoff_bundle"); let bundle = merged.sankoff_bundle(None, None, 0.5, &exclude_mask).expect("sankoff_bundle");