Relocation of narrowly used types and traits

differential-dataflow/benches/chunk_bench.rs

@@ -26,7 +26,7 @@ use differential_dataflow::trace::chunk::{merge_chains, Chunk, NavigableChunk};
 use differential_dataflow::trace::chunk::vec::VecChunk;
 use differential_dataflow::columnar::trace::ColChunk;
 use differential_dataflow::trace::cursor::Cursor;
-use differential_dataflow::trace::implementations::chunker::ContainerChunker;
+use differential_dataflow::trace::implementations::merge_batcher::chunker::ContainerChunker;
 
 /// A global allocator that tracks currently-resident bytes, so we can snapshot
 /// the heap footprint of a built chunk chain.

differential-dataflow/examples/spines.rs

@@ -81,7 +81,7 @@ fn main() {
                     use differential_dataflow::Hashable;
                     use differential_dataflow::columnar::trace::{Spine, ColChunk};
                     use differential_dataflow::trace::chunk::ChunkBatcher;
-                    use differential_dataflow::trace::implementations::chunker::ContainerChunker;
+                    use differential_dataflow::trace::implementations::merge_batcher::chunker::ContainerChunker;
                     use differential_dataflow::operators::arrange::arrangement::arrange_core;
                     use timely::dataflow::channels::pact::Exchange;
 

differential-dataflow/src/trace/chunk/mod.rs

@@ -21,7 +21,7 @@
 //! These are the `Batcher` / `Builder` / `Spine` to hand to
 //! [`arrange_core`](crate::operators::arrange::arrangement::arrange_core), along with a
 //! chunker that forms `C` from the input stream — typically
-//! [`ContainerChunker<C>`](crate::trace::implementations::chunker::ContainerChunker).
+//! [`ContainerChunker<C>`](crate::trace::implementations::merge_batcher::chunker::ContainerChunker).
 //! Trace *maintenance* needs only [`Chunk`]; cursor-driven *consumption* of the
 //! arrangement additionally asks `C` for the [`NavigableChunk`] capability.
 //! Everything else here ([`ChunkBatch`], [`ChunkMerger`], [`ChunkBatchMerger`],
@@ -688,7 +688,7 @@ where
 
 }
 
-impl<C> crate::trace::Sealer<C> for ChunkBatchBuilder<C>
+impl<C> crate::trace::implementations::merge_batcher::Sealer<C> for ChunkBatchBuilder<C>
 where
     C: Chunk + Default + 'static,
     C::Time: timely::progress::Timestamp,

differential-dataflow/src/trace/cursor/mod.rs

@@ -181,3 +181,75 @@ pub trait Cursor {
         out
     }
 }
+
+
+/// Blanket implementations for reference counted batches.
+pub mod rc_blanket_impls {
+
+    use std::rc::Rc;
+
+    use super::{Navigable, Cursor};
+
+    impl<B: Navigable> Navigable for Rc<B> {
+        /// The type used to enumerate the batch's contents.
+        type Cursor = RcBatchCursor<B::Cursor>;
+        /// Acquires a cursor to the batch's contents.
+        fn cursor(&self) -> Self::Cursor {
+            RcBatchCursor::new((**self).cursor())
+        }
+    }
+
+    /// Wrapper to provide cursor to nested scope.
+    pub struct RcBatchCursor<C> {
+        cursor: C,
+    }
+
+    impl<C> RcBatchCursor<C> {
+        fn new(cursor: C) -> Self {
+            RcBatchCursor {
+                cursor,
+            }
+        }
+    }
+
+    impl<C: Cursor> Cursor for RcBatchCursor<C> {
+
+        type Storage = Rc<C::Storage>;
+
+        type Key<'a> = C::Key<'a>;
+        type ValOwn = C::ValOwn;
+        type Val<'a> = C::Val<'a>;
+        type Time = C::Time;
+        type TimeGat<'a> = C::TimeGat<'a>;
+        type Diff = C::Diff;
+        type DiffGat<'a> = C::DiffGat<'a>;
+        type KeyContainer = C::KeyContainer;
+        type ValContainer = C::ValContainer;
+        type TimeContainer = C::TimeContainer;
+        type DiffContainer = C::DiffContainer;
+
+        #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) }
+        #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) }
+
+        #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage) }
+        #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage) }
+
+        #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { self.cursor.get_key(storage) }
+        #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(storage) }
+
+        #[inline]
+        fn map_times<L: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L) {
+            self.cursor.map_times(storage, logic)
+        }
+
+        #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage) }
+        #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage, key) }
+
+        #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage) }
+        #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage, val) }
+
+        #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage) }
+        #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage) }
+    }
+
+}

differential-dataflow/src/trace/implementations/chunker.rs → differential-dataflow/src/trace/implementations/merge_batcher/chunker.rs

@@ -1,4 +1,4 @@
-//! Organize streams of data into sorted chunks.
+//! Organizes streams of data into sorted chunks for a merge batcher.
 
 use std::collections::VecDeque;
 

differential-dataflow/src/trace/implementations/merge_batcher.rs → differential-dataflow/src/trace/implementations/merge_batcher/mod.rs

@@ -8,12 +8,14 @@
 //! its `Chu` before merging: forming sorted, consolidated chunks is the first stage of the
 //! batcher's own work rather than something a caller arranges.
 
+pub mod chunker;
+
 use timely::container::{ContainerBuilder, PushInto};
 use timely::progress::frontier::AntichainRef;
 use timely::progress::{frontier::Antichain, Timestamp};
 
 use crate::logging::{BatcherEvent, Logger};
-use crate::trace::{Batcher, Sealer};
+use crate::trace::Batcher;
 
 /// Creates batches from chunks of sorted, consolidated tuples.
 ///
@@ -233,6 +235,25 @@ pub trait Merger: Default {
     fn allocation(_chunk: &Self::Chunk) -> (usize, usize, usize) { (0, 0, 0) }
 }
 
+/// Forms a batch from a whole chain of updates at once.
+///
+/// Named rather than a bare `fn(&mut Vec<C>) -> Option<B>` so that implementors can name the
+/// batch they produce. There is no receiver: the chain goes in and the batch comes out, leaving
+/// nowhere for an update to be retained.
+pub trait Sealer<C> {
+    /// Output batch type.
+    type Output;
+
+    /// Builds a batch from a chain of updates.
+    ///
+    /// This method relies on the chain only containing updates greater or equal to the lower frontier,
+    /// and not greater or equal to the upper frontier, of the interval the caller means to describe.
+    /// Chains must also be sorted and consolidated.
+    ///
+    /// Having the whole chain in hand, an implementor can size itself before it fills.
+    fn seal(chain: &mut Vec<C>) -> Option<Self::Output>;
+}
+
 /// A `Merger` implementation for vector update containers.
 pub mod vec {
 
@@ -402,7 +423,7 @@ mod test {
     use super::MergeBatcher;
     use super::vec::VecMerger;
     use crate::trace::implementations::ord_neu::VecOrdKeyBuilder;
-    use crate::trace::implementations::chunker::ContainerChunker;
+    use crate::trace::implementations::merge_batcher::chunker::ContainerChunker;
 
     type In = Vec<((u64, ()), u64, i64)>;
     type Bt = MergeBatcher<ContainerChunker<In>, VecMerger<(u64, ()), u64, i64>, VecOrdKeyBuilder<u64, u64, i64>>;

differential-dataflow/src/trace/implementations/mod.rs

@@ -42,10 +42,9 @@ pub mod spine_fueled;
 
 pub mod merge_batcher;
 pub mod ord_neu;
-pub mod chunker;
 
 // Opinionated takes on default spines.
-pub use self::chunker::ContainerChunker;
+pub use self::merge_batcher::chunker::ContainerChunker;
 pub use self::ord_neu::OrdValSpine as ValSpine;
 pub use self::ord_neu::OrdValBatcher as ValBatcher;
 pub use self::ord_neu::VecOrdValBuilder as ValBuilder;
@@ -56,8 +55,7 @@ pub use self::ord_neu::VecOrdKeyBuilder as KeyBuilder;
 use std::convert::TryInto;
 
 use serde::{Deserialize, Serialize};
-use timely::container::{DrainContainer, PushInto};
-use timely::progress::Timestamp;
+use timely::container::PushInto;
 
 use crate::lattice::Lattice;
 use crate::difference::Semigroup;
@@ -289,82 +287,6 @@ impl BatchContainer for OffsetList {
     }
 }
 
-/// Behavior to split an update into principal components.
-pub trait BuilderInput<K: BatchContainer, V: BatchContainer>: DrainContainer + Sized {
-    /// Key portion
-    type Key<'a>: Ord;
-    /// Value portion
-    type Val<'a>: Ord;
-    /// Time
-    type Time;
-    /// Diff
-    type Diff;
-
-    /// Split an item into separate parts.
-    fn into_parts<'a>(item: Self::Item<'a>) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff);
-
-    /// Test that the key equals a key in the layout's key container.
-    fn key_eq(this: &Self::Key<'_>, other: K::ReadItem<'_>) -> bool;
-
-    /// Test that the value equals a key in the layout's value container.
-    fn val_eq(this: &Self::Val<'_>, other: V::ReadItem<'_>) -> bool;
-
-    /// Count the number of distinct keys, (key, val) pairs, and total updates.
-    fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize);
-}
-
-impl<K,KBC,V,VBC,T,R> BuilderInput<KBC, VBC> for Vec<((K, V), T, R)>
-where
-    K: Ord + Clone + 'static,
-    KBC: for<'a> BatchContainer<ReadItem<'a>: PartialEq<&'a K>>,
-    V: Ord + Clone + 'static,
-    VBC: for<'a> BatchContainer<ReadItem<'a>: PartialEq<&'a V>>,
-    T: Timestamp + Lattice + 'static,
-    R: Ord + Semigroup + 'static,
-{
-    type Key<'a> = K;
-    type Val<'a> = V;
-    type Time = T;
-    type Diff = R;
-
-    fn into_parts<'a>(((key, val), time, diff): Self::Item<'a>) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff) {
-        (key, val, time, diff)
-    }
-
-    fn key_eq(this: &K, other: KBC::ReadItem<'_>) -> bool {
-        KBC::reborrow(other) == this
-    }
-
-    fn val_eq(this: &V, other: VBC::ReadItem<'_>) -> bool {
-        VBC::reborrow(other) == this
-    }
-
-    fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize) {
-        let mut keys = 0;
-        let mut vals = 0;
-        let mut upds = 0;
-        let mut prev_keyval = None;
-        for link in chain.iter() {
-            for ((key, val), _, _) in link.iter() {
-                if let Some((p_key, p_val)) = prev_keyval {
-                    if p_key != key {
-                        keys += 1;
-                        vals += 1;
-                    } else if p_val != val {
-                        vals += 1;
-                    }
-                } else {
-                    keys += 1;
-                    vals += 1;
-                }
-                upds += 1;
-                prev_keyval = Some((key, val));
-            }
-        }
-        (keys, vals, upds)
-    }
-}
-
 pub use self::containers::{BatchContainer, SliceContainer};
 
 /// Containers for data that resemble `Vec<T>`, with leaner implementations.