Remove generics from upsert

Author: frankmcsherry

differential-dataflow/src/operators/arrange/upsert.rs

@@ -59,7 +59,7 @@
 //!         use differential_dataflow::operators::arrange::upsert;
 //!
 //!         let stream = scope.input_from(&mut input);
-//!         let arranged = upsert::arrange_from_upsert::<_, _, _, ValBuilder<Key, Val, _, _>, ValSpine<Key, Val, _, _>>(&stream, &"test");
+//!         let arranged = upsert::arrange_from_upsert::<_, ValBuilder<Key, Val, _, _>, ValSpine<Key, Val, _, _>>(&stream, &"test");
 //!
 //!         arranged
 //!             .as_collection(|k,v| (k.clone(), v.clone()))
@@ -127,21 +127,19 @@ use super::TraceAgent;
 /// This method is only implemented for totally ordered times, as we do not yet
 /// understand what a "sequence" of upserts would mean for partially ordered
 /// timestamps.
-pub fn arrange_from_upsert<G, K, V, Bu, Tr>(
-    stream: &Stream<G, (K, Option<V>, G::Timestamp)>,
+pub fn arrange_from_upsert<G, Bu, Tr>(
+    stream: &Stream<G, (Tr::KeyOwn, Option<Tr::ValOwn>, G::Timestamp)>,
     name: &str,
 ) -> Arranged<G, TraceAgent<Tr>>
 where
     G: Scope<Timestamp=Tr::Time>,
     Tr: for<'a> Trace<
-        KeyOwn = K,
-        ValOwn = V,
+        KeyOwn: ExchangeData+Hashable+std::hash::Hash,
+        ValOwn: ExchangeData,
         Time: TotalOrder+ExchangeData,
         Diff=isize,
     >+'static,
-    K: ExchangeData+Hashable+std::hash::Hash,
-    V: ExchangeData,
-    Bu: Builder<Time=G::Timestamp, Input = Vec<((K, V), Tr::Time, Tr::Diff)>, Output = Tr::Batch>,
+    Bu: Builder<Time=G::Timestamp, Input = Vec<((Tr::KeyOwn, Tr::ValOwn), Tr::Time, Tr::Diff)>, Output = Tr::Batch>,
 {
     let mut reader: Option<TraceAgent<Tr>> = None;
 
@@ -150,7 +148,7 @@ where
 
         let reader = &mut reader;
 
-        let exchange = Exchange::new(move |update: &(K,Option<V>,G::Timestamp)| (update.0).hashed().into());
+        let exchange = Exchange::new(move |update: &(Tr::KeyOwn,Option<Tr::ValOwn>,G::Timestamp)| (update.0).hashed().into());
 
         stream.unary_frontier(exchange, name, move |_capability, info| {
 
@@ -175,7 +173,7 @@ where
             let mut prev_frontier = Antichain::from_elem(<G::Timestamp as Timestamp>::minimum());
 
             // For stashing input upserts, ordered increasing by time (`BinaryHeap` is a max-heap).
-            let mut priority_queue = BinaryHeap::<std::cmp::Reverse<(G::Timestamp, K, Option<V>)>>::new();
+            let mut priority_queue = BinaryHeap::<std::cmp::Reverse<(G::Timestamp, Tr::KeyOwn, Option<Tr::ValOwn>)>>::new();
             let mut updates = Vec::new();
 
             move |input, output| {
@@ -238,7 +236,7 @@ where
                                 for (key, mut list) in to_process {
 
                                     // The prior value associated with the key.
-                                    let mut prev_value: Option<V> = None;
+                                    let mut prev_value: Option<Tr::ValOwn> = None;
 
                                     // Attempt to find the key in the trace.
                                     trace_cursor.seek_key(&trace_storage, Tr::KeyContainer::borrow_as(&key));