differential_dataflow/columnar/collection/
operators.rs

1//! Dataflow operators over the columnar [`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                // A message may carry several timestamps (a multi-element stamp): hold a
120                // capability for each, truncated exactly as the updates are.
121                let new_cap: timely::dataflow::operators::CapabilitySet<_> = cap
122                    .stamp()
123                    .iter()
124                    .map(|t| {
125                        let mut new_time = t.clone();
126                        let mut vec = std::mem::take(&mut new_time.inner).into_inner();
127                        vec.truncate(level - 1);
128                        new_time.inner = PointStamp::new(vec);
129                        cap.delayed(&new_time, 0)
130                    })
131                    .collect();
132                // Push updates with truncated times into the builder.
133                // The builder's form call on flush sorts and consolidates,
134                // handling the duplicate times that truncation can produce.
135                // TODO: The input trie is already sorted; a streaming form
136                // that accepts pre-sorted, potentially-collapsing timestamps
137                // could avoid the re-sort inside the builder.
138                for (k, v, t, d) in data.updates.view().iter() {
139                    Columnar::copy_from(&mut time, t);
140                    let mut inner_vec = std::mem::take(&mut time.inner).into_inner();
141                    inner_vec.truncate(level - 1);
142                    time.inner = PointStamp::new(inner_vec);
143                    col_builder.push_into((k, v, &time, d));
144                }
145                let mut session = output.session(&new_cap);
146                while let Some(container) = col_builder.finish() {
147                    session.give_container(container);
148                }
149            });
150        }
151    });
152
153    stream.as_collection()
154}
155
156/// Extract a `Collection<_, RecordedUpdates<U>>` from a columnar `Arranged`.
157///
158/// Cursors through each batch and pushes `(key, val, time, diff)` refs into
159/// a `Builder`, which sorts and consolidates on flush.
160pub fn as_recorded_updates<U>(
161    arranged: crate::operators::arrange::Arranged<
162        crate::operators::arrange::TraceAgent<Spine<U::Key, U::Val, U::Time, U::Diff>>,
163    >,
164) -> crate::Collection<U::Time, RecordedUpdates<U>>
165where
166    U: layout::ColumnarUpdate,
167{
168    use timely::dataflow::operators::generic::Operator;
169    use timely::dataflow::channels::pact::Pipeline;
170    use crate::trace::{Navigable, Cursor};
171    use crate::AsCollection;
172
173    arranged.stream
174        .unary::<Builder<U>, _, _, _>(Pipeline, "AsRecordedUpdates", |_, _| {
175            move |input, output| {
176                input.for_each(|time, batches| {
177                    let mut session = output.session_with_builder(&time);
178                    for batch in batches.drain(..) {
179                        let Some(batch) = batch.inner else { continue };
180                        let mut cursor = batch.cursor();
181                        while cursor.key_valid(&batch) {
182                            while cursor.val_valid(&batch) {
183                                let key = cursor.key(&batch);
184                                let val = cursor.val(&batch);
185                                cursor.map_times(&batch, |time, diff| {
186                                    session.give((key, val, time, diff));
187                                });
188                                cursor.step_val(&batch);
189                            }
190                            cursor.step_key(&batch);
191                        }
192                    }
193                });
194            }
195        })
196        .as_collection()
197}
differential_dataflow/columnar/collection/operators.rs

differential_dataflow/columnar/collection/
operators.rs
