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::order::Product;
17use timely::progress::Timestamp;
18use timely::dataflow::operators::generic::{OutputBuilder, builder_rc::OperatorBuilder};
19use timely::dataflow::operators::CapabilitySet;
20use timely::dataflow::channels::pact::Pipeline;
21use timely::progress::Antichain;
22
23use crate::difference::Semigroup;
24use crate::{VecCollection, Data};
25use crate::collection::AsCollection;
26use crate::dynamic::pointstamp::PointStamp;
27use crate::dynamic::pointstamp::PointStampSummary;
28
29impl<'scope, D, R, T, TOuter> VecCollection<'scope, Product<TOuter, PointStamp<T>>, D, R>
30where
31    D: Data,
32    R: Semigroup+'static,
33    T: Timestamp+Default,
34    TOuter: Timestamp,
35{
36    /// Enters a dynamically created scope which has `level` timestamp coordinates.
37    pub fn enter_dynamic(self, _level: usize) -> Self {
38        self
39    }
40    /// Leaves a dynamically created scope which has `level` timestamp coordinates.
41    pub fn leave_dynamic(self, level: usize) -> Self {
42        // Create a unary operator that will strip all but `level-1` timestamp coordinates.
43        let mut builder = OperatorBuilder::new("LeaveDynamic".to_string(), self.scope());
44        let (output, stream) = builder.new_output();
45        let mut output = OutputBuilder::from(output);
46        let mut input = builder.new_input_connection(self.inner, Pipeline, [(0, Antichain::from_elem(Product { outer: Default::default(), inner: PointStampSummary { retain: Some(level - 1), actions: Vec::new() } }))]);
47
48        builder.build(move |_capability| move |_frontier| {
49            let mut output = output.activate();
50            input.for_each(|cap, data| {
51                // A message may carry several timestamps (a multi-element stamp, e.g. late
52                // iterations of one epoch alongside early ones of the next): hold a capability
53                // for each, truncated exactly as the records are.
54                let truncate = |time: &Product<TOuter, PointStamp<T>>| {
55                    let mut new_time = time.clone();
56                    let mut vec = std::mem::take(&mut new_time.inner).into_inner();
57                    vec.truncate(level - 1);
58                    new_time.inner = PointStamp::new(vec);
59                    new_time
60                };
61                let caps: CapabilitySet<_> = cap.stamp().iter().map(|t| cap.delayed(&truncate(t), 0)).collect();
62                for (_data, time, _diff) in data.iter_mut() {
63                    *time = truncate(time);
64                }
65                output.session(&caps).give_container(data);
66            });
67        });
68
69        stream.as_collection()
70    }
71}
72
73/// Produces the summary for a feedback operator at `level`, applying `summary` to that coordinate.
74pub fn feedback_summary<T>(level: usize, summary: T::Summary) -> PointStampSummary<T::Summary>
75where
76    T: Timestamp+Default,
77{
78    PointStampSummary {
79        retain: None,
80        actions: std::iter::repeat(Default::default()).take(level-1).chain(std::iter::once(summary)).collect(),
81    }
82}
differential_dataflow/dynamic/mod.rs

differential_dataflow/dynamic/
mod.rs
