1use crate::circ_catalog::CircCatalogRecord;
8use crate::is_bsj_hg2::{report_scan2_hg_profile, IsBSJHg2};
9use crate::misd::misd;
10use crate::runtime::{
11 emit_perf_line, emit_trace_line, scan2_profile_enabled, should_trace_read, with_trace_hg2_scope,
12};
13use crate::utils::{
14 alignment_short_cs, bam_shard_count, bsj_is_summary_priority, bsj_payload_start,
15 clip_sequence_payload, local_clip_evidence_lines, part_path, reverse_complement,
16 AlignmentRecord,
17};
18use anyhow::{bail, Result};
19use indicatif::{ProgressBar, ProgressStyle};
20use memmap2::Mmap;
21use noodles::sam::{
22 self,
23 alignment::{
24 record::{
25 data::field::{Tag, Value},
26 Data as _, Sequence as _,
27 },
28 Record as _,
29 },
30};
31use rayon::prelude::*;
32use std::borrow::Cow;
33use std::collections::{HashMap, HashSet};
34use std::fmt::Write as FmtWrite;
35use std::fs::File;
36use std::io::{BufRead, BufReader, BufWriter, Seek, Write};
37use std::sync::atomic::{AtomicU64, Ordering};
38use std::sync::mpsc;
39use std::thread;
40use std::time::Duration;
41use std::time::Instant;
42
43const NON_BSJ_SEGMENT_MAPQ_THRES: i32 = 5;
44const NON_BSJ_SEGMENT_MAX_SPAN: i32 = 200000;
45
46pub struct Scan2 {
48 pub min_mapq_uni: i32,
50 pub linear_range_size_min: i32,
52 pub index1: HashMap<String, Vec<CandidateBreakpoint>>,
58 pub index2: HashMap<String, Vec<CandidateBreakpoint>>,
60 pub site_array1: HashMap<String, HashSet<i32>>,
62 pub site_array2: HashMap<String, HashSet<i32>>,
64 pub fsj_map: HashMap<String, i32>,
66 pub mem_limit: u64,
68 pub scan1_ids: HashSet<String>,
70 pub seq_len: i32,
72 pub rescued_reads: usize,
74 pub final_bsj_reads: usize,
76}
77
78#[derive(Default)]
85pub struct Scan2SegmentArtifacts {
86 pub non_bsj_segment_evidence_paths: Vec<String>,
88}
89
90fn xa_payload(raw: &str) -> &str {
96 if raw.is_empty() {
97 "*"
98 } else {
99 raw
100 }
101}
102
103fn raw_xa_from_sam_record(record: &sam::Record) -> Result<String> {
108 let tag = Tag::new(b'X', b'A');
109 let data = record.data();
110 let Some(value) = data.get(&tag).transpose()? else {
111 return Ok(String::new());
112 };
113 let Value::String(raw) = value else {
114 return Ok(String::new());
115 };
116 let Ok(raw) = std::str::from_utf8(raw.as_ref()) else {
117 return Ok(String::new());
118 };
119 Ok(raw.to_string())
120}
121
122fn raw_xa_from_bam_record(record: &noodles::bam::Record) -> Result<String> {
127 let tag = Tag::new(b'X', b'A');
128 let data = record.data();
129 let Some(value) = data.get(&tag).transpose()? else {
130 return Ok(String::new());
131 };
132 let Value::String(raw) = value else {
133 return Ok(String::new());
134 };
135 let Ok(raw) = std::str::from_utf8(raw.as_ref()) else {
136 return Ok(String::new());
137 };
138 Ok(raw.to_string())
139}
140
141fn reference_window_or_n(seq: &str, start: i32, end: i32) -> String {
148 let bytes = seq.as_bytes();
149 let mut out = String::with_capacity((end - start).max(0) as usize);
150 for pos in start..end {
151 if pos >= 0 {
152 let idx = pos as usize;
153 if idx < bytes.len() {
154 out.push(bytes[idx] as char);
155 continue;
156 }
157 }
158 out.push('N');
159 }
160 out
161}
162
163#[derive(Clone)]
164pub struct CandidateBreakpoint {
165 pub site: i32,
167 pub order: usize,
170 pub data: Vec<String>,
173}
174
175type OwnedAlignmentRecord = AlignmentRecord<'static>;
176type OwnedStandMap = HashMap<i32, (char, Cow<'static, str>)>;
177
178struct SamOwnedScan2Group {
179 read_id: String,
180 alignments: Vec<OwnedAlignmentRecord>,
181 all_alignments: Vec<OwnedAlignmentRecord>,
182 stand_map: OwnedStandMap,
183}
184
185#[derive(Clone, Copy)]
186struct NonBsjMsid {
187 kind: i32,
188 clip1: i32,
189 clip2: i32,
190 ref_len: i32,
191}
192
193impl NonBsjMsid {
194 fn invalid(ref_len: i32) -> Self {
195 Self {
196 kind: 0,
197 clip1: 0,
198 clip2: 0,
199 ref_len,
200 }
201 }
202}
203
204fn advise_dontneed(mmap: &Mmap, offset: usize, len: usize) {
210 if len == 0 {
211 return;
212 }
213 let page_size = 4096usize;
214 let aligned_offset = (offset / page_size) * page_size;
215 let aligned_len = ((offset + len + page_size - 1) / page_size) * page_size - aligned_offset;
216 unsafe {
217 let ptr = mmap.as_ptr().add(aligned_offset);
218 libc::madvise(ptr as *mut libc::c_void, aligned_len, libc::MADV_DONTNEED);
219 }
220}
221
222#[derive(Default)]
228pub(crate) struct Scan2Profile {
229 wall_total_ns: AtomicU64,
230 shard_total_ns: AtomicU64,
231 group_process_ns: AtomicU64,
232 validator_call_ns: AtomicU64,
233 display_ns: AtomicU64,
234 segment_sidecar_ns: AtomicU64,
235 non_bsj_sidecar_ns: AtomicU64,
236 write_ns: AtomicU64,
237 merge_ns: AtomicU64,
238 records: AtomicU64,
239 groups: AtomicU64,
240 candidate_checks: AtomicU64,
241 candidate_hits: AtomicU64,
242 segment_sidecar_rows: AtomicU64,
243 segment_sidecar_bytes: AtomicU64,
244 non_bsj_sidecar_rows: AtomicU64,
245 non_bsj_sidecar_bytes: AtomicU64,
246}
247
248struct Scan2DisplayClaims {
255 by_read: HashMap<String, u8>,
256}
257
258impl Scan2DisplayClaims {
259 fn new() -> Self {
261 Self {
262 by_read: HashMap::new(),
263 }
264 }
265
266 fn insert(&mut self, read_id: &str, mate_label: &str) {
268 let bit = match mate_label {
269 "R1" => 0b01,
270 "R2" => 0b10,
271 _ => 0,
272 };
273 if bit != 0 {
274 *self.by_read.entry(read_id.to_string()).or_insert(0) |= bit;
275 }
276 }
277
278 fn contains_mate(&self, read_id: &str, mate_label: &str) -> bool {
280 let bit = match mate_label {
281 "R1" => 0b01,
282 "R2" => 0b10,
283 _ => 0,
284 };
285 bit != 0
286 && self
287 .by_read
288 .get(read_id)
289 .is_some_and(|mask| mask & bit != 0)
290 }
291
292 fn contains_both_mates(&self, read_id: &str) -> bool {
294 self.by_read
295 .get(read_id)
296 .is_some_and(|mask| mask & 0b11 == 0b11)
297 }
298}
299
300impl Scan2Profile {
301 fn enabled_from_env() -> bool {
303 scan2_profile_enabled()
304 }
305
306 fn report(&self) {
308 let wall_total_ns = self.wall_total_ns.load(Ordering::Relaxed);
309 let shard_total_ns = self.shard_total_ns.load(Ordering::Relaxed);
310 let group_process_ns = self.group_process_ns.load(Ordering::Relaxed);
311 let validator_call_ns = self.validator_call_ns.load(Ordering::Relaxed);
312 let display_ns = self.display_ns.load(Ordering::Relaxed);
313 let segment_sidecar_ns = self.segment_sidecar_ns.load(Ordering::Relaxed);
314 let non_bsj_sidecar_ns = self.non_bsj_sidecar_ns.load(Ordering::Relaxed);
315 let write_ns = self.write_ns.load(Ordering::Relaxed);
316 let merge_ns = self.merge_ns.load(Ordering::Relaxed);
317 let records = self.records.load(Ordering::Relaxed);
318 let groups = self.groups.load(Ordering::Relaxed);
319 let candidate_checks = self.candidate_checks.load(Ordering::Relaxed);
320 let candidate_hits = self.candidate_hits.load(Ordering::Relaxed);
321 let segment_sidecar_rows = self.segment_sidecar_rows.load(Ordering::Relaxed);
322 let segment_sidecar_bytes = self.segment_sidecar_bytes.load(Ordering::Relaxed);
323 let non_bsj_sidecar_rows = self.non_bsj_sidecar_rows.load(Ordering::Relaxed);
324 let non_bsj_sidecar_bytes = self.non_bsj_sidecar_bytes.load(Ordering::Relaxed);
325 let accounted_ns = group_process_ns
326 .saturating_add(display_ns)
327 .saturating_add(segment_sidecar_ns)
328 .saturating_add(non_bsj_sidecar_ns)
329 .saturating_add(write_ns);
330 let other_shard_ns = shard_total_ns.saturating_sub(accounted_ns);
331 let pct = |part: u64, whole: u64| -> f64 {
332 if whole == 0 {
333 0.0
334 } else {
335 part as f64 * 100.0 / whole as f64
336 }
337 };
338 emit_perf_line(&format!(
339 "[PROFILE_SCAN2] wall_ms={:.3} shard_work_ms={:.3} merge_ms={:.3} records={} groups={} candidate_checks={} candidate_hits={}",
340 wall_total_ns as f64 / 1_000_000.0,
341 shard_total_ns as f64 / 1_000_000.0,
342 merge_ns as f64 / 1_000_000.0,
343 records,
344 groups,
345 candidate_checks,
346 candidate_hits,
347 ));
348 emit_perf_line(&format!(
349 "[PROFILE_SCAN2] shard_breakdown_ms group_process={:.3} ({:.1}%) validator={:.3} ({:.1}% of group) display={:.3} ({:.1}%) segment_sidecar={:.3} ({:.1}%) non_bsj_sidecar={:.3} ({:.1}%) write={:.3} ({:.1}%) other={:.3} ({:.1}%)",
350 group_process_ns as f64 / 1_000_000.0,
351 pct(group_process_ns, shard_total_ns),
352 validator_call_ns as f64 / 1_000_000.0,
353 pct(validator_call_ns, group_process_ns),
354 display_ns as f64 / 1_000_000.0,
355 pct(display_ns, shard_total_ns),
356 segment_sidecar_ns as f64 / 1_000_000.0,
357 pct(segment_sidecar_ns, shard_total_ns),
358 non_bsj_sidecar_ns as f64 / 1_000_000.0,
359 pct(non_bsj_sidecar_ns, shard_total_ns),
360 write_ns as f64 / 1_000_000.0,
361 pct(write_ns, shard_total_ns),
362 other_shard_ns as f64 / 1_000_000.0,
363 pct(other_shard_ns, shard_total_ns),
364 ));
365 emit_perf_line(&format!(
366 "[PROFILE_SCAN2_SIDECAR] bsj_rows={} bsj_bytes={} non_bsj_rows={} non_bsj_bytes={}",
367 segment_sidecar_rows,
368 segment_sidecar_bytes,
369 non_bsj_sidecar_rows,
370 non_bsj_sidecar_bytes,
371 ));
372 }
373}
374
375impl Scan2 {
376 #[inline]
379 fn java_string_hash(s: &str) -> i32 {
380 let mut h: i32 = 0;
382 for ch in s.chars() {
383 h = h.wrapping_mul(31).wrapping_add(ch as i32);
384 }
385 h
386 }
387
388 #[inline]
390 fn java_hash_spread(h: i32) -> u32 {
391 let x = h as u32;
392 x ^ (x >> 16)
393 }
394
395 #[inline]
397 fn java_hashset_capacity(n: usize) -> usize {
398 let mut cap = 16usize;
401 while n > ((cap as f64) * 0.75f64) as usize {
402 cap = cap.saturating_mul(2);
403 }
404 cap.max(1)
405 }
406
407 pub fn new(min_mapq_uni: i32, linear_range_size_min: i32, seq_len: i32) -> Self {
409 Self {
410 min_mapq_uni,
411 linear_range_size_min,
412 index1: HashMap::new(),
413 index2: HashMap::new(),
414 site_array1: HashMap::new(),
415 site_array2: HashMap::new(),
416 fsj_map: HashMap::new(),
417 mem_limit: 2 * 1024 * 1024 * 1024,
418 scan1_ids: HashSet::new(),
419 seq_len,
420 rescued_reads: 0,
421 final_bsj_reads: 0,
422 }
423 }
424
425 fn lower_bound(list: &[CandidateBreakpoint], key: i32) -> usize {
427 let mut lo = 0usize;
428 let mut hi = list.len();
429 while lo < hi {
430 let mid = (lo + hi) / 2;
431 if list[mid].site < key {
432 lo = mid + 1;
433 } else {
434 hi = mid;
435 }
436 }
437 lo
438 }
439
440 fn bucket_range(list: &[CandidateBreakpoint], bucket: i32, bucket_size: i32) -> (usize, usize) {
445 if bucket < 0 || bucket_size <= 0 {
446 return (0, 0);
447 }
448 let start = bucket.saturating_mul(bucket_size);
449 let end_exclusive = (bucket + 1).saturating_mul(bucket_size);
450 let l = Self::lower_bound(list, start);
451 let r = Self::lower_bound(list, end_exclusive);
452 (l, r)
453 }
454
455 pub fn set_mem_limit(&mut self, limit: u64) {
457 self.mem_limit = limit;
458 }
459
460 pub fn release_working_set(&mut self) {
466 self.index1.clear();
467 self.index1.shrink_to_fit();
468 self.index2.clear();
469 self.index2.shrink_to_fit();
470 self.site_array1.clear();
471 self.site_array1.shrink_to_fit();
472 self.site_array2.clear();
473 self.site_array2.shrink_to_fit();
474 self.scan1_ids.clear();
475 self.scan1_ids.shrink_to_fit();
476 }
477
478 #[inline]
480 fn shard_fsj_path(output_fsj: &str, shard_idx: usize) -> String {
481 part_path(output_fsj, shard_idx)
482 }
483
484 fn write_fsj_shard(path: &str, local_fsj: &HashMap<String, i32>) -> Result<()> {
487 let mut writer = BufWriter::with_capacity(256 * 1024, File::create(path)?);
488 for (key, count) in local_fsj {
489 writeln!(writer, "{}\t{}", key, count)?;
490 }
491 writer.flush()?;
492 Ok(())
493 }
494
495 pub fn build_index(&mut self, bsj1_file: &str) -> Result<()> {
508 let file = File::open(bsj1_file)?;
512 let reader = BufReader::new(file);
513 let mut chr_circ_site_seen: HashMap<String, HashSet<String>> = HashMap::new();
514 let mut chr_circ_site_insertion: HashMap<String, Vec<String>> = HashMap::new();
515 for line_res in reader.lines() {
516 let line = line_res?;
517 let p: Vec<&str> = line.split('\t').collect();
518 if p.len() < 10 || !bsj_is_summary_priority(&p) {
519 continue;
520 }
521 let payload_start = bsj_payload_start(&p);
522 if p.len() <= payload_start + 8 {
523 continue;
524 }
525 self.scan1_ids.insert(p[0].to_string());
526 let chr = p[payload_start + 2].to_string();
527 let site_infor = p[payload_start + 3..].join("\t");
528 let seen = chr_circ_site_seen
529 .entry(chr.clone())
530 .or_insert_with(HashSet::new);
531 if seen.insert(site_infor.clone()) {
532 chr_circ_site_insertion
533 .entry(chr)
534 .or_insert_with(Vec::new)
535 .push(site_infor);
536 }
537 }
538
539 let mut order_counter: usize = 0;
540 for (chr, insertion_list) in chr_circ_site_insertion {
541 let cap = Self::java_hashset_capacity(insertion_list.len());
544 let mut buckets: Vec<Vec<usize>> = vec![Vec::new(); cap];
545 for (idx, s) in insertion_list.iter().enumerate() {
546 let spread = Self::java_hash_spread(Self::java_string_hash(s));
547 let bucket_idx = (spread as usize) & (cap - 1);
548 buckets[bucket_idx].push(idx);
549 }
550
551 for bucket in buckets {
552 for idx in bucket {
553 let arr: Vec<String> = insertion_list[idx]
554 .split('\t')
555 .map(|s| s.to_string())
556 .collect();
557 if arr.len() < 6 {
558 continue;
559 }
560 let site1 = arr[0].parse().unwrap_or(0);
561 let site2 = arr[1].parse().unwrap_or(0);
562 self.fsj_map
563 .entry(format!("{}\t{}\t{}", chr, arr[0], arr[1]))
564 .or_insert(0);
565 self.index1
566 .entry(chr.clone())
567 .or_insert_with(Vec::new)
568 .push(CandidateBreakpoint {
569 site: site1,
570 order: order_counter,
571 data: arr.clone(),
572 });
573 self.index2
574 .entry(chr.clone())
575 .or_insert_with(Vec::new)
576 .push(CandidateBreakpoint {
577 site: site2,
578 order: order_counter,
579 data: arr,
580 });
581 let bucket_size = self.seq_len.max(1);
582 self.site_array1
583 .entry(chr.clone())
584 .or_insert_with(HashSet::new)
585 .insert(site1 / bucket_size);
586 self.site_array2
587 .entry(chr.clone())
588 .or_insert_with(HashSet::new)
589 .insert(site2 / bucket_size);
590 order_counter += 1;
591 }
592 }
593 }
594 for list in self.index1.values_mut() {
595 list.sort_by_key(|x| (x.site, x.order));
596 }
597 for list in self.index2.values_mut() {
598 list.sort_by_key(|x| (x.site, x.order));
599 }
600 Ok(())
601 }
602
603 pub fn add_external_circ_candidates(
613 &mut self,
614 candidates: &[CircCatalogRecord],
615 reference: &HashMap<String, String>,
616 ) -> Result<usize> {
617 let mut seen = HashSet::new();
618 let mut next_order = 0usize;
619 for (chrom, list) in &self.index1 {
620 for candidate in list {
621 next_order = next_order.max(candidate.order + 1);
622 if candidate.data.len() >= 3 {
623 seen.insert(format!(
624 "{}\t{}\t{}\t{}",
625 chrom, candidate.data[0], candidate.data[1], candidate.data[2]
626 ));
627 }
628 }
629 }
630
631 let mut added = 0usize;
632 let bucket_size = self.seq_len.max(1);
633 for candidate in candidates {
634 if candidate.start < 1 || candidate.end < candidate.start {
635 bail!(
636 "invalid external circ candidate {}:{}-{}",
637 candidate.chrom,
638 candidate.start,
639 candidate.end
640 );
641 }
642 let Some(chr_seq) = reference.get(&candidate.chrom) else {
643 bail!(
644 "external circ candidate {}:{}-{} uses chromosome absent from reference",
645 candidate.chrom,
646 candidate.start,
647 candidate.end
648 );
649 };
650 if candidate.end as usize > chr_seq.len() {
651 bail!(
652 "external circ candidate {}:{}-{} exceeds reference length {}",
653 candidate.chrom,
654 candidate.start,
655 candidate.end,
656 chr_seq.len()
657 );
658 }
659 let dedup_key = format!(
660 "{}\t{}\t{}\t{}",
661 candidate.chrom, candidate.start, candidate.end, candidate.strand
662 );
663 if !seen.insert(dedup_key) {
664 continue;
665 }
666 let signal_left =
667 reference_window_or_n(chr_seq, candidate.start - 3, candidate.start - 1);
668 let signal_right = reference_window_or_n(chr_seq, candidate.end, candidate.end + 2);
669 let payload = vec![
670 candidate.start.to_string(),
671 candidate.end.to_string(),
672 candidate.strand.clone(),
673 signal_left,
674 signal_right,
675 "1".to_string(),
676 ];
677 self.fsj_map
678 .entry(format!(
679 "{}\t{}\t{}",
680 candidate.chrom, candidate.start, candidate.end
681 ))
682 .or_insert(0);
683 self.index1
684 .entry(candidate.chrom.clone())
685 .or_insert_with(Vec::new)
686 .push(CandidateBreakpoint {
687 site: candidate.start,
688 order: next_order,
689 data: payload.clone(),
690 });
691 self.index2
692 .entry(candidate.chrom.clone())
693 .or_insert_with(Vec::new)
694 .push(CandidateBreakpoint {
695 site: candidate.end,
696 order: next_order,
697 data: payload,
698 });
699 self.site_array1
700 .entry(candidate.chrom.clone())
701 .or_insert_with(HashSet::new)
702 .insert(candidate.start / bucket_size);
703 self.site_array2
704 .entry(candidate.chrom.clone())
705 .or_insert_with(HashSet::new)
706 .insert(candidate.end / bucket_size);
707 next_order += 1;
708 added += 1;
709 }
710
711 for list in self.index1.values_mut() {
712 list.sort_by_key(|x| (x.site, x.order));
713 }
714 for list in self.index2.values_mut() {
715 list.sort_by_key(|x| (x.site, x.order));
716 }
717 Ok(added)
718 }
719
720 fn build_display_index(&mut self, bsj1_display_file: &str) -> Result<Scan2DisplayClaims> {
726 self.index1.clear();
727 self.index2.clear();
728 self.site_array1.clear();
729 self.site_array2.clear();
730 let file = File::open(bsj1_display_file)?;
731 let reader = BufReader::new(file);
732 let mut claims = Scan2DisplayClaims::new();
733 let mut chr_circ_site_seen: HashMap<String, HashSet<String>> = HashMap::new();
734 let mut chr_circ_site_insertion: HashMap<String, Vec<String>> = HashMap::new();
735 for line_res in reader.lines() {
736 let line = line_res?;
737 let p: Vec<&str> = line.split('\t').collect();
738 if p.len() < 10 {
739 continue;
740 }
741 let payload_start = bsj_payload_start(&p);
742 let mate_label = if payload_start == 3 { p[1] } else { "NA" };
743 if p.len() <= payload_start + 8 {
744 continue;
745 }
746 claims.insert(p[0], mate_label);
747 let chr = p[payload_start + 2].to_string();
748 let site_infor = p[payload_start + 3..].join("\t");
749 let seen = chr_circ_site_seen
750 .entry(chr.clone())
751 .or_insert_with(HashSet::new);
752 if seen.insert(site_infor.clone()) {
753 chr_circ_site_insertion
754 .entry(chr)
755 .or_insert_with(Vec::new)
756 .push(site_infor);
757 }
758 }
759
760 let mut order_counter: usize = 0;
761 for (chr, sites) in chr_circ_site_insertion {
762 let mut list1 = Vec::new();
763 let mut list2 = Vec::new();
764 let mut set1 = HashSet::new();
765 let mut set2 = HashSet::new();
766 for site in sites {
767 let data: Vec<String> = site.split('\t').map(|s| s.to_string()).collect();
768 if data.len() != 6 {
769 continue;
770 }
771 let site1 = data[0].parse::<i32>().unwrap_or(0);
772 let site2 = data[1].parse::<i32>().unwrap_or(0);
773 let num1 = site1 / self.seq_len.max(1);
774 let num2 = site2 / self.seq_len.max(1);
775 set1.insert(num1);
776 set2.insert(num2);
777 list1.push(CandidateBreakpoint {
778 site: site1,
779 order: order_counter,
780 data: data.clone(),
781 });
782 list2.push(CandidateBreakpoint {
783 site: site2,
784 order: order_counter,
785 data,
786 });
787 order_counter += 1;
788 }
789 list1.sort_by_key(|c| (c.site, c.order));
790 list2.sort_by_key(|c| (c.site, c.order));
791 self.index1.insert(chr.clone(), list1);
792 self.index2.insert(chr.clone(), list2);
793 self.site_array1.insert(chr.clone(), set1);
794 self.site_array2.insert(chr, set2);
795 }
796 Ok(claims)
797 }
798
799 fn collect_fsj_keys_in_range(
812 &self,
813 chr: &str,
814 start_tem: i32,
815 end_tem: i32,
816 style: i32,
817 out: &mut HashSet<String>,
818 ) {
819 let bucket_size = self.seq_len.max(1);
820 let num1 = start_tem / bucket_size;
821 let num2 = end_tem / bucket_size;
822 if let Some(list1) = self.index1.get(chr) {
823 if self.site_array1.get(chr).is_some_and(|s| s.contains(&num1)) {
824 let (l1, r1) = Self::bucket_range(list1, num1, bucket_size);
825 for idx in (l1..r1).rev() {
826 let cand = &list1[idx];
827 if cand.site >= start_tem {
828 if cand.site <= end_tem || style == 10 || style == 1 {
829 out.insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
830 }
831 } else {
832 break;
833 }
834 }
835 }
836 if num2 != num1 && self.site_array1.get(chr).is_some_and(|s| s.contains(&num2)) {
837 let (l2, r2) = Self::bucket_range(list1, num2, bucket_size);
838 for idx in l2..r2 {
839 let cand = &list1[idx];
840 if cand.site <= end_tem {
841 if cand.site >= start_tem || style == 10 || style == -1 {
842 out.insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
843 }
844 } else {
845 break;
846 }
847 }
848 }
849 }
850 if let Some(list2) = self.index2.get(chr) {
851 if self.site_array2.get(chr).is_some_and(|s| s.contains(&num1)) {
852 let (l1, r1) = Self::bucket_range(list2, num1, bucket_size);
853 for idx in (l1..r1).rev() {
854 let cand = &list2[idx];
855 if cand.site >= start_tem {
856 if cand.site <= end_tem || style == 10 || style == 1 {
857 out.insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
858 }
859 } else {
860 break;
861 }
862 }
863 }
864 if num2 != num1 && self.site_array2.get(chr).is_some_and(|s| s.contains(&num2)) {
865 let (l2, r2) = Self::bucket_range(list2, num2, bucket_size);
866 for idx in l2..r2 {
867 let cand = &list2[idx];
868 if cand.site <= end_tem {
869 if cand.site >= start_tem || style == 10 || style == -1 {
870 out.insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
871 }
872 } else {
873 break;
874 }
875 }
876 }
877 }
878 }
879
880 pub fn run(
882 &mut self,
883 sam_file: &str,
884 output_bsj2: &str,
885 output_fsj: &str,
886 chr_tcga_map: &HashMap<String, String>,
887 ) -> Result<()> {
888 self.run_with_display(sam_file, output_bsj2, output_fsj, None, None, chr_tcga_map)
889 }
890
891 pub fn run_with_display(
894 &mut self,
895 sam_file: &str,
896 output_bsj2: &str,
897 output_fsj: &str,
898 scan1_display_path: Option<&str>,
899 display_output_path: Option<&str>,
900 chr_tcga_map: &HashMap<String, String>,
901 ) -> Result<()> {
902 self.run_with_display_and_segments(
903 sam_file,
904 output_bsj2,
905 output_fsj,
906 scan1_display_path,
907 display_output_path,
908 None,
909 None,
910 chr_tcga_map,
911 )
912 .map(|_| ())
913 }
914
915 pub fn run_with_display_and_segments(
922 &mut self,
923 sam_file: &str,
924 output_bsj2: &str,
925 output_fsj: &str,
926 scan1_display_path: Option<&str>,
927 display_output_path: Option<&str>,
928 segments_output_path: Option<&str>,
929 non_bsj_segments_output_path: Option<&str>,
930 chr_tcga_map: &HashMap<String, String>,
931 ) -> Result<Scan2SegmentArtifacts> {
932 use crate::sam_bam::{detect_format, InputFormat};
933 let display_scan2 = if let Some(scan1_path) = scan1_display_path {
934 let mut helper =
935 Scan2::new(self.min_mapq_uni, self.linear_range_size_min, self.seq_len);
936 let claims = helper.build_display_index(scan1_path)?;
937 Some((helper, claims))
938 } else {
939 None
940 };
941 let format = detect_format(sam_file)?;
942 match format {
943 InputFormat::Sam => self.run_sam_with_display(
944 sam_file,
945 output_bsj2,
946 output_fsj,
947 display_output_path,
948 segments_output_path,
949 non_bsj_segments_output_path,
950 display_scan2.as_ref(),
951 chr_tcga_map,
952 ),
953 InputFormat::Bam => self.run_bam_with_display(
954 sam_file,
955 output_bsj2,
956 output_fsj,
957 display_output_path,
958 segments_output_path,
959 non_bsj_segments_output_path,
960 display_scan2.as_ref(),
961 chr_tcga_map,
962 ),
963 }
964 }
965
966 pub fn run_sam(
972 &mut self,
973 sam_file: &str,
974 output_bsj2: &str,
975 output_fsj: &str,
976 chr_tcga_map: &HashMap<String, String>,
977 ) -> Result<()> {
978 self.run_sam_with_display(
979 sam_file,
980 output_bsj2,
981 output_fsj,
982 None,
983 None,
984 None,
985 None,
986 chr_tcga_map,
987 )
988 .map(|_| ())
989 }
990
991 fn run_sam_with_display(
992 &mut self,
993 sam_file: &str,
994 output_bsj2: &str,
995 output_fsj: &str,
996 display_output_path: Option<&str>,
997 segments_output_path: Option<&str>,
998 non_bsj_segments_output_path: Option<&str>,
999 display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
1000 chr_tcga_map: &HashMap<String, String>,
1001 ) -> Result<Scan2SegmentArtifacts> {
1002 let run_started = Instant::now();
1003 let profile = if Scan2Profile::enabled_from_env() {
1004 Some(Scan2Profile::default())
1005 } else {
1006 None
1007 };
1008 let profile_ref = profile.as_ref();
1009 let file_size = std::fs::metadata(sam_file)?.len();
1010
1011 let pb = ProgressBar::new(file_size as u64);
1012 pb.set_style(ProgressStyle::default_bar().template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {percent:>3}% ({eta}) {msg}")?.progress_chars("#>-"));
1013 pb.set_message("");
1014 let shard_out = part_path(output_bsj2, 0);
1015 let display_shard_out = display_output_path.map(|path| part_path(path, 0));
1016 let segments_shard_out = segments_output_path.map(|path| part_path(path, 0));
1017 let non_bsj_segments_shard_out =
1018 non_bsj_segments_output_path.map(|path| part_path(path, 0));
1019 let fsj_out = Self::shard_fsj_path(output_fsj, 0);
1020 self.process_sam_file_to_file(
1021 sam_file,
1022 chr_tcga_map,
1023 &pb,
1024 &shard_out,
1025 &fsj_out,
1026 display_scan2,
1027 display_shard_out.as_deref(),
1028 segments_shard_out.as_deref(),
1029 non_bsj_segments_shard_out.as_deref(),
1030 profile_ref,
1031 )?;
1032
1033 pb.set_style(ProgressStyle::default_bar().template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {msg}")?.progress_chars("#>-"));
1034 pb.finish_with_message("Completed");
1035 let merge_started = Instant::now();
1036 self.merge_shards_and_fsj(output_bsj2, output_fsj, 1)?;
1037 if let Some(profile) = profile_ref {
1038 profile
1039 .merge_ns
1040 .fetch_add(merge_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
1041 profile
1042 .wall_total_ns
1043 .fetch_add(run_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
1044 profile.report();
1045 report_scan2_hg_profile();
1046 }
1047 if let Some(path) = display_output_path {
1048 self.merge_display_shards(path, 1)?;
1049 }
1050 if let Some(path) = segments_output_path {
1051 self.merge_segment_shards(path, 1)?;
1052 }
1053 let non_bsj_segment_evidence_paths = non_bsj_segments_output_path
1054 .map(|path| vec![part_path(path, 0)])
1055 .unwrap_or_default();
1056 Ok(Scan2SegmentArtifacts {
1057 non_bsj_segment_evidence_paths,
1058 })
1059 }
1060
1061 fn merge_shards_and_fsj(
1068 &mut self,
1069 output_bsj2: &str,
1070 output_fsj: &str,
1071 num_threads: usize,
1072 ) -> Result<()> {
1073 let mut writer = BufWriter::with_capacity(1024 * 1024, File::create(output_bsj2)?);
1074 let mut rescued_ids = HashSet::new();
1075 for i in 0..num_threads {
1076 let shard_path = part_path(output_bsj2, i);
1077 if let Ok(shard_file) = File::open(&shard_path) {
1078 let mut shard_reader = BufReader::new(shard_file);
1079 let mut line = String::new();
1080 while shard_reader.read_line(&mut line)? != 0 {
1081 let trimmed = line.trim_end();
1082 if !trimmed.is_empty() {
1083 let parts: Vec<&str> = trimmed.split('\t').collect();
1084 if !parts.is_empty() && bsj_is_summary_priority(&parts) {
1085 rescued_ids.insert(parts[0].to_string());
1086 }
1087 writeln!(writer, "{}", trimmed)?;
1088 }
1089 line.clear();
1090 }
1091 }
1092 let _ = std::fs::remove_file(shard_path);
1093 }
1094 for i in 0..num_threads {
1095 let fsj_path = Self::shard_fsj_path(output_fsj, i);
1096 if let Ok(fsj_file) = File::open(&fsj_path) {
1097 let mut fsj_reader = BufReader::new(fsj_file);
1098 let mut line = String::new();
1099 while fsj_reader.read_line(&mut line)? != 0 {
1100 let trimmed = line.trim_end();
1101 if !trimmed.is_empty() {
1102 let mut parts = trimmed.rsplitn(2, '\t');
1103 let count = parts
1104 .next()
1105 .and_then(|s| s.parse::<i32>().ok())
1106 .unwrap_or(0);
1107 if let Some(key) = parts.next() {
1108 *self.fsj_map.entry(key.to_string()).or_insert(0) += count;
1109 }
1110 }
1111 line.clear();
1112 }
1113 }
1114 let _ = std::fs::remove_file(fsj_path);
1115 }
1116 writer.flush()?;
1117 self.rescued_reads = rescued_ids.len();
1118 self.final_bsj_reads = self.scan1_ids.len() + self.rescued_reads;
1119 Ok(())
1120 }
1121
1122 pub fn run_bam(
1124 &mut self,
1125 bam_file: &str,
1126 output_bsj2: &str,
1127 output_fsj: &str,
1128 chr_tcga_map: &HashMap<String, String>,
1129 ) -> Result<()> {
1130 self.run_bam_with_display(
1131 bam_file,
1132 output_bsj2,
1133 output_fsj,
1134 None,
1135 None,
1136 None,
1137 None,
1138 chr_tcga_map,
1139 )
1140 .map(|_| ())
1141 }
1142
1143 fn run_bam_with_display(
1144 &mut self,
1145 bam_file: &str,
1146 output_bsj2: &str,
1147 output_fsj: &str,
1148 display_output_path: Option<&str>,
1149 segments_output_path: Option<&str>,
1150 non_bsj_segments_output_path: Option<&str>,
1151 display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
1152 chr_tcga_map: &HashMap<String, String>,
1153 ) -> Result<Scan2SegmentArtifacts> {
1154 use noodles::bam;
1155 let run_started = Instant::now();
1156 let profile = if Scan2Profile::enabled_from_env() {
1157 Some(Scan2Profile::default())
1158 } else {
1159 None
1160 };
1161 let profile_ref = profile.as_ref();
1162 let file = File::open(bam_file)?;
1163 let file_size = std::fs::metadata(bam_file)?.len();
1164 let mmap = unsafe { Mmap::map(&file)? };
1165 let num_threads = bam_shard_count(mmap.len(), rayon::current_num_threads());
1166 let shard_size = mmap.len() / num_threads;
1167 let mut reader = bam::io::Reader::new(file);
1168 let header = reader.read_header()?;
1169
1170 let pb = ProgressBar::new(file_size);
1171 pb.set_style(
1172 ProgressStyle::default_bar()
1173 .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {percent:>3}% ({eta}) {msg}")?
1174 .progress_chars("#>-"),
1175 );
1176 pb.set_message("");
1177 pb.enable_steady_tick(Duration::from_millis(120));
1178
1179 unsafe {
1180 libc::madvise(
1181 mmap.as_ptr() as *mut libc::c_void,
1182 mmap.len(),
1183 libc::MADV_SEQUENTIAL,
1184 );
1185 }
1186
1187 let header_ref = &header;
1188 (0..num_threads).into_par_iter().try_for_each(|i| {
1189 let start = i * shard_size;
1190 let end = if i == num_threads - 1 {
1191 mmap.len()
1192 } else {
1193 (i + 1) * shard_size
1194 };
1195 let shard_out = part_path(output_bsj2, i);
1196 let display_shard_out = display_output_path.map(|path| part_path(path, i));
1197 let segments_shard_out = segments_output_path.map(|path| part_path(path, i));
1198 let non_bsj_segments_shard_out =
1199 non_bsj_segments_output_path.map(|path| part_path(path, i));
1200 let fsj_out = Self::shard_fsj_path(output_fsj, i);
1201 self.process_bam_shard_to_file(
1202 &mmap,
1203 start,
1204 end,
1205 header_ref,
1206 chr_tcga_map,
1207 &pb,
1208 &shard_out,
1209 &fsj_out,
1210 display_scan2,
1211 display_shard_out.as_deref(),
1212 segments_shard_out.as_deref(),
1213 non_bsj_segments_shard_out.as_deref(),
1214 profile_ref,
1215 )
1216 })?;
1217
1218 pb.set_position(file_size);
1219 pb.set_style(
1220 ProgressStyle::default_bar()
1221 .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {msg}")?
1222 .progress_chars("#>-"),
1223 );
1224 pb.finish_with_message("Completed");
1225 let merge_started = Instant::now();
1226 self.merge_shards_and_fsj(output_bsj2, output_fsj, num_threads)?;
1227 if let Some(profile) = profile_ref {
1228 profile
1229 .merge_ns
1230 .fetch_add(merge_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
1231 profile
1232 .wall_total_ns
1233 .fetch_add(run_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
1234 profile.report();
1235 report_scan2_hg_profile();
1236 }
1237 if let Some(path) = display_output_path {
1238 self.merge_display_shards(path, num_threads)?;
1239 }
1240 if let Some(path) = segments_output_path {
1241 self.merge_segment_shards(path, num_threads)?;
1242 }
1243 let non_bsj_segment_evidence_paths = non_bsj_segments_output_path
1244 .map(|path| {
1245 (0..num_threads)
1246 .map(|idx| part_path(path, idx))
1247 .collect::<Vec<_>>()
1248 })
1249 .unwrap_or_default();
1250 Ok(Scan2SegmentArtifacts {
1251 non_bsj_segment_evidence_paths,
1252 })
1253 }
1254
1255 fn merge_display_shards(&self, output_path: &str, num_threads: usize) -> Result<()> {
1257 let mut writer = BufWriter::with_capacity(1024 * 1024, File::create(output_path)?);
1258 for i in 0..num_threads {
1259 let shard_path = part_path(output_path, i);
1260 if let Ok(shard_file) = File::open(&shard_path) {
1261 let mut reader = BufReader::new(shard_file);
1262 let mut line = String::new();
1263 loop {
1264 line.clear();
1265 if reader.read_line(&mut line)? == 0 {
1266 break;
1267 }
1268 writer.write_all(line.as_bytes())?;
1269 }
1270 }
1271 let _ = std::fs::remove_file(&shard_path);
1272 }
1273 writer.flush()?;
1274 Ok(())
1275 }
1276
1277 fn merge_segment_shards(&self, output_path: &str, num_threads: usize) -> Result<()> {
1283 let mut writer = BufWriter::with_capacity(1024 * 1024, File::create(output_path)?);
1284 for i in 0..num_threads {
1285 let shard_path = part_path(output_path, i);
1286 if let Ok(shard_file) = File::open(&shard_path) {
1287 let mut reader = BufReader::new(shard_file);
1288 let mut line = String::new();
1289 while reader.read_line(&mut line)? != 0 {
1290 writer.write_all(line.as_bytes())?;
1291 line.clear();
1292 }
1293 }
1294 let _ = std::fs::remove_file(&shard_path);
1295 }
1296 writer.flush()?;
1297 Ok(())
1298 }
1299
1300 fn segment_evidence_lines<'a>(
1307 read_id: &str,
1308 stage: &str,
1309 alignments: &[AlignmentRecord<'a>],
1310 reference: &HashMap<String, String>,
1311 ) -> Vec<String> {
1312 let mut rows = Vec::new();
1313 for aln in alignments {
1314 if aln.chrom.as_ref() == "*" || aln.cigar.as_ref() == "*" {
1315 continue;
1316 }
1317 let mate = if aln.flag & 0x40 != 0 { "R1" } else { "R2" };
1318 let clips = clip_sequence_payload(aln.cigar.as_ref(), aln.seq.as_ref());
1319 let cs = alignment_short_cs(
1320 aln.cigar.as_ref(),
1321 aln.seq.as_ref(),
1322 aln.chrom.as_ref(),
1323 aln.pos,
1324 reference,
1325 );
1326 rows.push(format!(
1327 "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
1328 read_id,
1329 stage,
1330 mate,
1331 aln.flag,
1332 aln.chrom,
1333 aln.pos,
1334 aln.mapq,
1335 aln.cigar,
1336 aln.seq.len(),
1337 clips,
1338 cs,
1339 xa_payload(aln.xa.as_ref())
1340 ));
1341 }
1342 rows
1343 }
1344
1345 fn non_bsj_segment_evidence_lines<'a>(
1356 read_id: &str,
1357 alignments: &[AlignmentRecord<'a>],
1358 reference: &HashMap<String, String>,
1359 ) -> Vec<String> {
1360 if !Self::may_support_non_bsj_segments(alignments) {
1361 return Vec::new();
1362 }
1363 let mut records = Vec::new();
1364 for aln in alignments {
1365 if aln.chrom.as_ref() == "*" || aln.cigar.as_ref() == "*" {
1366 continue;
1367 }
1368 let clips = clip_sequence_payload(aln.cigar.as_ref(), aln.seq.as_ref());
1369 let cs = alignment_short_cs(
1370 aln.cigar.as_ref(),
1371 aln.seq.as_ref(),
1372 aln.chrom.as_ref(),
1373 aln.pos,
1374 reference,
1375 );
1376 records.push(format!(
1377 "{}|{}|{}|{}|{}|{}|{}",
1378 aln.flag, aln.chrom, aln.pos, aln.mapq, aln.cigar, clips, cs
1379 ));
1380 }
1381 if records.is_empty() {
1382 Vec::new()
1383 } else {
1384 vec![format!("{}\t{}\t{}", read_id, "N", records.join(";"))]
1385 }
1386 }
1387
1388 fn may_support_non_bsj_segments<'a>(alignments: &[AlignmentRecord<'a>]) -> bool {
1390 Self::may_support_outward_segments(alignments)
1391 || Self::may_support_backward_segments(alignments)
1392 }
1393
1394 fn may_support_outward_segments<'a>(alignments: &[AlignmentRecord<'a>]) -> bool {
1396 let mut r1: Option<&AlignmentRecord<'_>> = None;
1397 let mut r2: Option<&AlignmentRecord<'_>> = None;
1398 for aln in alignments {
1399 if aln.flag & 0x4 != 0 || aln.flag & 0x100 != 0 || aln.flag & 0x800 != 0 {
1400 continue;
1401 }
1402 match aln.flag & 0x40 != 0 {
1403 true if r1.is_none() => r1 = Some(aln),
1404 true => return false,
1405 false if r2.is_none() => r2 = Some(aln),
1406 false => return false,
1407 }
1408 }
1409 let (Some(r1), Some(r2)) = (r1, r2) else {
1410 return false;
1411 };
1412 if r1.chrom != r2.chrom
1413 || r1.chrom.as_ref() == "*"
1414 || r1.mapq < NON_BSJ_SEGMENT_MAPQ_THRES
1415 || r2.mapq < NON_BSJ_SEGMENT_MAPQ_THRES
1416 {
1417 return false;
1418 }
1419 let Some(r1_span) = Self::alignment_ref_span(r1) else {
1420 return false;
1421 };
1422 let Some(r2_span) = Self::alignment_ref_span(r2) else {
1423 return false;
1424 };
1425 let r1_reverse = r1.flag & 0x10 != 0;
1426 let r2_reverse = r2.flag & 0x10 != 0;
1427 if r1_reverse == r2_reverse {
1428 return false;
1429 }
1430 let (reverse_span, forward_span) = if r1_reverse {
1431 (r1_span, r2_span)
1432 } else {
1433 (r2_span, r1_span)
1434 };
1435 reverse_span.0 < forward_span.0
1436 && forward_span.0 <= reverse_span.1
1437 && reverse_span.1 < forward_span.1
1438 && forward_span.1 - reverse_span.0 + 1 <= NON_BSJ_SEGMENT_MAX_SPAN
1439 }
1440
1441 fn may_support_backward_segments<'a>(alignments: &[AlignmentRecord<'a>]) -> bool {
1450 for i in 0..alignments.len().saturating_sub(1) {
1451 let left = &alignments[i];
1452 if left.chrom.as_ref() == "*" || left.cigar.as_ref() == "*" {
1453 continue;
1454 }
1455 for right in &alignments[i + 1..] {
1456 if left.chrom != right.chrom
1457 || right.cigar.as_ref() == "*"
1458 || (left.flag & 0x10 != 0) != (right.flag & 0x10 != 0)
1459 {
1460 continue;
1461 }
1462 let read_len = left.seq.len().max(right.seq.len()) as i32;
1463 if read_len <= 0 {
1464 continue;
1465 }
1466 let left_msid = Self::non_bsj_msid(left.cigar.as_ref(), read_len);
1467 let right_msid = Self::non_bsj_msid(right.cigar.as_ref(), read_len);
1468 if Self::non_bsj_backward_pair_possible(
1469 left, right, left_msid, right_msid, read_len,
1470 ) || Self::non_bsj_backward_pair_possible(
1471 right, left, right_msid, left_msid, read_len,
1472 ) {
1473 return true;
1474 }
1475 }
1476 }
1477 false
1478 }
1479
1480 fn non_bsj_backward_pair_possible<'a>(
1482 left: &AlignmentRecord<'a>,
1483 right: &AlignmentRecord<'a>,
1484 left_msid: NonBsjMsid,
1485 right_msid: NonBsjMsid,
1486 read_len: i32,
1487 ) -> bool {
1488 let product = left_msid.kind * right_msid.kind;
1489 if product == -1 {
1490 let cir_scale = left_msid.kind * (left.pos + left_msid.clip2)
1491 + right_msid.kind * (right.pos + right_msid.clip2);
1492 return (left_msid.clip1 - right_msid.clip1).abs() <= 6 && cir_scale < 0;
1493 }
1494 if product.abs() != 10 {
1495 return false;
1496 }
1497 let (mx, my, rx, ry) = if left_msid.kind <= right_msid.kind {
1498 (left_msid, right_msid, left, right)
1499 } else {
1500 (right_msid, left_msid, right, left)
1501 };
1502 if mx.kind == -1 {
1503 let cir_scale = ry.pos + my.ref_len - 1 - rx.pos;
1504 (read_len - my.clip2 - mx.clip1).abs() <= 6 && cir_scale < 0
1505 } else {
1506 let cir_scale = rx.pos + mx.ref_len - 1 - ry.pos;
1507 (mx.clip1 - my.clip1).abs() <= 6 && cir_scale < 0
1508 }
1509 }
1510
1511 fn non_bsj_msid(cigar: &str, read_len: i32) -> NonBsjMsid {
1513 if cigar == "*" || cigar.is_empty() {
1514 return NonBsjMsid::invalid(-1);
1515 }
1516 let mut counts = Vec::new();
1517 let mut styles = Vec::new();
1518 let mut count = 0i32;
1519 let mut has_count = false;
1520 for ch in cigar.chars() {
1521 if ch.is_ascii_digit() {
1522 let Some(next) = count
1523 .checked_mul(10)
1524 .and_then(|value| value.checked_add(ch.to_digit(10).unwrap_or(0) as i32))
1525 else {
1526 return NonBsjMsid::invalid(-2);
1527 };
1528 count = next;
1529 has_count = true;
1530 } else {
1531 if !has_count {
1532 return NonBsjMsid::invalid(-2);
1533 }
1534 counts.push(count);
1535 styles.push(if ch == 'H' { 'S' } else { ch });
1536 count = 0;
1537 has_count = false;
1538 }
1539 }
1540 if has_count || styles.is_empty() {
1541 return NonBsjMsid::invalid(-2);
1542 }
1543
1544 if counts.len() == 1 {
1545 return if styles[0] == 'M' && counts[0] == read_len {
1546 NonBsjMsid {
1547 kind: 0,
1548 clip1: 0,
1549 clip2: 0,
1550 ref_len: read_len,
1551 }
1552 } else {
1553 NonBsjMsid::invalid(-1)
1554 };
1555 }
1556
1557 match counts.len() {
1558 2 => match (styles[0], styles[1]) {
1559 ('M', 'S') => NonBsjMsid {
1560 kind: 1,
1561 clip1: counts[0],
1562 clip2: counts[0] - 1,
1563 ref_len: counts[0],
1564 },
1565 ('S', 'M') => NonBsjMsid {
1566 kind: -1,
1567 clip1: counts[0],
1568 clip2: 0,
1569 ref_len: counts[1],
1570 },
1571 _ => NonBsjMsid::invalid(-2),
1572 },
1573 3 => match (styles[0], styles[1], styles[2]) {
1574 ('S', _, 'S') => NonBsjMsid {
1575 kind: 10,
1576 clip1: counts[0],
1577 clip2: counts[2],
1578 ref_len: counts[1],
1579 },
1580 ('M', 'D', 'M') => NonBsjMsid {
1581 kind: 0,
1582 clip1: 0,
1583 clip2: 0,
1584 ref_len: read_len + counts[1],
1585 },
1586 ('M', 'I', 'M') => NonBsjMsid {
1587 kind: 0,
1588 clip1: 0,
1589 clip2: 0,
1590 ref_len: read_len - counts[1],
1591 },
1592 _ => NonBsjMsid::invalid(-2),
1593 },
1594 _ if styles[0] == 'M' && *styles.last().unwrap() == 'S' => {
1595 let (m_sum, d_sum) =
1596 Self::non_bsj_sum_md(&styles[..styles.len() - 1], &counts[..counts.len() - 1]);
1597 NonBsjMsid {
1598 kind: 1,
1599 clip1: read_len - counts[counts.len() - 1],
1600 clip2: m_sum + d_sum - 1,
1601 ref_len: m_sum + d_sum,
1602 }
1603 }
1604 _ if styles[0] == 'S' && *styles.last().unwrap() == 'M' => {
1605 let (m_sum, d_sum) = Self::non_bsj_sum_md(&styles[1..], &counts[1..]);
1606 NonBsjMsid {
1607 kind: -1,
1608 clip1: counts[0],
1609 clip2: 0,
1610 ref_len: m_sum + d_sum,
1611 }
1612 }
1613 _ if styles[0] == 'M' && *styles.last().unwrap() == 'M' => {
1614 let (m_sum, d_sum) = Self::non_bsj_sum_md(&styles, &counts);
1615 NonBsjMsid {
1616 kind: 0,
1617 clip1: 0,
1618 clip2: 0,
1619 ref_len: m_sum + d_sum,
1620 }
1621 }
1622 _ if styles[0] == 'S' && *styles.last().unwrap() == 'S' => {
1623 let (m_sum, d_sum) = Self::non_bsj_sum_md(
1624 &styles[1..styles.len() - 1],
1625 &counts[1..counts.len() - 1],
1626 );
1627 NonBsjMsid {
1628 kind: 10,
1629 clip1: counts[0],
1630 clip2: counts[counts.len() - 1],
1631 ref_len: m_sum + d_sum,
1632 }
1633 }
1634 _ => NonBsjMsid::invalid(-2),
1635 }
1636 }
1637
1638 fn non_bsj_sum_md(styles: &[char], counts: &[i32]) -> (i32, i32) {
1640 let mut match_sum = 0;
1641 let mut deletion_sum = 0;
1642 for (style, count) in styles.iter().zip(counts.iter()) {
1643 match style {
1644 'M' | '=' | 'X' => match_sum += count,
1645 'D' => deletion_sum += count,
1646 _ => {}
1647 }
1648 }
1649 (match_sum, deletion_sum)
1650 }
1651
1652 fn alignment_ref_span(aln: &AlignmentRecord<'_>) -> Option<(i32, i32)> {
1654 let mut ref_len = 0i32;
1655 let mut count = 0i32;
1656 let mut has_count = false;
1657 for op in aln.cigar.as_ref().chars() {
1658 if op.is_ascii_digit() {
1659 count = count
1660 .checked_mul(10)?
1661 .checked_add(op.to_digit(10)? as i32)?;
1662 has_count = true;
1663 continue;
1664 }
1665 if !has_count {
1666 return None;
1667 }
1668 if matches!(op, 'M' | 'D' | 'N' | '=' | 'X') {
1669 ref_len = ref_len.checked_add(count)?;
1670 }
1671 count = 0;
1672 has_count = false;
1673 }
1674 if has_count || ref_len <= 0 {
1675 return None;
1676 }
1677 Some((aln.pos, aln.pos + ref_len - 1))
1678 }
1679
1680 fn segment_evidence_lines_with_local<'a>(
1682 read_id: &str,
1683 stage: &str,
1684 local_stage: &str,
1685 alignments: &[AlignmentRecord<'a>],
1686 bsj_lines: &[String],
1687 reference: &HashMap<String, String>,
1688 ) -> Vec<String> {
1689 let mut rows = Self::segment_evidence_lines(read_id, stage, alignments, reference);
1690 let refs: Vec<&AlignmentRecord<'_>> = alignments.iter().collect();
1691 rows.extend(local_clip_evidence_lines(
1692 read_id,
1693 local_stage,
1694 &refs,
1695 bsj_lines,
1696 reference,
1697 10,
1698 ));
1699 rows
1700 }
1701
1702 fn scan2_priority_key(line: &str) -> Option<String> {
1707 let mut parts = line.splitn(4, '\t');
1708 let read_id = parts.next()?;
1709 let mate_label = parts.next()?;
1710 let _priority = parts.next()?;
1711 let payload = parts.next()?;
1712 Some(format!("{read_id}\t{mate_label}\t{payload}"))
1713 }
1714
1715 fn scan2_display_key_and_priority_line(line: &str, priority: &str) -> Option<(String, String)> {
1718 let mut parts = line.splitn(3, '\t');
1719 let read_id = parts.next()?;
1720 let mate_label = parts.next()?;
1721 let payload = parts.next()?;
1722 Some((
1723 format!("{read_id}\t{mate_label}\t{payload}"),
1724 format!("{read_id}\t{mate_label}\t{priority}\t{payload}"),
1725 ))
1726 }
1727
1728 fn process_sam_file_to_file(
1731 &self,
1732 sam_file: &str,
1733 chr_tcga_map: &HashMap<String, String>,
1734 pb: &ProgressBar,
1735 out_path: &str,
1736 fsj_path: &str,
1737 display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
1738 display_out_path: Option<&str>,
1739 segments_out_path: Option<&str>,
1740 non_bsj_segments_out_path: Option<&str>,
1741 profile: Option<&Scan2Profile>,
1742 ) -> Result<()> {
1743 let mut writer = BufWriter::with_capacity(256 * 1024, File::create(out_path)?);
1744 let mut display_writer = if let Some(path) = display_out_path {
1745 Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
1746 } else {
1747 None
1748 };
1749 let mut segments_writer = if let Some(path) = segments_out_path {
1750 Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
1751 } else {
1752 None
1753 };
1754 let mut non_bsj_segments_writer = if let Some(path) = non_bsj_segments_out_path {
1755 Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
1756 } else {
1757 None
1758 };
1759 let batch_size = (rayon::current_num_threads().max(1) * 256).max(1024);
1760 let (tx, rx) =
1761 mpsc::sync_channel::<Vec<SamOwnedScan2Group>>(rayon::current_num_threads().max(2));
1762 let pb_clone = pb.clone();
1763
1764 let local_fsj = thread::scope(|scope| -> Result<HashMap<String, i32>> {
1765 let producer = scope.spawn(|| {
1766 self.stream_sam_group_batches_for_scan2(
1767 sam_file, &pb_clone, tx, batch_size, profile,
1768 )
1769 });
1770 let mut merged_fsj = HashMap::new();
1771 for mut batch in rx {
1772 self.process_sam_group_batch_for_scan2(
1773 &mut writer,
1774 display_writer.as_mut(),
1775 segments_writer.as_mut(),
1776 non_bsj_segments_writer.as_mut(),
1777 &mut batch,
1778 &mut merged_fsj,
1779 chr_tcga_map,
1780 display_scan2,
1781 profile,
1782 )?;
1783 }
1784 producer
1785 .join()
1786 .map_err(|_| anyhow::anyhow!("SAM Scan2 group producer thread panicked"))??;
1787 Ok(merged_fsj)
1788 })?;
1789
1790 writer.flush()?;
1791 if let Some(writer) = display_writer.as_mut() {
1792 writer.flush()?;
1793 }
1794 if let Some(writer) = segments_writer.as_mut() {
1795 writer.flush()?;
1796 }
1797 if let Some(writer) = non_bsj_segments_writer.as_mut() {
1798 writer.flush()?;
1799 }
1800 Self::write_fsj_shard(fsj_path, &local_fsj)
1801 }
1802
1803 fn stream_sam_group_batches_for_scan2(
1805 &self,
1806 sam_file: &str,
1807 pb: &ProgressBar,
1808 tx: mpsc::SyncSender<Vec<SamOwnedScan2Group>>,
1809 batch_size: usize,
1810 profile: Option<&Scan2Profile>,
1811 ) -> Result<()> {
1812 let sam_reader = BufReader::with_capacity(1024 * 1024, File::open(sam_file)?);
1813 let mut reader = sam::io::Reader::new(sam_reader);
1814 let header = reader.read_header()?;
1815 let mut record = sam::Record::default();
1816 let mut current_id: Vec<u8> = Vec::new();
1817 let mut alignments: Vec<OwnedAlignmentRecord> = Vec::with_capacity(16);
1818 let mut all_alignments: Vec<OwnedAlignmentRecord> = Vec::with_capacity(16);
1819 let mut stand_map: OwnedStandMap = HashMap::with_capacity(4);
1820 let mut one_read_key: i32 = -1;
1821 let mut batch = Vec::with_capacity(batch_size);
1822 let mut records_since_progress = 0usize;
1823 let mut last_progress_pos = reader.get_mut().stream_position()?;
1824 let mut cigar_buf = String::with_capacity(64);
1825 let mut seq_buf = String::with_capacity(256);
1826
1827 while reader.read_record(&mut record)? != 0 {
1828 let read_id = record
1829 .name()
1830 .ok_or_else(|| anyhow::anyhow!("Missing read name"))?;
1831 if let Some(profile) = profile {
1832 profile.records.fetch_add(1, Ordering::Relaxed);
1833 }
1834 if read_id != current_id.as_slice() {
1835 if !current_id.is_empty() {
1836 self.push_sam_owned_group_for_scan2(
1837 &mut batch,
1838 ¤t_id,
1839 &mut alignments,
1840 &mut all_alignments,
1841 &mut stand_map,
1842 );
1843 if batch.len() >= batch_size {
1844 tx.send(std::mem::take(&mut batch))
1845 .map_err(|_| anyhow::anyhow!("SAM Scan2 consumer dropped"))?;
1846 batch = Vec::with_capacity(batch_size);
1847 }
1848 }
1849 current_id.clear();
1850 current_id.extend_from_slice(read_id);
1851 one_read_key = -1;
1852 }
1853
1854 let flag = i32::from(u16::from(record.flags()?));
1855 let chrom = match record.reference_sequence(&header) {
1856 Some(Ok((name, _))) => String::from_utf8_lossy(name).to_string(),
1857 _ => "*".to_string(),
1858 };
1859 let start_pos = record
1860 .alignment_start()
1861 .transpose()?
1862 .map(|p| p.get() as i32)
1863 .unwrap_or(0);
1864 let mapq = record
1865 .mapping_quality()
1866 .transpose()?
1867 .map(u8::from)
1868 .unwrap_or(0) as i32;
1869 cigar_buf.clear();
1870 for result in record.cigar().iter() {
1871 let op = result?;
1872 use noodles::sam::alignment::record::cigar::op::Kind;
1873 let op_char = match op.kind() {
1874 Kind::Match => 'M',
1875 Kind::Insertion => 'I',
1876 Kind::Deletion => 'D',
1877 Kind::Skip => 'N',
1878 Kind::SoftClip => 'S',
1879 Kind::HardClip => 'H',
1880 Kind::Pad => 'P',
1881 Kind::SequenceMatch => '=',
1882 Kind::SequenceMismatch => 'X',
1883 };
1884 let _ = write!(&mut cigar_buf, "{}{}", op.len(), op_char);
1885 }
1886 seq_buf.clear();
1887 for b in record.sequence().iter() {
1888 seq_buf.push(char::from(b));
1889 }
1890 let seq = seq_buf.clone();
1891 let s_idx = if flag & 0x40 != 0 { 1 } else { 0 };
1892 let st_c = if flag & 0x10 != 0 { '1' } else { '0' };
1893 if s_idx != one_read_key {
1894 one_read_key = s_idx;
1895 alignments.retain(|a| {
1896 let idx = if a.flag & 0x40 != 0 { 1 } else { 0 };
1897 idx != s_idx
1898 });
1899 if !seq.is_empty() && seq != "*" {
1900 stand_map.insert(s_idx, (st_c, Cow::Owned(seq.clone())));
1901 }
1902 }
1903 let alignment = AlignmentRecord {
1904 flag,
1905 chrom: Cow::Owned(chrom),
1906 pos: start_pos,
1907 mapq,
1908 cigar: Cow::Owned(cigar_buf.clone()),
1909 seq: Cow::Owned(seq),
1910 xa: Cow::Owned(raw_xa_from_sam_record(&record)?),
1911 };
1912 all_alignments.push(alignment.clone());
1913 alignments.push(alignment);
1914
1915 records_since_progress += 1;
1916 if records_since_progress >= 4096 {
1917 let pos = reader.get_mut().stream_position()?;
1918 if pos > last_progress_pos {
1919 pb.inc(pos - last_progress_pos);
1920 last_progress_pos = pos;
1921 }
1922 records_since_progress = 0;
1923 }
1924 }
1925
1926 if !current_id.is_empty() {
1927 self.push_sam_owned_group_for_scan2(
1928 &mut batch,
1929 ¤t_id,
1930 &mut alignments,
1931 &mut all_alignments,
1932 &mut stand_map,
1933 );
1934 }
1935 if !batch.is_empty() {
1936 tx.send(batch)
1937 .map_err(|_| anyhow::anyhow!("SAM Scan2 consumer dropped"))?;
1938 }
1939
1940 let final_pos = reader.get_mut().stream_position()?;
1941 if final_pos > last_progress_pos {
1942 pb.inc(final_pos - last_progress_pos);
1943 }
1944 Ok(())
1945 }
1946
1947 fn push_sam_owned_group_for_scan2(
1949 &self,
1950 batch: &mut Vec<SamOwnedScan2Group>,
1951 current_id: &[u8],
1952 alignments: &mut Vec<OwnedAlignmentRecord>,
1953 all_alignments: &mut Vec<OwnedAlignmentRecord>,
1954 stand_map: &mut OwnedStandMap,
1955 ) {
1956 batch.push(SamOwnedScan2Group {
1957 read_id: String::from_utf8_lossy(current_id).into_owned(),
1958 alignments: std::mem::take(alignments),
1959 all_alignments: std::mem::take(all_alignments),
1960 stand_map: std::mem::take(stand_map),
1961 });
1962 *alignments = Vec::with_capacity(16);
1963 *all_alignments = Vec::with_capacity(16);
1964 *stand_map = HashMap::with_capacity(4);
1965 }
1966
1967 fn process_sam_group_batch_for_scan2(
1969 &self,
1970 writer: &mut BufWriter<File>,
1971 mut display_writer: Option<&mut BufWriter<File>>,
1972 mut segments_writer: Option<&mut BufWriter<File>>,
1973 mut non_bsj_segments_writer: Option<&mut BufWriter<File>>,
1974 batch: &mut Vec<SamOwnedScan2Group>,
1975 merged_fsj: &mut HashMap<String, i32>,
1976 chr_tcga_map: &HashMap<String, String>,
1977 display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
1978 profile: Option<&Scan2Profile>,
1979 ) -> Result<()> {
1980 let groups = std::mem::take(batch);
1981 let results: Vec<(
1982 Vec<String>,
1983 HashMap<String, i32>,
1984 Vec<String>,
1985 Vec<String>,
1986 Vec<String>,
1987 )> = groups
1988 .into_par_iter()
1989 .map(|owned| {
1990 let mut local_lines = Vec::new();
1991 let mut local_fsj = HashMap::new();
1992 let mut validator = IsBSJHg2::new(self.linear_range_size_min, self.min_mapq_uni);
1993 let _ = self.process_group_view(
1994 &owned.read_id,
1995 &owned.alignments,
1996 &owned.stand_map,
1997 &mut local_lines,
1998 &mut local_fsj,
1999 chr_tcga_map,
2000 &mut validator,
2001 profile,
2002 );
2003 let display_lines = if let Some((display_helper, scan1_claims)) = display_scan2 {
2004 let mut display_validator =
2005 IsBSJHg2::new(self.linear_range_size_min, self.min_mapq_uni);
2006 display_helper
2007 .process_group_view_display(
2008 &owned.read_id,
2009 &owned.alignments,
2010 &owned.stand_map,
2011 scan1_claims,
2012 chr_tcga_map,
2013 &mut display_validator,
2014 )
2015 .unwrap_or_default()
2016 } else {
2017 Vec::new()
2018 };
2019 let has_bsj = !local_lines.is_empty() || !display_lines.is_empty();
2020 let evidence = if has_bsj {
2021 let mut bsj_lines = local_lines.clone();
2022 bsj_lines.extend(display_lines.iter().cloned());
2023 Self::segment_evidence_lines_with_local(
2024 &owned.read_id,
2025 "scan2",
2026 "scan2_local",
2027 &owned.alignments,
2028 &bsj_lines,
2029 chr_tcga_map,
2030 )
2031 } else {
2032 Vec::new()
2033 };
2034 let scan1_claimed = display_scan2.as_ref().is_some_and(|(_, scan1_claims)| {
2035 scan1_claims.by_read.contains_key(owned.read_id.as_str())
2036 });
2037 let non_bsj_evidence =
2038 if local_lines.is_empty() && display_lines.is_empty() && !scan1_claimed {
2039 Self::non_bsj_segment_evidence_lines(
2040 &owned.read_id,
2041 &owned.all_alignments,
2042 chr_tcga_map,
2043 )
2044 } else {
2045 Vec::new()
2046 };
2047 (
2048 local_lines,
2049 local_fsj,
2050 display_lines,
2051 evidence,
2052 non_bsj_evidence,
2053 )
2054 })
2055 .collect();
2056
2057 for (lines, fsj_map, display_lines, evidence, non_bsj_evidence) in results {
2058 let main_keys: HashSet<String> = lines
2059 .iter()
2060 .filter_map(|line| Self::scan2_priority_key(line))
2061 .collect();
2062 for line in lines {
2063 writeln!(writer, "{}", line)?;
2064 }
2065 if let Some(display_writer) = display_writer.as_deref_mut() {
2066 for line in display_lines {
2067 writeln!(display_writer, "{}", line)?;
2068 }
2069 } else {
2070 for line in display_lines {
2071 if let Some((key, priority_line)) =
2072 Self::scan2_display_key_and_priority_line(&line, "0")
2073 {
2074 if !main_keys.contains(&key) {
2075 writeln!(writer, "{}", priority_line)?;
2076 }
2077 }
2078 }
2079 }
2080 for (key, count) in fsj_map {
2081 *merged_fsj.entry(key).or_insert(0) += count;
2082 }
2083 if let Some(segments_writer) = segments_writer.as_deref_mut() {
2084 let mut rows = 0_u64;
2085 let mut bytes = 0_u64;
2086 for line in evidence {
2087 rows += 1;
2088 bytes += line.len() as u64 + 1;
2089 writeln!(segments_writer, "{}", line)?;
2090 }
2091 if let Some(profile) = profile {
2092 profile
2093 .segment_sidecar_rows
2094 .fetch_add(rows, Ordering::Relaxed);
2095 profile
2096 .segment_sidecar_bytes
2097 .fetch_add(bytes, Ordering::Relaxed);
2098 }
2099 }
2100 if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_deref_mut() {
2101 let mut rows = 0_u64;
2102 let mut bytes = 0_u64;
2103 for line in non_bsj_evidence {
2104 rows += 1;
2105 bytes += line.len() as u64 + 1;
2106 writeln!(non_bsj_segments_writer, "{}", line)?;
2107 }
2108 if let Some(profile) = profile {
2109 profile
2110 .non_bsj_sidecar_rows
2111 .fetch_add(rows, Ordering::Relaxed);
2112 profile
2113 .non_bsj_sidecar_bytes
2114 .fetch_add(bytes, Ordering::Relaxed);
2115 }
2116 }
2117 }
2118 Ok(())
2119 }
2120
2121 fn process_bam_shard_to_file(
2127 &self,
2128 mmap: &Mmap,
2129 start: usize,
2130 end: usize,
2131 header: &sam::Header,
2132 chr_tcga_map: &HashMap<String, String>,
2133 pb: &ProgressBar,
2134 out_path: &str,
2135 fsj_path: &str,
2136 display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
2137 display_out_path: Option<&str>,
2138 segments_out_path: Option<&str>,
2139 non_bsj_segments_out_path: Option<&str>,
2140 profile: Option<&Scan2Profile>,
2141 ) -> Result<()> {
2142 use noodles::bam;
2143 let shard_started = profile.map(|_| Instant::now());
2144 let mut writer = BufWriter::with_capacity(256 * 1024, File::create(out_path)?);
2145 let mut display_writer = if let Some(path) = display_out_path {
2146 Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
2147 } else {
2148 None
2149 };
2150 let mut segments_writer = if let Some(path) = segments_out_path {
2151 Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
2152 } else {
2153 None
2154 };
2155 let mut non_bsj_segments_writer = if let Some(path) = non_bsj_segments_out_path {
2156 Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
2157 } else {
2158 None
2159 };
2160 let mut local_fsj = HashMap::new();
2161 let mut validator = IsBSJHg2::new(self.linear_range_size_min, self.min_mapq_uni);
2162 let mut display_validator = display_scan2
2163 .as_ref()
2164 .map(|_| IsBSJHg2::new(self.linear_range_size_min, self.min_mapq_uni));
2165 let block_start = if start == 0 {
2166 0
2167 } else {
2168 let mut found = None;
2169 for i in start..mmap.len().saturating_sub(3) {
2170 if &mmap[i..i + 4] == b"\x1f\x8b\x08\x04" {
2171 found = Some(i);
2172 break;
2173 }
2174 }
2175 found.unwrap_or(mmap.len())
2176 };
2177 let pos = if start == 0 {
2178 0
2179 } else {
2180 let mut found = None;
2183 let mut i = block_start
2184 .saturating_sub(1)
2185 .min(mmap.len().saturating_sub(4));
2186 loop {
2187 if i + 3 < mmap.len() && &mmap[i..i + 4] == b"\x1f\x8b\x08\x04" {
2188 found = Some(i);
2189 break;
2190 }
2191 if i == 0 {
2192 break;
2193 }
2194 i -= 1;
2195 }
2196 found.unwrap_or(0)
2197 };
2198 if pos >= mmap.len() {
2199 writer.flush()?;
2200 if let Some(display_writer) = display_writer.as_mut() {
2201 display_writer.flush()?;
2202 }
2203 if let Some(segments_writer) = segments_writer.as_mut() {
2204 segments_writer.flush()?;
2205 }
2206 if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_mut() {
2207 non_bsj_segments_writer.flush()?;
2208 }
2209 return Self::write_fsj_shard(fsj_path, &local_fsj);
2210 }
2211
2212 let mut reader = bam::io::Reader::new(&mmap[pos..]);
2213 if start == 0 {
2214 let _ = reader.read_header()?;
2215 }
2216 let mut record = bam::Record::default();
2217 let mut current_id: Vec<u8> = Vec::new();
2218 let mut alignments: Vec<AlignmentRecord> = Vec::with_capacity(16);
2219 let mut all_alignments: Vec<AlignmentRecord> = Vec::with_capacity(16);
2220 let mut stand_map: HashMap<i32, (char, Cow<str>)> = HashMap::with_capacity(4);
2221 let mut one_read_key: i32 = -1;
2222 let mut crossed_start = start == 0;
2223 let mut leading_partial_id: Option<Vec<u8>> = None;
2224 let mut last_compressed_pos = block_start.saturating_sub(pos);
2225 let mut last_evicted_pos = pos;
2226 let eviction_threshold =
2229 (self.mem_limit as usize / 8).clamp(8 * 1024 * 1024, 64 * 1024 * 1024);
2230 let mut cigar_buf = String::with_capacity(64);
2231 let mut seq_buf = String::with_capacity(256);
2232 let mut res_batch = Vec::new();
2233
2234 while reader.read_record(&mut record)? != 0 {
2235 if let Some(profile) = profile {
2236 profile.records.fetch_add(1, Ordering::Relaxed);
2237 }
2238 let curr_c_pos = reader.get_ref().virtual_position().compressed() as usize;
2239 pb.inc(curr_c_pos.saturating_sub(last_compressed_pos) as u64);
2240 last_compressed_pos = curr_c_pos;
2241
2242 let evicted_rel = last_evicted_pos.saturating_sub(pos);
2243 if curr_c_pos.saturating_sub(evicted_rel) > eviction_threshold {
2244 advise_dontneed(
2245 mmap,
2246 last_evicted_pos,
2247 curr_c_pos.saturating_sub(evicted_rel),
2248 );
2249 last_evicted_pos = pos + curr_c_pos;
2250 }
2251
2252 let read_id = record
2253 .name()
2254 .ok_or_else(|| anyhow::anyhow!("Missing read name"))?;
2255 let abs_c_pos = pos + curr_c_pos;
2256 if !crossed_start {
2257 if abs_c_pos < block_start {
2258 leading_partial_id = Some(read_id.to_vec());
2259 continue;
2260 }
2261 crossed_start = true;
2262 }
2263 if let Some(partial_id) = &leading_partial_id {
2264 if read_id == partial_id.as_slice() {
2265 continue;
2269 }
2270 leading_partial_id = None;
2271 }
2272
2273 if read_id != current_id.as_slice() {
2274 if !current_id.is_empty() {
2275 let id_str = String::from_utf8_lossy(¤t_id);
2276 res_batch.clear();
2277 self.process_group_view(
2278 &id_str,
2279 &alignments,
2280 &stand_map,
2281 &mut res_batch,
2282 &mut local_fsj,
2283 chr_tcga_map,
2284 &mut validator,
2285 profile,
2286 )?;
2287 let write_started = profile.map(|_| Instant::now());
2288 let main_keys: HashSet<String> = res_batch
2289 .iter()
2290 .filter_map(|line| Self::scan2_priority_key(line))
2291 .collect();
2292 for line in &res_batch {
2293 writeln!(writer, "{}", line)?;
2294 }
2295 if let (Some(profile), Some(write_started)) = (profile, write_started) {
2296 profile.write_ns.fetch_add(
2297 write_started.elapsed().as_nanos() as u64,
2298 Ordering::Relaxed,
2299 );
2300 }
2301 let display_started = profile.map(|_| Instant::now());
2302 let display_lines =
2303 if let (Some((display_helper, scan1_claims)), Some(display_validator)) =
2304 (display_scan2, display_validator.as_mut())
2305 {
2306 display_helper.process_group_view_display(
2307 &id_str,
2308 &alignments,
2309 &stand_map,
2310 scan1_claims,
2311 chr_tcga_map,
2312 display_validator,
2313 )?
2314 } else {
2315 Vec::new()
2316 };
2317 if let (Some(profile), Some(display_started)) = (profile, display_started) {
2318 profile.display_ns.fetch_add(
2319 display_started.elapsed().as_nanos() as u64,
2320 Ordering::Relaxed,
2321 );
2322 }
2323 if !display_lines.is_empty() {
2324 let write_started = profile.map(|_| Instant::now());
2325 if let Some(display_writer) = display_writer.as_mut() {
2326 for line in &display_lines {
2327 writeln!(display_writer, "{}", line)?;
2328 }
2329 } else {
2330 for line in &display_lines {
2331 if let Some((key, priority_line)) =
2332 Self::scan2_display_key_and_priority_line(line, "0")
2333 {
2334 if !main_keys.contains(&key) {
2335 writeln!(writer, "{}", priority_line)?;
2336 }
2337 }
2338 }
2339 }
2340 if let (Some(profile), Some(write_started)) = (profile, write_started) {
2341 profile.write_ns.fetch_add(
2342 write_started.elapsed().as_nanos() as u64,
2343 Ordering::Relaxed,
2344 );
2345 }
2346 }
2347 let scan1_claimed = display_scan2.as_ref().is_some_and(|(_, scan1_claims)| {
2348 scan1_claims.by_read.contains_key(id_str.as_ref())
2349 });
2350 let has_bsj =
2351 !res_batch.is_empty() || !display_lines.is_empty() || scan1_claimed;
2352 if let Some(segments_writer) = segments_writer.as_mut() {
2353 let segment_started = profile.map(|_| Instant::now());
2354 if has_bsj {
2355 let mut bsj_lines = res_batch.clone();
2356 bsj_lines.extend(display_lines.iter().cloned());
2357 let mut rows = 0_u64;
2358 let mut bytes = 0_u64;
2359 for line in Self::segment_evidence_lines_with_local(
2360 &id_str,
2361 "scan2",
2362 "scan2_local",
2363 &alignments,
2364 &bsj_lines,
2365 chr_tcga_map,
2366 ) {
2367 rows += 1;
2368 bytes += line.len() as u64 + 1;
2369 writeln!(segments_writer, "{}", line)?;
2370 }
2371 if let Some(profile) = profile {
2372 profile
2373 .segment_sidecar_rows
2374 .fetch_add(rows, Ordering::Relaxed);
2375 profile
2376 .segment_sidecar_bytes
2377 .fetch_add(bytes, Ordering::Relaxed);
2378 }
2379 }
2380 if let (Some(profile), Some(segment_started)) = (profile, segment_started) {
2381 profile.segment_sidecar_ns.fetch_add(
2382 segment_started.elapsed().as_nanos() as u64,
2383 Ordering::Relaxed,
2384 );
2385 }
2386 }
2387 if !has_bsj {
2388 if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_mut() {
2389 let non_bsj_started = profile.map(|_| Instant::now());
2390 let mut rows = 0_u64;
2391 let mut bytes = 0_u64;
2392 for line in Self::non_bsj_segment_evidence_lines(
2393 &id_str,
2394 &all_alignments,
2395 chr_tcga_map,
2396 ) {
2397 rows += 1;
2398 bytes += line.len() as u64 + 1;
2399 writeln!(non_bsj_segments_writer, "{}", line)?;
2400 }
2401 if let (Some(profile), Some(non_bsj_started)) =
2402 (profile, non_bsj_started)
2403 {
2404 profile.non_bsj_sidecar_ns.fetch_add(
2405 non_bsj_started.elapsed().as_nanos() as u64,
2406 Ordering::Relaxed,
2407 );
2408 profile
2409 .non_bsj_sidecar_rows
2410 .fetch_add(rows, Ordering::Relaxed);
2411 profile
2412 .non_bsj_sidecar_bytes
2413 .fetch_add(bytes, Ordering::Relaxed);
2414 }
2415 }
2416 }
2417 if abs_c_pos > end {
2418 current_id.clear();
2419 break;
2420 }
2421 }
2422 current_id.clear();
2423 current_id.extend_from_slice(read_id);
2424 alignments.clear();
2425 all_alignments.clear();
2426 stand_map.clear();
2427 one_read_key = -1;
2428 }
2429
2430 let flag = i32::from(u16::from(record.flags()));
2431 let chrom = match record.reference_sequence(header) {
2432 Some(Ok((name, _))) => String::from_utf8_lossy(name).to_string(),
2433 _ => "*".to_string(),
2434 };
2435 let start_pos = record
2436 .alignment_start()
2437 .transpose()?
2438 .map(|p| p.get() as i32)
2439 .unwrap_or(0);
2440 let mapq = record.mapping_quality().map(u8::from).unwrap_or(0) as i32;
2441 cigar_buf.clear();
2442 for result in record.cigar().iter() {
2443 let op = result?;
2444 use noodles::sam::alignment::record::cigar::op::Kind;
2445 let op_char = match op.kind() {
2446 Kind::Match => 'M',
2447 Kind::Insertion => 'I',
2448 Kind::Deletion => 'D',
2449 Kind::Skip => 'N',
2450 Kind::SoftClip => 'S',
2451 Kind::HardClip => 'H',
2452 Kind::Pad => 'P',
2453 Kind::SequenceMatch => '=',
2454 Kind::SequenceMismatch => 'X',
2455 };
2456 let _ = write!(&mut cigar_buf, "{}{}", op.len(), op_char);
2457 }
2458 seq_buf.clear();
2459 for b in record.sequence().iter() {
2460 seq_buf.push(char::from(b));
2461 }
2462 let seq = seq_buf.clone();
2463 let s_idx = if flag & 0x40 != 0 { 1 } else { 0 };
2464 let st_c = if flag & 0x10 != 0 { '1' } else { '0' };
2465 if s_idx != one_read_key {
2466 one_read_key = s_idx;
2467 alignments.retain(|a| {
2475 let idx = if a.flag & 0x40 != 0 { 1 } else { 0 };
2476 idx != s_idx
2477 });
2478 if !seq.is_empty() && seq != "*" {
2486 stand_map.insert(s_idx, (st_c, Cow::Owned(seq.clone())));
2487 }
2488 }
2489 let alignment = AlignmentRecord {
2490 flag,
2491 chrom: Cow::Owned(chrom),
2492 pos: start_pos,
2493 mapq,
2494 cigar: Cow::Owned(cigar_buf.clone()),
2495 seq: Cow::Owned(seq),
2496 xa: Cow::Owned(raw_xa_from_bam_record(&record)?),
2497 };
2498 all_alignments.push(alignment.clone());
2499 alignments.push(alignment);
2500 }
2501 if !current_id.is_empty() {
2502 let id_str = String::from_utf8_lossy(¤t_id);
2503 res_batch.clear();
2504 self.process_group_view(
2505 &id_str,
2506 &alignments,
2507 &stand_map,
2508 &mut res_batch,
2509 &mut local_fsj,
2510 chr_tcga_map,
2511 &mut validator,
2512 profile,
2513 )?;
2514 let write_started = profile.map(|_| Instant::now());
2515 let main_keys: HashSet<String> = res_batch
2516 .iter()
2517 .filter_map(|line| Self::scan2_priority_key(line))
2518 .collect();
2519 for line in &res_batch {
2520 writeln!(writer, "{}", line)?;
2521 }
2522 if let (Some(profile), Some(write_started)) = (profile, write_started) {
2523 profile
2524 .write_ns
2525 .fetch_add(write_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
2526 }
2527 let display_started = profile.map(|_| Instant::now());
2528 let display_lines =
2529 if let (Some((display_helper, scan1_claims)), Some(display_validator)) =
2530 (display_scan2, display_validator.as_mut())
2531 {
2532 display_helper.process_group_view_display(
2533 &id_str,
2534 &alignments,
2535 &stand_map,
2536 scan1_claims,
2537 chr_tcga_map,
2538 display_validator,
2539 )?
2540 } else {
2541 Vec::new()
2542 };
2543 if let (Some(profile), Some(display_started)) = (profile, display_started) {
2544 profile.display_ns.fetch_add(
2545 display_started.elapsed().as_nanos() as u64,
2546 Ordering::Relaxed,
2547 );
2548 }
2549 if !display_lines.is_empty() {
2550 let write_started = profile.map(|_| Instant::now());
2551 if let Some(display_writer) = display_writer.as_mut() {
2552 for line in &display_lines {
2553 writeln!(display_writer, "{}", line)?;
2554 }
2555 } else {
2556 for line in &display_lines {
2557 if let Some((key, priority_line)) =
2558 Self::scan2_display_key_and_priority_line(line, "0")
2559 {
2560 if !main_keys.contains(&key) {
2561 writeln!(writer, "{}", priority_line)?;
2562 }
2563 }
2564 }
2565 }
2566 if let (Some(profile), Some(write_started)) = (profile, write_started) {
2567 profile
2568 .write_ns
2569 .fetch_add(write_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
2570 }
2571 }
2572 let scan1_claimed = display_scan2.as_ref().is_some_and(|(_, scan1_claims)| {
2573 scan1_claims.by_read.contains_key(id_str.as_ref())
2574 });
2575 let has_bsj = !res_batch.is_empty() || !display_lines.is_empty() || scan1_claimed;
2576 if let Some(segments_writer) = segments_writer.as_mut() {
2577 let segment_started = profile.map(|_| Instant::now());
2578 if has_bsj {
2579 let mut bsj_lines = res_batch.clone();
2580 bsj_lines.extend(display_lines.iter().cloned());
2581 let mut rows = 0_u64;
2582 let mut bytes = 0_u64;
2583 for line in Self::segment_evidence_lines_with_local(
2584 &id_str,
2585 "scan2",
2586 "scan2_local",
2587 &alignments,
2588 &bsj_lines,
2589 chr_tcga_map,
2590 ) {
2591 rows += 1;
2592 bytes += line.len() as u64 + 1;
2593 writeln!(segments_writer, "{}", line)?;
2594 }
2595 if let Some(profile) = profile {
2596 profile
2597 .segment_sidecar_rows
2598 .fetch_add(rows, Ordering::Relaxed);
2599 profile
2600 .segment_sidecar_bytes
2601 .fetch_add(bytes, Ordering::Relaxed);
2602 }
2603 }
2604 if let (Some(profile), Some(segment_started)) = (profile, segment_started) {
2605 profile.segment_sidecar_ns.fetch_add(
2606 segment_started.elapsed().as_nanos() as u64,
2607 Ordering::Relaxed,
2608 );
2609 }
2610 }
2611 if !has_bsj {
2612 if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_mut() {
2613 let non_bsj_started = profile.map(|_| Instant::now());
2614 let mut rows = 0_u64;
2615 let mut bytes = 0_u64;
2616 for line in
2617 Self::non_bsj_segment_evidence_lines(&id_str, &all_alignments, chr_tcga_map)
2618 {
2619 rows += 1;
2620 bytes += line.len() as u64 + 1;
2621 writeln!(non_bsj_segments_writer, "{}", line)?;
2622 }
2623 if let (Some(profile), Some(non_bsj_started)) = (profile, non_bsj_started) {
2624 profile.non_bsj_sidecar_ns.fetch_add(
2625 non_bsj_started.elapsed().as_nanos() as u64,
2626 Ordering::Relaxed,
2627 );
2628 profile
2629 .non_bsj_sidecar_rows
2630 .fetch_add(rows, Ordering::Relaxed);
2631 profile
2632 .non_bsj_sidecar_bytes
2633 .fetch_add(bytes, Ordering::Relaxed);
2634 }
2635 }
2636 }
2637 }
2638 let evicted_rel = last_evicted_pos.saturating_sub(pos);
2639 if last_compressed_pos > evicted_rel {
2640 advise_dontneed(mmap, last_evicted_pos, last_compressed_pos - evicted_rel);
2641 }
2642 writer.flush()?;
2643 if let Some(display_writer) = display_writer.as_mut() {
2644 display_writer.flush()?;
2645 }
2646 if let Some(segments_writer) = segments_writer.as_mut() {
2647 segments_writer.flush()?;
2648 }
2649 if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_mut() {
2650 non_bsj_segments_writer.flush()?;
2651 }
2652 if let (Some(profile), Some(shard_started)) = (profile, shard_started) {
2653 profile
2654 .shard_total_ns
2655 .fetch_add(shard_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
2656 }
2657 Self::write_fsj_shard(fsj_path, &local_fsj)
2658 }
2659
2660 pub(crate) fn process_group_view<'a>(
2669 &self,
2670 id: &str,
2671 alignments: &[AlignmentRecord<'a>],
2672 stand_map: &HashMap<i32, (char, Cow<'a, str>)>,
2673 results: &mut Vec<String>,
2674 local_fsj: &mut HashMap<String, i32>,
2675 chr_tcga_map: &HashMap<String, String>,
2676 is_bsj_hg2: &mut IsBSJHg2,
2677 profile: Option<&Scan2Profile>,
2678 ) -> Result<()> {
2679 let started = profile.map(|_| Instant::now());
2680 let trace_read = should_trace_read(id);
2681 let trace_all_candidates =
2682 trace_read && std::env::var("CIRI_TRACE_ALL_CANDS").ok().as_deref() == Some("1");
2683 if self.scan1_ids.contains(id) {
2684 return Ok(());
2685 }
2686 let mut segments: HashMap<i32, Vec<&AlignmentRecord<'a>>> = HashMap::new();
2687 for aln in alignments {
2688 segments
2689 .entry(if aln.flag & 0x40 != 0 { 1 } else { 0 })
2690 .or_default()
2691 .push(aln);
2692 }
2693 let mut tem_fsj_keys = HashSet::new();
2697 for seg_idx in [0_i32, 1_i32] {
2698 let Some(seg_alns) = segments.get(&seg_idx) else {
2699 continue;
2700 };
2701 let (read_strand, seq) = match stand_map.get(&seg_idx) {
2702 Some(&(st, ref s)) => (st, s.as_ref()),
2703 None => continue,
2704 };
2705 let slen = seq.len() as i32;
2706 let mate_seg_alns = segments.get(&(1 - seg_idx));
2709 for aln in seg_alns {
2710 let chr = aln.chrom.as_ref();
2711 if !self.index1.contains_key(chr) {
2712 continue;
2713 }
2714 let cigar_ref: Cow<'_, str> = if aln.cigar.contains('H') {
2715 Cow::Owned(aln.cigar.replace('H', "S"))
2716 } else {
2717 Cow::Borrowed(aln.cigar.as_ref())
2718 };
2719 if cigar_ref == "*" {
2720 continue;
2721 }
2722 let c = misd(cigar_ref.as_ref(), slen);
2723 if cigar_ref == format!("{}M", slen) {
2724 let start_tem = aln.pos + 6;
2725 let end_tem = aln.pos + slen - 7;
2726 self.collect_fsj_keys_in_range(chr, start_tem, end_tem, 0, &mut tem_fsj_keys);
2727 continue;
2728 }
2729 if c[0] == -1 || c[0] == 10 {
2730 if let Some(list) = self.index1.get(chr) {
2731 let curr_strand = if aln.flag & 0x10 != 0 { '1' } else { '0' };
2732 let str_e_owned = if aln.flag & 0x10 != 0 {
2736 if read_strand == '0' {
2737 reverse_complement(seq)
2738 } else {
2739 seq.to_string()
2740 }
2741 } else if read_strand == '0' {
2742 seq.to_string()
2743 } else {
2744 reverse_complement(seq)
2745 };
2746 let p_str_owned =
2747 if let Some(&(p_strand, ref p_seq)) = stand_map.get(&(1 - seg_idx)) {
2748 if p_strand != curr_strand {
2749 p_seq.to_string()
2750 } else {
2751 reverse_complement(p_seq)
2752 }
2753 } else {
2754 String::new()
2755 };
2756 let new_num_site = aln.pos;
2757 let bucket_size = self.seq_len.max(1);
2759 let num1 = (new_num_site - 6) / bucket_size;
2760 let num2 = (new_num_site + 6) / bucket_size;
2761
2762 let mut eval_candidate = |cand: &CandidateBreakpoint| -> Result<bool> {
2763 let e_idx = c[1] + (cand.site - new_num_site);
2764 if e_idx <= 0 || e_idx > slen {
2765 return Ok(false);
2766 }
2767 let str_f = &str_e_owned[0..e_idx as usize];
2768 let str3 = if c[0] == 10 {
2769 let si = slen - c[2];
2770 if si >= 0 && si <= slen {
2771 &str_e_owned[si as usize..]
2772 } else {
2773 ""
2774 }
2775 } else {
2776 "*"
2777 };
2778 let site1 = cand.data[0].parse::<i32>().unwrap_or(0);
2779 let site2 = cand.data[1].parse::<i32>().unwrap_or(0);
2780 let mut s2_ok = if mate_seg_alns.is_some() { 0 } else { 1 };
2781 if let Some(mate_seg_alns) = mate_seg_alns {
2782 for mate_aln in mate_seg_alns {
2783 if mate_aln.chrom.as_ref() == chr
2784 && mate_aln.mapq >= self.min_mapq_uni
2785 {
2786 let c_ano = misd(&mate_aln.cigar, slen);
2787 let mate_strand =
2788 if mate_aln.flag & 0x10 != 0 { '1' } else { '0' };
2789 if mate_strand != curr_strand
2790 && mate_aln.pos >= site1 - 6
2791 && mate_aln.pos + c_ano[3] - 1 <= site2 + 6
2792 {
2793 s2_ok = 1;
2794 } else {
2795 s2_ok = -1;
2796 }
2797 break;
2798 }
2799 }
2800 }
2801 let circ_c = vec![
2802 (if aln.flag & 0x10 != 0 { "1" } else { "0" }).to_string(),
2803 chr.to_string(),
2804 "sm".to_string(),
2805 cand.data[0].clone(),
2806 cand.data[1].clone(),
2807 str_f.to_string(),
2808 p_str_owned.clone(),
2809 str3.to_string(),
2810 s2_ok.to_string(),
2811 cand.data[2].clone(),
2812 cand.data[3].clone(),
2813 cand.data[4].clone(),
2814 aln.mapq.to_string(),
2815 ];
2816 let validator_started = profile.map(|_| Instant::now());
2817 let tag = with_trace_hg2_scope(trace_read, || {
2818 is_bsj_hg2.is_bsj_hg2(&circ_c, chr_tcga_map.get(chr).unwrap())
2819 });
2820 if let Some(profile) = profile {
2821 profile.candidate_checks.fetch_add(1, Ordering::Relaxed);
2822 if let Some(validator_started) = validator_started {
2823 profile.validator_call_ns.fetch_add(
2824 validator_started.elapsed().as_nanos() as u64,
2825 Ordering::Relaxed,
2826 );
2827 }
2828 }
2829 if trace_read {
2830 emit_trace_line(&format!(
2831 "[TRACE_SCAN2_CAND] id={} type=sm seg={} aln_pos={} chr={} site1={} site2={} cand_site={} cigar={} mapq={} s2_ok={} str_len={} pair_len={} tag={}",
2832 id,
2833 seg_idx,
2834 aln.pos,
2835 chr,
2836 cand.data[0],
2837 cand.data[1],
2838 cand.site,
2839 cigar_ref.as_ref(),
2840 aln.mapq,
2841 circ_c[8],
2842 circ_c[5].len(),
2843 circ_c[6].len(),
2844 tag
2845 ));
2846 }
2847 if tag == "0" {
2848 tem_fsj_keys
2855 .insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
2856 Ok(false)
2857 } else if tag != "2" {
2858 if let Some(profile) = profile {
2859 profile.candidate_hits.fetch_add(1, Ordering::Relaxed);
2860 }
2861 let tag_body = &tag[0..tag.len() - 1];
2862 let mate_label = if seg_idx == 1 { "R1" } else { "R2" };
2863 results.push(format!(
2864 "{}\t{}\t1\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
2865 id,
2866 mate_label,
2867 cigar_ref.as_ref(),
2868 tag_body,
2869 chr,
2870 cand.data[0],
2871 cand.data[1],
2872 cand.data[2],
2873 cand.data[3],
2874 cand.data[4],
2875 tag.chars().next_back().unwrap()
2876 ));
2877 Ok(true)
2878 } else {
2879 Ok(false)
2880 }
2881 };
2882
2883 if num1 > 0 && self.site_array1.get(chr).is_some_and(|s| s.contains(&num1))
2885 {
2886 let (l1, r1) = Self::bucket_range(list, num1, bucket_size);
2887 for idx in (l1..r1).rev() {
2888 let cand = &list[idx];
2889 let bias = cand.site - new_num_site;
2890 if bias >= -6 {
2891 if bias <= 6 && eval_candidate(cand)? && !trace_all_candidates {
2892 return Ok(());
2893 }
2894 } else {
2895 break;
2896 }
2897 }
2898 }
2899 if num2 != num1
2901 && self.site_array1.get(chr).is_some_and(|s| s.contains(&num2))
2902 {
2903 let (l2, r2) = Self::bucket_range(list, num2, bucket_size);
2904 for idx in l2..r2 {
2905 let cand = &list[idx];
2906 let bias = cand.site - new_num_site;
2907 if bias <= 6 {
2908 if eval_candidate(cand)? && !trace_all_candidates {
2909 return Ok(());
2910 }
2911 } else {
2912 break;
2913 }
2914 }
2915 }
2916 }
2917 }
2918 if c[0] == 1 || c[0] == 10 {
2919 let new_site = aln.pos + c[3] - 1;
2920 if let Some(list) = self.index2.get(chr) {
2921 let curr_strand = if aln.flag & 0x10 != 0 { '1' } else { '0' };
2922 let str_e_owned = if aln.flag & 0x10 != 0 {
2923 if read_strand == '0' {
2924 reverse_complement(seq)
2925 } else {
2926 seq.to_string()
2927 }
2928 } else if read_strand == '0' {
2929 seq.to_string()
2930 } else {
2931 reverse_complement(seq)
2932 };
2933 let p_str_owned =
2934 if let Some(&(p_strand, ref p_seq)) = stand_map.get(&(1 - seg_idx)) {
2935 if p_strand != curr_strand {
2936 p_seq.to_string()
2937 } else {
2938 reverse_complement(p_seq)
2939 }
2940 } else {
2941 String::new()
2942 };
2943 let bucket_size = self.seq_len.max(1);
2945 let num1 = (new_site - 6) / bucket_size;
2946 let num2 = (new_site + 6) / bucket_size;
2947
2948 let mut eval_candidate = |cand: &CandidateBreakpoint| -> Result<bool> {
2949 let s_idx = if c[0] == 10 {
2950 slen - c[2] + (cand.site - new_site)
2956 } else {
2957 c[1] + (cand.site - new_site)
2958 };
2959 if s_idx < 0 || s_idx >= slen {
2960 return Ok(false);
2961 }
2962 let str_f = &str_e_owned[s_idx as usize..];
2963 let str3 = if c[0] == 10 {
2964 &str_e_owned[0..c[1] as usize]
2965 } else {
2966 "*"
2967 };
2968 let site1 = cand.data[0].parse::<i32>().unwrap_or(0);
2969 let site2 = cand.data[1].parse::<i32>().unwrap_or(0);
2970 let mut s2_ok = if mate_seg_alns.is_some() { 0 } else { 1 };
2971 if let Some(mate_seg_alns) = mate_seg_alns {
2972 for mate_aln in mate_seg_alns {
2973 if mate_aln.chrom.as_ref() == chr
2974 && mate_aln.mapq >= self.min_mapq_uni
2975 {
2976 let c_ano = misd(&mate_aln.cigar, slen);
2977 let mate_strand =
2978 if mate_aln.flag & 0x10 != 0 { '1' } else { '0' };
2979 if mate_strand != curr_strand
2980 && mate_aln.pos >= site1 - 6
2981 && mate_aln.pos + c_ano[3] - 1 <= site2 + 6
2982 {
2983 s2_ok = 1;
2984 } else {
2985 s2_ok = -1;
2986 }
2987 break;
2988 }
2989 }
2990 }
2991 let circ_c = vec![
2992 (if aln.flag & 0x10 != 0 { "1" } else { "0" }).to_string(),
2993 chr.to_string(),
2994 "ms".to_string(),
2995 cand.data[0].clone(),
2996 cand.data[1].clone(),
2997 str_f.to_string(),
2998 p_str_owned.clone(),
2999 str3.to_string(),
3000 s2_ok.to_string(),
3001 cand.data[2].clone(),
3002 cand.data[3].clone(),
3003 cand.data[4].clone(),
3004 aln.mapq.to_string(),
3005 ];
3006 let validator_started = profile.map(|_| Instant::now());
3007 let tag = with_trace_hg2_scope(trace_read, || {
3008 is_bsj_hg2.is_bsj_hg2(&circ_c, chr_tcga_map.get(chr).unwrap())
3009 });
3010 if let Some(profile) = profile {
3011 profile.candidate_checks.fetch_add(1, Ordering::Relaxed);
3012 if let Some(validator_started) = validator_started {
3013 profile.validator_call_ns.fetch_add(
3014 validator_started.elapsed().as_nanos() as u64,
3015 Ordering::Relaxed,
3016 );
3017 }
3018 }
3019 if trace_read {
3020 emit_trace_line(&format!(
3021 "[TRACE_SCAN2_CAND] id={} type=ms seg={} aln_pos={} chr={} site1={} site2={} cand_site={} cigar={} mapq={} s2_ok={} str_len={} pair_len={} tag={}",
3022 id,
3023 seg_idx,
3024 aln.pos,
3025 chr,
3026 cand.data[0],
3027 cand.data[1],
3028 cand.site,
3029 cigar_ref.as_ref(),
3030 aln.mapq,
3031 circ_c[8],
3032 circ_c[5].len(),
3033 circ_c[6].len(),
3034 tag
3035 ));
3036 }
3037 if tag == "0" {
3038 tem_fsj_keys
3039 .insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
3040 Ok(false)
3041 } else if tag != "2" {
3042 if let Some(profile) = profile {
3043 profile.candidate_hits.fetch_add(1, Ordering::Relaxed);
3044 }
3045 let tag_body = &tag[0..tag.len() - 1];
3046 let mate_label = if seg_idx == 1 { "R1" } else { "R2" };
3047 results.push(format!(
3048 "{}\t{}\t1\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
3049 id,
3050 mate_label,
3051 cigar_ref.as_ref(),
3052 tag_body,
3053 chr,
3054 cand.data[0],
3055 cand.data[1],
3056 cand.data[2],
3057 cand.data[3],
3058 cand.data[4],
3059 tag.chars().next_back().unwrap()
3060 ));
3061 Ok(true)
3062 } else {
3063 Ok(false)
3064 }
3065 };
3066
3067 if self.site_array2.get(chr).is_some_and(|s| s.contains(&num1)) {
3069 let (l1, r1) = Self::bucket_range(list, num1, bucket_size);
3070 for idx in (l1..r1).rev() {
3071 let cand = &list[idx];
3072 let bias = cand.site - new_site;
3073 if bias >= -6 {
3074 if bias <= 6 && eval_candidate(cand)? && !trace_all_candidates {
3075 return Ok(());
3076 }
3077 } else {
3078 break;
3079 }
3080 }
3081 }
3082 if num2 != num1
3084 && self.site_array2.get(chr).is_some_and(|s| s.contains(&num2))
3085 {
3086 let (l2, r2) = Self::bucket_range(list, num2, bucket_size);
3087 for idx in l2..r2 {
3088 let cand = &list[idx];
3089 let bias = cand.site - new_site;
3090 if bias <= 6 {
3091 if eval_candidate(cand)? && !trace_all_candidates {
3092 return Ok(());
3093 }
3094 } else {
3095 break;
3096 }
3097 }
3098 }
3099 }
3100 }
3101 let start_tem = aln.pos + 6;
3105 let end_tem = aln.pos + c[3] - 7;
3106 self.collect_fsj_keys_in_range(chr, start_tem, end_tem, c[0], &mut tem_fsj_keys);
3107 }
3108 }
3109 for k in tem_fsj_keys {
3110 *local_fsj.entry(k).or_insert(0) += 1;
3111 }
3112 if let (Some(profile), Some(started)) = (profile, started) {
3113 profile
3114 .group_process_ns
3115 .fetch_add(started.elapsed().as_nanos() as u64, Ordering::Relaxed);
3116 profile.groups.fetch_add(1, Ordering::Relaxed);
3117 }
3118 Ok(())
3119 }
3120
3121 fn process_group_view_display<'a>(
3127 &self,
3128 id: &str,
3129 alignments: &[AlignmentRecord<'a>],
3130 stand_map: &HashMap<i32, (char, Cow<'a, str>)>,
3131 scan1_claims: &Scan2DisplayClaims,
3132 chr_tcga_map: &HashMap<String, String>,
3133 is_bsj_hg2: &mut IsBSJHg2,
3134 ) -> Result<Vec<String>> {
3135 let trace_read = should_trace_read(id);
3136 let trace_all_candidates =
3137 trace_read && std::env::var("CIRI_TRACE_ALL_CANDS").ok().as_deref() == Some("1");
3138 if scan1_claims.contains_both_mates(id) {
3139 return Ok(Vec::new());
3140 }
3141 let mut results = Vec::new();
3142 let mut segments: HashMap<i32, Vec<&AlignmentRecord<'a>>> = HashMap::new();
3143 for aln in alignments {
3144 segments
3145 .entry(if aln.flag & 0x40 != 0 { 1 } else { 0 })
3146 .or_default()
3147 .push(aln);
3148 }
3149
3150 'mate: for seg_idx in [1_i32, 0_i32] {
3151 let mate_label = if seg_idx == 1 { "R1" } else { "R2" };
3152 if scan1_claims.contains_mate(id, mate_label) {
3153 continue;
3154 }
3155 let Some(seg_alns) = segments.get(&seg_idx) else {
3156 continue;
3157 };
3158 let (read_strand, seq) = match stand_map.get(&seg_idx) {
3159 Some(&(st, ref s)) => (st, s.as_ref()),
3160 None => continue,
3161 };
3162 let slen = seq.len() as i32;
3163 let mate_seg_alns = segments.get(&(1 - seg_idx));
3164 for aln in seg_alns {
3165 let chr = aln.chrom.as_ref();
3166 if !self.index1.contains_key(chr) {
3167 continue;
3168 }
3169 let cigar_ref: Cow<'_, str> = if aln.cigar.contains('H') {
3170 Cow::Owned(aln.cigar.replace('H', "S"))
3171 } else {
3172 Cow::Borrowed(aln.cigar.as_ref())
3173 };
3174 if cigar_ref == "*" {
3175 continue;
3176 }
3177 let c = misd(cigar_ref.as_ref(), slen);
3178 if c[0] == -1 || c[0] == 10 {
3179 if let Some(list) = self.index1.get(chr) {
3180 let curr_strand = if aln.flag & 0x10 != 0 { '1' } else { '0' };
3181 let str_e_owned = if aln.flag & 0x10 != 0 {
3182 if read_strand == '0' {
3183 reverse_complement(seq)
3184 } else {
3185 seq.to_string()
3186 }
3187 } else if read_strand == '0' {
3188 seq.to_string()
3189 } else {
3190 reverse_complement(seq)
3191 };
3192 let p_str_owned =
3193 if let Some(&(p_strand, ref p_seq)) = stand_map.get(&(1 - seg_idx)) {
3194 if p_strand != curr_strand {
3195 p_seq.to_string()
3196 } else {
3197 reverse_complement(p_seq)
3198 }
3199 } else {
3200 String::new()
3201 };
3202 let new_num_site = aln.pos;
3203 let bucket_size = self.seq_len.max(1);
3204 let num1 = (new_num_site - 6) / bucket_size;
3205 let num2 = (new_num_site + 6) / bucket_size;
3206
3207 let mut eval_candidate = |cand: &CandidateBreakpoint| -> Result<bool> {
3208 let e_idx = c[1] + (cand.site - new_num_site);
3209 if e_idx <= 0 || e_idx > slen {
3210 return Ok(false);
3211 }
3212 let str_f = &str_e_owned[0..e_idx as usize];
3213 let str3 = if c[0] == 10 {
3214 let si = slen - c[2];
3215 if si >= 0 && si <= slen {
3216 &str_e_owned[si as usize..]
3217 } else {
3218 ""
3219 }
3220 } else {
3221 "*"
3222 };
3223 let site1 = cand.data[0].parse::<i32>().unwrap_or(0);
3224 let site2 = cand.data[1].parse::<i32>().unwrap_or(0);
3225 let mut s2_ok = if mate_seg_alns.is_some() { 0 } else { 1 };
3226 if let Some(mate_seg_alns) = mate_seg_alns {
3227 for mate_aln in mate_seg_alns {
3228 if mate_aln.chrom.as_ref() == chr
3229 && mate_aln.mapq >= self.min_mapq_uni
3230 {
3231 let c_ano = misd(&mate_aln.cigar, slen);
3232 let mate_strand =
3233 if mate_aln.flag & 0x10 != 0 { '1' } else { '0' };
3234 if mate_strand != curr_strand
3235 && mate_aln.pos >= site1 - 6
3236 && mate_aln.pos + c_ano[3] - 1 <= site2 + 6
3237 {
3238 s2_ok = 1;
3239 } else {
3240 s2_ok = -1;
3241 }
3242 break;
3243 }
3244 }
3245 }
3246 let circ_c = vec![
3247 (if aln.flag & 0x10 != 0 { "1" } else { "0" }).to_string(),
3248 chr.to_string(),
3249 "sm".to_string(),
3250 cand.data[0].clone(),
3251 cand.data[1].clone(),
3252 str_f.to_string(),
3253 p_str_owned.clone(),
3254 str3.to_string(),
3255 s2_ok.to_string(),
3256 cand.data[2].clone(),
3257 cand.data[3].clone(),
3258 cand.data[4].clone(),
3259 aln.mapq.to_string(),
3260 ];
3261 let tag = with_trace_hg2_scope(trace_read, || {
3262 is_bsj_hg2.is_bsj_hg2(&circ_c, chr_tcga_map.get(chr).unwrap())
3263 });
3264 if trace_read {
3265 emit_trace_line(&format!(
3266 "[TRACE_SCAN2_DISPLAY] id={} type=sm seg={} aln_pos={} chr={} site1={} site2={} cand_site={} cigar={} mapq={} tag={}",
3267 id,
3268 seg_idx,
3269 aln.pos,
3270 chr,
3271 cand.data[0],
3272 cand.data[1],
3273 cand.site,
3274 cigar_ref.as_ref(),
3275 aln.mapq,
3276 tag
3277 ));
3278 }
3279 if tag != "0" && tag != "2" {
3280 let tag_body = &tag[0..tag.len() - 1];
3281 results.push(format!(
3282 "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
3283 id,
3284 mate_label,
3285 cigar_ref.as_ref(),
3286 tag_body,
3287 chr,
3288 cand.data[0],
3289 cand.data[1],
3290 cand.data[2],
3291 cand.data[3],
3292 cand.data[4],
3293 tag.chars().next_back().unwrap()
3294 ));
3295 Ok(true)
3296 } else {
3297 Ok(false)
3298 }
3299 };
3300
3301 if num1 > 0 && self.site_array1.get(chr).is_some_and(|s| s.contains(&num1))
3302 {
3303 let (l1, r1) = Self::bucket_range(list, num1, bucket_size);
3304 for idx in (l1..r1).rev() {
3305 let cand = &list[idx];
3306 let bias = cand.site - new_num_site;
3307 if bias >= -6 {
3308 if bias <= 6 && eval_candidate(cand)? && !trace_all_candidates {
3309 continue 'mate;
3310 }
3311 } else {
3312 break;
3313 }
3314 }
3315 }
3316 if num2 != num1
3317 && self.site_array1.get(chr).is_some_and(|s| s.contains(&num2))
3318 {
3319 let (l2, r2) = Self::bucket_range(list, num2, bucket_size);
3320 for idx in l2..r2 {
3321 let cand = &list[idx];
3322 let bias = cand.site - new_num_site;
3323 if bias <= 6 {
3324 if eval_candidate(cand)? && !trace_all_candidates {
3325 continue 'mate;
3326 }
3327 } else {
3328 break;
3329 }
3330 }
3331 }
3332 }
3333 }
3334 if c[0] == 1 || c[0] == 10 {
3335 let new_site = aln.pos + c[3] - 1;
3336 if let Some(list) = self.index2.get(chr) {
3337 let curr_strand = if aln.flag & 0x10 != 0 { '1' } else { '0' };
3338 let str_e_owned = if aln.flag & 0x10 != 0 {
3339 if read_strand == '0' {
3340 reverse_complement(seq)
3341 } else {
3342 seq.to_string()
3343 }
3344 } else if read_strand == '0' {
3345 seq.to_string()
3346 } else {
3347 reverse_complement(seq)
3348 };
3349 let p_str_owned =
3350 if let Some(&(p_strand, ref p_seq)) = stand_map.get(&(1 - seg_idx)) {
3351 if p_strand != curr_strand {
3352 p_seq.to_string()
3353 } else {
3354 reverse_complement(p_seq)
3355 }
3356 } else {
3357 String::new()
3358 };
3359 let bucket_size = self.seq_len.max(1);
3360 let num1 = (new_site - 6) / bucket_size;
3361 let num2 = (new_site + 6) / bucket_size;
3362
3363 let mut eval_candidate = |cand: &CandidateBreakpoint| -> Result<bool> {
3364 let s_idx = if c[0] == 10 {
3365 slen - c[2] + (cand.site - new_site)
3366 } else {
3367 c[1] + (cand.site - new_site)
3368 };
3369 if s_idx < 0 || s_idx >= slen {
3370 return Ok(false);
3371 }
3372 let str_f = &str_e_owned[s_idx as usize..];
3373 let str3 = if c[0] == 10 {
3374 &str_e_owned[0..c[1] as usize]
3375 } else {
3376 "*"
3377 };
3378 let site1 = cand.data[0].parse::<i32>().unwrap_or(0);
3379 let site2 = cand.data[1].parse::<i32>().unwrap_or(0);
3380 let mut s2_ok = if mate_seg_alns.is_some() { 0 } else { 1 };
3381 if let Some(mate_seg_alns) = mate_seg_alns {
3382 for mate_aln in mate_seg_alns {
3383 if mate_aln.chrom.as_ref() == chr
3384 && mate_aln.mapq >= self.min_mapq_uni
3385 {
3386 let c_ano = misd(&mate_aln.cigar, slen);
3387 let mate_strand =
3388 if mate_aln.flag & 0x10 != 0 { '1' } else { '0' };
3389 if mate_strand != curr_strand
3390 && mate_aln.pos >= site1 - 6
3391 && mate_aln.pos + c_ano[3] - 1 <= site2 + 6
3392 {
3393 s2_ok = 1;
3394 } else {
3395 s2_ok = -1;
3396 }
3397 break;
3398 }
3399 }
3400 }
3401 let circ_c = vec![
3402 (if aln.flag & 0x10 != 0 { "1" } else { "0" }).to_string(),
3403 chr.to_string(),
3404 "ms".to_string(),
3405 cand.data[0].clone(),
3406 cand.data[1].clone(),
3407 str_f.to_string(),
3408 p_str_owned.clone(),
3409 str3.to_string(),
3410 s2_ok.to_string(),
3411 cand.data[2].clone(),
3412 cand.data[3].clone(),
3413 cand.data[4].clone(),
3414 aln.mapq.to_string(),
3415 ];
3416 let tag = with_trace_hg2_scope(trace_read, || {
3417 is_bsj_hg2.is_bsj_hg2(&circ_c, chr_tcga_map.get(chr).unwrap())
3418 });
3419 if trace_read {
3420 emit_trace_line(&format!(
3421 "[TRACE_SCAN2_DISPLAY] id={} type=ms seg={} aln_pos={} chr={} site1={} site2={} cand_site={} cigar={} mapq={} tag={}",
3422 id,
3423 seg_idx,
3424 aln.pos,
3425 chr,
3426 cand.data[0],
3427 cand.data[1],
3428 cand.site,
3429 cigar_ref.as_ref(),
3430 aln.mapq,
3431 tag
3432 ));
3433 }
3434 if tag != "0" && tag != "2" {
3435 let tag_body = &tag[0..tag.len() - 1];
3436 results.push(format!(
3437 "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
3438 id,
3439 mate_label,
3440 cigar_ref.as_ref(),
3441 tag_body,
3442 chr,
3443 cand.data[0],
3444 cand.data[1],
3445 cand.data[2],
3446 cand.data[3],
3447 cand.data[4],
3448 tag.chars().next_back().unwrap()
3449 ));
3450 Ok(true)
3451 } else {
3452 Ok(false)
3453 }
3454 };
3455
3456 if self.site_array2.get(chr).is_some_and(|s| s.contains(&num1)) {
3457 let (l1, r1) = Self::bucket_range(list, num1, bucket_size);
3458 for idx in (l1..r1).rev() {
3459 let cand = &list[idx];
3460 let bias = cand.site - new_site;
3461 if bias >= -6 {
3462 if bias <= 6 && eval_candidate(cand)? && !trace_all_candidates {
3463 continue 'mate;
3464 }
3465 } else {
3466 break;
3467 }
3468 }
3469 }
3470 if num2 != num1
3471 && self.site_array2.get(chr).is_some_and(|s| s.contains(&num2))
3472 {
3473 let (l2, r2) = Self::bucket_range(list, num2, bucket_size);
3474 for idx in l2..r2 {
3475 let cand = &list[idx];
3476 let bias = cand.site - new_site;
3477 if bias <= 6 {
3478 if eval_candidate(cand)? && !trace_all_candidates {
3479 continue 'mate;
3480 }
3481 } else {
3482 break;
3483 }
3484 }
3485 }
3486 }
3487 }
3488 }
3489 }
3490
3491 Ok(results)
3492 }
3493}