dino_seq/
source.rs

1use std::fs::File;
2#[cfg(all(feature = "gzip", feature = "libdeflate"))]
3use std::io::Cursor;
4use std::io::Read;
5#[cfg(any(feature = "bgzf", feature = "gzip"))]
6use std::io::{Seek, SeekFrom};
7use std::path::Path;
8
9use crate::error::Result;
10use crate::fasta::{FastaConfig, FastaReader};
11use crate::fastq::{FastqConfig, FastqReader, PairedFastqReader};
12#[cfg(feature = "bgzf")]
13use crate::{
14    BgzfAutoReader, BgzfInflateBackend, BgzfParallelConfig, BgzfParallelReader, BgzfReader,
15    bgzf::is_bgzf_header,
16};
17
18#[cfg(all(feature = "gzip", feature = "libdeflate"))]
19const DEFAULT_LIBDEFLATE_GZIP_MAX_COMPRESSED_BYTES: usize = if usize::BITS >= 64 {
20    1024 * 1024 * 1024
21} else {
22    usize::MAX / 2
23};
24#[cfg(all(feature = "gzip", feature = "libdeflate"))]
25const DEFAULT_LIBDEFLATE_GZIP_MAX_DECOMPRESSED_BYTES: usize = if usize::BITS >= 64 {
26    1024 * 1024 * 1024
27} else {
28    usize::MAX / 2
29};
30
31#[cfg(all(feature = "gzip", feature = "libdeflate"))]
32/// Memory limits for explicit buffered libdeflate gzip openers.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34pub struct LibdeflateGzipLimits {
35    /// Maximum compressed bytes accepted before buffering.
36    pub max_compressed_bytes: usize,
37    /// Maximum decompressed bytes accepted before parsing.
38    pub max_decompressed_bytes: usize,
39}
40
41#[cfg(all(feature = "gzip", feature = "libdeflate"))]
42impl Default for LibdeflateGzipLimits {
43    fn default() -> Self {
44        Self {
45            max_compressed_bytes: DEFAULT_LIBDEFLATE_GZIP_MAX_COMPRESSED_BYTES,
46            max_decompressed_bytes: DEFAULT_LIBDEFLATE_GZIP_MAX_DECOMPRESSED_BYTES,
47        }
48    }
49}
50
51#[cfg(any(feature = "bgzf", feature = "gzip"))]
52#[derive(Debug, Clone, Copy, PartialEq, Eq)]
53enum InputKind {
54    Raw,
55    #[cfg(feature = "gzip")]
56    Gzip,
57    #[cfg(feature = "bgzf")]
58    Bgzf,
59}
60
61/// Compression/container detected from input file magic.
62#[derive(Debug, Clone, Copy, PartialEq, Eq)]
63pub enum DetectedInputKind {
64    /// Plain uncompressed input.
65    Raw,
66    /// Ordinary gzip input.
67    #[cfg(feature = "gzip")]
68    Gzip,
69    /// BGZF blocked gzip input.
70    #[cfg(feature = "bgzf")]
71    Bgzf,
72}
73
74#[cfg(any(feature = "bgzf", feature = "gzip"))]
75impl From<InputKind> for DetectedInputKind {
76    fn from(value: InputKind) -> Self {
77        match value {
78            InputKind::Raw => Self::Raw,
79            #[cfg(feature = "gzip")]
80            InputKind::Gzip => Self::Gzip,
81            #[cfg(feature = "bgzf")]
82            InputKind::Bgzf => Self::Bgzf,
83        }
84    }
85}
86
87/// Detect raw, gzip, or BGZF input by file magic.
88pub fn detect_file_input_kind(path: impl AsRef<Path>) -> Result<DetectedInputKind> {
89    #[cfg(any(feature = "bgzf", feature = "gzip"))]
90    {
91        let mut file = File::open(path)?;
92        detect_input_kind(&mut file).map(DetectedInputKind::from)
93    }
94    #[cfg(not(any(feature = "bgzf", feature = "gzip")))]
95    {
96        let _ = path;
97        Ok(DetectedInputKind::Raw)
98    }
99}
100
101/// Open a FASTQ file with default configuration.
102///
103/// With default features, this opens raw FASTQ. Enabling `gzip` and/or `bgzf`
104/// adds file-magic transport detection. BGZF is checked before ordinary gzip
105/// when both features are enabled.
106pub fn open_fastq(path: impl AsRef<Path>) -> Result<FastqReader<Box<dyn Read + Send>>> {
107    open_fastq_with_config(path, FastqConfig::default())
108}
109
110/// Open a FASTQ file with explicit parser configuration.
111///
112/// This is the convenience file-path API for callers that want boxed transport
113/// detection with custom validation, slab, or pairing settings. Use
114/// `FastqReader<File>` directly for raw-file hot paths.
115pub fn open_fastq_with_config(
116    path: impl AsRef<Path>,
117    config: FastqConfig,
118) -> Result<FastqReader<Box<dyn Read + Send>>> {
119    let reader = open_read_by_magic(path)?;
120    Ok(FastqReader::with_config(reader, config))
121}
122
123/// Open a FASTA file with default configuration.
124///
125/// With default features, this opens raw FASTA. Enabling `gzip` and/or `bgzf`
126/// adds file-magic transport detection. BGZF is checked before ordinary gzip
127/// when both features are enabled.
128pub fn open_fasta(path: impl AsRef<Path>) -> Result<FastaReader<Box<dyn Read + Send>>> {
129    open_fasta_with_config(path, FastaConfig::default())
130}
131
132/// Open a FASTA file with explicit parser configuration.
133///
134/// This is the convenience file-path API for callers that want boxed transport
135/// detection. Use `FastaReader<File>` directly for raw-file hot paths.
136pub fn open_fasta_with_config(
137    path: impl AsRef<Path>,
138    config: FastaConfig,
139) -> Result<FastaReader<Box<dyn Read + Send>>> {
140    let reader = open_read_by_magic(path)?;
141    Ok(FastaReader::with_config(reader, config))
142}
143
144/// Open a FASTA reference genome with parser settings tuned for long records.
145///
146/// This uses the same raw/gzip/BGZF transport detection as [`open_fasta`], but
147/// lowers records per batch and increases I/O buffering and sequence
148/// preallocation hints for chromosome-scale records.
149pub fn open_fasta_for_reference(
150    path: impl AsRef<Path>,
151) -> Result<FastaReader<Box<dyn Read + Send>>> {
152    open_fasta_with_config(path, FastaConfig::reference())
153}
154
155/// Open ordered R1/R2 FASTQ files with default configuration.
156pub fn open_paired_fastq(
157    first_path: impl AsRef<Path>,
158    second_path: impl AsRef<Path>,
159) -> Result<PairedFastqReader<Box<dyn Read + Send>, Box<dyn Read + Send>>> {
160    open_paired_fastq_with_config(first_path, second_path, FastqConfig::default())
161}
162
163/// Open ordered R1/R2 FASTQ files with the same configuration for both mates.
164pub fn open_paired_fastq_with_config(
165    first_path: impl AsRef<Path>,
166    second_path: impl AsRef<Path>,
167    config: FastqConfig,
168) -> Result<PairedFastqReader<Box<dyn Read + Send>, Box<dyn Read + Send>>> {
169    open_paired_fastq_with_configs(first_path, config.clone(), second_path, config)
170}
171
172/// Open ordered R1/R2 FASTQ files with separate mate configurations.
173pub fn open_paired_fastq_with_configs(
174    first_path: impl AsRef<Path>,
175    first_config: FastqConfig,
176    second_path: impl AsRef<Path>,
177    second_config: FastqConfig,
178) -> Result<PairedFastqReader<Box<dyn Read + Send>, Box<dyn Read + Send>>> {
179    let first = open_read_by_magic(first_path)?;
180    let second = open_read_by_magic(second_path)?;
181    Ok(PairedFastqReader::with_configs(
182        first,
183        first_config,
184        second,
185        second_config,
186    ))
187}
188
189fn open_read_by_magic(path: impl AsRef<Path>) -> Result<Box<dyn Read + Send>> {
190    #[cfg(any(feature = "bgzf", feature = "gzip"))]
191    let mut file = File::open(path)?;
192    #[cfg(not(any(feature = "bgzf", feature = "gzip")))]
193    let file = File::open(path)?;
194
195    #[cfg(any(feature = "bgzf", feature = "gzip"))]
196    {
197        let kind = detect_input_kind(&mut file)?;
198        file.seek(SeekFrom::Start(0))?;
199
200        match kind {
201            #[cfg(feature = "bgzf")]
202            InputKind::Bgzf => {
203                let compressed_len = file.metadata()?.len();
204                let config = BgzfParallelConfig::new(default_bgzf_workers());
205                return Ok(Box::new(BgzfAutoReader::with_config(
206                    file,
207                    compressed_len,
208                    config,
209                )?));
210            }
211            #[cfg(feature = "gzip")]
212            InputKind::Gzip => return Ok(Box::new(flate2::read::MultiGzDecoder::new(file))),
213            InputKind::Raw => {}
214        }
215    }
216
217    let reader: Box<dyn Read + Send> = Box::new(file);
218    Ok(reader)
219}
220
221#[cfg(any(feature = "bgzf", feature = "gzip"))]
222fn detect_input_kind(file: &mut File) -> Result<InputKind> {
223    let mut prefix = [0_u8; 18];
224    let n = file.read(&mut prefix)?;
225    #[cfg(feature = "bgzf")]
226    if is_bgzf_header(&prefix[..n]) {
227        return Ok(InputKind::Bgzf);
228    }
229    #[cfg(feature = "gzip")]
230    if n >= 2 && prefix[..2] == [0x1f, 0x8b] {
231        return Ok(InputKind::Gzip);
232    }
233    Ok(InputKind::Raw)
234}
235
236#[cfg(feature = "bgzf")]
237fn default_bgzf_workers() -> usize {
238    std::thread::available_parallelism().map_or(1, usize::from)
239}
240
241#[cfg(all(feature = "gzip", feature = "libdeflate"))]
242/// Open an ordinary gzip FASTQ file through a buffered libdeflate path.
243///
244/// Unlike [`open_fastq`], this buffers the fully decompressed input before
245/// parsing. It accepts a single gzip member and applies default memory limits;
246/// use [`open_fastq_gzip_libdeflate_with_limits`] to set tighter bounds.
247pub fn open_fastq_gzip_libdeflate(path: impl AsRef<Path>) -> Result<FastqReader<Cursor<Vec<u8>>>> {
248    open_fastq_gzip_libdeflate_with_config(path, FastqConfig::default())
249}
250
251#[cfg(all(feature = "gzip", feature = "libdeflate"))]
252/// Open an ordinary gzip FASTQ file through buffered libdeflate with parser
253/// configuration.
254pub fn open_fastq_gzip_libdeflate_with_config(
255    path: impl AsRef<Path>,
256    config: FastqConfig,
257) -> Result<FastqReader<Cursor<Vec<u8>>>> {
258    open_fastq_gzip_libdeflate_with_limits(path, config, LibdeflateGzipLimits::default())
259}
260
261#[cfg(all(feature = "gzip", feature = "libdeflate"))]
262/// Open an ordinary single-member gzip FASTQ file through buffered libdeflate
263/// with explicit parser configuration and memory limits.
264pub fn open_fastq_gzip_libdeflate_with_limits(
265    path: impl AsRef<Path>,
266    config: FastqConfig,
267    limits: LibdeflateGzipLimits,
268) -> Result<FastqReader<Cursor<Vec<u8>>>> {
269    let compressed = read_limited(path, limits.max_compressed_bytes)?;
270    let decoded = decompress_gzip_libdeflate_buffered(&compressed, limits)?;
271    Ok(FastqReader::with_config(Cursor::new(decoded), config))
272}
273
274#[cfg(all(feature = "gzip", feature = "libdeflate"))]
275/// Open an ordinary gzip FASTA file through a buffered libdeflate path.
276///
277/// Unlike [`open_fasta`], this buffers the fully decompressed input before
278/// parsing. It accepts a single gzip member and applies default memory limits;
279/// use [`open_fasta_gzip_libdeflate_with_limits`] to set tighter bounds.
280pub fn open_fasta_gzip_libdeflate(path: impl AsRef<Path>) -> Result<FastaReader<Cursor<Vec<u8>>>> {
281    open_fasta_gzip_libdeflate_with_config(path, FastaConfig::default())
282}
283
284#[cfg(all(feature = "gzip", feature = "libdeflate"))]
285/// Open an ordinary gzip FASTA file through buffered libdeflate with parser
286/// configuration.
287pub fn open_fasta_gzip_libdeflate_with_config(
288    path: impl AsRef<Path>,
289    config: FastaConfig,
290) -> Result<FastaReader<Cursor<Vec<u8>>>> {
291    open_fasta_gzip_libdeflate_with_limits(path, config, LibdeflateGzipLimits::default())
292}
293
294#[cfg(all(feature = "gzip", feature = "libdeflate"))]
295/// Open an ordinary single-member gzip FASTA file through buffered libdeflate
296/// with explicit parser configuration and memory limits.
297pub fn open_fasta_gzip_libdeflate_with_limits(
298    path: impl AsRef<Path>,
299    config: FastaConfig,
300    limits: LibdeflateGzipLimits,
301) -> Result<FastaReader<Cursor<Vec<u8>>>> {
302    let compressed = read_limited(path, limits.max_compressed_bytes)?;
303    let decoded = decompress_gzip_libdeflate_buffered(&compressed, limits)?;
304    Ok(FastaReader::with_config(Cursor::new(decoded), config))
305}
306
307#[cfg(all(feature = "gzip", feature = "libdeflate"))]
308fn read_limited(path: impl AsRef<Path>, max_bytes: usize) -> Result<Vec<u8>> {
309    let path = path.as_ref();
310    let len = std::fs::metadata(path)?.len();
311    if len > max_bytes as u64 {
312        return Err(crate::FastqError::Format(format!(
313            "gzip input exceeds libdeflate compressed limit ({len} > {max_bytes} bytes)"
314        )));
315    }
316    let mut compressed = Vec::with_capacity(usize::try_from(len).unwrap_or(0));
317    File::open(path)?.read_to_end(&mut compressed)?;
318    if compressed.len() > max_bytes {
319        return Err(crate::FastqError::Format(format!(
320            "gzip input exceeds libdeflate compressed limit ({} > {max_bytes} bytes)",
321            compressed.len()
322        )));
323    }
324    Ok(compressed)
325}
326
327#[cfg(all(feature = "gzip", feature = "libdeflate"))]
328fn decompress_gzip_libdeflate_buffered(
329    compressed: &[u8],
330    limits: LibdeflateGzipLimits,
331) -> Result<Vec<u8>> {
332    if limits.max_decompressed_bytes == 0 {
333        return Err(crate::FastqError::Format(
334            "libdeflate decompressed limit must be greater than zero".into(),
335        ));
336    }
337    ensure_single_gzip_member(compressed)?;
338    let initial = initial_gzip_output_capacity(compressed).min(limits.max_decompressed_bytes);
339    let mut out = vec![0_u8; initial.max(1)];
340    let mut decompressor = libdeflater::Decompressor::new();
341    loop {
342        match decompressor.gzip_decompress(compressed, &mut out) {
343            Ok(n) => {
344                out.truncate(n);
345                return Ok(out);
346            }
347            Err(libdeflater::DecompressionError::InsufficientSpace) => {
348                let next = out.len().checked_mul(2).ok_or_else(|| {
349                    crate::FastqError::Format("gzip output size exceeds usize range".into())
350                })?;
351                if next > limits.max_decompressed_bytes {
352                    return Err(crate::FastqError::Format(format!(
353                        "gzip output exceeds libdeflate decompressed limit (>{} bytes)",
354                        limits.max_decompressed_bytes
355                    )));
356                }
357                out.resize(next.max(1), 0);
358            }
359            Err(err) => {
360                return Err(crate::FastqError::Format(format!(
361                    "libdeflate gzip inflate failed: {err}"
362                )));
363            }
364        }
365    }
366}
367
368#[cfg(all(feature = "gzip", feature = "libdeflate"))]
369fn ensure_single_gzip_member(compressed: &[u8]) -> Result<()> {
370    let end = gzip_first_member_end(compressed)?;
371    if end != compressed.len() {
372        return Err(crate::FastqError::Format(
373            "libdeflate gzip opener accepts exactly one gzip member; use open_fastq/open_fasta for concatenated gzip streams".into(),
374        ));
375    }
376    Ok(())
377}
378
379#[cfg(all(feature = "gzip", feature = "libdeflate"))]
380fn gzip_first_member_end(compressed: &[u8]) -> Result<usize> {
381    use flate2::{Decompress, FlushDecompress, Status};
382
383    if compressed.len() < 18 || compressed[..2] != [0x1f, 0x8b] || compressed[2] != 8 {
384        return Err(crate::FastqError::Format("invalid gzip header".into()));
385    }
386    let flags = compressed[3];
387    if flags & 0xe0 != 0 {
388        return Err(crate::FastqError::Format(
389            "gzip header uses reserved flags".into(),
390        ));
391    }
392
393    let mut pos = 10;
394    if flags & 0x04 != 0 {
395        let xlen = read_gzip_u16(compressed, pos)? as usize;
396        pos = pos
397            .checked_add(2)
398            .and_then(|p| p.checked_add(xlen))
399            .ok_or_else(|| crate::FastqError::Format("gzip extra field is too large".into()))?;
400        if pos > compressed.len() {
401            return Err(crate::FastqError::Format(
402                "truncated gzip extra field".into(),
403            ));
404        }
405    }
406    if flags & 0x08 != 0 {
407        pos = scan_gzip_cstring(compressed, pos, "name")?;
408    }
409    if flags & 0x10 != 0 {
410        pos = scan_gzip_cstring(compressed, pos, "comment")?;
411    }
412    if flags & 0x02 != 0 {
413        pos = pos
414            .checked_add(2)
415            .ok_or_else(|| crate::FastqError::Format("gzip header CRC is too large".into()))?;
416        if pos > compressed.len() {
417            return Err(crate::FastqError::Format(
418                "truncated gzip header CRC".into(),
419            ));
420        }
421    }
422    if pos + 8 > compressed.len() {
423        return Err(crate::FastqError::Format("truncated gzip member".into()));
424    }
425
426    let deflate = &compressed[pos..];
427    let mut decoder = Decompress::new(false);
428    let mut scratch = [0_u8; 8192];
429    loop {
430        let consumed_before = decoder.total_in();
431        let produced_before = decoder.total_out();
432        let consumed = usize::try_from(consumed_before)
433            .map_err(|_| crate::FastqError::Format("gzip member exceeds usize range".into()))?;
434        let status = decoder
435            .decompress(
436                deflate.get(consumed..).ok_or_else(|| {
437                    crate::FastqError::Format("truncated gzip deflate stream".into())
438                })?,
439                &mut scratch,
440                FlushDecompress::None,
441            )
442            .map_err(|err| {
443                crate::FastqError::Format(format!("gzip deflate parse failed: {err}"))
444            })?;
445        if status == Status::StreamEnd {
446            let deflate_len = usize::try_from(decoder.total_in()).map_err(|_| {
447                crate::FastqError::Format("gzip deflate stream exceeds usize range".into())
448            })?;
449            let end = pos
450                .checked_add(deflate_len)
451                .and_then(|p| p.checked_add(8))
452                .ok_or_else(|| crate::FastqError::Format("gzip member is too large".into()))?;
453            if end > compressed.len() {
454                return Err(crate::FastqError::Format("truncated gzip trailer".into()));
455            }
456            return Ok(end);
457        }
458        if decoder.total_in() == consumed_before && decoder.total_out() == produced_before {
459            return Err(crate::FastqError::Format(
460                "gzip deflate parser made no progress".into(),
461            ));
462        }
463    }
464}
465
466#[cfg(all(feature = "gzip", feature = "libdeflate"))]
467fn read_gzip_u16(bytes: &[u8], pos: usize) -> Result<u16> {
468    let Some(pair) = bytes.get(pos..pos + 2) else {
469        return Err(crate::FastqError::Format("truncated gzip header".into()));
470    };
471    Ok(u16::from_le_bytes([pair[0], pair[1]]))
472}
473
474#[cfg(all(feature = "gzip", feature = "libdeflate"))]
475fn scan_gzip_cstring(bytes: &[u8], pos: usize, field: &str) -> Result<usize> {
476    let Some(relative_end) = bytes[pos..].iter().position(|&b| b == 0) else {
477        return Err(crate::FastqError::Format(format!(
478            "unterminated gzip {field} field"
479        )));
480    };
481    Ok(pos + relative_end + 1)
482}
483
484#[cfg(all(feature = "gzip", feature = "libdeflate"))]
485fn initial_gzip_output_capacity(compressed: &[u8]) -> usize {
486    let isize = compressed
487        .len()
488        .checked_sub(4)
489        .and_then(|start| compressed.get(start..))
490        .map(|tail| u32::from_le_bytes([tail[0], tail[1], tail[2], tail[3]]) as usize)
491        .unwrap_or(0);
492    isize.max(compressed.len().saturating_mul(2)).max(1024)
493}
494
495#[cfg(feature = "bgzf")]
496/// Open a BGZF FASTQ file with an explicit parallel worker count.
497pub fn open_fastq_bgzf_parallel(
498    path: impl AsRef<Path>,
499    workers: usize,
500) -> Result<FastqReader<BgzfParallelReader>> {
501    open_fastq_bgzf_parallel_with_config(path, workers, FastqConfig::default())
502}
503
504#[cfg(feature = "bgzf")]
505/// Open a BGZF FASTQ file with an explicit parallel worker count and parser
506/// configuration.
507pub fn open_fastq_bgzf_parallel_with_config(
508    path: impl AsRef<Path>,
509    workers: usize,
510    config: FastqConfig,
511) -> Result<FastqReader<BgzfParallelReader>> {
512    open_fastq_bgzf_parallel_with_backend(path, workers, BgzfInflateBackend::default(), config)
513}
514
515#[cfg(feature = "bgzf")]
516/// Open a BGZF FASTQ file with explicit BGZF and FASTQ configuration.
517pub fn open_fastq_bgzf_parallel_with_options(
518    path: impl AsRef<Path>,
519    bgzf_config: BgzfParallelConfig,
520    fastq_config: FastqConfig,
521) -> Result<FastqReader<BgzfParallelReader>> {
522    let file = File::open(path)?;
523    Ok(FastqReader::with_config(
524        BgzfParallelReader::with_config(file, bgzf_config)?,
525        fastq_config,
526    ))
527}
528
529#[cfg(feature = "bgzf")]
530/// Open a BGZF FASTQ file with adaptive serial/parallel reading.
531///
532/// The supplied [`BgzfParallelConfig`] controls the parallelization threshold,
533/// backend, workers, queue depths, and optional metrics.
534pub fn open_fastq_bgzf_adaptive(
535    path: impl AsRef<Path>,
536    bgzf_config: BgzfParallelConfig,
537    fastq_config: FastqConfig,
538) -> Result<FastqReader<BgzfAutoReader<File>>> {
539    let path = path.as_ref();
540    let compressed_len = std::fs::metadata(path)?.len();
541    let file = File::open(path)?;
542    Ok(FastqReader::with_config(
543        BgzfAutoReader::with_config(file, compressed_len, bgzf_config)?,
544        fastq_config,
545    ))
546}
547
548#[cfg(feature = "bgzf")]
549/// Open a BGZF FASTQ file through the flate2 serial backend.
550pub fn open_fastq_bgzf_flate2(path: impl AsRef<Path>) -> Result<FastqReader<BgzfReader<File>>> {
551    open_fastq_bgzf_with_backend(path, BgzfInflateBackend::Flate2, FastqConfig::default())
552}
553
554#[cfg(feature = "bgzf")]
555/// Open a BGZF FASTQ file with a selected serial inflate backend.
556pub fn open_fastq_bgzf_with_backend(
557    path: impl AsRef<Path>,
558    backend: BgzfInflateBackend,
559    config: FastqConfig,
560) -> Result<FastqReader<BgzfReader<File>>> {
561    let file = File::open(path)?;
562    Ok(FastqReader::with_config(
563        BgzfReader::with_inflate_backend(file, backend),
564        config,
565    ))
566}
567
568#[cfg(feature = "bgzf")]
569/// Open a BGZF FASTQ file with a selected parallel inflate backend.
570pub fn open_fastq_bgzf_parallel_with_backend(
571    path: impl AsRef<Path>,
572    workers: usize,
573    backend: BgzfInflateBackend,
574    config: FastqConfig,
575) -> Result<FastqReader<BgzfParallelReader>> {
576    let file = File::open(path)?;
577    Ok(FastqReader::with_config(
578        BgzfParallelReader::with_inflate_backend(file, workers, backend)?,
579        config,
580    ))
581}
582
583#[cfg(all(test, feature = "gzip"))]
584mod tests {
585    use std::io::Write;
586
587    use super::*;
588
589    #[cfg(feature = "libdeflate")]
590    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
591    struct StreamStats {
592        records: u64,
593        bases: u64,
594        qualities: u64,
595        name_bytes: u64,
596        checksum: u64,
597    }
598
599    #[cfg(feature = "libdeflate")]
600    impl StreamStats {
601        fn observe_record(&mut self, name: &[u8], seq: &[u8], qual: &[u8]) {
602            self.records += 1;
603            self.bases += seq.len() as u64;
604            self.qualities += qual.len() as u64;
605            self.name_bytes += name.len() as u64;
606            self.checksum = self
607                .checksum
608                .wrapping_add(seq.first().copied().unwrap_or_default() as u64)
609                .wrapping_mul(1_099_511_628_211)
610                .wrapping_add(seq.len() as u64);
611        }
612    }
613
614    #[cfg(feature = "libdeflate")]
615    fn consume_fastq<R: Read>(reader: &mut FastqReader<R>) -> Result<StreamStats> {
616        let mut stats = StreamStats::default();
617        while let Some(batch) = reader.next_batch()? {
618            for record in batch.records() {
619                stats.observe_record(record.name(), record.seq(), record.qual());
620            }
621        }
622        Ok(stats)
623    }
624
625    #[cfg(feature = "libdeflate")]
626    fn consume_fasta<R: Read>(reader: &mut FastaReader<R>) -> Result<StreamStats> {
627        let mut stats = StreamStats::default();
628        while let Some(batch) = reader.next_batch()? {
629            for record in batch.records() {
630                stats.observe_record(record.name(), record.seq(), b"");
631            }
632        }
633        Ok(stats)
634    }
635
636    #[cfg(all(feature = "gzip", feature = "libdeflate"))]
637    fn gzip_member(payload: &[u8]) -> Vec<u8> {
638        let mut out = Vec::new();
639        {
640            let mut encoder = flate2::write::GzEncoder::new(&mut out, flate2::Compression::fast());
641            encoder.write_all(payload).unwrap();
642            encoder.finish().unwrap();
643        }
644        out
645    }
646
647    #[test]
648    #[cfg(feature = "gzip")]
649    fn opens_gzip_by_magic() {
650        let dir = std::env::temp_dir();
651        let path = dir.join(format!("dino_seq-{}.fq.gz", std::process::id()));
652        let file = File::create(&path).unwrap();
653        let mut encoder = flate2::write::GzEncoder::new(file, flate2::Compression::fast());
654        encoder.write_all(b"@r1\nACGT\n+\nIIII\n").unwrap();
655        encoder.finish().unwrap();
656
657        let mut reader = open_fastq(&path).unwrap();
658        let batch = reader.next_batch().unwrap().unwrap();
659        let rec = batch.records().next().unwrap();
660        assert_eq!(rec.seq(), b"ACGT");
661
662        std::fs::remove_file(path).unwrap();
663    }
664
665    #[test]
666    #[cfg(feature = "gzip")]
667    fn opens_gzip_fasta_by_magic() {
668        let dir = std::env::temp_dir();
669        let path = dir.join(format!("dino_seq-{}.fa.gz", std::process::id()));
670        let file = File::create(&path).unwrap();
671        let mut encoder = flate2::write::GzEncoder::new(file, flate2::Compression::fast());
672        encoder.write_all(b">seq1\nAC\nGT\n").unwrap();
673        encoder.finish().unwrap();
674
675        let mut reader = open_fasta(&path).unwrap();
676        let batch = reader.next_batch().unwrap().unwrap();
677        let rec = batch.records().next().unwrap();
678        assert_eq!(rec.name_without_gt(), b"seq1");
679        assert_eq!(rec.seq(), b"ACGT");
680
681        std::fs::remove_file(path).unwrap();
682    }
683
684    #[test]
685    #[cfg(feature = "bgzf")]
686    fn opens_bgzf_by_magic() {
687        let dir = std::env::temp_dir();
688        let path = dir.join(format!("dino_seq-{}.bgz", std::process::id()));
689        let mut writer = crate::BgzfWriter::new(File::create(&path).unwrap());
690        writer.write_all(b"@r1\nACGT\n+\nIIII\n").unwrap();
691        writer.finish().unwrap();
692
693        let mut reader = open_fastq(&path).unwrap();
694        let batch = reader.next_batch().unwrap().unwrap();
695        let rec = batch.records().next().unwrap();
696        assert_eq!(rec.seq(), b"ACGT");
697
698        let mut parallel = open_fastq_bgzf_parallel(&path, 2).unwrap();
699        let batch = parallel.next_batch().unwrap().unwrap();
700        let rec = batch.records().next().unwrap();
701        assert_eq!(rec.seq(), b"ACGT");
702
703        std::fs::remove_file(path).unwrap();
704    }
705
706    #[test]
707    #[cfg(feature = "bgzf")]
708    fn opens_bgzf_fasta_by_magic() {
709        let dir = std::env::temp_dir();
710        let path = dir.join(format!("dino_seq-fasta-{}.bgz", std::process::id()));
711        let mut writer = crate::BgzfWriter::new(File::create(&path).unwrap());
712        writer.write_all(b">seq1\nAC\nGT\n").unwrap();
713        writer.finish().unwrap();
714
715        let mut reader = open_fasta(&path).unwrap();
716        let batch = reader.next_batch().unwrap().unwrap();
717        let rec = batch.records().next().unwrap();
718        assert_eq!(rec.name_without_gt(), b"seq1");
719        assert_eq!(rec.seq(), b"ACGT");
720
721        std::fs::remove_file(path).unwrap();
722    }
723
724    #[test]
725    #[cfg(feature = "bgzf")]
726    fn public_bgzf_adaptive_opener_selects_serial_and_parallel() {
727        let dir = std::env::temp_dir();
728        let small_path = dir.join(format!("dino_seq-bgzf-small-{}.bgz", std::process::id()));
729        let large_path = dir.join(format!("dino_seq-bgzf-large-{}.bgz", std::process::id()));
730
731        let mut small_writer = crate::BgzfWriter::new(File::create(&small_path).unwrap());
732        small_writer.write_all(b"@r1\nACGT\n+\nIIII\n").unwrap();
733        small_writer.finish().unwrap();
734
735        let mut large_writer = crate::BgzfWriter::new(File::create(&large_path).unwrap());
736        for i in 0..2048 {
737            writeln!(large_writer, "@r{i}\nACGTACGTACGTACGT\n+\nIIIIIIIIIIIIIIII").unwrap();
738        }
739        large_writer.finish().unwrap();
740
741        let serial = open_fastq_bgzf_adaptive(
742            &small_path,
743            BgzfParallelConfig::new(2).with_parallel_min_compressed_bytes(u64::MAX),
744            FastqConfig::default(),
745        )
746        .unwrap()
747        .into_inner();
748        assert!(matches!(serial, crate::BgzfAutoReader::Serial(_)));
749
750        let parallel = open_fastq_bgzf_adaptive(
751            &large_path,
752            BgzfParallelConfig::new(2).with_parallel_min_compressed_bytes(0),
753            FastqConfig::default(),
754        )
755        .unwrap()
756        .into_inner();
757        assert!(matches!(parallel, crate::BgzfAutoReader::Parallel(_)));
758
759        std::fs::remove_file(small_path).unwrap();
760        std::fs::remove_file(large_path).unwrap();
761    }
762
763    #[test]
764    #[cfg(all(feature = "bgzf", feature = "libdeflate"))]
765    fn explicit_bgzf_backend_openers_parse_same_file() {
766        let dir = std::env::temp_dir();
767        let path = dir.join(format!("dino_seq-backend-{}.bgz", std::process::id()));
768        let mut writer = crate::BgzfWriter::new(File::create(&path).unwrap());
769        writer
770            .write_all(b"@r1\nACGT\n+\nIIII\n@r2\nTGCA\n+\nJJJJ\n")
771            .unwrap();
772        writer.finish().unwrap();
773
774        let mut auto = open_fastq(&path).unwrap();
775        let mut flate2 = open_fastq_bgzf_flate2(&path).unwrap();
776        let mut libdeflate = open_fastq_bgzf_with_backend(
777            &path,
778            BgzfInflateBackend::Libdeflate,
779            FastqConfig::default(),
780        )
781        .unwrap();
782
783        let auto_stats = consume_fastq(&mut auto).unwrap();
784        let flate2_stats = consume_fastq(&mut flate2).unwrap();
785        let libdeflate_stats = consume_fastq(&mut libdeflate).unwrap();
786
787        assert_eq!(auto_stats, flate2_stats);
788        assert_eq!(auto_stats, libdeflate_stats);
789
790        std::fs::remove_file(path).unwrap();
791    }
792
793    #[test]
794    #[cfg(all(feature = "gzip", feature = "libdeflate"))]
795    fn explicit_libdeflate_gzip_opener_parses_same_file() {
796        let dir = std::env::temp_dir();
797        let path = dir.join(format!(
798            "dino_seq-libdeflate-gzip-{}.fq.gz",
799            std::process::id()
800        ));
801        let file = File::create(&path).unwrap();
802        let mut encoder = flate2::write::GzEncoder::new(file, flate2::Compression::fast());
803        encoder
804            .write_all(b"@r1\nACGT\n+\nIIII\n@r2\nTGCA\n+\nJJJJ\n")
805            .unwrap();
806        encoder.finish().unwrap();
807
808        let mut flate2 = open_fastq(&path).unwrap();
809        let mut libdeflate = open_fastq_gzip_libdeflate(&path).unwrap();
810
811        let flate2_stats = consume_fastq(&mut flate2).unwrap();
812        let libdeflate_stats = consume_fastq(&mut libdeflate).unwrap();
813        assert_eq!(flate2_stats, libdeflate_stats);
814
815        std::fs::remove_file(path).unwrap();
816    }
817
818    #[test]
819    #[cfg(all(feature = "gzip", feature = "libdeflate"))]
820    fn explicit_libdeflate_gzip_rejects_concatenated_fastq_members() {
821        let dir = std::env::temp_dir();
822        let path = dir.join(format!(
823            "dino_seq-libdeflate-gzip-multi-{}.fq.gz",
824            std::process::id()
825        ));
826        let mut encoded = gzip_member(b"@r1\nACGT\n+\nIIII\n");
827        encoded.extend_from_slice(&gzip_member(b"@r2\nTGCA\n+\nJJJJ\n"));
828        std::fs::write(&path, encoded).unwrap();
829
830        let mut default_reader = open_fastq(&path).unwrap();
831        let default_stats = consume_fastq(&mut default_reader).unwrap();
832        assert_eq!(default_stats.records, 2);
833
834        let err = match open_fastq_gzip_libdeflate(&path) {
835            Ok(_) => panic!("concatenated gzip member was accepted"),
836            Err(err) => err,
837        };
838        assert!(err.to_string().contains("one gzip member"));
839
840        std::fs::remove_file(path).unwrap();
841    }
842
843    #[test]
844    #[cfg(all(feature = "gzip", feature = "libdeflate"))]
845    fn explicit_libdeflate_gzip_limits_compressed_input() {
846        let dir = std::env::temp_dir();
847        let path = dir.join(format!(
848            "dino_seq-libdeflate-gzip-limit-{}.fq.gz",
849            std::process::id()
850        ));
851        std::fs::write(&path, gzip_member(b"@r1\nACGT\n+\nIIII\n")).unwrap();
852
853        let err = match open_fastq_gzip_libdeflate_with_limits(
854            &path,
855            FastqConfig::default(),
856            LibdeflateGzipLimits {
857                max_compressed_bytes: 1,
858                max_decompressed_bytes: 1024,
859            },
860        ) {
861            Ok(_) => panic!("compressed limit was not enforced"),
862            Err(err) => err,
863        };
864        assert!(err.to_string().contains("compressed limit"));
865
866        std::fs::remove_file(path).unwrap();
867    }
868
869    #[test]
870    #[cfg(all(feature = "gzip", feature = "libdeflate"))]
871    fn explicit_libdeflate_gzip_limits_decompressed_output() {
872        let dir = std::env::temp_dir();
873        let path = dir.join(format!(
874            "dino_seq-libdeflate-gzip-output-limit-{}.fq.gz",
875            std::process::id()
876        ));
877        std::fs::write(&path, gzip_member(b"@r1\nACGT\n+\nIIII\n")).unwrap();
878
879        let err = match open_fastq_gzip_libdeflate_with_limits(
880            &path,
881            FastqConfig::default(),
882            LibdeflateGzipLimits {
883                max_compressed_bytes: 1024,
884                max_decompressed_bytes: 4,
885            },
886        ) {
887            Ok(_) => panic!("decompressed limit was not enforced"),
888            Err(err) => err,
889        };
890        assert!(err.to_string().contains("decompressed limit"));
891
892        std::fs::remove_file(path).unwrap();
893    }
894
895    #[test]
896    #[cfg(all(feature = "gzip", feature = "libdeflate"))]
897    fn explicit_libdeflate_gzip_fasta_opener_parses_same_file() {
898        let dir = std::env::temp_dir();
899        let path = dir.join(format!(
900            "dino_seq-libdeflate-gzip-fasta-{}.fa.gz",
901            std::process::id()
902        ));
903        let file = File::create(&path).unwrap();
904        let mut encoder = flate2::write::GzEncoder::new(file, flate2::Compression::fast());
905        encoder.write_all(b">seq1\nAC\nGT\n").unwrap();
906        encoder.finish().unwrap();
907
908        let mut flate2 = open_fasta(&path).unwrap();
909        let mut libdeflate = open_fasta_gzip_libdeflate(&path).unwrap();
910
911        let flate2_stats = consume_fasta(&mut flate2).unwrap();
912        let libdeflate_stats = consume_fasta(&mut libdeflate).unwrap();
913        assert_eq!(flate2_stats, libdeflate_stats);
914
915        std::fs::remove_file(path).unwrap();
916    }
917
918    #[test]
919    #[cfg(all(feature = "gzip", feature = "libdeflate"))]
920    fn explicit_libdeflate_gzip_rejects_concatenated_fasta_members() {
921        let dir = std::env::temp_dir();
922        let path = dir.join(format!(
923            "dino_seq-libdeflate-gzip-fasta-multi-{}.fa.gz",
924            std::process::id()
925        ));
926        let mut encoded = gzip_member(b">seq1\nAC\n");
927        encoded.extend_from_slice(&gzip_member(b">seq2\nGT\n"));
928        std::fs::write(&path, encoded).unwrap();
929
930        let mut default_reader = open_fasta(&path).unwrap();
931        let default_stats = consume_fasta(&mut default_reader).unwrap();
932        assert_eq!(default_stats.records, 2);
933
934        let err = match open_fasta_gzip_libdeflate(&path) {
935            Ok(_) => panic!("concatenated gzip member was accepted"),
936            Err(err) => err,
937        };
938        assert!(err.to_string().contains("one gzip member"));
939
940        std::fs::remove_file(path).unwrap();
941    }
942
943    #[test]
944    fn paired_openers_parse_matching_files() {
945        let dir = std::env::temp_dir();
946        let r1_path = dir.join(format!("dino_seq-r1-{}.fq", std::process::id()));
947        let r2_path = dir.join(format!("dino_seq-r2-{}.fq", std::process::id()));
948        std::fs::write(&r1_path, b"@frag/1\nACGT\n+\nIIII\n").unwrap();
949        std::fs::write(&r2_path, b"@frag/2\nTGCA\n+\nJJJJ\n").unwrap();
950
951        let mut reader = open_paired_fastq(&r1_path, &r2_path).unwrap();
952        let batch = reader.next_pair_batch().unwrap().unwrap();
953        let pair = batch.pairs().next().unwrap();
954        assert_eq!(pair.pair_id(), b"frag");
955        assert_eq!(pair.first().seq(), b"ACGT");
956        assert_eq!(pair.second().seq(), b"TGCA");
957
958        std::fs::remove_file(r1_path).unwrap();
959        std::fs::remove_file(r2_path).unwrap();
960    }
961
962    #[test]
963    fn open_fasta_parses_raw_file() {
964        let dir = std::env::temp_dir();
965        let path = dir.join(format!("dino_seq-{}.fa", std::process::id()));
966        std::fs::write(&path, b">seq1\nAC\nGT\n").unwrap();
967
968        let mut reader = open_fasta(&path).unwrap();
969        let batch = reader.next_batch().unwrap().unwrap();
970        let rec = batch.records().next().unwrap();
971        assert_eq!(rec.id_token(), b"seq1");
972        assert_eq!(rec.seq(), b"ACGT");
973
974        std::fs::remove_file(path).unwrap();
975    }
976}
dino_seq/source.rs

dino_seq/
source.rs
