differential_dataflow/batcher/
mod.rs

1//! Traits and implementations for forming batches from streams of updates.
2
3use timely::progress::frontier::AntichainRef;
4
5pub mod merge;
6
7/// A type capable of accepting containers of updates, and carving them out by time as batches.
8///
9/// Updates are accepted as `C0`, the containers that arrive on the dataflow edge, and released as
10/// `Output`, whatever the implementor means by a batch. The two need not agree: an implementor
11/// staging updates in a form of its own can release that form directly, and one whose batch is a
12/// sequence of chunks names a sequence as its output.
13///
14/// The implementor determines the meaning of extraction by a frontier; it is not required to be by
15/// antichain partial order.
16pub trait Batcher<C0> : Extract {
17    /// Takes the updates in `container`, leaving it in an undefined state.
18    ///
19    /// The implementor decides whether to claim the container's allocation or to drain it and
20    /// leave the allocation with the caller, who is free to reuse the container either way.
21    fn insert(&mut self, container: &mut C0);
22}
23
24/// Supertrait for `Batcher` that ensures a consistent time and output across `C0` implementations.
25pub trait Extract {
26    /// The timestamps by which updates are carved out.
27    type Time;
28    /// The batches released by extraction.
29    type Output;
30
31    /// Extracts the updates `upper` unblocks as a batch, and lower bounds the times of those retained.
32    ///
33    /// What `upper` unblocks is the implementor's to decide. It can be based on the antichain up
34    /// set, or it can be based on the total order of times (as used in delta join constructions).
35    /// Absent a batch, `upper` unblocked no updates.
36    ///
37    /// The reported lower bound should accurately reflect the times of all accepted updates that
38    /// have not been extracted. Over approximation can result in stalling dataflows, and under
39    /// approximation is simply incorrect.
40    fn extract<'a>(&'a mut self, upper: AntichainRef<'_, Self::Time>) -> (Option<Self::Output>, AntichainRef<'a, Self::Time>);
41}
differential_dataflow/batcher/mod.rs

differential_dataflow/batcher/
mod.rs
