Further tidying up of submitted PRs (#367)

* remove transmute and fix rust warning

while `()` is a ZST the potentially dangling reference is still
undefined behaviour. Making it a static is a trivial fix.

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

* trace: define `BatchReader` with associated types

The `TraceReader` trait uses associated types to define its `Key`,
`Val`, `Time`, `Diff` but the `BatchReader` trait did not, even though
they are very similar in nature. Usually the choice between asssociated
types or generic parameters on a trait is determined by whether or not a
particular type is expected to implement the same trait multiple times.

My starting point was that these two trait should at the very least be
consistent with respect to their structure and either both use generic
parameters or both use associated types.

All the uses in this repo (and also that I can imagine being useful)
don't really need `BatchReader` to be polymorphic for a particular type
and so I chose to change that one to make it consistent with
`TraceReader`.

The result is quite pleasing as in many cases a lot of generic
parameters are erased.

In order to keep this PR short I left the `Cursor` trait untouched, but
I believe a similar transformation would be beneficial there too,
simplifying further many type signatures.

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

* trace: redefine Cursor with associated types

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

* remove transmute and fix rust warning

while `()` is a ZST the potentially dangling reference is still
undefined behaviour. Making it a static is a trivial fix.

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

* simplify generics of batch related traits

* simplify generics of join's deferred struct

* merge CursorDebug trait methods into Cursor

* various code style changes

Co-authored-by: Frank McSherry <fmcsherry@me.com>

Author: petrosagg

dogsdogsdogs/src/operators/count.rs

@@ -4,7 +4,7 @@ use differential_dataflow::{ExchangeData, Collection, Hashable};
 use differential_dataflow::difference::{Monoid, Multiply};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{Cursor, TraceReader, BatchReader};
+use differential_dataflow::trace::TraceReader;
 
 /// Reports a number of extensions to a stream of prefixes.
 ///
@@ -23,8 +23,6 @@ where
     G::Timestamp: Lattice,
     Tr: TraceReader<Val=(), Time=G::Timestamp, R=isize>+Clone+'static,
     Tr::Key: Ord+Hashable+Default,
-    Tr::Batch: BatchReader<Tr::Key, (), Tr::Time, Tr::R>,
-    Tr::Cursor: Cursor<Tr::Key, (), Tr::Time, Tr::R>,
     R: Monoid+Multiply<Output = R>+ExchangeData,
     F: Fn(&P)->Tr::Key+Clone+'static,
     P: ExchangeData,

dogsdogsdogs/src/operators/half_join.rs

@@ -42,7 +42,7 @@ use differential_dataflow::{ExchangeData, Collection, AsCollection, Hashable};
 use differential_dataflow::difference::{Monoid, Semigroup};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{Cursor, TraceReader, BatchReader};
+use differential_dataflow::trace::{Cursor, TraceReader};
 use differential_dataflow::consolidation::{consolidate, consolidate_updates};
 
 /// A binary equijoin that responds to updates on only its first input.
@@ -81,8 +81,6 @@ where
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+ExchangeData,
     Tr::Val: Clone,
-    Tr::Batch: BatchReader<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
-    Tr::Cursor: Cursor<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
     Tr::R: Monoid+ExchangeData,
     FF: Fn(&G::Timestamp) -> G::Timestamp + 'static,
     CF: Fn(&G::Timestamp, &G::Timestamp) -> bool + 'static,
@@ -137,8 +135,6 @@ where
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+ExchangeData,
     Tr::Val: Clone,
-    Tr::Batch: BatchReader<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
-    Tr::Cursor: Cursor<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
     Tr::R: Monoid+ExchangeData,
     FF: Fn(&G::Timestamp) -> G::Timestamp + 'static,
     CF: Fn(&G::Timestamp, &G::Timestamp) -> bool + 'static,

dogsdogsdogs/src/operators/lookup_map.rs

@@ -10,7 +10,7 @@ use differential_dataflow::{ExchangeData, Collection, AsCollection, Hashable};
 use differential_dataflow::difference::{Semigroup, Monoid};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{Cursor, TraceReader, BatchReader};
+use differential_dataflow::trace::{Cursor, TraceReader};
 
 /// Proposes extensions to a stream of prefixes.
 ///
@@ -32,8 +32,6 @@ where
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable,
     Tr::Val: Clone,
-    Tr::Batch: BatchReader<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
-    Tr::Cursor: Cursor<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
     Tr::R: Monoid+ExchangeData,
     F: FnMut(&D, &mut Tr::Key)+Clone+'static,
     D: ExchangeData,

dogsdogsdogs/src/operators/propose.rs

@@ -4,7 +4,7 @@ use differential_dataflow::{ExchangeData, Collection, Hashable};
 use differential_dataflow::difference::{Monoid, Multiply};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{Cursor, TraceReader, BatchReader};
+use differential_dataflow::trace::TraceReader;
 
 /// Proposes extensions to a prefix stream.
 ///
@@ -25,8 +25,6 @@ where
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+Default,
     Tr::Val: Clone,
-    Tr::Batch: BatchReader<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
-    Tr::Cursor: Cursor<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
     Tr::R: Monoid+Multiply<Output = Tr::R>+ExchangeData,
     F: Fn(&P)->Tr::Key+Clone+'static,
     P: ExchangeData,
@@ -58,8 +56,6 @@ where
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+Default,
     Tr::Val: Clone,
-    Tr::Batch: BatchReader<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
-    Tr::Cursor: Cursor<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
     Tr::R: Monoid+Multiply<Output = Tr::R>+ExchangeData,
     F: Fn(&P)->Tr::Key+Clone+'static,
     P: ExchangeData,

dogsdogsdogs/src/operators/validate.rs

@@ -6,7 +6,7 @@ use differential_dataflow::{ExchangeData, Collection};
 use differential_dataflow::difference::{Monoid, Multiply};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{Cursor, TraceReader, BatchReader};
+use differential_dataflow::trace::TraceReader;
 
 /// Proposes extensions to a stream of prefixes.
 ///
@@ -24,8 +24,6 @@ where
     Tr: TraceReader<Key=(K,V), Val=(), Time=G::Timestamp>+Clone+'static,
     K: Ord+Hash+Clone+Default,
     V: ExchangeData+Hash+Default,
-    Tr::Batch: BatchReader<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
-    Tr::Cursor: Cursor<Tr::Key, Tr::Val, Tr::Time, Tr::R>,
     Tr::R: Monoid+Multiply<Output = Tr::R>+ExchangeData,
     F: Fn(&P)->K+Clone+'static,
     P: ExchangeData,

examples/cursors.rs

@@ -43,7 +43,7 @@ use timely::progress::frontier::AntichainRef;
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::arrange::ArrangeByKey;
 use differential_dataflow::operators::*;
-use differential_dataflow::trace::cursor::CursorDebug;
+use differential_dataflow::trace::cursor::Cursor;
 use differential_dataflow::trace::TraceReader;
 
 type Node = u32;

src/algorithms/graphs/bfs.rs

@@ -30,8 +30,6 @@ where
     G::Timestamp: Lattice+Ord,
     N: ExchangeData+Hash,
     Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, R=isize>+Clone+'static,
-    Tr::Batch: crate::trace::BatchReader<N, N, G::Timestamp, Tr::R>+'static,
-    Tr::Cursor: crate::trace::Cursor<N, N, G::Timestamp, Tr::R>+'static,
 {
     // initialize roots as reaching themselves at distance 0
     let nodes = roots.map(|x| (x, 0));

src/algorithms/graphs/bijkstra.rs

@@ -46,8 +46,6 @@ where
     G::Timestamp: Lattice+Ord,
     N: ExchangeData+Hash,
     Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, R=isize>+Clone+'static,
-    Tr::Batch: crate::trace::BatchReader<N, N, G::Timestamp, Tr::R>+'static,
-    Tr::Cursor: crate::trace::Cursor<N, N, G::Timestamp, Tr::R>+'static,
 {
     forward
         .stream

src/algorithms/graphs/propagate.rs

@@ -65,8 +65,6 @@ where
     R: From<i8>,
     L: ExchangeData,
     Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, R=R>+Clone+'static,
-    Tr::Batch: crate::trace::BatchReader<N, N, G::Timestamp, Tr::R>+'static,
-    Tr::Cursor: crate::trace::Cursor<N, N, G::Timestamp, Tr::R>+'static,
     F: Fn(&L)->u64+Clone+'static,
 {
     // Morally the code performs the following iterative computation. However, in the interest of a simplified

src/operators/arrange/agent.rs

@@ -77,7 +77,7 @@ where
     fn get_physical_compaction(&mut self) -> AntichainRef<Tr::Time> {
         self.physical_compaction.borrow()
     }
-    fn cursor_through(&mut self, frontier: AntichainRef<Tr::Time>) -> Option<(Tr::Cursor, <Tr::Cursor as Cursor<Tr::Key, Tr::Val, Tr::Time, Tr::R>>::Storage)> {
+    fn cursor_through(&mut self, frontier: AntichainRef<Tr::Time>) -> Option<(Tr::Cursor, <Tr::Cursor as Cursor>::Storage)> {
         self.trace.borrow_mut().trace.cursor_through(frontier)
     }
     fn map_batches<F: FnMut(&Self::Batch)>(&self, f: F) { self.trace.borrow().trace.map_batches(f) }
@@ -92,7 +92,7 @@ where
     pub fn new(trace: Tr, operator: ::timely::dataflow::operators::generic::OperatorInfo, logging: Option<::logging::Logger>) -> (Self, TraceWriter<Tr>)
     where
         Tr: Trace,
-        Tr::Batch: Batch<Tr::Key,Tr::Val,Tr::Time,Tr::R>,
+        Tr::Batch: Batch,
     {
         let trace = Rc::new(RefCell::new(TraceBox::new(trace)));
         let queues = Rc::new(RefCell::new(Vec::new()));