diff --git a/differential-dataflow/src/operators/co_reduce.rs b/differential-dataflow/src/operators/co_reduce.rs new file mode 100644 index 000000000..53d10d451 --- /dev/null +++ b/differential-dataflow/src/operators/co_reduce.rs @@ -0,0 +1,179 @@ +//! A key-wise reduction over several co-arranged inputs. +//! +//! The driver mirrors [`reduce_with_tactic`](crate::operators::reduce::reduce_with_tactic), with one +//! source trace per input. Batches reach the tactic tagged by their input, `(i, batch)`, so any +//! [`ReduceTactic`] whose backend reads tagged batches runs unchanged. +//! +//! The interval retired is bounded by the meet of the inputs' progress. An input running ahead of +//! the others can hand the tactic batches holding times beyond the interval. The tactic must defer +//! the times it finds there, as it does output times. + +use timely::progress::frontier::Antichain; +use timely::progress::Timestamp; +use timely::dataflow::operators::CapabilitySet; +use timely::dataflow::operators::generic::OutputBuilder; +use timely::dataflow::operators::generic::builder_rc::OperatorBuilder; +use timely::dataflow::channels::pact::Pipeline; + +use crate::operators::arrange::{Arranged, TraceAgent}; +use crate::operators::reduce::ReduceTactic; +use crate::trace::{ExertionLogic, Trace, TraceReader}; + +/// Per-input state of the driver. +struct Side { + trace: Tr, + /// Received batches not yet handed to the tactic, with their capabilities, in order. + queued: Vec<(CapabilitySet, Tr::Batch, Antichain)>, + /// The upper of the last received batch, advanced past empty batches in the trace. + received: Antichain, + /// The bound through which batches have been handed to the tactic, a batch boundary. + handed: Antichain, +} + +impl Side { + fn new(trace: Tr) -> Self { + let min = Antichain::from_elem(::minimum()); + Side { trace, queued: Vec::new(), received: min.clone(), handed: min } + } +} + +/// Drives a key-wise reduction over `inputs`, at least one, using a supplied [`ReduceTactic`]. +/// +/// The tactic sees each batch tagged by the index of the input it came from. Its output at a time +/// `t` must be a function of each key's input accumulations at `t`. Beyond the contract of +/// [`ReduceTactic::retire`], it must defer every time at or beyond `upper` that a new input batch +/// holds, and leave its input unread only when no held time lies below `upper`. +pub fn co_reduce_with_tactic<'scope, Tr1, Tr2, T>( + inputs: Vec>, + name: &str, + mut tactic: T, +) -> Arranged<'scope, TraceAgent> +where + Tr1: TraceReader + 'static, + Tr2: Trace