differential_dataflow/trace/implementations/
ord_neu.rs

1//! Trace and batch implementations based on sorted ranges.
2//!
3//! The types and type aliases in this module start with either
4//!
5//! * `OrdVal`: Collections whose data have the form `(key, val)` where `key` is ordered.
6//! * `OrdKey`: Collections whose data have the form `key` where `key` is ordered.
7//!
8//! Although `OrdVal` is more general than `OrdKey`, the latter has a simpler representation
9//! and should consume fewer resources (computation and memory) when it applies.
10
11use std::rc::Rc;
12
13use crate::trace::implementations::spine_fueled::Spine;
14use crate::batcher::merge::chunker::ContainerChunker;
15use crate::batcher::merge::MergeBatcher;
16use crate::batcher::merge::vec::VecMerger;
17
18use super::{Layout, Vector};
19
20pub use self::val_batch::{OrdValBatch, OrdValBuilder};
21pub use self::key_batch::{OrdKeyBatch, OrdKeyBuilder};
22
23/// A trace implementation using a spine of ordered lists.
24pub type OrdValSpine<K, V, T, R> = Spine<Rc<OrdValBatch<Vector<((K,V),T,R)>>>>;
25/// A batcher using ordered lists.
26pub type OrdValBatcher<K, V, T, R> = MergeBatcher<ContainerChunker<Vec<((K, V), T, R)>>, VecMerger<(K, V), T, R>, VecOrdValBuilder<K, V, T, R>>;
27/// A builder using ordered lists.
28pub type VecOrdValBuilder<K, V, T, R> = OrdValBuilder<Vector<((K,V),T,R)>, Vec<((K,V),T,R)>>;
29
30/// A trace implementation using a spine of ordered lists.
31pub type OrdKeySpine<K, T, R> = Spine<Rc<OrdKeyBatch<Vector<((K,()),T,R)>>>>;
32/// A batcher for ordered lists.
33pub type OrdKeyBatcher<K, T, R> = MergeBatcher<ContainerChunker<Vec<((K, ()), T, R)>>, VecMerger<(K, ()), T, R>, VecOrdKeyBuilder<K, T, R>>;
34/// A builder for ordered lists.
35pub type VecOrdKeyBuilder<K, T, R> = OrdKeyBuilder<Vector<((K,()),T,R)>, Vec<((K,()),T,R)>>;
36
37pub use layers::{Vals, Upds};
38/// Layers are containers of lists of some type.
39///
40/// The intent is that they "attach" to an outer layer which has as many values
41/// as the layer has lists, thereby associating a list with each outer value.
42/// A sequence of layers, each matching the number of values in its predecessor,
43/// forms a layered trie: a tree with values of some type on nodes at each depth.
44///
45/// We will form tries here by layering `[Keys, Vals, Upds]` or `[Keys, Upds]`.
46pub mod layers {
47
48    use serde::{Deserialize, Serialize};
49    use crate::trace::implementations::BatchContainer;
50
51    /// A container for non-empty lists of values.
52    #[derive(Debug, Serialize, Deserialize)]
53    pub struct Vals<O, V> {
54        /// Offsets used to provide indexes from keys to values.
55        ///
56        /// The length of this list is one longer than `keys`, so that we can avoid bounds logic.
57        pub offs: O,
58        /// Concatenated ordered lists of values, bracketed by offsets in `offs`.
59        pub vals: V,
60    }
61
62    impl<O: for<'a> BatchContainer<ReadItem<'a> = usize>, V: BatchContainer> Default for Vals<O, V> {
63        fn default() -> Self { Self::with_capacity(0, 0) }
64    }
65
66    impl<O: for<'a> BatchContainer<ReadItem<'a> = usize>, V: BatchContainer> Vals<O, V> {
67        /// Lower and upper bounds in `self.vals` of the indexed list.
68        #[inline(always)] pub fn bounds(&self, index: usize) -> (usize, usize) {
69            (self.offs.index(index), self.offs.index(index+1))
70        }
71        /// Retrieves a value using relative indexes.
72        ///
73        /// The first index identifies a list, and the second an item within the list.
74        /// The method adds the list's lower bound to the item index, and then calls
75        /// `get_abs`. Using absolute indexes within the list's bounds can be more
76        /// efficient than using relative indexing.
77        pub fn get_rel(&self, list_idx: usize, item_idx: usize) -> V::ReadItem<'_> {
78            self.get_abs(self.bounds(list_idx).0 + item_idx)
79        }
80
81        /// Number of lists in the container.
82        pub fn len(&self) -> usize { self.offs.len() - 1 }
83        /// Retrieves a value using an absolute rather than relative index.
84        pub fn get_abs(&self, index: usize) -> V::ReadItem<'_> {
85            self.vals.index(index)
86        }
87        /// Allocates with capacities for a number of lists and values.
88        pub fn with_capacity(o_size: usize, v_size: usize) -> Self {
89            let mut offs = <O as BatchContainer>::with_capacity(o_size);
90            offs.push_ref(0);
91            Self {
92                offs,
93                vals: <V as BatchContainer>::with_capacity(v_size),
94            }
95        }
96        /// Allocates with enough capacity to contain two inputs.
97        pub fn merge_capacity(this: &Self, that: &Self) -> Self {
98            let mut offs = <O as BatchContainer>::with_capacity(this.offs.len() + that.offs.len());
99            offs.push_ref(0);
100            Self {
101                offs,
102                vals: <V as BatchContainer>::merge_capacity(&this.vals, &that.vals),
103            }
104        }
105    }
106
107    /// A container for non-empty lists of updates.
108    ///
109    /// This container uses the special representiation of an empty slice to stand in for
110    /// "the previous single element". An empty slice is an otherwise invalid representation.
111    #[derive(Debug, Serialize, Deserialize)]
112    pub struct Upds<O, T, D> {
113        /// Offsets used to provide indexes from values to updates.
114        pub offs: O,
115        /// Concatenated ordered lists of update times, bracketed by offsets in `offs`.
116        pub times: T,
117        /// Concatenated ordered lists of update diffs, bracketed by offsets in `offs`.
118        pub diffs: D,
119    }
120
121    impl<O: for<'a> BatchContainer<ReadItem<'a> = usize>, T: BatchContainer, D: BatchContainer> Default for Upds<O, T, D> {
122        fn default() -> Self { Self::with_capacity(0, 0) }
123    }
124    impl<O: for<'a> BatchContainer<ReadItem<'a> = usize>, T: BatchContainer, D: BatchContainer> Upds<O, T, D> {
125        /// Lower and upper bounds in `self.times` and `self.diffs` of the indexed list.
126        pub fn bounds(&self, index: usize) -> (usize, usize) {
127            let mut lower = self.offs.index(index);
128            let upper = self.offs.index(index+1);
129            // We use equal lower and upper to encode "singleton update; just before here".
130            // It should only apply when there is a prior element, so `lower` should be greater than zero.
131            if lower == upper {
132                assert!(lower > 0);
133                lower -= 1;
134            }
135            (lower, upper)
136        }
137        /// Retrieves a value using relative indexes.
138        ///
139        /// The first index identifies a list, and the second an item within the list.
140        /// The method adds the list's lower bound to the item index, and then calls
141        /// `get_abs`. Using absolute indexes within the list's bounds can be more
142        /// efficient than using relative indexing.
143        pub fn get_rel(&self, list_idx: usize, item_idx: usize) -> (T::ReadItem<'_>, D::ReadItem<'_>) {
144            self.get_abs(self.bounds(list_idx).0 + item_idx)
145        }
146
147        /// Number of lists in the container.
148        pub fn len(&self) -> usize { self.offs.len() - 1 }
149        /// Retrieves a value using an absolute rather than relative index.
150        pub fn get_abs(&self, index: usize) -> (T::ReadItem<'_>, D::ReadItem<'_>) {
151            (self.times.index(index), self.diffs.index(index))
152        }
153        /// Allocates with capacities for a number of lists and values.
154        pub fn with_capacity(o_size: usize, u_size: usize) -> Self {
155            let mut offs = <O as BatchContainer>::with_capacity(o_size);
156            offs.push_ref(0);
157            Self {
158                offs,
159                times: <T as BatchContainer>::with_capacity(u_size),
160                diffs: <D as BatchContainer>::with_capacity(u_size),
161            }
162        }
163        /// Allocates with enough capacity to contain two inputs.
164        pub fn merge_capacity(this: &Self, that: &Self) -> Self {
165            let mut offs = <O as BatchContainer>::with_capacity(this.offs.len() + that.offs.len());
166            offs.push_ref(0);
167            Self {
168                offs,
169                times: <T as BatchContainer>::merge_capacity(&this.times, &that.times),
170                diffs: <D as BatchContainer>::merge_capacity(&this.diffs, &that.diffs),
171            }
172        }
173    }
174
175    /// Helper type for constructing `Upds` containers.
176    pub struct UpdsBuilder<T: BatchContainer, D: BatchContainer> {
177        /// Local stash of updates, to use for consolidation.
178        ///
179        /// We could emulate a `ChangeBatch` here, with related compaction smarts.
180        /// A `ChangeBatch` itself needs an `i64` diff type, which we have not.
181        stash: Vec<(T::Owned, D::Owned)>,
182        /// Total number of consolidated updates.
183        ///
184        /// Tracked independently to account for duplicate compression.
185        total: usize,
186
187        /// Time container to stage singleton times for evaluation.
188        time_con: T,
189        /// Diff container to stage singleton times for evaluation.
190        diff_con: D,
191    }
192
193    impl<T: BatchContainer, D: BatchContainer> Default for UpdsBuilder<T, D> {
194        fn default() -> Self { Self { stash: Vec::default(), total: 0, time_con: BatchContainer::with_capacity(1), diff_con: BatchContainer::with_capacity(1) } }
195    }
196
197
198    impl<T, D> UpdsBuilder<T, D>
199    where
200        T: BatchContainer<Owned: Ord>,
201        D: BatchContainer<Owned: crate::difference::Semigroup>,
202    {
203        /// Stages one update, but does not seal the set of updates.
204        pub fn push(&mut self, time: T::Owned, diff: D::Owned) {
205            self.stash.push((time, diff));
206        }
207
208        /// Consolidate and insert (if non-empty) the stashed updates.
209        ///
210        /// The return indicates whether the results were indeed non-empty.
211        pub fn seal<O: for<'a> BatchContainer<ReadItem<'a> = usize>>(&mut self, upds: &mut Upds<O, T, D>) -> bool {
212            use crate::consolidation;
213            consolidation::consolidate(&mut self.stash);
214            // If everything consolidates away, return false.
215            if self.stash.is_empty() { return false; }
216            // If there is a singleton, we may be able to optimize.
217            if self.stash.len() == 1 {
218                let (time, diff) = self.stash.last().unwrap();
219                self.time_con.clear(); self.time_con.push_own(time);
220                self.diff_con.clear(); self.diff_con.push_own(diff);
221                if upds.times.last() == self.time_con.get(0) && upds.diffs.last() == self.diff_con.get(0) {
222                    self.total += 1;
223                    self.stash.clear();
224                    upds.offs.push_ref(upds.times.len());
225                    return true;
226                }
227            }
228            // Conventional; move `stash` into `updates`.
229            self.total += self.stash.len();
230            for (time, diff) in self.stash.drain(..) {
231                upds.times.push_own(&time);
232                upds.diffs.push_own(&diff);
233            }
234            upds.offs.push_ref(upds.times.len());
235            true
236        }
237
238        /// Completes the building and returns the total updates sealed.
239        pub fn total(&self) -> usize { self.total }
240    }
241}
242
243/// Types related to forming batches with values.
244pub mod val_batch {
245
246    use std::marker::PhantomData;
247    use serde::{Deserialize, Serialize};
248    use timely::container::PushInto;
249    use timely::progress::{Antichain, frontier::AntichainRef};
250
251    use crate::trace::{Builder, Cursor};
252    use crate::trace::implementations::spine_fueled::{SpineBatch, Merger};
253    use crate::trace::implementations::{BatchContainer};
254    use crate::trace::implementations::layout;
255
256    use super::{Layout, Vals, Upds, layers::UpdsBuilder, BuilderInput};
257
258    /// An immutable collection of update tuples, from a contiguous interval of logical times.
259    #[derive(Debug, Serialize, Deserialize)]
260    #[serde(bound = "
261        L::KeyContainer: Serialize + for<'a> Deserialize<'a>,
262        L::ValContainer: Serialize + for<'a> Deserialize<'a>,
263        L::OffsetContainer: Serialize + for<'a> Deserialize<'a>,
264        L::TimeContainer: Serialize + for<'a> Deserialize<'a>,
265        L::DiffContainer: Serialize + for<'a> Deserialize<'a>,
266    ")]
267    pub struct OrdValStorage<L: Layout> {
268        /// An ordered list of keys.
269        pub keys: L::KeyContainer,
270        /// For each key in `keys`, a list of values.
271        pub vals: Vals<L::OffsetContainer, L::ValContainer>,
272        /// For each val in `vals`, a list of (time, diff) updates.
273        pub upds: Upds<L::OffsetContainer, L::TimeContainer, L::DiffContainer>,
274    }
275
276    /// The updates of a batch, as immutable update tuples.
277    ///
278    /// The `L` parameter captures how the updates should be laid out.
279    #[derive(Serialize, Deserialize)]
280    #[serde(bound = "
281        L::KeyContainer: Serialize + for<'a> Deserialize<'a>,
282        L::ValContainer: Serialize + for<'a> Deserialize<'a>,
283        L::OffsetContainer: Serialize + for<'a> Deserialize<'a>,
284        L::TimeContainer: Serialize + for<'a> Deserialize<'a>,
285        L::DiffContainer: Serialize + for<'a> Deserialize<'a>,
286    ")]
287    pub struct OrdValBatch<L: Layout> {
288        /// The updates themselves.
289        pub storage: OrdValStorage<L>,
290        /// The number of updates reflected in the batch.
291        ///
292        /// We track this separately from `storage` because due to the singleton optimization,
293        /// we may have many more updates than `storage.updates.len()`. It should equal that
294        /// length, plus the number of singleton optimizations employed.
295        pub updates: usize,
296    }
297
298    impl<L: Layout> crate::trace::Navigable for OrdValBatch<L> {
299        type Cursor = OrdValCursor<L>;
300        fn cursor(&self) -> Self::Cursor {
301            OrdValCursor {
302                key_cursor: 0,
303                val_cursor: 0,
304                phantom: PhantomData,
305            }
306        }
307    }
308
309    impl<L: Layout> SpineBatch for OrdValBatch<L> {
310        type Time = layout::Time<L>;
311        type Merger = OrdValMerger<L>;
312        fn len(&self) -> usize {
313            // Normally this would be `self.updates.len()`, but we have a clever compact encoding.
314            // Perhaps we should count such exceptions to the side, to provide a correct accounting.
315            self.updates
316        }
317    }
318
319    /// State for an in-progress merge.
320    pub struct OrdValMerger<L: Layout> {
321        /// Key position to merge next in the first batch.
322        key_cursor1: usize,
323        /// Key position to merge next in the second batch.
324        key_cursor2: usize,
325        /// result that we are currently assembling.
326        result: OrdValStorage<L>,
327        /// Frontier through which times should be advanced, from the result's description.
328        since: Antichain<layout::Time<L>>,
329        /// Staging area to consolidate owned times and diffs, before sealing.
330        staging: UpdsBuilder<L::TimeContainer, L::DiffContainer>,
331    }
332
333    impl<L: Layout> Merger<OrdValBatch<L>> for OrdValMerger<L> {
334        fn new(batch1: &OrdValBatch<L>, batch2: &OrdValBatch<L>, compaction_frontier: AntichainRef<layout::Time<L>>) -> Self {
335
336            let batch1 = &batch1.storage;
337            let batch2 = &batch2.storage;
338
339            OrdValMerger {
340                key_cursor1: 0,
341                key_cursor2: 0,
342                result: OrdValStorage {
343                    keys: L::KeyContainer::merge_capacity(&batch1.keys, &batch2.keys),
344                    vals: Vals::merge_capacity(&batch1.vals, &batch2.vals),
345                    upds: Upds::merge_capacity(&batch1.upds, &batch2.upds),
346                },
347                since: compaction_frontier.to_owned(),
348                staging: UpdsBuilder::default(),
349            }
350        }
351        fn done(self) -> Option<OrdValBatch<L>> {
352            let updates = self.staging.total();
353            (updates > 0).then(|| OrdValBatch { updates, storage: self.result })
354        }
355        fn work(&mut self, source1: &OrdValBatch<L>, source2: &OrdValBatch<L>, fuel: &mut isize) {
356
357            // An (incomplete) indication of the amount of work we've done so far.
358            let starting_updates = self.staging.total();
359            let mut effort = 0isize;
360
361            // While both mergees are still active, perform single-key merges.
362            while self.key_cursor1 < source1.storage.keys.len() && self.key_cursor2 < source2.storage.keys.len() && effort < *fuel {
363                self.merge_key(&source1.storage, &source2.storage);
364                // An (incomplete) accounting of the work we've done.
365                effort = (self.staging.total() - starting_updates) as isize;
366            }
367
368            // Merging is complete, and only copying remains.
369            // Key-by-key copying allows effort interruption, and compaction.
370            while self.key_cursor1 < source1.storage.keys.len() && effort < *fuel {
371                self.copy_key(&source1.storage, self.key_cursor1);
372                self.key_cursor1 += 1;
373                effort = (self.staging.total() - starting_updates) as isize;
374            }
375            while self.key_cursor2 < source2.storage.keys.len() && effort < *fuel {
376                self.copy_key(&source2.storage, self.key_cursor2);
377                self.key_cursor2 += 1;
378                effort = (self.staging.total() - starting_updates) as isize;
379            }
380
381            *fuel -= effort;
382        }
383    }
384
385    // Helper methods in support of merging batches.
386    impl<L: Layout> OrdValMerger<L> {
387        /// Copy the next key in `source`.
388        ///
389        /// The method extracts the key in `source` at `cursor`, and merges it in to `self`.
390        /// If the result does not wholly cancel, they key will be present in `self` with the
391        /// compacted values and updates.
392        ///
393        /// The caller should be certain to update the cursor, as this method does not do this.
394        fn copy_key(&mut self, source: &OrdValStorage<L>, cursor: usize) {
395            // Capture the initial number of values to determine if the merge was ultimately non-empty.
396            let init_vals = self.result.vals.vals.len();
397            let (mut lower, upper) = source.vals.bounds(cursor);
398            while lower < upper {
399                self.stash_updates_for_val(source, lower);
400                if self.staging.seal(&mut self.result.upds) {
401                    self.result.vals.vals.push_ref(source.vals.get_abs(lower));
402                }
403                lower += 1;
404            }
405
406            // If we have pushed any values, copy the key as well.
407            if self.result.vals.vals.len() > init_vals {
408                self.result.keys.push_ref(source.keys.index(cursor));
409                self.result.vals.offs.push_ref(self.result.vals.vals.len());
410            }
411        }
412        /// Merge the next key in each of `source1` and `source2` into `self`, updating the appropriate cursors.
413        ///
414        /// This method only merges a single key. It applies all compaction necessary, and may result in no output
415        /// if the updates cancel either directly or after compaction.
416        fn merge_key(&mut self, source1: &OrdValStorage<L>, source2: &OrdValStorage<L>) {
417            use ::std::cmp::Ordering;
418            match source1.keys.index(self.key_cursor1).cmp(&source2.keys.index(self.key_cursor2)) {
419                Ordering::Less => {
420                    self.copy_key(source1, self.key_cursor1);
421                    self.key_cursor1 += 1;
422                },
423                Ordering::Equal => {
424                    // Keys are equal; must merge all values from both sources for this one key.
425                    let (lower1, upper1) = source1.vals.bounds(self.key_cursor1);
426                    let (lower2, upper2) = source2.vals.bounds(self.key_cursor2);
427                    if let Some(off) = self.merge_vals((source1, lower1, upper1), (source2, lower2, upper2)) {
428                        self.result.keys.push_ref(source1.keys.index(self.key_cursor1));
429                        self.result.vals.offs.push_ref(off);
430                    }
431                    // Increment cursors in either case; the keys are merged.
432                    self.key_cursor1 += 1;
433                    self.key_cursor2 += 1;
434                },
435                Ordering::Greater => {
436                    self.copy_key(source2, self.key_cursor2);
437                    self.key_cursor2 += 1;
438                },
439            }
440        }
441        /// Merge two ranges of values into `self`.
442        ///
443        /// If the compacted result contains values with non-empty updates, the function returns
444        /// an offset that should be recorded to indicate the upper extent of the result values.
445        fn merge_vals(
446            &mut self,
447            (source1, mut lower1, upper1): (&OrdValStorage<L>, usize, usize),
448            (source2, mut lower2, upper2): (&OrdValStorage<L>, usize, usize),
449        ) -> Option<usize> {
450            // Capture the initial number of values to determine if the merge was ultimately non-empty.
451            let init_vals = self.result.vals.vals.len();
452            while lower1 < upper1 && lower2 < upper2 {
453                // We compare values, and fold in updates for the lowest values;
454                // if they are non-empty post-consolidation, we write the value.
455                // We could multi-way merge and it wouldn't be very complicated.
456                use ::std::cmp::Ordering;
457                match source1.vals.get_abs(lower1).cmp(&source2.vals.get_abs(lower2)) {
458                    Ordering::Less => {
459                        // Extend stash by updates, with logical compaction applied.
460                        self.stash_updates_for_val(source1, lower1);
461                        if self.staging.seal(&mut self.result.upds) {
462                            self.result.vals.vals.push_ref(source1.vals.get_abs(lower1));
463                        }
464                        lower1 += 1;
465                    },
466                    Ordering::Equal => {
467                        self.stash_updates_for_val(source1, lower1);
468                        self.stash_updates_for_val(source2, lower2);
469                        if self.staging.seal(&mut self.result.upds) {
470                            self.result.vals.vals.push_ref(source1.vals.get_abs(lower1));
471                        }
472                        lower1 += 1;
473                        lower2 += 1;
474                    },
475                    Ordering::Greater => {
476                        // Extend stash by updates, with logical compaction applied.
477                        self.stash_updates_for_val(source2, lower2);
478                        if self.staging.seal(&mut self.result.upds) {
479                            self.result.vals.vals.push_ref(source2.vals.get_abs(lower2));
480                        }
481                        lower2 += 1;
482                    },
483                }
484            }
485            // Merging is complete, but we may have remaining elements to push.
486            while lower1 < upper1 {
487                self.stash_updates_for_val(source1, lower1);
488                if self.staging.seal(&mut self.result.upds) {
489                    self.result.vals.vals.push_ref(source1.vals.get_abs(lower1));
490                }
491                lower1 += 1;
492            }
493            while lower2 < upper2 {
494                self.stash_updates_for_val(source2, lower2);
495                if self.staging.seal(&mut self.result.upds) {
496                    self.result.vals.vals.push_ref(source2.vals.get_abs(lower2));
497                }
498                lower2 += 1;
499            }
500
501            // Values being pushed indicate non-emptiness.
502            if self.result.vals.vals.len() > init_vals {
503                Some(self.result.vals.vals.len())
504            } else {
505                None
506            }
507        }
508
509        /// Transfer updates for an indexed value in `source` into `self`, with compaction applied.
510        fn stash_updates_for_val(&mut self, source: &OrdValStorage<L>, index: usize) {
511            let (lower, upper) = source.upds.bounds(index);
512            for i in lower .. upper {
513                // NB: Here is where we would need to look back if `lower == upper`.
514                let (time, diff) = source.upds.get_abs(i);
515                use crate::lattice::Lattice;
516                let mut new_time: layout::Time<L> = L::TimeContainer::into_owned(time);
517                new_time.advance_by(self.since.borrow());
518                self.staging.push(new_time, L::DiffContainer::into_owned(diff));
519            }
520        }
521    }
522
523    /// A cursor for navigating a single layer.
524    pub struct OrdValCursor<L: Layout> {
525        /// Absolute position of the current key.
526        key_cursor: usize,
527        /// Absolute position of the current value.
528        val_cursor: usize,
529        /// Phantom marker for Rust happiness.
530        phantom: PhantomData<L>,
531    }
532
533    impl<L: Layout> Cursor for OrdValCursor<L> {
534
535        type Storage = OrdValBatch<L>;
536
537        type KeyContainer = <L as Layout>::KeyContainer;
538        type Key<'a> = <<L as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
539        type ValContainer = <L as Layout>::ValContainer;
540        type Val<'a> = <<L as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
541        type ValOwn = <<L as Layout>::ValContainer as BatchContainer>::Owned;
542        type TimeContainer = <L as Layout>::TimeContainer;
543        type TimeGat<'a> = <<L as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
544        type Time = <<L as Layout>::TimeContainer as BatchContainer>::Owned;
545        type DiffContainer = <L as Layout>::DiffContainer;
546        type DiffGat<'a> = <<L as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
547        type Diff = <<L as Layout>::DiffContainer as BatchContainer>::Owned;
548
549        fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { storage.storage.keys.get(self.key_cursor) }
550        fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { if self.val_valid(storage) { Some(self.val(storage)) } else { None } }
551
552        fn key<'a>(&self, storage: &'a OrdValBatch<L>) -> Self::Key<'a> { storage.storage.keys.index(self.key_cursor) }
553        fn val<'a>(&self, storage: &'a OrdValBatch<L>) -> Self::Val<'a> { storage.storage.vals.get_abs(self.val_cursor) }
554        fn map_times<L2: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &OrdValBatch<L>, mut logic: L2) {
555            let (lower, upper) = storage.storage.upds.bounds(self.val_cursor);
556            for index in lower .. upper {
557                let (time, diff) = storage.storage.upds.get_abs(index);
558                logic(time, diff);
559            }
560        }
561        fn key_valid(&self, storage: &OrdValBatch<L>) -> bool { self.key_cursor < storage.storage.keys.len() }
562        fn val_valid(&self, storage: &OrdValBatch<L>) -> bool { self.val_cursor < storage.storage.vals.bounds(self.key_cursor).1 }
563        fn step_key(&mut self, storage: &OrdValBatch<L>){
564            self.key_cursor += 1;
565            if self.key_valid(storage) {
566                self.rewind_vals(storage);
567            }
568            else {
569                self.key_cursor = storage.storage.keys.len();
570            }
571        }
572        fn seek_key(&mut self, storage: &OrdValBatch<L>, key: Self::Key<'_>) {
573            self.key_cursor += storage.storage.keys.advance(self.key_cursor, storage.storage.keys.len(), |x| <L::KeyContainer as BatchContainer>::reborrow(x).lt(&<L::KeyContainer as BatchContainer>::reborrow(key)));
574            if self.key_valid(storage) {
575                self.rewind_vals(storage);
576            }
577        }
578        fn step_val(&mut self, storage: &OrdValBatch<L>) {
579            self.val_cursor += 1;
580            if !self.val_valid(storage) {
581                self.val_cursor = storage.storage.vals.bounds(self.key_cursor).1;
582            }
583        }
584        fn seek_val(&mut self, storage: &OrdValBatch<L>, val: Self::Val<'_>) {
585            self.val_cursor += storage.storage.vals.vals.advance(self.val_cursor, storage.storage.vals.bounds(self.key_cursor).1, |x| <L::ValContainer as BatchContainer>::reborrow(x).lt(&<L::ValContainer as BatchContainer>::reborrow(val)));
586        }
587        fn rewind_keys(&mut self, storage: &OrdValBatch<L>) {
588            self.key_cursor = 0;
589            if self.key_valid(storage) {
590                self.rewind_vals(storage)
591            }
592        }
593        fn rewind_vals(&mut self, storage: &OrdValBatch<L>) {
594            self.val_cursor = storage.storage.vals.bounds(self.key_cursor).0;
595        }
596    }
597
598    /// A builder for creating layers from unsorted update tuples.
599    pub struct OrdValBuilder<L: Layout, CI> {
600        /// The in-progress result.
601        ///
602        /// This is public to allow container implementors to set and inspect their container.
603        pub result: OrdValStorage<L>,
604        staging: UpdsBuilder<L::TimeContainer, L::DiffContainer>,
605        _marker: PhantomData<CI>,
606    }
607
608    impl<L, CI> OrdValBuilder<L, CI>
609    where
610        L: Layout,
611    {
612        /// Allocates a builder with capacity for the specified keys, values, and updates.
613        ///
614        /// They represent respectively the number of distinct `key`, `(key, val)`, and total updates.
615        pub fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self {
616            Self {
617                result: OrdValStorage {
618                    keys: L::KeyContainer::with_capacity(keys),
619                    vals: Vals::with_capacity(keys + 1, vals),
620                    upds: Upds::with_capacity(vals + 1, upds),
621                },
622                staging: UpdsBuilder::default(),
623                _marker: PhantomData,
624            }
625        }
626    }
627
628    impl<L: Layout, CI> Default for OrdValBuilder<L, CI> {
629        fn default() -> Self { Self::with_capacity(0, 0, 0) }
630    }
631
632    impl<L, CI> Builder for OrdValBuilder<L, CI>
633    where
634        L: for<'a> Layout<
635            KeyContainer: PushInto<CI::Key<'a>>,
636            ValContainer: PushInto<CI::Val<'a>>,
637        >,
638        CI: for<'a> BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
639    {
640
641        type Input = CI;
642        type Time = layout::Time<L>;
643        type Output = OrdValBatch<L>;
644
645        #[inline]
646        fn push(&mut self, chunk: &mut Self::Input) {
647            for item in chunk.drain() {
648                let (key, val, time, diff) = CI::into_parts(item);
649
650                // Pre-load the first update.
651                if self.result.keys.is_empty() {
652                    self.result.vals.vals.push_into(val);
653                    self.result.keys.push_into(key);
654                    self.staging.push(time, diff);
655                }
656                // Perhaps this is a continuation of an already received key.
657                else if self.result.keys.last().map(|k| CI::key_eq(&key, k)).unwrap_or(false) {
658                    // Perhaps this is a continuation of an already received value.
659                    if self.result.vals.vals.last().map(|v| CI::val_eq(&val, v)).unwrap_or(false) {
660                        self.staging.push(time, diff);
661                    } else {
662                        // New value; complete representation of prior value.
663                        self.staging.seal(&mut self.result.upds);
664                        self.staging.push(time, diff);
665                        self.result.vals.vals.push_into(val);
666                    }
667                } else {
668                    // New key; complete representation of prior key.
669                    self.staging.seal(&mut self.result.upds);
670                    self.staging.push(time, diff);
671                    self.result.vals.offs.push_ref(self.result.vals.vals.len());
672                    self.result.vals.vals.push_into(val);
673                    self.result.keys.push_into(key);
674                }
675            }
676        }
677
678        #[inline(never)]
679        fn done(mut self) -> Option<Self::Output> {
680            self.staging.seal(&mut self.result.upds);
681            self.result.vals.offs.push_ref(self.result.vals.vals.len());
682            let updates = self.staging.total();
683            (updates > 0).then(|| OrdValBatch { updates, storage: self.result })
684        }
685
686    }
687
688    impl<L, CI> crate::batcher::merge::Sealer<CI> for OrdValBuilder<L, CI>
689    where
690        L: for<'a> Layout<
691            KeyContainer: PushInto<CI::Key<'a>>,
692            ValContainer: PushInto<CI::Val<'a>>,
693        >,
694        CI: for<'a> BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
695    {
696        type Output = OrdValBatch<L>;
697
698        fn seal(chain: &mut Vec<CI>) -> Option<Self::Output> {
699            let (keys, vals, upds) = CI::key_val_upd_counts(&chain[..]);
700            let mut builder = Self::with_capacity(keys, vals, upds);
701            for mut chunk in chain.drain(..) {
702                builder.push(&mut chunk);
703            }
704
705            builder.done()
706        }
707    }
708}
709
710/// Types related to forming batches of keys.
711pub mod key_batch {
712
713    use std::marker::PhantomData;
714    use serde::{Deserialize, Serialize};
715    use timely::container::PushInto;
716    use timely::progress::{Antichain, frontier::AntichainRef};
717
718    use crate::trace::{Builder, Cursor};
719    use crate::trace::implementations::spine_fueled::{SpineBatch, Merger};
720    use crate::trace::implementations::BatchContainer;
721    use crate::trace::implementations::layout;
722    use crate::batcher::merge::Sealer;
723
724    use super::{Layout, Upds, layers::UpdsBuilder, BuilderInput};
725
726    /// An immutable collection of update tuples, from a contiguous interval of logical times.
727    #[derive(Debug, Serialize, Deserialize)]
728    #[serde(bound = "
729        L::KeyContainer: Serialize + for<'a> Deserialize<'a>,
730        L::OffsetContainer: Serialize + for<'a> Deserialize<'a>,
731        L::TimeContainer: Serialize + for<'a> Deserialize<'a>,
732        L::DiffContainer: Serialize + for<'a> Deserialize<'a>,
733    ")]
734    pub struct OrdKeyStorage<L: Layout> {
735        /// An ordered list of keys, corresponding to entries in `keys_offs`.
736        pub keys: L::KeyContainer,
737        /// For each key in `keys`, a list of (time, diff) updates.
738        pub upds: Upds<L::OffsetContainer, L::TimeContainer, L::DiffContainer>,
739    }
740
741    /// An immutable collection of update tuples, from a contiguous interval of logical times.
742    ///
743    /// The `L` parameter captures how the updates should be laid out, and `C` determines which
744    /// merge batcher to select.
745    #[derive(Serialize, Deserialize)]
746    #[serde(bound = "
747        L::KeyContainer: Serialize + for<'a> Deserialize<'a>,
748        L::ValContainer: Serialize + for<'a> Deserialize<'a>,
749        L::OffsetContainer: Serialize + for<'a> Deserialize<'a>,
750        L::TimeContainer: Serialize + for<'a> Deserialize<'a>,
751        L::DiffContainer: Serialize + for<'a> Deserialize<'a>,
752    ")]
753    pub struct OrdKeyBatch<L: Layout> {
754        /// The updates themselves.
755        pub storage: OrdKeyStorage<L>,
756        /// The number of updates reflected in the batch.
757        ///
758        /// We track this separately from `storage` because due to the singleton optimization,
759        /// we may have many more updates than `storage.updates.len()`. It should equal that
760        /// length, plus the number of singleton optimizations employed.
761        pub updates: usize,
762
763        /// Single value to return if asked.
764        pub value: L::ValContainer,
765    }
766
767    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> OrdKeyBatch<L> {
768        /// Creates a container with one value, to slot in to `self.value`.
769        pub fn create_value() -> L::ValContainer {
770            let mut value = L::ValContainer::with_capacity(1);
771            value.push_own(&Default::default());
772            value
773        }
774    }
775
776    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> crate::trace::Navigable for OrdKeyBatch<L> {
777        type Cursor = OrdKeyCursor<L>;
778        fn cursor(&self) -> Self::Cursor {
779            OrdKeyCursor {
780                key_cursor: 0,
781                val_stepped: false,
782                phantom: std::marker::PhantomData,
783            }
784        }
785    }
786
787    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> SpineBatch for OrdKeyBatch<L> {
788        type Time = layout::Time<L>;
789        type Merger = OrdKeyMerger<L>;
790        fn len(&self) -> usize {
791            // Normally this would be `self.updates.len()`, but we have a clever compact encoding.
792            // Perhaps we should count such exceptions to the side, to provide a correct accounting.
793            self.updates
794        }
795    }
796
797    /// State for an in-progress merge.
798    pub struct OrdKeyMerger<L: Layout> {
799        /// Key position to merge next in the first batch.
800        key_cursor1: usize,
801        /// Key position to merge next in the second batch.
802        key_cursor2: usize,
803        /// result that we are currently assembling.
804        result: OrdKeyStorage<L>,
805        /// Frontier through which times should be advanced, from the result's description.
806        since: Antichain<layout::Time<L>>,
807
808        /// Local stash of updates, to use for consolidation.
809        staging: UpdsBuilder<L::TimeContainer, L::DiffContainer>,
810    }
811
812    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> Merger<OrdKeyBatch<L>> for OrdKeyMerger<L> {
813        fn new(batch1: &OrdKeyBatch<L>, batch2: &OrdKeyBatch<L>, compaction_frontier: AntichainRef<layout::Time<L>>) -> Self {
814
815            let batch1 = &batch1.storage;
816            let batch2 = &batch2.storage;
817
818            OrdKeyMerger {
819                key_cursor1: 0,
820                key_cursor2: 0,
821                result: OrdKeyStorage {
822                    keys: L::KeyContainer::merge_capacity(&batch1.keys, &batch2.keys),
823                    upds: Upds::merge_capacity(&batch1.upds, &batch2.upds),
824                },
825                since: compaction_frontier.to_owned(),
826                staging: UpdsBuilder::default(),
827            }
828        }
829        fn done(self) -> Option<OrdKeyBatch<L>> {
830            let updates = self.staging.total();
831            (updates > 0).then(|| OrdKeyBatch {
832                updates,
833                storage: self.result,
834                value: OrdKeyBatch::<L>::create_value(),
835            })
836        }
837        fn work(&mut self, source1: &OrdKeyBatch<L>, source2: &OrdKeyBatch<L>, fuel: &mut isize) {
838
839            // An (incomplete) indication of the amount of work we've done so far.
840            let starting_updates = self.staging.total();
841            let mut effort = 0isize;
842
843            // While both mergees are still active, perform single-key merges.
844            while self.key_cursor1 < source1.storage.keys.len() && self.key_cursor2 < source2.storage.keys.len() && effort < *fuel {
845                self.merge_key(&source1.storage, &source2.storage);
846                // An (incomplete) accounting of the work we've done.
847                effort = (self.staging.total() - starting_updates) as isize;
848            }
849
850            // Merging is complete, and only copying remains.
851            // Key-by-key copying allows effort interruption, and compaction.
852            while self.key_cursor1 < source1.storage.keys.len() && effort < *fuel {
853                self.copy_key(&source1.storage, self.key_cursor1);
854                self.key_cursor1 += 1;
855                effort = (self.staging.total() - starting_updates) as isize;
856            }
857            while self.key_cursor2 < source2.storage.keys.len() && effort < *fuel {
858                self.copy_key(&source2.storage, self.key_cursor2);
859                self.key_cursor2 += 1;
860                effort = (self.staging.total() - starting_updates) as isize;
861            }
862
863            *fuel -= effort;
864        }
865    }
866
867    // Helper methods in support of merging batches.
868    impl<L: Layout> OrdKeyMerger<L> {
869        /// Copy the next key in `source`.
870        ///
871        /// The method extracts the key in `source` at `cursor`, and merges it in to `self`.
872        /// If the result does not wholly cancel, they key will be present in `self` with the
873        /// compacted values and updates.
874        ///
875        /// The caller should be certain to update the cursor, as this method does not do this.
876        fn copy_key(&mut self, source: &OrdKeyStorage<L>, cursor: usize) {
877            self.stash_updates_for_key(source, cursor);
878            if self.staging.seal(&mut self.result.upds) {
879                self.result.keys.push_ref(source.keys.index(cursor));
880            }
881        }
882        /// Merge the next key in each of `source1` and `source2` into `self`, updating the appropriate cursors.
883        ///
884        /// This method only merges a single key. It applies all compaction necessary, and may result in no output
885        /// if the updates cancel either directly or after compaction.
886        fn merge_key(&mut self, source1: &OrdKeyStorage<L>, source2: &OrdKeyStorage<L>) {
887            use ::std::cmp::Ordering;
888            match source1.keys.index(self.key_cursor1).cmp(&source2.keys.index(self.key_cursor2)) {
889                Ordering::Less => {
890                    self.copy_key(source1, self.key_cursor1);
891                    self.key_cursor1 += 1;
892                },
893                Ordering::Equal => {
894                    // Keys are equal; must merge all updates from both sources for this one key.
895                    self.stash_updates_for_key(source1, self.key_cursor1);
896                    self.stash_updates_for_key(source2, self.key_cursor2);
897                    if self.staging.seal(&mut self.result.upds) {
898                        self.result.keys.push_ref(source1.keys.index(self.key_cursor1));
899                    }
900                    // Increment cursors in either case; the keys are merged.
901                    self.key_cursor1 += 1;
902                    self.key_cursor2 += 1;
903                },
904                Ordering::Greater => {
905                    self.copy_key(source2, self.key_cursor2);
906                    self.key_cursor2 += 1;
907                },
908            }
909        }
910
911        /// Transfer updates for an indexed value in `source` into `self`, with compaction applied.
912        fn stash_updates_for_key(&mut self, source: &OrdKeyStorage<L>, index: usize) {
913            let (lower, upper) = source.upds.bounds(index);
914            for i in lower .. upper {
915                // NB: Here is where we would need to look back if `lower == upper`.
916                let (time, diff) = source.upds.get_abs(i);
917                use crate::lattice::Lattice;
918                let mut new_time = L::TimeContainer::into_owned(time);
919                new_time.advance_by(self.since.borrow());
920                self.staging.push(new_time, L::DiffContainer::into_owned(diff));
921            }
922        }
923    }
924
925    /// A cursor for navigating a single layer.
926    pub struct OrdKeyCursor<L: Layout> {
927        /// Absolute position of the current key.
928        key_cursor: usize,
929        /// If the value has been stepped for the key, there are no more values.
930        val_stepped: bool,
931        /// Phantom marker for Rust happiness.
932        phantom: PhantomData<L>,
933    }
934
935    impl<L: for<'a> Layout<ValContainer: BatchContainer<Owned: Default>>> Cursor for OrdKeyCursor<L> {
936
937        type Storage = OrdKeyBatch<L>;
938
939        type KeyContainer = <L as Layout>::KeyContainer;
940        type Key<'a> = <<L as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
941        type ValContainer = <L as Layout>::ValContainer;
942        type Val<'a> = <<L as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
943        type ValOwn = <<L as Layout>::ValContainer as BatchContainer>::Owned;
944        type TimeContainer = <L as Layout>::TimeContainer;
945        type TimeGat<'a> = <<L as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
946        type Time = <<L as Layout>::TimeContainer as BatchContainer>::Owned;
947        type DiffContainer = <L as Layout>::DiffContainer;
948        type DiffGat<'a> = <<L as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
949        type Diff = <<L as Layout>::DiffContainer as BatchContainer>::Owned;
950
951        fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { storage.storage.keys.get(self.key_cursor) }
952        fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { if self.val_valid(storage) { Some(self.val(storage)) } else { None } }
953
954        fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { storage.storage.keys.index(self.key_cursor) }
955        fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { storage.value.index(0) }
956        fn map_times<L2: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, mut logic: L2) {
957            let (lower, upper) = storage.storage.upds.bounds(self.key_cursor);
958            for index in lower .. upper {
959                let (time, diff) = storage.storage.upds.get_abs(index);
960                logic(time, diff);
961            }
962        }
963        fn key_valid(&self, storage: &Self::Storage) -> bool { self.key_cursor < storage.storage.keys.len() }
964        fn val_valid(&self, _storage: &Self::Storage) -> bool { !self.val_stepped }
965        fn step_key(&mut self, storage: &Self::Storage){
966            self.key_cursor += 1;
967            if self.key_valid(storage) {
968                self.rewind_vals(storage);
969            }
970            else {
971                self.key_cursor = storage.storage.keys.len();
972            }
973        }
974        fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) {
975            self.key_cursor += storage.storage.keys.advance(self.key_cursor, storage.storage.keys.len(), |x| <L::KeyContainer as BatchContainer>::reborrow(x).lt(&<L::KeyContainer as BatchContainer>::reborrow(key)));
976            if self.key_valid(storage) {
977                self.rewind_vals(storage);
978            }
979        }
980        fn step_val(&mut self, _storage: &Self::Storage) {
981            self.val_stepped = true;
982        }
983        fn seek_val(&mut self, _storage: &Self::Storage, _val: Self::Val<'_>) { }
984        fn rewind_keys(&mut self, storage: &Self::Storage) {
985            self.key_cursor = 0;
986            if self.key_valid(storage) {
987                self.rewind_vals(storage)
988            }
989        }
990        fn rewind_vals(&mut self, _storage: &Self::Storage) {
991            self.val_stepped = false;
992        }
993    }
994
995    /// A builder for creating layers from unsorted update tuples.
996    pub struct OrdKeyBuilder<L: Layout, CI> {
997        /// The in-progress result.
998        ///
999        /// This is public to allow container implementors to set and inspect their container.
1000        pub result: OrdKeyStorage<L>,
1001        staging: UpdsBuilder<L::TimeContainer, L::DiffContainer>,
1002        _marker: PhantomData<CI>,
1003    }
1004
1005    impl<L: Layout, CI> OrdKeyBuilder<L, CI> {
1006        /// Allocates a builder with capacity for the specified keys and updates.
1007        ///
1008        /// They represent respectively the number of distinct `key` and total updates.
1009        pub fn with_capacity(keys: usize, _vals: usize, upds: usize) -> Self {
1010            Self {
1011                result: OrdKeyStorage {
1012                    keys: L::KeyContainer::with_capacity(keys),
1013                    upds: Upds::with_capacity(keys+1, upds),
1014                },
1015                staging: UpdsBuilder::default(),
1016                _marker: PhantomData,
1017            }
1018        }
1019    }
1020
1021    impl<L: Layout, CI> Default for OrdKeyBuilder<L, CI> {
1022        fn default() -> Self { Self::with_capacity(0, 0, 0) }
1023    }
1024
1025    impl<L: Layout, CI> Builder for OrdKeyBuilder<L, CI>
1026    where
1027        L: for<'a> Layout<KeyContainer: PushInto<CI::Key<'a>>>,
1028        L: Layout<ValContainer: BatchContainer<Owned: Default>>,
1029        CI: BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
1030    {
1031
1032        type Input = CI;
1033        type Time = layout::Time<L>;
1034        type Output = OrdKeyBatch<L>;
1035
1036        #[inline]
1037        fn push(&mut self, chunk: &mut Self::Input) {
1038            for item in chunk.drain() {
1039                let (key, _val, time, diff) = CI::into_parts(item);
1040                if self.result.keys.is_empty() {
1041                    self.result.keys.push_into(key);
1042                    self.staging.push(time, diff);
1043                }
1044                // Perhaps this is a continuation of an already received key.
1045                else if self.result.keys.last().map(|k| CI::key_eq(&key, k)).unwrap_or(false) {
1046                    self.staging.push(time, diff);
1047                } else {
1048                    self.staging.seal(&mut self.result.upds);
1049                    self.staging.push(time, diff);
1050                    self.result.keys.push_into(key);
1051                }
1052            }
1053        }
1054
1055        #[inline(never)]
1056        fn done(mut self) -> Option<Self::Output> {
1057            self.staging.seal(&mut self.result.upds);
1058            let updates = self.staging.total();
1059            (updates > 0).then(|| OrdKeyBatch {
1060                updates,
1061                storage: self.result,
1062                value: OrdKeyBatch::<L>::create_value(),
1063            })
1064        }
1065
1066    }
1067
1068    impl<L: Layout, CI> Sealer<CI> for OrdKeyBuilder<L, CI>
1069    where
1070        L: for<'a> Layout<KeyContainer: PushInto<CI::Key<'a>>>,
1071        L: Layout<ValContainer: BatchContainer<Owned: Default>>,
1072        CI: BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
1073    {
1074        type Output = OrdKeyBatch<L>;
1075
1076        fn seal(chain: &mut Vec<CI>) -> Option<Self::Output> {
1077            let (keys, vals, upds) = CI::key_val_upd_counts(&chain[..]);
1078            let mut builder = Self::with_capacity(keys, vals, upds);
1079            for mut chunk in chain.drain(..) {
1080                builder.push(&mut chunk);
1081            }
1082
1083            builder.done()
1084        }
1085    }
1086
1087}
1088
1089use timely::container::DrainContainer;
1090use timely::progress::Timestamp;
1091use crate::trace::Lattice;
1092use crate::difference::Semigroup;
1093use crate::trace::implementations::BatchContainer;
1094
1095/// Behavior to split an update into principal components.
1096pub trait BuilderInput<K: BatchContainer, V: BatchContainer>: DrainContainer + Sized {
1097    /// Key portion
1098    type Key<'a>: Ord;
1099    /// Value portion
1100    type Val<'a>: Ord;
1101    /// Time
1102    type Time;
1103    /// Diff
1104    type Diff;
1105
1106    /// Split an item into separate parts.
1107    fn into_parts<'a>(item: Self::Item<'a>) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff);
1108
1109    /// Test that the key equals a key in the layout's key container.
1110    fn key_eq(this: &Self::Key<'_>, other: K::ReadItem<'_>) -> bool;
1111
1112    /// Test that the value equals a key in the layout's value container.
1113    fn val_eq(this: &Self::Val<'_>, other: V::ReadItem<'_>) -> bool;
1114
1115    /// Count the number of distinct keys, (key, val) pairs, and total updates.
1116    fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize);
1117}
1118
1119impl<K,KBC,V,VBC,T,R> BuilderInput<KBC, VBC> for Vec<((K, V), T, R)>
1120where
1121    K: Ord + Clone + 'static,
1122    KBC: for<'a> BatchContainer<ReadItem<'a>: PartialEq<&'a K>>,
1123    V: Ord + Clone + 'static,
1124    VBC: for<'a> BatchContainer<ReadItem<'a>: PartialEq<&'a V>>,
1125    T: Timestamp + Lattice + 'static,
1126    R: Ord + Semigroup + 'static,
1127{
1128    type Key<'a> = K;
1129    type Val<'a> = V;
1130    type Time = T;
1131    type Diff = R;
1132
1133    fn into_parts<'a>(((key, val), time, diff): Self::Item<'a>) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff) {
1134        (key, val, time, diff)
1135    }
1136
1137    fn key_eq(this: &K, other: KBC::ReadItem<'_>) -> bool {
1138        KBC::reborrow(other) == this
1139    }
1140
1141    fn val_eq(this: &V, other: VBC::ReadItem<'_>) -> bool {
1142        VBC::reborrow(other) == this
1143    }
1144
1145    fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize) {
1146        let mut keys = 0;
1147        let mut vals = 0;
1148        let mut upds = 0;
1149        let mut prev_keyval = None;
1150        for link in chain.iter() {
1151            for ((key, val), _, _) in link.iter() {
1152                if let Some((p_key, p_val)) = prev_keyval {
1153                    if p_key != key {
1154                        keys += 1;
1155                        vals += 1;
1156                    } else if p_val != val {
1157                        vals += 1;
1158                    }
1159                } else {
1160                    keys += 1;
1161                    vals += 1;
1162                }
1163                upds += 1;
1164                prev_keyval = Some((key, val));
1165            }
1166        }
1167        (keys, vals, upds)
1168    }
1169}
differential_dataflow/trace/implementations/ord_neu.rs

differential_dataflow/trace/implementations/
ord_neu.rs
