diff --git a/DevDocMD/implementation/partition_layer_cache.md b/DevDocMD/implementation/partition_layer_cache.md new file mode 100644 index 00000000..f3208fcd --- /dev/null +++ b/DevDocMD/implementation/partition_layer_cache.md @@ -0,0 +1,129 @@ +# Partition and layer caching (discussion) + +Status: problem confirmed, design direction agreed (2026-08-20). No +implementation started. Ownership split (below) still open. + +## The problem + +Reading a layer's data (MPHF + matrix) is not free: `MphfLayer::open` mmaps +`mphf.bin` plus (`evidence.bin`/`fingerprint.bin` + `unitigs.bin`), and the +matrix side mmaps `matrix.pbmx`/`matrix.pcmx` (or one file per genome column +if not yet packed). Any code path that reopens a layer per lookup instead of +once per run pays this cost repeatedly. + +`obikphylo::siblings::cache::PartitionCache` was built to avoid exactly this +for `build_sibling_annex`/`sibling_annex_stats`: those commands probe many +partitions, once per source layer, over the whole run. Profiling a real run +showed wall-clock time dominated by repeated `open()`/mmap syscalls, not +computation — parallelising the naive per-lookup opens spread the cost +across cores without reducing it. `PartitionCache::build` opens every +partition's every layer once, up front, in parallel, and keeps the handles +alive for the run. + +## Three independent implementations of the same bundle + +Searching the codebase for "who bundles MPHF + matrix, with per-layer format +auto-detection" turns up three unrelated implementations: + +| | lives in | scope | cached? | +|---|---|---|---| +| `Layer` | `obilayeredmap` | one layer, `D` fixed at compile time | held alive by whoever owns the `Layer`, no policy of its own | +| `Mat` | `obikphylo::siblings::cache` | one layer, format resolved per instance from an enum of 3 `Layer` variants | yes, via `PartitionCache` | +| `QueryLayer` | `obikpartitionner::query_layer` | one layer, `(MphfLayer, PersistentBitMatrix\|PersistentCompactIntMatrix)` pair, bypasses `Layer` entirely | **no** — opened fresh inside `query_partition_with` on every call | + +`query_partition_with` is `obikmer query`'s normal query path — the one +most exposed to repeated cross-partition lookups — and it is the one with +no cache at all. `obikphylo` built a cache first only because sibling-annex +construction hits the cost hardest, not because the need is sibling-specific. + +## The gap in `obilayeredmap`'s existing cache + +`obilayeredmap::LayeredMap` already caches correctly at the granularity +of one partition: `open(root)` opens every layer once, keeps `Vec>` +alive for the `LayeredMap`'s lifetime. But it is monomorphic — every layer +in the `Vec` must share the same concrete `D`. In practice this is false: +layers in the same partition are packed independently over time (`pack +--sparse` converts one layer's presence matrix at a time), so a partition +can genuinely mix `PersistentBitMatrix`-backed and +`PersistentSparseBitMatrix`-backed presence layers. `LayeredMap` cannot +represent that mix — `Mat`'s 3-way enum exists specifically to work around +this gap, one crate away from where the gap actually is. + +## Resource cost: mmap does not hold a file descriptor + +Before deciding how many layers/partitions a cache may hold open +simultaneously, the binding constraint needs to be identified correctly. + +Confirmed against upstream documentation, not inferred from behaviour: + +> "After the mmap() call has returned, the file descriptor, fd, can be +> closed immediately without invalidating the mapping." +> — [mmap(2), man7.org](https://man7.org/linux/man-pages/man2/mmap.2.html) + +> "The close(2) function does not unmap pages" +> — [mmap(2), Apple Developer](https://developer.apple.com/library/archive/documentation/System/Conceptual/ManPages_iPhoneOS/man2/mmap.2.html) + +> "A file backed Mmap ... will remain valid even after the File is dropped. +> ... the Mmap handle is completely independent of the File used to create +> it." +> — [memmap2::Mmap, docs.rs](https://docs.rs/memmap2/latest/memmap2/struct.Mmap.html) + +Every read-only mmap in this codebase already follows this: `Mmap::map(&File::open(path)?)?` +— the `File` is a temporary, dropped (fd closed) immediately after the +mapping is established; every persistent struct (`PersistentBitVec`, +`PersistentCompactIntVec`, `PackedBitMatrix`, `Evidence`, `FingerprintVec`, +...) stores only the `Mmap`, never the `File`. So a cache built on these +types does **not** consume the process's open-file-descriptor budget +(`ulimit -n`, notoriously low by default on macOS) proportionally to how +many mmapped files it holds. + +It does consume a different resource — the process's virtual-memory mapping +table (one entry per active `mmap()` region). Linux exposes this as +`vm.max_map_count` (default 65530). No documented macOS equivalent (fixed +numeric ceiling) was found; the constraint there appears to be virtual +address space rather than an explicit mapping counter, but this is not +sourced and should not be assumed. This is the resource actually worth +measuring before deciding on cache size, not fd count — and it is why +*packing* (`matrix.pbmx`/`matrix.pcmx`, one mmap for all columns) matters +independently of any caching decision: an unpacked `Columnar` matrix opens +one mmap **per genome column**, multiplying the mapping count a cache would +have to hold by `n_genomes`. + +## Layering: who owns what + +Established in discussion, not yet coded: + +- `obilayeredmap` operates within a single partition's index root; it has + no notion of "partition" (confirmed independently while fixing + `open_data`/`layer_dir` — see git history around 2026-08-20). One + `LayeredMap` (or its heterogeneous-`D` successor) = one partition. +- Nobody currently owns "the collection of partitions" as reusable state. + `obikpartitionner::KmerPartition` is the closest candidate — it already + owns `n_partitions()`/`part_dir(i)` — but today it is a pure + config/path resolver, not a state holder: `dereplicate`, `count_kmer`, + `obikindex`'s `distance.rs`/`stats.rs`, and + `obikphylo::PartitionCache::build` each independently write their own + `(0..n_partitions).into_par_iter().map(...)` loop and open what they need + from scratch. + +## Direction agreed, not yet implemented + +1. A heterogeneous-format layer type in `obilayeredmap` (name TBD), + generalising `Mat`'s 3-way enum + auto-detection — `obilayeredmap` + already implements `LayerData` for all three concrete matrix types + involved, so it already has the knowledge `Mat` needed and had to + duplicate. +2. A cache spanning multiple partitions, keeping (2)'s layer type open per + partition per layer, for the whole lifetime of a long-running command — + generalising `PartitionCache` minus its sibling-specific parts + (`fast_mode`, `find_presence_batch`) — living in `obikpartitionner` + (owner of the partition dimension) and built on top of (1). +3. `obikpartitionner::query_partition_with` — the normal query path, + currently uncached — becomes a consumer of (2), not just + `obikphylo::PartitionCache`. + +Open before implementing: exact API shape of (1) and (2), whether `Mat` is +deleted outright or kept as a thin sibling-specific wrapper, and whether (2) +needs an eviction policy or can simply hold every partition open for the +process lifetime (revisit once the VM-mapping-count question above has a +real number behind it for this codebase's scale). diff --git a/mkdocs.yml b/mkdocs.yml index b1660442..4fec9411 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -56,6 +56,7 @@ nav: - Select command: implementation/select.md - obitaxonomy crate: implementation/obitaxonomy.md - "Benchmark: query-path testing": implementation/benchmark_query_testing.md + - "Partition and layer caching (discussion)": implementation/partition_layer_cache.md - Architecture: - Sequences: architecture/sequences/invariant.md - Kmer index: architecture/index_architecture.md diff --git a/src/obikindex/examples/compare_sparse.rs b/src/obikindex/examples/compare_sparse.rs index 72534989..51376fad 100644 --- a/src/obikindex/examples/compare_sparse.rs +++ b/src/obikindex/examples/compare_sparse.rs @@ -5,7 +5,6 @@ use obicompactvec::{PersistentBitMatrix, PersistentSparseBitMatrix}; use obikindex::KmerIndex; -use obilayeredmap::layer_dir; fn main() -> anyhow::Result<()> { let sparse_root = std::env::args().nth(1).expect("usage: compare_sparse "); @@ -26,18 +25,16 @@ fn main() -> anyhow::Result<()> { let mut first_mismatch = None; for part in 0..n_parts { - let part_dir_sparse = sparse.partition().part_dir(part); - let part_dir_dense = dense.partition().part_dir(part); + let index_dir_sparse = sparse.partition().index_dir(part); + let index_dir_dense = dense.partition().index_dir(part); - if !part_dir_sparse.join("index").exists() || !part_dir_dense.join("index").exists() { + if !index_dir_sparse.exists() || !index_dir_dense.exists() { continue; } for layer in 0..n_layers { - let index_dir_sparse = part_dir_sparse.join("index"); - let index_dir_dense = part_dir_dense.join("index"); - let layer_dir_sparse = layer_dir(&index_dir_sparse, layer); - let layer_dir_dense = layer_dir(&index_dir_dense, layer); + let layer_dir_sparse = sparse.partition().layer_dir(part, layer); + let layer_dir_dense = dense.partition().layer_dir(part, layer); if !layer_dir_sparse.exists() || !layer_dir_dense.exists() { continue; diff --git a/src/obikindex/src/index.rs b/src/obikindex/src/index.rs index cadc7440..d178113f 100644 --- a/src/obikindex/src/index.rs +++ b/src/obikindex/src/index.rs @@ -149,7 +149,7 @@ impl KmerIndex { /// is enough, no need to scan every partition. pub fn n_layers_per_partition(&self) -> OKIResult { use obilayeredmap::meta::PartitionMeta; - let index_dir = self.partition.part_dir(0).join("index"); + let index_dir = self.partition.index_dir(0); let meta = PartitionMeta::load(&index_dir) .map_err(|e| OKIError::Io(std::io::Error::new(std::io::ErrorKind::Other, e.to_string())))?; Ok(meta.n_layers) @@ -264,8 +264,7 @@ impl KmerIndex { /// Path to the unitigs file for partition `part`, layer `layer`. pub fn layer_unitigs_path(&self, part: usize, layer: usize) -> PathBuf { - let index_dir = self.partition.part_dir(part).join("index"); - obilayeredmap::layer_dir(&index_dir, layer).join("unitigs.bin") + self.partition.layer_dir(part, layer).join("unitigs.bin") } /// Pack all partition matrices into single-file format (presence → .pbmx, counts → .pcmx). @@ -288,12 +287,12 @@ impl KmerIndex { crate::numa::PartitionRunner::new().run( &order, |i| -> OKIResult<()> { - let index_dir = self.partition.part_dir(i).join("index"); + let index_dir = self.partition.index_dir(i); if !index_dir.exists() { return Ok(()); } let meta = PartitionMeta::load(&index_dir) .map_err(|e| OKIError::Io(std::io::Error::new(std::io::ErrorKind::Other, e.to_string())))?; for l in 0..meta.n_layers { - let layer_dir = obilayeredmap::layer_dir(&index_dir, l); + let layer_dir = self.partition.layer_dir(i, l); let presence_dir = layer_dir.join("presence"); let counts_dir = layer_dir.join("counts"); if presence_dir.exists() { @@ -326,14 +325,14 @@ impl KmerIndex { let errors: Vec<_> = (0..n) .into_par_iter() .filter_map(|i| { - let index_dir = self.partition.part_dir(i).join("index"); + let index_dir = self.partition.index_dir(i); if !index_dir.exists() { return None; } let meta = match PartitionMeta::load(&index_dir) { Ok(m) => m, Err(e) => return Some(OKIError::Io(std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))), }; for l in 0..meta.n_layers { - let layer_dir = obilayeredmap::layer_dir(&index_dir, l); + let layer_dir = self.partition.layer_dir(i, l); let meta_path = layer_dir.join(LayerMeta::FILENAME); if meta_path.exists() { continue; } let unitigs_path = layer_dir.join("unitigs.bin"); diff --git a/src/obikindex/src/reindex.rs b/src/obikindex/src/reindex.rs index 7dbab9fd..05a919e1 100644 --- a/src/obikindex/src/reindex.rs +++ b/src/obikindex/src/reindex.rs @@ -48,7 +48,7 @@ impl KmerIndex { let runner = crate::numa::PartitionRunner::new(); runner.run( &order, - |i| reindex_partition(&self.partition.part_dir(i).join("index"), &target, block_bits) + |i| reindex_partition(&self.partition.index_dir(i), &target, block_bits) .map_err(|e| OKIError::InvalidInput(format!("partition {i}: {e}"))), |_, _, _| { pb.inc(1); }, )?; diff --git a/src/obikindex/src/stats.rs b/src/obikindex/src/stats.rs index 4db896a9..eef2a252 100644 --- a/src/obikindex/src/stats.rs +++ b/src/obikindex/src/stats.rs @@ -3,7 +3,6 @@ use std::path::Path; use obicompactvec::{LayerMeta, PersistentBitMatrix, PersistentCompactIntMatrix}; use obicompactvec::traits::ColumnWeights; -use obilayeredmap::layer_dir; use obilayeredmap::meta::PartitionMeta; use rayon::prelude::*; @@ -91,7 +90,7 @@ impl KmerIndex { let (n_kmers, mphf_b, evidence_b, matrix_b) = (0..n) .into_par_iter() .map(|i| { - let index_dir = self.partition.part_dir(i).join("index"); + let index_dir = self.partition.index_dir(i); if !index_dir.exists() { return (0usize, 0u64, 0u64, 0u64); } let n_layers = PartitionMeta::load(&index_dir) @@ -99,7 +98,7 @@ impl KmerIndex { .unwrap_or(0); (0..n_layers).fold((0usize, 0u64, 0u64, 0u64), |acc, l| { - let lb = layer_bytes(&layer_dir(&index_dir, l)); + let lb = layer_bytes(&self.partition.layer_dir(i, l)); (acc.0 + lb.n_kmers, acc.1 + lb.mphf, acc.2 + lb.evidence, acc.3 + lb.matrix) }) }) @@ -142,7 +141,7 @@ impl KmerIndex { let mut counts = vec![0u64; n_genomes]; let mut n_kmers = 0usize; - let index_dir = self.partition.part_dir(i).join("index"); + let index_dir = self.partition.index_dir(i); if !index_dir.exists() { return (0, counts); } let n_layers = PartitionMeta::load(&index_dir) @@ -150,7 +149,7 @@ impl KmerIndex { .unwrap_or(0); for l in 0..n_layers { - let this_layer_dir = obilayeredmap::layer_dir(&index_dir, l); + let this_layer_dir = self.partition.layer_dir(i, l); if !this_layer_dir.exists() { continue; } n_kmers += LayerMeta::load(&this_layer_dir).map(|m| m.n).unwrap_or(0); diff --git a/src/obikpartitionner/src/distance.rs b/src/obikpartitionner/src/distance.rs index 3cc33aeb..837d2cdc 100644 --- a/src/obikpartitionner/src/distance.rs +++ b/src/obikpartitionner/src/distance.rs @@ -1,24 +1,22 @@ use obicompactvec::{PersistentBitMatrix, PersistentCompactIntMatrix}; -use obilayeredmap::{layer_dir, open_data, LayeredStore}; +use obilayeredmap::{open_data, LayeredStore}; use obiskio::SKResult; use crate::common::{load_meta, olm_to_sk}; use crate::partition::KmerPartition; -const INDEX_SUBDIR: &str = "index"; - impl KmerPartition { /// Open all count matrices for partition `part`, one per layer. /// Layers without a `counts/` directory are skipped. pub fn count_store(&self, part: usize) -> SKResult> { - let index_dir = self.part_dir(part).join(INDEX_SUBDIR); + let index_dir = self.index_dir(part); if !index_dir.exists() { return Ok(LayeredStore::new(vec![])); } let n_layers = load_meta(&index_dir, "distance")?.n_layers; let matrices = (0..n_layers) .filter_map(|l| { - layer_dir(&index_dir, l).join("counts").exists() + self.layer_dir(part, l).join("counts").exists() .then(|| open_data(&index_dir, l).map_err(|e| olm_to_sk(e, "distance"))) }) .collect::>>()?; @@ -28,14 +26,14 @@ impl KmerPartition { /// Open all presence matrices for partition `part`, one per layer. /// Layers without a `presence/` directory are skipped. pub fn presence_store(&self, part: usize) -> SKResult> { - let index_dir = self.part_dir(part).join(INDEX_SUBDIR); + let index_dir = self.index_dir(part); if !index_dir.exists() { return Ok(LayeredStore::new(vec![])); } let n_layers = load_meta(&index_dir, "distance")?.n_layers; let matrices = (0..n_layers) .filter_map(|l| { - layer_dir(&index_dir, l).join("presence").exists() + self.layer_dir(part, l).join("presence").exists() .then(|| open_data(&index_dir, l).map_err(|e| olm_to_sk(e, "distance"))) }) .collect::>>()?; diff --git a/src/obikpartitionner/src/dump_layer.rs b/src/obikpartitionner/src/dump_layer.rs index 15252448..46e93cda 100644 --- a/src/obikpartitionner/src/dump_layer.rs +++ b/src/obikpartitionner/src/dump_layer.rs @@ -1,14 +1,12 @@ use obicompactvec::{PersistentBitMatrix, PersistentCompactIntMatrix}; use obikseq::CanonicalKmer; use obiskio::{SKError, SKResult, UnitigFileReader}; -use obilayeredmap::{layer_dir, IndexMode, MphfLayer, OLMError}; +use obilayeredmap::{IndexMode, MphfLayer, OLMError}; use obilayeredmap::meta::PartitionMeta; use crate::filter::{KmerFilter, passes_all}; use crate::partition::KmerPartition; -const INDEX_SUBDIR: &str = "index"; - fn olm_to_sk(e: OLMError) -> SKError { match e { OLMError::Io(e) => SKError::Io(e), @@ -36,7 +34,7 @@ impl KmerPartition { filters: &[Box], mut cb: impl FnMut(CanonicalKmer, Box<[u32]>) -> bool, ) -> SKResult { - let index_dir = self.part_dir(part).join(INDEX_SUBDIR); + let index_dir = self.index_dir(part); if !index_dir.exists() { return Ok(true); } @@ -47,7 +45,7 @@ impl KmerPartition { let mut l = 0; loop { - let layer_dir = layer_dir(&index_dir, l); + let layer_dir = self.layer_dir(part, l); if !layer_dir.exists() { break; } l += 1; let mphf = MphfLayer::open(&layer_dir, &index_mode).map_err(olm_to_sk)?; @@ -116,7 +114,7 @@ impl KmerPartition { filters: &[Box], mut cb: impl FnMut(usize, usize, CanonicalKmer, Box<[u32]>) -> bool, ) -> SKResult { - let index_dir = self.part_dir(part).join(INDEX_SUBDIR); + let index_dir = self.index_dir(part); if !index_dir.exists() { return Ok(true); } @@ -127,7 +125,7 @@ impl KmerPartition { let mut layer = 0; loop { - let layer_dir = layer_dir(&index_dir, layer); + let layer_dir = self.layer_dir(part, layer); if !layer_dir.exists() { break; } let mphf = MphfLayer::open(&layer_dir, &index_mode).map_err(olm_to_sk)?; let reader = UnitigFileReader::open_sequential(&layer_dir.join("unitigs.bin"))?; diff --git a/src/obikpartitionner/src/index_layer.rs b/src/obikpartitionner/src/index_layer.rs index 6b48df58..9217df69 100644 --- a/src/obikpartitionner/src/index_layer.rs +++ b/src/obikpartitionner/src/index_layer.rs @@ -42,13 +42,13 @@ impl KmerPartition { mode: &IndexMode, block_bits: u8, ) -> Result { - let part_dir = self.part_dir(i); - let dedup_path = part_dir.join("dereplicated.skmer.zst"); + let partition_dir = self.partition_dir(i); + let dedup_path = partition_dir.join("dereplicated.skmer.zst"); if !dedup_path.exists() { return Ok(0); } - let layer_dir = part_dir.join("index").join("layer_0"); + let layer_dir = self.layer_dir(i, 0); if layer_dir.join("mphf.bin").exists() { return Ok(0); } @@ -57,14 +57,14 @@ impl KmerPartition { let need_counts = filter_active || with_counts; let mphf1_opt: Option = if need_counts { - let p = part_dir.join("mphf1.bin"); + let p = partition_dir.join("mphf1.bin"); p.exists().then(|| Mphf::load_full(&p).ok()).flatten() } else { None }; let counts1_opt: Option = if need_counts { - let p = part_dir.join("counts1.bin"); + let p = partition_dir.join("counts1.bin"); p.exists() .then(|| PersistentCompactIntVec::open(&p).ok()) .flatten() @@ -125,11 +125,11 @@ impl KmerPartition { /// /// Deletes `dereplicated.skmer.zst` (+ sidecar), `mphf1.bin`, `counts1.bin`. pub fn remove_build_artifacts(&self, i: usize) { - let part_dir = self.part_dir(i); - let dedup = part_dir.join("dereplicated.skmer.zst"); + let partition_dir = self.partition_dir(i); + let dedup = partition_dir.join("dereplicated.skmer.zst"); remove_if_exists(&SKFileMeta::sidecar_path(&dedup)); remove_if_exists(&dedup); - remove_if_exists(&part_dir.join("mphf1.bin")); - remove_if_exists(&part_dir.join("counts1.bin")); + remove_if_exists(&partition_dir.join("mphf1.bin")); + remove_if_exists(&partition_dir.join("counts1.bin")); } } diff --git a/src/obikpartitionner/src/merge_layer/mod.rs b/src/obikpartitionner/src/merge_layer/mod.rs index a4046dd2..ea92e3f1 100644 --- a/src/obikpartitionner/src/merge_layer/mod.rs +++ b/src/obikpartitionner/src/merge_layer/mod.rs @@ -187,10 +187,6 @@ mod matrix_builder_tests { } } -// ── helpers ─────────────────────────────────────────────────────────────────── - -const INDEX_SUBDIR: &str = "index"; - // ── KmerPartition::merge_partition ──────────────────────────────────────────── impl KmerPartition { @@ -212,7 +208,7 @@ impl KmerPartition { block_bits: u8, evidence: &IndexMode, ) -> SKResult { - let dst_index_dir = self.part_dir(i).join(INDEX_SUBDIR); + let dst_index_dir = self.index_dir(i); if !dst_index_dir.exists() { return Ok(0); } @@ -241,7 +237,7 @@ impl KmerPartition { // Collect file paths (propagates load_meta errors before the pipeline starts) let mut unitig_paths: Vec = Vec::new(); for (src, _) in sources.iter() { - let src_index_dir = src.part_dir(i).join(INDEX_SUBDIR); + let src_index_dir = src.index_dir(i); if !src_index_dir.exists() { continue; } @@ -366,7 +362,7 @@ impl KmerPartition { { let mut col_offset = 0usize; for (src, src_n) in sources.iter() { - let src_index_dir = src.part_dir(i).join(INDEX_SUBDIR); + let src_index_dir = src.index_dir(i); if !src_index_dir.exists() { col_offset += src_n; continue; diff --git a/src/obikpartitionner/src/partition/kmer_partition.rs b/src/obikpartitionner/src/partition/kmer_partition.rs index 891e6f8a..ec3a5fa4 100644 --- a/src/obikpartitionner/src/partition/kmer_partition.rs +++ b/src/obikpartitionner/src/partition/kmer_partition.rs @@ -22,6 +22,13 @@ use super::count::count_partition; use super::dereplicate::{dereplicate_partition, optimal_buckets}; use super::{PARTITIONS_SUBDIR, SK_EXT}; +/// Name of a partition's layered-index subdirectory — the single source of +/// truth `index_dir`/`layer_dir` build on, replacing what used to be a +/// `const INDEX_SUBDIR: &str = "index";` (or a bare `"index"` literal) +/// redefined independently in every module that needed a partition's index +/// path. +const INDEX_SUBDIR: &str = "index"; + pub struct KmerSpectrum { pub f0: u64, pub f1: u64, @@ -168,10 +175,22 @@ impl KmerPartition { } /// Path of partition `i` directory. - pub fn part_dir(&self, i: usize) -> PathBuf { + pub fn partition_dir(&self, i: usize) -> PathBuf { self.root_path.join(PARTITIONS_SUBDIR).join(format!("part_{i:05}")) } + /// Path of partition `i`'s layered-index directory (`/index`). + pub fn index_dir(&self, i: usize) -> PathBuf { + self.partition_dir(i).join(INDEX_SUBDIR) + } + + /// Path of layer `l` within partition `i`'s layered index — composes + /// [`index_dir`](Self::index_dir) with `obilayeredmap`'s own + /// `layer_N` naming convention rather than reimplementing it. + pub fn layer_dir(&self, i: usize, l: usize) -> PathBuf { + obilayeredmap::layer_dir(&self.index_dir(i), l) + } + pub fn kmer_size(&self) -> usize { self.kmer_size } @@ -220,7 +239,7 @@ impl KmerPartition { let results: Vec> = (0..self.n_partitions) .into_par_iter() .map(|i| { - let dir = self.part_dir(i); + let dir = self.partition_dir(i); if !dir.exists() { pb.inc(1); return Ok(()); @@ -268,7 +287,7 @@ impl KmerPartition { let results: Vec> = (0..self.n_partitions) .into_par_iter() .map(|i| { - let dir = self.part_dir(i); + let dir = self.partition_dir(i); let dedup_path = dir.join(format!("dereplicated.{SK_EXT}")); if !dedup_path.exists() { pb.inc(1); @@ -293,7 +312,7 @@ impl KmerPartition { let mut f1: u64 = 0; for i in 0..self.n_partitions { - let path = self.part_dir(i).join("kmer_spectrum_raw.json"); + let path = self.partition_dir(i).join("kmer_spectrum_raw.json"); if !path.exists() { continue; } @@ -328,7 +347,7 @@ impl KmerPartition { fn ensure_writer(&mut self, partition: usize) -> SKResult<&mut SKFileWriter> { if self.writers[partition].is_none() { - let dir = self.root_path.join(PARTITIONS_SUBDIR).join(format!("part_{:05}", partition)); + let dir = self.partition_dir(partition); fs::create_dir_all(&dir)?; let file_path = dir.join(format!("raw.{SK_EXT}")); let writer = SKFileWriter::create_with(file_path, Format::Zstd, self.level)?; diff --git a/src/obikpartitionner/src/query_layer.rs b/src/obikpartitionner/src/query_layer.rs index c2827e0d..3c8cbc6a 100644 --- a/src/obikpartitionner/src/query_layer.rs +++ b/src/obikpartitionner/src/query_layer.rs @@ -4,13 +4,11 @@ use std::path::Path; use obicompactvec::{PersistentBitMatrix, PersistentCompactIntMatrix}; use obikseq::CanonicalKmer; use obiskio::{SKError, SKResult}; -use obilayeredmap::{layer_dir, IndexMode, MphfLayer, OLMError}; +use obilayeredmap::{IndexMode, MphfLayer, OLMError}; use obilayeredmap::meta::PartitionMeta; use crate::partition::KmerPartition; -const INDEX_SUBDIR: &str = "index"; - fn olm_to_sk(e: OLMError) -> SKError { match e { OLMError::Io(io_err) => SKError::Io(io_err), @@ -175,14 +173,14 @@ impl KmerPartition { return Ok(stats); } - let index_dir = self.part_dir(part_idx).join(INDEX_SUBDIR); + let index_dir = self.index_dir(part_idx); if !index_dir.exists() { return Ok(stats); } let meta = PartitionMeta::load(&index_dir).map_err(olm_to_sk)?; let layers: Vec = (0..meta.n_layers) - .map(|i| QueryLayer::open(&layer_dir(&index_dir, i), with_counts, &meta.mode)) + .map(|i| QueryLayer::open(&self.layer_dir(part_idx, i), with_counts, &meta.mode)) .collect::>()?; // ── Stage 1: MPHF-only pass, bucket hits by (layer_idx, slot) ──────── diff --git a/src/obikpartitionner/src/rebuild_layer.rs b/src/obikpartitionner/src/rebuild_layer.rs index d4dc5171..3d23915e 100644 --- a/src/obikpartitionner/src/rebuild_layer.rs +++ b/src/obikpartitionner/src/rebuild_layer.rs @@ -17,8 +17,6 @@ use crate::graph_pipeline::materialize_layer; use crate::merge_layer::{MergeMode, SrcLayerData}; use crate::partition::KmerPartition; -const INDEX_SUBDIR: &str = "index"; - // ── Builders — pair matrix builder + column builders for one mode ───────────── enum Builders { @@ -193,7 +191,7 @@ impl KmerPartition { n_genomes: usize, block_bits: u8, ) -> SKResult<()> { - let src_index_dir = src.part_dir(i).join(INDEX_SUBDIR); + let src_index_dir = src.index_dir(i); if !src_index_dir.exists() { return Ok(()); } @@ -214,8 +212,8 @@ impl KmerPartition { } // ── Build MPHF in dst layer_0 ───────────────────────────────────────── - let dst_index_dir = self.part_dir(i).join(INDEX_SUBDIR); - let dst_layer_dir = dst_index_dir.join("layer_0"); + let dst_index_dir = self.index_dir(i); + let dst_layer_dir = self.layer_dir(i, 0); let n_new = materialize_layer(g, &dst_layer_dir, block_bits, &IndexMode::Exact)?; let dst_mphf = MphfLayer::open(&dst_layer_dir, &IndexMode::Exact) diff --git a/src/obikpartitionner/src/select_layer.rs b/src/obikpartitionner/src/select_layer.rs index 847d1a4a..691dba55 100644 --- a/src/obikpartitionner/src/select_layer.rs +++ b/src/obikpartitionner/src/select_layer.rs @@ -8,13 +8,11 @@ use obicompactvec::{ PersistentCompactIntMatrix, PersistentCompactIntMatrixBuilder, }; use obilayeredmap::meta::PartitionMeta; -use obilayeredmap::{layer_dir, OLMError}; +use obilayeredmap::OLMError; use obiskio::{SKError, SKResult}; use crate::partition::KmerPartition; -const INDEX_SUBDIR: &str = "index"; - // ── AggOp ───────────────────────────────────────────────────────────────────── #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -147,7 +145,7 @@ impl KmerPartition { output_presence: bool, in_place: bool, ) -> SKResult<()> { - let src_index_dir = src.part_dir(i).join(INDEX_SUBDIR); + let src_index_dir = src.index_dir(i); if !src_index_dir.exists() { return Ok(()); } @@ -157,7 +155,7 @@ impl KmerPartition { return Ok(()); } - let dst_index_dir = self.part_dir(i).join(INDEX_SUBDIR); + let dst_index_dir = self.index_dir(i); if !in_place { fs::create_dir_all(&dst_index_dir)?; } @@ -165,10 +163,10 @@ impl KmerPartition { let data_subdir = if output_presence { "presence" } else { "counts" }; for l in 0..src_meta.n_layers { - let src_layer_dir = layer_dir(&src_index_dir, l); + let src_layer_dir = src.layer_dir(i, l); if !src_layer_dir.exists() { continue; } - let dst_layer_dir = layer_dir(&dst_index_dir, l); + let dst_layer_dir = self.layer_dir(i, l); let counts_dir = src_layer_dir.join("counts"); let presence_dir = src_layer_dir.join("presence"); diff --git a/src/obikphylo/src/siblings/build.rs b/src/obikphylo/src/siblings/build.rs index b2760c39..28c0768b 100644 --- a/src/obikphylo/src/siblings/build.rs +++ b/src/obikphylo/src/siblings/build.rs @@ -7,7 +7,7 @@ use rayon::prelude::*; use obikpartitionner::KmerPartition; use obipipeline::ThrottleGuard; use obikseq::CanonicalKmer; -use obilayeredmap::{layer_dir, MphfLayer}; +use obilayeredmap::MphfLayer; use obilayeredmap::meta::PartitionMeta; use obisys::progress_bar; @@ -16,7 +16,7 @@ use obikindex::KmerIndex; use super::cache::PartitionCache; use super::helpers::{central_base, is_minorant}; -use super::{olm_to_ok, FamilyMask, SiblingAnnexBuilder, ANNEX_FILE_NAME, INDEX_SUBDIR}; +use super::{olm_to_ok, FamilyMask, SiblingAnnexBuilder, ANNEX_FILE_NAME}; // ── obipipeline data types ───────────────────────────────────────────────── @@ -92,7 +92,7 @@ impl SiblingAnnexBuildExt for KmerIndex { let pb = progress_bar("sibling_annex", n_parts as u64, "partitions"); let mut total_slots: u64 = 0; for part in 0..n_parts { - let index_dir = self.partition().part_dir(part).join(INDEX_SUBDIR); + let index_dir = self.partition().index_dir(part); if !index_dir.exists() { pb.inc(1); continue; @@ -101,7 +101,7 @@ impl SiblingAnnexBuildExt for KmerIndex { let mut part_slots: u64 = 0; for l in 0..meta.n_layers { - part_slots += build_layer_sibling_annex(self, &layer_dir(&index_dir, l), n_parts, l, &cache)?; + part_slots += build_layer_sibling_annex(self, &self.partition().layer_dir(part, l), n_parts, l, &cache)?; } total_slots += part_slots; pb.inc(1); diff --git a/src/obikphylo/src/siblings/cache.rs b/src/obikphylo/src/siblings/cache.rs index 00cba141..8ffef9fe 100644 --- a/src/obikphylo/src/siblings/cache.rs +++ b/src/obikphylo/src/siblings/cache.rs @@ -5,14 +5,14 @@ use std::path::Path; use obicompactvec::{PersistentBitMatrix, PersistentCompactIntMatrix, PersistentSparseBitMatrix}; use obikpartitionner::KmerPartition; use obikseq::CanonicalKmer; -use obilayeredmap::{layer_dir, Layer, OLMResult}; +use obilayeredmap::{Layer, OLMResult}; use obilayeredmap::meta::{IndexMode, PartitionMeta}; use obisys::progress_bar; use obikindex::OKIResult; use super::iter::SiblingLayerExt; -use super::{olm_to_ok, SiblingAnnex, INDEX_SUBDIR}; +use super::{olm_to_ok, SiblingAnnex}; /// Every partition's already-open layers, built **once** for the whole /// `build_sibling_annex` run and shared (read-only) across every lookup, in @@ -174,7 +174,7 @@ impl PartitionCache { let built: Vec<(Vec, usize)> = (0..n_parts) .into_par_iter() .map(|part| -> OKIResult<(Vec, usize)> { - let index_dir = partition.part_dir(part).join(INDEX_SUBDIR); + let index_dir = partition.index_dir(part); if !index_dir.exists() { pb.inc(1); return Ok((Vec::new(), 0)); @@ -182,7 +182,7 @@ impl PartitionCache { let meta = PartitionMeta::load(&index_dir).map_err(olm_to_ok)?; let mut mats = Vec::with_capacity(meta.n_layers); for l in 0..meta.n_layers { - let Ok(mat) = Mat::open(&layer_dir(&index_dir, l), &meta.mode, with_counts) else { continue }; + let Ok(mat) = Mat::open(&partition.layer_dir(part, l), &meta.mode, with_counts) else { continue }; mats.push(mat); } pb.inc(1); diff --git a/src/obikphylo/src/siblings/family_scan.rs b/src/obikphylo/src/siblings/family_scan.rs index 227a5504..0215bc68 100644 --- a/src/obikphylo/src/siblings/family_scan.rs +++ b/src/obikphylo/src/siblings/family_scan.rs @@ -54,7 +54,6 @@ use std::sync::atomic::{AtomicU8, Ordering}; use rayon::prelude::*; use obikseq::CanonicalKmer; -use obilayeredmap::layer_dir; use obilayeredmap::meta::PartitionMeta; use obipipeline::{ThrottleGuard, throttle}; @@ -64,7 +63,7 @@ use obikindex::KmerIndex; use super::cache::{Mat, PartitionCache}; use super::helpers::central_base; use super::iter::SiblingEntry; -use super::{olm_to_ok, FamilyMask, SiblingAnnex, ANNEX_FILE_NAME, INDEX_SUBDIR}; +use super::{olm_to_ok, FamilyMask, SiblingAnnex, ANNEX_FILE_NAME}; /// Families per batch — see the module docs for the memory-vs-per-partition- /// density trade-off this picks a point on. At ~90 genomes and a few @@ -106,13 +105,13 @@ pub(crate) fn sibling_layer_dirs(index: &KmerIndex) -> OKIResult> { let n_parts = index.n_partitions(); let mut layer_dirs = Vec::new(); for part in 0..n_parts { - let index_dir = index.partition().part_dir(part).join(INDEX_SUBDIR); + let index_dir = index.partition().index_dir(part); if !index_dir.exists() { continue; } let meta = PartitionMeta::load(&index_dir).map_err(olm_to_ok)?; for l in 0..meta.n_layers { - let this_layer_dir = layer_dir(&index_dir, l); + let this_layer_dir = index.partition().layer_dir(part, l); let annex_path = this_layer_dir.join(ANNEX_FILE_NAME); if !annex_path.exists() { return Err(OKIError::InvalidInput(format!( diff --git a/src/obikphylo/src/siblings/mod.rs b/src/obikphylo/src/siblings/mod.rs index 9a1a3078..40458964 100644 --- a/src/obikphylo/src/siblings/mod.rs +++ b/src/obikphylo/src/siblings/mod.rs @@ -75,7 +75,6 @@ use obilayeredmap::OLMError; use obikindex::OKIError; -pub(super) const INDEX_SUBDIR: &str = "index"; pub(super) const ANNEX_FILE_NAME: &str = "siblings.psib"; pub(super) fn olm_to_ok(e: OLMError) -> OKIError { diff --git a/src/obikphylo/src/siblings/tests.rs b/src/obikphylo/src/siblings/tests.rs index 988349e2..93784782 100644 --- a/src/obikphylo/src/siblings/tests.rs +++ b/src/obikphylo/src/siblings/tests.rs @@ -3,7 +3,6 @@ use std::path::Path; use obikseq::{CanonicalKmer, Kmer, Sequence}; use obilayeredmap::MphfLayer; -use obilayeredmap::layer_dir; use obilayeredmap::meta::PartitionMeta; use obisys::Reporter; use tempfile::tempdir; @@ -20,7 +19,7 @@ use super::helpers::is_minorant; use super::sankoff_bundle::SankoffBundleExt; use super::stats::SiblingStatsExt; use super::subsample::EntropyBias; -use super::{FamilyMask, SiblingAnnex, ANNEX_FILE_NAME, INDEX_SUBDIR}; +use super::{FamilyMask, SiblingAnnex, ANNEX_FILE_NAME}; // k must be >= 11 (project constraint, "k ∈ [11,31]"); k=11, level_max=1, // theta=0.0 mirror `obiskbuilder`'s own tests (smaller k/level_max @@ -89,10 +88,10 @@ fn canonical(ascii: &[u8]) -> CanonicalKmer { /// Read back the annex entry for a given canonical k-mer from the merged /// index's (single) partition/layer, asserting it was found at all. fn annex_info_for(idx: &KmerIndex, kmer: CanonicalKmer) -> FamilyMask { - let index_dir = idx.partition().part_dir(0).join(INDEX_SUBDIR); + let index_dir = idx.partition().index_dir(0); let meta = PartitionMeta::load(&index_dir).unwrap(); for l in 0..meta.n_layers { - let layer_dir = layer_dir(&index_dir, l); + let layer_dir = idx.partition().layer_dir(0, l); let mphf = MphfLayer::open(&layer_dir, &meta.mode).unwrap(); if let Some(slot) = mphf.find(kmer) { let annex = SiblingAnnex::open(&layer_dir.join(ANNEX_FILE_NAME)).unwrap(); @@ -168,7 +167,7 @@ fn sibling_annex_works_after_pack_sparse() { let merged = merge_two(dir.path(), &g1, &g2); merged.pack_matrices(true).expect("pack_matrices(sparse)"); - let index_dir = merged.partition().part_dir(0).join(INDEX_SUBDIR); + let index_dir = merged.partition().index_dir(0); assert!( index_dir.join("layer_0").join("presence").join("is_multi.prsb").exists(), "pack_matrices(true) must leave the sparse marker file behind" @@ -311,10 +310,10 @@ fn sibling_annex_no_empty_masks_after_build() { let g1 = build_single_genome_index(dir.path(), "g1", &seq); g1.build_sibling_annex().expect("build_sibling_annex"); - let index_dir = g1.partition().part_dir(0).join(INDEX_SUBDIR); + let index_dir = g1.partition().index_dir(0); let meta = PartitionMeta::load(&index_dir).expect("partition meta"); for l in 0..meta.n_layers { - let layer_dir = layer_dir(&index_dir, l); + let layer_dir = g1.partition().layer_dir(0, l); let annex = SiblingAnnex::open(&layer_dir.join(ANNEX_FILE_NAME)).expect("annex open"); for slot in 0..annex.len() { let mask = annex.get(slot).expect("slot must have an entry"); @@ -362,7 +361,7 @@ fn sibling_annex_records_the_real_layer_of_each_family_member() { let merged = merge_two(dir.path(), &g1, &g2); merged.build_sibling_annex().expect("build_sibling_annex"); - let index_dir = merged.partition().part_dir(0).join(INDEX_SUBDIR); + let index_dir = merged.partition().index_dir(0); let meta = PartitionMeta::load(&index_dir).unwrap(); assert_eq!(meta.n_layers, 2, "fixture assumption: one merge, one new layer");