differential_dataflow/trace/implementations/
mod.rs

1//! Implementations of `Trace` and associated traits.
2//!
3//! The `Trace` trait provides access to an ordered collection of `(key, val, time, diff)` tuples, but
4//! there is substantial flexibility in implementations of this trait. Depending on characteristics of
5//! the data, we may wish to represent the data in different ways. This module contains several of these
6//! implementations, and combiners for merging the results of different traces.
7//!
8//! As examples of implementations,
9//!
10//! *  The `trie` module is meant to represent general update tuples, with no particular assumptions made
11//!    about their contents. It organizes the data first by key, then by val, and then leaves the rest
12//!    in an unordered pile.
13//!
14//! *  The `keys` module is meant for collections whose value type is `()`, which is to say there is no
15//!    (key, val) structure on the records; all of them are just viewed as "keys".
16//!
17//! *  The `time` module is meant for collections with a single time value. This can remove repetition
18//!    from the representation, at the cost of requiring more instances and run-time merging.
19//!
20//! *  The `base` module is meant for collections with a single time value equivalent to the least time.
21//!    These collections must always accumulate to non-negative collections, and as such we can indicate
22//!    the frequency of an element by its multiplicity. This removes both the time and weight from the
23//!    representation, but is only appropriate for a subset (often substantial) of the data.
24//!
25//! Each of these representations is best suited for different data, but they can be combined to get the
26//! benefits of each, as appropriate. There are several `Cursor` combiners, `CursorList` and `CursorPair`,
27//! for homogeneous and inhomogeneous cursors, respectively.
28//!
29//! #Musings
30//!
31//! What is less clear is how to transfer updates between the representations at merge time in a tasteful
32//! way. Perhaps we could put an ordering on the representations, each pair with a dominant representation,
33//! and part of merging the latter filters updates into the former. Although back and forth might be
34//! appealing, more thinking is required to negotiate all of these policies.
35//!
36//! One option would be to require the layer builder to handle these smarts. Merging is currently done by
37//! the layer as part of custom code, but we could make it simply be "iterate through cursor, push results
38//! into 'ordered builder'". Then the builder would be bright enough to emit a "batch" for the composite
39//! trace, rather than just a batch of the type merged.
40
41pub mod spine_fueled;
42
43pub mod merge_batcher;
44pub mod ord_neu;
45
46// Opinionated takes on default spines.
47pub use crate::batcher::merge::chunker::ContainerChunker;
48pub use self::ord_neu::OrdValSpine as ValSpine;
49pub use self::ord_neu::OrdValBatcher as ValBatcher;
50pub use self::ord_neu::VecOrdValBuilder as ValBuilder;
51pub use self::ord_neu::OrdKeySpine as KeySpine;
52pub use self::ord_neu::OrdKeyBatcher as KeyBatcher;
53pub use self::ord_neu::VecOrdKeyBuilder as KeyBuilder;
54
55use std::convert::TryInto;
56
57use serde::{Deserialize, Serialize};
58use timely::container::PushInto;
59
60use crate::lattice::Lattice;
61use crate::difference::Semigroup;
62
63/// A type that names constituent update types.
64pub trait Update {
65    /// Key by which data are grouped.
66    type Key: Ord + Clone + 'static;
67    /// Values associated with the key.
68    type Val: Ord + Clone + 'static;
69    /// Time at which updates occur.
70    type Time: Lattice + timely::progress::Timestamp;
71    /// Way in which updates occur.
72    type Diff: Ord + Semigroup + 'static;
73}
74
75impl<K,V,T,R> Update for ((K, V), T, R)
76where
77    K: Ord+Clone+'static,
78    V: Ord+Clone+'static,
79    T: Lattice+timely::progress::Timestamp,
80    R: Ord+Semigroup+'static,
81{
82    type Key = K;
83    type Val = V;
84    type Time = T;
85    type Diff = R;
86}
87
88/// A type with opinions on how updates should be laid out.
89pub trait Layout {
90    /// Container for update keys.
91    type KeyContainer: BatchContainer;
92    /// Container for update vals.
93    type ValContainer: BatchContainer;
94    /// Container for times.
95    type TimeContainer: BatchContainer<Owned: Lattice + timely::progress::Timestamp>;
96    /// Container for diffs.
97    type DiffContainer: BatchContainer<Owned: Semigroup + 'static>;
98    /// Container for offsets.
99    type OffsetContainer: for<'a> BatchContainer<ReadItem<'a> = usize>;
100}
101
102// An easy way to provide an explicit layout: as a 5-tuple.
103// Valuable when one wants to perform layout surgery.
104impl<KC, VC, TC, DC, OC> Layout for (KC, VC, TC, DC, OC)
105where
106    KC: BatchContainer,
107    VC: BatchContainer,
108    TC: BatchContainer<Owned: Lattice + timely::progress::Timestamp>,
109    DC: BatchContainer<Owned: Semigroup>,
110    OC: for<'a> BatchContainer<ReadItem<'a> = usize>,
111{
112    type KeyContainer = KC;
113    type ValContainer = VC;
114    type TimeContainer = TC;
115    type DiffContainer = DC;
116    type OffsetContainer = OC;
117}
118
119/// Aliases for types provided by the containers within a `Layout`.
120///
121/// For clarity, prefer `use layout;` and then `layout::Key<L>` to retain the layout context.
122pub mod layout {
123    use crate::trace::implementations::{BatchContainer, Layout};
124
125    /// Alias for an owned key of a layout.
126    pub type Key<L> = <<L as Layout>::KeyContainer as BatchContainer>::Owned;
127    /// Alias for an borrowed key of a layout.
128    pub type KeyRef<'a, L> = <<L as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
129    /// Alias for an owned val of a layout.
130    pub type Val<L> = <<L as Layout>::ValContainer as BatchContainer>::Owned;
131    /// Alias for an borrowed val of a layout.
132    pub type ValRef<'a, L> = <<L as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
133    /// Alias for an owned time of a layout.
134    pub type Time<L> = <<L as Layout>::TimeContainer as BatchContainer>::Owned;
135    /// Alias for an borrowed time of a layout.
136    pub type TimeRef<'a, L> = <<L as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
137    /// Alias for an owned diff of a layout.
138    pub type Diff<L> = <<L as Layout>::DiffContainer as BatchContainer>::Owned;
139    /// Alias for an borrowed diff of a layout.
140    pub type DiffRef<'a, L> = <<L as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
141}
142
143/// A layout that uses vectors
144pub struct Vector<U: Update> {
145    phantom: std::marker::PhantomData<U>,
146}
147
148impl<U: Update<Diff: Ord>> Layout for Vector<U> {
149    type KeyContainer = Vec<U::Key>;
150    type ValContainer = Vec<U::Val>;
151    type TimeContainer = Vec<U::Time>;
152    type DiffContainer = Vec<U::Diff>;
153    type OffsetContainer = OffsetList;
154}
155
156/// A list of unsigned integers that uses `u32` elements as long as they are small enough, and switches to `u64` once they are not.
157#[derive(Eq, PartialEq, Ord, PartialOrd, Clone, Serialize, Deserialize)]
158pub struct OffsetList {
159    /// Length of a prefix of zero elements.
160    pub zero_prefix: usize,
161    /// Offsets that fit within a `u32`.
162    pub smol: Vec<u32>,
163    /// Offsets that either do not fit in a `u32`, or are inserted after some offset that did not fit.
164    pub chonk: Vec<u64>,
165}
166
167impl std::fmt::Debug for OffsetList {
168    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
169        f.debug_list().entries(self.into_iter()).finish()
170    }
171}
172
173impl OffsetList {
174    /// Allocate a new list with a specified capacity.
175    pub fn with_capacity(cap: usize) -> Self {
176        Self {
177            zero_prefix: 0,
178            smol: Vec::with_capacity(cap),
179            chonk: Vec::new(),
180        }
181    }
182    /// Inserts the offset, as a `u32` if that is still on the table.
183    pub fn push(&mut self, offset: usize) {
184        if self.smol.is_empty() && self.chonk.is_empty() && offset == 0 {
185            self.zero_prefix += 1;
186        }
187        else if self.chonk.is_empty() {
188            if let Ok(smol) = offset.try_into() {
189                self.smol.push(smol);
190            }
191            else {
192                self.chonk.push(offset.try_into().unwrap())
193            }
194        }
195        else {
196            self.chonk.push(offset.try_into().unwrap())
197        }
198    }
199    /// Like `std::ops::Index`, which we cannot implement as it must return a `&usize`.
200    pub fn index(&self, index: usize) -> usize {
201        if index < self.zero_prefix {
202            0
203        }
204        else if index - self.zero_prefix < self.smol.len() {
205            self.smol[index - self.zero_prefix].try_into().unwrap()
206        }
207        else {
208            self.chonk[index - self.zero_prefix - self.smol.len()].try_into().unwrap()
209        }
210    }
211    /// The number of offsets in the list.
212    pub fn len(&self) -> usize {
213        self.zero_prefix + self.smol.len() + self.chonk.len()
214    }
215}
216
217impl<'a> IntoIterator for &'a OffsetList {
218    type Item = usize;
219    type IntoIter = OffsetListIter<'a>;
220
221    fn into_iter(self) -> Self::IntoIter {
222        OffsetListIter {list: self, index: 0 }
223    }
224}
225
226/// An iterator for [`OffsetList`].
227pub struct OffsetListIter<'a> {
228    list: &'a OffsetList,
229    index: usize,
230}
231
232impl<'a> Iterator for OffsetListIter<'a> {
233    type Item = usize;
234
235    fn next(&mut self) -> Option<Self::Item> {
236        if self.index < self.list.len() {
237            let res = Some(self.list.index(self.index));
238            self.index += 1;
239            res
240        } else {
241            None
242        }
243    }
244}
245
246impl PushInto<usize> for OffsetList {
247    fn push_into(&mut self, item: usize) {
248        self.push(item);
249    }
250}
251
252impl BatchContainer for OffsetList {
253    type Owned = usize;
254    type ReadItem<'a> = usize;
255
256    #[inline(always)]
257    fn into_owned<'a>(item: Self::ReadItem<'a>) -> Self::Owned { item }
258    #[inline(always)]
259    fn reborrow<'b, 'a: 'b>(item: Self::ReadItem<'a>) -> Self::ReadItem<'b> { item }
260
261    fn push_ref(&mut self, item: Self::ReadItem<'_>) { self.push_into(item) }
262    fn push_own(&mut self, item: &Self::Owned) { self.push_into(*item) }
263
264    fn clear(&mut self) { self.zero_prefix = 0; self.smol.clear(); self.chonk.clear(); }
265
266    fn with_capacity(size: usize) -> Self {
267        Self::with_capacity(size)
268    }
269
270    fn merge_capacity(cont1: &Self, cont2: &Self) -> Self {
271        Self::with_capacity(cont1.len() + cont2.len())
272    }
273
274    fn index(&self, index: usize) -> Self::ReadItem<'_> {
275        self.index(index)
276    }
277
278    fn len(&self) -> usize {
279        self.len()
280    }
281}
282
283pub use self::containers::{BatchContainer, SliceContainer};
284
285/// Containers for data that resemble `Vec<T>`, with leaner implementations.
286pub mod containers {
287
288    use timely::container::PushInto;
289
290    /// A general-purpose container resembling `Vec<T>`.
291    pub trait BatchContainer: 'static {
292        /// An owned instance of `Self::ReadItem<'_>`.
293        type Owned: Clone + Ord;
294
295        /// The type that can be read back out of the container.
296        type ReadItem<'a>: Copy + Ord;
297
298
299        /// Conversion from an instance of this type to the owned type.
300        #[must_use]
301        fn into_owned<'a>(item: Self::ReadItem<'a>) -> Self::Owned;
302        /// Clones `self` onto an existing instance of the owned type.
303        #[inline(always)]
304        fn clone_onto<'a>(item: Self::ReadItem<'a>, other: &mut Self::Owned) {
305            *other = Self::into_owned(item);
306        }
307
308        /// Push an item into this container
309        fn push_ref(&mut self, item: Self::ReadItem<'_>);
310        /// Push an item into this container
311        fn push_own(&mut self, item: &Self::Owned);
312
313        /// Clears the container. May not release resources.
314        fn clear(&mut self);
315
316        /// Creates a new container with sufficient capacity.
317        fn with_capacity(size: usize) -> Self;
318        /// Creates a new container with sufficient capacity.
319        fn merge_capacity(cont1: &Self, cont2: &Self) -> Self;
320
321        /// Converts a read item into one with a narrower lifetime.
322        fn reborrow<'b, 'a: 'b>(item: Self::ReadItem<'a>) -> Self::ReadItem<'b>;
323
324        /// Reference to the element at this position.
325        fn index(&self, index: usize) -> Self::ReadItem<'_>;
326
327        /// Reference to the element at this position, if it exists.
328        fn get(&self, index: usize) -> Option<Self::ReadItem<'_>> {
329            if index < self.len() {
330                Some(self.index(index))
331            }
332            else { None }
333        }
334
335        /// Number of contained elements
336        fn len(&self) -> usize;
337        /// Returns the last item if the container is non-empty.
338        fn last(&self) -> Option<Self::ReadItem<'_>> {
339            if self.len() > 0 {
340                Some(self.index(self.len()-1))
341            }
342            else {
343                None
344            }
345        }
346        /// Indicates if the length is zero.
347        fn is_empty(&self) -> bool { self.len() == 0 }
348
349        /// Reports the number of elements satisfying the predicate.
350        ///
351        /// This methods *relies strongly* on the assumption that the predicate
352        /// stays false once it becomes false, a joint property of the predicate
353        /// and the layout of `Self. This allows `advance` to use exponential search to
354        /// count the number of elements in time logarithmic in the result.
355        fn advance<F: for<'a> Fn(Self::ReadItem<'a>)->bool>(&self, start: usize, end: usize, function: F) -> usize {
356
357            let small_limit = 8;
358
359            // Exponential search if the answer isn't within `small_limit`.
360            if end > start + small_limit && function(self.index(start + small_limit)) {
361
362                // start with no advance
363                let mut index = small_limit + 1;
364                if start + index < end && function(self.index(start + index)) {
365
366                    // advance in exponentially growing steps.
367                    let mut step = 1;
368                    while start + index + step < end && function(self.index(start + index + step)) {
369                        index += step;
370                        step <<= 1;
371                    }
372
373                    // advance in exponentially shrinking steps.
374                    step >>= 1;
375                    while step > 0 {
376                        if start + index + step < end && function(self.index(start + index + step)) {
377                            index += step;
378                        }
379                        step >>= 1;
380                    }
381
382                    index += 1;
383                }
384
385                index
386            }
387            else {
388                let limit = std::cmp::min(end, start + small_limit);
389                (start .. limit).filter(|x| function(self.index(*x))).count()
390            }
391        }
392    }
393
394    // All `T: Clone` also implement `ToOwned<Owned = T>`, but without the constraint Rust
395    // struggles to understand why the owned type must be `T` (i.e. the one blanket impl).
396    impl<T: Ord + Clone + 'static> BatchContainer for Vec<T> {
397        type Owned = T;
398        type ReadItem<'a> = &'a T;
399
400        #[inline(always)] fn into_owned<'a>(item: Self::ReadItem<'a>) -> Self::Owned { item.clone() }
401        #[inline(always)] fn clone_onto<'a>(item: Self::ReadItem<'a>, other: &mut Self::Owned) { other.clone_from(item); }
402
403        fn reborrow<'b, 'a: 'b>(item: Self::ReadItem<'a>) -> Self::ReadItem<'b> { item }
404
405        fn push_ref(&mut self, item: Self::ReadItem<'_>) { self.push_into(item) }
406        fn push_own(&mut self, item: &Self::Owned) { self.push_into(item.clone()) }
407
408        fn clear(&mut self) { self.clear() }
409
410        fn with_capacity(size: usize) -> Self {
411            Vec::with_capacity(size)
412        }
413        fn merge_capacity(cont1: &Self, cont2: &Self) -> Self {
414            Vec::with_capacity(cont1.len() + cont2.len())
415        }
416        fn index(&self, index: usize) -> Self::ReadItem<'_> {
417            &self[index]
418        }
419        fn get(&self, index: usize) -> Option<Self::ReadItem<'_>> {
420            <[T]>::get(self, index)
421        }
422        fn len(&self) -> usize {
423            self[..].len()
424        }
425    }
426
427    /// A container that accepts slices `[B::Item]`.
428    pub struct SliceContainer<B> {
429        /// Offsets that bound each contained slice.
430        ///
431        /// The length will be one greater than the number of contained slices,
432        /// starting with zero and ending with `self.inner.len()`.
433        offsets: Vec<usize>,
434        /// An inner container for sequences of `B` that dereferences to a slice.
435        inner: Vec<B>,
436    }
437
438    impl<B: Ord + Clone + 'static> PushInto<&[B]> for SliceContainer<B> {
439        fn push_into(&mut self, item: &[B]) {
440            for x in item.iter() {
441                self.inner.push_into(x);
442            }
443            self.offsets.push(self.inner.len());
444        }
445    }
446
447    impl<B: Ord + Clone + 'static> PushInto<&Vec<B>> for SliceContainer<B> {
448        fn push_into(&mut self, item: &Vec<B>) {
449            self.push_into(&item[..]);
450        }
451    }
452
453    impl<B> BatchContainer for SliceContainer<B>
454    where
455        B: Ord + Clone + Sized + 'static,
456    {
457        type Owned = Vec<B>;
458        type ReadItem<'a> = &'a [B];
459
460        #[inline(always)] fn into_owned<'a>(item: Self::ReadItem<'a>) -> Self::Owned { item.to_vec() }
461        #[inline(always)] fn clone_onto<'a>(item: Self::ReadItem<'a>, other: &mut Self::Owned) { other.clone_from_slice(item); }
462
463        fn reborrow<'b, 'a: 'b>(item: Self::ReadItem<'a>) -> Self::ReadItem<'b> { item }
464
465        fn push_ref(&mut self, item: Self::ReadItem<'_>) { self.push_into(item) }
466        fn push_own(&mut self, item: &Self::Owned) { self.push_into(item) }
467
468        fn clear(&mut self) {
469            self.offsets.clear();
470            self.offsets.push(0);
471            self.inner.clear();
472        }
473
474        fn with_capacity(size: usize) -> Self {
475            let mut offsets = Vec::with_capacity(size + 1);
476            offsets.push(0);
477            Self {
478                offsets,
479                inner: Vec::with_capacity(size),
480            }
481        }
482        fn merge_capacity(cont1: &Self, cont2: &Self) -> Self {
483            let mut offsets = Vec::with_capacity(cont1.inner.len() + cont2.inner.len() + 1);
484            offsets.push(0);
485            Self {
486                offsets,
487                inner: Vec::with_capacity(cont1.inner.len() + cont2.inner.len()),
488            }
489        }
490        fn index(&self, index: usize) -> Self::ReadItem<'_> {
491            let lower = self.offsets[index];
492            let upper = self.offsets[index+1];
493            &self.inner[lower .. upper]
494        }
495        fn len(&self) -> usize {
496            self.offsets.len() - 1
497        }
498    }
499
500    /// Default implementation introduces a first offset.
501    impl<B> Default for SliceContainer<B> {
502        fn default() -> Self {
503            Self {
504                offsets: vec![0],
505                inner: Default::default(),
506            }
507        }
508    }
509}
differential_dataflow/trace/implementations/mod.rs

differential_dataflow/trace/implementations/
mod.rs
