1use crate::Side;
9#[cfg(test)]
10use crate::buckets::BucketRecord;
11use crate::buckets::{
12 BorrowedBucketPackedRecord, BucketError, BucketLocation, BucketManifestEntry, BucketStore,
13};
14use crate::color::{ColorError, ConcurrentColorRepository};
15use crate::dna::Base;
16use crate::hash::{FastBuildHasher, fast_u64_hash, hash_two_u64};
17use crate::kmer::{Kmer, KmerError};
18use crate::state::{ColorCoordinate, UnitigColor, VertexState, source_hash};
19use hashbrown::HashMap;
20use hashbrown::HashTable;
21use hashbrown::hash_map::RawEntryMut;
22use std::collections::HashSet;
23use std::path::Path;
24
25#[derive(Debug, Clone)]
26pub struct DenseLocalVertexMap {
27 indices: HashTable<u32>,
28 entries: Vec<(u64, VertexState)>,
29}
30
31impl DenseLocalVertexMap {
32 fn with_capacity(capacity: usize) -> Self {
33 Self {
34 indices: HashTable::with_capacity(capacity),
35 entries: Vec::with_capacity(capacity),
36 }
37 }
38
39 fn clear_and_reserve(&mut self, capacity: usize) {
40 self.indices.clear();
41 self.entries.clear();
42 if self.entries.capacity() < capacity {
43 self.entries.reserve(capacity - self.entries.capacity());
44 }
45 if self.indices.capacity() < capacity {
46 let entries = &self.entries;
47 self.indices
48 .reserve(capacity - self.indices.capacity(), |&index| {
49 local_u64_hash(entries[index as usize].0)
50 });
51 }
52 }
53
54 #[inline(always)]
55 fn get(&self, key: u64) -> Option<&VertexState> {
56 let hash = local_u64_hash(key);
57 let index = *self
58 .indices
59 .find(hash, |&index| self.entries[index as usize].0 == key)?;
60 Some(&self.entries[index as usize].1)
61 }
62
63 #[inline(always)]
64 fn get_mut(&mut self, key: u64) -> Option<&mut VertexState> {
65 let hash = local_u64_hash(key);
66 let index = *self
67 .indices
68 .find(hash, |&index| self.entries[index as usize].0 == key)?;
69 Some(&mut self.entries[index as usize].1)
70 }
71
72 #[inline(always)]
73 fn state_or_default(&mut self, key: u64) -> &mut VertexState {
74 let hash = local_u64_hash(key);
75 if let Some(&index) = self
76 .indices
77 .find(hash, |&index| self.entries[index as usize].0 == key)
78 {
79 return &mut self.entries[index as usize].1;
80 }
81 let index = self.entries.len() as u32;
82 self.entries.push((key, VertexState::default()));
83 let entries = &self.entries;
84 self.indices.insert_unique(hash, index, |&stored| {
85 local_u64_hash(entries[stored as usize].0)
86 });
87 &mut self.entries[index as usize].1
88 }
89}
90
91impl PartialEq for DenseLocalVertexMap {
92 fn eq(&self, other: &Self) -> bool {
93 self.entries.len() == other.entries.len()
94 && self
95 .entries
96 .iter()
97 .all(|&(key, state)| other.get(key) == Some(&state))
98 }
99}
100
101impl Eq for DenseLocalVertexMap {}
102
103#[derive(Default)]
104struct DenseWantedColorMap {
105 indices: HashTable<u32>,
106 entries: Vec<(u64, usize)>,
107}
108
109impl DenseWantedColorMap {
110 fn with_capacity(capacity: usize) -> Self {
111 Self {
112 indices: HashTable::with_capacity(capacity),
113 entries: Vec::with_capacity(capacity),
114 }
115 }
116
117 #[inline(always)]
118 fn get(&self, key: u64) -> Option<usize> {
119 let hash = local_u64_hash(key);
120 let index = *self
121 .indices
122 .find(hash, |&index| self.entries[index as usize].0 == key)?;
123 Some(self.entries[index as usize].1)
124 }
125
126 fn insert(&mut self, key: u64, value: usize) {
127 let hash = local_u64_hash(key);
128 if let Some(&index) = self
129 .indices
130 .find(hash, |&index| self.entries[index as usize].0 == key)
131 {
132 self.entries[index as usize].1 = value;
133 return;
134 }
135 let index = self.entries.len() as u32;
136 self.entries.push((key, value));
137 let entries = &self.entries;
138 self.indices.insert_unique(hash, index, |&stored| {
139 local_u64_hash(entries[stored as usize].0)
140 });
141 }
142}
143
144enum WantedColorMap<const K: usize> {
145 OneWord(DenseWantedColorMap),
146 TwoWord(HashMap<Kmer<K>, usize, FastBuildHasher>),
147}
148
149impl<const K: usize> WantedColorMap<K> {
150 fn with_capacity(capacity: usize) -> Self {
151 if K <= 32 {
152 Self::OneWord(DenseWantedColorMap::with_capacity(capacity))
153 } else {
154 Self::TwoWord(HashMap::with_capacity_and_hasher(
155 capacity,
156 FastBuildHasher::default(),
157 ))
158 }
159 }
160
161 fn insert(&mut self, key: Kmer<K>, value: usize) {
162 match self {
163 Self::OneWord(map) => map.insert(key.as_u128() as u64, value),
164 Self::TwoWord(map) => {
165 map.insert(key, value);
166 }
167 }
168 }
169
170 #[inline(always)]
171 fn get(&self, key: Kmer<K>) -> Option<usize> {
172 match self {
173 Self::OneWord(map) => map.get(key.as_u128() as u64),
174 Self::TwoWord(map) => map.get(&key).copied(),
175 }
176 }
177}
178
179#[derive(Debug, Clone, PartialEq, Eq)]
180pub enum LocalVertexMap<const K: usize> {
181 OneWord(DenseLocalVertexMap),
182 TwoWord(HashMap<Kmer<K>, VertexState, FastBuildHasher>),
183}
184
185impl<const K: usize> LocalVertexMap<K> {
186 #[inline(always)]
187 fn with_capacity(capacity: usize) -> Self {
188 if K <= 32 {
189 Self::OneWord(DenseLocalVertexMap::with_capacity(capacity))
190 } else {
191 Self::TwoWord(HashMap::with_capacity_and_hasher(
192 capacity,
193 FastBuildHasher::default(),
194 ))
195 }
196 }
197
198 fn clear_and_reserve(&mut self, capacity: usize) {
199 match self {
200 Self::OneWord(map) => {
201 map.clear_and_reserve(capacity);
202 }
203 Self::TwoWord(map) => {
204 map.clear();
205 if map.capacity() < capacity {
206 map.reserve(capacity);
207 }
208 }
209 }
210 }
211
212 pub fn len(&self) -> usize {
213 match self {
214 Self::OneWord(map) => map.entries.len(),
215 Self::TwoWord(map) => map.len(),
216 }
217 }
218
219 pub fn is_empty(&self) -> bool {
220 self.len() == 0
221 }
222
223 pub fn capacity(&self) -> usize {
228 match self {
229 Self::OneWord(map) => map.indices.capacity(),
230 Self::TwoWord(map) => map.capacity(),
231 }
232 }
233
234 #[inline(always)]
235 fn get(&self, kmer: &Kmer<K>) -> Option<&VertexState> {
236 match self {
237 Self::OneWord(map) => map.get(kmer.as_u128() as u64),
238 Self::TwoWord(map) => map.get(kmer),
239 }
240 }
241
242 #[inline(always)]
243 fn get_mut(&mut self, kmer: &Kmer<K>) -> Option<&mut VertexState> {
244 match self {
245 Self::OneWord(map) => map.get_mut(kmer.as_u128() as u64),
246 Self::TwoWord(map) => map.get_mut(kmer),
247 }
248 }
249
250 fn keys_vec(&self) -> Vec<Kmer<K>> {
251 match self {
252 Self::OneWord(map) => map
253 .entries
254 .iter()
255 .map(|&(key, _)| Kmer::<K>::from_bits(key as u128))
256 .collect(),
257 Self::TwoWord(map) => map.keys().copied().collect(),
258 }
259 }
260
261 fn dense_key_state(&self, index: usize) -> Option<(Kmer<K>, VertexState)> {
262 match self {
263 Self::OneWord(map) => map
264 .entries
265 .get(index)
266 .map(|&(key, state)| (Kmer::<K>::from_bits(key as u128), state)),
267 Self::TwoWord(_) => None,
268 }
269 }
270
271 fn dense_len(&self) -> Option<usize> {
272 match self {
273 Self::OneWord(map) => Some(map.entries.len()),
274 Self::TwoWord(_) => None,
275 }
276 }
277
278 #[inline(always)]
279 fn state_or_default(&mut self, kmer: Kmer<K>) -> &mut VertexState {
280 match self {
281 Self::OneWord(map) => {
282 let key = kmer.as_u128() as u64;
283 map.state_or_default(key)
284 }
285 Self::TwoWord(map) => {
286 let hash = local_vertex_hash(kmer);
287 match map
288 .raw_entry_mut()
289 .from_hash(hash, |stored| *stored == kmer)
290 {
291 RawEntryMut::Occupied(entry) => entry.into_mut(),
292 RawEntryMut::Vacant(entry) => {
293 entry
294 .insert_with_hasher(hash, kmer, VertexState::default(), |stored| {
295 local_vertex_hash(*stored)
296 })
297 .1
298 }
299 }
300 }
301 }
302 }
303}
304type LocalEdgeSet<const K: usize> = HashSet<LocalEdge<K>, FastBuildHasher>;
305
306#[derive(Debug, Clone, PartialEq, Eq)]
307pub struct LocalSubgraph<const K: usize> {
308 pub graph_id: usize,
309 pub colored: bool,
310 pub cutoff: u32,
311 pub vertices: LocalVertexMap<K>,
312 pub edges: LocalEdgeSet<K>,
313 pub stats: LocalSubgraphStats,
314}
315
316#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
317pub struct LocalEdge<const K: usize> {
318 pub from: Kmer<K>,
319 pub to: Kmer<K>,
320}
321
322#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
323pub struct LocalSubgraphStats {
324 pub weak_superkmers: u64,
325 pub weak_superkmer_bases: u64,
326 pub observed_vertices: u64,
327 pub unique_vertices: u64,
328 pub observed_edges: u64,
329 pub unique_edges: u64,
330 pub discontinuity_fronts: u64,
331 pub discontinuity_backs: u64,
332 pub isolated_vertices: u64,
333 pub unitigs: u64,
334 pub trivial_unitigs: u64,
335 pub cyclic_unitigs: u64,
336 pub discontinuity_exits: u64,
337 pub unitig_bases: u64,
338}
339
340#[derive(Debug, Clone, PartialEq, Eq)]
341pub struct LocalUnitig<const K: usize> {
342 pub label: Vec<u8>,
343 pub vertices: Vec<Kmer<K>>,
344 color_hashes: Vec<u64>,
345 pub left_exit: Option<(Kmer<K>, Side)>,
346 pub right_exit: Option<(Kmer<K>, Side)>,
347 pub is_cycle: bool,
348}
349
350#[derive(Debug, Clone, PartialEq, Eq)]
351pub struct LocalUnitigColorRun {
352 pub offset: u32,
353 pub color_hash: u64,
354 pub coordinate: Option<ColorCoordinate>,
355 pub sources: Vec<u32>,
356}
357
358#[derive(Debug, Clone, PartialEq, Eq)]
359pub struct ColoredLocalUnitig<const K: usize> {
360 pub unitig: LocalUnitig<K>,
361 pub colors: Vec<LocalUnitigColorRun>,
362}
363
364type PendingColorRun = (u32, u64, Option<ColorCoordinate>);
365
366struct PendingColoredContraction<const K: usize> {
367 unitigs: Vec<LocalUnitig<K>>,
368 runs: Vec<Vec<PendingColorRun>>,
369 representative_indices: HashMap<u64, usize, FastBuildHasher>,
370 source_sets: Vec<Vec<u32>>,
371}
372
373struct PendingColoredData {
374 runs: Vec<Vec<PendingColorRun>>,
375 representative_indices: HashMap<u64, usize, FastBuildHasher>,
376 source_sets: Vec<Vec<u32>>,
377}
378
379#[derive(Debug, Clone, PartialEq, Eq)]
380struct UnitigWalk<const K: usize> {
381 label: Vec<u8>,
382 vertices: WalkVertices<K>,
383 color_hashes: Vec<u64>,
384 is_cycle: bool,
385}
386
387#[derive(Debug, Clone, PartialEq, Eq)]
388struct WalkVertices<const K: usize> {
389 low: Vec<u64>,
390 high: Vec<u64>,
391}
392
393impl<const K: usize> WalkVertices<K> {
394 #[inline]
395 fn clear(&mut self) {
396 self.low.clear();
397 self.high.clear();
398 }
399
400 #[inline]
401 fn push(&mut self, vertex: Kmer<K>) {
402 let words = vertex.words();
403 self.low.push(words[0]);
404 if K > 32 {
405 self.high.push(words[1]);
406 }
407 }
408
409 #[inline]
410 fn len(&self) -> usize {
411 self.low.len()
412 }
413
414 #[inline]
415 fn get(&self, index: usize) -> Kmer<K> {
416 let high = if K > 32 { self.high[index] } else { 0 };
417 Kmer::from_bits(self.low[index] as u128 | ((high as u128) << 64))
418 }
419}
420
421impl<const K: usize> Default for WalkVertices<K> {
422 fn default() -> Self {
423 Self {
424 low: Vec::new(),
425 high: Vec::new(),
426 }
427 }
428}
429
430#[derive(Debug, Clone, PartialEq, Eq)]
431enum WalkTermination<const K: usize> {
432 Branched,
433 Crossed,
434 DeadEnded,
435 Exited(DirectedKmer<K>),
436}
437
438#[derive(Debug, Clone, Copy, PartialEq, Eq)]
439struct DirectedKmer<const K: usize> {
440 observed: Kmer<K>,
441 reverse: Kmer<K>,
442}
443
444impl<const K: usize> DirectedKmer<K> {
445 #[inline]
446 fn canonical(self) -> Kmer<K> {
447 self.observed.min(self.reverse)
448 }
449
450 #[inline]
451 fn in_canonical_form(self) -> bool {
452 self.observed <= self.reverse
453 }
454
455 #[inline]
456 fn entrance_side(self) -> Side {
457 if self.in_canonical_form() {
458 Side::Front
459 } else {
460 Side::Back
461 }
462 }
463
464 #[inline]
465 fn roll_forward(self, base: Base) -> Self {
466 Self {
467 observed: self.observed.roll_forward(base),
468 reverse: self.reverse.roll_backward(base.complement()),
469 }
470 }
471}
472
473impl<const K: usize> UnitigWalk<K> {
474 fn reset(&mut self, v: DirectedKmer<K>, collect_vertices: bool, color_hash: u64) {
475 self.label.clear();
476 v.observed.append_ascii(&mut self.label);
477 self.vertices.clear();
478 self.color_hashes.clear();
479 if collect_vertices {
480 self.vertices.push(v.canonical());
481 self.color_hashes.push(color_hash);
482 }
483 self.is_cycle = false;
484 }
485
486 fn extend(
487 &mut self,
488 v: DirectedKmer<K>,
489 base: Base,
490 anchor: Kmer<K>,
491 collect_vertices: bool,
492 color_hash: u64,
493 ) -> bool {
494 if v.canonical() == anchor {
495 self.is_cycle = true;
496 return false;
497 }
498
499 self.label.push(base.to_ascii());
500 if collect_vertices {
501 self.vertices.push(v.canonical());
502 self.color_hashes.push(color_hash);
503 }
504 true
505 }
506}
507
508impl<const K: usize> Default for UnitigWalk<K> {
509 fn default() -> Self {
510 Self {
511 label: Vec::new(),
512 vertices: WalkVertices::default(),
513 color_hashes: Vec::new(),
514 is_cycle: false,
515 }
516 }
517}
518
519fn reverse_complement_label(label: &[u8]) -> Vec<u8> {
520 label
521 .iter()
522 .rev()
523 .map(|&b| complement_valid_ascii(b))
524 .collect()
525}
526
527#[inline(always)]
528fn complement_valid_ascii(base: u8) -> u8 {
529 const COMPLEMENT: [u8; 8] = *b"NTNGANNC";
530 COMPLEMENT[(base & 7) as usize]
531}
532
533fn canonical_cycle_label<const K: usize>(label: Vec<u8>) -> Vec<u8> {
534 if label.len() <= K {
535 return label;
536 }
537
538 let cycle_len = label.len() - K + 1;
539 let forward = minimal_rotation(&label[..cycle_len]);
540 let reverse = minimal_rotation(&reverse_complement_label(&label[..cycle_len]));
541 let canonical_body = if reverse < forward {
542 reverse.as_slice()
543 } else {
544 forward.as_slice()
545 };
546 linearize_cycle_body::<K>(canonical_body)
547}
548
549fn linearize_cycle_body<const K: usize>(body: &[u8]) -> Vec<u8> {
550 let mut label = Vec::with_capacity(body.len() + K - 1);
551 label.extend_from_slice(body);
552 for i in 0..K - 1 {
553 label.push(body[i % body.len()]);
554 }
555 label
556}
557
558fn minimal_rotation(label: &[u8]) -> Vec<u8> {
559 let start = least_rotation_start(label);
560 label[start..]
561 .iter()
562 .chain(label[..start].iter())
563 .copied()
564 .collect()
565}
566
567fn least_rotation_start(s: &[u8]) -> usize {
568 let n = s.len();
569 if n <= 1 {
570 return 0;
571 }
572
573 let mut i = 0;
574 let mut j = 1;
575 let mut k = 0;
576 while i < n && j < n && k < n {
577 let a = s[(i + k) % n];
578 let b = s[(j + k) % n];
579 if a == b {
580 k += 1;
581 } else if a > b {
582 i += k + 1;
583 if i <= j {
584 i = j + 1;
585 }
586 k = 0;
587 } else {
588 j += k + 1;
589 if j <= i {
590 j = i + 1;
591 }
592 k = 0;
593 }
594 }
595
596 i.min(j)
597}
598
599fn decode_only_diagnostic() -> bool {
601 static DECODE_ONLY: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
602 *DECODE_ONLY.get_or_init(|| std::env::var_os("CF3_RS_DECODE_ONLY").is_some())
603}
604
605impl<const K: usize> LocalSubgraph<K> {
606 pub fn from_bucket_path(
607 path: impl AsRef<Path>,
608 cutoff: u32,
609 ) -> Result<Self, LocalSubgraphError> {
610 let store = BucketStore::files_only();
611 let entries = [BucketManifestEntry {
612 graph_id: 0,
613 records: 0,
614 location: BucketLocation::File(path.as_ref().to_path_buf()),
615 }];
616 Self::from_entries_with_capacity(&store, &entries, cutoff, 0, None)
617 }
618
619 pub fn from_manifest_entries(
620 store: &BucketStore,
621 entries: &[BucketManifestEntry],
622 cutoff: u32,
623 ) -> Result<Self, LocalSubgraphError> {
624 Self::from_entries_with_capacity(store, entries, cutoff, 0, None)
625 }
626
627 pub(crate) fn from_manifest_entries_reusing(
628 store: &BucketStore,
629 entries: &[BucketManifestEntry],
630 cutoff: u32,
631 vertices: Option<LocalVertexMap<K>>,
632 ) -> Result<Self, LocalSubgraphError> {
633 Self::from_entries_with_capacity(store, entries, cutoff, 0, vertices)
634 }
635
636 fn from_entries_with_capacity(
637 store: &BucketStore,
638 entries: &[BucketManifestEntry],
639 cutoff: u32,
640 vertex_capacity: usize,
641 reusable_vertices: Option<LocalVertexMap<K>>,
642 ) -> Result<Self, LocalSubgraphError> {
643 if cutoff == 0 {
644 return Err(LocalSubgraphError::InvalidCutoff);
645 }
646 let Some(first_entry) = entries.first() else {
647 return Err(LocalSubgraphError::EmptyBucketGroup);
648 };
649 let mut reader = store.reader(first_entry)?;
650 if reader.header().k as usize != K {
651 return Err(LocalSubgraphError::KMismatch {
652 expected: K,
653 got: reader.header().k as usize,
654 });
655 }
656
657 let graph_id = reader.header().graph_id;
658 let colored = reader.header().colored;
659 let mut vertices =
660 reusable_vertices.unwrap_or_else(|| LocalVertexMap::with_capacity(vertex_capacity));
661 vertices.clear_and_reserve(vertex_capacity);
662 let mut subgraph = Self {
663 graph_id,
664 colored,
665 cutoff,
666 vertices,
667 edges: HashSet::with_hasher(FastBuildHasher::default()),
668 stats: LocalSubgraphStats::default(),
669 };
670
671 let decode_only = decode_only_diagnostic();
674 reader.try_for_each_borrowed_packed_record(|record| {
675 if decode_only {
676 std::hint::black_box(record.words.first());
677 return Ok(());
678 }
679 subgraph.add_borrowed_packed_record(record)
680 })?;
681 for entry in &entries[1..] {
682 let mut reader = store.reader(entry)?;
683 if reader.header().k as usize != K {
684 return Err(LocalSubgraphError::KMismatch {
685 expected: K,
686 got: reader.header().k as usize,
687 });
688 }
689 if reader.header().graph_id != subgraph.graph_id {
690 return Err(LocalSubgraphError::GraphMismatch {
691 expected: subgraph.graph_id,
692 got: reader.header().graph_id,
693 });
694 }
695 if reader.header().colored != subgraph.colored {
696 return Err(LocalSubgraphError::MalformedRecord);
697 }
698 reader.try_for_each_borrowed_packed_record(|record| {
699 if decode_only {
700 std::hint::black_box(record.words.first());
701 return Ok(());
702 }
703 subgraph.add_borrowed_packed_record(record)
704 })?;
705 }
706
707 subgraph.stats.unique_vertices = subgraph.vertices.len() as u64;
708 subgraph.stats.unique_edges = subgraph.edges.len() as u64;
709
710 Ok(subgraph)
711 }
712
713 pub(crate) fn into_vertex_map(self) -> LocalVertexMap<K> {
714 self.vertices
715 }
716
717 pub fn vertex_state(&self, kmer: Kmer<K>) -> Option<&VertexState> {
718 self.vertices.get(&kmer)
719 }
720
721 pub fn contract(&mut self) -> Result<Vec<LocalUnitig<K>>, LocalSubgraphError> {
722 self.contract_internal(true)
723 }
724
725 pub fn contract_compact(&mut self) -> Result<Vec<LocalUnitig<K>>, LocalSubgraphError> {
726 self.contract_internal(false)
727 }
728
729 pub(crate) fn contract_compact_with<F>(&mut self, emit: F) -> Result<(), LocalSubgraphError>
730 where
731 F: FnMut(LocalUnitig<K>),
732 {
733 self.contract_internal_with(false, emit)
734 }
735
736 pub fn contract_colored(
737 &mut self,
738 store: &BucketStore,
739 entries: &[BucketManifestEntry],
740 ) -> Result<Vec<ColoredLocalUnitig<K>>, LocalSubgraphError> {
741 self.contract_colored_with_known(store, entries, |_| None)
742 }
743
744 pub fn contract_colored_with_known<F>(
745 &mut self,
746 store: &BucketStore,
747 entries: &[BucketManifestEntry],
748 color_is_known: F,
749 ) -> Result<Vec<ColoredLocalUnitig<K>>, LocalSubgraphError>
750 where
751 F: Fn(u64) -> Option<ColorCoordinate>,
752 {
753 let pending = self.contract_colored_impl(store, entries, color_is_known)?;
754 pending
755 .unitigs
756 .into_iter()
757 .zip(pending.runs)
758 .map(|(unitig, runs)| {
759 let colors = runs
760 .into_iter()
761 .map(|(offset, color_hash, coordinate)| {
762 let sources = match coordinate {
763 Some(_) => Vec::new(),
764 None => {
765 let &index = pending
766 .representative_indices
767 .get(&color_hash)
768 .ok_or(LocalSubgraphError::MissingVertex)?;
769 pending.source_sets[index].clone()
770 }
771 };
772 Ok(LocalUnitigColorRun {
773 offset,
774 color_hash,
775 coordinate,
776 sources,
777 })
778 })
779 .collect::<Result<Vec<_>, LocalSubgraphError>>()?;
780 Ok(ColoredLocalUnitig { unitig, colors })
781 })
782 .collect()
783 }
784
785 pub(crate) fn contract_colored_resolved_with<F>(
786 &mut self,
787 store: &BucketStore,
788 entries: &[BucketManifestEntry],
789 repository: &ConcurrentColorRepository,
790 worker: usize,
791 emit: F,
792 ) -> Result<Vec<Vec<UnitigColor>>, LocalSubgraphError>
793 where
794 F: FnMut(LocalUnitig<K>),
795 {
796 let pending = self.contract_colored_impl_with(
797 store,
798 entries,
799 |color_hash| repository.get(color_hash),
800 emit,
801 )?;
802 let mut color_runs = Vec::with_capacity(pending.runs.len());
803 for runs in pending.runs {
804 let mut packed = Vec::with_capacity(runs.len());
805 for (offset, color_hash, coordinate) in runs {
806 let coordinate = match coordinate {
807 Some(coordinate) => coordinate,
808 None => {
809 let &index = pending
810 .representative_indices
811 .get(&color_hash)
812 .ok_or(LocalSubgraphError::MissingVertex)?;
813 repository
814 .resolve_or_insert(color_hash, &pending.source_sets[index], worker)
815 .map_err(LocalSubgraphError::Color)?
816 }
817 };
818 packed.push(UnitigColor::new(offset, coordinate));
819 }
820 color_runs.push(packed);
821 }
822 Ok(color_runs)
823 }
824
825 fn contract_colored_impl<F>(
826 &mut self,
827 store: &BucketStore,
828 entries: &[BucketManifestEntry],
829 color_is_known: F,
830 ) -> Result<PendingColoredContraction<K>, LocalSubgraphError>
831 where
832 F: Fn(u64) -> Option<ColorCoordinate>,
833 {
834 let mut unitigs = Vec::new();
835 let pending =
836 self.contract_colored_impl_with(store, entries, color_is_known, |unitig| {
837 unitigs.push(unitig)
838 })?;
839 Ok(PendingColoredContraction {
840 unitigs,
841 runs: pending.runs,
842 representative_indices: pending.representative_indices,
843 source_sets: pending.source_sets,
844 })
845 }
846
847 fn contract_colored_impl_with<F, G>(
848 &mut self,
849 store: &BucketStore,
850 entries: &[BucketManifestEntry],
851 color_is_known: F,
852 mut emit: G,
853 ) -> Result<PendingColoredData, LocalSubgraphError>
854 where
855 F: Fn(u64) -> Option<ColorCoordinate>,
856 G: FnMut(LocalUnitig<K>),
857 {
858 if !self.colored {
859 return Err(LocalSubgraphError::MalformedRecord);
860 }
861 let mut representative_indices = HashMap::<u64, usize, FastBuildHasher>::default();
862 let mut coordinate_cache =
863 HashMap::<u64, Option<ColorCoordinate>, FastBuildHasher>::default();
864 let mut representatives = Vec::<Kmer<K>>::new();
865 let mut unitig_hash_runs = Vec::new();
866 let mut back_walk = UnitigWalk::default();
867 let mut front_walk = UnitigWalk::default();
868 if let Some(vertex_count) = self.vertices.dense_len()
869 && std::env::var_os("CF3_RS_SORT_LOCAL_VERTICES").is_none()
870 {
871 for index in 0..vertex_count {
872 let (v_hat, state) = self
873 .vertices
874 .dense_key_state(index)
875 .expect("dense vertex index is in bounds");
876 self.contract_colored_vertex(
877 v_hat,
878 state,
879 &color_is_known,
880 &mut emit,
881 &mut unitig_hash_runs,
882 &mut representative_indices,
883 &mut coordinate_cache,
884 &mut representatives,
885 &mut back_walk,
886 &mut front_walk,
887 )?;
888 }
889 } else {
890 let mut vertices = self.vertices.keys_vec();
891 if std::env::var_os("CF3_RS_SORT_LOCAL_VERTICES").is_some() {
892 vertices.sort_unstable();
893 }
894 for v_hat in vertices {
895 let Some(state) = self.vertices.get(&v_hat).copied() else {
896 continue;
897 };
898 self.contract_colored_vertex(
899 v_hat,
900 state,
901 &color_is_known,
902 &mut emit,
903 &mut unitig_hash_runs,
904 &mut representative_indices,
905 &mut coordinate_cache,
906 &mut representatives,
907 &mut back_walk,
908 &mut front_walk,
909 )?;
910 }
911 }
912
913 let mut wanted = WantedColorMap::<K>::with_capacity(representatives.len());
914 for (index, &vertex) in representatives.iter().enumerate() {
915 wanted.insert(vertex, index);
916 }
917 let mut source_sets = vec![Vec::<u32>::new(); representatives.len()];
918 for entry in entries {
919 let mut reader = store.reader(entry)?;
920 reader.try_for_each_borrowed_packed_record(|record| {
921 collect_wanted_color_relations::<K>(record, &wanted, &mut source_sets)
922 })?;
923 }
924 normalize_source_sets(&mut source_sets);
925
926 Ok(PendingColoredData {
927 runs: unitig_hash_runs,
928 representative_indices,
929 source_sets,
930 })
931 }
932
933 #[allow(clippy::too_many_arguments)]
934 fn contract_colored_vertex<F, G>(
935 &mut self,
936 v_hat: Kmer<K>,
937 state: VertexState,
938 color_is_known: &F,
939 emit: &mut G,
940 unitig_hash_runs: &mut Vec<Vec<PendingColorRun>>,
941 representative_indices: &mut HashMap<u64, usize, FastBuildHasher>,
942 coordinate_cache: &mut HashMap<u64, Option<ColorCoordinate>, FastBuildHasher>,
943 representatives: &mut Vec<Kmer<K>>,
944 back_walk: &mut UnitigWalk<K>,
945 front_walk: &mut UnitigWalk<K>,
946 ) -> Result<(), LocalSubgraphError>
947 where
948 F: Fn(u64) -> Option<ColorCoordinate>,
949 G: FnMut(LocalUnitig<K>),
950 {
951 if state.is_visited() {
952 return Ok(());
953 }
954 if state.is_isolated(self.cutoff) {
955 if let Some(state) = self.vertices.get_mut(&v_hat) {
956 state.mark_visited();
957 }
958 self.stats.isolated_vertices += 1;
959 return Ok(());
960 }
961
962 let unitig = self.extract_maximal_unitig_compact(v_hat, back_walk, front_walk)?;
963 self.stats.unitigs += 1;
964 self.stats.unitig_bases += unitig.label.len() as u64;
965 if unitig.is_cycle {
966 self.stats.cyclic_unitigs += 1;
967 }
968 if unitig.left_exit.is_none() && unitig.right_exit.is_none() {
969 self.stats.trivial_unitigs += 1;
970 } else {
971 self.stats.discontinuity_exits +=
972 u64::from(unitig.left_exit.is_some()) + u64::from(unitig.right_exit.is_some());
973 }
974
975 let mut runs = Vec::<(u32, u64, Option<ColorCoordinate>)>::new();
976 let mut record_hash = |offset: usize, vertex: Kmer<K>, color_hash: u64| {
977 if runs
978 .last()
979 .is_none_or(|&(_, previous, _)| previous != color_hash)
980 {
981 let coordinate = *coordinate_cache
982 .entry(color_hash)
983 .or_insert_with(|| color_is_known(color_hash));
984 runs.push((offset as u32, color_hash, coordinate));
985 if coordinate.is_none() {
986 representative_indices.entry(color_hash).or_insert_with(|| {
987 let index = representatives.len();
988 representatives.push(vertex);
989 index
990 });
991 }
992 }
993 };
994 if unitig.is_cycle {
995 let mut vertex = Kmer::<K>::from_ascii(&unitig.label[..K])?;
996 for offset in 0..=unitig.label.len() - K {
997 let canonical = vertex.canonical();
998 let color_hash = self
999 .vertices
1000 .get(&canonical)
1001 .ok_or(LocalSubgraphError::MissingVertex)?
1002 .color_hash();
1003 record_hash(offset, canonical, color_hash);
1004 if offset < unitig.label.len() - K {
1005 vertex = vertex.roll_forward(Base::from_ascii(unitig.label[offset + K]));
1006 }
1007 }
1008 } else {
1009 debug_assert_eq!(front_walk.vertices.len(), front_walk.color_hashes.len());
1010 debug_assert_eq!(back_walk.vertices.len(), back_walk.color_hashes.len());
1011 let mut offset = 0;
1012 for index in (0..front_walk.vertices.len()).rev() {
1013 record_hash(
1014 offset,
1015 front_walk.vertices.get(index),
1016 front_walk.color_hashes[index],
1017 );
1018 offset += 1;
1019 }
1020 for index in 1..back_walk.vertices.len() {
1021 record_hash(
1022 offset,
1023 back_walk.vertices.get(index),
1024 back_walk.color_hashes[index],
1025 );
1026 offset += 1;
1027 }
1028 }
1029 emit(unitig);
1030 unitig_hash_runs.push(runs);
1031 Ok(())
1032 }
1033
1034 fn contract_internal(
1035 &mut self,
1036 collect_vertices: bool,
1037 ) -> Result<Vec<LocalUnitig<K>>, LocalSubgraphError> {
1038 let mut unitigs = Vec::new();
1039 self.contract_internal_with(collect_vertices, |unitig| unitigs.push(unitig))?;
1040 Ok(unitigs)
1041 }
1042
1043 fn contract_internal_with<F>(
1044 &mut self,
1045 collect_vertices: bool,
1046 mut emit: F,
1047 ) -> Result<(), LocalSubgraphError>
1048 where
1049 F: FnMut(LocalUnitig<K>),
1050 {
1051 let mut back_walk = UnitigWalk::default();
1052 let mut front_walk = UnitigWalk::default();
1053 if let Some(vertex_count) = self.vertices.dense_len()
1054 && std::env::var_os("CF3_RS_SORT_LOCAL_VERTICES").is_none()
1055 {
1056 for index in 0..vertex_count {
1057 let (v_hat, state) = self
1058 .vertices
1059 .dense_key_state(index)
1060 .expect("dense vertex index is in bounds");
1061 self.contract_vertex(
1062 v_hat,
1063 state,
1064 collect_vertices,
1065 &mut emit,
1066 &mut back_walk,
1067 &mut front_walk,
1068 )?;
1069 }
1070 } else {
1071 let mut vertices = self.vertices.keys_vec();
1072 if std::env::var_os("CF3_RS_SORT_LOCAL_VERTICES").is_some() {
1073 vertices.sort_unstable();
1074 }
1075 for v_hat in vertices {
1076 let Some(state) = self.vertices.get(&v_hat).copied() else {
1077 continue;
1078 };
1079 self.contract_vertex(
1080 v_hat,
1081 state,
1082 collect_vertices,
1083 &mut emit,
1084 &mut back_walk,
1085 &mut front_walk,
1086 )?;
1087 }
1088 }
1089
1090 Ok(())
1091 }
1092
1093 fn contract_vertex<F>(
1094 &mut self,
1095 v_hat: Kmer<K>,
1096 state: VertexState,
1097 collect_vertices: bool,
1098 emit: &mut F,
1099 back_walk: &mut UnitigWalk<K>,
1100 front_walk: &mut UnitigWalk<K>,
1101 ) -> Result<(), LocalSubgraphError>
1102 where
1103 F: FnMut(LocalUnitig<K>),
1104 {
1105 if state.is_visited() {
1106 return Ok(());
1107 }
1108 if state.is_isolated(self.cutoff) {
1109 if let Some(st) = self.vertices.get_mut(&v_hat) {
1110 st.mark_visited();
1111 }
1112 self.stats.isolated_vertices += 1;
1113 return Ok(());
1114 }
1115
1116 let unitig = self.extract_maximal_unitig(v_hat, collect_vertices, back_walk, front_walk)?;
1117 self.stats.unitigs += 1;
1118 self.stats.unitig_bases += unitig.label.len() as u64;
1119 if unitig.is_cycle {
1120 self.stats.cyclic_unitigs += 1;
1121 }
1122 if unitig.left_exit.is_none() && unitig.right_exit.is_none() {
1123 self.stats.trivial_unitigs += 1;
1124 } else {
1125 self.stats.discontinuity_exits +=
1126 u64::from(unitig.left_exit.is_some()) + u64::from(unitig.right_exit.is_some());
1127 }
1128 emit(unitig);
1129 Ok(())
1130 }
1131
1132 fn extract_maximal_unitig(
1133 &mut self,
1134 v_hat: Kmer<K>,
1135 collect_vertices: bool,
1136 back_walk: &mut UnitigWalk<K>,
1137 front_walk: &mut UnitigWalk<K>,
1138 ) -> Result<LocalUnitig<K>, LocalSubgraphError> {
1139 let back_term = self.walk_unitig(v_hat, Side::Back, collect_vertices, back_walk)?;
1140
1141 if back_walk.is_cycle {
1142 return Ok(LocalUnitig {
1143 label: canonical_cycle_label::<K>(back_walk.label.clone()),
1144 vertices: (0..back_walk.vertices.len())
1145 .map(|index| back_walk.vertices.get(index))
1146 .collect(),
1147 color_hashes: back_walk.color_hashes.clone(),
1148 left_exit: None,
1149 right_exit: None,
1150 is_cycle: true,
1151 });
1152 }
1153
1154 let front_term = self.walk_unitig(v_hat, Side::Front, collect_vertices, front_walk)?;
1155 let vertices = if collect_vertices {
1156 (0..front_walk.vertices.len())
1157 .rev()
1158 .map(|index| front_walk.vertices.get(index))
1159 .chain((1..back_walk.vertices.len()).map(|index| back_walk.vertices.get(index)))
1160 .collect()
1161 } else {
1162 Vec::new()
1163 };
1164 let color_hashes = if collect_vertices {
1165 front_walk
1166 .color_hashes
1167 .iter()
1168 .rev()
1169 .chain(back_walk.color_hashes.iter().skip(1))
1170 .copied()
1171 .collect()
1172 } else {
1173 Vec::new()
1174 };
1175
1176 let mut label = Vec::with_capacity(front_walk.label.len() + back_walk.label.len() - K);
1177 label.extend(
1178 front_walk
1179 .label
1180 .iter()
1181 .rev()
1182 .map(|&base| complement_valid_ascii(base)),
1183 );
1184 label.extend_from_slice(&back_walk.label[K..]);
1185 let left_exit = match front_term {
1186 WalkTermination::Exited(v) => Some((v.canonical(), v.entrance_side())),
1187 _ => None,
1188 };
1189 let right_exit = match back_term {
1190 WalkTermination::Exited(v) => Some((v.canonical(), v.entrance_side())),
1191 _ => None,
1192 };
1193
1194 Ok(LocalUnitig {
1195 label,
1196 vertices,
1197 color_hashes,
1198 left_exit,
1199 right_exit,
1200 is_cycle: false,
1201 })
1202 }
1203
1204 fn extract_maximal_unitig_compact(
1205 &mut self,
1206 v_hat: Kmer<K>,
1207 back_walk: &mut UnitigWalk<K>,
1208 front_walk: &mut UnitigWalk<K>,
1209 ) -> Result<LocalUnitig<K>, LocalSubgraphError> {
1210 let back_term = self.walk_unitig(v_hat, Side::Back, true, back_walk)?;
1211
1212 if back_walk.is_cycle {
1213 return Ok(LocalUnitig {
1214 label: canonical_cycle_label::<K>(back_walk.label.clone()),
1215 vertices: Vec::new(),
1216 color_hashes: Vec::new(),
1217 left_exit: None,
1218 right_exit: None,
1219 is_cycle: true,
1220 });
1221 }
1222
1223 let front_term = self.walk_unitig(v_hat, Side::Front, true, front_walk)?;
1224 let mut label = Vec::with_capacity(front_walk.label.len() + back_walk.label.len() - K);
1225 label.extend(
1226 front_walk
1227 .label
1228 .iter()
1229 .rev()
1230 .map(|&base| complement_valid_ascii(base)),
1231 );
1232 label.extend_from_slice(&back_walk.label[K..]);
1233 let left_exit = match front_term {
1234 WalkTermination::Exited(v) => Some((v.canonical(), v.entrance_side())),
1235 _ => None,
1236 };
1237 let right_exit = match back_term {
1238 WalkTermination::Exited(v) => Some((v.canonical(), v.entrance_side())),
1239 _ => None,
1240 };
1241
1242 Ok(LocalUnitig {
1243 label,
1244 vertices: Vec::new(),
1245 color_hashes: Vec::new(),
1246 left_exit,
1247 right_exit,
1248 is_cycle: false,
1249 })
1250 }
1251
1252 fn walk_unitig(
1253 &mut self,
1254 v_hat: Kmer<K>,
1255 start_side: Side,
1256 collect_vertices: bool,
1257 walk: &mut UnitigWalk<K>,
1258 ) -> Result<WalkTermination<K>, LocalSubgraphError> {
1259 let icc_return_side = start_side.inverse();
1260 let v_hat_reverse = v_hat.reverse_complement();
1261 let mut v = if start_side == Side::Back {
1262 DirectedKmer {
1263 observed: v_hat,
1264 reverse: v_hat_reverse,
1265 }
1266 } else {
1267 DirectedKmer {
1268 observed: v_hat_reverse,
1269 reverse: v_hat,
1270 }
1271 };
1272 let mut side = start_side;
1273 let mut state = {
1274 let state = self
1275 .vertices
1276 .get_mut(&v.canonical())
1277 .ok_or(LocalSubgraphError::MissingVertex)?;
1278 let copied = *state;
1279 state.mark_visited();
1280 copied
1281 };
1282 walk.reset(v, collect_vertices, state.color_hash());
1283
1284 loop {
1285 let mut edge = state.edge_at(side, self.cutoff);
1286 if edge == Base::N {
1287 return Ok(WalkTermination::Branched);
1288 }
1289 if edge == Base::E {
1290 if !state.is_discontinuous(side) {
1291 return Ok(WalkTermination::DeadEnded);
1292 }
1293 return Ok(WalkTermination::Exited(v));
1294 }
1295
1296 if side == Side::Front {
1297 edge = edge.complement();
1298 }
1299 v = v.roll_forward(edge);
1300
1301 let next_state = self
1302 .vertices
1303 .get_mut(&v.canonical())
1304 .ok_or(LocalSubgraphError::MissingVertex)?;
1305 let next_state_copy = *next_state;
1306 side = v.entrance_side();
1307 if next_state_copy.is_branching_side(side, self.cutoff) {
1308 return Ok(WalkTermination::Crossed);
1309 }
1310 if next_state_copy.is_visited() {
1311 if v.canonical() == v_hat && side == icc_return_side {
1312 walk.is_cycle = true;
1313 }
1314 return Ok(WalkTermination::Crossed);
1315 }
1316
1317 next_state.mark_visited();
1318 if !walk.extend(
1319 v,
1320 edge,
1321 v_hat,
1322 collect_vertices,
1323 next_state_copy.color_hash(),
1324 ) {
1325 return Ok(WalkTermination::Crossed);
1326 }
1327 state = next_state_copy;
1328 side = side.inverse();
1329 }
1330 }
1331
1332 #[cfg(test)]
1333 fn add_record(&mut self, record: &BucketRecord) -> Result<(), LocalSubgraphError> {
1334 if record.graph_id != self.graph_id {
1335 return Err(LocalSubgraphError::GraphMismatch {
1336 expected: self.graph_id,
1337 got: record.graph_id,
1338 });
1339 }
1340 if record.len != record.label.len() || record.label.len() < K {
1341 return Err(LocalSubgraphError::MalformedRecord);
1342 }
1343
1344 self.stats.weak_superkmers += 1;
1345 self.stats.weak_superkmer_bases += record.label.len() as u64;
1346
1347 let last_vertex_offset = record.label.len() - K;
1348 let mut observed = Kmer::<K>::from_ascii(&record.label[..K])?;
1349 let mut reverse = observed.reverse_complement();
1350 let mut prev = None;
1351 for offset in 0..=last_vertex_offset {
1352 let in_canonical_form = observed <= reverse;
1353 let canonical = if in_canonical_form { observed } else { reverse };
1354 let pred_base = if offset == 0 {
1355 Base::E
1356 } else {
1357 Base::from_ascii(record.label[offset - 1])
1358 };
1359 let succ_base = if offset == last_vertex_offset {
1360 Base::E
1361 } else {
1362 Base::from_ascii(record.label[offset + K])
1363 };
1364 let mut front = if in_canonical_form {
1365 pred_base
1366 } else {
1367 succ_base.complement()
1368 };
1369 let mut back = if in_canonical_form {
1370 succ_base
1371 } else {
1372 pred_base.complement()
1373 };
1374
1375 if offset > 0 && Some(canonical) == prev {
1376 if in_canonical_form {
1377 front = Base::E;
1378 } else {
1379 back = Base::E;
1380 }
1381 }
1382
1383 let mut discontinuity_fronts = 0;
1384 let mut discontinuity_backs = 0;
1385 {
1386 let state = self.vertex_state_or_default(canonical);
1387 state.update_edges(front, back);
1388 if let Some(source_id) = record.source_id {
1389 state.add_source(source_id);
1390 }
1391 if offset == 0 && record.left_discontinuous {
1392 let side = if in_canonical_form {
1393 Side::Front
1394 } else {
1395 Side::Back
1396 };
1397 state.mark_discontinuous(side);
1398 match side {
1399 Side::Front => discontinuity_fronts += 1,
1400 Side::Back => discontinuity_backs += 1,
1401 }
1402 }
1403 if offset == last_vertex_offset && record.right_discontinuous {
1404 let side = if in_canonical_form {
1405 Side::Back
1406 } else {
1407 Side::Front
1408 };
1409 state.mark_discontinuous(side);
1410 match side {
1411 Side::Front => discontinuity_fronts += 1,
1412 Side::Back => discontinuity_backs += 1,
1413 }
1414 }
1415 }
1416 self.stats.discontinuity_fronts += discontinuity_fronts;
1417 self.stats.discontinuity_backs += discontinuity_backs;
1418 self.stats.observed_vertices += 1;
1419
1420 if let Some(from) = prev {
1421 #[cfg(debug_assertions)]
1422 {
1423 self.edges.insert(LocalEdge {
1424 from,
1425 to: canonical,
1426 });
1427 }
1428 #[cfg(not(debug_assertions))]
1429 let _ = from;
1430 self.stats.observed_edges += 1;
1431 }
1432 prev = Some(canonical);
1433 if offset < last_vertex_offset {
1434 observed = observed.roll_forward(succ_base);
1435 reverse = reverse.roll_backward(succ_base.complement());
1436 }
1437 }
1438
1439 Ok(())
1440 }
1441
1442 fn add_borrowed_packed_record(
1443 &mut self,
1444 record: BorrowedBucketPackedRecord<'_>,
1445 ) -> Result<(), LocalSubgraphError> {
1446 self.add_packed_parts(
1447 record.graph_id,
1448 record.len,
1449 record.source_id,
1450 record.left_discontinuous,
1451 record.right_discontinuous,
1452 record.words,
1453 )
1454 }
1455
1456 #[allow(clippy::too_many_arguments)]
1457 fn add_packed_parts(
1458 &mut self,
1459 graph_id: usize,
1460 len: usize,
1461 source_id: Option<u32>,
1462 left_discontinuous: bool,
1463 right_discontinuous: bool,
1464 words: &[u64],
1465 ) -> Result<(), LocalSubgraphError> {
1466 if graph_id != self.graph_id {
1467 return Err(LocalSubgraphError::GraphMismatch {
1468 expected: self.graph_id,
1469 got: graph_id,
1470 });
1471 }
1472 if len < K || len > words.len() * 32 {
1473 return Err(LocalSubgraphError::MalformedRecord);
1474 }
1475
1476 self.stats.weak_superkmers += 1;
1477 self.stats.weak_superkmer_bases += len as u64;
1478
1479 let last_vertex_offset = len - K;
1480 let observed_bits = packed_prefix_bits::<K>(words);
1481 let mut observed = Kmer::<K>::from_bits(observed_bits);
1482 let mut reverse = observed.reverse_complement();
1483 let mut prev = None;
1484 let source = source_id.map(|source| (source, source_hash(source)));
1485 for offset in 0..=last_vertex_offset {
1486 let in_canonical_form = observed <= reverse;
1487 let canonical = if in_canonical_form { observed } else { reverse };
1488 let pred_base = if offset == 0 {
1489 Base::E
1490 } else {
1491 packed_base(words, offset - 1)
1492 };
1493 let succ_base = if offset == last_vertex_offset {
1494 Base::E
1495 } else {
1496 packed_base(words, offset + K)
1497 };
1498 let mut front = if in_canonical_form {
1499 pred_base
1500 } else {
1501 succ_base.complement()
1502 };
1503 let mut back = if in_canonical_form {
1504 succ_base
1505 } else {
1506 pred_base.complement()
1507 };
1508
1509 if offset > 0 && Some(canonical) == prev {
1510 if in_canonical_form {
1511 front = Base::E;
1512 } else {
1513 back = Base::E;
1514 }
1515 }
1516
1517 let mut discontinuity_fronts = 0;
1518 let mut discontinuity_backs = 0;
1519 {
1520 let state = self.vertex_state_or_default(canonical);
1521 state.update_edges(front, back);
1522 if let Some((source_id, hash)) = source {
1523 state.add_source_hashed(source_id, hash);
1524 }
1525 if offset == 0 && left_discontinuous {
1526 let side = if in_canonical_form {
1527 Side::Front
1528 } else {
1529 Side::Back
1530 };
1531 state.mark_discontinuous(side);
1532 match side {
1533 Side::Front => discontinuity_fronts += 1,
1534 Side::Back => discontinuity_backs += 1,
1535 }
1536 }
1537 if offset == last_vertex_offset && right_discontinuous {
1538 let side = if in_canonical_form {
1539 Side::Back
1540 } else {
1541 Side::Front
1542 };
1543 state.mark_discontinuous(side);
1544 match side {
1545 Side::Front => discontinuity_fronts += 1,
1546 Side::Back => discontinuity_backs += 1,
1547 }
1548 }
1549 }
1550 self.stats.discontinuity_fronts += discontinuity_fronts;
1551 self.stats.discontinuity_backs += discontinuity_backs;
1552 self.stats.observed_vertices += 1;
1553
1554 if let Some(from) = prev {
1555 #[cfg(debug_assertions)]
1556 {
1557 self.edges.insert(LocalEdge {
1558 from,
1559 to: canonical,
1560 });
1561 }
1562 #[cfg(not(debug_assertions))]
1563 let _ = from;
1564 self.stats.observed_edges += 1;
1565 }
1566 prev = Some(canonical);
1567 if offset < last_vertex_offset {
1568 observed = observed.roll_forward(succ_base);
1569 reverse = reverse.roll_backward(succ_base.complement());
1570 }
1571 }
1572
1573 Ok(())
1574 }
1575
1576 #[inline]
1577 fn vertex_state_or_default(&mut self, kmer: Kmer<K>) -> &mut VertexState {
1578 self.vertices.state_or_default(kmer)
1579 }
1580}
1581
1582fn normalize_source_sets(source_sets: &mut [Vec<u32>]) {
1583 let max_source = source_sets
1584 .iter()
1585 .flat_map(|sources| sources.iter().copied())
1586 .max()
1587 .unwrap_or(0) as usize;
1588 let mut words = vec![0u64; max_source / 64 + 1];
1589 let mut touched = Vec::<usize>::new();
1590 for sources in source_sets {
1591 if sources.len() < 2 {
1592 continue;
1593 }
1594 touched.clear();
1595 for source in sources.drain(..) {
1596 let word_index = source as usize / 64;
1597 if words[word_index] == 0 {
1598 touched.push(word_index);
1599 }
1600 words[word_index] |= 1u64 << (source & 63);
1601 }
1602 touched.sort_unstable();
1603 for &word_index in &touched {
1604 let mut word = words[word_index];
1605 while word != 0 {
1606 let bit = word.trailing_zeros();
1607 sources.push((word_index as u32) * 64 + bit);
1608 word &= word - 1;
1609 }
1610 words[word_index] = 0;
1611 }
1612 }
1613}
1614
1615#[allow(dead_code)]
1617fn canonical_vertices_from_label<const K: usize>(
1618 label: &[u8],
1619) -> Result<Vec<Kmer<K>>, LocalSubgraphError> {
1620 if label.len() < K {
1621 return Err(LocalSubgraphError::MalformedRecord);
1622 }
1623 let mut vertices = Vec::with_capacity(label.len() - K + 1);
1624 let mut vertex = Kmer::<K>::from_ascii(&label[..K])?;
1625 vertices.push(vertex.canonical());
1626 for &base in &label[K..] {
1627 vertex = vertex.roll_forward(Base::from_ascii(base));
1628 vertices.push(vertex.canonical());
1629 }
1630 Ok(vertices)
1631}
1632
1633fn collect_wanted_color_relations<const K: usize>(
1634 record: BorrowedBucketPackedRecord<'_>,
1635 wanted: &WantedColorMap<K>,
1636 source_sets: &mut [Vec<u32>],
1637) -> Result<(), LocalSubgraphError> {
1638 let source = record
1639 .source_id
1640 .ok_or(LocalSubgraphError::MalformedRecord)?;
1641 if record.len < K || record.len > record.words.len() * 32 {
1642 return Err(LocalSubgraphError::MalformedRecord);
1643 }
1644 let mut vertex = Kmer::<K>::from_bits(packed_prefix_bits::<K>(record.words));
1645 let mut reverse = vertex.reverse_complement();
1646 for offset in 0..=record.len - K {
1647 let canonical = vertex.min(reverse);
1648 if let Some(color_index) = wanted.get(canonical) {
1649 let sources = &mut source_sets[color_index];
1650 if sources.last().copied() != Some(source) {
1651 sources.push(source);
1652 }
1653 }
1654 if offset < record.len - K {
1655 let next = packed_base(record.words, offset + K);
1656 vertex = vertex.roll_forward(next);
1657 reverse = reverse.roll_backward(next.complement());
1658 }
1659 }
1660 Ok(())
1661}
1662
1663#[inline]
1664fn packed_base(words: &[u64], idx: usize) -> Base {
1665 let word_idx = idx / 32;
1666 let shift = 2 * (31 - (idx % 32));
1667 match ((words[word_idx] >> shift) & 0b11) as u8 {
1668 0 => Base::A,
1669 1 => Base::C,
1670 2 => Base::G,
1671 3 => Base::T,
1672 _ => unreachable!(),
1673 }
1674}
1675
1676#[inline]
1677fn packed_prefix_bits<const K: usize>(words: &[u64]) -> u128 {
1678 debug_assert!((1..=63).contains(&K));
1679 debug_assert!(words.len() * 32 >= K);
1680 if K <= 32 {
1681 (words[0] >> (2 * (32 - K))) as u128
1682 } else {
1683 let tail_bases = K - 32;
1684 ((words[0] as u128) << (2 * tail_bases)) | (words[1] >> (2 * (32 - tail_bases))) as u128
1685 }
1686}
1687
1688#[inline]
1689fn local_vertex_hash<const K: usize>(kmer: Kmer<K>) -> u64 {
1690 let words = kmer.words();
1691 hash_two_u64(words[0], words[1])
1692}
1693
1694#[inline]
1695fn local_u64_hash(key: u64) -> u64 {
1696 fast_u64_hash(key, 0xAAAA_AAAA_5555_5555)
1697}
1698
1699#[derive(Debug)]
1700pub enum LocalSubgraphError {
1701 Bucket(BucketError),
1702 Color(ColorError),
1703 Kmer(KmerError),
1704 InvalidCutoff,
1705 EmptyBucketGroup,
1706 KMismatch { expected: usize, got: usize },
1707 GraphMismatch { expected: usize, got: usize },
1708 MalformedRecord,
1709 MissingVertex,
1710}
1711
1712impl From<BucketError> for LocalSubgraphError {
1713 fn from(value: BucketError) -> Self {
1714 Self::Bucket(value)
1715 }
1716}
1717
1718impl From<KmerError> for LocalSubgraphError {
1719 fn from(value: KmerError) -> Self {
1720 Self::Kmer(value)
1721 }
1722}
1723
1724impl std::fmt::Display for LocalSubgraphError {
1725 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1726 match self {
1727 Self::Bucket(err) => write!(f, "{err}"),
1728 Self::Color(err) => write!(f, "{err}"),
1729 Self::Kmer(err) => write!(f, "{err}"),
1730 Self::InvalidCutoff => write!(f, "local subgraph cutoff must be at least 1"),
1731 Self::EmptyBucketGroup => write!(f, "local subgraph bucket group is empty"),
1732 Self::KMismatch { expected, got } => {
1733 write!(f, "bucket k mismatch: expected {expected}, got {got}")
1734 }
1735 Self::GraphMismatch { expected, got } => {
1736 write!(
1737 f,
1738 "bucket record graph mismatch: expected {expected}, got {got}"
1739 )
1740 }
1741 Self::MalformedRecord => write!(f, "malformed bucket record for local subgraph"),
1742 Self::MissingVertex => write!(f, "local subgraph edge references a missing vertex"),
1743 }
1744 }
1745}
1746
1747impl std::error::Error for LocalSubgraphError {}
1748
1749#[cfg(test)]
1750mod tests {
1751 use super::*;
1752 use crate::GraphInput;
1753 use crate::buckets::BucketEmitter;
1754 use crate::params::BuildParams;
1755 use crate::partition::WeakSuperKmer;
1756 use std::fs;
1757
1758 fn assert_packed_prefix<const K: usize>(seq: &[u8]) {
1759 let mut words = vec![0u64; seq.len().div_ceil(32)];
1760 for (idx, &base) in seq.iter().enumerate() {
1761 words[idx / 32] |= (Base::from_ascii(base).bits() as u64) << (2 * (31 - (idx % 32)));
1762 }
1763 assert_eq!(
1764 packed_prefix_bits::<K>(&words),
1765 Kmer::<K>::from_ascii(&seq[..K]).unwrap().as_u128()
1766 );
1767 }
1768
1769 #[test]
1770 fn packed_prefix_decode_spans_word_boundary() {
1771 let seq = b"ACGTTGCATGTCGCATACGATCGTAGCTAGCTTGCATGACCTAGGCTAACGTTCGATGCATAC";
1772 assert_packed_prefix::<31>(seq);
1773 assert_packed_prefix::<33>(seq);
1774 assert_packed_prefix::<63>(seq);
1775 }
1776
1777 #[test]
1778 fn builds_state_from_record_label() {
1779 let record = BucketRecord {
1780 graph_id: 3,
1781 len: 5,
1782 source_id: Some(1),
1783 left_discontinuous: true,
1784 right_discontinuous: true,
1785 label: b"ACGTT".to_vec(),
1786 };
1787 let mut subgraph = LocalSubgraph::<3> {
1788 graph_id: 3,
1789 colored: true,
1790 cutoff: 1,
1791 vertices: LocalVertexMap::with_capacity(0),
1792 edges: HashSet::with_hasher(FastBuildHasher::default()),
1793 stats: LocalSubgraphStats::default(),
1794 };
1795
1796 subgraph.add_record(&record).unwrap();
1797 subgraph.stats.unique_vertices = subgraph.vertices.len() as u64;
1798 #[cfg(debug_assertions)]
1799 {
1800 subgraph.stats.unique_edges = subgraph.edges.len() as u64;
1801 }
1802
1803 assert_eq!(subgraph.stats.observed_vertices, 3);
1804 assert_eq!(subgraph.stats.observed_edges, 2);
1805 assert_eq!(
1806 subgraph.stats.discontinuity_fronts + subgraph.stats.discontinuity_backs,
1807 2
1808 );
1809 #[cfg(debug_assertions)]
1810 assert_eq!(subgraph.stats.unique_edges, 2);
1811
1812 let acg = Kmer::<3>::from_ascii(b"ACG").unwrap();
1813 let state = subgraph.vertex_state(acg).unwrap();
1814 assert!(state.is_discontinuous(Side::Front));
1815 assert_eq!(state.edge_at(Side::Back, 1), Base::T);
1816 assert_ne!(state.color_hash(), 0);
1817 }
1818
1819 #[test]
1820 fn canonical_cycle_label_normalizes_rotation_and_strand() {
1821 let expected = b"AAACCGTTAAAC".to_vec();
1822
1823 assert_eq!(
1824 canonical_cycle_label::<5>(b"CGTTAAACCGTT".to_vec()),
1825 expected
1826 );
1827 assert_eq!(
1828 canonical_cycle_label::<5>(b"GTTTAACGGTTT".to_vec()),
1829 expected
1830 );
1831 }
1832
1833 #[test]
1834 fn colored_contraction_extracts_source_sets_at_color_transitions() {
1835 let dir = std::env::temp_dir().join(format!(
1836 "cf3-colored-local-{}-{:?}",
1837 std::process::id(),
1838 std::thread::current().id()
1839 ));
1840 let mut params = BuildParams::new(GraphInput::References, "colored-local".into());
1841 params.k = 3;
1842 params.minimizer_len = 2;
1843 params.color = true;
1844 let mut emitter = BucketEmitter::create_in_dir(¶ms, 1, dir.clone()).unwrap();
1845 for source_id in [2, 1, 2] {
1846 emitter
1847 .add(
1848 &WeakSuperKmer {
1849 graph_id: 0,
1850 offset: 0,
1851 len: 5,
1852 source_id: Some(source_id),
1853 left_discontinuous: false,
1854 right_discontinuous: false,
1855 },
1856 b"AACGT",
1857 )
1858 .unwrap();
1859 }
1860 emitter.finish().unwrap();
1861 let (store, entries) = BucketStore::open_dir(&dir).unwrap();
1862 let mut subgraph = LocalSubgraph::<3>::from_manifest_entries(&store, &entries, 1).unwrap();
1863 let unitigs = subgraph.contract_colored(&store, &entries).unwrap();
1864 assert_eq!(unitigs.len(), 1);
1865 assert_eq!(unitigs[0].colors.len(), 1);
1866 assert_eq!(unitigs[0].colors[0].offset, 0);
1867 assert_eq!(unitigs[0].colors[0].sources, [1, 2]);
1868 fs::remove_dir_all(dir).unwrap();
1869 }
1870}