From 5a23ad64d37cb4e7a7a4ce335fdf67dc8d9f06c0 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Mon, 27 Nov 2023 22:53:12 -0500
Subject: [PATCH] Cursor repivoting (#435)

* Make Storage an associated type, again

* Reorient reduce::perkeycompute to depend on cursors

* Pivot history replay to be based on cursors

* Rename Cursor::R associated type to Diff

* Encourage idiomatic cursor use

* Update dogs3
---
 dogsdogsdogs/src/operators/count.rs       |   2 +-
 dogsdogsdogs/src/operators/half_join.rs   |  20 +--
 dogsdogsdogs/src/operators/lookup_map.rs  |   6 +-
 dogsdogsdogs/src/operators/propose.rs     |  12 +-
 dogsdogsdogs/src/operators/validate.rs    |   6 +-
 examples/cursors.rs                       |   2 +-
 src/algorithms/graphs/bfs.rs              |   2 +-
 src/algorithms/graphs/bijkstra.rs         |   2 +-
 src/algorithms/graphs/propagate.rs        |   2 +-
 src/operators/arrange/agent.rs            |   2 +-
 src/operators/arrange/arrangement.rs      |  28 ++--
 src/operators/arrange/upsert.rs           |   4 +-
 src/operators/consolidate.rs              |   2 +-
 src/operators/count.rs                    |  12 +-
 src/operators/join.rs                     | 130 +++++++++-------
 src/operators/mod.rs                      | 106 +++++++------
 src/operators/reduce.rs                   | 179 ++++++++++++----------
 src/operators/threshold.rs                |  11 +-
 src/trace/cursor/cursor_list.rs           |  42 ++---
 src/trace/cursor/cursor_pair.rs           |  32 ++--
 src/trace/cursor/mod.rs                   |  37 +++--
 src/trace/implementations/ord.rs          |  24 +--
 src/trace/implementations/ord_neu.rs      |  12 +-
 src/trace/implementations/rhh.rs          |  12 +-
 src/trace/implementations/spine_fueled.rs |  14 +-
 src/trace/mod.rs                          |  30 ++--
 src/trace/wrappers/enter.rs               |  44 +++---
 src/trace/wrappers/enter_at.rs            |  44 +++---
 src/trace/wrappers/filter.rs              |  44 +++---
 src/trace/wrappers/freeze.rs              |  48 +++---
 src/trace/wrappers/frontier.rs            |  44 +++---
 src/trace/wrappers/rc.rs                  |   2 +-
 32 files changed, 512 insertions(+), 445 deletions(-)

diff --git a/dogsdogsdogs/src/operators/count.rs b/dogsdogsdogs/src/operators/count.rs
index d86fb5089..df26ddee7 100644
--- a/dogsdogsdogs/src/operators/count.rs
+++ b/dogsdogsdogs/src/operators/count.rs
@@ -21,7 +21,7 @@ pub fn count<G, Tr, R, F, P>(
 where
     G: Scope,
     G::Timestamp: Lattice,
-    Tr: TraceReader<Val=(), Time=G::Timestamp, R=isize>+Clone+'static,
+    Tr: TraceReader<Val=(), Time=G::Timestamp, Diff=isize>+Clone+'static,
     Tr::Key: Ord+Hashable+Default,
     R: Monoid+Multiply<Output = R>+ExchangeData,
     F: Fn(&P)->Tr::Key+Clone+'static,
diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs
index a2920c163..b713d07d1 100644
--- a/dogsdogsdogs/src/operators/half_join.rs
+++ b/dogsdogsdogs/src/operators/half_join.rs
@@ -68,12 +68,12 @@ use differential_dataflow::consolidation::{consolidate, consolidate_updates};
 /// once out of the "delta flow region", the updates will be `delay`d to the
 /// times specified in the payloads.
 pub fn half_join<G, V, Tr, FF, CF, DOut, S>(
-    stream: &Collection<G, (Tr::Key, V, G::Timestamp), Tr::R>,
+    stream: &Collection<G, (Tr::Key, V, G::Timestamp), Tr::Diff>,
     arrangement: Arranged<G, Tr>,
     frontier_func: FF,
     comparison: CF,
     mut output_func: S,
-) -> Collection<G, (DOut, G::Timestamp), Tr::R>
+) -> Collection<G, (DOut, G::Timestamp), Tr::Diff>
 where
     G: Scope,
     G::Timestamp: Lattice,
@@ -81,14 +81,14 @@ where
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+ExchangeData,
     Tr::Val: Clone,
-    Tr::R: Monoid+ExchangeData,
+    Tr::Diff: Monoid+ExchangeData,
     FF: Fn(&G::Timestamp) -> G::Timestamp + 'static,
     CF: Fn(&G::Timestamp, &G::Timestamp) -> bool + 'static,
     DOut: Clone+'static,
-    Tr::R: std::ops::Mul<Tr::R, Output=Tr::R>,
+    Tr::Diff: std::ops::Mul<Tr::Diff, Output=Tr::Diff>,
     S: FnMut(&Tr::Key, &V, &Tr::Val)->DOut+'static,
 {
-    let output_func = move |k: &Tr::Key, v1: &V, v2: &Tr::Val, initial: &G::Timestamp, time: &G::Timestamp, diff1: &Tr::R, diff2: &Tr::R| {
+    let output_func = move |k: &Tr::Key, v1: &V, v2: &Tr::Val, initial: &G::Timestamp, time: &G::Timestamp, diff1: &Tr::Diff, diff2: &Tr::Diff| {
         let diff = diff1.clone() * diff2.clone();
         let dout = (output_func(k, v1, v2), time.clone());
         Some((dout, initial.clone(), diff))
@@ -121,7 +121,7 @@ where
 /// records. Note this is not the number of *output* records, owing mainly to
 /// the number of matched records being easiest to record with low overhead.
 pub fn half_join_internal_unsafe<G, V, Tr, FF, CF, DOut, ROut, Y, I, S>(
-    stream: &Collection<G, (Tr::Key, V, G::Timestamp), Tr::R>,
+    stream: &Collection<G, (Tr::Key, V, G::Timestamp), Tr::Diff>,
     mut arrangement: Arranged<G, Tr>,
     frontier_func: FF,
     comparison: CF,
@@ -135,14 +135,14 @@ where
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+ExchangeData,
     Tr::Val: Clone,
-    Tr::R: Monoid+ExchangeData,
+    Tr::Diff: Monoid+ExchangeData,
     FF: Fn(&G::Timestamp) -> G::Timestamp + 'static,
     CF: Fn(&G::Timestamp, &G::Timestamp) -> bool + 'static,
     DOut: Clone+'static,
     ROut: Monoid,
     Y: Fn(std::time::Instant, usize) -> bool + 'static,
     I: IntoIterator<Item=(DOut, G::Timestamp, ROut)>,
-    S: FnMut(&Tr::Key, &V, &Tr::Val, &G::Timestamp, &G::Timestamp, &Tr::R, &Tr::R)-> I + 'static,
+    S: FnMut(&Tr::Key, &V, &Tr::Val, &G::Timestamp, &G::Timestamp, &Tr::Diff, &Tr::Diff)-> I + 'static,
 {
     // No need to block physical merging for this operator.
     arrangement.trace.set_physical_compaction(Antichain::new().borrow());
@@ -152,7 +152,7 @@ where
     let mut stash = HashMap::new();
     let mut buffer = Vec::new();
 
-    let exchange = Exchange::new(move |update: &((Tr::Key, V, G::Timestamp),G::Timestamp,Tr::R)| (update.0).0.hashed().into());
+    let exchange = Exchange::new(move |update: &((Tr::Key, V, G::Timestamp),G::Timestamp,Tr::Diff)| (update.0).0.hashed().into());
 
     // Stash for (time, diff) accumulation.
     let mut output_buffer = Vec::new();
@@ -229,7 +229,7 @@ where
                                     }
                                     cursor.rewind_vals(&storage);
                                 }
-                                *diff1 = Tr::R::zero();
+                                *diff1 = Tr::Diff::zero();
                             }
                         }
 
diff --git a/dogsdogsdogs/src/operators/lookup_map.rs b/dogsdogsdogs/src/operators/lookup_map.rs
index a35472479..9228d2d4d 100644
--- a/dogsdogsdogs/src/operators/lookup_map.rs
+++ b/dogsdogsdogs/src/operators/lookup_map.rs
@@ -32,13 +32,13 @@ where
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+Sized,
     Tr::Val: Clone,
-    Tr::R: Monoid+ExchangeData,
+    Tr::Diff: Monoid+ExchangeData,
     F: FnMut(&D, &mut Tr::Key)+Clone+'static,
     D: ExchangeData,
     R: ExchangeData+Monoid,
     DOut: Clone+'static,
     ROut: Monoid,
-    S: FnMut(&D, &R, &Tr::Val, &Tr::R)->(DOut, ROut)+'static,
+    S: FnMut(&D, &R, &Tr::Val, &Tr::Diff)->(DOut, ROut)+'static,
 {
     // No need to block physical merging for this operator.
     arrangement.trace.set_physical_compaction(Antichain::new().borrow());
@@ -99,7 +99,7 @@ where
                             cursor.seek_key(&storage, &key1);
                             if cursor.get_key(&storage) == Some(&key1) {
                                 while let Some(value) = cursor.get_val(&storage) {
-                                    let mut count = Tr::R::zero();
+                                    let mut count = Tr::Diff::zero();
                                     cursor.map_times(&storage, |t, d| {
                                         if t.less_equal(time) { count.plus_equals(d); }
                                     });
diff --git a/dogsdogsdogs/src/operators/propose.rs b/dogsdogsdogs/src/operators/propose.rs
index 2a5e6747a..6d6ed95f3 100644
--- a/dogsdogsdogs/src/operators/propose.rs
+++ b/dogsdogsdogs/src/operators/propose.rs
@@ -15,17 +15,17 @@ use differential_dataflow::trace::TraceReader;
 /// `arrangement` undergoes. More complicated patterns are also appropriate, as in the case
 /// of delta queries.
 pub fn propose<G, Tr, F, P>(
-    prefixes: &Collection<G, P, Tr::R>,
+    prefixes: &Collection<G, P, Tr::Diff>,
     arrangement: Arranged<G, Tr>,
     key_selector: F,
-) -> Collection<G, (P, Tr::Val), Tr::R>
+) -> Collection<G, (P, Tr::Val), Tr::Diff>
 where
     G: Scope,
     G::Timestamp: Lattice,
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+Default,
     Tr::Val: Clone,
-    Tr::R: Monoid+Multiply<Output = Tr::R>+ExchangeData,
+    Tr::Diff: Monoid+Multiply<Output = Tr::Diff>+ExchangeData,
     F: Fn(&P)->Tr::Key+Clone+'static,
     P: ExchangeData,
 {
@@ -46,17 +46,17 @@ where
 /// prefixes by the number of matches in `arrangement`. This can be useful to
 /// avoid the need to prepare an arrangement of distinct extensions.
 pub fn propose_distinct<G, Tr, F, P>(
-    prefixes: &Collection<G, P, Tr::R>,
+    prefixes: &Collection<G, P, Tr::Diff>,
     arrangement: Arranged<G, Tr>,
     key_selector: F,
-) -> Collection<G, (P, Tr::Val), Tr::R>
+) -> Collection<G, (P, Tr::Val), Tr::Diff>
 where
     G: Scope,
     G::Timestamp: Lattice,
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Ord+Hashable+Default,
     Tr::Val: Clone,
-    Tr::R: Monoid+Multiply<Output = Tr::R>+ExchangeData,
+    Tr::Diff: Monoid+Multiply<Output = Tr::Diff>+ExchangeData,
     F: Fn(&P)->Tr::Key+Clone+'static,
     P: ExchangeData,
 {
diff --git a/dogsdogsdogs/src/operators/validate.rs b/dogsdogsdogs/src/operators/validate.rs
index 41efa8df2..b41ea8687 100644
--- a/dogsdogsdogs/src/operators/validate.rs
+++ b/dogsdogsdogs/src/operators/validate.rs
@@ -14,17 +14,17 @@ use differential_dataflow::trace::TraceReader;
 /// key with `key_selector` and then proposes all pair af the prefix
 /// and values associated with the key in `arrangement`.
 pub fn validate<G, K, V, Tr, F, P>(
-    extensions: &Collection<G, (P, V), Tr::R>,
+    extensions: &Collection<G, (P, V), Tr::Diff>,
     arrangement: Arranged<G, Tr>,
     key_selector: F,
-) -> Collection<G, (P, V), Tr::R>
+) -> Collection<G, (P, V), Tr::Diff>
 where
     G: Scope,
     G::Timestamp: Lattice,
     Tr: TraceReader<Key=(K,V), Val=(), Time=G::Timestamp>+Clone+'static,
     K: Ord+Hash+Clone+Default,
     V: ExchangeData+Hash+Default,
-    Tr::R: Monoid+Multiply<Output = Tr::R>+ExchangeData,
+    Tr::Diff: Monoid+Multiply<Output = Tr::Diff>+ExchangeData,
     F: Fn(&P)->K+Clone+'static,
     P: ExchangeData,
 {
diff --git a/examples/cursors.rs b/examples/cursors.rs
index cd916530c..068a69a42 100644
--- a/examples/cursors.rs
+++ b/examples/cursors.rs
@@ -138,7 +138,7 @@ where
     Tr::Key: Debug + Clone,
     Tr::Val: Debug + Clone,
     Tr::Time: Debug + Clone,
-    Tr::R: Debug + Clone,
+    Tr::Diff: Debug + Clone,
 {
     let (mut cursor, storage) = trace.cursor();
     for ((k, v), diffs) in cursor.to_vec(&storage).iter() {
diff --git a/src/algorithms/graphs/bfs.rs b/src/algorithms/graphs/bfs.rs
index 6774f535f..8f42cb47d 100644
--- a/src/algorithms/graphs/bfs.rs
+++ b/src/algorithms/graphs/bfs.rs
@@ -29,7 +29,7 @@ where
     G: Scope,
     G::Timestamp: Lattice+Ord,
     N: ExchangeData+Hash,
-    Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, R=isize>+Clone+'static,
+    Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, Diff=isize>+Clone+'static,
 {
     // initialize roots as reaching themselves at distance 0
     let nodes = roots.map(|x| (x, 0));
diff --git a/src/algorithms/graphs/bijkstra.rs b/src/algorithms/graphs/bijkstra.rs
index eb803f160..f8a4662be 100644
--- a/src/algorithms/graphs/bijkstra.rs
+++ b/src/algorithms/graphs/bijkstra.rs
@@ -45,7 +45,7 @@ where
     G: Scope,
     G::Timestamp: Lattice+Ord,
     N: ExchangeData+Hash,
-    Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, R=isize>+Clone+'static,
+    Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, Diff=isize>+Clone+'static,
 {
     forward
         .stream
diff --git a/src/algorithms/graphs/propagate.rs b/src/algorithms/graphs/propagate.rs
index b0404c8ee..32947d4c7 100644
--- a/src/algorithms/graphs/propagate.rs
+++ b/src/algorithms/graphs/propagate.rs
@@ -64,7 +64,7 @@ where
     R: Multiply<R, Output=R>,
     R: From<i8>,
     L: ExchangeData,
-    Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, R=R>+Clone+'static,
+    Tr: TraceReader<Key=N, Val=N, Time=G::Timestamp, Diff=R>+Clone+'static,
     F: Fn(&L)->u64+Clone+'static,
 {
     // Morally the code performs the following iterative computation. However, in the interest of a simplified
diff --git a/src/operators/arrange/agent.rs b/src/operators/arrange/agent.rs
index c477adaa3..03ea5af1b 100644
--- a/src/operators/arrange/agent.rs
+++ b/src/operators/arrange/agent.rs
@@ -50,7 +50,7 @@ where
     type Key = Tr::Key;
     type Val = Tr::Val;
     type Time = Tr::Time;
-    type R = Tr::R;
+    type Diff = Tr::Diff;
 
     type Batch = Tr::Batch;
     type Storage = Tr::Storage;
diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index 7373dc188..42bf28f87 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -91,7 +91,7 @@ where
         where
             Tr::Key: 'static,
             Tr::Val: 'static,
-            Tr::R: 'static,
+            Tr::Diff: 'static,
             G::Timestamp: Clone+'static,
             TInner: Refines<G::Timestamp>+Lattice+Timestamp+Clone+'static,
     {
@@ -110,7 +110,7 @@ where
         where
             Tr::Key: 'static,
             Tr::Val: 'static,
-            Tr::R: 'static,
+            Tr::Diff: 'static,
             G::Timestamp: Clone+'static,
     {
         Arranged {
@@ -129,7 +129,7 @@ where
         where
             Tr::Key: 'static,
             Tr::Val: 'static,
-            Tr::R: 'static,
+            Tr::Diff: 'static,
             G::Timestamp: Clone+'static,
             TInner: Refines<G::Timestamp>+Lattice+Timestamp+Clone+'static,
             F: FnMut(&Tr::Key, &Tr::Val, &G::Timestamp)->TInner+Clone+'static,
@@ -179,7 +179,7 @@ where
         where
             Tr::Key: 'static,
             Tr::Val: 'static,
-            Tr::R: 'static,
+            Tr::Diff: 'static,
             G::Timestamp: Clone+'static,
             F: FnMut(&Tr::Key, &Tr::Val)->bool+Clone+'static,
     {
@@ -195,9 +195,9 @@ where
     /// The underlying `Stream<G, BatchWrapper<T::Batch>>` is a much more efficient way to access the data,
     /// and this method should only be used when the data need to be transformed or exchanged, rather than
     /// supplied as arguments to an operator using the same key-value structure.
-    pub fn as_collection<D: Data, L>(&self, mut logic: L) -> Collection<G, D, Tr::R>
+    pub fn as_collection<D: Data, L>(&self, mut logic: L) -> Collection<G, D, Tr::Diff>
         where
-            Tr::R: Semigroup,
+            Tr::Diff: Semigroup,
             L: FnMut(&Tr::Key, &Tr::Val) -> D+'static,
     {
         self.flat_map_ref(move |key, val| Some(logic(key,val)))
@@ -207,9 +207,9 @@ where
     ///
     /// The supplied logic may produce an iterator over output values, allowing either
     /// filtering or flat mapping as part of the extraction.
-    pub fn flat_map_ref<I, L>(&self, logic: L) -> Collection<G, I::Item, Tr::R>
+    pub fn flat_map_ref<I, L>(&self, logic: L) -> Collection<G, I::Item, Tr::Diff>
         where
-            Tr::R: Semigroup,
+            Tr::Diff: Semigroup,
             I: IntoIterator,
             I::Item: Data,
             L: FnMut(&Tr::Key, &Tr::Val) -> I+'static,
@@ -224,9 +224,9 @@ where
     ///
     /// This method exists for streams of batches without the corresponding arrangement.
     /// If you have the arrangement, its `flat_map_ref` method is equivalent to this.
-    pub fn flat_map_batches<I, L>(stream: &Stream<G, Tr::Batch>, mut logic: L) -> Collection<G, I::Item, Tr::R>
+    pub fn flat_map_batches<I, L>(stream: &Stream<G, Tr::Batch>, mut logic: L) -> Collection<G, I::Item, Tr::Diff>
     where
-        Tr::R: Semigroup,
+        Tr::Diff: Semigroup,
         I: IntoIterator,
         I::Item: Data,
         L: FnMut(&Tr::Key, &Tr::Val) -> I+'static,
@@ -258,12 +258,12 @@ where
     ///
     /// This method consumes a stream of (key, time) queries and reports the corresponding stream of
     /// (key, value, time, diff) accumulations in the `self` trace.
-    pub fn lookup(&self, queries: &Stream<G, (Tr::Key, G::Timestamp)>) -> Stream<G, (Tr::Key, Tr::Val, G::Timestamp, Tr::R)>
+    pub fn lookup(&self, queries: &Stream<G, (Tr::Key, G::Timestamp)>) -> Stream<G, (Tr::Key, Tr::Val, G::Timestamp, Tr::Diff)>
     where
         G::Timestamp: Data+Lattice+Ord+TotalOrder,
         Tr::Key: ExchangeData+Hashable,
         Tr::Val: ExchangeData,
-        Tr::R: ExchangeData+Semigroup,
+        Tr::Diff: ExchangeData+Semigroup,
         Tr: 'static,
     {
         // while the arrangement is already correctly distributed, the query stream may not be.
@@ -280,8 +280,8 @@ where
             let mut active = Vec::new();
             let mut retain = Vec::new();
 
-            let mut working: Vec<(G::Timestamp, Tr::Val, Tr::R)> = Vec::new();
-            let mut working2: Vec<(Tr::Val, Tr::R)> = Vec::new();
+            let mut working: Vec<(G::Timestamp, Tr::Val, Tr::Diff)> = Vec::new();
+            let mut working2: Vec<(Tr::Val, Tr::Diff)> = Vec::new();
 
             move |input1, input2, output| {
 
diff --git a/src/operators/arrange/upsert.rs b/src/operators/arrange/upsert.rs
index a374dbc10..b5b4618bc 100644
--- a/src/operators/arrange/upsert.rs
+++ b/src/operators/arrange/upsert.rs
@@ -144,9 +144,9 @@ where
     G::Timestamp: Lattice+Ord+TotalOrder+ExchangeData,
     Tr::Key: ExchangeData+Hashable+std::hash::Hash,
     Tr::Val: ExchangeData,
-    Tr: Trace+TraceReader<Time=G::Timestamp,R=isize>+'static,
+    Tr: Trace+TraceReader<Time=G::Timestamp,Diff=isize>+'static,
     Tr::Batch: Batch,
-    Tr::Builder: Builder<Item = ((Tr::Key, Tr::Val), Tr::Time, Tr::R)>,
+    Tr::Builder: Builder<Item = ((Tr::Key, Tr::Val), Tr::Time, Tr::Diff)>,
 {
     let mut reader: Option<TraceAgent<Tr>> = None;
 
diff --git a/src/operators/consolidate.rs b/src/operators/consolidate.rs
index 73e29a80f..7345a3ef9 100644
--- a/src/operators/consolidate.rs
+++ b/src/operators/consolidate.rs
@@ -56,7 +56,7 @@ where
     /// As `consolidate` but with the ability to name the operator and specify the trace type.
     pub fn consolidate_named<Tr>(&self, name: &str) -> Self
     where
-        Tr: crate::trace::Trace+crate::trace::TraceReader<Key=D,Val=(),Time=G::Timestamp,R=R>+'static,
+        Tr: crate::trace::Trace+crate::trace::TraceReader<Key=D,Val=(),Time=G::Timestamp,Diff=R>+'static,
         Tr::Batch: crate::trace::Batch,
         Tr::Batcher: Batcher<Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
         Tr::Builder: Builder<Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
diff --git a/src/operators/count.rs b/src/operators/count.rs
index 0fbb9bb15..5d8a6dbeb 100644
--- a/src/operators/count.rs
+++ b/src/operators/count.rs
@@ -55,14 +55,14 @@ where G::Timestamp: TotalOrder+Lattice+Ord {
     }
 }
 
-impl<G: Scope, T1> CountTotal<G, T1::Key, T1::R> for Arranged<G, T1>
+impl<G: Scope, T1> CountTotal<G, T1::Key, T1::Diff> for Arranged<G, T1>
 where
     G::Timestamp: TotalOrder+Lattice+Ord,
     T1: TraceReader<Val=(), Time=G::Timestamp>+Clone+'static,
     T1::Key: ExchangeData,
-    T1::R: ExchangeData+Semigroup,
+    T1::Diff: ExchangeData+Semigroup,
 {
-    fn count_total_core<R2: Semigroup + From<i8>>(&self) -> Collection<G, (T1::Key, T1::R), R2> {
+    fn count_total_core<R2: Semigroup + From<i8>>(&self) -> Collection<G, (T1::Key, T1::Diff), R2> {
 
         let mut trace = self.trace.clone();
         let mut buffer = Vec::new();
@@ -82,10 +82,8 @@ where
                         let (mut trace_cursor, trace_storage) = trace.cursor_through(batch.lower().borrow()).unwrap();
                         upper_limit.clone_from(batch.upper());
 
-                        while batch_cursor.key_valid(&batch) {
-
-                            let key = batch_cursor.key(&batch);
-                            let mut count: Option<T1::R> = None;
+                        while let Some(key) = batch_cursor.get_key(&batch) {
+                            let mut count: Option<T1::Diff> = None;
 
                             trace_cursor.seek_key(&trace_storage, key);
                             if trace_cursor.get_key(&trace_storage) == Some(key) {
diff --git a/src/operators/join.rs b/src/operators/join.rs
index 48e508ab6..ae397a5fa 100644
--- a/src/operators/join.rs
+++ b/src/operators/join.rs
@@ -181,29 +181,29 @@ where
     }
 }
 
-impl<G, Tr> Join<G, Tr::Key, Tr::Val, Tr::R> for Arranged<G, Tr>
+impl<G, Tr> Join<G, Tr::Key, Tr::Val, Tr::Diff> for Arranged<G, Tr>
 where
     G: Scope,
     G::Timestamp: Lattice+Ord,
     Tr: TraceReader<Time=G::Timestamp>+Clone+'static,
     Tr::Key: Data+Hashable,
     Tr::Val: Data,
-    Tr::R: Semigroup,
+    Tr::Diff: Semigroup,
 {
-    fn join_map<V2: ExchangeData, R2: ExchangeData+Semigroup, D: Data, L>(&self, other: &Collection<G, (Tr::Key, V2), R2>, mut logic: L) -> Collection<G, D, <Tr::R as Multiply<R2>>::Output>
-    where Tr::Key: ExchangeData, Tr::R: Multiply<R2>, <Tr::R as Multiply<R2>>::Output: Semigroup, L: FnMut(&Tr::Key, &Tr::Val, &V2)->D+'static {
+    fn join_map<V2: ExchangeData, R2: ExchangeData+Semigroup, D: Data, L>(&self, other: &Collection<G, (Tr::Key, V2), R2>, mut logic: L) -> Collection<G, D, <Tr::Diff as Multiply<R2>>::Output>
+    where Tr::Key: ExchangeData, Tr::Diff: Multiply<R2>, <Tr::Diff as Multiply<R2>>::Output: Semigroup, L: FnMut(&Tr::Key, &Tr::Val, &V2)->D+'static {
         let arranged2 = other.arrange_by_key();
         self.join_core(&arranged2, move |k,v1,v2| Some(logic(k,v1,v2)))
     }
 
-    fn semijoin<R2: ExchangeData+Semigroup>(&self, other: &Collection<G, Tr::Key, R2>) -> Collection<G, (Tr::Key, Tr::Val), <Tr::R as Multiply<R2>>::Output>
-    where Tr::Key: ExchangeData, Tr::R: Multiply<R2>, <Tr::R as Multiply<R2>>::Output: Semigroup {
+    fn semijoin<R2: ExchangeData+Semigroup>(&self, other: &Collection<G, Tr::Key, R2>) -> Collection<G, (Tr::Key, Tr::Val), <Tr::Diff as Multiply<R2>>::Output>
+    where Tr::Key: ExchangeData, Tr::Diff: Multiply<R2>, <Tr::Diff as Multiply<R2>>::Output: Semigroup {
         let arranged2 = other.arrange_by_self();
         self.join_core(&arranged2, |k,v,_| Some((k.clone(), v.clone())))
     }
 
-    fn antijoin<R2: ExchangeData+Semigroup>(&self, other: &Collection<G, Tr::Key, R2>) -> Collection<G, (Tr::Key, Tr::Val), Tr::R>
-    where Tr::Key: ExchangeData, Tr::R: Multiply<R2, Output=Tr::R>, Tr::R: Abelian {
+    fn antijoin<R2: ExchangeData+Semigroup>(&self, other: &Collection<G, Tr::Key, R2>) -> Collection<G, (Tr::Key, Tr::Val), Tr::Diff>
+    where Tr::Key: ExchangeData, Tr::Diff: Multiply<R2, Output=Tr::Diff>, Tr::Diff: Abelian {
         self.as_collection(|k,v| (k.clone(), v.clone()))
             .concat(&self.semijoin(other).negate())
     }
@@ -251,13 +251,13 @@ pub trait JoinCore<G: Scope, K: 'static + ?Sized, V: 'static + ?Sized, R: Semigr
     ///     });
     /// }
     /// ```
-    fn join_core<Tr2,I,L> (&self, stream2: &Arranged<G,Tr2>, result: L) -> Collection<G,I::Item,<R as Multiply<Tr2::R>>::Output>
+    fn join_core<Tr2,I,L> (&self, stream2: &Arranged<G,Tr2>, result: L) -> Collection<G,I::Item,<R as Multiply<Tr2::Diff>>::Output>
     where
         Tr2: TraceReader<Key=K, Time=G::Timestamp>+Clone+'static,
         Tr2::Val: Ord+'static,
-        Tr2::R: Semigroup,
-        R: Multiply<Tr2::R>,
-        <R as Multiply<Tr2::R>>::Output: Semigroup,
+        Tr2::Diff: Semigroup,
+        R: Multiply<Tr2::Diff>,
+        <R as Multiply<Tr2::Diff>>::Output: Semigroup,
         I: IntoIterator,
         I::Item: Data,
         L: FnMut(&K,&V,&Tr2::Val)->I+'static,
@@ -305,11 +305,11 @@ pub trait JoinCore<G: Scope, K: 'static + ?Sized, V: 'static + ?Sized, R: Semigr
     where
         Tr2: TraceReader<Key=K, Time=G::Timestamp>+Clone+'static,
         Tr2::Val: Ord+'static,
-        Tr2::R: Semigroup,
+        Tr2::Diff: Semigroup,
         D: Data,
         ROut: Semigroup,
         I: IntoIterator<Item=(D, G::Timestamp, ROut)>,
-        L: FnMut(&K,&V,&Tr2::Val,&G::Timestamp,&R,&Tr2::R)->I+'static,
+        L: FnMut(&K,&V,&Tr2::Val,&G::Timestamp,&R,&Tr2::Diff)->I+'static,
         ;
 }
 
@@ -322,13 +322,13 @@ where
     R: ExchangeData+Semigroup,
     G::Timestamp: Lattice+Ord,
 {
-    fn join_core<Tr2,I,L> (&self, stream2: &Arranged<G,Tr2>, result: L) -> Collection<G,I::Item,<R as Multiply<Tr2::R>>::Output>
+    fn join_core<Tr2,I,L> (&self, stream2: &Arranged<G,Tr2>, result: L) -> Collection<G,I::Item,<R as Multiply<Tr2::Diff>>::Output>
     where
         Tr2: TraceReader<Key=K, Time=G::Timestamp>+Clone+'static,
         Tr2::Val: Ord+'static,
-        Tr2::R: Semigroup,
-        R: Multiply<Tr2::R>,
-        <R as Multiply<Tr2::R>>::Output: Semigroup,
+        Tr2::Diff: Semigroup,
+        R: Multiply<Tr2::Diff>,
+        <R as Multiply<Tr2::Diff>>::Output: Semigroup,
         I: IntoIterator,
         I::Item: Data,
         L: FnMut(&K,&V,&Tr2::Val)->I+'static,
@@ -341,39 +341,39 @@ where
     where
         Tr2: TraceReader<Key=K, Time=G::Timestamp>+Clone+'static,
         Tr2::Val: Ord+'static,
-        Tr2::R: Semigroup,
+        Tr2::Diff: Semigroup,
         R: Semigroup,
         D: Data,
         ROut: Semigroup,
         I: IntoIterator<Item=(D, G::Timestamp, ROut)>,
-        L: FnMut(&K,&V,&Tr2::Val,&G::Timestamp,&R,&Tr2::R)->I+'static,
+        L: FnMut(&K,&V,&Tr2::Val,&G::Timestamp,&R,&Tr2::Diff)->I+'static,
     {
         self.arrange_by_key().join_core_internal_unsafe(stream2, result)
     }
 
 }
 
-impl<G, T1> JoinCore<G, T1::Key, T1::Val, T1::R> for Arranged<G,T1>
+impl<G, T1> JoinCore<G, T1::Key, T1::Val, T1::Diff> for Arranged<G,T1>
     where
         G: Scope,
         G::Timestamp: Lattice+Ord,
         T1: TraceReader<Time=G::Timestamp>+Clone+'static,
         T1::Key: Ord+'static,
         T1::Val: Ord+'static,
-        T1::R: Semigroup,
+        T1::Diff: Semigroup,
 {
-    fn join_core<Tr2,I,L>(&self, other: &Arranged<G,Tr2>, mut result: L) -> Collection<G,I::Item,<T1::R as Multiply<Tr2::R>>::Output>
+    fn join_core<Tr2,I,L>(&self, other: &Arranged<G,Tr2>, mut result: L) -> Collection<G,I::Item,<T1::Diff as Multiply<Tr2::Diff>>::Output>
     where
         Tr2::Val: Ord+'static,
         Tr2: TraceReader<Key=T1::Key,Time=G::Timestamp>+Clone+'static,
-        Tr2::R: Semigroup,
-        T1::R: Multiply<Tr2::R>,
-        <T1::R as Multiply<Tr2::R>>::Output: Semigroup,
+        Tr2::Diff: Semigroup,
+        T1::Diff: Multiply<Tr2::Diff>,
+        <T1::Diff as Multiply<Tr2::Diff>>::Output: Semigroup,
         I: IntoIterator,
         I::Item: Data,
         L: FnMut(&T1::Key,&T1::Val,&Tr2::Val)->I+'static
     {
-        let result = move |k: &T1::Key, v1: &T1::Val, v2: &Tr2::Val, t: &G::Timestamp, r1: &T1::R, r2: &Tr2::R| {
+        let result = move |k: &T1::Key, v1: &T1::Val, v2: &Tr2::Val, t: &G::Timestamp, r1: &T1::Diff, r2: &Tr2::Diff| {
             let t = t.clone();
             let r = (r1.clone()).multiply(r2);
             result(k, v1, v2).into_iter().map(move |d| (d, t.clone(), r.clone()))
@@ -385,11 +385,11 @@ impl<G, T1> JoinCore<G, T1::Key, T1::Val, T1::R> for Arranged<G,T1>
     where
         Tr2: TraceReader<Key=T1::Key, Time=G::Timestamp>+Clone+'static,
         Tr2::Val: Ord+'static,
-        Tr2::R: Semigroup,
+        Tr2::Diff: Semigroup,
         D: Data,
         ROut: Semigroup,
         I: IntoIterator<Item=(D, G::Timestamp, ROut)>,
-        L: FnMut(&T1::Key,&T1::Val,&Tr2::Val,&G::Timestamp,&T1::R,&Tr2::R)->I+'static,
+        L: FnMut(&T1::Key,&T1::Val,&Tr2::Val,&G::Timestamp,&T1::Diff,&Tr2::Diff)->I+'static,
     {
         join_traces(self, other, result)
     }
@@ -410,14 +410,14 @@ where
     T1: TraceReader<Time=G::Timestamp>+Clone+'static,
     T1::Key: Ord,
     T1::Val: Ord,
-    T1::R: Semigroup,
+    T1::Diff: Semigroup,
     T2: TraceReader<Key=T1::Key, Time=G::Timestamp>+Clone+'static,
     T2::Val: Ord,
-    T2::R: Semigroup,
+    T2::Diff: Semigroup,
     D: Data,
     R: Semigroup,
     I: IntoIterator<Item=(D, G::Timestamp, R)>,
-    L: FnMut(&T1::Key,&T1::Val,&T2::Val,&G::Timestamp,&T1::R,&T2::R)->I+'static,
+    L: FnMut(&T1::Key,&T1::Val,&T2::Val,&G::Timestamp,&T1::Diff,&T2::Diff)->I+'static,
 {
     // Rename traces for symmetry from here on out.
     let mut trace1 = arranged1.trace.clone();
@@ -660,41 +660,41 @@ where
 /// The structure wraps cursors which allow us to play out join computation at whatever rate we like.
 /// This allows us to avoid producing and buffering massive amounts of data, without giving the timely
 /// dataflow system a chance to run operators that can consume and aggregate the data.
-struct Deferred<T, R, S1, S2, C1, C2, D>
+struct Deferred<T, R, C1, C2, D>
 where
     T: Timestamp+Lattice+Ord,
     R: Semigroup,
-    C1: Cursor<S1, Time=T>,
-    C2: Cursor<S2, Key=C1::Key, Time=T>,
+    C1: Cursor<Time=T>,
+    C2: Cursor<Key=C1::Key, Time=T>,
     C1::Val: Ord,
     C2::Val: Ord,
-    C1::R: Semigroup,
-    C2::R: Semigroup,
+    C1::Diff: Semigroup,
+    C2::Diff: Semigroup,
     D: Ord+Clone+Data,
 {
     trace: C1,
-    trace_storage: S1,
+    trace_storage: C1::Storage,
     batch: C2,
-    batch_storage: S2,
+    batch_storage: C2::Storage,
     capability: Capability<T>,
     done: bool,
     temp: Vec<((D, T), R)>,
 }
 
-impl<T, R, S1, S2, C1, C2, D> Deferred<T, R, S1, S2, C1, C2, D>
+impl<T, R, C1, C2, D> Deferred<T, R, C1, C2, D>
 where
     C1::Key: Ord+Eq,
-    C1: Cursor<S1, Time=T>,
-    C2: Cursor<S2, Key=C1::Key, Time=T>,
+    C1: Cursor<Time=T>,
+    C2: Cursor<Key=C1::Key, Time=T>,
     C1::Val: Ord,
     C2::Val: Ord,
-    C1::R: Semigroup,
-    C2::R: Semigroup,
+    C1::Diff: Semigroup,
+    C2::Diff: Semigroup,
     T: Timestamp+Lattice+Ord,
     R: Semigroup,
     D: Clone+Data,
 {
-    fn new(trace: C1, trace_storage: S1, batch: C2, batch_storage: S2, capability: Capability<T>) -> Self {
+    fn new(trace: C1, trace_storage: C1::Storage, batch: C2, batch_storage: C2::Storage, capability: Capability<T>) -> Self {
         Deferred {
             trace,
             trace_storage,
@@ -713,7 +713,7 @@ where
     /// Process keys until at least `fuel` output tuples produced, or the work is exhausted.
     #[inline(never)]
     fn work<L, I>(&mut self, output: &mut OutputHandle<T, (D, T, R), Tee<T, (D, T, R)>>, mut logic: L, fuel: &mut usize)
-    where I: IntoIterator<Item=(D, T, R)>, L: FnMut(&C1::Key, &C1::Val, &C2::Val, &T, &C1::R, &C2::R)->I {
+    where I: IntoIterator<Item=(D, T, R)>, L: FnMut(&C1::Key, &C1::Val, &C2::Val, &T, &C1::Diff, &C2::Diff)->I {
 
         let meet = self.capability.time();
 
@@ -777,12 +777,30 @@ where
     }
 }
 
-struct JoinThinker<'a, V1: Ord+'a + ?Sized, V2: Ord+'a + ?Sized, T: Lattice+Ord+Clone, R1: Semigroup, R2: Semigroup> {
-    pub history1: ValueHistory<'a, V1, T, R1>,
-    pub history2: ValueHistory<'a, V2, T, R2>,
+struct JoinThinker<'a, C1, C2>
+where
+    C1: Cursor,
+    C2: Cursor<Time = C1::Time>,
+    C1::Val: Ord,
+    C2::Val: Ord,
+    C1::Time: Lattice+Ord+Clone,
+    C1::Diff: Semigroup,
+    C2::Diff: Semigroup,
+{
+    pub history1: ValueHistory<'a, C1>,
+    pub history2: ValueHistory<'a, C2>,
 }
 
-impl<'a, V1: Ord + ?Sized, V2: Ord + ?Sized, T: Lattice+Ord+Clone, R1: Semigroup, R2: Semigroup> JoinThinker<'a, V1, V2, T, R1, R2> {
+impl<'a, C1, C2> JoinThinker<'a, C1, C2>
+where
+    C1: Cursor,
+    C2: Cursor<Time = C1::Time>,
+    C1::Val: Ord,
+    C2::Val: Ord,
+    C1::Time: Lattice+Ord+Clone,
+    C1::Diff: Semigroup,
+    C2::Diff: Semigroup,
+{
     fn new() -> Self {
         JoinThinker {
             history1: ValueHistory::new(),
@@ -790,13 +808,13 @@ impl<'a, V1: Ord + ?Sized, V2: Ord + ?Sized, T: Lattice+Ord+Clone, R1: Semigroup
         }
     }
 
-    fn think<F: FnMut(&V1,&V2,T,&R1,&R2)>(&mut self, mut results: F) {
+    fn think<F: FnMut(&C1::Val,&C2::Val,C1::Time,&C1::Diff,&C2::Diff)>(&mut self, mut results: F) {
 
         // for reasonably sized edits, do the dead-simple thing.
         if self.history1.edits.len() < 10 || self.history2.edits.len() < 10 {
             self.history1.edits.map(|v1, t1, d1| {
                 self.history2.edits.map(|v2, t2, d2| {
-                    results(v1, v2, t1.join(t2), &d1, &d2);
+                    results(v1, v2, t1.join(t2), d1, d2);
                 })
             })
         }
@@ -816,7 +834,7 @@ impl<'a, V1: Ord + ?Sized, V2: Ord + ?Sized, T: Lattice+Ord+Clone, R1: Semigroup
                 if replay1.time().unwrap().cmp(&replay2.time().unwrap()) == ::std::cmp::Ordering::Less {
                     replay2.advance_buffer_by(replay1.meet().unwrap());
                     for &((ref val2, ref time2), ref diff2) in replay2.buffer().iter() {
-                        let (val1, time1, ref diff1) = replay1.edit().unwrap();
+                        let (val1, time1, diff1) = replay1.edit().unwrap();
                         results(val1, val2, time1.join(time2), diff1, diff2);
                     }
                     replay1.step();
@@ -824,7 +842,7 @@ impl<'a, V1: Ord + ?Sized, V2: Ord + ?Sized, T: Lattice+Ord+Clone, R1: Semigroup
                 else {
                     replay1.advance_buffer_by(replay2.meet().unwrap());
                     for &((ref val1, ref time1), ref diff1) in replay1.buffer().iter() {
-                        let (val2, time2, ref diff2) = replay2.edit().unwrap();
+                        let (val2, time2, diff2) = replay2.edit().unwrap();
                         results(val1, val2, time1.join(time2), diff1, diff2);
                     }
                     replay2.step();
@@ -834,7 +852,7 @@ impl<'a, V1: Ord + ?Sized, V2: Ord + ?Sized, T: Lattice+Ord+Clone, R1: Semigroup
             while !replay1.is_done() {
                 replay2.advance_buffer_by(replay1.meet().unwrap());
                 for &((ref val2, ref time2), ref diff2) in replay2.buffer().iter() {
-                    let (val1, time1, ref diff1) = replay1.edit().unwrap();
+                    let (val1, time1, diff1) = replay1.edit().unwrap();
                     results(val1, val2, time1.join(time2), diff1, diff2);
                 }
                 replay1.step();
@@ -842,7 +860,7 @@ impl<'a, V1: Ord + ?Sized, V2: Ord + ?Sized, T: Lattice+Ord+Clone, R1: Semigroup
             while !replay2.is_done() {
                 replay1.advance_buffer_by(replay2.meet().unwrap());
                 for &((ref val1, ref time1), ref diff1) in replay1.buffer().iter() {
-                    let (val2, time2, ref diff2) = replay2.edit().unwrap();
+                    let (val2, time2, diff2) = replay2.edit().unwrap();
                     results(val1, val2, time1.join(time2), diff1, diff2);
                 }
                 replay2.step();
diff --git a/src/operators/mod.rs b/src/operators/mod.rs
index 1dd8cb516..8ebca59f8 100644
--- a/src/operators/mod.rs
+++ b/src/operators/mod.rs
@@ -23,12 +23,12 @@ use lattice::Lattice;
 use trace::Cursor;
 
 /// An accumulation of (value, time, diff) updates.
-struct EditList<'a, V: 'a + ?Sized, T, R> {
-    values: Vec<(&'a V, usize)>,
-    edits: Vec<(T, R)>,
+struct EditList<'a, C: Cursor> where C::Time: Sized, C::Diff: Sized {
+    values: Vec<(&'a C::Val, usize)>,
+    edits: Vec<(C::Time, C::Diff)>,
 }
 
-impl<'a, V:'a + ?Sized, T, R> EditList<'a, V, T, R> where T: Ord+Clone, R: Semigroup {
+impl<'a, C: Cursor> EditList<'a, C> where C::Time: Ord+Clone, C::Diff: Semigroup {
     /// Creates an empty list of edits.
     #[inline]
     fn new() -> Self {
@@ -38,8 +38,10 @@ impl<'a, V:'a + ?Sized, T, R> EditList<'a, V, T, R> where T: Ord+Clone, R: Semig
         }
     }
     /// Loads the contents of a cursor.
-    fn load<S, C, L>(&mut self, cursor: &mut C, storage: &'a S, logic: L)
-    where C: Cursor<S, Val=V, Time=T, R=R>, C::Key: Eq, L: Fn(&T)->T {
+    fn load<L>(&mut self, cursor: &mut C, storage: &'a C::Storage, logic: L)
+    where
+        L: Fn(&C::Time)->C::Time,
+    {
         self.clear();
         while cursor.val_valid(storage) {
             cursor.map_times(storage, |time1, diff1| self.push(logic(time1), diff1.clone()));
@@ -56,39 +58,43 @@ impl<'a, V:'a + ?Sized, T, R> EditList<'a, V, T, R> where T: Ord+Clone, R: Semig
     fn len(&self) -> usize { self.edits.len() }
     /// Inserts a new edit for an as-yet undetermined value.
     #[inline]
-    fn push(&mut self, time: T, diff: R) {
+    fn push(&mut self, time: C::Time, diff: C::Diff) {
         // TODO: Could attempt "insertion-sort" like behavior here, where we collapse if possible.
         self.edits.push((time, diff));
     }
     /// Associates all edits pushed since the previous `seal_value` call with `value`.
     #[inline]
-    fn seal(&mut self, value: &'a V) {
+    fn seal(&mut self, value: &'a C::Val) {
         let prev = self.values.last().map(|x| x.1).unwrap_or(0);
         crate::consolidation::consolidate_from(&mut self.edits, prev);
         if self.edits.len() > prev {
             self.values.push((value, self.edits.len()));
         }
     }
-    fn map<F: FnMut(&V, &T, R)>(&self, mut logic: F) {
+    fn map<F: FnMut(&C::Val, &C::Time, &C::Diff)>(&self, mut logic: F) {
         for index in 0 .. self.values.len() {
             let lower = if index == 0 { 0 } else { self.values[index-1].1 };
             let upper = self.values[index].1;
             for edit in lower .. upper {
-                logic(&self.values[index].0, &self.edits[edit].0, self.edits[edit].1.clone());
+                logic(&self.values[index].0, &self.edits[edit].0, &self.edits[edit].1);
             }
         }
     }
 }
 
-struct ValueHistory<'storage, V: 'storage + ?Sized, T, R> {
+struct ValueHistory<'storage, C: Cursor> where C::Time: Sized, C::Diff: Sized {
 
-    edits: EditList<'storage, V, T, R>,
-    history: Vec<(T, T, usize, usize)>,        // (time, meet, value_index, edit_offset)
-    // frontier: FrontierHistory<T>,           // tracks frontiers of remaining times.
-    buffer: Vec<((&'storage V, T), R)>,               // where we accumulate / collapse updates.
+    edits: EditList<'storage, C>,
+    history: Vec<(C::Time, C::Time, usize, usize)>,     // (time, meet, value_index, edit_offset)
+    buffer: Vec<((&'storage C::Val, C::Time), C::Diff)>,   // where we accumulate / collapse updates.
 }
 
-impl<'storage, V: Ord+'storage + ?Sized, T: Lattice+Ord+Clone, R: Semigroup> ValueHistory<'storage, V, T, R> {
+impl<'storage, C: Cursor> ValueHistory<'storage, C>
+where
+    C::Val: Ord+'storage,
+    C::Time: Lattice+Ord+Clone,
+    C::Diff: Semigroup,
+{
     fn new() -> Self {
         ValueHistory {
             edits: EditList::new(),
@@ -101,22 +107,26 @@ impl<'storage, V: Ord+'storage + ?Sized, T: Lattice+Ord+Clone, R: Semigroup> Val
         self.history.clear();
         self.buffer.clear();
     }
-    fn load<S, C, L>(&mut self, cursor: &mut C, storage: &'storage S, logic: L)
-    where C: Cursor<S, Val=V, Time=T, R=R>, C::Key: Eq, L: Fn(&T)->T {
+    fn load<L>(&mut self, cursor: &mut C, storage: &'storage C::Storage, logic: L)
+    where
+        L: Fn(&C::Time)->C::Time,
+    {
         self.edits.load(cursor, storage, logic);
     }
 
     /// Loads and replays a specified key.
     ///
     /// If the key is absent, the replayed history will be empty.
-    fn replay_key<'history, S, C, L>(
+    fn replay_key<'history, L>(
         &'history mut self,
         cursor: &mut C,
-        storage: &'storage S,
+        storage: &'storage C::Storage,
         key: &C::Key,
         logic: L
-    ) -> HistoryReplay<'storage, 'history, V, T, R>
-    where C: Cursor<S, Val=V, Time=T, R=R>, C::Key: Eq, L: Fn(&T)->T
+    ) -> HistoryReplay<'storage, 'history, C>
+    where
+        C::Key: Eq,
+        L: Fn(&C::Time)->C::Time,
     {
         self.clear();
         cursor.seek_key(storage, key);
@@ -127,7 +137,7 @@ impl<'storage, V: Ord+'storage + ?Sized, T: Lattice+Ord+Clone, R: Semigroup> Val
     }
 
     /// Organizes history based on current contents of edits.
-    fn replay<'history>(&'history mut self) -> HistoryReplay<'storage, 'history, V, T, R> {
+    fn replay<'history>(&'history mut self) -> HistoryReplay<'storage, 'history, C> {
 
         self.buffer.clear();
         self.history.clear();
@@ -136,7 +146,7 @@ impl<'storage, V: Ord+'storage + ?Sized, T: Lattice+Ord+Clone, R: Semigroup> Val
             let upper = self.edits.values[value_index].1;
             for edit_index in lower .. upper {
                 let time = self.edits.edits[edit_index].0.clone();
-                self.history.push((time.clone(), time.clone(), value_index, edit_index));
+                self.history.push((time.clone(), time, value_index, edit_index));
             }
         }
 
@@ -151,30 +161,32 @@ impl<'storage, V: Ord+'storage + ?Sized, T: Lattice+Ord+Clone, R: Semigroup> Val
     }
 }
 
-struct HistoryReplay<'storage, 'history, V, T, R>
+struct HistoryReplay<'storage, 'history, C>
 where
     'storage: 'history,
-    V: Ord+'storage + ?Sized,
-    T: Lattice+Ord+Clone+'history,
-    R: Semigroup+'history,
+    C: Cursor,
+    C::Val: Ord+'storage,
+    C::Time: Lattice+Ord+Clone+'history,
+    C::Diff: Semigroup+'history,
 {
-    replay: &'history mut ValueHistory<'storage, V, T, R>
+    replay: &'history mut ValueHistory<'storage, C>
 }
 
-impl<'storage, 'history, V, T, R> HistoryReplay<'storage, 'history, V, T, R>
+impl<'storage, 'history, C> HistoryReplay<'storage, 'history, C>
 where
     'storage: 'history,
-    V: Ord+'storage + ?Sized,
-    T: Lattice+Ord+Clone+'history,
-    R: Semigroup+'history,
+    C: Cursor,
+    C::Val: Ord+'storage,
+    C::Time: Lattice+Ord+Clone+'history,
+    C::Diff: Semigroup+'history,
 {
-    fn time(&self) -> Option<&T> { self.replay.history.last().map(|x| &x.0) }
-    fn meet(&self) -> Option<&T> { self.replay.history.last().map(|x| &x.1) }
-    fn edit(&self) -> Option<(&V, &T, R)> {
-        self.replay.history.last().map(|&(ref t, _, v, e)| (self.replay.edits.values[v].0, t, self.replay.edits.edits[e].1.clone()))
+    fn time(&self) -> Option<&C::Time> { self.replay.history.last().map(|x| &x.0) }
+    fn meet(&self) -> Option<&C::Time> { self.replay.history.last().map(|x| &x.1) }
+    fn edit(&self) -> Option<(&C::Val, &C::Time, &C::Diff)> {
+        self.replay.history.last().map(|&(ref t, _, v, e)| (self.replay.edits.values[v].0, t, &self.replay.edits.edits[e].1))
     }
 
-    fn buffer(&self) -> &[((&'storage V, T), R)] {
+    fn buffer(&self) -> &[((&'storage C::Val, C::Time), C::Diff)] {
         &self.replay.buffer[..]
     }
 
@@ -182,7 +194,7 @@ where
         let (time, _, value_index, edit_offset) = self.replay.history.pop().unwrap();
         self.replay.buffer.push(((self.replay.edits.values[value_index].0, time), self.replay.edits.edits[edit_offset].1.clone()));
     }
-    fn step_while_time_is(&mut self, time: &T) -> bool {
+    fn step_while_time_is(&mut self, time: &C::Time) -> bool {
         let mut found = false;
         while self.time() == Some(time) {
             found = true;
@@ -190,25 +202,11 @@ where
         }
         found
     }
-    fn advance_buffer_by(&mut self, meet: &T) {
+    fn advance_buffer_by(&mut self, meet: &C::Time) {
         for element in self.replay.buffer.iter_mut() {
             (element.0).1 = (element.0).1.join(meet);
         }
         crate::consolidation::consolidate(&mut self.replay.buffer);
     }
     fn is_done(&self) -> bool { self.replay.history.len() == 0 }
-
-    fn _print(&self) where V: ::std::fmt::Debug, T: ::std::fmt::Debug, R: ::std::fmt::Debug {
-        for value_index in 0 .. self.replay.edits.values.len() {
-            let lower = if value_index > 0 { self.replay.edits.values[value_index-1].1 } else { 0 };
-            let upper = self.replay.edits.values[value_index].1;
-            for edit_index in lower .. upper {
-                println!("{:?}, {:?}, {:?}",
-                    self.replay.edits.values[value_index].0,
-                    self.replay.edits.edits[edit_index].0,
-                    self.replay.edits.edits[edit_index].1
-                );
-            }
-        }
-    }
 }
diff --git a/src/operators/reduce.rs b/src/operators/reduce.rs
index 4fbc04904..265ac09c3 100644
--- a/src/operators/reduce.rs
+++ b/src/operators/reduce.rs
@@ -88,7 +88,7 @@ impl<G, K, V, R> Reduce<G, K, V, R> for Collection<G, (K, V), R>
 impl<G: Scope, K: Data, V: Data, T1, R: Semigroup> Reduce<G, K, V, R> for Arranged<G, T1>
 where
     G::Timestamp: Lattice+Ord,
-    T1: TraceReader<Key=K, Val=V, Time=G::Timestamp, R=R>+Clone+'static,
+    T1: TraceReader<Key=K, Val=V, Time=G::Timestamp, Diff=R>+Clone+'static,
 {
     fn reduce_named<L, V2: Data, R2: Abelian>(&self, name: &str, logic: L) -> Collection<G, (K, V2), R2>
         where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>)+'static {
@@ -175,7 +175,7 @@ where G::Timestamp: Lattice+Ord {
 impl<G: Scope, K: Data, T1, R1: Semigroup> Threshold<G, K, R1> for Arranged<G, T1>
 where
     G::Timestamp: Lattice+Ord,
-    T1: TraceReader<Key=K, Val=(), Time=G::Timestamp, R=R1>+Clone+'static,
+    T1: TraceReader<Key=K, Val=(), Time=G::Timestamp, Diff=R1>+Clone+'static,
 {
     fn threshold_named<R2: Abelian, F: FnMut(&K,&R1)->R2+'static>(&self, name: &str, mut thresh: F) -> Collection<G, K, R2> {
         self.reduce_abelian::<_,KeySpine<_,_,_>>(name, move |k,s,t| t.push(((), thresh(k, &s[0].1))))
@@ -230,7 +230,7 @@ where
 impl<G: Scope, K: Data, T1, R: Semigroup> Count<G, K, R> for Arranged<G, T1>
 where
     G::Timestamp: Lattice+Ord,
-    T1: TraceReader<Key=K, Val=(), Time=G::Timestamp, R=R>+Clone+'static,
+    T1: TraceReader<Key=K, Val=(), Time=G::Timestamp, Diff=R>+Clone+'static,
 {
     fn count_core<R2: Abelian + From<i8>>(&self) -> Collection<G, (K, R), R2> {
         self.reduce_abelian::<_,ValSpine<_,_,_,_>>("Count", |_k,s,t| t.push((s[0].1.clone(), R2::from(1i8))))
@@ -274,10 +274,10 @@ pub trait ReduceCore<G: Scope, K: ToOwned + ?Sized, V: ToOwned + ?Sized, R: Semi
         where
             T2: Trace+TraceReader<Key=K, Time=G::Timestamp>+'static,
             T2::Val: Data,
-            T2::R: Abelian,
+            T2::Diff: Abelian,
             T2::Batch: Batch,
-            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, T2::Val), G::Timestamp, T2::R)>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val, T2::R)>)+'static,
+            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, T2::Val), G::Timestamp, T2::Diff)>,
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val, T2::Diff)>)+'static,
         {
             self.reduce_core::<_,T2>(name, move |key, input, output, change| {
                 if !input.is_empty() {
@@ -297,10 +297,10 @@ pub trait ReduceCore<G: Scope, K: ToOwned + ?Sized, V: ToOwned + ?Sized, R: Semi
         where
             T2: Trace+TraceReader<Key=K, Time=G::Timestamp>+'static,
             T2::Val: Data,
-            T2::R: Semigroup,
+            T2::Diff: Semigroup,
             T2::Batch: Batch,
-            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, T2::Val), G::Timestamp, T2::R)>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::R)>, &mut Vec<(T2::Val,T2::R)>)+'static
+            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, T2::Val), G::Timestamp, T2::Diff)>,
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::Diff)>, &mut Vec<(T2::Val,T2::Diff)>)+'static
             ;
 }
 
@@ -316,11 +316,11 @@ where
     fn reduce_core<L, T2>(&self, name: &str, logic: L) -> Arranged<G, TraceAgent<T2>>
         where
             T2::Val: Data,
-            T2::R: Semigroup,
+            T2::Diff: Semigroup,
             T2: Trace+TraceReader<Key=K, Time=G::Timestamp>+'static,
             T2::Batch: Batch,
-            T2::Builder: Builder<Output=T2::Batch, Item = ((K, T2::Val), G::Timestamp, T2::R)>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::R)>, &mut Vec<(T2::Val, T2::R)>)+'static
+            T2::Builder: Builder<Output=T2::Batch, Item = ((K, T2::Val), G::Timestamp, T2::Diff)>,
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::Diff)>, &mut Vec<(T2::Val, T2::Diff)>)+'static
     {
         self.arrange_by_key_named(&format!("Arrange: {}", name))
             .reduce_core(name, logic)
@@ -333,17 +333,17 @@ where
     K::Owned: Data,
     V: ToOwned + Ord + ?Sized,
     G::Timestamp: Lattice+Ord,
-    T1: TraceReader<Key=K, Val=V, Time=G::Timestamp, R=R>+Clone+'static,
+    T1: TraceReader<Key=K, Val=V, Time=G::Timestamp, Diff=R>+Clone+'static,
 {
     fn reduce_core<L, T2>(&self, name: &str, logic: L) -> Arranged<G, TraceAgent<T2>>
         where
             T2: Trace+TraceReader<Key=K, Time=G::Timestamp>+'static,
             T2::Val: Ord + ToOwned,
             <T2::Val as ToOwned>::Owned: Data,
-            T2::R: Semigroup,
+            T2::Diff: Semigroup,
             T2::Batch: Batch,
-            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, <T2::Val as ToOwned>::Owned), T2::Time, T2::R)>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(<T2::Val as ToOwned>::Owned,T2::R)>, &mut Vec<(<T2::Val as ToOwned>::Owned, T2::R)>)+'static,
+            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, <T2::Val as ToOwned>::Owned), T2::Time, T2::Diff)>,
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(<T2::Val as ToOwned>::Owned,T2::Diff)>, &mut Vec<(<T2::Val as ToOwned>::Owned, T2::Diff)>)+'static,
         {
             reduce_trace(self, name, logic)
         }
@@ -357,14 +357,14 @@ where
     T1::Key: Ord + ToOwned,
     <T1::Key as ToOwned>::Owned: Ord,
     T1::Val: Ord,
-    T1::R: Semigroup,
+    T1::Diff: Semigroup,
     T2: Trace+TraceReader<Key=T1::Key, Time=G::Timestamp> + 'static,
     T2::Val: Ord + ToOwned,
     <T2::Val as ToOwned>::Owned: Data,
-    T2::R: Semigroup,
+    T2::Diff: Semigroup,
     T2::Batch: Batch,
-    T2::Builder: Builder<Output=T2::Batch, Item = ((<T1::Key as ToOwned>::Owned, <T2::Val as ToOwned>::Owned), T2::Time, T2::R)>,
-    L: FnMut(&T1::Key, &[(&T1::Val, T1::R)], &mut Vec<(<T2::Val as ToOwned>::Owned,T2::R)>, &mut Vec<(<T2::Val as ToOwned>::Owned, T2::R)>)+'static,
+    T2::Builder: Builder<Output=T2::Batch, Item = ((<T1::Key as ToOwned>::Owned, <T2::Val as ToOwned>::Owned), T2::Time, T2::Diff)>,
+    L: FnMut(&T1::Key, &[(&T1::Val, T1::Diff)], &mut Vec<(<T2::Val as ToOwned>::Owned,T2::Diff)>, &mut Vec<(<T2::Val as ToOwned>::Owned, T2::Diff)>)+'static,
 {
     let mut result_trace = None;
 
@@ -498,7 +498,7 @@ where
                         //
                         // TODO: It would be better if all updates went into one batch, but timely dataflow prevents
                         //       this as long as it requires that there is only one capability for each message.
-                        let mut buffers = Vec::<(G::Timestamp, Vec<(<T2::Val as ToOwned>::Owned, G::Timestamp, T2::R)>)>::new();
+                        let mut buffers = Vec::<(G::Timestamp, Vec<(<T2::Val as ToOwned>::Owned, G::Timestamp, T2::Diff)>)>::new();
                         let mut builders = Vec::new();
                         for i in 0 .. capabilities.len() {
                             buffers.push((capabilities[i].time().clone(), Vec::new()));
@@ -513,7 +513,7 @@ where
                         let (mut batch_cursor, batch_storage) = (CursorList::new(batch_cursors, &batch_storage), batch_storage);
                         let batch_storage = &batch_storage;
 
-                        let mut thinker = history_replay::HistoryReplayer::<T1::Val, T2::Val, G::Timestamp, T1::R, T2::R>::new();
+                        let mut thinker = history_replay::HistoryReplayer::new();
 
                         // We now march through the keys we must work on, drawing from `batch_cursors` and `exposed`.
                         //
@@ -674,33 +674,37 @@ fn sort_dedup<T: Ord>(list: &mut Vec<T>) {
     list.dedup();
 }
 
-trait PerKeyCompute<'a, V1, V2, T, R1, R2>
+trait PerKeyCompute<'a, C1, C2, C3>
 where
-    V1: Ord + ?Sized,
-    V2: Ord + ToOwned+'a + ?Sized,
-    V2::Owned: Ord + Clone + 'a,
-    T: Lattice+Ord+Clone,
-    R1: Semigroup,
-    R2: Semigroup,
+    C1: Cursor,
+    C2: Cursor<Key = C1::Key, Time = C1::Time>,
+    C3: Cursor<Key = C1::Key, Val = C1::Val, Time = C1::Time, Diff = C1::Diff>,
+    C1::Val: Ord,
+    C2::Val: Ord + ToOwned,
+    <C2::Val as ToOwned>::Owned: Ord + Clone,
+    C1::Time: Lattice+Ord+Clone,
+    C1::Diff: Semigroup,
+    C2::Diff: Semigroup,
 {
     fn new() -> Self;
-    fn compute<K, S1, S2, S3, C1, C2, C3, L>(
+    fn compute<L>(
         &mut self,
-        key: &K,
-        source_cursor: (&mut C1, &'a S1),
-        output_cursor: (&mut C2, &'a S2),
-        batch_cursor: (&mut C3, &'a S3),
-        times: &mut Vec<T>,
+        key: &C1::Key,
+        source_cursor: (&mut C1, &'a C1::Storage),
+        output_cursor: (&mut C2, &'a C2::Storage),
+        batch_cursor: (&mut C3, &'a C3::Storage),
+        times: &mut Vec<C1::Time>,
         logic: &mut L,
-        upper_limit: &Antichain<T>,
-        outputs: &mut [(T, Vec<(V2::Owned, T, R2)>)],
-        new_interesting: &mut Vec<T>) -> (usize, usize)
+        upper_limit: &Antichain<C1::Time>,
+        outputs: &mut [(C2::Time, Vec<(<C2::Val as ToOwned>::Owned, C2::Time, C2::Diff)>)],
+        new_interesting: &mut Vec<C1::Time>) -> (usize, usize)
     where
-        K: Eq + ?Sized,
-        C1: Cursor<S1, Key = K, Val = V1, Time = T, R = R1>,
-        C2: Cursor<S2, Key = K, Val = V2, Time = T, R = R2>,
-        C3: Cursor<S3, Key = K, Val = V1, Time = T, R = R1>,
-        L: FnMut(&K, &[(&V1, R1)], &mut Vec<(V2::Owned, R2)>, &mut Vec<(V2::Owned, R2)>);
+        C1::Key: Eq,
+        L: FnMut(
+            &C1::Key, &[(&C1::Val, C1::Diff)],
+            &mut Vec<(<C2::Val as ToOwned>::Owned, C2::Diff)>,
+            &mut Vec<(<C2::Val as ToOwned>::Owned, C2::Diff)>,
+        );
 }
 
 
@@ -713,46 +717,54 @@ mod history_replay {
     use operators::ValueHistory;
     use timely::progress::Antichain;
 
+    use timely::PartialOrder;
+
     use super::{PerKeyCompute, sort_dedup};
 
     /// The `HistoryReplayer` is a compute strategy based on moving through existing inputs, interesting times, etc in
     /// time order, maintaining consolidated representations of updates with respect to future interesting times.
-    pub struct HistoryReplayer<'a, V1, V2, T, R1, R2>
+    pub struct HistoryReplayer<'a, C1, C2, C3>//V1, V2, T, R1, R2>
     where
-        V1: Ord+'a + ?Sized,
-        V2: Ord+'a + ToOwned + ?Sized,
-        V2::Owned: Ord + 'a,
-        T: Lattice+Ord+Clone,
-        R1: Semigroup,
-        R2: Semigroup,
+        C1: Cursor,
+        C2: Cursor<Key = C1::Key, Time = C1::Time>,
+        C3: Cursor<Key = C1::Key, Val = C1::Val, Time = C1::Time, Diff = C1::Diff>,
+        C1::Val: Ord,
+        C2::Val: Ord + ToOwned,
+        <C2::Val as ToOwned>::Owned: Ord + Clone,
+        C1::Time: Lattice+Ord+Clone,
+        C1::Diff: Semigroup,
+        C2::Diff: Semigroup,
     {
-        batch_history: ValueHistory<'a, V1, T, R1>,
-        input_history: ValueHistory<'a, V1, T, R1>,
-        output_history: ValueHistory<'a, V2, T, R2>,
-        input_buffer: Vec<(&'a V1, R1)>,
-        output_buffer: Vec<(V2::Owned, R2)>,
-        update_buffer: Vec<(V2::Owned, R2)>,
-        output_produced: Vec<((V2::Owned, T), R2)>,
-        synth_times: Vec<T>,
-        meets: Vec<T>,
-        times_current: Vec<T>,
-        temporary: Vec<T>,
+        input_history: ValueHistory<'a, C1>,
+        output_history: ValueHistory<'a, C2>,
+        batch_history: ValueHistory<'a, C3>,
+        input_buffer: Vec<(&'a C1::Val, C1::Diff)>,
+        output_buffer: Vec<(<C2::Val as ToOwned>::Owned, C2::Diff)>,
+        update_buffer: Vec<(<C2::Val as ToOwned>::Owned, C2::Diff)>,
+        output_produced: Vec<((<C2::Val as ToOwned>::Owned, C2::Time), C2::Diff)>,
+        synth_times: Vec<C1::Time>,
+        meets: Vec<C1::Time>,
+        times_current: Vec<C1::Time>,
+        temporary: Vec<C1::Time>,
     }
 
-    impl<'a, V1, V2, T, R1, R2> PerKeyCompute<'a, V1, V2, T, R1, R2> for HistoryReplayer<'a, V1, V2, T, R1, R2>
+    impl<'a, C1, C2, C3> PerKeyCompute<'a, C1, C2, C3> for HistoryReplayer<'a, C1, C2, C3>
     where
-        V1: Ord + ?Sized,
-        V2: Ord + ToOwned + ?Sized,
-        V2::Owned: Ord + Clone + 'a,
-        T: Lattice+Ord+Clone,
-        R1: Semigroup,
-        R2: Semigroup,
+        C1: Cursor,
+        C2: Cursor<Key = C1::Key, Time = C1::Time>,
+        C3: Cursor<Key = C1::Key, Val = C1::Val, Time = C1::Time, Diff = C1::Diff>,
+        C1::Val: Ord,
+        C2::Val: Ord + ToOwned,
+        <C2::Val as ToOwned>::Owned: Ord + Clone,
+        C1::Time: Lattice+Ord+Clone,
+        C1::Diff: Semigroup,
+        C2::Diff: Semigroup,
     {
         fn new() -> Self {
             HistoryReplayer {
-                batch_history: ValueHistory::new(),
                 input_history: ValueHistory::new(),
                 output_history: ValueHistory::new(),
+                batch_history: ValueHistory::new(),
                 input_buffer: Vec::new(),
                 output_buffer: Vec::new(),
                 update_buffer: Vec::new(),
@@ -764,23 +776,24 @@ mod history_replay {
             }
         }
         #[inline(never)]
-        fn compute<K, S1, S2, S3, C1, C2, C3, L>(
+        fn compute<L>(
             &mut self,
-            key: &K,
-            (source_cursor, source_storage): (&mut C1, &'a S1),
-            (output_cursor, output_storage): (&mut C2, &'a S2),
-            (batch_cursor, batch_storage): (&mut C3, &'a S3),
-            times: &mut Vec<T>,
+            key: &C1::Key,
+            (source_cursor, source_storage): (&mut C1, &'a C1::Storage),
+            (output_cursor, output_storage): (&mut C2, &'a C2::Storage),
+            (batch_cursor, batch_storage): (&mut C3, &'a C3::Storage),
+            times: &mut Vec<C1::Time>,
             logic: &mut L,
-            upper_limit: &Antichain<T>,
-            outputs: &mut [(T, Vec<(V2::Owned, T, R2)>)],
-            new_interesting: &mut Vec<T>) -> (usize, usize)
+            upper_limit: &Antichain<C1::Time>,
+            outputs: &mut [(C2::Time, Vec<(<C2::Val as ToOwned>::Owned, C2::Time, C2::Diff)>)],
+            new_interesting: &mut Vec<C1::Time>) -> (usize, usize)
         where
-            K: Eq + ?Sized,
-            C1: Cursor<S1, Key = K, Val = V1, Time = T, R = R1>,
-            C2: Cursor<S2, Key = K, Val = V2, Time = T, R = R2>,
-            C3: Cursor<S3, Key = K, Val = V1, Time = T, R = R1>,
-            L: FnMut(&K, &[(&V1, R1)], &mut Vec<(V2::Owned, R2)>, &mut Vec<(V2::Owned, R2)>)
+            C1::Key: Eq,
+            L: FnMut(
+                &C1::Key, &[(&C1::Val, C1::Diff)],
+                &mut Vec<(<C2::Val as ToOwned>::Owned, C2::Diff)>,
+                &mut Vec<(<C2::Val as ToOwned>::Owned, C2::Diff)>,
+            )
         {
 
             // The work we need to perform is at times defined principally by the contents of `batch_cursor`
diff --git a/src/operators/threshold.rs b/src/operators/threshold.rs
index cbe1b8d8f..d17078b12 100644
--- a/src/operators/threshold.rs
+++ b/src/operators/threshold.rs
@@ -102,17 +102,17 @@ where G::Timestamp: TotalOrder+Lattice+Ord {
     }
 }
 
-impl<G: Scope, T1> ThresholdTotal<G, T1::Key, T1::R> for Arranged<G, T1>
+impl<G: Scope, T1> ThresholdTotal<G, T1::Key, T1::Diff> for Arranged<G, T1>
 where
     G::Timestamp: TotalOrder+Lattice+Ord,
     T1: TraceReader<Val=(), Time=G::Timestamp>+Clone+'static,
     T1::Key: ExchangeData,
-    T1::R: ExchangeData+Semigroup,
+    T1::Diff: ExchangeData+Semigroup,
 {
     fn threshold_semigroup<R2, F>(&self, mut thresh: F) -> Collection<G, T1::Key, R2>
     where
         R2: Semigroup,
-        F: FnMut(&T1::Key,&T1::R,Option<&T1::R>)->Option<R2>+'static,
+        F: FnMut(&T1::Key,&T1::Diff,Option<&T1::Diff>)->Option<R2>+'static,
     {
 
         let mut trace = self.trace.clone();
@@ -135,9 +135,8 @@ where
 
                         upper_limit.clone_from(batch.upper());
 
-                        while batch_cursor.key_valid(&batch) {
-                            let key = batch_cursor.key(&batch);
-                            let mut count: Option<T1::R> = None;
+                        while let Some(key) = batch_cursor.get_key(&batch) {
+                            let mut count: Option<T1::Diff> = None;
 
                             // Compute the multiplicity of this key before the current batch.
                             trace_cursor.seek_key(&trace_storage, key);
diff --git a/src/trace/cursor/cursor_list.rs b/src/trace/cursor/cursor_list.rs
index 32e0dc5d5..035fb5719 100644
--- a/src/trace/cursor/cursor_list.rs
+++ b/src/trace/cursor/cursor_list.rs
@@ -13,10 +13,14 @@ pub struct CursorList<C> {
     min_val: Vec<usize>,
 }
 
-impl<C> CursorList<C> {
+impl<C> CursorList<C>
+where
+    C: Cursor, 
+    C::Key: Ord, 
+    C::Val: Ord,
+{
     /// Creates a new cursor list from pre-existing cursors.
-    pub fn new<S>(cursors: Vec<C>, storage: &[S]) -> Self where C: Cursor<S>, C::Key: Ord, C::Val: Ord {
-
+    pub fn new(cursors: Vec<C>, storage: &[C::Storage]) -> Self  {
         let mut result = CursorList {
             cursors,
             min_key: Vec::new(),
@@ -36,7 +40,7 @@ impl<C> CursorList<C> {
     //
     // Once finished, it invokes `minimize_vals()` to ensure the value cursor is
     // in a consistent state as well.
-    fn minimize_keys<S>(&mut self, storage: &[S]) where C: Cursor<S>, C::Key: Ord, C::Val: Ord {
+    fn minimize_keys(&mut self, storage: &[C::Storage]) {
 
         self.min_key.clear();
 
@@ -64,7 +68,7 @@ impl<C> CursorList<C> {
     // indices of cursors whose value equals the minimum valid value seen so far. As it
     // goes, if it observes an improved value it clears the current list, updates the minimum
     // value, and continues.
-    fn minimize_vals<S>(&mut self, storage: &[S]) where C: Cursor<S>, C::Key: Ord, C::Val: Ord {
+    fn minimize_vals(&mut self, storage: &[C::Storage]) {
 
         self.min_val.clear();
 
@@ -85,7 +89,7 @@ impl<C> CursorList<C> {
     }
 }
 
-impl<S, C: Cursor<S>> Cursor<Vec<S>> for CursorList<C>
+impl<C: Cursor> Cursor for CursorList<C>
 where
     C::Key: Ord,
     C::Val: Ord,
@@ -93,30 +97,32 @@ where
     type Key = C::Key;
     type Val = C::Val;
     type Time = C::Time;
-    type R = C::R;
+    type Diff = C::Diff;
+
+    type Storage = Vec<C::Storage>;
 
     // validation methods
     #[inline]
-    fn key_valid(&self, _storage: &Vec<S>) -> bool { !self.min_key.is_empty() }
+    fn key_valid(&self, _storage: &Vec<C::Storage>) -> bool { !self.min_key.is_empty() }
     #[inline]
-    fn val_valid(&self, _storage: &Vec<S>) -> bool { !self.min_val.is_empty() }
+    fn val_valid(&self, _storage: &Vec<C::Storage>) -> bool { !self.min_val.is_empty() }
 
     // accessors
     #[inline]
-    fn key<'a>(&self, storage: &'a Vec<S>) -> &'a Self::Key {
+    fn key<'a>(&self, storage: &'a Vec<C::Storage>) -> &'a Self::Key {
         debug_assert!(self.key_valid(storage));
         debug_assert!(self.cursors[self.min_key[0]].key_valid(&storage[self.min_key[0]]));
         self.cursors[self.min_key[0]].key(&storage[self.min_key[0]])
     }
     #[inline]
-    fn val<'a>(&self, storage: &'a Vec<S>) -> &'a Self::Val {
+    fn val<'a>(&self, storage: &'a Vec<C::Storage>) -> &'a Self::Val {
         debug_assert!(self.key_valid(storage));
         debug_assert!(self.val_valid(storage));
         debug_assert!(self.cursors[self.min_val[0]].val_valid(&storage[self.min_val[0]]));
         self.cursors[self.min_val[0]].val(&storage[self.min_val[0]])
     }
     #[inline]
-    fn map_times<L: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &Vec<S>, mut logic: L) {
+    fn map_times<L: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &Vec<C::Storage>, mut logic: L) {
         for &index in self.min_val.iter() {
             self.cursors[index].map_times(&storage[index], |t,d| logic(t,d));
         }
@@ -124,14 +130,14 @@ where
 
     // key methods
     #[inline]
-    fn step_key(&mut self, storage: &Vec<S>) {
+    fn step_key(&mut self, storage: &Vec<C::Storage>) {
         for &index in self.min_key.iter() {
             self.cursors[index].step_key(&storage[index]);
         }
         self.minimize_keys(storage);
     }
     #[inline]
-    fn seek_key(&mut self, storage: &Vec<S>, key: &Self::Key) {
+    fn seek_key(&mut self, storage: &Vec<C::Storage>, key: &Self::Key) {
         for index in 0 .. self.cursors.len() {
             self.cursors[index].seek_key(&storage[index], key);
         }
@@ -140,14 +146,14 @@ where
 
     // value methods
     #[inline]
-    fn step_val(&mut self, storage: &Vec<S>) {
+    fn step_val(&mut self, storage: &Vec<C::Storage>) {
         for &index in self.min_val.iter() {
             self.cursors[index].step_val(&storage[index]);
         }
         self.minimize_vals(storage);
     }
     #[inline]
-    fn seek_val(&mut self, storage: &Vec<S>, val: &Self::Val) {
+    fn seek_val(&mut self, storage: &Vec<C::Storage>, val: &Self::Val) {
         for &index in self.min_key.iter() {
             self.cursors[index].seek_val(&storage[index], val);
         }
@@ -156,14 +162,14 @@ where
 
     // rewinding methods
     #[inline]
-    fn rewind_keys(&mut self, storage: &Vec<S>) {
+    fn rewind_keys(&mut self, storage: &Vec<C::Storage>) {
         for index in 0 .. self.cursors.len() {
             self.cursors[index].rewind_keys(&storage[index]);
         }
         self.minimize_keys(storage);
     }
     #[inline]
-    fn rewind_vals(&mut self, storage: &Vec<S>) {
+    fn rewind_vals(&mut self, storage: &Vec<C::Storage>) {
         for &index in self.min_key.iter() {
             self.cursors[index].rewind_vals(&storage[index]);
         }
diff --git a/src/trace/cursor/cursor_pair.rs b/src/trace/cursor/cursor_pair.rs
index bdbd8b79e..7ca40a4af 100644
--- a/src/trace/cursor/cursor_pair.rs
+++ b/src/trace/cursor/cursor_pair.rs
@@ -15,27 +15,29 @@ pub struct CursorPair<C1, C2> {
     val_order: Ordering,    // Invalid vals are `Greater` than all other vals. `Equal` implies both valid.
 }
 
-impl<K, V, T, R, C1, C2, S1, S2> Cursor<(S1, S2)> for CursorPair<C1, C2>
+impl<K, V, T, R, C1, C2> Cursor for CursorPair<C1, C2>
 where
     K: Ord,
     V: Ord,
-    C1: Cursor<S1, Key = K, Val = V, Time = T, R = R>,
-    C2: Cursor<S2, Key = K, Val = V, Time = T, R = R>,
+    C1: Cursor<Key = K, Val = V, Time = T, Diff = R>,
+    C2: Cursor<Key = K, Val = V, Time = T, Diff = R>,
 {
     type Key = K;
     type Val = V;
     type Time = T;
-    type R = R;
+    type Diff = R;
+
+    type Storage = (C1::Storage, C2::Storage);
 
     // validation methods
-    fn key_valid(&self, storage: &(S1, S2)) -> bool {
+    fn key_valid(&self, storage: &(C1::Storage, C2::Storage)) -> bool {
         match self.key_order {
             Ordering::Less => self.cursor1.key_valid(&storage.0),
             Ordering::Equal => true,
             Ordering::Greater => self.cursor2.key_valid(&storage.1),
         }
     }
-    fn val_valid(&self, storage: &(S1, S2)) -> bool {
+    fn val_valid(&self, storage: &(C1::Storage, C2::Storage)) -> bool {
         match (self.key_order, self.val_order) {
             (Ordering::Less, _) => self.cursor1.val_valid(&storage.0),
             (Ordering::Greater, _) => self.cursor2.val_valid(&storage.1),
@@ -46,13 +48,13 @@ where
     }
 
     // accessors
-    fn key<'a>(&self, storage: &'a (S1, S2)) -> &'a K {
+    fn key<'a>(&self, storage: &'a (C1::Storage, C2::Storage)) -> &'a K {
         match self.key_order {
             Ordering::Less => self.cursor1.key(&storage.0),
             _ => self.cursor2.key(&storage.1),
         }
     }
-    fn val<'a>(&self, storage: &'a (S1, S2)) -> &'a V {
+    fn val<'a>(&self, storage: &'a (C1::Storage, C2::Storage)) -> &'a V {
         if self.key_order == Ordering::Less || (self.key_order == Ordering::Equal && self.val_order != Ordering::Greater) {
             self.cursor1.val(&storage.0)
         }
@@ -60,7 +62,7 @@ where
             self.cursor2.val(&storage.1)
         }
     }
-    fn map_times<L: FnMut(&T, &R)>(&mut self, storage: &(S1, S2), mut logic: L) {
+    fn map_times<L: FnMut(&T, &R)>(&mut self, storage: &(C1::Storage, C2::Storage), mut logic: L) {
         if self.key_order == Ordering::Less || (self.key_order == Ordering::Equal && self.val_order != Ordering::Greater) {
             self.cursor1.map_times(&storage.0, |t,d| logic(t,d));
         }
@@ -70,7 +72,7 @@ where
     }
 
     // key methods
-    fn step_key(&mut self, storage: &(S1, S2)) {
+    fn step_key(&mut self, storage: &(C1::Storage, C2::Storage)) {
 
         if self.key_order != Ordering::Greater { self.cursor1.step_key(&storage.0); }
         if self.key_order != Ordering::Less { self.cursor2.step_key(&storage.1); }
@@ -81,7 +83,7 @@ where
             (true, true) => self.cursor1.key(&storage.0).cmp(self.cursor2.key(&storage.1)),
         };
     }
-    fn seek_key(&mut self, storage: &(S1, S2), key: &K) {
+    fn seek_key(&mut self, storage: &(C1::Storage, C2::Storage), key: &K) {
 
         self.cursor1.seek_key(&storage.0, key);
         self.cursor2.seek_key(&storage.1, key);
@@ -94,7 +96,7 @@ where
     }
 
     // value methods
-    fn step_val(&mut self, storage: &(S1, S2)) {
+    fn step_val(&mut self, storage: &(C1::Storage, C2::Storage)) {
         match self.key_order {
             Ordering::Less => self.cursor1.step_val(&storage.0),
             Ordering::Equal => {
@@ -109,7 +111,7 @@ where
             Ordering::Greater => self.cursor2.step_val(&storage.1),
         }
     }
-    fn seek_val(&mut self, storage: &(S1, S2), val: &V) {
+    fn seek_val(&mut self, storage: &(C1::Storage, C2::Storage), val: &V) {
         match self.key_order {
             Ordering::Less => self.cursor1.seek_val(&storage.0, val),
             Ordering::Equal => {
@@ -126,11 +128,11 @@ where
     }
 
     // rewinding methods
-    fn rewind_keys(&mut self, storage: &(S1, S2)) {
+    fn rewind_keys(&mut self, storage: &(C1::Storage, C2::Storage)) {
         self.cursor1.rewind_keys(&storage.0);
         self.cursor2.rewind_keys(&storage.1);
     }
-    fn rewind_vals(&mut self, storage: &(S1, S2)) {
+    fn rewind_vals(&mut self, storage: &(C1::Storage, C2::Storage)) {
         if self.key_order != Ordering::Greater { self.cursor1.rewind_vals(&storage.0); }
         if self.key_order != Ordering::Less { self.cursor2.rewind_vals(&storage.1); }
     }
diff --git a/src/trace/cursor/mod.rs b/src/trace/cursor/mod.rs
index 2cf5c4293..4dcd2062e 100644
--- a/src/trace/cursor/mod.rs
+++ b/src/trace/cursor/mod.rs
@@ -12,7 +12,7 @@ pub mod cursor_list;
 pub use self::cursor_list::CursorList;
 
 /// A cursor for navigating ordered `(key, val, time, diff)` updates.
-pub trait Cursor<Storage: ?Sized> {
+pub trait Cursor {
 
     /// Key by which updates are indexed.
     type Key: ?Sized;
@@ -21,57 +21,60 @@ pub trait Cursor<Storage: ?Sized> {
     /// Timestamps associated with updates
     type Time: ?Sized;
     /// Associated update.
-    type R: ?Sized;
+    type Diff: ?Sized;
+
+    /// Storage required by the cursor.
+    type Storage;
 
     /// Indicates if the current key is valid.
     ///
     /// A value of `false` indicates that the cursor has exhausted all keys.
-    fn key_valid(&self, storage: &Storage) -> bool;
+    fn key_valid(&self, storage: &Self::Storage) -> bool;
     /// Indicates if the current value is valid.
     ///
     /// A value of `false` indicates that the cursor has exhausted all values for this key.
-    fn val_valid(&self, storage: &Storage) -> bool;
+    fn val_valid(&self, storage: &Self::Storage) -> bool;
 
     /// A reference to the current key. Asserts if invalid.
-    fn key<'a>(&self, storage: &'a Storage) -> &'a Self::Key;
+    fn key<'a>(&self, storage: &'a Self::Storage) -> &'a Self::Key;
     /// A reference to the current value. Asserts if invalid.
-    fn val<'a>(&self, storage: &'a Storage) -> &'a Self::Val;
+    fn val<'a>(&self, storage: &'a Self::Storage) -> &'a Self::Val;
 
     /// Returns a reference to the current key, if valid.
-    fn get_key<'a>(&self, storage: &'a Storage) -> Option<&'a Self::Key> {
+    fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<&'a Self::Key> {
         if self.key_valid(storage) { Some(self.key(storage)) } else { None }
     }
     /// Returns a reference to the current value, if valid.
-    fn get_val<'a>(&self, storage: &'a Storage) -> Option<&'a Self::Val> {
+    fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<&'a Self::Val> {
         if self.val_valid(storage) { Some(self.val(storage)) } else { None }
     }
 
     /// Applies `logic` to each pair of time and difference. Intended for mutation of the
     /// closure's scope.
-    fn map_times<L: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &Storage, logic: L);
+    fn map_times<L: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &Self::Storage, logic: L);
 
     /// Advances the cursor to the next key.
-    fn step_key(&mut self, storage: &Storage);
+    fn step_key(&mut self, storage: &Self::Storage);
     /// Advances the cursor to the specified key.
-    fn seek_key(&mut self, storage: &Storage, key: &Self::Key);
+    fn seek_key(&mut self, storage: &Self::Storage, key: &Self::Key);
 
     /// Advances the cursor to the next value.
-    fn step_val(&mut self, storage: &Storage);
+    fn step_val(&mut self, storage: &Self::Storage);
     /// Advances the cursor to the specified value.
-    fn seek_val(&mut self, storage: &Storage, val: &Self::Val);
+    fn seek_val(&mut self, storage: &Self::Storage, val: &Self::Val);
 
     /// Rewinds the cursor to the first key.
-    fn rewind_keys(&mut self, storage: &Storage);
+    fn rewind_keys(&mut self, storage: &Self::Storage);
     /// Rewinds the cursor to the first value for current key.
-    fn rewind_vals(&mut self, storage: &Storage);
+    fn rewind_vals(&mut self, storage: &Self::Storage);
 
     /// Rewinds the cursor and outputs its contents to a Vec
-    fn to_vec(&mut self, storage: &Storage) -> Vec<((Self::Key, Self::Val), Vec<(Self::Time, Self::R)>)>
+    fn to_vec(&mut self, storage: &Self::Storage) -> Vec<((Self::Key, Self::Val), Vec<(Self::Time, Self::Diff)>)>
     where
         Self::Key: Clone,
         Self::Val: Clone,
         Self::Time: Clone,
-        Self::R: Clone,
+        Self::Diff: Clone,
     {
         let mut out = Vec::new();
         self.rewind_keys(storage);
diff --git a/src/trace/implementations/ord.rs b/src/trace/implementations/ord.rs
index 8aadd6d02..f63b58a14 100644
--- a/src/trace/implementations/ord.rs
+++ b/src/trace/implementations/ord.rs
@@ -114,7 +114,7 @@ where
     type Key = <L::Target as Update>::Key;
     type Val = <L::Target as Update>::Val;
     type Time = <L::Target as Update>::Time;
-    type R = <L::Target as Update>::Diff;
+    type Diff = <L::Target as Update>::Diff;
 
     type Cursor = OrdValCursor<L>;
     fn cursor(&self) -> Self::Cursor { OrdValCursor { cursor: self.layer.cursor(), phantom: std::marker::PhantomData } }
@@ -337,7 +337,7 @@ where
     cursor: OrderedCursor<VTDLayer<L>>,
 }
 
-impl<L: Layout> Cursor<OrdValBatch<L>> for OrdValCursor<L>
+impl<L: Layout> Cursor for OrdValCursor<L>
 where
     <L::Target as Update>::Key: Sized + Clone,
     <L::Target as Update>::Val: Sized + Clone,
@@ -345,13 +345,13 @@ where
     type Key = <L::Target as Update>::Key;
     type Val = <L::Target as Update>::Val;
     type Time = <L::Target as Update>::Time;
-    type R = <L::Target as Update>::Diff;
+    type Diff = <L::Target as Update>::Diff;
 
-    // type Storage = OrdValBatch<L>;
+    type Storage = OrdValBatch<L>;
 
     fn key<'a>(&self, storage: &'a OrdValBatch<L>) -> &'a Self::Key { &self.cursor.key(&storage.layer) }
     fn val<'a>(&self, storage: &'a OrdValBatch<L>) -> &'a Self::Val { &self.cursor.child.key(&storage.layer.vals) }
-    fn map_times<L2: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &OrdValBatch<L>, mut logic: L2) {
+    fn map_times<L2: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &OrdValBatch<L>, mut logic: L2) {
         self.cursor.child.child.rewind(&storage.layer.vals.vals);
         while self.cursor.child.child.valid(&storage.layer.vals.vals) {
             logic(&self.cursor.child.child.key(&storage.layer.vals.vals).0, &self.cursor.child.child.key(&storage.layer.vals.vals).1);
@@ -382,7 +382,7 @@ impl<L: Layout> Builder for OrdValBuilder<L>
 where
     <L::Target as Update>::Key: Sized + Clone,
     <L::Target as Update>::Val: Sized + Clone,
-    OrdValBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, R=<L::Target as Update>::Diff>
+    OrdValBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>
 {
     type Item = ((<L::Target as Update>::Key, <L::Target as Update>::Val), <L::Target as Update>::Time, <L::Target as Update>::Diff);
     type Time = <L::Target as Update>::Time;
@@ -434,7 +434,7 @@ where
     type Key = <L::Target as Update>::Key;
     type Val = ();
     type Time = <L::Target as Update>::Time;
-    type R = <L::Target as Update>::Diff;
+    type Diff = <L::Target as Update>::Diff;
 
     type Cursor = OrdKeyCursor<L>;
     fn cursor(&self) -> Self::Cursor {
@@ -636,18 +636,20 @@ pub struct OrdKeyCursor<L: Layout> {
     cursor: OrderedCursor<OrderedLeaf<<L::Target as Update>::Time, <L::Target as Update>::Diff>>,
 }
 
-impl<L: Layout> Cursor<OrdKeyBatch<L>> for OrdKeyCursor<L>
+impl<L: Layout> Cursor for OrdKeyCursor<L>
 where
     <L::Target as Update>::Key: Sized,
 {
     type Key = <L::Target as Update>::Key;
     type Val = ();
     type Time = <L::Target as Update>::Time;
-    type R = <L::Target as Update>::Diff;
+    type Diff = <L::Target as Update>::Diff;
+
+    type Storage = OrdKeyBatch<L>;
 
     fn key<'a>(&self, storage: &'a OrdKeyBatch<L>) -> &'a Self::Key { &self.cursor.key(&storage.layer) }
     fn val<'a>(&self, _storage: &'a OrdKeyBatch<L>) -> &'a () { &() }
-    fn map_times<L2: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &OrdKeyBatch<L>, mut logic: L2) {
+    fn map_times<L2: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &OrdKeyBatch<L>, mut logic: L2) {
         self.cursor.child.rewind(&storage.layer.vals);
         while self.cursor.child.valid(&storage.layer.vals) {
             logic(&self.cursor.child.key(&storage.layer.vals).0, &self.cursor.child.key(&storage.layer.vals).1);
@@ -676,7 +678,7 @@ where
 impl<L: Layout> Builder for OrdKeyBuilder<L>
 where
     <L::Target as Update>::Key: Sized + Clone,
-    OrdKeyBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=(), Time=<L::Target as Update>::Time, R=<L::Target as Update>::Diff>
+    OrdKeyBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=(), Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>
 {
     type Item = ((<L::Target as Update>::Key, ()), <L::Target as Update>::Time, <L::Target as Update>::Diff);
     type Time = <L::Target as Update>::Time;
diff --git a/src/trace/implementations/ord_neu.rs b/src/trace/implementations/ord_neu.rs
index beab03dd3..b9d39a7e6 100644
--- a/src/trace/implementations/ord_neu.rs
+++ b/src/trace/implementations/ord_neu.rs
@@ -123,7 +123,7 @@ mod val_batch {
         type Key = <L::Target as Update>::Key;
         type Val = <L::Target as Update>::Val;
         type Time = <L::Target as Update>::Time;
-        type R = <L::Target as Update>::Diff;
+        type Diff = <L::Target as Update>::Diff;
 
         type Cursor = OrdValCursor<L>;
         fn cursor(&self) -> Self::Cursor { 
@@ -420,15 +420,17 @@ mod val_batch {
         phantom: PhantomData<L>,
     }
 
-    impl<L: Layout> Cursor<OrdValBatch<L>> for OrdValCursor<L> {
+    impl<L: Layout> Cursor for OrdValCursor<L> {
         type Key = <L::Target as Update>::Key;
         type Val = <L::Target as Update>::Val;
         type Time = <L::Target as Update>::Time;
-        type R = <L::Target as Update>::Diff;
+        type Diff = <L::Target as Update>::Diff;
+
+        type Storage = OrdValBatch<L>;
 
         fn key<'a>(&self, storage: &'a OrdValBatch<L>) -> &'a Self::Key { storage.storage.keys.index(self.key_cursor) }
         fn val<'a>(&self, storage: &'a OrdValBatch<L>) -> &'a Self::Val { storage.storage.vals.index(self.val_cursor) }
-        fn map_times<L2: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &OrdValBatch<L>, mut logic: L2) {
+        fn map_times<L2: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &OrdValBatch<L>, mut logic: L2) {
             let (lower, upper) = storage.storage.updates_for_value(self.val_cursor);
             for index in lower .. upper {
                 let (time, diff) = &storage.storage.updates.index(index);
@@ -513,7 +515,7 @@ mod val_batch {
 
     impl<L: Layout> Builder for OrdValBuilder<L>
     where
-        OrdValBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, R=<L::Target as Update>::Diff>,
+        OrdValBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
         <L::Target as Update>::KeyOwned: Borrow<<L::Target as Update>::Key>,
     {
         type Item = ((<L::Target as Update>::KeyOwned, <L::Target as Update>::ValOwned), <L::Target as Update>::Time, <L::Target as Update>::Diff);
diff --git a/src/trace/implementations/rhh.rs b/src/trace/implementations/rhh.rs
index e01cf4d45..5d67a8c9a 100644
--- a/src/trace/implementations/rhh.rs
+++ b/src/trace/implementations/rhh.rs
@@ -252,7 +252,7 @@ mod val_batch {
         type Key = <L::Target as Update>::Key;
         type Val = <L::Target as Update>::Val;
         type Time = <L::Target as Update>::Time;
-        type R = <L::Target as Update>::Diff;
+        type Diff = <L::Target as Update>::Diff;
 
         type Cursor = RhhValCursor<L>;
         fn cursor(&self) -> Self::Cursor { 
@@ -591,7 +591,7 @@ mod val_batch {
         phantom: PhantomData<L>,
     }
 
-    impl<L: Layout> Cursor<RhhValBatch<L>> for RhhValCursor<L> 
+    impl<L: Layout> Cursor for RhhValCursor<L> 
     where 
         <L::Target as Update>::Key: HashOrdered,
         <L::Target as Update>::KeyOwned: Default + HashOrdered,
@@ -599,13 +599,15 @@ mod val_batch {
         type Key = <L::Target as Update>::Key;
         type Val = <L::Target as Update>::Val;
         type Time = <L::Target as Update>::Time;
-        type R = <L::Target as Update>::Diff;
+        type Diff = <L::Target as Update>::Diff;
+
+        type Storage = RhhValBatch<L>;
 
         fn key<'a>(&self, storage: &'a RhhValBatch<L>) -> &'a Self::Key { 
             storage.storage.keys.index(self.key_cursor) 
         }
         fn val<'a>(&self, storage: &'a RhhValBatch<L>) -> &'a Self::Val { storage.storage.vals.index(self.val_cursor) }
-        fn map_times<L2: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &RhhValBatch<L>, mut logic: L2) {
+        fn map_times<L2: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &RhhValBatch<L>, mut logic: L2) {
             let (lower, upper) = storage.storage.updates_for_value(self.val_cursor);
             for index in lower .. upper {
                 let (time, diff) = &storage.storage.updates.index(index);
@@ -720,7 +722,7 @@ mod val_batch {
         <L::Target as Update>::Key: HashOrdered,
         <L::Target as Update>::KeyOwned: Default + HashOrdered,
         <L::Target as Update>::KeyOwned: Borrow<<L::Target as Update>::Key>,
-        RhhValBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, R=<L::Target as Update>::Diff>
+        RhhValBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>
     {
         type Item = ((<L::Target as Update>::KeyOwned, <L::Target as Update>::ValOwned), <L::Target as Update>::Time, <L::Target as Update>::Diff);
         type Time = <L::Target as Update>::Time;
diff --git a/src/trace/implementations/spine_fueled.rs b/src/trace/implementations/spine_fueled.rs
index 4d464d324..adab0d121 100644
--- a/src/trace/implementations/spine_fueled.rs
+++ b/src/trace/implementations/spine_fueled.rs
@@ -90,7 +90,7 @@ use ::timely::order::PartialOrder;
 pub struct Spine<B: Batch, BA, BU>
 where
     B::Time: Lattice+Ord,
-    B::R: Semigroup,
+    B::Diff: Semigroup,
     // Intended constraints:
     // BA: Batcher<Time = B::Time>,
     // BU: Builder<Item=BA::Item, Time=BA::Time, Output = B>,
@@ -115,12 +115,12 @@ where
     B::Key: Ord,           // Clone is required by `batch::advance_*` (in-place could remove).
     B::Val: Ord,           // Clone is required by `batch::advance_*` (in-place could remove).
     B::Time: Lattice+timely::progress::Timestamp+Ord+Clone+Debug,
-    B::R: Semigroup,
+    B::Diff: Semigroup,
 {
     type Key = B::Key;
     type Val = B::Val;
     type Time = B::Time;
-    type R = B::R;
+    type Diff = B::Diff;
 
     type Batch = B;
     type Storage = Vec<B>;
@@ -263,7 +263,7 @@ where
     B::Key: Ord,
     B::Val: Ord,
     B::Time: Lattice+timely::progress::Timestamp+Ord+Clone+Debug,
-    B::R: Semigroup,
+    B::Diff: Semigroup,
     BA: Batcher<Time = B::Time>,
     BU: Builder<Item=BA::Item, Time=BA::Time, Output = B>,
 {
@@ -346,7 +346,7 @@ impl<B, BA, BU> Drop for Spine<B, BA, BU>
 where
     B: Batch,
     B::Time: Lattice+Ord,
-    B::R: Semigroup,
+    B::Diff: Semigroup,
 {
     fn drop(&mut self) {
         self.drop_batches();
@@ -358,7 +358,7 @@ impl<B, BA, BU> Spine<B, BA, BU>
 where
     B: Batch,
     B::Time: Lattice+Ord,
-    B::R: Semigroup,
+    B::Diff: Semigroup,
 {
     /// Drops and logs batches. Used in `set_logical_compaction` and drop.
     fn drop_batches(&mut self) {
@@ -406,7 +406,7 @@ where
     B::Key: Ord,
     B::Val: Ord,
     B::Time: Lattice+timely::progress::Timestamp+Ord+Clone+Debug,
-    B::R: Semigroup,
+    B::Diff: Semigroup,
 {
     /// Determine the amount of effort we should exert in the absence of updates.
     ///
diff --git a/src/trace/mod.rs b/src/trace/mod.rs
index f843d140a..da3b7516a 100644
--- a/src/trace/mod.rs
+++ b/src/trace/mod.rs
@@ -53,16 +53,16 @@ pub trait TraceReader {
     /// Timestamps associated with updates
     type Time;
     /// Associated update.
-    type R;
+    type Diff;
 
     /// The type of an immutable collection of updates.
-    type Batch: BatchReader<Key = Self::Key, Val = Self::Val, Time = Self::Time, R = Self::R>+Clone+'static;
+    type Batch: BatchReader<Key = Self::Key, Val = Self::Val, Time = Self::Time, Diff = Self::Diff>+Clone+'static;
 
     /// Storage type for `Self::Cursor`. Likely related to `Self::Batch`.
     type Storage;
 
     /// The type used to enumerate the collections contents.
-    type Cursor: Cursor<Self::Storage, Key = Self::Key, Val = Self::Val, Time = Self::Time, R = Self::R>;
+    type Cursor: Cursor<Storage=Self::Storage, Key = Self::Key, Val = Self::Val, Time = Self::Time, Diff = Self::Diff>;
 
     /// Provides a cursor over updates contained in the trace.
     fn cursor(&mut self) -> (Self::Cursor, Self::Storage) {
@@ -263,10 +263,10 @@ where
     /// Timestamps associated with updates
     type Time: Timestamp;
     /// Associated update.
-    type R;
+    type Diff;
 
     /// The type used to enumerate the batch's contents.
-    type Cursor: Cursor<Self, Key = Self::Key, Val = Self::Val, Time = Self::Time, R = Self::R>;
+    type Cursor: Cursor<Storage=Self, Key = Self::Key, Val = Self::Val, Time = Self::Time, Diff = Self::Diff>;
     /// Acquires a cursor to the batch's contents.
     fn cursor(&self) -> Self::Cursor;
     /// The number of updates in the batch.
@@ -379,7 +379,7 @@ pub mod rc_blanket_impls {
         type Key = B::Key;
         type Val = B::Val;
         type Time = B::Time;
-        type R = B::R;
+        type Diff = B::Diff;
 
         /// The type used to enumerate the batch's contents.
         type Cursor = RcBatchCursor<B>;
@@ -407,12 +407,14 @@ pub mod rc_blanket_impls {
         }
     }
 
-    impl<B: BatchReader> Cursor<Rc<B>> for RcBatchCursor<B> {
+    impl<B: BatchReader> Cursor for RcBatchCursor<B> {
 
         type Key = B::Key;
         type Val = B::Val;
         type Time = B::Time;
-        type R = B::R;
+        type Diff = B::Diff;
+
+        type Storage = Rc<B>;
 
         #[inline] fn key_valid(&self, storage: &Rc<B>) -> bool { self.cursor.key_valid(storage) }
         #[inline] fn val_valid(&self, storage: &Rc<B>) -> bool { self.cursor.val_valid(storage) }
@@ -421,7 +423,7 @@ pub mod rc_blanket_impls {
         #[inline] fn val<'a>(&self, storage: &'a Rc<B>) -> &'a Self::Val { self.cursor.val(storage) }
 
         #[inline]
-        fn map_times<L: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &Rc<B>, logic: L) {
+        fn map_times<L: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &Rc<B>, logic: L) {
             self.cursor.map_times(storage, logic)
         }
 
@@ -482,7 +484,7 @@ pub mod abomonated_blanket_impls {
         type Key = B::Key;
         type Val = B::Val;
         type Time = B::Time;
-        type R = B::R;
+        type Diff = B::Diff;
 
         /// The type used to enumerate the batch's contents.
         type Cursor = AbomonatedBatchCursor<B>;
@@ -510,12 +512,14 @@ pub mod abomonated_blanket_impls {
         }
     }
 
-    impl<B: BatchReader+Abomonation> Cursor<Abomonated<B, Vec<u8>>> for AbomonatedBatchCursor<B> {
+    impl<B: BatchReader+Abomonation> Cursor for AbomonatedBatchCursor<B> {
 
         type Key = B::Key;
         type Val = B::Val;
         type Time = B::Time;
-        type R = B::R;
+        type Diff = B::Diff;
+
+        type Storage = Abomonated<B, Vec<u8>>;
 
         #[inline] fn key_valid(&self, storage: &Abomonated<B, Vec<u8>>) -> bool { self.cursor.key_valid(storage) }
         #[inline] fn val_valid(&self, storage: &Abomonated<B, Vec<u8>>) -> bool { self.cursor.val_valid(storage) }
@@ -524,7 +528,7 @@ pub mod abomonated_blanket_impls {
         #[inline] fn val<'a>(&self, storage: &'a Abomonated<B, Vec<u8>>) -> &'a Self::Val { self.cursor.val(storage) }
 
         #[inline]
-        fn map_times<L: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &Abomonated<B, Vec<u8>>, logic: L) {
+        fn map_times<L: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &Abomonated<B, Vec<u8>>, logic: L) {
             self.cursor.map_times(storage, logic)
         }
 
diff --git a/src/trace/wrappers/enter.rs b/src/trace/wrappers/enter.rs
index 9ed2d439d..00a279b04 100644
--- a/src/trace/wrappers/enter.rs
+++ b/src/trace/wrappers/enter.rs
@@ -39,13 +39,13 @@ where
     Tr::Key: 'static,
     Tr::Val: 'static,
     Tr::Time: Timestamp,
-    Tr::R: 'static,
+    Tr::Diff: 'static,
     TInner: Refines<Tr::Time>+Lattice,
 {
     type Key = Tr::Key;
     type Val = Tr::Val;
     type Time = TInner;
-    type R = Tr::R;
+    type Diff = Tr::Diff;
 
     type Batch = BatchEnter<Tr::Batch, TInner>;
     type Storage = Tr::Storage;
@@ -129,7 +129,7 @@ where
     type Key = B::Key;
     type Val = B::Val;
     type Time = TInner;
-    type R = B::R;
+    type Diff = B::Diff;
 
     type Cursor = BatchCursorEnter<B, TInner>;
 
@@ -174,38 +174,40 @@ impl<C, TInner> CursorEnter<C, TInner> {
     }
 }
 
-impl<S, C, TInner> Cursor<S> for CursorEnter<C, TInner>
+impl<C, TInner> Cursor for CursorEnter<C, TInner>
 where
-    C: Cursor<S>,
+    C: Cursor,
     C::Time: Timestamp,
     TInner: Refines<C::Time>+Lattice,
 {
     type Key = C::Key;
     type Val = C::Val;
     type Time = TInner;
-    type R = C::R;
+    type Diff = C::Diff;
 
-    #[inline] fn key_valid(&self, storage: &S) -> bool { self.cursor.key_valid(storage) }
-    #[inline] fn val_valid(&self, storage: &S) -> bool { self.cursor.val_valid(storage) }
+    type Storage = C::Storage;
 
-    #[inline] fn key<'a>(&self, storage: &'a S) -> &'a Self::Key { self.cursor.key(storage) }
-    #[inline] fn val<'a>(&self, storage: &'a S) -> &'a Self::Val { self.cursor.val(storage) }
+    #[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) -> &'a Self::Key { self.cursor.key(storage) }
+    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> &'a Self::Val { self.cursor.val(storage) }
 
     #[inline]
-    fn map_times<L: FnMut(&TInner, &Self::R)>(&mut self, storage: &S, mut logic: L) {
+    fn map_times<L: FnMut(&TInner, &Self::Diff)>(&mut self, storage: &Self::Storage, mut logic: L) {
         self.cursor.map_times(storage, |time, diff| {
             logic(&TInner::to_inner(time.clone()), diff)
         })
     }
 
-    #[inline] fn step_key(&mut self, storage: &S) { self.cursor.step_key(storage) }
-    #[inline] fn seek_key(&mut self, storage: &S, key: &Self::Key) { self.cursor.seek_key(storage, key) }
+    #[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: &S) { self.cursor.step_val(storage) }
-    #[inline] fn seek_val(&mut self, storage: &S, val: &Self::Val) { self.cursor.seek_val(storage, val) }
+    #[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: &S) { self.cursor.rewind_keys(storage) }
-    #[inline] fn rewind_vals(&mut self, storage: &S) { self.cursor.rewind_vals(storage) }
+    #[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) }
 }
 
 
@@ -225,7 +227,7 @@ impl<B: BatchReader, TInner> BatchCursorEnter<B, TInner> {
     }
 }
 
-impl<TInner, B: BatchReader> Cursor<BatchEnter<B, TInner>> for BatchCursorEnter<B, TInner>
+impl<TInner, B: BatchReader> Cursor for BatchCursorEnter<B, TInner>
 where
     B::Time: Timestamp,
     TInner: Refines<B::Time>+Lattice,
@@ -233,7 +235,9 @@ where
     type Key = B::Key;
     type Val = B::Val;
     type Time = TInner;
-    type R = B::R;
+    type Diff = B::Diff;
+
+    type Storage = BatchEnter<B, TInner>;
 
     #[inline] fn key_valid(&self, storage: &BatchEnter<B, TInner>) -> bool { self.cursor.key_valid(&storage.batch) }
     #[inline] fn val_valid(&self, storage: &BatchEnter<B, TInner>) -> bool { self.cursor.val_valid(&storage.batch) }
@@ -242,7 +246,7 @@ where
     #[inline] fn val<'a>(&self, storage: &'a BatchEnter<B, TInner>) -> &'a Self::Val { self.cursor.val(&storage.batch) }
 
     #[inline]
-    fn map_times<L: FnMut(&TInner, &Self::R)>(&mut self, storage: &BatchEnter<B, TInner>, mut logic: L) {
+    fn map_times<L: FnMut(&TInner, &Self::Diff)>(&mut self, storage: &BatchEnter<B, TInner>, mut logic: L) {
         self.cursor.map_times(&storage.batch, |time, diff| {
             logic(&TInner::to_inner(time.clone()), diff)
         })
diff --git a/src/trace/wrappers/enter_at.rs b/src/trace/wrappers/enter_at.rs
index 027817bce..18ed0cf01 100644
--- a/src/trace/wrappers/enter_at.rs
+++ b/src/trace/wrappers/enter_at.rs
@@ -52,7 +52,7 @@ where
     Tr::Val: 'static,
     Tr::Time: Timestamp,
     TInner: Refines<Tr::Time>+Lattice,
-    Tr::R: 'static,
+    Tr::Diff: 'static,
     F: 'static,
     F: FnMut(&Tr::Key, &Tr::Val, &Tr::Time)->TInner+Clone,
     G: FnMut(&TInner)->Tr::Time+Clone+'static,
@@ -60,7 +60,7 @@ where
     type Key = Tr::Key;
     type Val = Tr::Val;
     type Time = TInner;
-    type R = Tr::R;
+    type Diff = Tr::Diff;
 
     type Batch = BatchEnter<Tr::Batch, TInner,F>;
     type Storage = Tr::Storage;
@@ -149,7 +149,7 @@ where
     type Key = B::Key;
     type Val = B::Val;
     type Time = TInner;
-    type R = B::R;
+    type Diff = B::Diff;
 
     type Cursor = BatchCursorEnter<B, TInner, F>;
 
@@ -197,9 +197,9 @@ impl<C, TInner, F> CursorEnter<C, TInner, F> {
     }
 }
 
-impl<S, C, TInner, F> Cursor<S> for CursorEnter<C, TInner, F>
+impl<C, TInner, F> Cursor for CursorEnter<C, TInner, F>
 where
-    C: Cursor<S>,
+    C: Cursor,
     C::Time: Timestamp,
     TInner: Refines<C::Time>+Lattice,
     F: FnMut(&C::Key, &C::Val, &C::Time)->TInner,
@@ -207,16 +207,18 @@ where
     type Key = C::Key;
     type Val = C::Val;
     type Time = TInner;
-    type R = C::R;
+    type Diff = C::Diff;
 
-    #[inline] fn key_valid(&self, storage: &S) -> bool { self.cursor.key_valid(storage) }
-    #[inline] fn val_valid(&self, storage: &S) -> bool { self.cursor.val_valid(storage) }
+    type Storage = C::Storage;
 
-    #[inline] fn key<'a>(&self, storage: &'a S) -> &'a Self::Key { self.cursor.key(storage) }
-    #[inline] fn val<'a>(&self, storage: &'a S) -> &'a Self::Val { self.cursor.val(storage) }
+    #[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) -> &'a Self::Key { self.cursor.key(storage) }
+    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> &'a Self::Val { self.cursor.val(storage) }
 
     #[inline]
-    fn map_times<L: FnMut(&TInner, &Self::R)>(&mut self, storage: &S, mut logic: L) {
+    fn map_times<L: FnMut(&TInner, &Self::Diff)>(&mut self, storage: &Self::Storage, mut logic: L) {
         let key = self.key(storage);
         let val = self.val(storage);
         let logic2 = &mut self.logic;
@@ -225,14 +227,14 @@ where
         })
     }
 
-    #[inline] fn step_key(&mut self, storage: &S) { self.cursor.step_key(storage) }
-    #[inline] fn seek_key(&mut self, storage: &S, key: &Self::Key) { self.cursor.seek_key(storage, key) }
+    #[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: &S) { self.cursor.step_val(storage) }
-    #[inline] fn seek_val(&mut self, storage: &S, val: &Self::Val) { self.cursor.seek_val(storage, val) }
+    #[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: &S) { self.cursor.rewind_keys(storage) }
-    #[inline] fn rewind_vals(&mut self, storage: &S) { self.cursor.rewind_vals(storage) }
+    #[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) }
 }
 
 
@@ -254,7 +256,7 @@ impl<B: BatchReader, TInner, F> BatchCursorEnter<B, TInner, F> {
     }
 }
 
-impl<TInner, B: BatchReader, F> Cursor<BatchEnter<B, TInner, F>> for BatchCursorEnter<B, TInner, F>
+impl<TInner, B: BatchReader, F> Cursor for BatchCursorEnter<B, TInner, F>
 where
     B::Time: Timestamp,
     TInner: Refines<B::Time>+Lattice,
@@ -263,7 +265,9 @@ where
     type Key = B::Key;
     type Val = B::Val;
     type Time = TInner;
-    type R = B::R;
+    type Diff = B::Diff;
+
+    type Storage = BatchEnter<B, TInner, F>;
 
     #[inline] fn key_valid(&self, storage: &BatchEnter<B, TInner, F>) -> bool { self.cursor.key_valid(&storage.batch) }
     #[inline] fn val_valid(&self, storage: &BatchEnter<B, TInner, F>) -> bool { self.cursor.val_valid(&storage.batch) }
@@ -272,7 +276,7 @@ where
     #[inline] fn val<'a>(&self, storage: &'a BatchEnter<B, TInner, F>) -> &'a Self::Val { self.cursor.val(&storage.batch) }
 
     #[inline]
-    fn map_times<L: FnMut(&TInner, &Self::R)>(&mut self, storage: &BatchEnter<B, TInner, F>, mut logic: L) {
+    fn map_times<L: FnMut(&TInner, &Self::Diff)>(&mut self, storage: &BatchEnter<B, TInner, F>, mut logic: L) {
         let key = self.key(storage);
         let val = self.val(storage);
         let logic2 = &mut self.logic;
diff --git a/src/trace/wrappers/filter.rs b/src/trace/wrappers/filter.rs
index 64ff943b9..837111048 100644
--- a/src/trace/wrappers/filter.rs
+++ b/src/trace/wrappers/filter.rs
@@ -32,13 +32,13 @@ where
     Tr::Key: 'static,
     Tr::Val: 'static,
     Tr::Time: Timestamp,
-    Tr::R: 'static,
+    Tr::Diff: 'static,
     F: FnMut(&Tr::Key, &Tr::Val)->bool+Clone+'static,
 {
     type Key = Tr::Key;
     type Val = Tr::Val;
     type Time = Tr::Time;
-    type R = Tr::R;
+    type Diff = Tr::Diff;
 
     type Batch = BatchFilter<Tr::Batch, F>;
     type Storage = Tr::Storage;
@@ -92,7 +92,7 @@ where
     type Key = B::Key;
     type Val = B::Val;
     type Time = B::Time;
-    type R = B::R;
+    type Diff = B::Diff;
 
     type Cursor = BatchCursorFilter<B, F>;
 
@@ -132,25 +132,27 @@ impl<C, F> CursorFilter<C, F> {
     }
 }
 
-impl<S, C, F> Cursor<S> for CursorFilter<C, F>
+impl<C, F> Cursor for CursorFilter<C, F>
 where
-    C: Cursor<S>,
+    C: Cursor,
     C::Time: Timestamp,
     F: FnMut(&C::Key, &C::Val)->bool+'static
 {
     type Key = C::Key;
     type Val = C::Val;
     type Time = C::Time;
-    type R = C::R;
+    type Diff = C::Diff;
 
-    #[inline] fn key_valid(&self, storage: &S) -> bool { self.cursor.key_valid(storage) }
-    #[inline] fn val_valid(&self, storage: &S) -> bool { self.cursor.val_valid(storage) }
+    type Storage = C::Storage;
 
-    #[inline] fn key<'a>(&self, storage: &'a S) -> &'a Self::Key { self.cursor.key(storage) }
-    #[inline] fn val<'a>(&self, storage: &'a S) -> &'a Self::Val { self.cursor.val(storage) }
+    #[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) -> &'a Self::Key { self.cursor.key(storage) }
+    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> &'a Self::Val { self.cursor.val(storage) }
 
     #[inline]
-    fn map_times<L: FnMut(&Self::Time,&Self::R)>(&mut self, storage: &S, logic: L) {
+    fn map_times<L: FnMut(&Self::Time,&Self::Diff)>(&mut self, storage: &Self::Storage, logic: L) {
         let key = self.key(storage);
         let val = self.val(storage);
         if (self.logic)(key, val) {
@@ -158,14 +160,14 @@ where
         }
     }
 
-    #[inline] fn step_key(&mut self, storage: &S) { self.cursor.step_key(storage) }
-    #[inline] fn seek_key(&mut self, storage: &S, key: &Self::Key) { self.cursor.seek_key(storage, key) }
+    #[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: &S) { self.cursor.step_val(storage) }
-    #[inline] fn seek_val(&mut self, storage: &S, val: &Self::Val) { self.cursor.seek_val(storage, val) }
+    #[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: &S) { self.cursor.rewind_keys(storage) }
-    #[inline] fn rewind_vals(&mut self, storage: &S) { self.cursor.rewind_vals(storage) }
+    #[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) }
 }
 
 
@@ -185,7 +187,7 @@ impl<B: BatchReader, F> BatchCursorFilter<B, F> {
     }
 }
 
-impl<B: BatchReader, F> Cursor<BatchFilter<B, F>> for BatchCursorFilter<B, F>
+impl<B: BatchReader, F> Cursor for BatchCursorFilter<B, F>
 where
     B::Time: Timestamp,
     F: FnMut(&B::Key, &B::Val)->bool+'static,
@@ -193,7 +195,9 @@ where
     type Key = B::Key;
     type Val = B::Val;
     type Time = B::Time;
-    type R = B::R;
+    type Diff = B::Diff;
+
+    type Storage = BatchFilter<B, F>;
 
     #[inline] fn key_valid(&self, storage: &BatchFilter<B, F>) -> bool { self.cursor.key_valid(&storage.batch) }
     #[inline] fn val_valid(&self, storage: &BatchFilter<B, F>) -> bool { self.cursor.val_valid(&storage.batch) }
@@ -202,7 +206,7 @@ where
     #[inline] fn val<'a>(&self, storage: &'a BatchFilter<B, F>) -> &'a Self::Val { self.cursor.val(&storage.batch) }
 
     #[inline]
-    fn map_times<L: FnMut(&Self::Time,&Self::R)>(&mut self, storage: &BatchFilter<B, F>, logic: L) {
+    fn map_times<L: FnMut(&Self::Time,&Self::Diff)>(&mut self, storage: &BatchFilter<B, F>, logic: L) {
         let key = self.key(storage);
         let val = self.val(storage);
         if (self.logic)(key, val) {
diff --git a/src/trace/wrappers/freeze.rs b/src/trace/wrappers/freeze.rs
index ab3800291..2080a9a22 100644
--- a/src/trace/wrappers/freeze.rs
+++ b/src/trace/wrappers/freeze.rs
@@ -40,7 +40,7 @@ where
     T: TraceReader<Time=G::Timestamp>+Clone,
     T::Key: 'static,
     T::Val: 'static,
-    T::R: 'static,
+    T::Diff: 'static,
     F: Fn(&G::Timestamp)->Option<G::Timestamp>+'static,
 {
     let func1 = Rc::new(func);
@@ -83,13 +83,13 @@ where
     Tr::Key: 'static,
     Tr::Val: 'static,
     Tr::Time: Lattice+Clone+'static,
-    Tr::R: 'static,
+    Tr::Diff: 'static,
     F: Fn(&Tr::Time)->Option<Tr::Time>+'static,
 {
     type Key = Tr::Key;
     type Val = Tr::Val;
     type Time = Tr::Time;
-    type R = Tr::R;
+    type Diff = Tr::Diff;
 
     type Batch = BatchFreeze<Tr::Batch, F>;
     type Storage = Tr::Storage;
@@ -122,7 +122,7 @@ where
     Tr::Key: 'static,
     Tr::Val: 'static,
     Tr::Time: Lattice+Clone+'static,
-    Tr::R: 'static,
+    Tr::Diff: 'static,
     F: Fn(&Tr::Time)->Option<Tr::Time>,
 {
     /// Makes a new trace wrapper
@@ -159,7 +159,7 @@ where
     type Key = B::Key;
     type Val = B::Val;
     type Time = B::Time;
-    type R = B::R;
+    type Diff = B::Diff;
 
     type Cursor = BatchCursorFreeze<B, F>;
 
@@ -200,24 +200,26 @@ impl<C, F> CursorFreeze<C, F> {
     }
 }
 
-impl<S, C, F> Cursor<S> for CursorFreeze<C, F>
+impl<C, F> Cursor for CursorFreeze<C, F>
 where
-    C: Cursor<S>,
+    C: Cursor,
     C::Time: Sized,
     F: Fn(&C::Time)->Option<C::Time>,
 {
     type Key = C::Key;
     type Val = C::Val;
     type Time = C::Time;
-    type R = C::R;
+    type Diff = C::Diff;
 
-    #[inline] fn key_valid(&self, storage: &S) -> bool { self.cursor.key_valid(storage) }
-    #[inline] fn val_valid(&self, storage: &S) -> bool { self.cursor.val_valid(storage) }
+    type Storage = C::Storage;
 
-    #[inline] fn key<'a>(&self, storage: &'a S) -> &'a Self::Key { self.cursor.key(storage) }
-    #[inline] fn val<'a>(&self, storage: &'a S) -> &'a Self::Val { self.cursor.val(storage) }
+    #[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 map_times<L: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &S, mut logic: L) {
+    #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> &'a Self::Key { self.cursor.key(storage) }
+    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> &'a Self::Val { self.cursor.val(storage) }
+
+    #[inline] fn map_times<L: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &Self::Storage, mut logic: L) {
         let func = &self.func;
         self.cursor.map_times(storage, |time, diff| {
             if let Some(time) = func(time) {
@@ -226,14 +228,14 @@ where
         })
     }
 
-    #[inline] fn step_key(&mut self, storage: &S) { self.cursor.step_key(storage) }
-    #[inline] fn seek_key(&mut self, storage: &S, key: &Self::Key) { self.cursor.seek_key(storage, key) }
+    #[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: &S) { self.cursor.step_val(storage) }
-    #[inline] fn seek_val(&mut self, storage: &S, val: &Self::Val) { self.cursor.seek_val(storage, val) }
+    #[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: &S) { self.cursor.rewind_keys(storage) }
-    #[inline] fn rewind_vals(&mut self, storage: &S) { self.cursor.rewind_vals(storage) }
+    #[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) }
 }
 
 
@@ -252,14 +254,16 @@ impl<B: BatchReader, F> BatchCursorFreeze<B, F> {
     }
 }
 
-impl<B: BatchReader, F> Cursor<BatchFreeze<B, F>> for BatchCursorFreeze<B, F>
+impl<B: BatchReader, F> Cursor for BatchCursorFreeze<B, F>
 where
     F: Fn(&B::Time)->Option<B::Time>,
 {
     type Key = B::Key;
     type Val = B::Val;
     type Time = B::Time;
-    type R = B::R;
+    type Diff = B::Diff;
+
+    type Storage = BatchFreeze<B, F>;
 
     #[inline] fn key_valid(&self, storage: &BatchFreeze<B, F>) -> bool { self.cursor.key_valid(&storage.batch) }
     #[inline] fn val_valid(&self, storage: &BatchFreeze<B, F>) -> bool { self.cursor.val_valid(&storage.batch) }
@@ -267,7 +271,7 @@ where
     #[inline] fn key<'a>(&self, storage: &'a BatchFreeze<B, F>) -> &'a Self::Key { self.cursor.key(&storage.batch) }
     #[inline] fn val<'a>(&self, storage: &'a BatchFreeze<B, F>) -> &'a Self::Val { self.cursor.val(&storage.batch) }
 
-    #[inline] fn map_times<L: FnMut(&Self::Time, &Self::R)>(&mut self, storage: &BatchFreeze<B, F>, mut logic: L) {
+    #[inline] fn map_times<L: FnMut(&Self::Time, &Self::Diff)>(&mut self, storage: &BatchFreeze<B, F>, mut logic: L) {
         let func = &self.func;
         self.cursor.map_times(&storage.batch, |time, diff| {
             if let Some(time) = func(time) {
diff --git a/src/trace/wrappers/frontier.rs b/src/trace/wrappers/frontier.rs
index 54be6fade..ffe9c073a 100644
--- a/src/trace/wrappers/frontier.rs
+++ b/src/trace/wrappers/frontier.rs
@@ -46,12 +46,12 @@ where
     Tr::Key: 'static,
     Tr::Val: 'static,
     Tr::Time: Timestamp+Lattice,
-    Tr::R: 'static,
+    Tr::Diff: 'static,
 {
     type Key = Tr::Key;
     type Val = Tr::Val;
     type Time = Tr::Time;
-    type R = Tr::R;
+    type Diff = Tr::Diff;
 
     type Batch = BatchFrontier<Tr::Batch>;
     type Storage = Tr::Storage;
@@ -108,7 +108,7 @@ where
     type Key = B::Key;
     type Val = B::Val;
     type Time = B::Time;
-    type R = B::R;
+    type Diff = B::Diff;
 
     type Cursor = BatchCursorFrontier<B>;
 
@@ -151,24 +151,26 @@ impl<C, T> CursorFrontier<C, T> where T: Clone {
     }
 }
 
-impl<S, C, T> Cursor<S> for CursorFrontier<C, T>
+impl<C, T> Cursor for CursorFrontier<C, T>
 where
-    C: Cursor<S, Time=T>,
+    C: Cursor<Time=T>,
     T: Timestamp+Lattice,
 {
     type Key = C::Key;
     type Val = C::Val;
     type Time = C::Time;
-    type R = C::R;
+    type Diff = C::Diff;
 
-    #[inline] fn key_valid(&self, storage: &S) -> bool { self.cursor.key_valid(storage) }
-    #[inline] fn val_valid(&self, storage: &S) -> bool { self.cursor.val_valid(storage) }
+    type Storage = C::Storage;
 
-    #[inline] fn key<'a>(&self, storage: &'a S) -> &'a Self::Key { self.cursor.key(storage) }
-    #[inline] fn val<'a>(&self, storage: &'a S) -> &'a Self::Val { self.cursor.val(storage) }
+    #[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) -> &'a Self::Key { self.cursor.key(storage) }
+    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> &'a Self::Val { self.cursor.val(storage) }
 
     #[inline]
-    fn map_times<L: FnMut(&Self::Time,&Self::R)>(&mut self, storage: &S, mut logic: L) {
+    fn map_times<L: FnMut(&Self::Time,&Self::Diff)>(&mut self, storage: &Self::Storage, mut logic: L) {
         let since = self.since.borrow();
         let until = self.until.borrow();
         let mut temp: C::Time = <C::Time as timely::progress::Timestamp>::minimum();
@@ -181,14 +183,14 @@ where
         })
     }
 
-    #[inline] fn step_key(&mut self, storage: &S) { self.cursor.step_key(storage) }
-    #[inline] fn seek_key(&mut self, storage: &S, key: &Self::Key) { self.cursor.seek_key(storage, key) }
+    #[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: &S) { self.cursor.step_val(storage) }
-    #[inline] fn seek_val(&mut self, storage: &S, val: &Self::Val) { self.cursor.seek_val(storage, val) }
+    #[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: &S) { self.cursor.rewind_keys(storage) }
-    #[inline] fn rewind_vals(&mut self, storage: &S) { self.cursor.rewind_vals(storage) }
+    #[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) }
 }
 
 
@@ -210,14 +212,16 @@ impl<B: BatchReader> BatchCursorFrontier<B> where B::Time: Clone {
     }
 }
 
-impl<B: BatchReader> Cursor<BatchFrontier<B>> for BatchCursorFrontier<B>
+impl<B: BatchReader> Cursor for BatchCursorFrontier<B>
 where
     B::Time: Timestamp+Lattice,
 {
     type Key = B::Key;
     type Val = B::Val;
     type Time = B::Time;
-    type R = B::R;
+    type Diff = B::Diff;
+
+    type Storage = BatchFrontier<B>;
 
     #[inline] fn key_valid(&self, storage: &BatchFrontier<B>) -> bool { self.cursor.key_valid(&storage.batch) }
     #[inline] fn val_valid(&self, storage: &BatchFrontier<B>) -> bool { self.cursor.val_valid(&storage.batch) }
@@ -226,7 +230,7 @@ where
     #[inline] fn val<'a>(&self, storage: &'a BatchFrontier<B>) -> &'a Self::Val { self.cursor.val(&storage.batch) }
 
     #[inline]
-    fn map_times<L: FnMut(&Self::Time,&Self::R)>(&mut self, storage: &BatchFrontier<B>, mut logic: L) {
+    fn map_times<L: FnMut(&Self::Time,&Self::Diff)>(&mut self, storage: &BatchFrontier<B>, mut logic: L) {
         let since = self.since.borrow();
         let until = self.until.borrow();
         let mut temp: B::Time = <B::Time as timely::progress::Timestamp>::minimum();
diff --git a/src/trace/wrappers/rc.rs b/src/trace/wrappers/rc.rs
index 67854c991..5dc094230 100644
--- a/src/trace/wrappers/rc.rs
+++ b/src/trace/wrappers/rc.rs
@@ -99,7 +99,7 @@ where
     type Key = Tr::Key;
     type Val = Tr::Val;
     type Time = Tr::Time;
-    type R = Tr::R;
+    type Diff = Tr::Diff;
 
     type Batch = Tr::Batch;
     type Storage = Tr::Storage;