diff --git a/src/obikindex/src/siblings/build.rs b/src/obikindex/src/siblings/build.rs index a23896ee..b51a7f32 100644 --- a/src/obikindex/src/siblings/build.rs +++ b/src/obikindex/src/siblings/build.rs @@ -21,13 +21,17 @@ use super::{olm_to_ok, ANNEX_FILE_NAME, INDEX_SUBDIR}; // ── obipipeline data types ───────────────────────────────────────────────── -/// A batch of this layer's distinct k-mers (local MPHF slot + k-mer), the -/// pipeline's source item — batched, not one k-mer per item, so that +/// A batch of this layer's distinct k-mers (iteration-order index + k-mer), +/// the pipeline's source item — batched, not one k-mer per item, so that /// pipeline messages and their synchronisation cost stay amortised over /// thousands of lookups (see `build_layer_sibling_annex`'s comment on /// `BATCH_SIZE`). Carries the throttle permit for the whole batch, moved /// (not cloned) into the corresponding `VariantBatch` — a 1-to-1 transform, /// unlike the fan-out `Flat` stage this replaced, needs no `Arc` sharing. +/// +/// The `usize` is this k-mer's position in `iter_kmers()`'s enumeration +/// order, **not** an MPHF slot — see `docmd/architecture/siblings.md`. It +/// is the same index the annex file is written under. struct SourceBatch { items: Vec<(usize, CanonicalKmer)>, _permit: ThrottleGuard, @@ -36,7 +40,9 @@ struct SourceBatch { /// One batch's worth of central-substitution variants (up to 3 per source /// k-mer), each already routed to its destination partition and carrying /// its own central base (0=A/1=C/2=G/3=T) — the mask bit it will set on a -/// hit. `(dest_partition, variant, source_slot, base)` per entry. +/// hit. `(dest_partition, variant, source_order, base)` per entry, where +/// `source_order` is the source k-mer's iteration-order index (see +/// `SourceBatch`), not an MPHF slot. struct VariantBatch { items: Vec<(usize, CanonicalKmer, usize, u8)>, _permit: ThrottleGuard, @@ -103,7 +109,7 @@ impl KmerIndex { Ok(()) } - /// Returns the number of distinct k-mers (annex slots) processed, for + /// Returns the number of distinct k-mers (annex entries) processed, for /// progress reporting. fn build_layer_sibling_annex( &self, @@ -114,29 +120,31 @@ impl KmerIndex { let index_dir = layer_dir.parent().expect("layer_dir has a parent index dir"); let meta = PartitionMeta::load(index_dir).map_err(olm_to_ok)?; let mphf = MphfLayer::open(layer_dir, &meta.mode).map_err(olm_to_ok)?; - let n_slots = mphf.n(); - - // ── Enumerate this layer's distinct k-mers, one per slot — direct - // slot -> k-mer reconstruction (evidence + direct-access unitigs, no - // MPHF hashing, no file scan), not scan-and-hash-every-k-mer-forward. - let slot_kmer: Vec> = (0..n_slots).map(|slot| mphf.kmer_at(slot)).collect(); - let k = self.kmer_size(); + let n = mphf.n(); - // ── Reconciliation state, initialised with each slot's own base — - // that member is trivially present, no lookup needed. Built before - // the pipeline runs, from the same enumeration, since `sources` - // below is consumed as a throttled iterator, not collected. - // `AtomicU8`, not `FamilyMask`, because the gather phase below - // parallelises across destination partitions (independent - // `query_partition_with` calls, safe to run concurrently) and their - // `Found` hits can land on arbitrary, possibly-shared slots — a - // lock-free `fetch_or` avoids needing any synchronisation beyond - // that. ───────────────────────────────────────────────────────── - let mask: Vec = (0..n_slots).map(|_| AtomicU8::new(0)).collect(); - for (slot, kmer) in slot_kmer.iter().enumerate().filter_map(|(s, k)| k.map(|k| (s, k))) { - mask[slot].fetch_or(1 << central_base(kmer, k), Ordering::Relaxed); - } + // ── Reconciliation state, indexed by this layer's k-mer iteration + // order (the physical layout of `unitigs.bin`), never by MPHF slot + // — this layer's own k-mers are known members by construction, so + // no evidence check, no MPHF slot, and no slot -> k-mer + // reconstruction is needed or legitimate here (see + // `docmd/architecture/siblings.md`: the MPHF is not invertible, and + // evidence answers membership, not identity). The annex is written + // under this same iteration order end to end, so a reader can later + // zip `iter_kmers()` with the annex file directly, with no + // MPHF/slot indirection at read time either. + // + // `Arc>`, not `Vec` — `Pipe::apply` requires + // its source iterator to be `Send + 'static` (its items are + // dispatched to worker threads that outlive this call), so the + // batch-generating closure below needs an owned handle it can move + // in, not a borrow of a local. `AtomicU8`, not `FamilyMask`, + // because the gather phase below parallelises across destination + // partitions (independent `query_partition_with` calls, safe to run + // concurrently) and their `Found` hits can land on arbitrary, + // possibly-shared source entries — a lock-free `fetch_or` avoids + // needing any synchronisation beyond that. ───────────────────────── + let mask: Arc> = Arc::new((0..n).map(|_| AtomicU8::new(0)).collect()); // ── obipipeline: a *batch* transform, not a per-k-mer `Flat` one — // the actual cross-partition lookup reuses @@ -169,16 +177,25 @@ impl KmerIndex { // the pipeline by the accumulation loop below. See // `obipipeline::throttle`'s docs for why this is required, not // optional, once a `Flat`-style stage sits in the pipeline. - let sources: Vec<(usize, CanonicalKmer)> = slot_kmer - .iter() - .enumerate() - .filter_map(|(slot, maybe_kmer)| maybe_kmer.map(|kmer| (slot, kmer))) - .collect(); - let batches: Vec> = sources - .chunks(BATCH_SIZE) - .map(|chunk| chunk.to_vec()) - .collect(); - let throttled = obipipeline::throttle(batches.into_iter(), n_workers).map(|t| SourceBatch { + // + // Batches stream straight from `unitigs.bin` via + // `enumerate_kmers_batch` — never a full-layer `Vec` collect (a + // layer can hold billions of k-mers; see the "no full collect" + // rule). Each k-mer's own base is seeded into `mask` in this same + // pass, since this is exactly the iteration order `mask` is keyed + // on — no separate seeding pass needed. + let seed_mask = Arc::clone(&mask); + let batches = mphf.enumerate_kmers_batch(BATCH_SIZE).map(move |(start, kmers)| { + kmers + .into_iter() + .enumerate() + .map(|(i, kmer)| { + seed_mask[start + i].fetch_or(1 << central_base(kmer, k), Ordering::Relaxed); + (start + i, kmer) + }) + .collect::>() + }); + let throttled = obipipeline::throttle(batches, n_workers).map(|t| SourceBatch { items: t.item, _permit: t.guard, }); @@ -188,7 +205,7 @@ impl KmerIndex { | { move |batch: SourceBatch| -> VariantBatch { let mut items = Vec::with_capacity(batch.items.len() * 3); - for (slot, kmer) in batch.items { + for (order, kmer) in batch.items { for variant in kmer.central_canonical_neighbors() { if variant == kmer { continue; @@ -196,7 +213,7 @@ impl KmerIndex { items.push(( partition_of(variant, n_parts), variant, - slot, + order, central_base(variant, k), )); } @@ -216,8 +233,8 @@ impl KmerIndex { // them. Each batch's throttle permit drops here, once accumulated. let mut outgoing: Vec> = (0..n_parts).map(|_| Vec::new()).collect(); for vb in pipe.apply(throttled, n_workers, capacity) { - for (dest_partition, variant, source_slot, base) in vb.items { - outgoing[dest_partition].push((variant, source_slot, base)); + for (dest_partition, variant, source_order, base) in vb.items { + outgoing[dest_partition].push((variant, source_order, base)); } } @@ -226,33 +243,35 @@ impl KmerIndex { // read-only) so this keeps using multiple cores without giving up // the per-partition locality above. ───────────────────────────── outgoing.par_iter().enumerate().filter(|(_, q)| !q.is_empty()).for_each(|(dest, queries)| { - for &(variant, source_slot, base) in queries { + for &(variant, source_order, base) in queries { if cache.find(dest, variant) { - mask[source_slot].fetch_or(1 << base, Ordering::Relaxed); + mask[source_order].fetch_or(1 << base, Ordering::Relaxed); } } }); - // ── Write the layer's annex file ───────────────────────────────────── - // The minorant flag is computed here, not by every later reader: - // this is the one place the whole family's *final* mask and this - // slot's own k-mer (already in hand, no extra lookup) are both - // available together. Every consumer that only needs to know - // "is this slot the family's minorant" — the common case, since a - // family is tallied once, at its minorant — reads the bit straight - // back instead of re-deriving it (re-scanning `unitigs.bin` and - // re-hashing through the MPHF, the cost `is_minorant` was cheap to - // *compute* but expensive to *get the inputs for* every time). + // ── Write the layer's annex file, indexed by iteration order — a + // second streamed pass over `unitigs.bin` (via `enumerate_kmers`), + // now that every entry's mask is final; never a `Vec` hold of the + // whole layer. The minorant flag is computed here, not by every + // later reader: this is the one place the whole family's *final* + // mask and this entry's own k-mer (already in hand, no extra + // lookup) are both available together. Every consumer that only + // needs to know "is this k-mer the family's minorant" — the common + // case, since a family is tallied once, at its minorant — reads + // the bit straight back instead of re-deriving it (re-scanning + // `unitigs.bin` and re-hashing through the MPHF, the cost + // `is_minorant` was cheap to *compute* but expensive to *get the + // inputs for* every time). let annex_path = layer_dir.join(ANNEX_FILE_NAME); - let mut builder = SiblingAnnexBuilder::new(n_slots, &annex_path)?; - for (slot, m) in mask.iter().enumerate() { - let Some(kmer) = slot_kmer[slot] else { continue }; // unused MPHF slot, if any — leave at the sentinel - let final_mask = FamilyMask::from_bits(m.load(Ordering::Relaxed)); + let mut builder = SiblingAnnexBuilder::new(n, &annex_path)?; + for (order, kmer) in mphf.enumerate_kmers() { + let final_mask = FamilyMask::from_bits(mask[order].load(Ordering::Relaxed)); let minorant = is_minorant(kmer, final_mask, k); - builder.set(slot, final_mask.with_minorant(minorant)); + builder.set(order, final_mask.with_minorant(minorant)); } builder.close()?; - Ok(n_slots as u64) + Ok(n as u64) } } diff --git a/src/obikindex/src/siblings/iter.rs b/src/obikindex/src/siblings/iter.rs new file mode 100644 index 00000000..70b26c1c --- /dev/null +++ b/src/obikindex/src/siblings/iter.rs @@ -0,0 +1,78 @@ +//! Phylo/sibling-domain iteration over a layer — an extension trait, not a +//! new field on `MphfLayer`/`Layer`: "family"/"minorant" are phylo +//! concepts, `obilayeredmap` stays kmer/slot-mapping only (see +//! `docmd/architecture/siblings.md`). +//! +//! The sibling annex is persisted in the same order as `iter_kmers()` +//! (`build_sibling_annex`, see `build.rs`), so pairing them is a plain zip — +//! no MPHF, no slot, no `kmer_at`. Both sides are already `Send + 'static` +//! (`KmerIter` owns an `Arc` clone; `SiblingAnnex` is +//! mmap-backed and handed in as an `Arc` by the caller), so `SiblingIter` +//! streams straight from disk and can be fed to `obipipeline` batch by +//! batch — never collected whole into memory (see the project's "no full +//! collect" rule). + +use std::sync::Arc; + +use obicompactvec::{FamilyMask, SiblingAnnex}; +use obikseq::CanonicalKmer; +use obilayeredmap::{KmerIter, MphfLayer}; + +/// One layer entry: a k-mer's position in the layer's iteration order (the +/// same index the sibling annex is keyed on — not an MPHF slot), the k-mer +/// itself, and its family mask. +#[derive(Debug, Clone, Copy)] +pub struct SiblingEntry { + pub order: usize, + pub kmer: CanonicalKmer, + pub mask: FamilyMask, +} + +/// Streams `(order, kmer, mask)` triples for one layer, in iteration order. +/// Produced by [`SiblingLayerExt::iter_siblings`]. +pub struct SiblingIter { + kmers: KmerIter, + annex: Arc, + order: usize, +} + +impl Iterator for SiblingIter { + type Item = SiblingEntry; + + fn next(&mut self) -> Option { + loop { + let kmer = self.kmers.next()?; + let order = self.order; + self.order += 1; + // `None` means "not yet computed" (see `SiblingAnnex` module + // docs) — shouldn't happen against a fully-built annex, but + // skip rather than misalign the two streams if it does. + if let Some(mask) = self.annex.get(order) { + return Some(SiblingEntry { order, kmer, mask }); + } + } + } +} + +/// Adds phylo/sibling iteration to `MphfLayer`. +pub trait SiblingLayerExt { + /// Zip this layer's k-mers with their sibling-annex entry, in iteration + /// order. `annex` must have been built from this same layer (its length + /// must match the layer's k-mer count). + fn iter_siblings(&self, annex: Arc) -> SiblingIter; + + /// Like [`iter_siblings`](Self::iter_siblings), filtered to the + /// minorant of each family — the common case, since a family is + /// tallied once, at its minorant. + fn iter_minorants(&self, annex: Arc) -> impl Iterator; +} + +impl SiblingLayerExt for MphfLayer { + fn iter_siblings(&self, annex: Arc) -> SiblingIter { + SiblingIter { kmers: self.iter_kmers(), annex, order: 0 } + } + + fn iter_minorants(&self, annex: Arc) -> impl Iterator { + self.iter_siblings(annex).filter(|e| e.mask.is_minorant()) + } +} diff --git a/src/obikindex/src/siblings/mod.rs b/src/obikindex/src/siblings/mod.rs index 2ff427e9..a91711a1 100644 --- a/src/obikindex/src/siblings/mod.rs +++ b/src/obikindex/src/siblings/mod.rs @@ -49,6 +49,7 @@ mod cardinality; mod distance; mod family_scan; mod helpers; +mod iter; mod stats; #[cfg(test)] @@ -57,6 +58,7 @@ mod tests; pub use alignment::SnpAlignment; pub use cardinality::CardinalityTally; pub use distance::{BasePairTally, RawSnpDistanceOutput}; +pub use iter::{SiblingEntry, SiblingIter, SiblingLayerExt}; pub use stats::SiblingAnnexStats; use obilayeredmap::OLMError; diff --git a/src/obilayeredmap/src/layer.rs b/src/obilayeredmap/src/layer.rs index db71218f..44797318 100644 --- a/src/obilayeredmap/src/layer.rs +++ b/src/obilayeredmap/src/layer.rs @@ -85,24 +85,24 @@ impl Layer { } /// Iterate over all canonical kmers in the layer, in deterministic order. - pub fn iter_kmers(&self) -> crate::mphf_layer::KmerIter<'_> { + pub fn iter_kmers(&self) -> crate::mphf_layer::KmerIter { self.mphf.iter_kmers() } /// Iterate over all canonical kmers, each paired with its zero-based /// sequence index in `unitigs.bin`. - pub fn enumerate_kmers(&self) -> std::iter::Enumerate> { + pub fn enumerate_kmers(&self) -> std::iter::Enumerate { self.mphf.enumerate_kmers() } /// Iterate over the layer's canonical kmers in batches of `n`. - pub fn iter_kmers_batch(&self, n: usize) -> crate::mphf_layer::KmerBatchIter<'_> { + pub fn iter_kmers_batch(&self, n: usize) -> crate::mphf_layer::KmerBatchIter { self.mphf.iter_kmers_batch(n) } /// Iterate over batches, each paired with the zero-based index of the /// first kmer in the batch. - pub fn enumerate_kmers_batch(&self, n: usize) -> std::iter::Enumerate> { + pub fn enumerate_kmers_batch(&self, n: usize) -> std::iter::Enumerate { self.mphf.enumerate_kmers_batch(n) } diff --git a/src/obilayeredmap/src/lib.rs b/src/obilayeredmap/src/lib.rs index 65ae2e2a..9c7f9975 100644 --- a/src/obilayeredmap/src/lib.rs +++ b/src/obilayeredmap/src/lib.rs @@ -12,4 +12,4 @@ pub use layer::{Hit, Layer, LayerData}; pub use layered_store::LayeredStore; pub use map::LayeredMap; pub use meta::{IndexMode, PartitionMeta}; -pub use mphf_layer::{MphfLayer, MphfOnly}; +pub use mphf_layer::{KmerBatchIter, KmerIter, MphfLayer, MphfOnly}; diff --git a/src/obilayeredmap/src/mphf_layer.rs b/src/obilayeredmap/src/mphf_layer.rs index 6aa67660..98f6682f 100644 --- a/src/obilayeredmap/src/mphf_layer.rs +++ b/src/obilayeredmap/src/mphf_layer.rs @@ -1,6 +1,7 @@ use std::fs; use std::iter::Enumerate; use std::path::{Path, PathBuf}; +use std::sync::Arc; use cacheline_ef::{CachelineEf, CachelineEfVec}; use epserde::prelude::*; @@ -29,9 +30,9 @@ type MphfEps = PtrHash, Xx // ── LayerEvidence ───────────────────────────────────────────────────────────── enum LayerEvidence { - Exact { evidence: Evidence, unitigs: UnitigFileReader }, - Approx { fingerprint: FingerprintVec, unitigs: UnitigFileReader, unitigs_path: PathBuf }, - Hybrid { evidence: Evidence, unitigs: UnitigFileReader, fingerprint: FingerprintVec }, + Exact { evidence: Evidence, unitigs: Arc }, + Approx { fingerprint: FingerprintVec, unitigs: Arc, unitigs_path: PathBuf }, + Hybrid { evidence: Evidence, unitigs: Arc, fingerprint: FingerprintVec }, } // ── MphfLayer ───────────────────────────────────────────────────────────────── @@ -58,13 +59,13 @@ impl MphfLayer { IndexMode::Exact => { let evidence = Evidence::open(&dir.join(EVIDENCE_FILE))?; let n = evidence.len(); - let unitigs = UnitigFileReader::open(&dir.join(UNITIGS_FILE))?; + let unitigs = Arc::new(UnitigFileReader::open(&dir.join(UNITIGS_FILE))?); (LayerEvidence::Exact { evidence, unitigs }, n) } IndexMode::Approx { .. } => { let fingerprint = FingerprintVec::open(&dir.join(FINGERPRINT_FILE))?; let n = fingerprint.n(); - let unitigs = UnitigFileReader::open(&dir.join(UNITIGS_FILE))?; + let unitigs = Arc::new(UnitigFileReader::open(&dir.join(UNITIGS_FILE))?); let unitigs_path = dir.join(UNITIGS_FILE); (LayerEvidence::Approx { fingerprint, unitigs, unitigs_path }, n) } @@ -72,7 +73,7 @@ impl MphfLayer { let evidence = Evidence::open(&dir.join(EVIDENCE_FILE))?; let fingerprint = FingerprintVec::open(&dir.join(FINGERPRINT_FILE))?; let n = evidence.len(); - let unitigs = UnitigFileReader::open(&dir.join(UNITIGS_FILE))?; + let unitigs = Arc::new(UnitigFileReader::open(&dir.join(UNITIGS_FILE))?); (LayerEvidence::Hybrid { evidence, unitigs, fingerprint }, n) } }; @@ -184,16 +185,20 @@ impl MphfLayer { /// Iterate over all canonical kmers in the layer, in deterministic order. /// /// The iteration order follows the physical layout of `unitigs.bin` and is - /// **not** correlated with MPHF slot numbers. Multiple `KmerIter` instances - /// can be held concurrently for as long as the layer lives, because each + /// **not** correlated with MPHF slot numbers. Owns a clone of the + /// underlying `Arc` rather than borrowing `self` — `Send + /// + 'static`, so it can be handed to a threaded consumer (e.g. + /// `obipipeline`) directly, streamed from disk, with no need to collect + /// the layer's kmers into memory first. Multiple instances can be held + /// concurrently for as long as the underlying file exists, because each /// carries its own cursor. - pub fn iter_kmers(&self) -> KmerIter<'_> { + pub fn iter_kmers(&self) -> KmerIter { let reader = match &self.ev { LayerEvidence::Exact { unitigs, .. } => unitigs, LayerEvidence::Approx { unitigs, .. } => unitigs, LayerEvidence::Hybrid { unitigs, .. } => unitigs, }; - KmerIter { inner: Box::new(reader.iter_indexed_canonical_kmers()) } + KmerIter { inner: Box::new(reader.iter_indexed_canonical_kmers_owned()) } } /// Iterate over all canonical kmers, each paired with its zero-based @@ -202,7 +207,7 @@ impl MphfLayer { /// Yields `(index, kmer)` where `index` starts at `0` for the first kmer /// stored in the layer. This is the standard Rust `enumerate` adapter /// applied to [`iter_kmers`](Self::iter_kmers). - pub fn enumerate_kmers(&self) -> Enumerate> { + pub fn enumerate_kmers(&self) -> Enumerate { self.iter_kmers().enumerate() } @@ -210,7 +215,7 @@ impl MphfLayer { /// /// Each call to [`next`](Iterator::next) returns a [`Vec`] of up to `n` /// kmers. The final batch may be shorter when the layer is exhausted. - pub fn iter_kmers_batch(&self, n: usize) -> KmerBatchIter<'_> { + pub fn iter_kmers_batch(&self, n: usize) -> KmerBatchIter { KmerBatchIter { inner: self.iter_kmers(), batch_size: n } } @@ -220,7 +225,7 @@ impl MphfLayer { /// Yields `(batch_start_index, Vec)` where /// `batch_start_index` is a multiple of `n`. This is the standard Rust /// `enumerate` adapter applied to [`iter_kmers_batch`](Self::iter_kmers_batch). - pub fn enumerate_kmers_batch(&self, n: usize) -> Enumerate> { + pub fn enumerate_kmers_batch(&self, n: usize) -> Enumerate { self.iter_kmers_batch(n).enumerate() } } @@ -229,14 +234,16 @@ impl MphfLayer { /// Iterator over the canonical kmers stored in a layer. /// -/// Produced by [`MphfLayer::iter_kmers`]. Multiple `KmerIter` instances can -/// coexist concurrently for as long as the parent layer lives, because each -/// holds its own cursor over the underlying [`UnitigFileReader`]. -pub struct KmerIter<'a> { - inner: Box + 'a>, +/// Produced by [`MphfLayer::iter_kmers`]. Owns an `Arc` +/// clone internally (via `iter_indexed_canonical_kmers_owned`) instead of +/// borrowing the parent layer — `Send + 'static`, streamed from disk one +/// kmer at a time, never materialised as a whole. Multiple `KmerIter` +/// instances can coexist concurrently, because each holds its own cursor. +pub struct KmerIter { + inner: Box + Send>, } -impl<'a> Iterator for KmerIter<'a> { +impl Iterator for KmerIter { type Item = CanonicalKmer; /// Return the next canonical kmer in iteration order. @@ -250,12 +257,12 @@ impl<'a> Iterator for KmerIter<'a> { /// Produced by [`MphfLayer::iter_kmers_batch`]. Each call to [`next`](Self::next) /// returns a [`Vec`] of up to `batch_size` kmers. The last batch may be shorter /// than `batch_size` when the layer is exhausted. -pub struct KmerBatchIter<'a> { - inner: KmerIter<'a>, +pub struct KmerBatchIter { + inner: KmerIter, batch_size: usize, } -impl<'a> Iterator for KmerBatchIter<'a> { +impl Iterator for KmerBatchIter { type Item = Vec; /// Return the next batch of kmers, or `None` if the layer is exhausted. diff --git a/src/obiskio/src/unitig_index/reader.rs b/src/obiskio/src/unitig_index/reader.rs index e5f0d978..781b8f44 100644 --- a/src/obiskio/src/unitig_index/reader.rs +++ b/src/obiskio/src/unitig_index/reader.rs @@ -1,5 +1,6 @@ use std::fs::File; use std::path::Path; +use std::sync::Arc; use memmap2::Mmap; use obikseq::{CanonicalKmer, Kmer, Unitig}; @@ -203,6 +204,34 @@ impl UnitigFileReader { .map(move |(rank, kmer)| (kmer, chunk_id, rank)) }) } + + /// Same streamed sequence as [`iter_indexed_canonical_kmers`](Self::iter_indexed_canonical_kmers), + /// but owning a clone of `self` instead of borrowing it — `Send + 'static`, + /// so it can be handed to a threaded consumer (e.g. `obipipeline`) without + /// first collecting the layer's k-mers into memory. Reads `mmap` fresh + /// through the `Arc` on every step; no data is duplicated up front. + pub fn iter_indexed_canonical_kmers_owned( + self: &Arc, + ) -> impl Iterator + Send + 'static { + let this = Arc::clone(self); + let k = this.k; + let n = this.n_unitigs; + let mut offset = 0usize; + (0..n) + .map(move |chunk_id| { + let mmap = &*this.mmap; + let seql = mmap[offset] as usize + k; + let byte_len = (seql + 3) / 4; + let bytes = mmap[offset + 1..offset + 1 + byte_len].to_vec().into_boxed_slice(); + offset += 1 + byte_len; + (chunk_id, Unitig::new((seql % 4) as u8, bytes)) + }) + .flat_map(|(chunk_id, u)| { + u.into_canonical_kmers() + .enumerate() + .map(move |(rank, kmer)| (kmer, chunk_id, rank)) + }) + } } fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec)> {