Push zunrplorkwkt #70

Merged
coissac merged 93 commits from push-zunrplorkwkt into main 2026-08-28 23:15:38 +00:00
7 changed files with 134 additions and 90 deletions
Showing only changes of commit 31bb324752 - Show all commits
+42 -42
View File
@@ -39,7 +39,7 @@ use crate::layer::typed_layer::{COUNTS_DIR, LayerContent, PRESENCE_DIR, TypedLay
/// (`Count`/`Presence`) exist so far; states in between (unitigs written,
/// MPHF built, evidence built, matrix not yet built) are not represented
/// yet.
pub enum Layer {
pub enum KmerLayer {
/// Just a directory — nothing built yet. Every method below other than
/// the path accessors panics on this variant: calling them means the
/// caller assumed a layer was ready when it wasn't, an implementation
@@ -51,13 +51,13 @@ pub enum Layer {
Presence(TypedLayer<PersistentBitMatrix>),
}
impl Layer {
impl KmerLayer {
/// An empty shell at `dir` — creates the directory (if it doesn't
/// already exist) but nothing inside it. The starting state for
/// building a new layer.
pub fn create(dir: &Path) -> std::io::Result<Self> {
std::fs::create_dir_all(dir)?;
Ok(Layer::Empty {
Ok(KmerLayer::Empty {
dir: dir.to_owned(),
})
}
@@ -73,9 +73,9 @@ impl Layer {
pub fn open(layer_dir: &Path, mode: &IndexMode, with_counts: bool) -> OLMResult<Self> {
if with_counts && layer_dir.join(COUNTS_DIR).exists() {
return TypedLayer::<PersistentCompactIntMatrix>::open(layer_dir, mode)
.map(Layer::Count);
.map(KmerLayer::Count);
}
TypedLayer::<PersistentBitMatrix>::open(layer_dir, mode).map(Layer::Presence)
TypedLayer::<PersistentBitMatrix>::open(layer_dir, mode).map(KmerLayer::Presence)
}
// ── Paths — only meaningful before anything is built ───────────────
@@ -93,7 +93,7 @@ impl Layer {
/// again here would mean it lost track of its own state.
pub fn dir(&self) -> &Path {
match self {
Layer::Empty { dir } => dir,
KmerLayer::Empty { dir } => dir,
_ => panic!("Layer::dir() only available on Empty — caller already has this path"),
}
}
@@ -121,33 +121,33 @@ impl Layer {
pub fn content(&self) -> LayerContent {
match self {
Layer::Count(_) => LayerContent::Count,
Layer::Presence(_) => LayerContent::Presence,
Layer::Empty { .. } => panic!("Layer::content() called on an Empty layer"),
KmerLayer::Count(_) => LayerContent::Count,
KmerLayer::Presence(_) => LayerContent::Presence,
KmerLayer::Empty { .. } => panic!("Layer::content() called on an Empty layer"),
}
}
pub fn evidence_kind(&self) -> EvidenceKind {
match self {
Layer::Count(l) => l.evidence_kind(),
Layer::Presence(l) => l.evidence_kind(),
Layer::Empty { .. } => panic!("Layer::evidence_kind() called on an Empty layer"),
KmerLayer::Count(l) => l.evidence_kind(),
KmerLayer::Presence(l) => l.evidence_kind(),
KmerLayer::Empty { .. } => panic!("Layer::evidence_kind() called on an Empty layer"),
}
}
pub fn n(&self) -> usize {
match self {
Layer::Count(l) => l.n(),
Layer::Presence(l) => l.n(),
Layer::Empty { .. } => panic!("Layer::n() called on an Empty layer"),
KmerLayer::Count(l) => l.n(),
KmerLayer::Presence(l) => l.n(),
KmerLayer::Empty { .. } => panic!("Layer::n() called on an Empty layer"),
}
}
pub fn find_slot(&self, kmer: CanonicalKmer) -> Option<usize> {
match self {
Layer::Count(l) => l.find_slot(kmer),
Layer::Presence(l) => l.find_slot(kmer),
Layer::Empty { .. } => panic!("Layer::find_slot() called on an Empty layer"),
KmerLayer::Count(l) => l.find_slot(kmer),
KmerLayer::Presence(l) => l.find_slot(kmer),
KmerLayer::Empty { .. } => panic!("Layer::find_slot() called on an Empty layer"),
}
}
@@ -156,17 +156,17 @@ impl Layer {
/// the evidence check `find_slot`/`find` would perform is redundant.
pub fn hash_batch(&self, kmers: &[CanonicalKmer]) -> Vec<usize> {
match self {
Layer::Count(l) => l.hash_batch(kmers),
Layer::Presence(l) => l.hash_batch(kmers),
Layer::Empty { .. } => panic!("Layer::hash_batch() called on an Empty layer"),
KmerLayer::Count(l) => l.hash_batch(kmers),
KmerLayer::Presence(l) => l.hash_batch(kmers),
KmerLayer::Empty { .. } => panic!("Layer::hash_batch() called on an Empty layer"),
}
}
pub fn n_cols(&self) -> usize {
match self {
Layer::Count(l) => l.n_cols(),
Layer::Presence(l) => l.n_cols(),
Layer::Empty { .. } => panic!("Layer::n_cols() called on an Empty layer"),
KmerLayer::Count(l) => l.n_cols(),
KmerLayer::Presence(l) => l.n_cols(),
KmerLayer::Empty { .. } => panic!("Layer::n_cols() called on an Empty layer"),
}
}
@@ -178,8 +178,8 @@ impl Layer {
/// presence (`!= 0`) in place.
pub fn fill_sub_matrix_carries(&self, slots: &[usize], out: &mut [Vec<bool>]) {
match self {
Layer::Presence(l) => l.fill_sub_matrix(slots, out),
Layer::Count(l) => {
KmerLayer::Presence(l) => l.fill_sub_matrix(slots, out),
KmerLayer::Count(l) => {
let mut counts: Vec<Vec<u32>> = out.iter().map(|_| Vec::new()).collect();
l.fill_sub_matrix(slots, &mut counts);
for (o, c) in out.iter_mut().zip(counts.iter()) {
@@ -187,7 +187,7 @@ impl Layer {
o.extend(c.iter().map(|&v| v != 0));
}
}
Layer::Empty { .. } => {
KmerLayer::Empty { .. } => {
panic!("Layer::fill_sub_matrix_carries() called on an Empty layer")
}
}
@@ -196,18 +196,18 @@ impl Layer {
/// Raw MPHF lookup: kmer → slot, no membership check.
pub fn hash(&self, kmer: CanonicalKmer) -> usize {
match self {
Layer::Count(l) => l.hash(kmer),
Layer::Presence(l) => l.hash(kmer),
Layer::Empty { .. } => panic!("Layer::hash() called on an Empty layer"),
KmerLayer::Count(l) => l.hash(kmer),
KmerLayer::Presence(l) => l.hash(kmer),
KmerLayer::Empty { .. } => panic!("Layer::hash() called on an Empty layer"),
}
}
/// Iterate over all canonical kmers in the layer, in deterministic order.
pub fn iter_kmers(&self) -> crate::layer::mphf_layer::KmerIter {
match self {
Layer::Count(l) => l.iter_kmers(),
Layer::Presence(l) => l.iter_kmers(),
Layer::Empty { .. } => panic!("Layer::iter_kmers() called on an Empty layer"),
KmerLayer::Count(l) => l.iter_kmers(),
KmerLayer::Presence(l) => l.iter_kmers(),
KmerLayer::Empty { .. } => panic!("Layer::iter_kmers() called on an Empty layer"),
}
}
@@ -215,18 +215,18 @@ impl Layer {
/// sequence index in `unitigs.bin`.
pub fn enumerate_kmers(&self) -> std::iter::Enumerate<crate::layer::mphf_layer::KmerIter> {
match self {
Layer::Count(l) => l.enumerate_kmers(),
Layer::Presence(l) => l.enumerate_kmers(),
Layer::Empty { .. } => panic!("Layer::enumerate_kmers() called on an Empty layer"),
KmerLayer::Count(l) => l.enumerate_kmers(),
KmerLayer::Presence(l) => l.enumerate_kmers(),
KmerLayer::Empty { .. } => panic!("Layer::enumerate_kmers() called on an Empty layer"),
}
}
/// Iterate over the layer's canonical kmers in batches of `n`.
pub fn iter_kmers_batch(&self, n: usize) -> crate::layer::mphf_layer::KmerBatchIter {
match self {
Layer::Count(l) => l.iter_kmers_batch(n),
Layer::Presence(l) => l.iter_kmers_batch(n),
Layer::Empty { .. } => panic!("Layer::iter_kmers_batch() called on an Empty layer"),
KmerLayer::Count(l) => l.iter_kmers_batch(n),
KmerLayer::Presence(l) => l.iter_kmers_batch(n),
KmerLayer::Empty { .. } => panic!("Layer::iter_kmers_batch() called on an Empty layer"),
}
}
@@ -237,9 +237,9 @@ impl Layer {
n: usize,
) -> Box<dyn Iterator<Item = (usize, Vec<CanonicalKmer>)> + Send + 'static> {
match self {
Layer::Count(l) => Box::new(l.enumerate_kmers_batch(n)),
Layer::Presence(l) => Box::new(l.enumerate_kmers_batch(n)),
Layer::Empty { .. } => {
KmerLayer::Count(l) => Box::new(l.enumerate_kmers_batch(n)),
KmerLayer::Presence(l) => Box::new(l.enumerate_kmers_batch(n)),
KmerLayer::Empty { .. } => {
panic!("Layer::enumerate_kmers_batch() called on an Empty layer")
}
}
+6 -6
View File
@@ -2,19 +2,19 @@ pub mod content_layer;
pub mod error;
pub mod evidence;
pub mod fingerprint;
pub mod typed_layer;
pub mod layered_store;
pub mod map;
pub mod meta;
pub(crate) mod mphf_layer;
pub mod typed_layer;
pub use content_layer::Layer;
pub use content_layer::KmerLayer;
pub use error::{OLMError, OLMResult};
pub use typed_layer::{
dereplicated_superkmers_path, layer_dir, open_data, raw_superkmers_path, HasLayerContent,
HasStorageKind, Hit, LayerContent, LayerData, TypedLayer,
};
pub use layered_store::LayeredStore;
pub use map::LayeredMap;
pub use meta::{IndexMode, PartitionMeta};
pub use mphf_layer::{EvidenceKind, KmerBatchIter, KmerIter, MphfLayer, MphfOnly};
pub use typed_layer::{
HasLayerContent, HasStorageKind, Hit, LayerContent, LayerData, TypedLayer,
dereplicated_superkmers_path, layer_dir, open_data, raw_superkmers_path,
};
+11 -6
View File
@@ -21,8 +21,8 @@
use std::path::{Path, PathBuf};
use crate::layer::{IndexMode, KmerLayer, OLMResult, layer_dir};
use obikseq::CanonicalKmer;
use crate::layer::{layer_dir, IndexMode, Layer, OLMResult};
/// Partition subdirectory name, under an index's root — the single source
/// of truth for the on-disk `partitions/part_NNNNN` naming convention.
@@ -47,7 +47,7 @@ pub fn index_dir(root: &Path, i: usize) -> PathBuf {
/// One partition's open layers, in layer order (layer 0 first).
pub struct KmerPartition {
layers: Vec<Layer>,
layers: Vec<KmerLayer>,
}
impl KmerPartition {
@@ -55,9 +55,14 @@ impl KmerPartition {
/// `index_dir/layer_1`, ... up to `n_layers`), eagerly — not lazily on
/// first access, so the caller pays the mmap cost once, up front,
/// rather than at an unpredictable point during later lookups.
pub fn open(index_dir: &Path, mode: &IndexMode, n_layers: usize, with_counts: bool) -> OLMResult<Self> {
pub fn open(
index_dir: &Path,
mode: &IndexMode,
n_layers: usize,
with_counts: bool,
) -> OLMResult<Self> {
let layers = (0..n_layers)
.map(|l| Layer::open(&layer_dir(index_dir, l), mode, with_counts))
.map(|l| KmerLayer::open(&layer_dir(index_dir, l), mode, with_counts))
.collect::<OLMResult<Vec<_>>>()?;
Ok(Self { layers })
}
@@ -66,11 +71,11 @@ impl KmerPartition {
self.layers.len()
}
pub fn layer(&self, i: usize) -> &Layer {
pub fn layer(&self, i: usize) -> &KmerLayer {
&self.layers[i]
}
pub fn layers(&self) -> &[Layer] {
pub fn layers(&self) -> &[KmerLayer] {
&self.layers
}
@@ -1,12 +1,12 @@
use std::io;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::time::Instant;
use obikindex::KmerIndex;
use obikindex::layer::KmerLayer;
use obikseq::RoutableSuperKmer;
use obikindex::layer::Layer;
use obiskio::SKResult;
use obisys::Progress;
use tracing::info;
@@ -15,7 +15,7 @@ use niffler::Level;
use niffler::send::compression::Format;
use obiskio::SKFileWriter;
use obipipeline::{throttle, ThrottleGuard, Throttled};
use obipipeline::{ThrottleGuard, Throttled, throttle};
use obiread::NucPage;
// ── Pipeline plumbing, private to `run` ─────────────────────────────────────
@@ -240,13 +240,19 @@ impl<'a> PartitionRouter<'a> {
let now = Instant::now();
if now.duration_since(last_report).as_secs_f64() > REPORT_INTERVAL {
last_report = now;
cb(Progress { position: total_bases, total: None });
cb(Progress {
position: total_bases,
total: None,
});
}
}
self.write_batch(batch)?;
}
if let Some(cb) = on_progress.as_mut() {
cb(Progress { position: total_bases, total: None });
cb(Progress {
position: total_bases,
total: None,
});
}
self.close()
}
@@ -272,7 +278,7 @@ impl<'a> PartitionRouter<'a> {
fn ensure_writer(&mut self, partition: usize) -> SKResult<&mut SKFileWriter> {
if self.writers[partition].is_none() {
let dir = self.layer0_dir(partition);
Layer::create(&dir).map_err(|e| io::Error::other(e.to_string()))?;
KmerLayer::create(&dir).map_err(|e| io::Error::other(e.to_string()))?;
let file_path = obikindex::layer::raw_superkmers_path(&dir);
let writer = SKFileWriter::create_with(file_path, Format::Zstd, self.level)?;
self.writers[partition] = Some(writer);
+7 -6
View File
@@ -1,6 +1,6 @@
use rayon::prelude::*;
use obikindex::layer::Layer;
use obikindex::layer::KmerLayer;
use obikseq::CanonicalKmer;
use obisys::progress_bar;
@@ -31,7 +31,7 @@ use obikindex::{KmerIndex, OKIResult};
/// `DevDocMD/implementation/partition_layer_cache.md`). Its
/// sibling-specific extension (`iter_minorants_batch`) lives in `iter.rs`.
pub(super) struct PartitionCache {
mats: Vec<Vec<Layer>>,
mats: Vec<Vec<KmerLayer>>,
/// Whether every `FamilyMask` field in this index's sibling annexes can
/// be trusted as a real layer index (`n_layers <= 7`, the same decision
/// `build_layer_sibling_annex` makes when writing them) — computed once
@@ -47,9 +47,9 @@ pub(super) struct PartitionCache {
impl PartitionCache {
pub(super) fn build(index: &KmerIndex, n_parts: usize, with_counts: bool) -> OKIResult<Self> {
let pb = progress_bar("open_partitions", n_parts as u64, "partitions");
let built: Vec<(Vec<Layer>, usize)> = (0..n_parts)
let built: Vec<(Vec<KmerLayer>, usize)> = (0..n_parts)
.into_par_iter()
.map(|part| -> OKIResult<(Vec<Layer>, usize)> {
.map(|part| -> OKIResult<(Vec<KmerLayer>, usize)> {
let index_dir = index.index_dir(part);
if !index_dir.exists() {
pb.inc(1);
@@ -58,7 +58,8 @@ impl PartitionCache {
let meta = index.partition_meta(part)?;
let mut mats = Vec::with_capacity(meta.n_layers);
for l in 0..meta.n_layers {
let Ok(mat) = Layer::open(&index.layer_dir(part, l), &meta.mode, with_counts)
let Ok(mat) =
KmerLayer::open(&index.layer_dir(part, l), &meta.mode, with_counts)
else {
continue;
};
@@ -215,7 +216,7 @@ impl PartitionCache {
/// `hits` gets its slots (evidence-probed vs. trusted `index_batch`) — this
/// is everything after that.
fn resolve_layer_hits(
mat: &Layer,
mat: &KmerLayer,
hits: &[(usize, usize, u8)],
n_genomes: usize,
on_hit: &mut impl FnMut(usize, u8, usize),
+3 -3
View File
@@ -60,7 +60,7 @@ use obipipeline::{ThrottleGuard, throttle};
use obikindex::KmerIndex;
use obikindex::{OKIError, OKIResult};
use obikindex::layer::Layer;
use obikindex::layer::KmerLayer;
use super::cache::PartitionCache;
use super::helpers::central_base;
@@ -134,7 +134,7 @@ pub(crate) fn sibling_layer_dirs(index: &KmerIndex) -> OKIResult<Vec<PathBuf>> {
/// `mat` (a `TypedLayer<D>`) already bundles the MPHF, and each `SiblingEntry`
/// arrives with its kmer and mask already in hand from `iter_minorants_batch`.
struct LayerCtx {
mat: Layer,
mat: KmerLayer,
n_parts: usize,
n_genomes: usize,
n_cols: usize,
@@ -197,7 +197,7 @@ pub(super) fn scan_layer_families(
let meta = PartitionMeta::load(index_dir).map_err(olm_to_ok)?;
let annex = Arc::new(SiblingAnnex::open(&layer_dir.join(ANNEX_FILE_NAME))?);
let mat = Layer::open(layer_dir, &meta.mode, with_counts).map_err(olm_to_ok)?;
let mat = KmerLayer::open(layer_dir, &meta.mode, with_counts).map_err(olm_to_ok)?;
let n_cols = mat.n_cols().min(n_genomes);
let ctx = Arc::new(LayerCtx {
+53 -21
View File
@@ -28,8 +28,8 @@
use std::sync::Arc;
use obikindex::layer::{KmerIter, LayerData, TypedLayer};
use obikseq::CanonicalKmer;
use obikindex::layer::{KmerIter, TypedLayer, LayerData};
use super::{FamilyMask, SiblingAnnex};
@@ -151,24 +151,44 @@ pub trait SiblingLayerExt {
/// Like [`iter_minorants`](Self::iter_minorants), yielding `batch_size`
/// minorants at a time.
fn iter_minorants_batch(&self, annex: Arc<SiblingAnnex>, batch_size: usize) -> MinorantBatchIter;
fn iter_minorants_batch(
&self,
annex: Arc<SiblingAnnex>,
batch_size: usize,
) -> MinorantBatchIter;
}
impl<D: LayerData> SiblingLayerExt for TypedLayer<D> {
fn iter_siblings(&self, annex: Arc<SiblingAnnex>) -> SiblingIter {
SiblingIter { kmers: self.iter_kmers(), annex, order: 0 }
SiblingIter {
kmers: self.iter_kmers(),
annex,
order: 0,
}
}
fn iter_siblings_batch(&self, annex: Arc<SiblingAnnex>, batch_size: usize) -> SiblingBatchIter {
SiblingBatchIter { inner: self.iter_siblings(annex), batch_size }
SiblingBatchIter {
inner: self.iter_siblings(annex),
batch_size,
}
}
fn iter_minorants(&self, annex: Arc<SiblingAnnex>) -> MinorantIter {
MinorantIter { inner: self.iter_siblings(annex) }
MinorantIter {
inner: self.iter_siblings(annex),
}
}
fn iter_minorants_batch(&self, annex: Arc<SiblingAnnex>, batch_size: usize) -> MinorantBatchIter {
MinorantBatchIter { inner: self.iter_minorants(annex), batch_size }
fn iter_minorants_batch(
&self,
annex: Arc<SiblingAnnex>,
batch_size: usize,
) -> MinorantBatchIter {
MinorantBatchIter {
inner: self.iter_minorants(annex),
batch_size,
}
}
}
@@ -178,36 +198,48 @@ impl<D: LayerData> SiblingLayerExt for TypedLayer<D> {
/// depend on which `D` is inside), so no boxing is needed. `obikindex::layer`
/// itself can't implement this: "family"/"minorant" are phylo concepts, it
/// stays kmer/slot-mapping only (see the module docs above).
impl SiblingLayerExt for obikindex::layer::Layer {
impl SiblingLayerExt for obikindex::layer::KmerLayer {
fn iter_siblings(&self, annex: Arc<SiblingAnnex>) -> SiblingIter {
match self {
obikindex::layer::Layer::Count(l) => l.iter_siblings(annex),
obikindex::layer::Layer::Presence(l) => l.iter_siblings(annex),
obikindex::layer::Layer::Empty { .. } => panic!("iter_siblings() called on an Empty layer"),
obikindex::layer::KmerLayer::Count(l) => l.iter_siblings(annex),
obikindex::layer::KmerLayer::Presence(l) => l.iter_siblings(annex),
obikindex::layer::KmerLayer::Empty { .. } => {
panic!("iter_siblings() called on an Empty layer")
}
}
}
fn iter_siblings_batch(&self, annex: Arc<SiblingAnnex>, batch_size: usize) -> SiblingBatchIter {
match self {
obikindex::layer::Layer::Count(l) => l.iter_siblings_batch(annex, batch_size),
obikindex::layer::Layer::Presence(l) => l.iter_siblings_batch(annex, batch_size),
obikindex::layer::Layer::Empty { .. } => panic!("iter_siblings_batch() called on an Empty layer"),
obikindex::layer::KmerLayer::Count(l) => l.iter_siblings_batch(annex, batch_size),
obikindex::layer::KmerLayer::Presence(l) => l.iter_siblings_batch(annex, batch_size),
obikindex::layer::KmerLayer::Empty { .. } => {
panic!("iter_siblings_batch() called on an Empty layer")
}
}
}
fn iter_minorants(&self, annex: Arc<SiblingAnnex>) -> MinorantIter {
match self {
obikindex::layer::Layer::Count(l) => l.iter_minorants(annex),
obikindex::layer::Layer::Presence(l) => l.iter_minorants(annex),
obikindex::layer::Layer::Empty { .. } => panic!("iter_minorants() called on an Empty layer"),
obikindex::layer::KmerLayer::Count(l) => l.iter_minorants(annex),
obikindex::layer::KmerLayer::Presence(l) => l.iter_minorants(annex),
obikindex::layer::KmerLayer::Empty { .. } => {
panic!("iter_minorants() called on an Empty layer")
}
}
}
fn iter_minorants_batch(&self, annex: Arc<SiblingAnnex>, batch_size: usize) -> MinorantBatchIter {
fn iter_minorants_batch(
&self,
annex: Arc<SiblingAnnex>,
batch_size: usize,
) -> MinorantBatchIter {
match self {
obikindex::layer::Layer::Count(l) => l.iter_minorants_batch(annex, batch_size),
obikindex::layer::Layer::Presence(l) => l.iter_minorants_batch(annex, batch_size),
obikindex::layer::Layer::Empty { .. } => panic!("iter_minorants_batch() called on an Empty layer"),
obikindex::layer::KmerLayer::Count(l) => l.iter_minorants_batch(annex, batch_size),
obikindex::layer::KmerLayer::Presence(l) => l.iter_minorants_batch(annex, batch_size),
obikindex::layer::KmerLayer::Empty { .. } => {
panic!("iter_minorants_batch() called on an Empty layer")
}
}
}
}