differential_dataflow/dynamic/
mod.rs

1//! Types and operators for dynamically scoped iterative dataflows.
2//! 
3//! Scopes in timely dataflow are expressed statically, as part of the type system.
4//! This affords many efficiencies, as well as type-driven reassurance of correctness.
5//! However, there are times you need scopes whose organization is discovered only at runtime.
6//! Naiad and Materialize are examples: the latter taking arbitrary SQL into iterative dataflows.
7//! 
8//! This module provides a timestamp type `Pointstamp` that can represent an update with an 
9//! unboundedly long sequence of some `T: Timestamp`, ordered by the product order by which times
10//! in iterative dataflows are ordered. The module also provides methods for manipulating these 
11//! timestamps to emulate the movement of update streams in to, within, and out of iterative scopes.
12//! 
13
14pub mod pointstamp;
15
16use timely::dataflow::Scope;
17use timely::order::Product;
18use timely::progress::Timestamp;
19use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
20use timely::dataflow::channels::pact::Pipeline;
21use timely::progress::Antichain;
22
23use crate::difference::Semigroup;
24use crate::{Collection, Data};
25use crate::collection::AsCollection;
26use crate::dynamic::pointstamp::PointStamp;
27use crate::dynamic::pointstamp::PointStampSummary;
28
29impl<G, D, R, T, TOuter> Collection<G, D, R>
30where
31    G: Scope<Timestamp = Product<TOuter, PointStamp<T>>>,
32    D: Data,
33    R: Semigroup,
34    T: Timestamp+Default,
35    TOuter: Timestamp,
36{
37    /// Enters a dynamically created scope which has `level` timestamp coordinates.
38    pub fn enter_dynamic(&self, _level: usize) -> Self {
39        (*self).clone()
40    }
41    /// Leaves a dynamically created scope which has `level` timestamp coordinates.
42    pub fn leave_dynamic(&self, level: usize) -> Self {
43        // Create a unary operator that will strip all but `level-1` timestamp coordinates.
44        let mut builder = OperatorBuilder::new("LeaveDynamic".to_string(), self.scope());
45        let (mut output, stream) = builder.new_output();
46        let mut input = builder.new_input_connection(&self.inner, Pipeline, vec![Antichain::from_elem(Product { outer: Default::default(), inner: PointStampSummary { retain: Some(level - 1), actions: Vec::new() } })]);
47
48        let mut vector = Default::default();
49        builder.build(move |_capability| move |_frontier| {
50            let mut output = output.activate();
51            input.for_each(|cap, data| {
52                data.swap(&mut vector);
53                let mut new_time = cap.time().clone();
54                new_time.inner.vector.truncate(level - 1);
55                let new_cap = cap.delayed(&new_time);
56                for (_data, time, _diff) in vector.iter_mut() {
57                    time.inner.vector.truncate(level - 1);
58                }
59                output.session(&new_cap).give_vec(&mut vector);
60            });
61        });
62
63        stream.as_collection()
64    }
65}
66
67/// Produces the summary for a feedback operator at `level`, applying `summary` to that coordinate.
68pub fn feedback_summary<T>(level: usize, summary: T::Summary) -> PointStampSummary<T::Summary> 
69where
70    T: Timestamp+Default,
71{
72    PointStampSummary {
73        retain: None,
74        actions: std::iter::repeat(Default::default()).take(level-1).chain(std::iter::once(summary)).collect(),
75    }
76}
differential_dataflow/dynamic/mod.rs

differential_dataflow/dynamic/
mod.rs
