Remove flatcontainer, absorb TimelyStack (#570)

Adjust to changes in TimelyDataflow/timely-dataflow#647 which removes
flatcontainer and columnation as Timely doesn't have any opinion around
the types and defers to downstream crates instead.

Ideally, we'd move the TimelyStack to an example, but that would be a
bigger change, so I'm holding off for the moment.

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

Cargo.toml

@@ -42,10 +42,12 @@ graph_map = "0.1"
 bytemuck = "1.18.0"
 
 [dependencies]
-serde = { version = "1.0", features = ["derive"] }
+columnar = "0.3"
+columnation = "0.1.0"
 fnv="1.0.2"
+paste = "1.0"
+serde = { version = "1.0", features = ["derive"] }
 timely = {workspace = true}
-columnar = "0.3"
 
 [workspace.dependencies]
 timely = { version = "0.18", default-features = false }

examples/columnar.rs

@@ -348,7 +348,7 @@ mod builder {
 
 use differential_dataflow::trace::implementations::merge_batcher::MergeBatcher;
 use differential_dataflow::trace::implementations::merge_batcher::ColMerger;
-use timely::container::columnation::TimelyStack;
+use differential_dataflow::containers::TimelyStack;
 
 /// A batcher for columnar storage.
 pub type Col2ValBatcher<K, V, T, R> = MergeBatcher<Column<((K,V),T,R)>, batcher::Chunker<TimelyStack<((K,V),T,R)>>, ColMerger<(K,V),T,R>>;

examples/spines.rs

@@ -64,13 +64,6 @@ fn main() {
                     keys.join_core(&data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                 },
-                "flat" => {
-                    use differential_dataflow::trace::implementations::ord_neu::{FlatKeyBatcherDefault, FlatKeyBuilderDefault, FlatKeySpineDefault};
-                    let data = data.arrange::<FlatKeyBatcherDefault<String,usize,isize,_>, FlatKeyBuilderDefault<String,usize,isize>, FlatKeySpineDefault<String,usize,isize>>();
-                    let keys = keys.arrange::<FlatKeyBatcherDefault<String,usize,isize,_>, FlatKeyBuilderDefault<String,usize,isize>, FlatKeySpineDefault<String,usize,isize>>();
-                    keys.join_core(&data, |_k, (), ()| Option::<()>::None)
-                        .probe_with(&mut probe);
-                }
                 _ => {
                     println!("unrecognized mode: {:?}", mode)
                 }

src/consolidation.rs

@@ -14,11 +14,8 @@ use std::cmp::Ordering;
 use std::collections::VecDeque;
 use timely::Container;
 use timely::container::{ContainerBuilder, PushInto};
-use timely::container::flatcontainer::{FlatStack, Push, Region};
-use timely::container::flatcontainer::impls::tuple::{TupleABCRegion, TupleABRegion};
-use crate::Data;
+use crate::{IntoOwned, Data};
 use crate::difference::{IsZero, Semigroup};
-use crate::trace::cursor::IntoOwned;
 
 /// Sorts and consolidates `vec`.
 ///
@@ -321,36 +318,6 @@ where
     }
 }
 
-impl<K, V, T, R> ConsolidateLayout for FlatStack<TupleABCRegion<TupleABRegion<K, V>, T, R>>
-where
-    for<'a> K: Region + Push<<K as Region>::ReadItem<'a>> + Clone + 'static,
-    for<'a> K::ReadItem<'a>: Ord + Copy,
-    for<'a> V: Region + Push<<V as Region>::ReadItem<'a>> + Clone + 'static,
-    for<'a> V::ReadItem<'a>: Ord + Copy,
-    for<'a> T: Region + Push<<T as Region>::ReadItem<'a>> + Clone + 'static,
-    for<'a> T::ReadItem<'a>: Ord + Copy,
-    R: Region + Push<<R as Region>::Owned> + Clone + 'static,
-    for<'a> R::Owned: Semigroup<R::ReadItem<'a>>,
-{
-    type Key<'a> = (K::ReadItem<'a>, V::ReadItem<'a>, T::ReadItem<'a>) where Self: 'a;
-    type Diff<'a> = R::ReadItem<'a> where Self: 'a;
-    type DiffOwned = R::Owned;
-
-    fn into_parts(((key, val), time, diff): Self::Item<'_>) -> (Self::Key<'_>, Self::Diff<'_>) {
-        ((key, val, time), diff)
-    }
-
-    fn cmp<'a>(((key1, val1), time1, _diff1): &Self::Item<'_>, ((key2, val2), time2, _diff2): &Self::Item<'_>) -> Ordering {
-        (K::reborrow(*key1), V::reborrow(*val1), T::reborrow(*time1)).cmp(&(K::reborrow(*key2), V::reborrow(*val2), T::reborrow(*time2)))
-    }
-
-    fn push_with_diff(&mut self, (key, value, time): Self::Key<'_>, diff: Self::DiffOwned) {
-        self.copy(((key, value), time, diff));
-    }
-}
-
-
-
 #[cfg(test)]
 mod tests {
     use super::*;