differential_dataflow/columnar/
layout.rs

1//! Layout traits for columnar arrangements.
2//!
3//! `ColumnarUpdate` names the four constituent columnar types of an update,
4//! and `ColumnarLayout` glues them into a DD `Layout` backed by `Coltainer`.
5
6use std::fmt::Debug;
7use columnar::Columnar;
8use crate::trace::implementations::{Layout, OffsetList};
9use crate::difference::Semigroup;
10use crate::lattice::Lattice;
11use timely::progress::Timestamp;
12
13/// A layout based on columnar
14pub struct ColumnarLayout<U: ColumnarUpdate> {
15    phantom: std::marker::PhantomData<U>,
16}
17
18impl<K, V, T, R> ColumnarUpdate for (K, V, T, R)
19where
20    K: Columnar<Container: OrdContainer + Default> + Debug + Ord + Clone + 'static,
21    V: Columnar<Container: OrdContainer + Default> + Debug + Ord + Clone + 'static,
22    T: Columnar<Container: OrdContainer + Default> + Debug + Ord + Default + Clone + Lattice + Timestamp,
23    R: Columnar<Container: OrdContainer + Default> + Debug + Ord + Default + Semigroup + 'static,
24{
25    type Key = K;
26    type Val = V;
27    type Time = T;
28    type Diff = R;
29}
30
31impl<U: ColumnarUpdate> Layout for ColumnarLayout<U> {
32    type KeyContainer    = Coltainer<U::Key>;
33    type ValContainer    = Coltainer<U::Val>;
34    type TimeContainer   = Coltainer<U::Time>;
35    type DiffContainer   = Coltainer<U::Diff>;
36    type OffsetContainer = OffsetList;
37}
38
39/// A type that names constituent update types.
40///
41/// We will use their associated `Columnar::Container`
42pub trait ColumnarUpdate : Debug + 'static {
43    /// The key type.
44    type Key:  Columnar<Container: OrdContainer + Default> + Debug + Ord + Clone + 'static;
45    /// The value type.
46    type Val:  Columnar<Container: OrdContainer + Default> + Debug + Ord + Clone + 'static;
47    /// The time type.
48    type Time: Columnar<Container: OrdContainer + Default> + Debug + Ord + Default + Clone + Lattice + Timestamp;
49    /// The difference type.
50    type Diff: Columnar<Container: OrdContainer + Default> + Debug + Ord + Default + Semigroup + 'static;
51}
52
53/// A container whose references can be ordered.
54pub trait OrdContainer : for<'a> columnar::Container<Ref<'a> : Ord> { }
55impl<C: for<'a> columnar::Container<Ref<'a> : Ord>> OrdContainer for C { }
56
57pub use batch_container::Coltainer;
58mod batch_container {
59    //! [`Coltainer`] wraps a columnar container as a DD [`BatchContainer`].
60
61    use columnar::{Borrow, Columnar, Container, Clear, Push, Index, Len};
62    use crate::trace::implementations::BatchContainer;
63
64    /// Container, anchored by `C` to provide an owned type.
65    pub struct Coltainer<C: Columnar> {
66        /// The underlying columnar container.
67        pub container: C::Container,
68    }
69
70    impl<C: Columnar> Default for Coltainer<C> {
71        fn default() -> Self { Self { container: Default::default() } }
72    }
73
74    impl<C: Columnar + Ord + Clone> BatchContainer for Coltainer<C> where for<'a> columnar::Ref<'a, C> : Ord {
75
76        type ReadItem<'a> = columnar::Ref<'a, C>;
77        type Owned = C;
78
79        #[inline(always)] fn into_owned<'a>(item: Self::ReadItem<'a>) -> Self::Owned { C::into_owned(item) }
80        #[inline(always)] fn clone_onto<'a>(item: Self::ReadItem<'a>, other: &mut Self::Owned) { other.copy_from(item) }
81
82        #[inline(always)] fn push_ref(&mut self, item: Self::ReadItem<'_>) { self.container.push(item) }
83        #[inline(always)] fn push_own(&mut self, item: &Self::Owned) { self.container.push(item) }
84
85        /// Clears the container. May not release resources.
86        fn clear(&mut self) { self.container.clear() }
87
88        /// Creates a new container with sufficient capacity.
89        fn with_capacity(_size: usize) -> Self { Self::default() }
90        /// Creates a new container with sufficient capacity.
91        fn merge_capacity(cont1: &Self, cont2: &Self) -> Self {
92            Self {
93                container: <C as Columnar>::Container::with_capacity_for([cont1.container.borrow(), cont2.container.borrow()].into_iter()),
94            }
95         }
96
97        /// Converts a read item into one with a narrower lifetime.
98        #[inline(always)] fn reborrow<'b, 'a: 'b>(item: Self::ReadItem<'a>) -> Self::ReadItem<'b> { columnar::ContainerOf::<C>::reborrow_ref(item) }
99
100        /// Reference to the element at this position.
101        #[inline(always)] fn index(&self, index: usize) -> Self::ReadItem<'_> { self.container.borrow().get(index) }
102
103        #[inline(always)] fn len(&self) -> usize { self.container.len() }
104
105        /// Reports the number of elements satisfying the predicate.
106        ///
107        /// This methods *relies strongly* on the assumption that the predicate
108        /// stays false once it becomes false, a joint property of the predicate
109        /// and the layout of `Self. This allows `advance` to use exponential search to
110        /// count the number of elements in time logarithmic in the result.
111        fn advance<F: for<'a> Fn(Self::ReadItem<'a>)->bool>(&self, start: usize, end: usize, function: F) -> usize {
112
113            let borrow = self.container.borrow();
114
115            let small_limit = 8;
116
117            // Exponential search if the answer isn't within `small_limit`.
118            if end > start + small_limit && function(borrow.get(start + small_limit)) {
119
120                // start with no advance
121                let mut index = small_limit + 1;
122                if start + index < end && function(borrow.get(start + index)) {
123
124                    // advance in exponentially growing steps.
125                    let mut step = 1;
126                    while start + index + step < end && function(borrow.get(start + index + step)) {
127                        index += step;
128                        step <<= 1;
129                    }
130
131                    // advance in exponentially shrinking steps.
132                    step >>= 1;
133                    while step > 0 {
134                        if start + index + step < end && function(borrow.get(start + index + step)) {
135                            index += step;
136                        }
137                        step >>= 1;
138                    }
139
140                    index += 1;
141                }
142
143                index
144            }
145            else {
146                let limit = std::cmp::min(end, start + small_limit);
147                (start .. limit).filter(|x| function(borrow.get(*x))).count()
148            }
149        }
150    }
151}
differential_dataflow/columnar/layout.rs

differential_dataflow/columnar/
layout.rs
