feat(fastq): UMI-in-read-name output and BAM→FASTQ on the typed-step chain

Cargo.toml

@@ -148,6 +148,8 @@ rstest = "0"
 proptest = "1.10"
 criterion = { version = "0.8", features = ["html_reports"] }
 fgumi-raw-bam = { workspace = true, features = ["test-utils", "noodles"] }
+fgumi-tag = { workspace = true }
+flate2 = { version = "1.1", features = ["zlib-rs"] }
 noodles = { version = "0.111.0", features = ["bam", "sam"] }
 trybuild = "1.0"
 

crates/fgumi-bam-io/src/grouping.rs

@@ -282,6 +282,14 @@ impl DecodedRecord {
         self.data.as_ref()
     }
 
+    /// Immutable access to the underlying [`RawRecord`], for read-only
+    /// consumers (e.g. the FASTQ-encode step) that need the typed accessors
+    /// (`flags()`, sequence, qualities, tags) rather than the raw byte slice.
+    #[must_use]
+    pub fn record(&self) -> &RawRecord {
+        &self.data
+    }
+
     /// Mutable access to the underlying [`RawRecord`]. Used by mid-chain
     /// `Parallel` Process steps that need to mutate record bytes (e.g.,
     /// MQ bumping, tag rewriting) without rebuilding the `DecodedRecord`

crates/fgumi-bam-io/src/lib.rs

@@ -40,6 +40,6 @@ pub use reader::{
 pub use reorder::{DrainReady, ReorderBuffer};
 pub use writer::{
     BamWriter, BgzfWriterEnum, RawBamWriter, bai_sidecar_path, create_bam_writer,
-    create_optional_bam_writer, create_raw_bam_writer, write_bai_index, write_bai_sidecar,
-    write_bam_header,
+    create_bgzf_writer, create_optional_bam_writer, create_raw_bam_writer, write_bai_index,
+    write_bai_sidecar, write_bam_header,
 };

crates/fgumi-bam-io/src/writer.rs

@@ -112,6 +112,25 @@ pub(crate) fn make_bgzf_writer(
     }
 }
 
+/// Create a BGZF-compressed writer over an arbitrary `output`, selecting a
+/// multi-threaded encoder when `threads > 1`.
+///
+/// This exposes the same BGZF backend fgumi uses for BAM output so other output
+/// formats (e.g. gzipped FASTQ) get identical, block-gzip (BGZF) framing and
+/// multi-threaded compression rather than a separate implementation.
+///
+/// `compression_level` is the BGZF/DEFLATE level (0–12, where 0 means
+/// uncompressed); values outside the valid range fall back to the backend
+/// default.
+#[must_use]
+pub fn create_bgzf_writer(
+    output: Box<dyn Write + Send>,
+    threads: usize,
+    compression_level: u32,
+) -> BgzfWriterEnum {
+    make_bgzf_writer(output, threads, compression_level)
+}
+
 /// Write a BAM header (magic, SAM text, reference sequences) to any writer.
 ///
 /// Shared implementation used by the BAM writer factories (e.g. [`RawBamWriter`]).

crates/fgumi-pipeline-core/src/handles.rs

@@ -1031,6 +1031,32 @@ where
     }
 }
 
+impl<A, B, C> OutputHandles<crate::outputs::OrderedBytesTuple3<A, B, C>>
+where
+    A: Send + HeapSize + Ordered + 'static,
+    B: Send + HeapSize + Ordered + 'static,
+    C: Send + HeapSize + Ordered + 'static,
+{
+    /// Borrow the typed per-branch view for an ordered + byte-bounded 3-tuple
+    /// output. Same view type as plain `(A, B, C)` — the view carries only
+    /// `BranchOutputHandle`s; the ordering is encoded in the queue transport.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the type-erased outputs view doesn't downcast to
+    /// `Tuple3OutputsView<A, B, C>` (a framework invariant violation).
+    #[must_use]
+    #[inline]
+    pub fn view(&self) -> Tuple3View<'_, A, B, C> {
+        let v = self
+            .inner
+            .inner
+            .downcast_ref::<Tuple3OutputsView<A, B, C>>()
+            .expect("OrderedBytesTuple3 outputs view downcast failed");
+        Tuple3View { a: &v.a, b: &v.b, c: &v.c }
+    }
+}
+
 pub struct Tuple2View<'a, A: Send + HeapSize + 'static, B: Send + HeapSize + 'static> {
     pub a: &'a BranchOutputHandle<A>,
     pub b: &'a BranchOutputHandle<B>,
@@ -1196,6 +1222,29 @@ where
     }
 }
 
+impl<A, B, C> OutputHandles<crate::outputs::OrderedBytesTuple3<A, B, C>>
+where
+    A: Send + HeapSize + Ordered + 'static,
+    B: Send + HeapSize + Ordered + 'static,
+    C: Send + HeapSize + Ordered + 'static,
+{
+    /// Mark all output branches drained.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the type-erased outputs view doesn't downcast to
+    /// `Tuple3OutputsView<A, B, C>` (the view shape is shared with plain
+    /// `(A, B, C)`).
+    pub fn mark_all_drained(&self) {
+        let view = self
+            .inner
+            .inner
+            .downcast_ref::<Tuple3OutputsView<A, B, C>>()
+            .expect("OrderedBytesTuple3 outputs view downcast failed");
+        view.mark_all_drained();
+    }
+}
+
 impl<A, B, C> OutputHandles<(A, B, C)>
 where
     A: Send + HeapSize + 'static,
@@ -1402,6 +1451,49 @@ where
     (queue_set, outputs_view)
 }
 
+/// Build queues for `OrderedBytesTuple3<A, B, C>` outputs. All three branches
+/// support the full `QueueSpec × BranchOrdering` cross-product because each is
+/// `Ordered + HeapSize`. The `Tuple3OutputsView` carries only
+/// `BranchOutputHandle`s (no ordering metadata), so the view type is the same
+/// as for plain `(A, B, C)`.
+pub(crate) fn build_tuple3_queues_ordered_bytes<A, B, C>(
+    specs: &[QueueSpec],
+    ordering: &[BranchOrdering],
+    level: crate::builder::InstrumentationLevel,
+) -> (OutputQueueSet, OutputsViewAny)
+where
+    A: Send + HeapSize + Ordered + 'static,
+    B: Send + HeapSize + Ordered + 'static,
+    C: Send + HeapSize + Ordered + 'static,
+{
+    assert_eq!(specs.len(), 3, "OrderedBytesTuple3<A, B, C>::build_queues requires 3 specs");
+    assert_eq!(ordering.len(), 3, "OrderedBytesTuple3<A, B, C>::build_queues requires 3 orderings");
+
+    let ba = build_branch_ordered_bytes::<A>(specs[0], ordering[0], level);
+    let bb = build_branch_ordered_bytes::<B>(specs[1], ordering[1], level);
+    let bc = build_branch_ordered_bytes::<C>(specs[2], ordering[2], level);
+    let view = Tuple3OutputsView { a: ba.output, b: bb.output, c: bc.output };
+    let outputs_view = OutputsViewAny { inner: Box::new(view) };
+    let queue_set = OutputQueueSet::new(vec![
+        BranchEntry {
+            input_handle: Box::new(ba.input),
+            bounded_queue_handle: ba.bounded_queue_handle,
+            metrics: ba.metrics,
+        },
+        BranchEntry {
+            input_handle: Box::new(bb.input),
+            bounded_queue_handle: bb.bounded_queue_handle,
+            metrics: bb.metrics,
+        },
+        BranchEntry {
+            input_handle: Box::new(bc.input),
+            bounded_queue_handle: bc.bounded_queue_handle,
+            metrics: bc.metrics,
+        },
+    ]);
+    (queue_set, outputs_view)
+}
+
 pub(crate) fn build_tuple4_queues<A, B, C, D>(
     specs: &[QueueSpec],
     ordering: &[BranchOrdering],

crates/fgumi-pipeline-core/src/outputs.rs

@@ -171,6 +171,46 @@ where
     }
 }
 
+/// Ordered + byte-bounded tuple-3 outputs. The 3-branch analog of
+/// [`OrderedBytesTuple2`]: use when a step fans out to three branches that all
+/// need `BranchOrdering::ByItemOrdinal` + byte-bounded queues — e.g. paired
+/// FASTQ output splitting one record batch into R1 / R2 / other byte streams,
+/// each feeding an ordered `WriteRawFile` sink.
+///
+/// The plain `(A, B, C)` `StepOutputs` impl only bounds each branch `HeapSize`.
+/// To use `ByItemOrdinal` on any branch, all three must satisfy
+/// `Ordered + HeapSize` — that's what this shape encodes at the type level.
+#[allow(clippy::type_complexity)]
+pub struct OrderedBytesTuple3<A, B, C>(PhantomData<fn() -> (A, B, C)>)
+where
+    A: Send + HeapSize + Ordered + 'static,
+    B: Send + HeapSize + Ordered + 'static,
+    C: Send + HeapSize + Ordered + 'static;
+
+impl<A, B, C> StepOutputs for OrderedBytesTuple3<A, B, C>
+where
+    A: Send + HeapSize + Ordered + 'static,
+    B: Send + HeapSize + Ordered + 'static,
+    C: Send + HeapSize + Ordered + 'static,
+{
+    #[inline]
+    fn arity() -> usize {
+        3
+    }
+
+    fn build_queues(
+        specs: &[QueueSpec],
+        ordering: &[BranchOrdering],
+        level: crate::builder::InstrumentationLevel,
+    ) -> (OutputQueueSet, OutputsViewAny) {
+        super::handles::build_tuple3_queues_ordered_bytes::<A, B, C>(specs, ordering, level)
+    }
+
+    fn mark_all_drained(handles: &super::step::OutputHandles<Self>) {
+        handles.mark_all_drained();
+    }
+}
+
 impl<A, B, C> StepOutputs for (A, B, C)
 where
     A: Send + HeapSize + 'static,
@@ -258,6 +298,26 @@ mod tests {
         assert_eq!(<(u32, u64, String) as StepOutputs>::arity(), 3);
     }
 
+    #[derive(Clone, Copy)]
+    struct OrdU64(u64);
+    impl crate::item::HeapSize for OrdU64 {}
+    impl crate::item::Ordered for OrdU64 {
+        fn ordinal(&self) -> u64 {
+            self.0
+        }
+    }
+
+    #[test]
+    fn ordered_bytes_tuple_3_arity_is_three() {
+        // OrderedBytesTuple3 is the ordered + byte-bounded 3-way fan-out shape.
+        // Its three branches carry Ordered + HeapSize items (here the u64/u32
+        // stand-ins just need to satisfy the bounds at the type level).
+        fn assert_arity<O: StepOutputs>() -> usize {
+            O::arity()
+        }
+        assert_eq!(assert_arity::<OrderedBytesTuple3<OrdU64, OrdU64, OrdU64>>(), 3);
+    }
+
     #[test]
     fn tuple_4_arity_is_four() {
         assert_eq!(<(u32, u64, String, Vec<u8>) as StepOutputs>::arity(), 4);

crates/fgumi-pipeline-io/src/sink/mod.rs

@@ -1 +1,2 @@
 pub mod write_bgzf;
+pub mod write_raw;