From a5dcaa1f77c523f0c12a42abb54d2c4015b60f41 Mon Sep 17 00:00:00 2001 From: Nils Homer Date: Fri, 19 Jun 2026 14:06:17 -0700 Subject: [PATCH] docs(pipeline): pin the deliberate Step2 reorder-stage asymmetry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A review of `TypedStep2::build_output_set` asked why it doesn't apply the Serial/Exclusive reorder-collapse that `TypedStep::build_output_set` has. Document that the asymmetry is intentional and add a test that pins it. The single-input collapse is sound because of a universal property: a Serial single-input step consumes one already-ordered stream, so any input ordinal it propagates onto an output is emitted in push order, making the reorder stage redundant. A two-input MERGE has no such guarantee — it interleaves two branches and can push a `ByItemOrdinal` output out of ordinal order even under serial execution, so its reorder stage is load-bearing. Collapsing it would silently deliver records out of order. Today's production Step2s happen not to need it (`PairRawFastq` uses `BranchOrdering::None`; `ZipperMergeStep` assigns fresh sequential ordinals), but the framework cannot assume that for an arbitrary merge. Adds `step2_serial_byitemordinal_output_is_reordered_not_collapsed`, which pushes a Serial Step2's ordered output out of ordinal order and asserts the consumer receives it in ordinal order — it fails if the single-input collapse is ever added to the Step2 path (verified). Stacked on the fgumi-pipeline-core extraction (#440); rebases onto feat-runall once that merges. --- crates/fgumi-pipeline-core/src/erased.rs | 93 ++++++++++++++++++++++++ 1 file changed, 93 insertions(+) diff --git a/crates/fgumi-pipeline-core/src/erased.rs b/crates/fgumi-pipeline-core/src/erased.rs index 2cb047613..1f10e7b60 100644 --- a/crates/fgumi-pipeline-core/src/erased.rs +++ b/crates/fgumi-pipeline-core/src/erased.rs @@ -556,6 +556,31 @@ impl ErasedStep for TypedStep2 { } fn build_output_set(&self) -> (OutputQueueSet, OutputsViewAny) { + // Unlike `TypedStep::build_output_set`, a `Step2` does NOT collapse a + // `ByOrdinal` / `ByItemOrdinal` output to `None` for `Serial` / + // `Exclusive` kinds — the ordering is passed through verbatim and the + // reorder stage is kept. + // + // The single-input collapse rests on a universal property: a Serial + // single-input step consumes one already-ordered stream, so any input + // ordinal it propagates onto an output is emitted in push order, making + // the reorder stage redundant. A two-input MERGE has no such guarantee. + // It interleaves two branches, so a `ByItemOrdinal` output that + // propagates an *input's* ordinal can be pushed out of ordinal order + // even under serial (mutex-serialized, single-dispatcher) execution — + // e.g. when branch B's next-needed ordinal hasn't arrived yet but + // branch A's later one has. For such a step the reorder stage is + // load-bearing: dropping it would deliver records out of order. + // + // Today's production `Step2`s happen not to need it (`PairRawFastq` + // declares `BranchOrdering::None`; `ZipperMergeStep` assigns fresh + // *sequential* output ordinals, so its push order already equals its + // ordinal order). But the framework cannot assume that for an arbitrary + // merge, so it conservatively keeps the reorder for every ordered Step2 + // output. Do not add the single-input collapse here: a `Step2` that + // propagates an input ordinal can emit out of ordinal order even under + // serial execution, so dropping the reorder would deliver records out of + // order. ::build_queues( &self.inner.profile().output_queues, &self.inner.profile().branch_ordering, @@ -872,4 +897,72 @@ mod tests { assert!(source.is_source()); assert!(!mid.is_source()); } + + /// Pins the deliberate asymmetry between `TypedStep::build_output_set` + /// (collapses ordered output → `None` for Serial/Exclusive) and + /// `TypedStep2::build_output_set` (keeps the reorder stage). A two-input + /// merge can push a `ByItemOrdinal` output out of ordinal order even under + /// serial execution, so the reorder stage is load-bearing. This test fails + /// if a future change adds the single-input collapse to the Step2 path. + #[test] + fn step2_serial_byitemordinal_output_is_reordered_not_collapsed() { + use crate::item::{HeapSize, Ordered}; + use crate::outputs::OrderedBytesSingle; + use crate::step::{Step2, StepCtx2}; + + #[derive(Debug)] + struct Ord32 { + ordinal: u64, + } + impl HeapSize for Ord32 { + fn heap_size(&self) -> usize { + 0 + } + } + impl Ordered for Ord32 { + fn ordinal(&self) -> u64 { + self.ordinal + } + } + + // A Serial two-input merge with a `ByItemOrdinal` output. We only need + // its output edge (via `build_output_set`), never run it, so `try_run` + // is trivial. + struct OutOfOrderMerge; + impl Step2 for OutOfOrderMerge { + type InputA = u32; + type InputB = u32; + type Outputs = OrderedBytesSingle; + fn profile(&self) -> StepProfile { + StepProfile { + name: "OutOfOrderMerge", + kind: StepKind::Serial, + sticky: false, + output_queues: vec![QueueSpec::ByteBounded { limit_bytes: 64 * 1024 }], + branch_ordering: vec![BranchOrdering::ByItemOrdinal], + } + } + fn try_run(&mut self, _ctx: &mut StepCtx2<'_, Self>) -> io::Result { + Ok(StepOutcome::NoProgress) + } + } + + let step: Box = Box::new(TypedStep2::new(OutOfOrderMerge)); + let (mut queue_set, view) = step.build_output_set(); + let outputs_any = step.wrap_outputs_view(view); + let outputs = outputs_any + .downcast_ref::>>() + .expect("OutputHandles downcast"); + + // Push OUT of ordinal order: 2, then 0, then 1. + outputs.push(Ord32 { ordinal: 2 }).unwrap(); + outputs.push(Ord32 { ordinal: 0 }).unwrap(); + outputs.push(Ord32 { ordinal: 1 }).unwrap(); + + // The reorder stage must deliver them in ORDINAL order (0, 1, 2). With + // the single-input collapse the consumer would see push order (2, 0, 1). + let input = queue_set.take_typed_input::(0); + let got: Vec = std::iter::from_fn(|| input.pop().map(|o| o.ordinal)).collect(); + assert_eq!(got, vec![0, 1, 2], "ordered Step2 output must be reordered by ordinal"); + } }