OBIKmers/obikmer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 34 Wiki Activity

refactor: extract index construction state tracking into extension trait

Browse Source 
Moves pipeline state bookkeeping, including sentinel file marking and spectrum persistence, into the algorithms' run and close methods. Introduces a crate-private extension trait to satisfy Rust's orphan rule while implementing construction-only operations on KmerIndex. Updates helper function visibility for cross-crate access and removes the now-empty index_layer module.
This commit is contained in:
Eric Coissac
2026-08-21 10:38:51 +02:00
parent 31bb324752
commit da3aa5a2cb
16 changed files with 527 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Cargo.lock
Generated
+1
View File
@@ -1563,6 +1563,7 @@ dependencies = [
"memmap2",
"niffler",
"obicompactvec",
"obidebruinj",
"obikindex",
"obikrope",
"obikseq",
src/obikindex/src/index/common.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use obiskio::{SKError, SKResult};
// ── olm_to_sk ────────────────────────────────────────────────────────────────
pub(crate) fn olm_to_sk(e: OLMError, context: &'static str) -> SKError {
pub fn olm_to_sk(e: OLMError, context: &'static str) -> SKError {
match e {
OLMError::Io(e) => SKError::Io(e),
other => SKError::InvalidData {
src/obikindex/src/index/graph_pipeline.rs
+2 -2
View File
@@ -121,7 +121,7 @@ where
/// Phase 2 (write unitigs only): compute degrees, write unitigs to `layer_dir`, drop graph.
///
/// Returns n_kmers. Does NOT build the MPHF — caller does it.
pub(crate) fn write_graph_as_unitigs(g: GraphDeBruijn, layer_dir: &Path) -> SKResult<usize> {
pub fn write_graph_as_unitigs(g: GraphDeBruijn, layer_dir: &Path) -> SKResult<usize> {
let n_kmers = g.len();
g.compute_degrees_and_mark_starts();
std::fs::create_dir_all(layer_dir)?;
@@ -137,7 +137,7 @@ pub(crate) fn write_graph_as_unitigs(g: GraphDeBruijn, layer_dir: &Path) -> SKRe
/// Phase 2 (full): write_graph_as_unitigs + `TypedLayer::<()>::build`.
///
/// Returns n_kmers.
pub(crate) fn materialize_layer(
pub fn materialize_layer(
g: GraphDeBruijn,
layer_dir: &Path,
block_bits: u8,
src/obikindex/src/index/index_layer.rs
-132
View File
@@ -1,132 +0,0 @@
use std::fs;
use std::io;
use cacheline_ef::{CachelineEf, CachelineEfVec};
use epserde::prelude::*;
use obicompactvec::{PersistentCompactIntMatrix, PersistentCompactIntVec};
use obidebruinj::GraphDeBruijn;
use crate::layer::meta::PartitionMeta;
use crate::layer::{IndexMode, TypedLayer};
use obiskio::{SKError, SKFileMeta, SKFileReader};
use ptr_hash::{PtrHash, bucket_fn::CubicEps, hash::Xx64};
use crate::index::common::olm_to_sk;
use crate::index::graph_pipeline::{materialize_layer, write_graph_as_unitigs};
use crate::index::kmer_index::KmerIndex;
type Mphf = PtrHash<u64, CubicEps, CachelineEfVec<Vec<CachelineEf>>, Xx64, Vec<u8>>;
fn remove_if_exists(path: &std::path::Path) {
if let Err(e) = fs::remove_file(path) {
if e.kind() != io::ErrorKind::NotFound {
eprintln!("warning: could not remove {}: {e}", path.display());
}
}
}
impl KmerIndex {
/// Build the layered MPHF index for partition `i`.
///
/// Returns the number of canonical k-mers indexed, or 0 if the partition
/// has no data or its layer was already built (resume-safe).
///
/// Abundance filtering is applied when `min_ab > 1` or `max_ab.is_some()`,
/// using `mphf1.bin` + `counts1.bin` if they exist.
/// Count payload is stored iff `with_counts` is true.
pub fn build_index_layer(
&self,
i: usize,
min_ab: u32,
max_ab: Option<u32>,
with_counts: bool,
mode: &IndexMode,
block_bits: u8,
) -> Result<usize, SKError> {
let layer0_dir = self.layer_dir(i, 0);
let dedup_path = crate::layer::dereplicated_superkmers_path(&layer0_dir);
if !dedup_path.exists() {
return Ok(0);
}
if layer0_dir.join("mphf.bin").exists() {
return Ok(0);
}
let filter_active = min_ab > 1 || max_ab.is_some();
let need_counts = filter_active || with_counts;
let mphf1_opt: Option<Mphf> = if need_counts {
let p = layer0_dir.join("mphf1.bin");
p.exists().then(|| Mphf::load_full(&p).ok()).flatten()
} else {
None
};
let counts1_opt: Option<PersistentCompactIntVec> = if need_counts {
let p = layer0_dir.join("counts1.bin");
p.exists()
.then(|| PersistentCompactIntVec::open(&p).ok())
.flatten()
} else {
None
};
let mut g = GraphDeBruijn::new();
let mut reader = SKFileReader::open(&dedup_path)?;
for sk in reader.iter() {
for kmer in sk.iter_canonical_kmers() {
let accept = if filter_active {
match (&mphf1_opt, &counts1_opt) {
(Some(mphf), Some(counts)) => {
let ab = counts.get(mphf.index(&kmer.raw()));
ab >= min_ab && max_ab.map_or(true, |max| ab <= max)
}
_ => true,
}
} else {
true
};
if accept {
g.push(kmer);
}
}
}
let n_kmers =
if with_counts {
let n = write_graph_as_unitigs(g, &layer0_dir)?;
TypedLayer::<PersistentCompactIntMatrix>::build(&layer0_dir, block_bits, mode, |kmer| {
match (&mphf1_opt, &counts1_opt) {
(Some(mphf), Some(counts)) => counts.get(mphf.index(&kmer.raw())),
_ => 1,
}
})
.map_err(|e| olm_to_sk(e, "layer build"))?;
n
} else {
materialize_layer(g, &layer0_dir, block_bits, mode)?
};
let index_dir = layer0_dir.parent().expect("layer_dir has a parent");
PartitionMeta {
n_layers: 1,
mode: mode.clone(),
}
.save(index_dir)
.map_err(|e| olm_to_sk(e, "layer build"))?;
Ok(n_kmers)
}
/// Remove intermediate build artifacts for partition `i`.
///
/// Deletes `dereplicated.skmer.zst` (+ sidecar), `mphf1.bin`, `counts1.bin`.
pub fn remove_build_artifacts(&self, i: usize) {
let layer0_dir = self.layer_dir(i, 0);
let dedup = crate::layer::dereplicated_superkmers_path(&layer0_dir);
remove_if_exists(&SKFileMeta::sidecar_path(&dedup));
remove_if_exists(&dedup);
remove_if_exists(&layer0_dir.join("mphf1.bin"));
remove_if_exists(&layer0_dir.join("counts1.bin"));
}
}
src/obikindex/src/index/kmer_index.rs
+1 -77
View File
@@ -1,4 +1,3 @@
use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};
@@ -11,7 +10,7 @@ use obikseq::{set_k, set_m};
use crate::index::common::load_meta;
use crate::index::error::{OKIError, OKIResult};
use crate::index::meta::{GenomeInfo, IndexConfig, IndexMeta};
use crate::index::state::{IndexState, SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED};
use crate::index::state::{IndexState, SENTINEL_INDEXED};
pub struct KmerIndex {
pub(crate) root_path: PathBuf,
@@ -209,62 +208,6 @@ impl KmerIndex {
Ok(self.n_layers(0)?)
}
/// Mark scatter as complete and write `scatter.done`.
///
/// If no genome label was set at creation time, one is derived from
/// the index root directory name (stripped of all extensions).
pub fn mark_scattered(&mut self) -> OKIResult<()> {
if self.meta.genomes.is_empty() {
let label = label_from_path(&self.root_path);
self.meta.genomes.push(GenomeInfo::new(label));
self.meta.write(&self.root_path)?;
}
touch(&self.root_path.join(SENTINEL_SCATTERED))?;
Ok(())
}
/// Mark dereplicate+count as complete and write `count.done`.
pub fn mark_counted(&self) -> OKIResult<()> {
touch(&self.root_path.join(SENTINEL_COUNTED))?;
Ok(())
}
/// Mark layer construction as complete and write `index.done`.
pub fn mark_indexed(&self) -> OKIResult<()> {
touch(&self.root_path.join(SENTINEL_INDEXED))?;
Ok(())
}
/// Write `spectrums/{label}.json` from an already-computed kmer
/// spectrum (`f0`/`f1`/abundance histogram). Takes plain values rather
/// than `obikindexer::algorithms::partitionner::KmerSpectrum` — `KmerIndex`
/// is the data model, `PartitionRouter` the algorithm that depends on
/// it (see `DevDocMD/implementation/partition_layer_cache.md`), not
/// the other way around, and this is the only field of that type it
/// actually uses.
pub fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap<u32, u64>) -> OKIResult<()> {
let label = self
.meta
.genomes
.first()
.map(|g| g.label.as_str())
.unwrap_or("unknown");
let spectrums_dir = self.root_path.join("spectrums");
fs::create_dir_all(&spectrums_dir)?;
let path = spectrums_dir.join(format!("{label}.json"));
let spectrum_map: BTreeMap<String, u64> = counts
.iter()
.map(|(&c, &f)| (format!("{c:010}"), f))
.collect();
let f = fs::File::create(&path)?;
serde_json::to_writer_pretty(
f,
&serde_json::json!({ "f0": f0, "f1": f1, "spectrum": spectrum_map }),
)
.map_err(OKIError::Json)?;
Ok(())
}
/// Path to the unitigs file for partition `part`, layer `layer`.
pub fn layer_unitigs_path(&self, part: usize, layer: usize) -> PathBuf {
self.layer_dir(part, layer).join("unitigs.bin")
@@ -370,23 +313,4 @@ impl KmerIndex {
}
}
fn label_from_path(path: &Path) -> String {
let name = path
.file_name()
.unwrap_or(path.as_os_str())
.to_string_lossy()
.into_owned();
let mut s = name;
while let Some(pos) = s.rfind('.') {
s.truncate(pos);
}
if s.is_empty() {
"unknown".to_string()
} else {
s
}
}
fn touch(path: &Path) -> Result<(), std::io::Error> {
fs::File::create(path).map(|_| ())
}
src/obikindex/src/index/mod.rs
+2 -1
View File
@@ -9,7 +9,6 @@ mod dump_layer;
pub mod filter;
mod graph_pipeline;
mod kmer_index;
mod index_layer;
mod matrix_store;
mod merge;
mod merge_layer;
@@ -23,6 +22,8 @@ mod select_layer;
mod stats;
pub use error::{OKIError, OKIResult};
pub use common::olm_to_sk;
pub use graph_pipeline::{materialize_layer, write_graph_as_unitigs};
pub use distance::{DistanceMetric, DistanceOutput};
pub use filter::{GroupQuorumFilter, KmerFilter, passes_all};
pub use kmer_index::KmerIndex;
src/obikindex/src/lib.rs
+1 -1
View File
@@ -17,6 +17,6 @@ pub use index::{
GroupQuorumFilter, IndexBitsPerKmer, IndexConfig, IndexMeta, IndexState, KmerDesc, KmerFilter,
KmerIndex, MergeMode, MetaPred, OKIError, OKIResult, OutputCol, PartitionRunner, QueryHit,
QueryStats, META_FILENAME, SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED,
passes_all,
materialize_layer, olm_to_sk, write_graph_as_unitigs, passes_all,
};
pub use index::{filter, meta};
src/obikindexer/Cargo.toml
+1
View File
@@ -6,6 +6,7 @@ edition = "2024"
[dependencies]
obikindex = { path = "../obikindex" }
obikseq = { path = "../obikseq" }
obidebruinj = { path = "../obidebruinj" }
obiskio = { path = "../obiskio" }
obisys = { path = "../obisys" }
obicompactvec = { path = "../obicompactvec" }
src/obikindexer/src/algorithms/counter/mod.rs
+16 -2
View File
@@ -22,6 +22,8 @@ use sysinfo::System;
use count::count_partition;
use kmer_sort::chunk_size_from_ram;
use crate::extensions::PrivateBuilder;
pub struct KmerSpectrum {
pub f0: u64,
pub f1: u64,
@@ -67,7 +69,9 @@ impl<'a> Counter<'a> {
/// `FnMut`, same reason as `Dereplicator::run`: the counting pass itself
/// is a parallel `par_iter`, so the callback must tolerate concurrent
/// calls. `total: Some(n_partitions)` — known up front. This algorithm
/// never renders anything itself.
/// never renders anything itself. Writes `spectrums/{label}.json` and
/// marks the index as counted (`count.done`) once every partition
/// succeeds.
pub fn run(&self, on_progress: Option<impl Fn(Progress) + Sync>) -> SKResult<KmerSpectrum> {
let sys = System::new_all();
// available_memory() can return 0 on macOS when the compressor page count exceeds
@@ -92,7 +96,10 @@ impl<'a> Counter<'a> {
};
if let Some(cb) = &on_progress {
let pos = done.fetch_add(1, Ordering::Relaxed) + 1;
cb(Progress { position: pos, total: Some(self.n_partitions as u64) });
cb(Progress {
position: pos,
total: Some(self.n_partitions as u64),
});
}
result
})
@@ -128,6 +135,13 @@ impl<'a> Counter<'a> {
}
}
self.index
.write_spectrum(f0, f1, &counts)
.map_err(|e| io::Error::other(e.to_string()))?;
self.index
.mark_counted()
.map_err(|e| io::Error::other(e.to_string()))?;
Ok(KmerSpectrum { f0, f1, counts })
}
}
src/obikindexer/src/algorithms/layer_builder/mod.rs
+17 -4
View File
@@ -8,6 +8,8 @@ use obikindex::{KmerIndex, PartitionRunner};
use obiskio::SKResult;
use obisys::Progress;
use crate::extensions::PrivateBuilder;
/// Builds every partition's real layer 0 in parallel.
///
/// Two-phase construction, same shape as the other pipeline algorithms:
@@ -65,8 +67,9 @@ impl<'a> LayerBuilder<'a> {
/// Build every partition's layer 0 in parallel via `PartitionRunner`
/// (NUMA-aware scheduling — this stage is more CPU/memory-intensive
/// per partition than scatter/dereplicate/count, unlike those it
/// doesn't use plain rayon). Returns the total number of kmers built
/// across all partitions.
/// doesn't use plain rayon). Marks the index as fully built
/// (`index.done`) once every partition succeeds. Returns the total
/// number of kmers built across all partitions.
///
/// `on_progress`, when set, is called once per completed partition —
/// `FnMut + Send`, not `Fn + Sync`: `PartitionRunner::run` calls
@@ -88,12 +91,18 @@ impl<'a> LayerBuilder<'a> {
let runner = PartitionRunner::new();
runner.run(
&order,
|i| self.index.build_index_layer(i, min_ab, max_ab, with_counts, &evidence, block_bits),
|i| {
self.index
.build_index_layer(i, min_ab, max_ab, with_counts, &evidence, block_bits)
},
|_i, n_kmers, _elapsed| {
total_kmers += n_kmers;
done += 1;
if let Some(cb) = on_progress.as_mut() {
cb(Progress { position: done, total: Some(self.n_partitions as u64) });
cb(Progress {
position: done,
total: Some(self.n_partitions as u64),
});
}
},
)?;
@@ -104,6 +113,10 @@ impl<'a> LayerBuilder<'a> {
}
}
self.index
.mark_indexed()
.map_err(|e| std::io::Error::other(e.to_string()))?;
Ok(total_kmers)
}
}
src/obikindexer/src/algorithms/partitionner/router.rs
+11
View File
@@ -18,6 +18,8 @@ use obiskio::SKFileWriter;
use obipipeline::{ThrottleGuard, Throttled, throttle};
use obiread::NucPage;
use crate::extensions::PrivateBuilder;
// ── Pipeline plumbing, private to `run` ─────────────────────────────────────
/// Carrier enum for `obipipeline::make_pipe!`'s two-stage transform — local
@@ -160,6 +162,12 @@ impl<'a> PartitionRouter<'a> {
Ok(())
}
/// Closes every open writer and marks scatter as complete
/// (`scatter.done`) — the single completion point shared by `run`
/// (the file-driven pipeline) and manual `write`/`write_batch` callers
/// (e.g. tests feeding in-memory superkmers directly): whichever path
/// was used, `close` is what both agree means "scatter is done".
/// Idempotent — a second call is a no-op, sentinel included.
pub fn close(&mut self) -> SKResult<()> {
if self.closed {
return Ok(());
@@ -168,6 +176,9 @@ impl<'a> PartitionRouter<'a> {
for writer in self.writers.iter_mut().flatten() {
writer.close()?;
}
self.index
.mark_scattered()
.map_err(|e| std::io::Error::other(e.to_string()))?;
Ok(())
}
src/obikindexer/src/extensions/mod.rs
+261
View File
@@ -0,0 +1,261 @@
//! Construction-only `KmerIndex` operations — semantically meaningless
//! once an index is built and being used for reading (query, dump, phylo,
//! or as a `merge`/`select`/`rebuild` *source*): sentinel marking, the
//! per-partition layer-build primitive, and cleanup of build-time scratch
//! files. Private to this crate — no consumer outside the
//! indexing-pipeline algorithms in `obikindexer::algorithms` should ever
//! need them. See `DevDocMD/implementation/partition_layer_cache.md`,
//! "(8)".
//!
//! Defined here, not in `obikindex`, so it can be genuinely `pub(crate)`
//! while extending a foreign type (`KmerIndex`) — Rust's orphan rule
//! requires the trait itself to be local to the crate that implements it
//! for a foreign type; `obikindex` has no way to make a trait "private
//! except to `obikindexer`" from its own side.
//!
//! Scoped to the six methods exclusive to this crate's own pipeline
//! (`partitionner`/`dereplicator`/`counter`/`layer_builder`).
//! `clear_output_for_create`/`create_skeleton`/`finalize_indexed`/`state`
//! stay inherent on `KmerIndex`, deliberately not moved here: they're also
//! called from `obikindex`'s own `merge`/`select`/`rebuild`/`reindex`
//! modules, which — living inside `obikindex` itself — could never reach a
//! trait defined in `obikindexer` (the dependency only runs
//! `obikindexer → obikindex`, never the other way). Moving those four
//! would require extracting `merge`/`select`/`rebuild`/`reindex` into
//! algorithms first — separate, larger future work, not this one.
use std::collections::BTreeMap;
use std::fs;
use std::io;
use std::path::Path;
use cacheline_ef::{CachelineEf, CachelineEfVec};
use epserde::prelude::*;
use obicompactvec::{PersistentCompactIntMatrix, PersistentCompactIntVec};
use obidebruinj::GraphDeBruijn;
use obikindex::layer::IndexMode;
use obikindex::layer::{TypedLayer, meta::PartitionMeta};
use obikindex::{GenomeInfo, KmerIndex, OKIError, OKIResult};
use obikindex::{SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED};
use obikindex::{materialize_layer, olm_to_sk, write_graph_as_unitigs};
use obiskio::{SKError, SKFileMeta, SKFileReader};
use ptr_hash::{PtrHash, bucket_fn::CubicEps, hash::Xx64};
type Mphf = PtrHash<u64, CubicEps, CachelineEfVec<Vec<CachelineEf>>, Xx64, Vec<u8>>;
pub(crate) trait PrivateBuilder {
/// Mark scatter as complete and write `scatter.done`.
///
/// If no genome label was set at creation time, one is derived from
/// the index root directory name (stripped of all extensions).
fn mark_scattered(&mut self) -> OKIResult<()>;
/// Mark dereplicate+count as complete and write `count.done`.
fn mark_counted(&self) -> OKIResult<()>;
/// Mark layer construction as complete and write `index.done`.
fn mark_indexed(&self) -> OKIResult<()>;
/// Write `spectrums/{label}.json` from an already-computed kmer
/// spectrum (`f0`/`f1`/abundance histogram). Takes plain values rather
/// than `algorithms::counter::KmerSpectrum` — `KmerIndex` is the data
/// model, `Counter` the algorithm that depends on it, not the other
/// way around, and this is the only field of that type it actually
/// uses.
fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap<u32, u64>) -> OKIResult<()>;
/// Build the layered MPHF index for partition `i`.
///
/// Returns the number of canonical k-mers indexed, or 0 if the
/// partition has no data or its layer was already built
/// (resume-safe).
///
/// Abundance filtering is applied when `min_ab > 1` or
/// `max_ab.is_some()`, using `mphf1.bin` + `counts1.bin` if they
/// exist. Count payload is stored iff `with_counts` is true.
fn build_index_layer(
&self,
i: usize,
min_ab: u32,
max_ab: Option<u32>,
with_counts: bool,
mode: &IndexMode,
block_bits: u8,
) -> Result<usize, SKError>;
/// Remove intermediate build artifacts for partition `i`: dereplicated
/// superkmers (+ sidecar), `mphf1.bin`, `counts1.bin`.
fn remove_build_artifacts(&self, i: usize);
}
impl PrivateBuilder for KmerIndex {
fn mark_scattered(&mut self) -> OKIResult<()> {
if self.meta().genomes.is_empty() {
let label = label_from_path(self.root_path());
self.meta_mut().genomes.push(GenomeInfo::new(label));
self.meta().write(self.root_path())?;
}
touch(&self.root_path().join(SENTINEL_SCATTERED))?;
Ok(())
}
fn mark_counted(&self) -> OKIResult<()> {
touch(&self.root_path().join(SENTINEL_COUNTED))?;
Ok(())
}
fn mark_indexed(&self) -> OKIResult<()> {
touch(&self.root_path().join(SENTINEL_INDEXED))?;
Ok(())
}
fn write_spectrum(&self, f0: u64, f1: u64, counts: &BTreeMap<u32, u64>) -> OKIResult<()> {
let label = self
.meta()
.genomes
.first()
.map(|g| g.label.as_str())
.unwrap_or("unknown");
let spectrums_dir = self.root_path().join("spectrums");
fs::create_dir_all(&spectrums_dir)?;
let path = spectrums_dir.join(format!("{label}.json"));
let spectrum_map: BTreeMap<String, u64> = counts
.iter()
.map(|(&c, &f)| (format!("{c:010}"), f))
.collect();
let f = fs::File::create(&path)?;
serde_json::to_writer_pretty(
f,
&serde_json::json!({ "f0": f0, "f1": f1, "spectrum": spectrum_map }),
)
.map_err(OKIError::Json)?;
Ok(())
}
fn build_index_layer(
&self,
i: usize,
min_ab: u32,
max_ab: Option<u32>,
with_counts: bool,
mode: &IndexMode,
block_bits: u8,
) -> Result<usize, SKError> {
let layer0_dir = self.layer_dir(i, 0);
let dedup_path = obikindex::layer::dereplicated_superkmers_path(&layer0_dir);
if !dedup_path.exists() {
return Ok(0);
}
if layer0_dir.join("mphf.bin").exists() {
return Ok(0);
}
let filter_active = min_ab > 1 || max_ab.is_some();
let need_counts = filter_active || with_counts;
let mphf1_opt: Option<Mphf> = if need_counts {
let p = layer0_dir.join("mphf1.bin");
p.exists().then(|| Mphf::load_full(&p).ok()).flatten()
} else {
None
};
let counts1_opt: Option<PersistentCompactIntVec> = if need_counts {
let p = layer0_dir.join("counts1.bin");
p.exists()
.then(|| PersistentCompactIntVec::open(&p).ok())
.flatten()
} else {
None
};
let mut g = GraphDeBruijn::new();
let mut reader = SKFileReader::open(&dedup_path)?;
for sk in reader.iter() {
for kmer in sk.iter_canonical_kmers() {
let accept = if filter_active {
match (&mphf1_opt, &counts1_opt) {
(Some(mphf), Some(counts)) => {
let ab = counts.get(mphf.index(&kmer.raw()));
ab >= min_ab && max_ab.map_or(true, |max| ab <= max)
}
_ => true,
}
} else {
true
};
if accept {
g.push(kmer);
}
}
}
let n_kmers = if with_counts {
let n = write_graph_as_unitigs(g, &layer0_dir)?;
TypedLayer::<PersistentCompactIntMatrix>::build(
&layer0_dir,
block_bits,
mode,
|kmer| match (&mphf1_opt, &counts1_opt) {
(Some(mphf), Some(counts)) => counts.get(mphf.index(&kmer.raw())),
_ => 1,
},
)
.map_err(|e| olm_to_sk(e, "layer build"))?;
n
} else {
materialize_layer(g, &layer0_dir, block_bits, mode)?
};
let index_dir = layer0_dir.parent().expect("layer_dir has a parent");
PartitionMeta {
n_layers: 1,
mode: mode.clone(),
}
.save(index_dir)
.map_err(|e| olm_to_sk(e, "layer build"))?;
Ok(n_kmers)
}
fn remove_build_artifacts(&self, i: usize) {
let layer0_dir = self.layer_dir(i, 0);
let dedup = obikindex::layer::dereplicated_superkmers_path(&layer0_dir);
remove_if_exists(&SKFileMeta::sidecar_path(&dedup));
remove_if_exists(&dedup);
remove_if_exists(&layer0_dir.join("mphf1.bin"));
remove_if_exists(&layer0_dir.join("counts1.bin"));
}
}
// ── private helpers ──────────────────────────────────────────────────────
fn label_from_path(path: &Path) -> String {
let name = path
.file_name()
.unwrap_or(path.as_os_str())
.to_string_lossy()
.into_owned();
let mut s = name;
while let Some(pos) = s.rfind('.') {
s.truncate(pos);
}
if s.is_empty() {
"unknown".to_string()
} else {
s
}
}
fn touch(path: &Path) -> Result<(), io::Error> {
fs::File::create(path).map(|_| ())
}
fn remove_if_exists(path: &Path) {
if let Err(e) = fs::remove_file(path) {
if e.kind() != io::ErrorKind::NotFound {
eprintln!("warning: could not remove {}: {e}", path.display());
}
}
}
src/obikindexer/src/lib.rs
+6 -3
View File
@@ -6,8 +6,11 @@
//! and `obikindex` doesn't grow every algorithm's own dependencies.
//!
//! [`algorithms`] holds each algorithm as its own submodule: `partitionner`
//! (routing raw super-kmers into partitions, then counting), `dereplicator`
//! (deduplicating a partition's raw super-kmers before counting). A future
//! `extensions` module will sit alongside it.
//! (routing raw super-kmers into partitions), `dereplicator` (deduplicating
//! a partition's raw super-kmers), `counter` (counting unique canonical
//! kmers), `layer_builder` (turning dereplicated super-kmers + counts into
//! the real layer 0). [`extensions`] (private — see its own module docs)
//! holds construction-only `KmerIndex` operations these algorithms share.
pub mod algorithms;
pub(crate) mod extensions;
src/obikmer/src/cmd/index/mod.rs
+6 -21
View File
@@ -280,12 +280,7 @@ pub fn run(args: IndexArgs) {
});
pb.finish_and_clear();
rep.push(t.stop());
drop(router); // ends the borrow of `idx` early — `PartitionRouter`'s `Drop` impl would otherwise extend it to the end of scope
idx.mark_scattered().unwrap_or_else(|e| {
eprintln!("error marking scatter done: {e}");
std::process::exit(1);
});
drop(router); // ends the borrow of `idx` early — `PartitionRouter`'s `Drop` impl would otherwise extend it to the end of scope (`run()` already called `close()`, which marks scatter done, internally)
} else {
info!("scatter already done, skipping");
}
@@ -305,7 +300,9 @@ pub fn run(args: IndexArgs) {
let t = Stage::start("count_kmer");
let pb = progress_bar("counting", idx.n_partitions() as u64, "partitions");
let spectrum = Counter::new(&idx)
// `Counter::run` writes `spectrums/{label}.json` and marks count
// done (`count.done`) internally once every partition succeeds.
Counter::new(&idx)
.keep_partial(args.keep_intermediate)
.run(Some(|_: Progress| pb.inc(1)))
.unwrap_or_else(|e| {
@@ -314,15 +311,6 @@ pub fn run(args: IndexArgs) {
});
pb.finish_and_clear();
rep.push(t.stop());
idx.write_spectrum(spectrum.f0, spectrum.f1, &spectrum.counts).unwrap_or_else(|e| {
eprintln!("error: {e}");
std::process::exit(1);
});
idx.mark_counted().unwrap_or_else(|e| {
eprintln!("error marking count done: {e}");
std::process::exit(1);
});
} else {
info!("dereplicate+count already done, skipping");
}
@@ -343,11 +331,8 @@ pub fn run(args: IndexArgs) {
pb.finish_and_clear();
info!("done — {total_kmers} total kmers indexed");
rep.push(t.stop());
idx.mark_indexed().unwrap_or_else(|e| {
eprintln!("error marking index done: {e}");
std::process::exit(1);
});
// `LayerBuilder::run` marks the index done (`index.done`) internally
// once every partition succeeds.
} else {
info!("index already built, skipping");
}
src/obikphylo/src/siblings/tests.rs
+3 -8
View File
@@ -77,19 +77,14 @@ fn build_single_genome_index(dir: &Path, label: &str, seq: &[u8]) -> KmerIndex {
let batch = obiskbuilder::build_superkmers_page(page, K, /* level_max */ 1, /* theta */ 0.0);
router.write_batch(batch).expect("write_batch");
}
router.close().expect("close partition writers");
router.close().expect("close partition writers"); // also marks scatter done
drop(router); // ends the borrow of `idx` early — `PartitionRouter`'s `Drop` impl would otherwise extend it to the end of scope
Dereplicator::new(&idx).run(None::<fn(obisys::Progress)>).expect("dereplicate");
let spectrum = Counter::new(&idx).run(None::<fn(obisys::Progress)>).expect("count_kmer");
idx.mark_scattered().expect("mark_scattered");
idx.write_spectrum(spectrum.f0, spectrum.f1, &spectrum.counts).expect("write_spectrum");
idx.mark_counted().expect("mark_counted");
Counter::new(&idx).run(None::<fn(obisys::Progress)>).expect("count_kmer"); // also writes the spectrum + marks counted
LayerBuilder::new(&idx)
.run(None::<fn(obisys::Progress)>)
.expect("build_layers");
idx.mark_indexed().expect("mark_indexed");
.expect("build_layers"); // also marks indexed
idx
}