Push zunrplorkwkt #70
@@ -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<D>` | `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<D>` variants | yes, via `PartitionCache` |
|
||||
| `QueryLayer` | `obikpartitionner::query_layer` | one layer, `(MphfLayer, PersistentBitMatrix\|PersistentCompactIntMatrix)` pair, bypasses `Layer<D>` 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<D>` already caches correctly at the granularity
|
||||
of one partition: `open(root)` opens every layer once, keeps `Vec<Layer<D>>`
|
||||
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<D>` 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<D>` (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).
|
||||
@@ -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
|
||||
|
||||
@@ -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 <sparse_root> <dense_root>");
|
||||
@@ -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;
|
||||
|
||||
@@ -149,7 +149,7 @@ impl KmerIndex {
|
||||
/// is enough, no need to scan every partition.
|
||||
pub fn n_layers_per_partition(&self) -> OKIResult<usize> {
|
||||
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");
|
||||
|
||||
@@ -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); },
|
||||
)?;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<LayeredStore<PersistentCompactIntMatrix>> {
|
||||
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::<SKResult<Vec<_>>>()?;
|
||||
@@ -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<LayeredStore<PersistentBitMatrix>> {
|
||||
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::<SKResult<Vec<_>>>()?;
|
||||
|
||||
@@ -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<dyn KmerFilter>],
|
||||
mut cb: impl FnMut(CanonicalKmer, Box<[u32]>) -> bool,
|
||||
) -> SKResult<bool> {
|
||||
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<dyn KmerFilter>],
|
||||
mut cb: impl FnMut(usize, usize, CanonicalKmer, Box<[u32]>) -> bool,
|
||||
) -> SKResult<bool> {
|
||||
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"))?;
|
||||
|
||||
@@ -42,13 +42,13 @@ impl KmerPartition {
|
||||
mode: &IndexMode,
|
||||
block_bits: u8,
|
||||
) -> Result<usize, SKError> {
|
||||
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<Mphf> = 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<PersistentCompactIntVec> = 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"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<usize> {
|
||||
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<PathBuf> = 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;
|
||||
|
||||
@@ -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 (`<partition>/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<SKResult<()>> = (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<SKResult<()>> = (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)?;
|
||||
|
||||
@@ -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<QueryLayer> = (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::<SKResult<_>>()?;
|
||||
|
||||
// ── Stage 1: MPHF-only pass, bucket hits by (layer_idx, slot) ────────
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<Mat>, usize)> = (0..n_parts)
|
||||
.into_par_iter()
|
||||
.map(|part| -> OKIResult<(Vec<Mat>, 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);
|
||||
|
||||
@@ -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<Vec<PathBuf>> {
|
||||
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!(
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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");
|
||||
|
||||
|
||||
Reference in New Issue
Block a user