differential_dataflow/columnar/collection/
operators.rs

1//! Dataflow operators over the columnar [`RecordedUpdates`](super::RecordedUpdates)
2//! container — the collection-level surface of the columnar chunk.
3//!
4//! [`join_function`] is a columnar `flat_map` (subsuming map/filter/negate/
5//! enter_at); [`leave_dynamic`] truncates dynamic-scope timestamps; and
6//! [`as_recorded_updates`] extracts a `RecordedUpdates` collection from a
7//! columnar arrangement's batch stream.
8
9use crate::columnar::layout;
10use crate::columnar::trace::Spine;
11use super::{Builder, RecordedUpdates};
12
13/// A columnar flat_map: iterates RecordedUpdates, calls logic per (key, val, time, diff),
14/// joins output times with input times, multiplies output diffs with input diffs.
15///
16/// This subsumes map, filter, negate, and enter_at for columnar collections.
17pub fn join_function<U, I, L>(
18    input: crate::Collection<U::Time, RecordedUpdates<U>>,
19    mut logic: L,
20) -> crate::Collection<U::Time, RecordedUpdates<U>>
21where
22    U::Time: crate::lattice::Lattice,
23    U: layout::ColumnarUpdate<Diff: crate::difference::Multiply<U::Diff, Output = U::Diff>>,
24    I: IntoIterator<Item = (U::Key, U::Val, U::Time, U::Diff)>,
25    L: FnMut(
26        columnar::Ref<'_, U::Key>,
27        columnar::Ref<'_, U::Val>,
28        columnar::Ref<'_, U::Time>,
29        columnar::Ref<'_, U::Diff>,
30    ) -> I + 'static,
31{
32    use timely::dataflow::operators::generic::Operator;
33    use timely::dataflow::channels::pact::Pipeline;
34    use crate::AsCollection;
35    use crate::difference::Multiply;
36    use crate::lattice::Lattice;
37    use columnar::Columnar;
38
39    input
40        .inner
41        .unary::<Builder<U>, _, _, _>(Pipeline, "JoinFunction", move |_, _| {
42            move |input, output| {
43                let mut t1o = U::Time::default();
44                let mut d1o = U::Diff::default();
45                input.for_each(|time, data| {
46                    let mut session = output.session_with_builder(&time);
47                    for (k1, v1, t1, d1) in data.updates.view().iter() {
48                        Columnar::copy_from(&mut t1o, t1);
49                        Columnar::copy_from(&mut d1o, d1);
50                        for (k2, v2, t2, d2) in logic(k1, v1, t1, d1) {
51                            let t3 = t2.join(&t1o);
52                            let d3 = d2.multiply(&d1o);
53                            session.give((&k2, &v2, &t3, &d3));
54                        }
55                    }
56                });
57            }
58        })
59        .as_collection()
60}
61
62/// Timestamp shape of a dynamic iterative scope: an outer timestamp paired
63/// with a per-level `PointStamp` of loop counters.
64pub type DynTime<TOuter, T> = timely::order::Product<TOuter, crate::dynamic::pointstamp::PointStamp<T>>;
65
66/// Leave a dynamic iterative scope, truncating PointStamp coordinates.
67///
68/// Uses OperatorBuilder (not unary) for the custom input connection summary
69/// that tells timely how the PointStamp is affected (retain `level - 1` coordinates).
70///
71/// Consolidates after truncation since distinct PointStamp coordinates can collapse.
72pub fn leave_dynamic<K, V, R, TOuter, T>(
73    input: crate::Collection<DynTime<TOuter, T>, RecordedUpdates<(K, V, DynTime<TOuter, T>, R)>>,
74    level: usize,
75) -> crate::Collection<DynTime<TOuter, T>, RecordedUpdates<(K, V, DynTime<TOuter, T>, R)>>
76where
77    K: columnar::Columnar,
78    V: columnar::Columnar,
79    R: columnar::Columnar,
80    TOuter: timely::progress::Timestamp + Default + columnar::Columnar,
81    T: timely::progress::Timestamp + Default + columnar::Columnar,
82    (K, V, DynTime<TOuter, T>, R): layout::ColumnarUpdate<Key = K, Val = V, Time = DynTime<TOuter, T>, Diff = R>,
83{
84    assert!(level > 0, "leave_dynamic requires level > 0");
85    use timely::dataflow::channels::pact::Pipeline;
86    use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
87    use timely::dataflow::operators::generic::OutputBuilder;
88    use timely::order::Product;
89    use timely::progress::Antichain;
90    use timely::container::{ContainerBuilder, PushInto};
91    use crate::AsCollection;
92    use crate::dynamic::pointstamp::{PointStamp, PointStampSummary};
93    use columnar::Columnar;
94
95    let mut builder = OperatorBuilder::new("LeaveDynamic".to_string(), input.inner.scope());
96    let (output, stream) = builder.new_output();
97    let mut output = OutputBuilder::from(output);
98    let mut op_input = builder.new_input_connection(
99        input.inner,
100        Pipeline,
101        [(
102            0,
103            Antichain::from_elem(Product {
104                outer: Default::default(),
105                inner: PointStampSummary {
106                    retain: Some(level - 1),
107                    actions: Vec::new(),
108                },
109            }),
110        )],
111    );
112
113    builder.build(move |_capability| {
114        let mut col_builder = Builder::<(K, V, DynTime<TOuter, T>, R)>::default();
115        let mut time = DynTime::<TOuter, T>::default();
116        move |_frontier| {
117            let mut output = output.activate();
118            op_input.for_each(|cap, data| {
119                // Truncate the capability's timestamp.
120                let mut new_time = cap.time().clone();
121                let mut vec = std::mem::take(&mut new_time.inner).into_inner();
122                vec.truncate(level - 1);
123                new_time.inner = PointStamp::new(vec);
124                let new_cap = cap.delayed(&new_time, 0);
125                // Push updates with truncated times into the builder.
126                // The builder's form call on flush sorts and consolidates,
127                // handling the duplicate times that truncation can produce.
128                // TODO: The input trie is already sorted; a streaming form
129                // that accepts pre-sorted, potentially-collapsing timestamps
130                // could avoid the re-sort inside the builder.
131                for (k, v, t, d) in data.updates.view().iter() {
132                    Columnar::copy_from(&mut time, t);
133                    let mut inner_vec = std::mem::take(&mut time.inner).into_inner();
134                    inner_vec.truncate(level - 1);
135                    time.inner = PointStamp::new(inner_vec);
136                    col_builder.push_into((k, v, &time, d));
137                }
138                let mut session = output.session(&new_cap);
139                while let Some(container) = col_builder.finish() {
140                    session.give_container(container);
141                }
142            });
143        }
144    });
145
146    stream.as_collection()
147}
148
149/// Extract a `Collection<_, RecordedUpdates<U>>` from a columnar `Arranged`.
150///
151/// Cursors through each batch and pushes `(key, val, time, diff)` refs into
152/// a `Builder`, which sorts and consolidates on flush.
153pub fn as_recorded_updates<U>(
154    arranged: crate::operators::arrange::Arranged<
155        crate::operators::arrange::TraceAgent<Spine<U::Key, U::Val, U::Time, U::Diff>>,
156    >,
157) -> crate::Collection<U::Time, RecordedUpdates<U>>
158where
159    U: layout::ColumnarUpdate,
160{
161    use timely::dataflow::operators::generic::Operator;
162    use timely::dataflow::channels::pact::Pipeline;
163    use crate::trace::{Navigable, Cursor};
164    use crate::AsCollection;
165
166    arranged.stream
167        .unary::<Builder<U>, _, _, _>(Pipeline, "AsRecordedUpdates", |_, _| {
168            move |input, output| {
169                input.for_each(|time, batches| {
170                    let mut session = output.session_with_builder(&time);
171                    for batch in batches.drain(..) {
172                        let mut cursor = batch.cursor();
173                        while cursor.key_valid(&batch) {
174                            while cursor.val_valid(&batch) {
175                                let key = cursor.key(&batch);
176                                let val = cursor.val(&batch);
177                                cursor.map_times(&batch, |time, diff| {
178                                    session.give((key, val, time, diff));
179                                });
180                                cursor.step_val(&batch);
181                            }
182                            cursor.step_key(&batch);
183                        }
184                    }
185                });
186            }
187        })
188        .as_collection()
189}
differential_dataflow/columnar/collection/operators.rs

differential_dataflow/columnar/collection/
operators.rs
