dino_seq/fastq/
pair.rs

1use std::io::Read;
2
3use crate::error::{FastqError, Result};
4use crate::fastq_frame;
5
6use super::record::{FastqRecord, RecordRef};
7use super::{FastqBatch, FastqConfig, FastqReader};
8
9/// Pairing interpretation for a single FASTQ stream.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum PairingMode {
12    /// Records are yielded independently.
13    None,
14    /// Adjacent records are expected to be ordered mates.
15    Interleaved,
16}
17
18/// Identifier validation policy for paired-end data.
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub enum PairValidation {
21    /// Compare normalized first-token identifiers after removing `/1` or `/2`.
22    Full,
23    /// Fast path for ordered `/1` and `/2` mate suffixes, with fallback to
24    /// [`Full`](Self::Full) when that shape is not present.
25    FastSlash,
26    /// Skip identifier checks for trusted, already synchronized inputs.
27    None,
28}
29
30/// Borrowed view of an ordered read pair.
31#[derive(Debug, Clone, Copy)]
32pub struct FastqPair<'a> {
33    pub(crate) first: FastqRecord<'a>,
34    pub(crate) second: FastqRecord<'a>,
35}
36
37impl<'a> FastqPair<'a> {
38    /// First mate.
39    pub fn first(&self) -> FastqRecord<'a> {
40        self.first
41    }
42
43    /// Second mate.
44    pub fn second(&self) -> FastqRecord<'a> {
45        self.second
46    }
47
48    /// Normalized pair identifier from the first mate.
49    pub fn pair_id(&self) -> &'a [u8] {
50        self.first.pair_normalized_id()
51    }
52}
53
54/// Iterator over adjacent pairs in an interleaved [`FastqBatch`].
55#[derive(Debug)]
56pub struct InterleavedPairs<'a> {
57    pub(crate) batch: &'a FastqBatch<'a>,
58    pub(crate) next: usize,
59}
60
61impl<'a> Iterator for InterleavedPairs<'a> {
62    type Item = FastqPair<'a>;
63
64    fn next(&mut self) -> Option<Self::Item> {
65        if self.next >= self.batch.records.len() {
66            return None;
67        }
68        let pair = FastqPair {
69            first: self.batch.record_at(self.next),
70            second: self.batch.record_at(self.next + 1),
71        };
72        self.next += 2;
73        Some(pair)
74    }
75}
76
77/// Iterator over paired records from two separate [`FastqBatch`] values.
78#[derive(Debug)]
79pub struct PairedRecords<'a> {
80    pub(crate) first: &'a FastqBatch<'a>,
81    pub(crate) second: &'a FastqBatch<'a>,
82    pub(crate) next: usize,
83}
84
85impl<'a> Iterator for PairedRecords<'a> {
86    type Item = FastqPair<'a>;
87
88    fn next(&mut self) -> Option<Self::Item> {
89        if self.next >= self.first.records.len() {
90            return None;
91        }
92        let pair = FastqPair {
93            first: self.first.record_at(self.next),
94            second: self.second.record_at(self.next),
95        };
96        self.next += 1;
97        Some(pair)
98    }
99}
100
101/// Validate and zip two separate FASTQ batches by ordered mate identifiers.
102pub fn paired_records<'a>(
103    first: &'a FastqBatch<'a>,
104    second: &'a FastqBatch<'a>,
105) -> Result<PairedRecords<'a>> {
106    first.paired_with(second)
107}
108
109/// A paired-end batch produced by [`PairedFastqReader`].
110///
111/// The batch contains the validated common prefix from the two underlying
112/// readers. If one reader yields more records than the other in a slab, the
113/// extra records are retained for the next call.
114#[derive(Debug)]
115pub struct PairedFastqBatch<'a> {
116    pub(crate) first_bytes: &'a [u8],
117    pub(crate) first_records: &'a [RecordRef],
118    pub(crate) second_bytes: &'a [u8],
119    pub(crate) second_records: &'a [RecordRef],
120}
121
122impl<'a> PairedFastqBatch<'a> {
123    /// Number of read pairs in the batch.
124    pub fn len(&self) -> usize {
125        self.first_records.len()
126    }
127
128    /// Whether this batch contains no read pairs.
129    pub fn is_empty(&self) -> bool {
130        self.first_records.is_empty()
131    }
132
133    /// Iterate borrowed read-pair views.
134    pub fn pairs(&'a self) -> PairedFastqPairs<'a> {
135        PairedFastqPairs {
136            batch: self,
137            next: 0,
138        }
139    }
140
141    fn first_record_at(&self, index: usize) -> FastqRecord<'a> {
142        FastqRecord {
143            bytes: self.first_bytes,
144            record: &self.first_records[index],
145        }
146    }
147
148    fn second_record_at(&self, index: usize) -> FastqRecord<'a> {
149        FastqRecord {
150            bytes: self.second_bytes,
151            record: &self.second_records[index],
152        }
153    }
154}
155
156/// Iterator over read pairs in a [`PairedFastqBatch`].
157#[derive(Debug)]
158pub struct PairedFastqPairs<'a> {
159    batch: &'a PairedFastqBatch<'a>,
160    next: usize,
161}
162
163impl<'a> Iterator for PairedFastqPairs<'a> {
164    type Item = FastqPair<'a>;
165
166    fn next(&mut self) -> Option<Self::Item> {
167        if self.next >= self.batch.first_records.len() {
168            return None;
169        }
170        let pair = FastqPair {
171            first: self.batch.first_record_at(self.next),
172            second: self.batch.second_record_at(self.next),
173        };
174        self.next += 1;
175        Some(pair)
176    }
177}
178
179/// Stateful reader for ordered separate-file paired-end FASTQ streams.
180///
181/// This reader validates matching records as it advances both streams. It does
182/// not reorder or synchronize mates that appear in different orders.
183#[derive(Debug)]
184pub struct PairedFastqReader<R1, R2> {
185    first: FastqReader<R1>,
186    second: FastqReader<R2>,
187    pair_validation: PairValidation,
188    first_pending_retain: Option<(usize, usize)>,
189    second_pending_retain: Option<(usize, usize)>,
190}
191
192impl<R1: Read, R2: Read> PairedFastqReader<R1, R2> {
193    /// Create a paired reader with default FASTQ configuration.
194    pub fn new(first: R1, second: R2) -> Self {
195        Self::with_config(first, second, FastqConfig::default())
196    }
197
198    /// Create a paired reader using the same configuration for both streams.
199    pub fn with_config(first: R1, second: R2, config: FastqConfig) -> Self {
200        Self::with_configs(
201            first,
202            FastqConfig {
203                pairing: PairingMode::None,
204                ..config.clone()
205            },
206            second,
207            FastqConfig {
208                pairing: PairingMode::None,
209                ..config
210            },
211        )
212    }
213
214    /// Create a paired reader with separate per-stream configurations.
215    pub fn with_configs(
216        first: R1,
217        first_config: FastqConfig,
218        second: R2,
219        second_config: FastqConfig,
220    ) -> Self {
221        let pair_validation = first_config.pair_validation;
222        Self {
223            first: FastqReader::with_config(
224                first,
225                FastqConfig {
226                    pairing: PairingMode::None,
227                    ..first_config
228                },
229            ),
230            second: FastqReader::with_config(
231                second,
232                FastqConfig {
233                    pairing: PairingMode::None,
234                    ..second_config
235                },
236            ),
237            pair_validation,
238            first_pending_retain: None,
239            second_pending_retain: None,
240        }
241    }
242
243    /// Create a paired reader from two existing [`FastqReader`] values.
244    pub fn from_fastq_readers(first: FastqReader<R1>, second: FastqReader<R2>) -> Self {
245        Self {
246            pair_validation: first.config.pair_validation,
247            first,
248            second,
249            first_pending_retain: None,
250            second_pending_retain: None,
251        }
252    }
253
254    /// Override the paired identifier validation policy.
255    pub fn with_pair_validation(mut self, pair_validation: PairValidation) -> Self {
256        self.pair_validation = pair_validation;
257        self
258    }
259
260    /// Read the next validated paired batch.
261    ///
262    /// Returns `Ok(None)` when both streams end together.
263    pub fn next_pair_batch(&mut self) -> Result<Option<PairedFastqBatch<'_>>> {
264        if let Some((next_start, record_count)) = self.first_pending_retain.take() {
265            self.first
266                .retain_records_from_parts(next_start, record_count);
267        }
268        if let Some((next_start, record_count)) = self.second_pending_retain.take() {
269            self.second
270                .retain_records_from_parts(next_start, record_count);
271        }
272
273        let first = self.first.next_batch()?;
274        let second = self.second.next_batch()?;
275
276        match (first, second) {
277            (None, None) => Ok(None),
278            (Some(first), None) => Err(extra_record_error(&first, 0)),
279            (None, Some(second)) => Err(extra_record_error(&second, 0)),
280            (Some(first), Some(second)) => {
281                let pair_count = first.len().min(second.len());
282                validate_pair_ids_prefix(&first, &second, pair_count, self.pair_validation)?;
283
284                self.first_pending_retain = retain_from(&first, pair_count);
285                self.second_pending_retain = retain_from(&second, pair_count);
286
287                Ok(Some(PairedFastqBatch {
288                    first_bytes: first.bytes,
289                    first_records: &first.records[..pair_count],
290                    second_bytes: second.bytes,
291                    second_records: &second.records[..pair_count],
292                }))
293            }
294        }
295    }
296}
297
298fn retain_from(batch: &FastqBatch<'_>, index: usize) -> Option<(usize, usize)> {
299    if index >= batch.records.len() {
300        return None;
301    }
302    Some((
303        batch.records[index].name.start as usize,
304        batch.records.len() - index,
305    ))
306}
307
308pub(crate) fn validate_even_pair_count(batch: &FastqBatch<'_>) -> Result<()> {
309    if batch.records.len().is_multiple_of(2) {
310        return Ok(());
311    }
312    let Some(last) = batch.records.last() else {
313        return Ok(());
314    };
315    Err(fastq_frame::format_at(
316        "interleaved FASTQ batch has an odd record count",
317        batch.base_offset,
318        last.name.start as usize,
319        batch.first_record_index + (batch.records.len() - 1) as u64,
320        0,
321    ))
322}
323
324pub(crate) fn validate_paired_batches(
325    first: &FastqBatch<'_>,
326    second: &FastqBatch<'_>,
327    pair_validation: PairValidation,
328) -> Result<()> {
329    if first.records.len() == second.records.len() {
330        return validate_pair_ids(first, second, pair_validation);
331    }
332
333    let (batch, index) = if first.records.len() > second.records.len() {
334        (first, second.records.len())
335    } else {
336        (second, first.records.len())
337    };
338    let record = &batch.records[index];
339    Err(fastq_frame::format_at(
340        "paired FASTQ batches have different record counts",
341        batch.base_offset,
342        record.name.start as usize,
343        batch.first_record_index + index as u64,
344        0,
345    ))
346}
347
348fn validate_pair_ids(
349    first: &FastqBatch<'_>,
350    second: &FastqBatch<'_>,
351    pair_validation: PairValidation,
352) -> Result<()> {
353    if pair_validation == PairValidation::None {
354        return Ok(());
355    }
356    for index in 0..first.records.len() {
357        let r1 = first.record_at(index);
358        let r2 = second.record_at(index);
359        if !pair_ids_match(r1, r2, pair_validation) {
360            return Err(pair_id_mismatch(second, index));
361        }
362    }
363    Ok(())
364}
365
366pub(crate) fn validate_interleaved_pair_ids(
367    batch: &FastqBatch<'_>,
368    pair_validation: PairValidation,
369) -> Result<()> {
370    if pair_validation == PairValidation::None {
371        return Ok(());
372    }
373    for index in (0..batch.records.len()).step_by(2) {
374        let r1 = batch.record_at(index);
375        let r2 = batch.record_at(index + 1);
376        if !pair_ids_match(r1, r2, pair_validation) {
377            return Err(pair_id_mismatch(batch, index + 1));
378        }
379    }
380    Ok(())
381}
382
383fn validate_pair_ids_prefix(
384    first: &FastqBatch<'_>,
385    second: &FastqBatch<'_>,
386    len: usize,
387    pair_validation: PairValidation,
388) -> Result<()> {
389    if pair_validation == PairValidation::None {
390        return Ok(());
391    }
392    for index in 0..len {
393        let r1 = first.record_at(index);
394        let r2 = second.record_at(index);
395        if !pair_ids_match(r1, r2, pair_validation) {
396            return Err(pair_id_mismatch(second, index));
397        }
398    }
399    Ok(())
400}
401
402fn pair_ids_match(first: FastqRecord<'_>, second: FastqRecord<'_>, mode: PairValidation) -> bool {
403    match mode {
404        PairValidation::None => true,
405        PairValidation::Full => first.pair_normalized_id() == second.pair_normalized_id(),
406        PairValidation::FastSlash => {
407            fastq_frame::fast_slash_pair_ids_match(first.name(), second.name())
408                .unwrap_or_else(|| first.pair_normalized_id() == second.pair_normalized_id())
409        }
410    }
411}
412
413fn pair_id_mismatch(batch: &FastqBatch<'_>, index: usize) -> FastqError {
414    let record = &batch.records[index];
415    fastq_frame::format_at(
416        "paired FASTQ record identifiers do not match",
417        batch.base_offset,
418        record.name.start as usize,
419        batch.first_record_index + index as u64,
420        0,
421    )
422}
423
424pub(crate) fn extra_record_error(batch: &FastqBatch<'_>, index: usize) -> FastqError {
425    let record = &batch.records[index];
426    fastq_frame::format_at(
427        "paired FASTQ inputs have different record counts",
428        batch.base_offset,
429        record.name.start as usize,
430        batch.first_record_index + index as u64,
431        0,
432    )
433}
dino_seq/fastq/pair.rs

dino_seq/fastq/
pair.rs
