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::trace::implementations::merge_batcher::MergeBatcher;
15use crate::trace::implementations::merge_batcher::vec::VecMerger;
16use crate::trace::rc_blanket_impls::RcBuilder;
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<VecMerger<(K, V), T, R>>;
27/// A builder using ordered lists.
28pub type RcOrdValBuilder<K, V, T, R> = RcBuilder<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<VecMerger<(K, ()), T, R>>;
34/// A builder for ordered lists.
35pub type RcOrdKeyBuilder<K, T, R> = RcBuilder<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::{Batch, BatchReader, Builder, Cursor, Description, Merger};
252    use crate::trace::implementations::{BatchContainer, BuilderInput};
253    use crate::trace::implementations::layout;
254
255    use super::{Layout, Vals, Upds, layers::UpdsBuilder};
256
257    /// An immutable collection of update tuples, from a contiguous interval of logical times.
258    #[derive(Debug, Serialize, Deserialize)]
259    #[serde(bound = "
260        L::KeyContainer: Serialize + for<'a> Deserialize<'a>,
261        L::ValContainer: Serialize + for<'a> Deserialize<'a>,
262        L::OffsetContainer: Serialize + for<'a> Deserialize<'a>,
263        L::TimeContainer: Serialize + for<'a> Deserialize<'a>,
264        L::DiffContainer: Serialize + for<'a> Deserialize<'a>,
265    ")]
266    pub struct OrdValStorage<L: Layout> {
267        /// An ordered list of keys.
268        pub keys: L::KeyContainer,
269        /// For each key in `keys`, a list of values.
270        pub vals: Vals<L::OffsetContainer, L::ValContainer>,
271        /// For each val in `vals`, a list of (time, diff) updates.
272        pub upds: Upds<L::OffsetContainer, L::TimeContainer, L::DiffContainer>,
273    }
274
275    /// An immutable collection of update tuples, from a contiguous interval of logical times.
276    ///
277    /// The `L` parameter captures how the updates should be laid out, and `C` determines which
278    /// merge batcher to select.
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        /// Description of the update times this layer represents.
291        pub description: Description<layout::Time<L>>,
292        /// The number of updates reflected in the batch.
293        ///
294        /// We track this separately from `storage` because due to the singleton optimization,
295        /// we may have many more updates than `storage.updates.len()`. It should equal that
296        /// length, plus the number of singleton optimizations employed.
297        pub updates: usize,
298    }
299
300    impl<L: Layout> WithLayout for OrdValBatch<L> {
301        type Layout = L;
302    }
303
304    impl<L: Layout> crate::trace::Navigable for OrdValBatch<L> {
305        type Cursor = OrdValCursor<L>;
306        fn cursor(&self) -> Self::Cursor {
307            OrdValCursor {
308                key_cursor: 0,
309                val_cursor: 0,
310                phantom: PhantomData,
311            }
312        }
313    }
314
315    impl<L: Layout> BatchReader for OrdValBatch<L> {
316
317        type Time = layout::Time<L>;
318        fn len(&self) -> usize {
319            // Normally this would be `self.updates.len()`, but we have a clever compact encoding.
320            // Perhaps we should count such exceptions to the side, to provide a correct accounting.
321            self.updates
322        }
323        fn description(&self) -> &Description<layout::Time<L>> { &self.description }
324    }
325
326    impl<L: Layout> Batch for OrdValBatch<L> {
327        type Merger = OrdValMerger<L>;
328
329        fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef<layout::Time<L>>) -> Self::Merger {
330            OrdValMerger::new(self, other, compaction_frontier)
331        }
332
333        fn empty(lower: Antichain<Self::Time>, upper: Antichain<Self::Time>) -> Self {
334            use timely::progress::Timestamp;
335            Self {
336                storage: OrdValStorage {
337                    keys: L::KeyContainer::with_capacity(0),
338                    vals: Default::default(),
339                    upds: Default::default(),
340                },
341                description: Description::new(lower, upper, Antichain::from_elem(Self::Time::minimum())),
342                updates: 0,
343            }
344        }
345    }
346
347    /// State for an in-progress merge.
348    pub struct OrdValMerger<L: Layout> {
349        /// Key position to merge next in the first batch.
350        key_cursor1: usize,
351        /// Key position to merge next in the second batch.
352        key_cursor2: usize,
353        /// result that we are currently assembling.
354        result: OrdValStorage<L>,
355        /// description
356        description: Description<layout::Time<L>>,
357        /// Staging area to consolidate owned times and diffs, before sealing.
358        staging: UpdsBuilder<L::TimeContainer, L::DiffContainer>,
359    }
360
361    impl<L: Layout> Merger<OrdValBatch<L>> for OrdValMerger<L>
362    where
363        OrdValBatch<L>: Batch<Time=layout::Time<L>>,
364    {
365        fn new(batch1: &OrdValBatch<L>, batch2: &OrdValBatch<L>, compaction_frontier: AntichainRef<layout::Time<L>>) -> Self {
366
367            assert!(batch1.upper() == batch2.lower());
368            use crate::lattice::Lattice;
369            let mut since = batch1.description().since().join(batch2.description().since());
370            since = since.join(&compaction_frontier.to_owned());
371
372            let description = Description::new(batch1.lower().clone(), batch2.upper().clone(), since);
373
374            let batch1 = &batch1.storage;
375            let batch2 = &batch2.storage;
376
377            OrdValMerger {
378                key_cursor1: 0,
379                key_cursor2: 0,
380                result: OrdValStorage {
381                    keys: L::KeyContainer::merge_capacity(&batch1.keys, &batch2.keys),
382                    vals: Vals::merge_capacity(&batch1.vals, &batch2.vals),
383                    upds: Upds::merge_capacity(&batch1.upds, &batch2.upds),
384                },
385                description,
386                staging: UpdsBuilder::default(),
387            }
388        }
389        fn done(self) -> OrdValBatch<L> {
390            OrdValBatch {
391                updates: self.staging.total(),
392                storage: self.result,
393                description: self.description,
394            }
395        }
396        fn work(&mut self, source1: &OrdValBatch<L>, source2: &OrdValBatch<L>, fuel: &mut isize) {
397
398            // An (incomplete) indication of the amount of work we've done so far.
399            let starting_updates = self.staging.total();
400            let mut effort = 0isize;
401
402            // While both mergees are still active, perform single-key merges.
403            while self.key_cursor1 < source1.storage.keys.len() && self.key_cursor2 < source2.storage.keys.len() && effort < *fuel {
404                self.merge_key(&source1.storage, &source2.storage);
405                // An (incomplete) accounting of the work we've done.
406                effort = (self.staging.total() - starting_updates) as isize;
407            }
408
409            // Merging is complete, and only copying remains.
410            // Key-by-key copying allows effort interruption, and compaction.
411            while self.key_cursor1 < source1.storage.keys.len() && effort < *fuel {
412                self.copy_key(&source1.storage, self.key_cursor1);
413                self.key_cursor1 += 1;
414                effort = (self.staging.total() - starting_updates) as isize;
415            }
416            while self.key_cursor2 < source2.storage.keys.len() && effort < *fuel {
417                self.copy_key(&source2.storage, self.key_cursor2);
418                self.key_cursor2 += 1;
419                effort = (self.staging.total() - starting_updates) as isize;
420            }
421
422            *fuel -= effort;
423        }
424    }
425
426    // Helper methods in support of merging batches.
427    impl<L: Layout> OrdValMerger<L> {
428        /// Copy the next key in `source`.
429        ///
430        /// The method extracts the key in `source` at `cursor`, and merges it in to `self`.
431        /// If the result does not wholly cancel, they key will be present in `self` with the
432        /// compacted values and updates.
433        ///
434        /// The caller should be certain to update the cursor, as this method does not do this.
435        fn copy_key(&mut self, source: &OrdValStorage<L>, cursor: usize) {
436            // Capture the initial number of values to determine if the merge was ultimately non-empty.
437            let init_vals = self.result.vals.vals.len();
438            let (mut lower, upper) = source.vals.bounds(cursor);
439            while lower < upper {
440                self.stash_updates_for_val(source, lower);
441                if self.staging.seal(&mut self.result.upds) {
442                    self.result.vals.vals.push_ref(source.vals.get_abs(lower));
443                }
444                lower += 1;
445            }
446
447            // If we have pushed any values, copy the key as well.
448            if self.result.vals.vals.len() > init_vals {
449                self.result.keys.push_ref(source.keys.index(cursor));
450                self.result.vals.offs.push_ref(self.result.vals.vals.len());
451            }
452        }
453        /// Merge the next key in each of `source1` and `source2` into `self`, updating the appropriate cursors.
454        ///
455        /// This method only merges a single key. It applies all compaction necessary, and may result in no output
456        /// if the updates cancel either directly or after compaction.
457        fn merge_key(&mut self, source1: &OrdValStorage<L>, source2: &OrdValStorage<L>) {
458            use ::std::cmp::Ordering;
459            match source1.keys.index(self.key_cursor1).cmp(&source2.keys.index(self.key_cursor2)) {
460                Ordering::Less => {
461                    self.copy_key(source1, self.key_cursor1);
462                    self.key_cursor1 += 1;
463                },
464                Ordering::Equal => {
465                    // Keys are equal; must merge all values from both sources for this one key.
466                    let (lower1, upper1) = source1.vals.bounds(self.key_cursor1);
467                    let (lower2, upper2) = source2.vals.bounds(self.key_cursor2);
468                    if let Some(off) = self.merge_vals((source1, lower1, upper1), (source2, lower2, upper2)) {
469                        self.result.keys.push_ref(source1.keys.index(self.key_cursor1));
470                        self.result.vals.offs.push_ref(off);
471                    }
472                    // Increment cursors in either case; the keys are merged.
473                    self.key_cursor1 += 1;
474                    self.key_cursor2 += 1;
475                },
476                Ordering::Greater => {
477                    self.copy_key(source2, self.key_cursor2);
478                    self.key_cursor2 += 1;
479                },
480            }
481        }
482        /// Merge two ranges of values into `self`.
483        ///
484        /// If the compacted result contains values with non-empty updates, the function returns
485        /// an offset that should be recorded to indicate the upper extent of the result values.
486        fn merge_vals(
487            &mut self,
488            (source1, mut lower1, upper1): (&OrdValStorage<L>, usize, usize),
489            (source2, mut lower2, upper2): (&OrdValStorage<L>, usize, usize),
490        ) -> Option<usize> {
491            // Capture the initial number of values to determine if the merge was ultimately non-empty.
492            let init_vals = self.result.vals.vals.len();
493            while lower1 < upper1 && lower2 < upper2 {
494                // We compare values, and fold in updates for the lowest values;
495                // if they are non-empty post-consolidation, we write the value.
496                // We could multi-way merge and it wouldn't be very complicated.
497                use ::std::cmp::Ordering;
498                match source1.vals.get_abs(lower1).cmp(&source2.vals.get_abs(lower2)) {
499                    Ordering::Less => {
500                        // Extend stash by updates, with logical compaction applied.
501                        self.stash_updates_for_val(source1, lower1);
502                        if self.staging.seal(&mut self.result.upds) {
503                            self.result.vals.vals.push_ref(source1.vals.get_abs(lower1));
504                        }
505                        lower1 += 1;
506                    },
507                    Ordering::Equal => {
508                        self.stash_updates_for_val(source1, lower1);
509                        self.stash_updates_for_val(source2, lower2);
510                        if self.staging.seal(&mut self.result.upds) {
511                            self.result.vals.vals.push_ref(source1.vals.get_abs(lower1));
512                        }
513                        lower1 += 1;
514                        lower2 += 1;
515                    },
516                    Ordering::Greater => {
517                        // Extend stash by updates, with logical compaction applied.
518                        self.stash_updates_for_val(source2, lower2);
519                        if self.staging.seal(&mut self.result.upds) {
520                            self.result.vals.vals.push_ref(source2.vals.get_abs(lower2));
521                        }
522                        lower2 += 1;
523                    },
524                }
525            }
526            // Merging is complete, but we may have remaining elements to push.
527            while lower1 < upper1 {
528                self.stash_updates_for_val(source1, lower1);
529                if self.staging.seal(&mut self.result.upds) {
530                    self.result.vals.vals.push_ref(source1.vals.get_abs(lower1));
531                }
532                lower1 += 1;
533            }
534            while lower2 < upper2 {
535                self.stash_updates_for_val(source2, lower2);
536                if self.staging.seal(&mut self.result.upds) {
537                    self.result.vals.vals.push_ref(source2.vals.get_abs(lower2));
538                }
539                lower2 += 1;
540            }
541
542            // Values being pushed indicate non-emptiness.
543            if self.result.vals.vals.len() > init_vals {
544                Some(self.result.vals.vals.len())
545            } else {
546                None
547            }
548        }
549
550        /// Transfer updates for an indexed value in `source` into `self`, with compaction applied.
551        fn stash_updates_for_val(&mut self, source: &OrdValStorage<L>, index: usize) {
552            let (lower, upper) = source.upds.bounds(index);
553            for i in lower .. upper {
554                // NB: Here is where we would need to look back if `lower == upper`.
555                let (time, diff) = source.upds.get_abs(i);
556                use crate::lattice::Lattice;
557                let mut new_time: layout::Time<L> = L::TimeContainer::into_owned(time);
558                new_time.advance_by(self.description.since().borrow());
559                self.staging.push(new_time, L::DiffContainer::into_owned(diff));
560            }
561        }
562    }
563
564    /// A cursor for navigating a single layer.
565    pub struct OrdValCursor<L: Layout> {
566        /// Absolute position of the current key.
567        key_cursor: usize,
568        /// Absolute position of the current value.
569        val_cursor: usize,
570        /// Phantom marker for Rust happiness.
571        phantom: PhantomData<L>,
572    }
573
574    use crate::trace::implementations::WithLayout;
575    impl<L: Layout> WithLayout for OrdValCursor<L> {
576        type Layout = L;
577    }
578
579    impl<L: Layout> Cursor for OrdValCursor<L> {
580
581        type Storage = OrdValBatch<L>;
582
583        type KeyContainer = <L as Layout>::KeyContainer;
584        type Key<'a> = <<L as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
585        type ValContainer = <L as Layout>::ValContainer;
586        type Val<'a> = <<L as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
587        type ValOwn = <<L as Layout>::ValContainer as BatchContainer>::Owned;
588        type TimeContainer = <L as Layout>::TimeContainer;
589        type TimeGat<'a> = <<L as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
590        type Time = <<L as Layout>::TimeContainer as BatchContainer>::Owned;
591        type DiffContainer = <L as Layout>::DiffContainer;
592        type DiffGat<'a> = <<L as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
593        type Diff = <<L as Layout>::DiffContainer as BatchContainer>::Owned;
594
595        fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { storage.storage.keys.get(self.key_cursor) }
596        fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { if self.val_valid(storage) { Some(self.val(storage)) } else { None } }
597
598        fn key<'a>(&self, storage: &'a OrdValBatch<L>) -> Self::Key<'a> { storage.storage.keys.index(self.key_cursor) }
599        fn val<'a>(&self, storage: &'a OrdValBatch<L>) -> Self::Val<'a> { storage.storage.vals.get_abs(self.val_cursor) }
600        fn map_times<L2: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &OrdValBatch<L>, mut logic: L2) {
601            let (lower, upper) = storage.storage.upds.bounds(self.val_cursor);
602            for index in lower .. upper {
603                let (time, diff) = storage.storage.upds.get_abs(index);
604                logic(time, diff);
605            }
606        }
607        fn key_valid(&self, storage: &OrdValBatch<L>) -> bool { self.key_cursor < storage.storage.keys.len() }
608        fn val_valid(&self, storage: &OrdValBatch<L>) -> bool { self.val_cursor < storage.storage.vals.bounds(self.key_cursor).1 }
609        fn step_key(&mut self, storage: &OrdValBatch<L>){
610            self.key_cursor += 1;
611            if self.key_valid(storage) {
612                self.rewind_vals(storage);
613            }
614            else {
615                self.key_cursor = storage.storage.keys.len();
616            }
617        }
618        fn seek_key(&mut self, storage: &OrdValBatch<L>, key: Self::Key<'_>) {
619            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)));
620            if self.key_valid(storage) {
621                self.rewind_vals(storage);
622            }
623        }
624        fn step_val(&mut self, storage: &OrdValBatch<L>) {
625            self.val_cursor += 1;
626            if !self.val_valid(storage) {
627                self.val_cursor = storage.storage.vals.bounds(self.key_cursor).1;
628            }
629        }
630        fn seek_val(&mut self, storage: &OrdValBatch<L>, val: Self::Val<'_>) {
631            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)));
632        }
633        fn rewind_keys(&mut self, storage: &OrdValBatch<L>) {
634            self.key_cursor = 0;
635            if self.key_valid(storage) {
636                self.rewind_vals(storage)
637            }
638        }
639        fn rewind_vals(&mut self, storage: &OrdValBatch<L>) {
640            self.val_cursor = storage.storage.vals.bounds(self.key_cursor).0;
641        }
642    }
643
644    /// A builder for creating layers from unsorted update tuples.
645    pub struct OrdValBuilder<L: Layout, CI> {
646        /// The in-progress result.
647        ///
648        /// This is public to allow container implementors to set and inspect their container.
649        pub result: OrdValStorage<L>,
650        staging: UpdsBuilder<L::TimeContainer, L::DiffContainer>,
651        _marker: PhantomData<CI>,
652    }
653
654    impl<L, CI> Builder for OrdValBuilder<L, CI>
655    where
656        L: for<'a> Layout<
657            KeyContainer: PushInto<CI::Key<'a>>,
658            ValContainer: PushInto<CI::Val<'a>>,
659        >,
660        CI: for<'a> BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
661    {
662
663        type Input = CI;
664        type Time = layout::Time<L>;
665        type Output = OrdValBatch<L>;
666
667        fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self {
668            Self {
669                result: OrdValStorage {
670                    keys: L::KeyContainer::with_capacity(keys),
671                    vals: Vals::with_capacity(keys + 1, vals),
672                    upds: Upds::with_capacity(vals + 1, upds),
673                },
674                staging: UpdsBuilder::default(),
675                _marker: PhantomData,
676            }
677        }
678
679        #[inline]
680        fn push(&mut self, chunk: &mut Self::Input) {
681            for item in chunk.drain() {
682                let (key, val, time, diff) = CI::into_parts(item);
683
684                // Pre-load the first update.
685                if self.result.keys.is_empty() {
686                    self.result.vals.vals.push_into(val);
687                    self.result.keys.push_into(key);
688                    self.staging.push(time, diff);
689                }
690                // Perhaps this is a continuation of an already received key.
691                else if self.result.keys.last().map(|k| CI::key_eq(&key, k)).unwrap_or(false) {
692                    // Perhaps this is a continuation of an already received value.
693                    if self.result.vals.vals.last().map(|v| CI::val_eq(&val, v)).unwrap_or(false) {
694                        self.staging.push(time, diff);
695                    } else {
696                        // New value; complete representation of prior value.
697                        self.staging.seal(&mut self.result.upds);
698                        self.staging.push(time, diff);
699                        self.result.vals.vals.push_into(val);
700                    }
701                } else {
702                    // New key; complete representation of prior key.
703                    self.staging.seal(&mut self.result.upds);
704                    self.staging.push(time, diff);
705                    self.result.vals.offs.push_ref(self.result.vals.vals.len());
706                    self.result.vals.vals.push_into(val);
707                    self.result.keys.push_into(key);
708                }
709            }
710        }
711
712        #[inline(never)]
713        fn done(mut self, description: Description<Self::Time>) -> OrdValBatch<L> {
714            self.staging.seal(&mut self.result.upds);
715            self.result.vals.offs.push_ref(self.result.vals.vals.len());
716            OrdValBatch {
717                updates: self.staging.total(),
718                storage: self.result,
719                description,
720            }
721        }
722
723        fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {
724            let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);
725            let mut builder = Self::with_capacity(keys, vals, upds);
726            for mut chunk in chain.drain(..) {
727                builder.push(&mut chunk);
728            }
729
730            builder.done(description)
731        }
732    }
733}
734
735/// Types related to forming batches of keys.
736pub mod key_batch {
737
738    use std::marker::PhantomData;
739    use serde::{Deserialize, Serialize};
740    use timely::container::PushInto;
741    use timely::progress::{Antichain, frontier::AntichainRef};
742
743    use crate::trace::{Batch, BatchReader, Builder, Cursor, Description, Merger};
744    use crate::trace::implementations::{BatchContainer, BuilderInput};
745    use crate::trace::implementations::layout;
746
747    use super::{Layout, Upds, layers::UpdsBuilder};
748
749    /// An immutable collection of update tuples, from a contiguous interval of logical times.
750    #[derive(Debug, Serialize, Deserialize)]
751    #[serde(bound = "
752        L::KeyContainer: Serialize + for<'a> Deserialize<'a>,
753        L::OffsetContainer: Serialize + for<'a> Deserialize<'a>,
754        L::TimeContainer: Serialize + for<'a> Deserialize<'a>,
755        L::DiffContainer: Serialize + for<'a> Deserialize<'a>,
756    ")]
757    pub struct OrdKeyStorage<L: Layout> {
758        /// An ordered list of keys, corresponding to entries in `keys_offs`.
759        pub keys: L::KeyContainer,
760        /// For each key in `keys`, a list of (time, diff) updates.
761        pub upds: Upds<L::OffsetContainer, L::TimeContainer, L::DiffContainer>,
762    }
763
764    /// An immutable collection of update tuples, from a contiguous interval of logical times.
765    ///
766    /// The `L` parameter captures how the updates should be laid out, and `C` determines which
767    /// merge batcher to select.
768    #[derive(Serialize, Deserialize)]
769    #[serde(bound = "
770        L::KeyContainer: Serialize + for<'a> Deserialize<'a>,
771        L::ValContainer: Serialize + for<'a> Deserialize<'a>,
772        L::OffsetContainer: Serialize + for<'a> Deserialize<'a>,
773        L::TimeContainer: Serialize + for<'a> Deserialize<'a>,
774        L::DiffContainer: Serialize + for<'a> Deserialize<'a>,
775    ")]
776    pub struct OrdKeyBatch<L: Layout> {
777        /// The updates themselves.
778        pub storage: OrdKeyStorage<L>,
779        /// Description of the update times this layer represents.
780        pub description: Description<layout::Time<L>>,
781        /// The number of updates reflected in the batch.
782        ///
783        /// We track this separately from `storage` because due to the singleton optimization,
784        /// we may have many more updates than `storage.updates.len()`. It should equal that
785        /// length, plus the number of singleton optimizations employed.
786        pub updates: usize,
787
788        /// Single value to return if asked.
789        pub value: L::ValContainer,
790    }
791
792    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> OrdKeyBatch<L> {
793        /// Creates a container with one value, to slot in to `self.value`.
794        pub fn create_value() -> L::ValContainer {
795            let mut value = L::ValContainer::with_capacity(1);
796            value.push_own(&Default::default());
797            value
798        }
799    }
800
801    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> WithLayout for OrdKeyBatch<L> {
802        type Layout = L;
803    }
804
805    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> crate::trace::Navigable for OrdKeyBatch<L> {
806        type Cursor = OrdKeyCursor<L>;
807        fn cursor(&self) -> Self::Cursor {
808            OrdKeyCursor {
809                key_cursor: 0,
810                val_stepped: false,
811                phantom: std::marker::PhantomData,
812            }
813        }
814    }
815
816    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> BatchReader for OrdKeyBatch<L> {
817
818        type Time = layout::Time<L>;
819        fn len(&self) -> usize {
820            // Normally this would be `self.updates.len()`, but we have a clever compact encoding.
821            // Perhaps we should count such exceptions to the side, to provide a correct accounting.
822            self.updates
823        }
824        fn description(&self) -> &Description<layout::Time<L>> { &self.description }
825    }
826
827    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> Batch for OrdKeyBatch<L> {
828        type Merger = OrdKeyMerger<L>;
829
830        fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef<layout::Time<L>>) -> Self::Merger {
831            OrdKeyMerger::new(self, other, compaction_frontier)
832        }
833
834        fn empty(lower: Antichain<Self::Time>, upper: Antichain<Self::Time>) -> Self {
835            use timely::progress::Timestamp;
836            Self {
837                storage: OrdKeyStorage {
838                    keys: L::KeyContainer::with_capacity(0),
839                    upds: Upds::default(),
840                },
841                description: Description::new(lower, upper, Antichain::from_elem(Self::Time::minimum())),
842                updates: 0,
843                value: Self::create_value(),
844            }
845        }
846    }
847
848    /// State for an in-progress merge.
849    pub struct OrdKeyMerger<L: Layout> {
850        /// Key position to merge next in the first batch.
851        key_cursor1: usize,
852        /// Key position to merge next in the second batch.
853        key_cursor2: usize,
854        /// result that we are currently assembling.
855        result: OrdKeyStorage<L>,
856        /// description
857        description: Description<layout::Time<L>>,
858
859        /// Local stash of updates, to use for consolidation.
860        staging: UpdsBuilder<L::TimeContainer, L::DiffContainer>,
861    }
862
863    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> Merger<OrdKeyBatch<L>> for OrdKeyMerger<L>
864    where
865        OrdKeyBatch<L>: Batch<Time=layout::Time<L>>,
866    {
867        fn new(batch1: &OrdKeyBatch<L>, batch2: &OrdKeyBatch<L>, compaction_frontier: AntichainRef<layout::Time<L>>) -> Self {
868
869            assert!(batch1.upper() == batch2.lower());
870            use crate::lattice::Lattice;
871            let mut since = batch1.description().since().join(batch2.description().since());
872            since = since.join(&compaction_frontier.to_owned());
873
874            let description = Description::new(batch1.lower().clone(), batch2.upper().clone(), since);
875
876            let batch1 = &batch1.storage;
877            let batch2 = &batch2.storage;
878
879            OrdKeyMerger {
880                key_cursor1: 0,
881                key_cursor2: 0,
882                result: OrdKeyStorage {
883                    keys: L::KeyContainer::merge_capacity(&batch1.keys, &batch2.keys),
884                    upds: Upds::merge_capacity(&batch1.upds, &batch2.upds),
885                },
886                description,
887                staging: UpdsBuilder::default(),
888            }
889        }
890        fn done(self) -> OrdKeyBatch<L> {
891            OrdKeyBatch {
892                updates: self.staging.total(),
893                storage: self.result,
894                description: self.description,
895                value: OrdKeyBatch::<L>::create_value(),
896            }
897        }
898        fn work(&mut self, source1: &OrdKeyBatch<L>, source2: &OrdKeyBatch<L>, fuel: &mut isize) {
899
900            // An (incomplete) indication of the amount of work we've done so far.
901            let starting_updates = self.staging.total();
902            let mut effort = 0isize;
903
904            // While both mergees are still active, perform single-key merges.
905            while self.key_cursor1 < source1.storage.keys.len() && self.key_cursor2 < source2.storage.keys.len() && effort < *fuel {
906                self.merge_key(&source1.storage, &source2.storage);
907                // An (incomplete) accounting of the work we've done.
908                effort = (self.staging.total() - starting_updates) as isize;
909            }
910
911            // Merging is complete, and only copying remains.
912            // Key-by-key copying allows effort interruption, and compaction.
913            while self.key_cursor1 < source1.storage.keys.len() && effort < *fuel {
914                self.copy_key(&source1.storage, self.key_cursor1);
915                self.key_cursor1 += 1;
916                effort = (self.staging.total() - starting_updates) as isize;
917            }
918            while self.key_cursor2 < source2.storage.keys.len() && effort < *fuel {
919                self.copy_key(&source2.storage, self.key_cursor2);
920                self.key_cursor2 += 1;
921                effort = (self.staging.total() - starting_updates) as isize;
922            }
923
924            *fuel -= effort;
925        }
926    }
927
928    // Helper methods in support of merging batches.
929    impl<L: Layout> OrdKeyMerger<L> {
930        /// Copy the next key in `source`.
931        ///
932        /// The method extracts the key in `source` at `cursor`, and merges it in to `self`.
933        /// If the result does not wholly cancel, they key will be present in `self` with the
934        /// compacted values and updates.
935        ///
936        /// The caller should be certain to update the cursor, as this method does not do this.
937        fn copy_key(&mut self, source: &OrdKeyStorage<L>, cursor: usize) {
938            self.stash_updates_for_key(source, cursor);
939            if self.staging.seal(&mut self.result.upds) {
940                self.result.keys.push_ref(source.keys.index(cursor));
941            }
942        }
943        /// Merge the next key in each of `source1` and `source2` into `self`, updating the appropriate cursors.
944        ///
945        /// This method only merges a single key. It applies all compaction necessary, and may result in no output
946        /// if the updates cancel either directly or after compaction.
947        fn merge_key(&mut self, source1: &OrdKeyStorage<L>, source2: &OrdKeyStorage<L>) {
948            use ::std::cmp::Ordering;
949            match source1.keys.index(self.key_cursor1).cmp(&source2.keys.index(self.key_cursor2)) {
950                Ordering::Less => {
951                    self.copy_key(source1, self.key_cursor1);
952                    self.key_cursor1 += 1;
953                },
954                Ordering::Equal => {
955                    // Keys are equal; must merge all updates from both sources for this one key.
956                    self.stash_updates_for_key(source1, self.key_cursor1);
957                    self.stash_updates_for_key(source2, self.key_cursor2);
958                    if self.staging.seal(&mut self.result.upds) {
959                        self.result.keys.push_ref(source1.keys.index(self.key_cursor1));
960                    }
961                    // Increment cursors in either case; the keys are merged.
962                    self.key_cursor1 += 1;
963                    self.key_cursor2 += 1;
964                },
965                Ordering::Greater => {
966                    self.copy_key(source2, self.key_cursor2);
967                    self.key_cursor2 += 1;
968                },
969            }
970        }
971
972        /// Transfer updates for an indexed value in `source` into `self`, with compaction applied.
973        fn stash_updates_for_key(&mut self, source: &OrdKeyStorage<L>, index: usize) {
974            let (lower, upper) = source.upds.bounds(index);
975            for i in lower .. upper {
976                // NB: Here is where we would need to look back if `lower == upper`.
977                let (time, diff) = source.upds.get_abs(i);
978                use crate::lattice::Lattice;
979                let mut new_time = L::TimeContainer::into_owned(time);
980                new_time.advance_by(self.description.since().borrow());
981                self.staging.push(new_time, L::DiffContainer::into_owned(diff));
982            }
983        }
984    }
985
986    /// A cursor for navigating a single layer.
987    pub struct OrdKeyCursor<L: Layout> {
988        /// Absolute position of the current key.
989        key_cursor: usize,
990        /// If the value has been stepped for the key, there are no more values.
991        val_stepped: bool,
992        /// Phantom marker for Rust happiness.
993        phantom: PhantomData<L>,
994    }
995
996    use crate::trace::implementations::WithLayout;
997    impl<L: Layout<ValContainer: BatchContainer>> WithLayout for OrdKeyCursor<L> {
998        type Layout = L;
999    }
1000
1001    impl<L: for<'a> Layout<ValContainer: BatchContainer<Owned: Default>>> Cursor for OrdKeyCursor<L> {
1002
1003        type Storage = OrdKeyBatch<L>;
1004
1005        type KeyContainer = <L as Layout>::KeyContainer;
1006        type Key<'a> = <<L as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
1007        type ValContainer = <L as Layout>::ValContainer;
1008        type Val<'a> = <<L as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
1009        type ValOwn = <<L as Layout>::ValContainer as BatchContainer>::Owned;
1010        type TimeContainer = <L as Layout>::TimeContainer;
1011        type TimeGat<'a> = <<L as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
1012        type Time = <<L as Layout>::TimeContainer as BatchContainer>::Owned;
1013        type DiffContainer = <L as Layout>::DiffContainer;
1014        type DiffGat<'a> = <<L as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
1015        type Diff = <<L as Layout>::DiffContainer as BatchContainer>::Owned;
1016
1017        fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { storage.storage.keys.get(self.key_cursor) }
1018        fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { if self.val_valid(storage) { Some(self.val(storage)) } else { None } }
1019
1020        fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { storage.storage.keys.index(self.key_cursor) }
1021        fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { storage.value.index(0) }
1022        fn map_times<L2: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, mut logic: L2) {
1023            let (lower, upper) = storage.storage.upds.bounds(self.key_cursor);
1024            for index in lower .. upper {
1025                let (time, diff) = storage.storage.upds.get_abs(index);
1026                logic(time, diff);
1027            }
1028        }
1029        fn key_valid(&self, storage: &Self::Storage) -> bool { self.key_cursor < storage.storage.keys.len() }
1030        fn val_valid(&self, _storage: &Self::Storage) -> bool { !self.val_stepped }
1031        fn step_key(&mut self, storage: &Self::Storage){
1032            self.key_cursor += 1;
1033            if self.key_valid(storage) {
1034                self.rewind_vals(storage);
1035            }
1036            else {
1037                self.key_cursor = storage.storage.keys.len();
1038            }
1039        }
1040        fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) {
1041            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)));
1042            if self.key_valid(storage) {
1043                self.rewind_vals(storage);
1044            }
1045        }
1046        fn step_val(&mut self, _storage: &Self::Storage) {
1047            self.val_stepped = true;
1048        }
1049        fn seek_val(&mut self, _storage: &Self::Storage, _val: Self::Val<'_>) { }
1050        fn rewind_keys(&mut self, storage: &Self::Storage) {
1051            self.key_cursor = 0;
1052            if self.key_valid(storage) {
1053                self.rewind_vals(storage)
1054            }
1055        }
1056        fn rewind_vals(&mut self, _storage: &Self::Storage) {
1057            self.val_stepped = false;
1058        }
1059    }
1060
1061    /// A builder for creating layers from unsorted update tuples.
1062    pub struct OrdKeyBuilder<L: Layout, CI> {
1063        /// The in-progress result.
1064        ///
1065        /// This is public to allow container implementors to set and inspect their container.
1066        pub result: OrdKeyStorage<L>,
1067        staging: UpdsBuilder<L::TimeContainer, L::DiffContainer>,
1068        _marker: PhantomData<CI>,
1069    }
1070
1071    impl<L: Layout, CI> Builder for OrdKeyBuilder<L, CI>
1072    where
1073        L: for<'a> Layout<KeyContainer: PushInto<CI::Key<'a>>>,
1074        L: Layout<ValContainer: BatchContainer<Owned: Default>>,
1075        CI: BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
1076    {
1077
1078        type Input = CI;
1079        type Time = layout::Time<L>;
1080        type Output = OrdKeyBatch<L>;
1081
1082        fn with_capacity(keys: usize, _vals: usize, upds: usize) -> Self {
1083            Self {
1084                result: OrdKeyStorage {
1085                    keys: L::KeyContainer::with_capacity(keys),
1086                    upds: Upds::with_capacity(keys+1, upds),
1087                },
1088                staging: UpdsBuilder::default(),
1089                _marker: PhantomData,
1090            }
1091        }
1092
1093        #[inline]
1094        fn push(&mut self, chunk: &mut Self::Input) {
1095            for item in chunk.drain() {
1096                let (key, _val, time, diff) = CI::into_parts(item);
1097                if self.result.keys.is_empty() {
1098                    self.result.keys.push_into(key);
1099                    self.staging.push(time, diff);
1100                }
1101                // Perhaps this is a continuation of an already received key.
1102                else if self.result.keys.last().map(|k| CI::key_eq(&key, k)).unwrap_or(false) {
1103                    self.staging.push(time, diff);
1104                } else {
1105                    self.staging.seal(&mut self.result.upds);
1106                    self.staging.push(time, diff);
1107                    self.result.keys.push_into(key);
1108                }
1109            }
1110        }
1111
1112        #[inline(never)]
1113        fn done(mut self, description: Description<Self::Time>) -> OrdKeyBatch<L> {
1114            self.staging.seal(&mut self.result.upds);
1115            OrdKeyBatch {
1116                updates: self.staging.total(),
1117                storage: self.result,
1118                description,
1119                value: OrdKeyBatch::<L>::create_value(),
1120            }
1121        }
1122
1123        fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {
1124            let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);
1125            let mut builder = Self::with_capacity(keys, vals, upds);
1126            for mut chunk in chain.drain(..) {
1127                builder.push(&mut chunk);
1128            }
1129
1130            builder.done(description)
1131        }
1132    }
1133
1134}
differential_dataflow/trace/implementations/ord_neu.rs

differential_dataflow/trace/implementations/
ord_neu.rs
