diff --git a/differential-dataflow/src/algorithms/graphs/bfs.rs b/differential-dataflow/src/algorithms/graphs/bfs.rs
index 8a9428993..1a419b8a5 100644
--- a/differential-dataflow/src/algorithms/graphs/bfs.rs
+++ b/differential-dataflow/src/algorithms/graphs/bfs.rs
@@ -18,14 +18,15 @@ where
     bfs_arranged(edges, roots)
 }
 
-use crate::trace::TraceReader;
+use crate::trace::{BatchCursor, Cursor, Navigable, TraceReader};
 use crate::operators::arrange::Arranged;
 
 /// Returns pairs (node, dist) indicating distance of each node from a root.
 pub fn bfs_arranged<'scope, N, Tr>(edges: Arranged<'scope, Tr>, roots: VecCollection<'scope, Tr::Time, N>) -> VecCollection<'scope, Tr::Time, (N, u32)>
 where
     N: ExchangeData+Hash,
-    Tr: for<'a> TraceReader<Key<'a>=&'a N, Val<'a>=&'a N, Diff=isize>+Clone+'static,
+    Tr: TraceReader<Batch: Navigable>+Clone+'static,
+    for<'a> BatchCursor<Tr>: Cursor<Key<'a>=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=isize>,
 {
     // initialize roots as reaching themselves at distance 0
     let nodes = roots.map(|x| (x, 0));
diff --git a/differential-dataflow/src/algorithms/graphs/bijkstra.rs b/differential-dataflow/src/algorithms/graphs/bijkstra.rs
index fce691af5..4d2034382 100644
--- a/differential-dataflow/src/algorithms/graphs/bijkstra.rs
+++ b/differential-dataflow/src/algorithms/graphs/bijkstra.rs
@@ -29,7 +29,7 @@ where
     bidijkstra_arranged(forward, reverse, goals)
 }
 
-use crate::trace::TraceReader;
+use crate::trace::{BatchCursor, Cursor, Navigable, TraceReader};
 use crate::operators::arrange::Arranged;
 
 /// Bi-directional Dijkstra search using arranged forward and reverse edge collections.
@@ -40,7 +40,8 @@ pub fn bidijkstra_arranged<'scope, N, Tr>(
 ) -> VecCollection<'scope, Tr::Time, ((N,N), u32)>
 where
     N: ExchangeData+Hash,
-    Tr: for<'a> TraceReader<Key<'a>=&'a N, Val<'a>=&'a N, Diff=isize>+Clone+'static,
+    Tr: TraceReader<Batch: Navigable>+Clone+'static,
+    for<'a> BatchCursor<Tr>: Cursor<Key<'a>=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=isize>,
 {
     let outer = forward.stream.scope();
     outer.iterative::<u64,_,_>(|inner| {
diff --git a/differential-dataflow/src/algorithms/graphs/propagate.rs b/differential-dataflow/src/algorithms/graphs/propagate.rs
index a1afe57ff..a691b3279 100644
--- a/differential-dataflow/src/algorithms/graphs/propagate.rs
+++ b/differential-dataflow/src/algorithms/graphs/propagate.rs
@@ -43,7 +43,7 @@ where
     propagate_core(edges.arrange_by_key(), nodes, logic)
 }
 
-use crate::trace::TraceReader;
+use crate::trace::{BatchCursor, Cursor, Navigable, TraceReader};
 use crate::operators::arrange::arrangement::Arranged;
 
 /// Propagates labels forward, retaining the minimum label.
@@ -58,7 +58,8 @@ where
     R: Multiply<R, Output=R>,
     R: From<i8>,
     L: ExchangeData,
-    Tr: for<'a> TraceReader<Key<'a>=&'a N, Val<'a>=&'a N, Diff=R, Time: Hash>+Clone+'static,
+    Tr: TraceReader<Batch: Navigable, Time: Hash>+Clone+'static,
+    for<'a> BatchCursor<Tr>: Cursor<Key<'a>=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=R>,
     F: Fn(&L)->u64+Clone+'static,
 {
     // Morally the code performs the following iterative computation. However, in the interest of a simplified
diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs
index a319f5c0b..5defe78c2 100644
--- a/differential-dataflow/src/collection.rs
+++ b/differential-dataflow/src/collection.rs
@@ -329,6 +329,8 @@ pub mod vec {
     use crate::difference::{Semigroup, Abelian, Multiply};
     use crate::lattice::Lattice;
     use crate::hashable::Hashable;
+    use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchVal, Navigable};
+    use crate::trace::Cursor;
 
     /// An evolving collection of values of type `D`, backed by Rust `Vec` types as containers.
     ///
@@ -781,9 +783,10 @@ pub mod vec {
         /// ```
         pub fn reduce_abelian<L, Bu, T2>(self, name: &str, mut logic: L) -> Arranged<'scope, TraceAgent<T2>>
         where
-            T2: for<'a> Trace<Key<'a>= &'a K, ValOwn = V, Time=T, Diff: Abelian>+'static,
-            Bu: Builder<Time=T2::Time, Input = Vec<((K, V), T2::Time, T2::Diff)>, Output = T2::Batch>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(V, T2::Diff)>)+'static,
+            T2: Trace<Batch: Navigable, Time=T>+'static,
+            for<'a> BatchCursor<T2>: Cursor<Key<'a>= &'a K, ValOwn = V, Time = T2::Time, Diff: Abelian>,
+            Bu: Builder<Time=T2::Time, Input = Vec<((K, V), T2::Time, BatchDiff<T2>)>, Output = T2::Batch>,
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(V, BatchDiff<T2>)>)+'static,
         {
             self.reduce_core::<_,Bu,T2>(name, move |key, input, output, change| {
                 if !input.is_empty() { logic(key, input, change); }
@@ -800,9 +803,10 @@ pub mod vec {
         pub fn reduce_core<L, Bu, T2>(self, name: &str, logic: L) -> Arranged<'scope, TraceAgent<T2>>
         where
             V: Clone+'static,
-            T2: for<'a> Trace<Key<'a>=&'a K, ValOwn = V, Time=T>+'static,
-            Bu: Builder<Time=T2::Time, Input = Vec<((K, V), T2::Time, T2::Diff)>, Output = T2::Batch>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(V,T2::Diff)>, &mut Vec<(V, T2::Diff)>)+'static,
+            T2: Trace<Batch: Navigable, Time=T>+'static,
+            for<'a> BatchCursor<T2>: Cursor<Key<'a>=&'a K, ValOwn = V, Time = T2::Time>,
+            Bu: Builder<Time=T2::Time, Input = Vec<((K, V), T2::Time, BatchDiff<T2>)>, Output = T2::Batch>,
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(V,BatchDiff<T2>)>, &mut Vec<(V, BatchDiff<T2>)>)+'static,
         {
             self.arrange_by_key_named(&format!("Arrange: {}", name))
                 .reduce_core::<_,Bu,_,_>(
@@ -962,9 +966,10 @@ pub mod vec {
         pub fn consolidate_named<Ba, Bu, Tr, F>(self, name: &str, reify: F) -> Self
         where
             Ba: crate::trace::Batcher<Output=Vec<((D, ()), T, R)>, Time=T> + 'static,
-            Tr: for<'a> crate::trace::Trace<Time=T,Diff=R>+'static,
+            Tr: crate::trace::Trace<Batch: Navigable, Time=T>+'static,
+            for<'a> BatchCursor<Tr>: Cursor<Time=Tr::Time, Diff=R>,
             Bu: crate::trace::Builder<Time=Tr::Time, Input=Vec<((D, ()), T, R)>, Output=Tr::Batch>,
-            F: Fn(Tr::Key<'_>, Tr::Val<'_>) -> D + 'static,
+            F: Fn(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D + 'static,
         {
             use crate::operators::arrange::arrangement::Arrange;
             self.map(|k| (k, ()))
@@ -1237,12 +1242,16 @@ pub mod vec {
         ///      .assert_eq(z);
         /// });
         /// ```
-        pub fn join_core<Tr2,I,L> (self, stream2: Arranged<'scope, Tr2>, result: L) -> Collection<'scope, T,I::Item,<R as Multiply<Tr2::Diff>>::Output>
+        pub fn join_core<Tr2,I,L,R2> (self, stream2: Arranged<'scope, Tr2>, result: L) -> Collection<'scope, T,I::Item,<R as Multiply<R2>>::Output>
         where
-            Tr2: for<'a> crate::trace::TraceReader<Key<'a>=&'a K, Time=T>+Clone+'static,
-            R: Multiply<Tr2::Diff, Output: Semigroup+'static>,
+            Tr2: crate::trace::TraceReader<Batch: Navigable, Time=T>+Clone+'static,
+            for<'a> BatchCursor<Tr2>: Cursor<Key<'a>=&'a K>,
+            // Pin the cursor diff to a named param `R2`: a `Multiply` bound on a projection does not
+            // connect to its use-site (the solver normalizes the use but not the bound's subject).
+            BatchCursor<Tr2>: Cursor<Diff = R2, Time = T>,
+            R: Multiply<R2, Output: Semigroup+'static>,
             I: IntoIterator<Item: crate::Data>,
-            L: FnMut(&K,&V,Tr2::Val<'_>)->I+'static,
+            L: FnMut(&K,&V,BatchVal<'_, Tr2>)->I+'static,
         {
             self.arrange_by_key()
                 .join_core(stream2, result)
diff --git a/differential-dataflow/src/columnar/collection/operators.rs b/differential-dataflow/src/columnar/collection/operators.rs
index f3208b104..c5f00d540 100644
--- a/differential-dataflow/src/columnar/collection/operators.rs
+++ b/differential-dataflow/src/columnar/collection/operators.rs
@@ -160,7 +160,7 @@ where
 {
     use timely::dataflow::operators::generic::Operator;
     use timely::dataflow::channels::pact::Pipeline;
-    use crate::trace::{BatchReader, Cursor};
+    use crate::trace::{Navigable, Cursor};
     use crate::AsCollection;
 
     arranged.stream
diff --git a/differential-dataflow/src/columnar/trace/chunk.rs b/differential-dataflow/src/columnar/trace/chunk.rs
index 75c363bd3..f4cb6718f 100644
--- a/differential-dataflow/src/columnar/trace/chunk.rs
+++ b/differential-dataflow/src/columnar/trace/chunk.rs
@@ -40,8 +40,9 @@ use timely::progress::frontier::AntichainRef;
 
 use crate::consolidation::Consolidate;
 use crate::lattice::Lattice;
+use crate::trace::Navigable;
 use crate::trace::cursor::Cursor;
-use crate::trace::implementations::{BatchContainer, WithLayout};
+use crate::trace::implementations::{BatchContainer, Layout, WithLayout};
 
 use crate::columnar::layout::{ColumnarLayout, ColumnarUpdate, Coltainer};
 use crate::columnar::updates::{child_range, UpdatesBuilder, UpdatesTyped};
@@ -205,6 +206,18 @@ impl<U: ColumnarUpdate> WithLayout for ColChunkCursor<U> {
 impl<U: ColumnarUpdate> Cursor for ColChunkCursor<U> {
     type Storage = ColChunk<U>;
 
+    type KeyContainer = <ColumnarLayout<U> as Layout>::KeyContainer;
+    type Key<'a> = <<ColumnarLayout<U> as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
+    type ValContainer = <ColumnarLayout<U> as Layout>::ValContainer;
+    type Val<'a> = <<ColumnarLayout<U> as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
+    type ValOwn = <<ColumnarLayout<U> as Layout>::ValContainer as BatchContainer>::Owned;
+    type TimeContainer = <ColumnarLayout<U> as Layout>::TimeContainer;
+    type TimeGat<'a> = <<ColumnarLayout<U> as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
+    type Time = <<ColumnarLayout<U> as Layout>::TimeContainer as BatchContainer>::Owned;
+    type DiffContainer = <ColumnarLayout<U> as Layout>::DiffContainer;
+    type DiffGat<'a> = <<ColumnarLayout<U> as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
+    type Diff = <<ColumnarLayout<U> as Layout>::DiffContainer as BatchContainer>::Owned;
+
     fn key_valid(&self, s: &Self::Storage) -> bool { self.key_cursor < s.trie().view().keys.values.len() }
     fn val_valid(&self, s: &Self::Storage) -> bool {
         let view = s.trie().view();
@@ -279,19 +292,24 @@ fn consolidated<U: ColumnarUpdate>(merged: UpdatesTyped<U>) -> UpdatesTyped<U> {
     if merged.diffs.values.len() == merged.times.values.len() { merged } else { merged.filter_zero() }
 }
 
-impl<U: ColumnarUpdate> Chunk for ColChunk<U>
+impl<U: ColumnarUpdate> Navigable for ColChunk<U>
 where U::Time: 'static {
     type Cursor = ColChunkCursor<U>;
 
-    const TARGET: usize = TARGET;
-
     fn cursor(&self) -> Self::Cursor {
         ColChunkCursor { key_cursor: 0, val_cursor: 0, phantom: PhantomData }
     }
+}
+
+impl<U: ColumnarUpdate> Chunk for ColChunk<U>
+where U::Time: 'static {
+    type Time = <<ColumnarLayout<U> as Layout>::TimeContainer as BatchContainer>::Owned;
+
+    const TARGET: usize = TARGET;
 
     fn bounds(&self) -> (
-        (Self::Key<'_>, Self::Val<'_>, Self::TimeGat<'_>),
-        (Self::Key<'_>, Self::Val<'_>, Self::TimeGat<'_>),
+        (<Self::Cursor as Cursor>::Key<'_>, <Self::Cursor as Cursor>::Val<'_>, <Self::Cursor as Cursor>::TimeGat<'_>),
+        (<Self::Cursor as Cursor>::Key<'_>, <Self::Cursor as Cursor>::Val<'_>, <Self::Cursor as Cursor>::TimeGat<'_>),
     ) {
         match self {
             ColChunk::Resident(rc) => {
@@ -530,6 +548,7 @@ mod test {
     use super::{ColChunk, Chunk};
     use crate::columnar::updates::UpdatesTyped;
     use crate::trace::chunk::merge_chains;
+    use crate::trace::Navigable;
 
     type Upd = (u64, u64, u64, i64);
 
@@ -625,7 +644,7 @@ mod test {
     #[test]
     fn cursor_handles_straddle() {
         use crate::trace::cursor::Cursor;
-        use crate::trace::{BatchReader, Description};
+        use crate::trace::Description;
         use crate::trace::chunk::ChunkBatch;
         use timely::progress::Antichain;
 
@@ -656,7 +675,7 @@ mod test {
     // resumable merge -> advance -> settle pipeline end to end.
     #[test]
     fn batch_merger_resumable_matches_reference() {
-        use crate::trace::{BatchReader, Description, Merger};
+        use crate::trace::{Description, Merger};
         use crate::trace::chunk::{ChunkBatch, ChunkBatchMerger, is_graded};
         use crate::trace::cursor::Cursor;
         use crate::consolidation::consolidate_updates;
diff --git a/differential-dataflow/src/operators/arrange/agent.rs b/differential-dataflow/src/operators/arrange/agent.rs
index 41f964072..47fafaf62 100644
--- a/differential-dataflow/src/operators/arrange/agent.rs
+++ b/differential-dataflow/src/operators/arrange/agent.rs
@@ -35,13 +35,9 @@ pub struct TraceAgent<Tr: TraceReader> {
     logging: Option<crate::logging::Logger>,
 }
 
-use crate::trace::implementations::WithLayout;
-impl<Tr: TraceReader> WithLayout for TraceAgent<Tr> {
-    type Layout = Tr::Layout;
-}
-
 impl<Tr: TraceReader> TraceReader for TraceAgent<Tr> {
 
+    type Time = Tr::Time;
     type Batch = Tr::Batch;
 
     fn set_logical_compaction(&mut self, frontier: AntichainRef<Tr::Time>) {
diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index 06ba1f005..0e908cf6d 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -30,7 +30,7 @@ use timely::dataflow::operators::Capability;
 use crate::{Data, VecCollection, AsCollection};
 use crate::difference::Semigroup;
 use crate::lattice::Lattice;
-use crate::trace::{self, Trace, TraceReader, BatchReader, Batcher, Builder, Cursor};
+use crate::trace::{self, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchTimeGat, BatchVal, BatchValOwn};
 
 use trace::wrappers::enter::{TraceEnter, BatchEnter,};
 use trace::wrappers::enter_at::TraceEnter as TraceEnterAt;
@@ -99,8 +99,9 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
     /// not vary with the new timestamp coordinate.
     pub fn enter_at<'inner, TInner, F, P>(self, child: Scope<'inner, TInner>, logic: F, prior: P) -> Arranged<'inner, TraceEnterAt<Tr, TInner, F, P>>
     where
+        Tr::Batch: Navigable,
         TInner: Refines<Tr::Time>+Lattice+'static,
-        F: FnMut(Tr::Key<'_>, Tr::Val<'_>, Tr::TimeGat<'_>)->TInner+Clone+'static,
+        F: FnMut(BatchKey<'_, Tr>, BatchVal<'_, Tr>, BatchTimeGat<'_, Tr>)->TInner+Clone+'static,
         P: FnMut(&TInner)->Tr::Time+Clone+'static,
     {
         let logic1 = logic.clone();
@@ -138,9 +139,11 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
     /// The underlying `Stream<T, Vec<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) -> VecCollection<'scope, Tr::Time, D, Tr::Diff>
+    pub fn as_collection<D: Data, L>(self, mut logic: L) -> VecCollection<'scope, Tr::Time, D, BatchDiff<Tr>>
         where
-            L: FnMut(Tr::Key<'_>, Tr::Val<'_>) -> D+'static,
+            Tr::Batch: Navigable,
+            BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+            L: FnMut(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D+'static,
     {
         self.flat_map_ref(move |key, val| Some(logic(key,val)))
     }
@@ -154,11 +157,13 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
     /// The method takes `K` and `V` as generic arguments, in order to constrain the reference types to support
     /// cloning into owned types. If this bound does not work, the `as_collection` method allows arbitrary logic
     /// on the reference types.
-    pub fn as_vecs<K, V>(self) -> VecCollection<'scope, Tr::Time, (K, V), Tr::Diff>
+    pub fn as_vecs<K, V>(self) -> VecCollection<'scope, Tr::Time, (K, V), BatchDiff<Tr>>
     where
         K: crate::ExchangeData,
         V: crate::ExchangeData,
-        Tr: for<'a> TraceReader<Key<'a> = &'a K, Val<'a> = &'a V>,
+        Tr::Batch: Navigable,
+        BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+        for<'a> BatchCursor<Tr>: Cursor<Key<'a> = &'a K, Val<'a> = &'a V>,
     {
         self.flat_map_ref(move |key, val| [(key.clone(), val.clone())])
     }
@@ -167,10 +172,12 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
     ///
     /// 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) -> VecCollection<'scope, Tr::Time, I::Item, Tr::Diff>
+    pub fn flat_map_ref<I, L>(self, logic: L) -> VecCollection<'scope, Tr::Time, I::Item, BatchDiff<Tr>>
         where
+            Tr::Batch: Navigable,
+            BatchCursor<Tr>: Cursor<Time = Tr::Time>,
             I: IntoIterator<Item: Data>,
-            L: FnMut(Tr::Key<'_>, Tr::Val<'_>) -> I+'static,
+            L: FnMut(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> I+'static,
     {
         Self::flat_map_batches(self.stream, logic)
     }
@@ -182,10 +189,12 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
     ///
     /// 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<'scope, Tr::Time, Vec<Tr::Batch>>, mut logic: L) -> VecCollection<'scope, Tr::Time, I::Item, Tr::Diff>
+    pub fn flat_map_batches<I, L>(stream: Stream<'scope, Tr::Time, Vec<Tr::Batch>>, mut logic: L) -> VecCollection<'scope, Tr::Time, I::Item, BatchDiff<Tr>>
     where
+        Tr::Batch: Navigable,
+        BatchCursor<Tr>: Cursor<Time = Tr::Time>,
         I: IntoIterator<Item: Data>,
-        L: FnMut(Tr::Key<'_>, Tr::Val<'_>) -> I+'static,
+        L: FnMut(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> I+'static,
     {
         stream.unary(Pipeline, "AsCollection", move |_,_| move |input, output| {
             input.for_each(|time, data| {
@@ -197,7 +206,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
                         while let Some(val) = cursor.get_val(batch) {
                             for datum in logic(key, val) {
                                 cursor.map_times(batch, |time, diff| {
-                                    session.give((datum.clone(), Tr::owned_time(time), Tr::owned_diff(diff)));
+                                    session.give((datum.clone(), <BatchCursor<Tr> as Cursor>::owned_time(time), <BatchCursor<Tr> as Cursor>::owned_diff(diff)));
                                 });
                             }
                             cursor.step_val(batch);
@@ -214,18 +223,24 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
 
 use crate::difference::Multiply;
 // Direct join implementations.
-impl<'scope, Tr1: TraceReader+'static> Arranged<'scope, Tr1> {
+impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
     /// A convenience method to join and produce `VecCollection` output.
     ///
     /// Avoid this method, as it is likely to evolve into one without the `VecCollection` opinion.
-    pub fn join_core<Tr2,I,L>(self, other: Arranged<'scope, Tr2>, mut result: L) -> VecCollection<'scope, Tr1::Time,I::Item,<Tr1::Diff as Multiply<Tr2::Diff>>::Output>
+    pub fn join_core<Tr2,I,L,R1,R2>(self, other: Arranged<'scope, Tr2>, mut result: L) -> VecCollection<'scope, Tr1::Time,I::Item,<R1 as Multiply<R2>>::Output>
     where
-        Tr2: for<'a> TraceReader<Key<'a>=Tr1::Key<'a>,Time=Tr1::Time>+Clone+'static,
-        Tr1::Diff: Multiply<Tr2::Diff, Output: Semigroup+'static>,
+        Tr2: TraceReader<Batch: Navigable, Time=Tr1::Time>+Clone+'static,
+        // Pin the cursor diffs to named params `R1`/`R2`: a `Multiply` bound on a projection
+        // does not connect to its use-site (the solver normalizes the use but not the bound's
+        // subject), so we constrain plain params instead.
+        BatchCursor<Tr1>: Cursor<Diff = R1, Time = Tr1::Time>,
+        BatchCursor<Tr2>: Cursor<Diff = R2, Time = Tr1::Time>,
+        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>>,
+        R1: Multiply<R2, Output: Semigroup+'static> + Clone,
         I: IntoIterator<Item: Data>,
-        L: FnMut(Tr1::Key<'_>,Tr1::Val<'_>,Tr2::Val<'_>)->I+'static
+        L: FnMut(BatchKey<'_, Tr1>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>)->I+'static
     {
-        let mut result = move |k: Tr1::Key<'_>, v1: Tr1::Val<'_>, v2: Tr2::Val<'_>, t: Tr1::Time, r1: &Tr1::Diff, r2: &Tr2::Diff| {
+        let mut result = move |k: BatchKey<'_, Tr1>, v1: BatchVal<'_, Tr1>, v2: BatchVal<'_, Tr2>, t: Tr1::Time, r1: &R1, r2: &R2| {
             let r = (r1.clone()).multiply(r2);
             result(k, v1, v2).into_iter().map(move |d| (d, t.clone(), r.clone()))
         };
@@ -246,19 +261,16 @@ impl<'scope, Tr1: TraceReader+'static> Arranged<'scope, Tr1> {
 
 // Direct reduce implementations.
 use crate::difference::Abelian;
-impl<'scope, Tr1: TraceReader+'static> Arranged<'scope, Tr1> {
+impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
     /// A direct implementation of `ReduceCore::reduce_abelian`.
     pub fn reduce_abelian<L, Bu, Tr2, P>(self, name: &str, mut logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
     where
-        Tr2: for<'a> Trace<
-            Key<'a>= Tr1::Key<'a>,
-            ValOwn: Data,
-            Time=Tr1::Time,
-            Diff: Abelian,
-        >+'static,
+        Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
+        BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time, Diff: Abelian>,
         Bu: Builder<Time=Tr1::Time, Output = Tr2::Batch, Input: Default>,
-        L: FnMut(Tr1::Key<'_>, &[(Tr1::Val<'_>, Tr1::Diff)], &mut Vec<(Tr2::ValOwn, Tr2::Diff)>)+'static,
-        P: FnMut(&mut Bu::Input, Tr1::Key<'_>, &mut Vec<(Tr2::ValOwn, Tr2::Time, Tr2::Diff)>) + 'static,
+        L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
+        P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
     {
         self.reduce_core::<_,Bu,Tr2,_>(name, move |key, input, output, change| {
             if !input.is_empty() {
@@ -272,14 +284,12 @@ impl<'scope, Tr1: TraceReader+'static> Arranged<'scope, Tr1> {
     /// A direct implementation of `ReduceCore::reduce_core`.
     pub fn reduce_core<L, Bu, Tr2, P>(self, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
     where
-        Tr2: for<'a> Trace<
-            Key<'a>=Tr1::Key<'a>,
-            ValOwn: Data,
-            Time=Tr1::Time,
-        >+'static,
+        Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
+        BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
         Bu: Builder<Time=Tr1::Time, Output = Tr2::Batch, Input: Default>,
-        L: FnMut(Tr1::Key<'_>, &[(Tr1::Val<'_>, Tr1::Diff)], &mut Vec<(Tr2::ValOwn, Tr2::Diff)>, &mut Vec<(Tr2::ValOwn, Tr2::Diff)>)+'static,
-        P: FnMut(&mut Bu::Input, Tr1::Key<'_>, &mut Vec<(Tr2::ValOwn, Tr2::Time, Tr2::Diff)>) + 'static,
+        L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
+        P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
     {
         use crate::operators::reduce::reduce_trace;
         reduce_trace::<_,Bu,_,_,_>(self, name, logic, push)
diff --git a/differential-dataflow/src/operators/arrange/upsert.rs b/differential-dataflow/src/operators/arrange/upsert.rs
index 60fb51f26..33dcc4cc8 100644
--- a/differential-dataflow/src/operators/arrange/upsert.rs
+++ b/differential-dataflow/src/operators/arrange/upsert.rs
@@ -108,8 +108,7 @@ use timely::progress::{Antichain, Timestamp};
 use timely::dataflow::operators::Capability;
 
 use crate::operators::arrange::arrangement::Arranged;
-use crate::trace::{Builder, Description};
-use crate::trace::{self, Trace, TraceReader, Cursor};
+use crate::trace::{self, BatchCursor, BatchDiff, Builder, Cursor, Description, Navigable, Trace, TraceReader};
 use crate::{ExchangeData, Hashable};
 
 use crate::trace::implementations::containers::BatchContainer;
@@ -133,13 +132,13 @@ pub fn arrange_from_upsert<'scope, Bu, Tr, K, V>(
 where
     K: ExchangeData+Hashable+std::hash::Hash,
     V: ExchangeData,
-    Tr: for<'a> Trace<
+    Tr: Trace<Batch: Navigable, Time: TotalOrder+ExchangeData>+'static,
+    for<'a> BatchCursor<Tr>: Cursor<
         Key<'a> = &'a K,
         Val<'a> = &'a V,
-        Time: TotalOrder+ExchangeData,
         Diff=isize,
-    >+'static,
-    Bu: Builder<Time=Tr::Time, Input = Vec<((K, V), Tr::Time, Tr::Diff)>, Output = Tr::Batch>,
+    >,
+    Bu: Builder<Time=Tr::Time, Input = Vec<((K, V), Tr::Time, BatchDiff<Tr>)>, Output = Tr::Batch>,
 {
     let mut reader: Option<TraceAgent<Tr>> = None;
 
@@ -234,7 +233,7 @@ where
                                 // new stuff that we add.
                                 let (mut trace_cursor, trace_storage) = reader_local.cursor();
                                 let mut builder = Bu::new();
-                                let mut key_con = Tr::KeyContainer::with_capacity(1);
+                                let mut key_con = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
                                 for (key, mut list) in to_process {
 
                                     key_con.clear(); key_con.push_ref(&key);
@@ -248,7 +247,7 @@ where
                                         // Determine the prior value associated with the key.
                                         while let Some(val) = trace_cursor.get_val(&trace_storage) {
                                             let mut count = 0;
-                                            trace_cursor.map_times(&trace_storage, |_time, diff| count += Tr::owned_diff(diff));
+                                            trace_cursor.map_times(&trace_storage, |_time, diff| count += <BatchCursor<Tr> as Cursor>::owned_diff(diff));
                                             assert!(count == 0 || count == 1);
                                             if count == 1 {
                                                 assert!(prev_value.is_none());
diff --git a/differential-dataflow/src/operators/count.rs b/differential-dataflow/src/operators/count.rs
index b0de75c62..842e828de 100644
--- a/differential-dataflow/src/operators/count.rs
+++ b/differential-dataflow/src/operators/count.rs
@@ -11,7 +11,7 @@ use crate::difference::{IsZero, Semigroup};
 use crate::hashable::Hashable;
 use crate::collection::AsCollection;
 use crate::operators::arrange::Arranged;
-use crate::trace::{BatchReader, Cursor, TraceReader};
+use crate::trace::{BatchCursor, BatchDiff, BatchDiffGat, BatchReader, Cursor, Navigable, TraceReader};
 
 /// Extension trait for the `count` differential dataflow method.
 pub trait CountTotal<'scope, T: Timestamp + TotalOrder + Lattice, K: ExchangeData, R: Semigroup> : Sized {
@@ -52,17 +52,18 @@ where
     }
 }
 
-impl<'scope, K, Tr> CountTotal<'scope, Tr::Time, K, Tr::Diff> for Arranged<'scope, Tr>
+impl<'scope, K, Tr> CountTotal<'scope, Tr::Time, K, BatchDiff<Tr>> for Arranged<'scope, Tr>
 where
-    Tr: for<'a> TraceReader<
+    Tr: TraceReader<Batch: Navigable, Time: TotalOrder> + Clone + 'static,
+    for<'a> BatchCursor<Tr>: Cursor<
         Key<'a> = &'a K,
-        Val<'a>=&'a (),
-        Time: TotalOrder,
-        Diff: ExchangeData+Semigroup<Tr::DiffGat<'a>>
-    >+Clone+'static,
+        Val<'a> = &'a (),
+        Time = Tr::Time,
+        Diff: ExchangeData + Semigroup<BatchDiffGat<'a, Tr>>,
+    >,
     K: ExchangeData,
 {
-    fn count_total_core<R2: Semigroup + From<i8> + 'static>(self) -> VecCollection<'scope, Tr::Time, (K, Tr::Diff), R2> {
+    fn count_total_core<R2: Semigroup + From<i8> + 'static>(self) -> VecCollection<'scope, Tr::Time, (K, BatchDiff<Tr>), R2> {
 
         let mut trace = self.trace.clone();
 
@@ -74,7 +75,6 @@ where
 
             move |(input, _frontier), output| {
 
-                let mut batch_cursors = Vec::new();
                 let mut batch_storage = Vec::new();
 
                 // Downgrade previous upper limit to be current lower limit.
@@ -88,7 +88,6 @@ where
                     }
                     for batch in batches.drain(..) {
                         upper_limit.clone_from(batch.upper());  // NB: Assumes batches are in-order
-                        batch_cursors.push(batch.cursor());
                         batch_storage.push(batch);
                     }
                 });
@@ -97,18 +96,17 @@ where
 
                     let mut session = output.session(&capability);
 
-                    use crate::trace::cursor::CursorList;
-                    let mut batch_cursor = CursorList::new(batch_cursors, &batch_storage);
+                    let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage);
                     let (mut trace_cursor, trace_storage) = trace.cursor_through(lower_limit.borrow()).unwrap();
 
                     while let Some(key) = batch_cursor.get_key(&batch_storage) {
-                        let mut count: Option<Tr::Diff> = None;
+                        let mut count: Option<BatchDiff<Tr>> = None;
 
                         trace_cursor.seek_key(&trace_storage, key);
                         if trace_cursor.get_key(&trace_storage) == Some(key) {
                             trace_cursor.map_times(&trace_storage, |_, diff| {
                                 count.as_mut().map(|c| c.plus_equals(&diff));
-                                if count.is_none() { count = Some(Tr::owned_diff(diff)); }
+                                if count.is_none() { count = Some(<BatchCursor<Tr> as Cursor>::owned_diff(diff)); }
                             });
                         }
 
@@ -116,14 +114,14 @@ where
 
                             if let Some(count) = count.as_ref() {
                                 if !count.is_zero() {
-                                    session.give(((key.clone(), count.clone()), Tr::owned_time(time), R2::from(-1i8)));
+                                    session.give(((key.clone(), count.clone()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(-1i8)));
                                 }
                             }
                             count.as_mut().map(|c| c.plus_equals(&diff));
-                            if count.is_none() { count = Some(Tr::owned_diff(diff)); }
+                            if count.is_none() { count = Some(<BatchCursor<Tr> as Cursor>::owned_diff(diff)); }
                             if let Some(count) = count.as_ref() {
                                 if !count.is_zero() {
-                                    session.give(((key.clone(), count.clone()), Tr::owned_time(time), R2::from(1i8)));
+                                    session.give(((key.clone(), count.clone()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(1i8)));
                                 }
                             }
                         });
diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs
index 15829b605..f96cf0959 100644
--- a/differential-dataflow/src/operators/join.rs
+++ b/differential-dataflow/src/operators/join.rs
@@ -16,11 +16,9 @@ use timely::dataflow::operators::Capability;
 
 use crate::lattice::Lattice;
 use crate::operators::arrange::Arranged;
-use crate::trace::{BatchReader, Cursor};
+use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchReader, BatchVal, Cursor, Navigable, TraceReader};
 use crate::operators::ValueHistory;
 
-use crate::trace::TraceReader;
-
 /// The session passed to join closures.
 pub type JoinSession<'a, 'b, T, CB, CT> = Session<'a, 'b, T, EffortBuilder<CB>, CT>;
 
@@ -68,9 +66,11 @@ impl<CB: PushInto<D>, D> PushInto<D> for EffortBuilder<CB> {
 /// [`AsCollection`]: crate::collection::AsCollection
 pub fn join_traces<'scope, Tr1, Tr2, L, CB>(arranged1: Arranged<'scope, Tr1>, arranged2: Arranged<'scope, Tr2>, mut result: L) -> Stream<'scope, Tr1::Time, CB::Container>
 where
-    Tr1: TraceReader+'static,
-    Tr2: for<'a> TraceReader<Key<'a>=Tr1::Key<'a>, Time = Tr1::Time>+'static,
-    L: FnMut(Tr1::Key<'_>,Tr1::Val<'_>,Tr2::Val<'_>,Tr1::Time,&Tr1::Diff,&Tr2::Diff,&mut JoinSession<Tr1::Time, CB, Capability<Tr1::Time>>)+'static,
+    Tr1: TraceReader<Batch: Navigable>+'static,
+    Tr2: TraceReader<Batch: Navigable, Time = Tr1::Time>+'static,
+    BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+    for<'a> BatchCursor<Tr2>: Cursor<Key<'a>=BatchKey<'a, Tr1>, Time = Tr1::Time>,
+    L: FnMut(BatchKey<'_, Tr1>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>,Tr1::Time,&BatchDiff<Tr1>,&BatchDiff<Tr2>,&mut JoinSession<Tr1::Time, CB, Capability<Tr1::Time>>)+'static,
     CB: ContainerBuilder,
 {
     // Rename traces for symmetry from here on out.
diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index 592c360a2..e7498dff6 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -13,10 +13,8 @@ use timely::dataflow::operators::Operator;
 use timely::dataflow::channels::pact::Pipeline;
 
 use crate::operators::arrange::{Arranged, TraceAgent};
-use crate::trace::{BatchReader, Cursor, Trace, Builder, ExertionLogic, Description};
-use crate::trace::cursor::CursorList;
+use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchReader, BatchVal, BatchValOwn, Builder, Cursor, Description, ExertionLogic, Navigable, Trace, TraceReader};
 use crate::trace::implementations::containers::BatchContainer;
-use crate::trace::TraceReader;
 
 /// A key-wise reduction of values in an input trace.
 ///
@@ -33,11 +31,13 @@ use crate::trace::TraceReader;
 /// key's computation to another, and will likely introduce non-determinism.
 pub fn reduce_trace<'scope, Tr1, Bu, Tr2, L, P>(trace: Arranged<'scope, Tr1>, name: &str, mut logic: L, mut push: P) -> Arranged<'scope, TraceAgent<Tr2>>
 where
-    Tr1: TraceReader + 'static,
-    Tr2: for<'a> Trace<Key<'a>=Tr1::Key<'a>, ValOwn: Data, Time = Tr1::Time> + 'static,
+    Tr1: TraceReader<Batch: Navigable> + 'static,
+    Tr2: Trace<Batch: Navigable, Time = Tr1::Time> + 'static,
+    BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+    for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
     Bu: Builder<Time=Tr2::Time, Output = Tr2::Batch, Input: Default>,
-    L: FnMut(Tr1::Key<'_>, &[(Tr1::Val<'_>, Tr1::Diff)], &mut Vec<(Tr2::ValOwn,Tr2::Diff)>, &mut Vec<(Tr2::ValOwn, Tr2::Diff)>)+'static,
-    P: FnMut(&mut Bu::Input, Tr1::Key<'_>, &mut Vec<(Tr2::ValOwn, Tr2::Time, Tr2::Diff)>) + 'static,
+    L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
+    P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
 {
     let mut result_trace = None;
 
@@ -67,10 +67,10 @@ where
 
             // Our implementation maintains a list of outstanding `(key, time)` synthetic interesting times,
             // sorted by (key, time), as well as capabilities for the lower envelope of the times.
-            let mut pending_keys = Tr1::KeyContainer::with_capacity(0);
-            let mut pending_time = Tr1::TimeContainer::with_capacity(0);
-            let mut next_pending_keys = Tr1::KeyContainer::with_capacity(0);
-            let mut next_pending_time = Tr1::TimeContainer::with_capacity(0);
+            let mut pending_keys = <BatchCursor<Tr1> as Cursor>::KeyContainer::with_capacity(0);
+            let mut pending_time = <BatchCursor<Tr1> as Cursor>::TimeContainer::with_capacity(0);
+            let mut next_pending_keys = <BatchCursor<Tr1> as Cursor>::KeyContainer::with_capacity(0);
+            let mut next_pending_time = <BatchCursor<Tr1> as Cursor>::TimeContainer::with_capacity(0);
             let mut capabilities = timely::dataflow::operators::CapabilitySet::<Tr1::Time>::default();
 
             // buffers and logic for computing per-key interesting times "efficiently".
@@ -94,7 +94,6 @@ where
                 // may not be seen in its input. The standard example is that updates at `(0, 1)` and `(1, 0)`
                 // may result in outputs at `(1, 1)` as well, even with no input at that time.
 
-                let mut batch_cursors = Vec::new();
                 let mut batch_storage = Vec::new();
 
                 // Downgrade previous upper limit to be current lower limit.
@@ -106,7 +105,6 @@ where
                     capabilities.insert(capability.retain(0));
                     for batch in batches.drain(..) {
                         upper_limit.clone_from(batch.upper());
-                        batch_cursors.push(batch.cursor());
                         batch_storage.push(batch);
                     }
                 });
@@ -129,12 +127,12 @@ where
                         // cursors for navigating input and output traces.
                         let (mut source_cursor, ref source_storage) = source_trace.cursor_through(lower_limit.borrow()).expect("failed to acquire source cursor");
                         let (mut output_cursor, ref output_storage) = output_reader.cursor_through(lower_limit.borrow()).expect("failed to acquire output cursor");
-                        let (mut batch_cursor, ref batch_storage) = (CursorList::new(batch_cursors, &batch_storage), batch_storage);
+                        let (mut batch_cursor, ref batch_storage) = crate::trace::cursor::cursor_list(batch_storage);
 
                         // Prepare an output buffer and builder for each capability.
                         // 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::<(Tr1::Time, Vec<(Tr2::ValOwn, Tr1::Time, Tr2::Diff)>)>::new();
+                        let mut buffers = Vec::<(Tr1::Time, Vec<(BatchValOwn<Tr2>, Tr1::Time, BatchDiff<Tr2>)>)>::new();
                         let mut builders = Vec::new();
                         for cap in capabilities.iter() {
                             buffers.push((cap.time().clone(), Vec::new()));
@@ -168,7 +166,7 @@ where
                             let prior_pos = pending_pos;
                             interesting_times.clear();
                             while pending_keys.get(pending_pos) == Some(key) {
-                                let owned_time = Tr1::owned_time(pending_time.index(pending_pos));
+                                let owned_time = <BatchCursor<Tr1> as Cursor>::owned_time(pending_time.index(pending_pos));
                                 if !upper_limit.less_equal(&owned_time) { interesting_times.push(owned_time); }
                                 pending_pos += 1;
                             }
@@ -198,7 +196,7 @@ where
                                 // Merge novel pending times with any prior pending times we did not process.
                                 // TODO: This could be a merge, not a sort_dedup, because both lists should be sorted.
                                 for pos in prior_pos .. pending_pos {
-                                    let owned_time = Tr1::owned_time(pending_time.index(pos));
+                                    let owned_time = <BatchCursor<Tr1> as Cursor>::owned_time(pending_time.index(pos));
                                     if upper_limit.less_equal(&owned_time) { new_interesting_times.push(owned_time); }
                                 }
                                 sort_dedup(&mut new_interesting_times);
@@ -272,7 +270,7 @@ where
                         let mut frontier = Antichain::<Tr1::Time>::new();
                         let mut owned_time = Tr1::Time::minimum();
                         for pos in 0 .. pending_time.len() {
-                            Tr1::clone_time_onto(pending_time.index(pos), &mut owned_time);
+                            <BatchCursor<Tr1> as Cursor>::clone_time_onto(pending_time.index(pos), &mut owned_time);
                             frontier.insert_ref(&owned_time);
                         }
                         capabilities.downgrade(frontier);
diff --git a/differential-dataflow/src/operators/threshold.rs b/differential-dataflow/src/operators/threshold.rs
index f712ae7ba..2909a0c53 100644
--- a/differential-dataflow/src/operators/threshold.rs
+++ b/differential-dataflow/src/operators/threshold.rs
@@ -14,7 +14,7 @@ use crate::difference::{Semigroup, Abelian};
 use crate::hashable::Hashable;
 use crate::collection::AsCollection;
 use crate::operators::arrange::Arranged;
-use crate::trace::{BatchReader, Cursor, TraceReader};
+use crate::trace::{BatchCursor, BatchDiff, BatchDiffGat, BatchKey, BatchReader, Cursor, Navigable, TraceReader};
 
 /// Extension trait for the `distinct` differential dataflow method.
 pub trait ThresholdTotal<'scope, T: Timestamp + TotalOrder + Lattice, K: ExchangeData, R: ExchangeData+Semigroup> : Sized {
@@ -98,20 +98,21 @@ where
     }
 }
 
-impl<'scope, K, Tr> ThresholdTotal<'scope, Tr::Time, K, Tr::Diff> for Arranged<'scope, Tr>
+impl<'scope, K, Tr> ThresholdTotal<'scope, Tr::Time, K, BatchDiff<Tr>> for Arranged<'scope, Tr>
 where
-    Tr: for<'a> TraceReader<
+    Tr: TraceReader<Batch: Navigable, Time: TotalOrder> + Clone + 'static,
+    for<'a> BatchCursor<Tr>: Cursor<
         Key<'a>=&'a K,
         Val<'a>=&'a (),
-        Time: TotalOrder,
-        Diff : ExchangeData + Semigroup<Tr::DiffGat<'a>>,
-    >+Clone+'static,
+        Time = Tr::Time,
+        Diff : ExchangeData + Semigroup<BatchDiffGat<'a, Tr>>,
+    >,
     K: ExchangeData,
 {
     fn threshold_semigroup<R2, F>(self, mut thresh: F) -> VecCollection<'scope, Tr::Time, K, R2>
     where
         R2: Semigroup+'static,
-        F: for<'a> FnMut(Tr::Key<'a>,&Tr::Diff,Option<&Tr::Diff>)->Option<R2>+'static,
+        F: for<'a> FnMut(BatchKey<'a, Tr>,&BatchDiff<Tr>,Option<&BatchDiff<Tr>>)->Option<R2>+'static,
     {
 
         let mut trace = self.trace.clone();
@@ -124,7 +125,6 @@ where
 
             move |(input, _frontier), output| {
 
-                let mut batch_cursors = Vec::new();
                 let mut batch_storage = Vec::new();
 
                 // Downgrde previous upper limit to be current lower limit.
@@ -138,7 +138,6 @@ where
                     }
                     for batch in batches.drain(..) {
                         upper_limit.clone_from(batch.upper());  // NB: Assumes batches are in-order
-                        batch_cursors.push(batch.cursor());
                         batch_storage.push(batch);
                     }
                 });
@@ -147,19 +146,18 @@ where
 
                     let mut session = output.session(&capability);
 
-                    use crate::trace::cursor::CursorList;
-                    let mut batch_cursor = CursorList::new(batch_cursors, &batch_storage);
+                    let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage);
                     let (mut trace_cursor, trace_storage) = trace.cursor_through(lower_limit.borrow()).unwrap();
 
                     while let Some(key) = batch_cursor.get_key(&batch_storage) {
-                        let mut count: Option<Tr::Diff> = None;
+                        let mut count: Option<BatchDiff<Tr>> = None;
 
                         // Compute the multiplicity of this key before the current batch.
                         trace_cursor.seek_key(&trace_storage, key);
                         if trace_cursor.get_key(&trace_storage) == Some(key) {
                             trace_cursor.map_times(&trace_storage, |_, diff| {
                                 count.as_mut().map(|c| c.plus_equals(&diff));
-                                if count.is_none() { count = Some(Tr::owned_diff(diff)); }
+                                if count.is_none() { count = Some(<BatchCursor<Tr> as Cursor>::owned_diff(diff)); }
                             });
                         }
 
@@ -174,7 +172,7 @@ where
                                     temp.plus_equals(&diff);
                                     thresh(key, &temp, Some(old))
                                 },
-                                None => { thresh(key, &Tr::owned_diff(diff), None) },
+                                None => { thresh(key, &<BatchCursor<Tr> as Cursor>::owned_diff(diff), None) },
                             };
 
                             // Either add or assign `diff` to `count`.
@@ -182,12 +180,12 @@ where
                                 count.plus_equals(&diff);
                             }
                             else {
-                                count = Some(Tr::owned_diff(diff));
+                                count = Some(<BatchCursor<Tr> as Cursor>::owned_diff(diff));
                             }
 
                             if let Some(difference) = difference {
                                 if !difference.is_zero() {
-                                    session.give((key.clone(), Tr::owned_time(time), difference));
+                                    session.give((key.clone(), <BatchCursor<Tr> as Cursor>::owned_time(time), difference));
                                 }
                             }
                         });
diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs
index 730f208f6..940a28764 100644
--- a/differential-dataflow/src/trace/chunk/mod.rs
+++ b/differential-dataflow/src/trace/chunk/mod.rs
@@ -55,9 +55,9 @@ use std::collections::VecDeque;
 use timely::progress::Antichain;
 use timely::progress::frontier::AntichainRef;
 use crate::lattice::Lattice;
-use crate::trace::{Batch, BatchReader, Description};
+use crate::trace::{Batch, BatchReader, Description, Navigable};
 use crate::trace::cursor::Cursor;
-use crate::trace::implementations::{BatchContainer, Layout, LayoutExt, WithLayout};
+use crate::trace::implementations::BatchContainer;
 
 pub mod vec;
 
@@ -70,36 +70,20 @@ pub mod vec;
 /// The "data" operations transform lists of chunks, are expected to do roughly
 /// "one chunk's worth" of work at a time; they can afford to compress and page.
 /// The "metadata" operations provide chunk information, and should be lightweight.
-pub trait Chunk: Sized + Clone + LayoutExt {
+pub trait Chunk: Navigable<Cursor: Cursor<Time = Self::Time>> + Sized + Clone {
+    /// The timestamp type of the chunk's updates.
+    ///
+    /// Key/val/diff opinions live on the chunk's [`Navigable::Cursor`]; the chunk itself only needs
+    /// time, to bound its interval and participate in advancement and compaction.
+    type Time: Lattice + timely::progress::Timestamp;
 
     /// The intended maximum chunk size.
     const TARGET: usize;
 
-    /// A cursor navigating this chunk's contents.
-    type Cursor:
-        Cursor<Storage = Self> +
-        WithLayout<Layout = Self::Layout> +
-        for<'a> LayoutExt<
-            Key<'a> = Self::Key<'a>,
-            Val<'a> = Self::Val<'a>,
-            ValOwn = Self::ValOwn,
-            Time = Self::Time,
-            TimeGat<'a> = Self::TimeGat<'a>,
-            Diff = Self::Diff,
-            DiffGat<'a> = Self::DiffGat<'a>,
-            KeyContainer = Self::KeyContainer,
-            ValContainer = Self::ValContainer,
-            TimeContainer = Self::TimeContainer,
-            DiffContainer = Self::DiffContainer,
-        >;
-
-    /// Acquire a cursor over this chunk.
-    fn cursor(&self) -> Self::Cursor;
-
     /// The first and last `(key, val, time)` triples in the chunk.
     fn bounds(&self) -> (
-        (Self::Key<'_>, Self::Val<'_>, Self::TimeGat<'_>),
-        (Self::Key<'_>, Self::Val<'_>, Self::TimeGat<'_>),
+        (<Self::Cursor as Cursor>::Key<'_>, <Self::Cursor as Cursor>::Val<'_>, <Self::Cursor as Cursor>::TimeGat<'_>),
+        (<Self::Cursor as Cursor>::Key<'_>, <Self::Cursor as Cursor>::Val<'_>, <Self::Cursor as Cursor>::TimeGat<'_>),
     );
 
     /// The number of updates in the chunk.
@@ -231,8 +215,8 @@ where
     }
 }
 
-type KeyCon<C> = <<C as WithLayout>::Layout as Layout>::KeyContainer;
-type ValCon<C> = <<C as WithLayout>::Layout as Layout>::ValContainer;
+type KeyCon<C> = <<C as Navigable>::Cursor as Cursor>::KeyContainer;
+type ValCon<C> = <<C as Navigable>::Cursor as Cursor>::ValContainer;
 
 /// A batch is a [`Chunk`] sequence plus a [`Description`].
 ///
@@ -269,15 +253,15 @@ impl<C: Chunk> ChunkBatch<C> {
     }
 }
 
-impl<C: Chunk> WithLayout for ChunkBatch<C> {
-    type Layout = C::Layout;
-}
-
-impl<C: Chunk> BatchReader for ChunkBatch<C> {
+impl<C: Chunk> crate::trace::Navigable for ChunkBatch<C> {
     type Cursor = ChunkBatchCursor<C>;
     fn cursor(&self) -> Self::Cursor {
         ChunkBatchCursor { key_chunk: 0, chunk: 0, inner: self.chunks.first().map(C::cursor) }
     }
+}
+
+impl<C: Chunk> BatchReader for ChunkBatch<C> {
+    type Time = C::Time;
     fn len(&self) -> usize { self.chunks.iter().map(C::len).sum() }
     fn description(&self) -> &Description<Self::Time> { &self.description }
 }
@@ -330,10 +314,6 @@ pub struct ChunkBatchCursor<C: Chunk> {
     inner: Option<C::Cursor>,
 }
 
-impl<C: Chunk> WithLayout for ChunkBatchCursor<C> {
-    type Layout = C::Layout;
-}
-
 impl<C: Chunk> ChunkBatchCursor<C> {
     /// Move the active chunk to `c`, opening a fresh inner cursor at its start.
     fn goto(&mut self, c: usize, storage: &ChunkBatch<C>) {
@@ -345,6 +325,18 @@ impl<C: Chunk> ChunkBatchCursor<C> {
 impl<C: Chunk> Cursor for ChunkBatchCursor<C> {
     type Storage = ChunkBatch<C>;
 
+    type KeyContainer = <C::Cursor as Cursor>::KeyContainer;
+    type Key<'a> = <C::Cursor as Cursor>::Key<'a>;
+    type ValContainer = <C::Cursor as Cursor>::ValContainer;
+    type Val<'a> = <C::Cursor as Cursor>::Val<'a>;
+    type ValOwn = <C::Cursor as Cursor>::ValOwn;
+    type TimeContainer = <C::Cursor as Cursor>::TimeContainer;
+    type TimeGat<'a> = <C::Cursor as Cursor>::TimeGat<'a>;
+    type Time = <C::Cursor as Cursor>::Time;
+    type DiffContainer = <C::Cursor as Cursor>::DiffContainer;
+    type DiffGat<'a> = <C::Cursor as Cursor>::DiffGat<'a>;
+    type Diff = <C::Cursor as Cursor>::Diff;
+
     fn key_valid(&self, s: &Self::Storage) -> bool { self.chunk < s.chunks.len() && self.inner.as_ref().is_some_and(|i| i.key_valid(&s.chunks[self.chunk])) }
     fn val_valid(&self, s: &Self::Storage) -> bool { self.chunk < s.chunks.len() && self.inner.as_ref().is_some_and(|i| i.val_valid(&s.chunks[self.chunk])) }
     fn key<'a>(&self, s: &'a Self::Storage) -> Self::Key<'a> { self.inner.as_ref().unwrap().key(&s.chunks[self.chunk]) }
diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs
index 15b7a390d..71170835b 100644
--- a/differential-dataflow/src/trace/chunk/vec.rs
+++ b/differential-dataflow/src/trace/chunk/vec.rs
@@ -25,8 +25,9 @@ use timely::progress::frontier::AntichainRef;
 use crate::consolidation::Consolidate;
 use crate::difference::Semigroup;
 use crate::lattice::Lattice;
+use crate::trace::Navigable;
 use crate::trace::cursor::Cursor;
-use crate::trace::implementations::{Vector, WithLayout};
+use crate::trace::implementations::{BatchContainer, Layout, Vector, WithLayout};
 
 use super::Chunk;
 
@@ -120,6 +121,18 @@ impl<K, V, T, R> Cursor for VecChunkCursor<K, V, T, R>
 where K: Ord+Clone+'static, V: Ord+Clone+'static, T: Lattice+Timestamp, R: Ord+Semigroup+'static {
     type Storage = VecChunk<K, V, T, R>;
 
+    type KeyContainer = <Vector<((K, V), T, R)> as Layout>::KeyContainer;
+    type Key<'a> = <<Vector<((K, V), T, R)> as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
+    type ValContainer = <Vector<((K, V), T, R)> as Layout>::ValContainer;
+    type Val<'a> = <<Vector<((K, V), T, R)> as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
+    type ValOwn = <<Vector<((K, V), T, R)> as Layout>::ValContainer as BatchContainer>::Owned;
+    type TimeContainer = <Vector<((K, V), T, R)> as Layout>::TimeContainer;
+    type TimeGat<'a> = <<Vector<((K, V), T, R)> as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
+    type Time = <<Vector<((K, V), T, R)> as Layout>::TimeContainer as BatchContainer>::Owned;
+    type DiffContainer = <Vector<((K, V), T, R)> as Layout>::DiffContainer;
+    type DiffGat<'a> = <<Vector<((K, V), T, R)> as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
+    type Diff = <<Vector<((K, V), T, R)> as Layout>::DiffContainer as BatchContainer>::Owned;
+
     fn key_valid(&self, s: &Self::Storage) -> bool { self.key_pos < s.0.len() }
     fn val_valid(&self, s: &Self::Storage) -> bool {
         self.key_pos < s.0.len() && self.val_pos < s.0.len() && s.0[self.val_pos].0.0 == s.0[self.key_pos].0.0
@@ -177,15 +190,20 @@ fn take<K: Clone, V: Clone, T: Clone, R: Clone>(chunk: VecChunk<K, V, T, R>) ->
     Rc::try_unwrap(chunk.0).unwrap_or_else(|rc| (*rc).clone())
 }
 
-impl<K, V, T, R> Chunk for VecChunk<K, V, T, R>
+impl<K, V, T, R> Navigable for VecChunk<K, V, T, R>
 where K: Ord+Clone+'static, V: Ord+Clone+'static, T: Lattice+Timestamp, R: Ord+Semigroup+'static {
     type Cursor = VecChunkCursor<K, V, T, R>;
 
-    const TARGET: usize = TARGET;
-
     fn cursor(&self) -> Self::Cursor {
         VecChunkCursor { key_pos: 0, val_pos: 0, phantom: PhantomData }
     }
+}
+
+impl<K, V, T, R> Chunk for VecChunk<K, V, T, R>
+where K: Ord+Clone+'static, V: Ord+Clone+'static, T: Lattice+Timestamp, R: Ord+Semigroup+'static {
+    type Time = <<Vector<((K, V), T, R)> as Layout>::TimeContainer as BatchContainer>::Owned;
+
+    const TARGET: usize = TARGET;
 
     fn bounds(&self) -> ((&K, &V, &T), (&K, &V, &T)) {
         let s = &self.0[..];
@@ -369,6 +387,7 @@ mod test {
     use std::collections::VecDeque;
     use super::{Chunk, VecChunk};
     use crate::trace::chunk::merge_chains;
+    use crate::trace::Navigable;
     use std::rc::Rc;
 
     fn chunk(updates: Vec<((u64, u64), u64, i64)>) -> VecChunk<u64, u64, u64, i64> {
@@ -574,7 +593,7 @@ mod test {
     // resumable merge→advance→settle pipeline and the grade-at-yield invariant.
     #[test]
     fn batch_merger_resumable_matches_reference() {
-        use crate::trace::{BatchReader, Description, Merger};
+        use crate::trace::{Description, Merger};
         use crate::trace::chunk::{ChunkBatch, ChunkBatchMerger, is_graded};
         use crate::trace::cursor::Cursor;
         use crate::consolidation::consolidate_updates;
@@ -686,7 +705,7 @@ mod test {
     #[test]
     fn cursor_handles_straddle() {
         use crate::trace::cursor::Cursor;
-        use crate::trace::{BatchReader, Description};
+        use crate::trace::Description;
         use crate::trace::chunk::ChunkBatch;
         use timely::progress::Antichain;
 
diff --git a/differential-dataflow/src/trace/cursor/cursor_list.rs b/differential-dataflow/src/trace/cursor/cursor_list.rs
index 6d2d5b57b..b5345e6bd 100644
--- a/differential-dataflow/src/trace/cursor/cursor_list.rs
+++ b/differential-dataflow/src/trace/cursor/cursor_list.rs
@@ -94,15 +94,22 @@ impl<C: Cursor> CursorList<C> {
     }
 }
 
-use crate::trace::implementations::WithLayout;
-impl<C: Cursor> WithLayout for CursorList<C> {
-    type Layout = C::Layout;
-}
-
 impl<C: Cursor> Cursor for CursorList<C> {
 
     type Storage = Vec<C::Storage>;
 
+    type Key<'a> = C::Key<'a>;
+    type ValOwn = C::ValOwn;
+    type Val<'a> = C::Val<'a>;
+    type Time = C::Time;
+    type TimeGat<'a> = C::TimeGat<'a>;
+    type Diff = C::Diff;
+    type DiffGat<'a> = C::DiffGat<'a>;
+    type KeyContainer = C::KeyContainer;
+    type ValContainer = C::ValContainer;
+    type TimeContainer = C::TimeContainer;
+    type DiffContainer = C::DiffContainer;
+
     // validation methods
     #[inline]
     fn key_valid(&self, _storage: &Vec<C::Storage>) -> bool { !self.min_key.is_empty() }
diff --git a/differential-dataflow/src/trace/cursor/mod.rs b/differential-dataflow/src/trace/cursor/mod.rs
index 71ebc0966..c8d068cfa 100644
--- a/differential-dataflow/src/trace/cursor/mod.rs
+++ b/differential-dataflow/src/trace/cursor/mod.rs
@@ -9,14 +9,97 @@ pub mod cursor_list;
 
 pub use self::cursor_list::CursorList;
 
-use crate::trace::implementations::LayoutExt;
+use timely::progress::Timestamp;
+use crate::lattice::Lattice;
+use crate::difference::Semigroup;
+use crate::trace::implementations::containers::BatchContainer;
+
+/// A batch type that has a cursor for navigation.
+///
+/// This is the entry point for accessing batch data through cursors, and the place that opinions
+/// about keys and values are introduced (via the `Cursor` associated type). Cut-and-merge assembly
+/// is the trace's concern: [`TraceReader::cursor_storage`](crate::trace::TraceReader::cursor_storage)
+/// selects the batches and the defaulted
+/// [`TraceReader::cursor_through`](crate::trace::TraceReader::cursor_through) builds a [`CursorList`]
+/// over their per-batch cursors.
+pub trait Navigable {
+
+    /// The cursor type.
+    ///
+    /// The cursor carries the layout opinions (keys, values, containers); a `Navigable` type only
+    /// promises that it can produce one.
+    type Cursor: Cursor<Storage = Self>;
+
+    /// Acquire a cursor suitable for the instance.
+    fn cursor(&self) -> Self::Cursor;
+}
+
+/// The cursor type for a trace's batches.
+pub type BatchCursor<Tr> = <<Tr as crate::trace::TraceReader>::Batch as Navigable>::Cursor;
+
+/// The borrowed key type of a trace's batch cursor.
+pub type BatchKey<'a, Tr> = <BatchCursor<Tr> as Cursor>::Key<'a>;
+/// The borrowed val type of a trace's batch cursor.
+pub type BatchVal<'a, Tr> = <BatchCursor<Tr> as Cursor>::Val<'a>;
+/// The owned val type of a trace's batch cursor.
+pub type BatchValOwn<Tr> = <BatchCursor<Tr> as Cursor>::ValOwn;
+/// The borrowed diff type of a trace's batch cursor.
+pub type BatchDiffGat<'a, Tr> = <BatchCursor<Tr> as Cursor>::DiffGat<'a>;
+/// The owned diff type of a trace's batch cursor.
+pub type BatchDiff<Tr> = <BatchCursor<Tr> as Cursor>::Diff;
+/// The borrowed time type of a trace's batch cursor.
+pub type BatchTimeGat<'a, Tr> = <BatchCursor<Tr> as Cursor>::TimeGat<'a>;
+
+/// Assembles a merged cursor over a sequence of batches.
+///
+/// The batches become the cursor's storage and are returned alongside the cursor; they must be kept
+/// alive and handed to the cursor's navigation methods. This is the shared assembly behind
+/// `TraceReader::cursor_through` and the per-round input cursors in `reduce` / `count` / `threshold`.
+pub fn cursor_list<B: crate::trace::BatchReader + Navigable>(batches: Vec<B>) -> (CursorList<B::Cursor>, Vec<B>) {
+    let cursors = batches.iter().map(|batch| batch.cursor()).collect::<Vec<_>>();
+    let cursor = CursorList::new(cursors, &batches);
+    (cursor, batches)
+}
 
 /// A cursor for navigating ordered `(key, val, time, diff)` updates.
-pub trait Cursor : LayoutExt {
+pub trait Cursor {
 
     /// Storage required by the cursor.
     type Storage;
 
+    /// Alias for a borrowed key.
+    type Key<'a>: Copy + Ord;
+    /// Alias for an owned val.
+    type ValOwn: Clone + Ord;
+    /// Alias for a borrowed val.
+    type Val<'a>: Copy + Ord;
+    /// Alias for an owned time.
+    type Time: Lattice + Timestamp;
+    /// Alias for a borrowed time.
+    type TimeGat<'a>: Copy + Ord;
+    /// Alias for an owned diff.
+    type Diff: Semigroup + 'static;
+    /// Alias for a borrowed diff.
+    type DiffGat<'a>: Copy + Ord;
+
+    /// Container for update keys.
+    type KeyContainer: for<'a> BatchContainer<ReadItem<'a> = Self::Key<'a>>;
+    /// Container for update vals.
+    type ValContainer: for<'a> BatchContainer<ReadItem<'a> = Self::Val<'a>, Owned = Self::ValOwn>;
+    /// Container for times.
+    type TimeContainer: for<'a> BatchContainer<ReadItem<'a> = Self::TimeGat<'a>, Owned = Self::Time>;
+    /// Container for diffs.
+    type DiffContainer: for<'a> BatchContainer<ReadItem<'a> = Self::DiffGat<'a>, Owned = Self::Diff>;
+
+    /// Construct an owned val from a reference.
+    #[inline(always)] fn owned_val(val: Self::Val<'_>) -> Self::ValOwn { <Self::ValContainer as BatchContainer>::into_owned(val) }
+    /// Construct an owned time from a reference.
+    #[inline(always)] fn owned_time(time: Self::TimeGat<'_>) -> Self::Time { <Self::TimeContainer as BatchContainer>::into_owned(time) }
+    /// Construct an owned diff from a reference.
+    #[inline(always)] fn owned_diff(diff: Self::DiffGat<'_>) -> Self::Diff { <Self::DiffContainer as BatchContainer>::into_owned(diff) }
+    /// Clones a reference time onto an owned time.
+    #[inline(always)] fn clone_time_onto(time: Self::TimeGat<'_>, onto: &mut Self::Time) { <Self::TimeContainer as BatchContainer>::clone_onto(time, onto) }
+
     /// Indicates if the current key is valid.
     ///
     /// A value of `false` indicates that the cursor has exhausted all keys.
diff --git a/differential-dataflow/src/trace/implementations/mod.rs b/differential-dataflow/src/trace/implementations/mod.rs
index 46bb27e90..9b69cb12d 100644
--- a/differential-dataflow/src/trace/implementations/mod.rs
+++ b/differential-dataflow/src/trace/implementations/mod.rs
@@ -107,63 +107,6 @@ pub trait WithLayout {
     type Layout: Layout;
 }
 
-/// Automatically implemented trait for types with layouts.
-pub trait LayoutExt : WithLayout<Layout: Layout<KeyContainer = Self::KeyContainer, ValContainer = Self::ValContainer, TimeContainer = Self::TimeContainer, DiffContainer = Self::DiffContainer>> {
-    /// Alias for an borrowed key of a layout.
-    type Key<'a>: Copy + Ord;
-    /// Alias for an owned val of a layout.
-    type ValOwn: Clone + Ord;
-    /// Alias for an borrowed val of a layout.
-    type Val<'a>: Copy + Ord;
-    /// Alias for an owned time of a layout.
-    type Time: Lattice + timely::progress::Timestamp;
-    /// Alias for an borrowed time of a layout.
-    type TimeGat<'a>: Copy + Ord;
-    /// Alias for an owned diff of a layout.
-    type Diff: Semigroup + 'static;
-    /// Alias for an borrowed diff of a layout.
-    type DiffGat<'a>: Copy + Ord;
-
-    /// Container for update keys.
-    type KeyContainer: for<'a> BatchContainer<ReadItem<'a> = Self::Key<'a>>;
-    /// Container for update vals.
-    type ValContainer: for<'a> BatchContainer<ReadItem<'a> = Self::Val<'a>, Owned = Self::ValOwn>;
-    /// Container for times.
-    type TimeContainer: for<'a> BatchContainer<ReadItem<'a> = Self::TimeGat<'a>, Owned = Self::Time>;
-    /// Container for diffs.
-    type DiffContainer: for<'a> BatchContainer<ReadItem<'a> = Self::DiffGat<'a>, Owned = Self::Diff>;
-
-    /// Construct an owned val from a reference.
-    fn owned_val(val: Self::Val<'_>) -> Self::ValOwn;
-    /// Construct an owned time from a reference.
-    fn owned_time(time: Self::TimeGat<'_>) -> Self::Time;
-    /// Construct an owned diff from a reference.
-    fn owned_diff(diff: Self::DiffGat<'_>) -> Self::Diff;
-
-    /// Clones a reference time onto an owned time.
-    fn clone_time_onto(time: Self::TimeGat<'_>, onto: &mut Self::Time);
-}
-
-impl<L: WithLayout> LayoutExt for L {
-    type Key<'a> = <<L::Layout as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
-    type ValOwn = <<L::Layout as Layout>::ValContainer as BatchContainer>::Owned;
-    type Val<'a> = <<L::Layout as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
-    type Time = <<L::Layout as Layout>::TimeContainer as BatchContainer>::Owned;
-    type TimeGat<'a> = <<L::Layout as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
-    type Diff = <<L::Layout as Layout>::DiffContainer as BatchContainer>::Owned;
-    type DiffGat<'a> = <<L::Layout as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
-
-    type KeyContainer = <L::Layout as Layout>::KeyContainer;
-    type ValContainer = <L::Layout as Layout>::ValContainer;
-    type TimeContainer = <L::Layout as Layout>::TimeContainer;
-    type DiffContainer = <L::Layout as Layout>::DiffContainer;
-
-    #[inline(always)] fn owned_val(val: Self::Val<'_>) -> Self::ValOwn { <Self::Layout as Layout>::ValContainer::into_owned(val) }
-    #[inline(always)] fn owned_time(time: Self::TimeGat<'_>) -> Self::Time { <Self::Layout as Layout>::TimeContainer::into_owned(time) }
-    #[inline(always)] fn owned_diff(diff: Self::DiffGat<'_>) -> Self::Diff { <Self::Layout as Layout>::DiffContainer::into_owned(diff) }
-    #[inline(always)] fn clone_time_onto(time: Self::TimeGat<'_>, onto: &mut Self::Time) { <Self::Layout as Layout>::TimeContainer::clone_onto(time, onto) }
-
-}
 
 // An easy way to provide an explicit layout: as a 5-tuple.
 // Valuable when one wants to perform layout surgery.
diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs
index 309ef6694..cf0c22a5f 100644
--- a/differential-dataflow/src/trace/implementations/ord_neu.rs
+++ b/differential-dataflow/src/trace/implementations/ord_neu.rs
@@ -301,8 +301,7 @@ pub mod val_batch {
         type Layout = L;
     }
 
-    impl<L: Layout> BatchReader for OrdValBatch<L> {
-
+    impl<L: Layout> crate::trace::Navigable for OrdValBatch<L> {
         type Cursor = OrdValCursor<L>;
         fn cursor(&self) -> Self::Cursor {
             OrdValCursor {
@@ -311,6 +310,11 @@ pub mod val_batch {
                 phantom: PhantomData,
             }
         }
+    }
+
+    impl<L: Layout> BatchReader for OrdValBatch<L> {
+
+        type Time = layout::Time<L>;
         fn len(&self) -> usize {
             // Normally this would be `self.updates.len()`, but we have a clever compact encoding.
             // Perhaps we should count such exceptions to the side, to provide a correct accounting.
@@ -576,6 +580,18 @@ pub mod val_batch {
 
         type Storage = OrdValBatch<L>;
 
+        type KeyContainer = <L as Layout>::KeyContainer;
+        type Key<'a> = <<L as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
+        type ValContainer = <L as Layout>::ValContainer;
+        type Val<'a> = <<L as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
+        type ValOwn = <<L as Layout>::ValContainer as BatchContainer>::Owned;
+        type TimeContainer = <L as Layout>::TimeContainer;
+        type TimeGat<'a> = <<L as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
+        type Time = <<L as Layout>::TimeContainer as BatchContainer>::Owned;
+        type DiffContainer = <L as Layout>::DiffContainer;
+        type DiffGat<'a> = <<L as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
+        type Diff = <<L as Layout>::DiffContainer as BatchContainer>::Owned;
+
         fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { storage.storage.keys.get(self.key_cursor) }
         fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { if self.val_valid(storage) { Some(self.val(storage)) } else { None } }
 
@@ -786,8 +802,7 @@ pub mod key_batch {
         type Layout = L;
     }
 
-    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> BatchReader for OrdKeyBatch<L> {
-
+    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> crate::trace::Navigable for OrdKeyBatch<L> {
         type Cursor = OrdKeyCursor<L>;
         fn cursor(&self) -> Self::Cursor {
             OrdKeyCursor {
@@ -796,6 +811,11 @@ pub mod key_batch {
                 phantom: std::marker::PhantomData,
             }
         }
+    }
+
+    impl<L: Layout<ValContainer: BatchContainer<Owned: Default>>> BatchReader for OrdKeyBatch<L> {
+
+        type Time = layout::Time<L>;
         fn len(&self) -> usize {
             // Normally this would be `self.updates.len()`, but we have a clever compact encoding.
             // Perhaps we should count such exceptions to the side, to provide a correct accounting.
@@ -982,6 +1002,18 @@ pub mod key_batch {
 
         type Storage = OrdKeyBatch<L>;
 
+        type KeyContainer = <L as Layout>::KeyContainer;
+        type Key<'a> = <<L as Layout>::KeyContainer as BatchContainer>::ReadItem<'a>;
+        type ValContainer = <L as Layout>::ValContainer;
+        type Val<'a> = <<L as Layout>::ValContainer as BatchContainer>::ReadItem<'a>;
+        type ValOwn = <<L as Layout>::ValContainer as BatchContainer>::Owned;
+        type TimeContainer = <L as Layout>::TimeContainer;
+        type TimeGat<'a> = <<L as Layout>::TimeContainer as BatchContainer>::ReadItem<'a>;
+        type Time = <<L as Layout>::TimeContainer as BatchContainer>::Owned;
+        type DiffContainer = <L as Layout>::DiffContainer;
+        type DiffGat<'a> = <<L as Layout>::DiffContainer as BatchContainer>::ReadItem<'a>;
+        type Diff = <<L as Layout>::DiffContainer as BatchContainer>::Owned;
+
         fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { storage.storage.keys.get(self.key_cursor) }
         fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { if self.val_valid(storage) { Some(self.val(storage)) } else { None } }
 
diff --git a/differential-dataflow/src/trace/implementations/spine_fueled.rs b/differential-dataflow/src/trace/implementations/spine_fueled.rs
index 5cdb115da..b6b5af6e2 100644
--- a/differential-dataflow/src/trace/implementations/spine_fueled.rs
+++ b/differential-dataflow/src/trace/implementations/spine_fueled.rs
@@ -70,8 +70,7 @@
 
 
 use crate::logging::Logger;
-use crate::trace::{Batch, Trace, TraceReader, ExertionLogic};
-use crate::trace::Merger;
+use crate::trace::{Batch, ExertionLogic, Merger, Trace, TraceReader};
 
 use ::timely::dataflow::operators::generic::OperatorInfo;
 use ::timely::progress::{Antichain, frontier::AntichainRef};
@@ -98,13 +97,9 @@ pub struct Spine<B: Batch> {
     exert_logic: Option<ExertionLogic>,
 }
 
-use crate::trace::WithLayout;
-impl<B: Batch> WithLayout for Spine<B> {
-    type Layout = B::Layout;
-}
-
 impl<B: Batch+Clone+'static> TraceReader for Spine<B> {
 
+    type Time = B::Time;
     type Batch = B;
 
     fn cursor_storage(&mut self, upper: AntichainRef<Self::Time>) -> Option<Vec<Self::Batch>> {
diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs
index 65e7abdd8..d11532150 100644
--- a/differential-dataflow/src/trace/mod.rs
+++ b/differential-dataflow/src/trace/mod.rs
@@ -16,59 +16,36 @@ pub mod wrappers;
 use timely::container::PushInto;
 use timely::progress::{Antichain, frontier::AntichainRef};
 use timely::progress::Timestamp;
+use crate::lattice::Lattice;
 
 use crate::logging::Logger;
 pub use self::cursor::Cursor;
+pub use self::cursor::Navigable;
+pub use self::cursor::{BatchCursor, BatchKey, BatchVal, BatchValOwn, BatchDiff, BatchDiffGat, BatchTimeGat};
 use self::cursor::CursorList;
 pub use self::description::Description;
 
-use crate::trace::implementations::LayoutExt;
-
 /// A type used to express how much effort a trace should exert even in the absence of updates.
 pub type ExertionLogic = std::sync::Arc<dyn for<'a> Fn(&'a [(usize, usize, usize)])->Option<usize>+Send+Sync>;
 
-//     The traces and batch and cursors want the flexibility to appear as if they manage certain types of keys and
-//     values and such, while perhaps using other representations, I'm thinking mostly of wrappers around the keys
-//     and vals that change the `Ord` implementation, or stash hash codes, or the like.
-//
-//     This complicates what requirements we make so that the trace is still usable by someone who knows only about
-//     the base key and value types. For example, the complex types should likely dereference to the simpler types,
-//    so that the user can make sense of the result as if they were given references to the simpler types. At the
-//  same time, the collection should be formable from base types (perhaps we need an `Into` or `From` constraint)
-//  and we should, somehow, be able to take a reference to the simple types to compare against the more complex
-//  types. This second one is also like an `Into` or `From` constraint, except that we start with a reference and
-//  really don't need anything more complex than a reference, but we can't form an owned copy of the complex type
-//  without cloning it.
-//
-//  We could just start by cloning things. Worry about wrapping references later on.
-
 /// A trace whose contents may be read.
 ///
 /// This is a restricted interface to the more general `Trace` trait, which extends this trait with further methods
 /// to update the contents of the trace. These methods are used to examine the contents, and to update the reader's
 /// capabilities (which may release restrictions on the mutations to the underlying trace and cause work to happen).
-pub trait TraceReader : LayoutExt {
+pub trait TraceReader {
+
+    /// The timestamp type of the trace's updates.
+    ///
+    /// Key/val/diff opinions live on the batches' cursors; the trace itself only needs time, to
+    /// bound its contents and to drive compaction.
+    type Time: Timestamp + Lattice;
 
     /// The type of an immutable collection of updates.
     type Batch:
         'static +
         Clone +
-        BatchReader +
-        WithLayout<Layout = Self::Layout> +
-        for<'a> LayoutExt<
-            Key<'a> = Self::Key<'a>,
-            Val<'a> = Self::Val<'a>,
-            ValOwn = Self::ValOwn,
-            Time = Self::Time,
-            TimeGat<'a> = Self::TimeGat<'a>,
-            Diff = Self::Diff,
-            DiffGat<'a> = Self::DiffGat<'a>,
-            KeyContainer = Self::KeyContainer,
-            ValContainer = Self::ValContainer,
-            TimeContainer = Self::TimeContainer,
-            DiffContainer = Self::DiffContainer,
-        >;
-
+        BatchReader<Time = Self::Time>;
 
     /// Acquires the non-empty sequence of batches a cursor would draw from, restricted to updates
     /// at times not greater or equal to an element of `upper`.
@@ -84,7 +61,7 @@ pub trait TraceReader : LayoutExt {
     fn cursor_storage(&mut self, upper: AntichainRef<Self::Time>) -> Option<Vec<Self::Batch>>;
 
     /// Provides a cursor over updates contained in the trace.
-    fn cursor(&mut self) -> (CursorList<<Self::Batch as BatchReader>::Cursor>, Vec<Self::Batch>) {
+    fn cursor(&mut self) -> (CursorList<<Self::Batch as Navigable>::Cursor>, Vec<Self::Batch>) where Self::Batch: Navigable {
         if let Some(cursor) = self.cursor_through(Antichain::new().borrow()) {
             cursor
         }
@@ -98,11 +75,8 @@ pub trait TraceReader : LayoutExt {
     ///
     /// The cursor is a [`CursorList`] that merges the cursors of the batches returned by
     /// [`cursor_storage`](TraceReader::cursor_storage); see that method for the contract on `upper`.
-    fn cursor_through(&mut self, upper: AntichainRef<Self::Time>) -> Option<(CursorList<<Self::Batch as BatchReader>::Cursor>, Vec<Self::Batch>)> {
-        let storage = self.cursor_storage(upper)?;
-        let cursors = storage.iter().map(|batch| batch.cursor()).collect::<Vec<_>>();
-        let cursor = CursorList::new(cursors, &storage);
-        Some((cursor, storage))
+    fn cursor_through(&mut self, upper: AntichainRef<Self::Time>) -> Option<(CursorList<<Self::Batch as Navigable>::Cursor>, Vec<Self::Batch>)> where Self::Batch: Navigable {
+        Some(self::cursor::cursor_list(self.cursor_storage(upper)?))
     }
 
     /// Advances the frontier that constrains logical compaction.
@@ -232,36 +206,20 @@ pub trait Trace : TraceReader<Batch: Batch> {
     fn close(&mut self);
 }
 
-use crate::trace::implementations::WithLayout;
-
 /// A batch of updates whose contents may be read.
 ///
 /// This is a restricted interface to batches of updates, which support the reading of the batch's contents,
 /// but do not expose ways to construct the batches. This trait is appropriate for views of the batch, and is
 /// especially useful for views derived from other sources in ways that prevent the construction of batches
 /// from the type of data in the view (for example, filtered views, or views with extended time coordinates).
-pub trait BatchReader : LayoutExt + Sized {
-
-    /// The type used to enumerate the batch's contents.
-    type Cursor:
-        Cursor<Storage=Self> +
-        WithLayout<Layout = Self::Layout> +
-        for<'a> LayoutExt<
-            Key<'a> = Self::Key<'a>,
-            Val<'a> = Self::Val<'a>,
-            ValOwn = Self::ValOwn,
-            Time = Self::Time,
-            TimeGat<'a> = Self::TimeGat<'a>,
-            Diff = Self::Diff,
-            DiffGat<'a> = Self::DiffGat<'a>,
-            KeyContainer = Self::KeyContainer,
-            ValContainer = Self::ValContainer,
-            TimeContainer = Self::TimeContainer,
-            DiffContainer = Self::DiffContainer,
-        >;
-
-      /// Acquires a cursor to the batch's contents.
-    fn cursor(&self) -> Self::Cursor;
+pub trait BatchReader : Sized {
+
+    /// The timestamp type of the batch's updates.
+    ///
+    /// A batch carries only time; navigating its contents is the separate, optional [`Navigable`]
+    /// capability, which an operator requests with a `Self: Navigable` bound when it needs a cursor.
+    type Time: Timestamp + Lattice;
+
     /// The number of updates in the batch.
     fn len(&self) -> usize;
     /// True if the batch is empty.
@@ -372,21 +330,20 @@ pub mod rc_blanket_impls {
     use std::rc::Rc;
 
     use timely::progress::{Antichain, frontier::AntichainRef};
-    use super::{Batch, BatchReader, Builder, Merger, Cursor, Description};
-
-    impl<B: BatchReader> WithLayout for Rc<B> {
-        type Layout = B::Layout;
-    }
-
-    impl<B: BatchReader> BatchReader for Rc<B> {
+    use super::{Batch, BatchReader, Builder, Merger, Navigable, Cursor, Description};
 
+    impl<B: BatchReader + Navigable> Navigable for Rc<B> {
         /// The type used to enumerate the batch's contents.
         type Cursor = RcBatchCursor<B::Cursor>;
         /// Acquires a cursor to the batch's contents.
         fn cursor(&self) -> Self::Cursor {
             RcBatchCursor::new((**self).cursor())
         }
+    }
+
+    impl<B: BatchReader> BatchReader for Rc<B> {
 
+        type Time = B::Time;
         /// The number of updates in the batch.
         fn len(&self) -> usize { (**self).len() }
         /// Describes the times of the updates in the batch.
@@ -398,11 +355,6 @@ pub mod rc_blanket_impls {
         cursor: C,
     }
 
-    use crate::trace::implementations::WithLayout;
-    impl<C: Cursor> WithLayout for RcBatchCursor<C> {
-        type Layout = C::Layout;
-    }
-
     impl<C> RcBatchCursor<C> {
         fn new(cursor: C) -> Self {
             RcBatchCursor {
@@ -415,6 +367,18 @@ pub mod rc_blanket_impls {
 
         type Storage = Rc<C::Storage>;
 
+        type Key<'a> = C::Key<'a>;
+        type ValOwn = C::ValOwn;
+        type Val<'a> = C::Val<'a>;
+        type Time = C::Time;
+        type TimeGat<'a> = C::TimeGat<'a>;
+        type Diff = C::Diff;
+        type DiffGat<'a> = C::DiffGat<'a>;
+        type KeyContainer = C::KeyContainer;
+        type ValContainer = C::ValContainer;
+        type TimeContainer = C::TimeContainer;
+        type DiffContainer = C::DiffContainer;
+
         #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) }
         #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) }
 
diff --git a/differential-dataflow/src/trace/wrappers/enter.rs b/differential-dataflow/src/trace/wrappers/enter.rs
index f4f9909fd..66eeadc14 100644
--- a/differential-dataflow/src/trace/wrappers/enter.rs
+++ b/differential-dataflow/src/trace/wrappers/enter.rs
@@ -5,7 +5,7 @@ use timely::progress::timestamp::Refines;
 use timely::progress::{Antichain, frontier::AntichainRef};
 
 use crate::lattice::Lattice;
-use crate::trace::{TraceReader, BatchReader, Description};
+use crate::trace::{BatchReader, Description, Navigable, TraceReader};
 use crate::trace::cursor::Cursor;
 
 /// Wrapper to provide trace to nested scope.
@@ -25,26 +25,13 @@ impl<Tr: TraceReader + Clone, TInner> Clone for TraceEnter<Tr, TInner> {
     }
 }
 
-impl<Tr, TInner> WithLayout for TraceEnter<Tr, TInner>
-where
-    Tr: TraceReader,
-    TInner: Refines<Tr::Time>+Lattice,
-{
-    type Layout = (
-        <Tr::Layout as Layout>::KeyContainer,
-        <Tr::Layout as Layout>::ValContainer,
-        Vec<TInner>,
-        <Tr::Layout as Layout>::DiffContainer,
-        <Tr::Layout as Layout>::OffsetContainer,
-    );
-}
-
 impl<Tr, TInner> TraceReader for TraceEnter<Tr, TInner>
 where
     Tr: TraceReader,
     TInner: Refines<Tr::Time>+Lattice,
 {
     type Batch = BatchEnter<Tr::Batch, TInner>;
+    type Time = TInner;
 
     fn map_batches<F: FnMut(&Self::Batch)>(&self, mut f: F) {
         self.trace.map_batches(|batch| {
@@ -115,18 +102,17 @@ pub struct BatchEnter<B, TInner> {
     description: Description<TInner>,
 }
 
-impl<B, TInner> WithLayout for BatchEnter<B, TInner>
+impl<B, TInner> Navigable for BatchEnter<B, TInner>
 where
-    B: BatchReader,
+    B: BatchReader + Navigable,
     TInner: Refines<B::Time>+Lattice,
+    TInner: Refines<<B::Cursor as Cursor>::Time>,
 {
-    type Layout = (
-        <B::Layout as Layout>::KeyContainer,
-        <B::Layout as Layout>::ValContainer,
-        Vec<TInner>,
-        <B::Layout as Layout>::DiffContainer,
-        <B::Layout as Layout>::OffsetContainer,
-    );
+    type Cursor = BatchCursorEnter<B::Cursor, TInner>;
+
+    fn cursor(&self) -> Self::Cursor {
+        BatchCursorEnter::new(self.batch.cursor())
+    }
 }
 
 impl<B, TInner> BatchReader for BatchEnter<B, TInner>
@@ -134,11 +120,7 @@ where
     B: BatchReader,
     TInner: Refines<B::Time>+Lattice,
 {
-    type Cursor = BatchCursorEnter<B::Cursor, TInner>;
-
-    fn cursor(&self) -> Self::Cursor {
-        BatchCursorEnter::new(self.batch.cursor())
-    }
+    type Time = TInner;
     fn len(&self) -> usize { self.batch.len() }
     fn description(&self) -> &Description<TInner> { &self.description }
 }
@@ -161,7 +143,7 @@ where
     }
 }
 
-use crate::trace::implementations::{Layout, WithLayout};
+use crate::trace::implementations::BatchContainer;
 
 /// Wrapper to provide cursor to nested scope.
 pub struct BatchCursorEnter<C, TInner> {
@@ -178,26 +160,24 @@ impl<C, TInner> BatchCursorEnter<C, TInner> {
     }
 }
 
-impl<C, TInner> WithLayout for BatchCursorEnter<C, TInner>
-where
-    C: Cursor,
-    TInner: Refines<C::Time>+Lattice,
-{
-    type Layout = (
-        <C::Layout as Layout>::KeyContainer,
-        <C::Layout as Layout>::ValContainer,
-        Vec<TInner>,
-        <C::Layout as Layout>::DiffContainer,
-        <C::Layout as Layout>::OffsetContainer,
-    );
-}
-
 impl<TInner, C: Cursor> Cursor for BatchCursorEnter<C, TInner>
 where
     TInner: Refines<C::Time>+Lattice,
 {
     type Storage = BatchEnter<C::Storage, TInner>;
 
+    type Key<'a> = C::Key<'a>;
+    type ValOwn = C::ValOwn;
+    type Val<'a> = C::Val<'a>;
+    type KeyContainer = C::KeyContainer;
+    type ValContainer = C::ValContainer;
+    type DiffContainer = C::DiffContainer;
+    type Diff = C::Diff;
+    type DiffGat<'a> = C::DiffGat<'a>;
+    type TimeContainer = Vec<TInner>;
+    type Time = <Vec<TInner> as BatchContainer>::Owned;
+    type TimeGat<'a> = <Vec<TInner> as BatchContainer>::ReadItem<'a>;
+
     #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(&storage.batch) }
     #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(&storage.batch) }
 
diff --git a/differential-dataflow/src/trace/wrappers/enter_at.rs b/differential-dataflow/src/trace/wrappers/enter_at.rs
index 84fcbaf68..f0c56a564 100644
--- a/differential-dataflow/src/trace/wrappers/enter_at.rs
+++ b/differential-dataflow/src/trace/wrappers/enter_at.rs
@@ -4,7 +4,7 @@ use timely::progress::timestamp::Refines;
 use timely::progress::{Antichain, frontier::AntichainRef};
 
 use crate::lattice::Lattice;
-use crate::trace::{TraceReader, BatchReader, Description};
+use crate::trace::{BatchKey, BatchReader, BatchTimeGat, BatchVal, Description, Navigable, TraceReader};
 use crate::trace::cursor::Cursor;
 
 /// Wrapper to provide trace to nested scope.
@@ -40,30 +40,15 @@ where
     }
 }
 
-impl<Tr, TInner, F, G> WithLayout for TraceEnter<Tr, TInner, F, G>
-where
-    Tr: TraceReader,
-    TInner: Refines<Tr::Time>+Lattice,
-    F: Clone,
-    G: Clone,
-{
-    type Layout = (
-        <Tr::Layout as Layout>::KeyContainer,
-        <Tr::Layout as Layout>::ValContainer,
-        Vec<TInner>,
-        <Tr::Layout as Layout>::DiffContainer,
-        <Tr::Layout as Layout>::OffsetContainer,
-    );
-}
-
 impl<Tr, TInner, F, G> TraceReader for TraceEnter<Tr, TInner, F, G>
 where
-    Tr: TraceReader,
+    Tr: TraceReader<Batch: Navigable>,
     TInner: Refines<Tr::Time>+Lattice,
     F: 'static,
-    F: FnMut(Tr::Key<'_>, Tr::Val<'_>, Tr::TimeGat<'_>)->TInner+Clone,
+    F: FnMut(BatchKey<'_, Tr>, BatchVal<'_, Tr>, BatchTimeGat<'_, Tr>)->TInner+Clone,
     G: FnMut(&TInner)->Tr::Time+Clone+'static,
 {
+    type Time = TInner;
     type Batch = BatchEnter<Tr::Batch, TInner,F>;
 
     fn map_batches<F2: FnMut(&Self::Batch)>(&self, mut f: F2) {
@@ -140,32 +125,27 @@ pub struct BatchEnter<B, TInner, F> {
     logic: F,
 }
 
-impl<B, TInner, F> WithLayout for BatchEnter<B, TInner, F>
+impl<B, TInner, F> Navigable for BatchEnter<B, TInner, F>
 where
-    B: BatchReader,
+    B: BatchReader + Navigable,
     TInner: Refines<B::Time>+Lattice,
+    TInner: Refines<<B::Cursor as Cursor>::Time>,
+    F: FnMut(<B::Cursor as Cursor>::Key<'_>, <B::Cursor as Cursor>::Val<'_>, <B::Cursor as Cursor>::TimeGat<'_>)->TInner+Clone,
 {
-    type Layout = (
-        <B::Layout as Layout>::KeyContainer,
-        <B::Layout as Layout>::ValContainer,
-        Vec<TInner>,
-        <B::Layout as Layout>::DiffContainer,
-        <B::Layout as Layout>::OffsetContainer,
-    );
+    type Cursor = BatchCursorEnter<B::Cursor, TInner, F>;
+
+    fn cursor(&self) -> Self::Cursor {
+        BatchCursorEnter::new(self.batch.cursor(), self.logic.clone())
+    }
 }
 
-use crate::trace::implementations::LayoutExt;
 impl<B, TInner, F> BatchReader for BatchEnter<B, TInner, F>
 where
-    B: BatchReader,
+    B: BatchReader + Navigable,
     TInner: Refines<B::Time>+Lattice,
-    F: FnMut(B::Key<'_>, <B::Cursor as LayoutExt>::Val<'_>, B::TimeGat<'_>)->TInner+Clone,
+    F: FnMut(<B::Cursor as Cursor>::Key<'_>, <B::Cursor as Cursor>::Val<'_>, <B::Cursor as Cursor>::TimeGat<'_>)->TInner+Clone,
 {
-    type Cursor = BatchCursorEnter<B::Cursor, TInner, F>;
-
-    fn cursor(&self) -> Self::Cursor {
-        BatchCursorEnter::new(self.batch.cursor(), self.logic.clone())
-    }
+    type Time = TInner;
     fn len(&self) -> usize { self.batch.len() }
     fn description(&self) -> &Description<TInner> { &self.description }
 }
@@ -189,7 +169,7 @@ where
     }
 }
 
-use crate::trace::implementations::{Layout, WithLayout};
+use crate::trace::implementations::BatchContainer;
 
 /// Wrapper to provide cursor to nested scope.
 pub struct BatchCursorEnter<C, TInner, F> {
@@ -198,20 +178,6 @@ pub struct BatchCursorEnter<C, TInner, F> {
     logic: F,
 }
 
-impl<C, TInner, F> WithLayout for BatchCursorEnter<C, TInner, F>
-where
-    C: Cursor,
-    TInner: Refines<C::Time>+Lattice,
-{
-    type Layout = (
-        <C::Layout as Layout>::KeyContainer,
-        <C::Layout as Layout>::ValContainer,
-        Vec<TInner>,
-        <C::Layout as Layout>::DiffContainer,
-        <C::Layout as Layout>::OffsetContainer,
-    );
-}
-
 impl<C, TInner, F> BatchCursorEnter<C, TInner, F> {
     fn new(cursor: C, logic: F) -> Self {
         BatchCursorEnter {
@@ -229,6 +195,18 @@ where
 {
     type Storage = BatchEnter<C::Storage, TInner, F>;
 
+    type Key<'a> = C::Key<'a>;
+    type ValOwn = C::ValOwn;
+    type Val<'a> = C::Val<'a>;
+    type KeyContainer = C::KeyContainer;
+    type ValContainer = C::ValContainer;
+    type DiffContainer = C::DiffContainer;
+    type Diff = C::Diff;
+    type DiffGat<'a> = C::DiffGat<'a>;
+    type TimeContainer = Vec<TInner>;
+    type Time = <Vec<TInner> as BatchContainer>::Owned;
+    type TimeGat<'a> = <Vec<TInner> as BatchContainer>::ReadItem<'a>;
+
     #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(&storage.batch) }
     #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(&storage.batch) }
 
diff --git a/differential-dataflow/src/trace/wrappers/frontier.rs b/differential-dataflow/src/trace/wrappers/frontier.rs
index de8b29c35..36170fd56 100644
--- a/differential-dataflow/src/trace/wrappers/frontier.rs
+++ b/differential-dataflow/src/trace/wrappers/frontier.rs
@@ -8,7 +8,7 @@
 
 use timely::progress::{Antichain, frontier::AntichainRef};
 
-use crate::trace::{TraceReader, BatchReader, Description};
+use crate::trace::{BatchReader, Description, Navigable, TraceReader};
 use crate::trace::cursor::Cursor;
 use crate::lattice::Lattice;
 
@@ -31,18 +31,9 @@ impl<Tr: TraceReader + Clone> Clone for TraceFrontier<Tr> {
     }
 }
 
-impl<Tr: TraceReader> WithLayout for TraceFrontier<Tr> {
-    type Layout = (
-        <Tr::Layout as Layout>::KeyContainer,
-        <Tr::Layout as Layout>::ValContainer,
-        Vec<Tr::Time>,
-        <Tr::Layout as Layout>::DiffContainer,
-        <Tr::Layout as Layout>::OffsetContainer,
-    );
-}
-
 impl<Tr: TraceReader> TraceReader for TraceFrontier<Tr> {
 
+    type Time = Tr::Time;
     type Batch = BatchFrontier<Tr::Batch>;
 
     fn map_batches<F: FnMut(&Self::Batch)>(&self, mut f: F) {
@@ -85,23 +76,21 @@ pub struct BatchFrontier<B: BatchReader> {
     until: Antichain<B::Time>,
 }
 
-impl<B: BatchReader> WithLayout for BatchFrontier<B> {
-    type Layout = (
-        <B::Layout as Layout>::KeyContainer,
-        <B::Layout as Layout>::ValContainer,
-        Vec<B::Time>,
-        <B::Layout as Layout>::DiffContainer,
-        <B::Layout as Layout>::OffsetContainer,
-    );
-}
-
-impl<B: BatchReader> BatchReader for BatchFrontier<B> {
+impl<B> Navigable for BatchFrontier<B>
+where
+    B: BatchReader + Navigable,
+    B::Cursor: Cursor<Time = B::Time>,
+{
 
     type Cursor = BatchCursorFrontier<B::Cursor>;
 
     fn cursor(&self) -> Self::Cursor {
         BatchCursorFrontier::new(self.batch.cursor(), self.since.borrow(), self.until.borrow())
     }
+}
+
+impl<B: BatchReader> BatchReader for BatchFrontier<B> {
+    type Time = B::Time;
     fn len(&self) -> usize { self.batch.len() }
     fn description(&self) -> &Description<B::Time> { self.batch.description() }
 }
@@ -117,7 +106,7 @@ impl<B: BatchReader> BatchFrontier<B> {
     }
 }
 
-use crate::trace::implementations::{Layout, WithLayout};
+use crate::trace::implementations::BatchContainer;
 
 /// Wrapper to provide cursor to nested scope.
 pub struct BatchCursorFrontier<C: Cursor> {
@@ -126,16 +115,6 @@ pub struct BatchCursorFrontier<C: Cursor> {
     until: Antichain<C::Time>,
 }
 
-impl<C: Cursor> WithLayout for BatchCursorFrontier<C> {
-    type Layout = (
-        <C::Layout as Layout>::KeyContainer,
-        <C::Layout as Layout>::ValContainer,
-        Vec<C::Time>,
-        <C::Layout as Layout>::DiffContainer,
-        <C::Layout as Layout>::OffsetContainer,
-    );
-}
-
 impl<C: Cursor> BatchCursorFrontier<C> {
     fn new(cursor: C, since: AntichainRef<C::Time>, until: AntichainRef<C::Time>) -> Self {
         BatchCursorFrontier {
@@ -150,6 +129,18 @@ impl<C: Cursor<Storage: BatchReader>> Cursor for BatchCursorFrontier<C> {
 
     type Storage = BatchFrontier<C::Storage>;
 
+    type Key<'a> = C::Key<'a>;
+    type ValOwn = C::ValOwn;
+    type Val<'a> = C::Val<'a>;
+    type KeyContainer = C::KeyContainer;
+    type ValContainer = C::ValContainer;
+    type DiffContainer = C::DiffContainer;
+    type Diff = C::Diff;
+    type DiffGat<'a> = C::DiffGat<'a>;
+    type TimeContainer = Vec<C::Time>;
+    type Time = <Vec<C::Time> as BatchContainer>::Owned;
+    type TimeGat<'a> = <Vec<C::Time> as BatchContainer>::ReadItem<'a>;
+
     #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(&storage.batch) }
     #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(&storage.batch) }
 
diff --git a/dogsdogsdogs/src/operators/count.rs b/dogsdogsdogs/src/operators/count.rs
index 34aebc738..c46be8c25 100644
--- a/dogsdogsdogs/src/operators/count.rs
+++ b/dogsdogsdogs/src/operators/count.rs
@@ -1,7 +1,7 @@
 use differential_dataflow::{ExchangeData, VecCollection, Hashable};
 use differential_dataflow::difference::{Semigroup, Monoid, Multiply};
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::TraceReader;
+use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, Cursor, Navigable, TraceReader};
 
 /// Reports a number of extensions to a stream of prefixes.
 ///
@@ -16,9 +16,10 @@ pub fn count<'scope, Tr, K, R, F, P>(
     index: usize,
 ) -> VecCollection<'scope, Tr::Time, (P, usize, usize), R>
 where
-    Tr: TraceReader<Time: std::hash::Hash, Diff=isize>+Clone+'static,
-    Tr::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
-    for<'a> Tr::Diff : Semigroup<Tr::DiffGat<'a>>,
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time, Diff=isize>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
+    for<'a> BatchDiff<Tr> : Semigroup<BatchDiffGat<'a, Tr>>,
     K: Hashable + Ord + Default + 'static,
     R: Monoid+Multiply<Output = R>+ExchangeData,
     F: Fn(&P)->K+Clone+'static,
diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs
index 36836357c..c5a9ae9b9 100644
--- a/dogsdogsdogs/src/operators/half_join.rs
+++ b/dogsdogsdogs/src/operators/half_join.rs
@@ -48,7 +48,7 @@ use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable}
 use differential_dataflow::difference::{Monoid, Semigroup};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{Cursor, TraceReader};
+use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchTimeGat, BatchVal, Cursor, Navigable, TraceReader};
 use differential_dataflow::consolidation::{consolidate, consolidate_updates};
 use differential_dataflow::trace::implementations::BatchContainer;
 
@@ -80,20 +80,21 @@ pub fn half_join<'scope, K, V, R, Tr, FF, CF, DOut, S>(
     frontier_func: FF,
     comparison: CF,
     mut output_func: S,
-) -> VecCollection<'scope, Tr::Time, (DOut, Tr::Time), <R as Mul<Tr::Diff>>::Output>
+) -> VecCollection<'scope, Tr::Time, (DOut, Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
 where
     K: Hashable + ExchangeData,
     V: ExchangeData,
     R: ExchangeData + Monoid,
-    Tr: TraceReader<Time: std::hash::Hash>+Clone+'static,
-    Tr::KeyContainer: BatchContainer<Owned=K>,
-    R: Mul<Tr::Diff, Output: Semigroup>,
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
+    R: Mul<BatchDiff<Tr>, Output: Semigroup>,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(Tr::TimeGat<'_>, &Tr::Time) -> bool + 'static,
+    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
     DOut: Clone+'static,
-    S: FnMut(&K, &V, Tr::Val<'_>)->DOut+'static,
+    S: FnMut(&K, &V, BatchVal<'_, Tr>)->DOut+'static,
 {
-    let output_func = move |session: &mut SessionFor<Tr::Time, _>, k: &K, v1: &V, v2: Tr::Val<'_>, initial: &Tr::Time, diff1: &R, output: &mut Vec<(Tr::Time, Tr::Diff)>| {
+    let output_func = move |session: &mut SessionFor<Tr::Time, _>, k: &K, v1: &V, v2: BatchVal<'_, Tr>, initial: &Tr::Time, diff1: &R, output: &mut Vec<(Tr::Time, BatchDiff<Tr>)>| {
         for (time, diff2) in output.drain(..) {
             let diff = diff1.clone() * diff2.clone();
             let dout = (output_func(k, v1, v2), time.clone());
@@ -151,12 +152,13 @@ where
     K: Hashable + ExchangeData,
     V: ExchangeData,
     R: ExchangeData + Monoid,
-    Tr: TraceReader<Time: std::hash::Hash>+Clone+'static,
-    Tr::KeyContainer: BatchContainer<Owned=K>,
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(Tr::TimeGat<'_>, &Tr::Time) -> bool + 'static,
+    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
     Y: Fn(std::time::Instant, usize) -> bool + 'static,
-    S: FnMut(&mut SessionFor<Tr::Time, CB>, &K, &V, Tr::Val<'_>, &Tr::Time, &R, &mut Vec<(Tr::Time, Tr::Diff)>) + 'static,
+    S: FnMut(&mut SessionFor<Tr::Time, CB>, &K, &V, BatchVal<'_, Tr>, &Tr::Time, &R, &mut Vec<(Tr::Time, BatchDiff<Tr>)>) + 'static,
     CB: ContainerBuilder,
 {
     // No need to block physical merging for this operator.
@@ -302,7 +304,7 @@ fn process_proposals<Tr, CF, Y, S, CB, K, V, R>(
     comparison: &CF,
     yield_function: &Y,
     output_func: &mut S,
-    mut output_buffer: &mut Vec<(Tr::Time, Tr::Diff)>,
+    mut output_buffer: &mut Vec<(Tr::Time, BatchDiff<Tr>)>,
     timer: Instant,
     work: &mut usize,
     trace: &mut Tr,
@@ -311,11 +313,12 @@ fn process_proposals<Tr, CF, Y, S, CB, K, V, R>(
     frontier: AntichainRef<Tr::Time>
 ) -> bool
 where
-    Tr: TraceReader,
-    Tr::KeyContainer: BatchContainer<Owned=K>,
-    CF: Fn(Tr::TimeGat<'_>, &Tr::Time) -> bool + 'static,
+    Tr: TraceReader<Batch: Navigable>,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
+    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
     Y: Fn(Instant, usize) -> bool + 'static,
-    S: FnMut(&mut SessionFor<Tr::Time, CB>, &K, &V, Tr::Val<'_>, &Tr::Time, &R, &mut Vec<(Tr::Time, Tr::Diff)>) + 'static,
+    S: FnMut(&mut SessionFor<Tr::Time, CB>, &K, &V, BatchVal<'_, Tr>, &Tr::Time, &R, &mut Vec<(Tr::Time, BatchDiff<Tr>)>) + 'static,
     CB: ContainerBuilder,
     K: Ord,
     V: Ord,
@@ -327,8 +330,8 @@ where
     let (mut cursor, storage) = trace.cursor();
     let mut yielded = false;
 
-    let mut key_con = Tr::KeyContainer::with_capacity(1);
-    let mut time_con = Tr::TimeContainer::with_capacity(1);
+    let mut key_con = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
+    let mut time_con = <BatchCursor<Tr> as Cursor>::TimeContainer::with_capacity(1);
     for time in frontier.iter() {
         time_con.push_own(time);
     }
@@ -345,9 +348,9 @@ where
                 while let Some(val2) = cursor.get_val(&storage) {
                     cursor.map_times(&storage, |t, d| {
                         if comparison(t, initial) {
-                            let mut t = Tr::owned_time(t);
+                            let mut t = <BatchCursor<Tr> as Cursor>::owned_time(t);
                             t.join_assign(time);
-                            output_buffer.push((t, Tr::owned_diff(d)))
+                            output_buffer.push((t, <BatchCursor<Tr> as Cursor>::owned_diff(d)))
                         }
                     });
                     consolidate(&mut output_buffer);
diff --git a/dogsdogsdogs/src/operators/half_join2.rs b/dogsdogsdogs/src/operators/half_join2.rs
index 2c61b8e3f..e9deb66ce 100644
--- a/dogsdogsdogs/src/operators/half_join2.rs
+++ b/dogsdogsdogs/src/operators/half_join2.rs
@@ -33,7 +33,7 @@ use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable}
 use differential_dataflow::difference::{Monoid, Semigroup};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{Cursor, TraceReader};
+use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchTimeGat, BatchVal, Cursor, Navigable, TraceReader};
 use differential_dataflow::consolidation::{consolidate, consolidate_updates};
 use differential_dataflow::trace::implementations::BatchContainer;
 
@@ -67,20 +67,21 @@ pub fn half_join<'scope, K, V, R, Tr, FF, CF, DOut, S>(
     frontier_func: FF,
     comparison: CF,
     mut output_func: S,
-) -> VecCollection<'scope, Tr::Time, (DOut, Tr::Time), <R as Mul<Tr::Diff>>::Output>
+) -> VecCollection<'scope, Tr::Time, (DOut, Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
 where
     K: Hashable + ExchangeData,
     V: ExchangeData,
     R: ExchangeData + Monoid,
-    Tr: TraceReader<Time: std::hash::Hash>+Clone+'static,
-    Tr::KeyContainer: BatchContainer<Owned=K>,
-    R: Mul<Tr::Diff, Output: Semigroup>,
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
+    R: Mul<BatchDiff<Tr>, Output: Semigroup>,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(Tr::TimeGat<'_>, &Tr::Time) -> bool + 'static,
+    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
     DOut: Clone+'static,
-    S: FnMut(&K, &V, Tr::Val<'_>)->DOut+'static,
+    S: FnMut(&K, &V, BatchVal<'_, Tr>)->DOut+'static,
 {
-    let output_func = move |builder: &mut CapacityContainerBuilder<Vec<_>>, k: &K, v1: &V, v2: Tr::Val<'_>, initial: &Tr::Time, diff1: &R, output: &mut Vec<(Tr::Time, Tr::Diff)>| {
+    let output_func = move |builder: &mut CapacityContainerBuilder<Vec<_>>, k: &K, v1: &V, v2: BatchVal<'_, Tr>, initial: &Tr::Time, diff1: &R, output: &mut Vec<(Tr::Time, BatchDiff<Tr>)>| {
         for (time, diff2) in output.drain(..) {
             let diff = diff1.clone() * diff2.clone();
             let dout = (output_func(k, v1, v2), time.clone());
@@ -129,12 +130,13 @@ where
     K: Hashable + ExchangeData,
     V: ExchangeData,
     R: ExchangeData + Monoid,
-    Tr: TraceReader<Time: std::hash::Hash>+Clone+'static,
-    Tr::KeyContainer: BatchContainer<Owned=K>,
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(Tr::TimeGat<'_>, &Tr::Time) -> bool + 'static,
+    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
     Y: Fn(std::time::Instant, usize) -> bool + 'static,
-    S: FnMut(&mut CB, &K, &V, Tr::Val<'_>, &Tr::Time, &R, &mut Vec<(Tr::Time, Tr::Diff)>) + 'static,
+    S: FnMut(&mut CB, &K, &V, BatchVal<'_, Tr>, &Tr::Time, &R, &mut Vec<(Tr::Time, BatchDiff<Tr>)>) + 'static,
     CB: ContainerBuilder,
 {
     // No need to block physical merging for this operator.
@@ -183,7 +185,7 @@ where
 
                 // Determine the total-order minimum of the arrangement frontier,
                 // used to partition arrivals into immediately-eligible vs stuck.
-                let mut time_con = Tr::TimeContainer::with_capacity(1);
+                let mut time_con = <BatchCursor<Tr> as Cursor>::TimeContainer::with_capacity(1);
                 if let Some(min_time) = frontier.iter().min() {
                     time_con.push_own(min_time);
                 }
@@ -262,7 +264,7 @@ where
                     let mut builders = (0..blob.caps.len()).map(|_| CB::default()).collect::<Vec<_>>();
 
                     let (mut cursor, storage) = trace.cursor();
-                    let mut key_con = Tr::KeyContainer::with_capacity(1);
+                    let mut key_con = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
                     let mut removals: ChangeBatch<Tr::Time> = ChangeBatch::new();
 
                     // Process ready elements from the front.
@@ -282,9 +284,9 @@ where
                             while let Some(val2) = cursor.get_val(&storage) {
                                 cursor.map_times(&storage, |t, d| {
                                     if comparison(t, initial) {
-                                        let mut t = Tr::owned_time(t);
+                                        let mut t = <BatchCursor<Tr> as Cursor>::owned_time(t);
                                         t.join_assign(time);
-                                        output_buffer.push((t, Tr::owned_diff(d)))
+                                        output_buffer.push((t, <BatchCursor<Tr> as Cursor>::owned_diff(d)))
                                     }
                                 });
                                 consolidate(&mut output_buffer);
diff --git a/dogsdogsdogs/src/operators/lookup_map.rs b/dogsdogsdogs/src/operators/lookup_map.rs
index 702bdf138..82fd2ed47 100644
--- a/dogsdogsdogs/src/operators/lookup_map.rs
+++ b/dogsdogsdogs/src/operators/lookup_map.rs
@@ -8,7 +8,7 @@ use timely::progress::Antichain;
 use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable};
 use differential_dataflow::difference::{IsZero, Semigroup, Monoid};
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{Cursor, TraceReader};
+use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, BatchVal, Cursor, Navigable, TraceReader};
 use differential_dataflow::trace::implementations::BatchContainer;
 
 /// Proposes extensions to a stream of prefixes.
@@ -26,18 +26,19 @@ pub fn lookup_map<'scope, D, K, R, Tr, F, DOut, ROut, S>(
     supplied_key2: K,
 ) -> VecCollection<'scope, Tr::Time, DOut, ROut>
 where
-    Tr: for<'a> TraceReader<
-        Time: std::hash::Hash,
-        Diff : Semigroup<Tr::DiffGat<'a>>+Monoid+ExchangeData,
-    >+Clone+'static,
-    Tr::KeyContainer: BatchContainer<Owned=K>,
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    for<'a> BatchCursor<Tr>: Cursor<
+        Time = Tr::Time,
+        Diff : Semigroup<BatchDiffGat<'a, Tr>>+Monoid+ExchangeData,
+    >,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
     K: Hashable + Ord + 'static,
     F: FnMut(&D, &mut K)+Clone+'static,
     D: ExchangeData,
     R: ExchangeData+Monoid,
     DOut: Clone+'static,
     ROut: Monoid + 'static,
-    S: FnMut(&D, &R, Tr::Val<'_>, &Tr::Diff)->(DOut, ROut)+'static,
+    S: FnMut(&D, &R, BatchVal<'_, Tr>, &BatchDiff<Tr>)->(DOut, ROut)+'static,
 {
     // No need to block physical merging for this operator.
     arrangement.trace.set_physical_compaction(Antichain::new().borrow());
@@ -89,7 +90,7 @@ where
 
                     let (mut cursor, storage) = trace.cursor();
                     // Key container to stage keys for comparison.
-                    let mut key_con = Tr::KeyContainer::with_capacity(1);
+                    let mut key_con = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
                     for &mut (ref prefix, ref time, ref mut diff) in prefixes.iter_mut() {
                         if !frontier2.less_equal(time) {
                             logic2(prefix, &mut key1);
@@ -97,9 +98,9 @@ where
                             cursor.seek_key(&storage, key_con.index(1));
                             if cursor.get_key(&storage) == Some(key_con.index(1)) {
                                 while let Some(value) = cursor.get_val(&storage) {
-                                    let mut count = Tr::Diff::zero();
+                                    let mut count = BatchDiff::<Tr>::zero();
                                     cursor.map_times(&storage, |t, d| {
-                                        if Tr::owned_time(t).less_equal(time) { count.plus_equals(&d); }
+                                        if <BatchCursor<Tr> as Cursor>::owned_time(t).less_equal(time) { count.plus_equals(&d); }
                                     });
                                     if !count.is_zero() {
                                         let (dout, rout) = output_func(prefix, diff, value, &count);
diff --git a/dogsdogsdogs/src/operators/propose.rs b/dogsdogsdogs/src/operators/propose.rs
index 187c79256..9415985bd 100644
--- a/dogsdogsdogs/src/operators/propose.rs
+++ b/dogsdogsdogs/src/operators/propose.rs
@@ -1,7 +1,7 @@
 use differential_dataflow::{ExchangeData, VecCollection, Hashable};
 use differential_dataflow::difference::{Semigroup, Monoid, Multiply};
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::TraceReader;
+use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, Cursor, Navigable, TraceReader};
 
 /// Proposes extensions to a prefix stream.
 ///
@@ -12,17 +12,18 @@ use differential_dataflow::trace::TraceReader;
 /// `arrangement` undergoes. More complicated patterns are also appropriate, as in the case
 /// of delta queries.
 pub fn propose<'scope, Tr, K, F, P, V>(
-    prefixes: VecCollection<'scope, Tr::Time, P, Tr::Diff>,
+    prefixes: VecCollection<'scope, Tr::Time, P, BatchDiff<Tr>>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
-) -> VecCollection<'scope, Tr::Time, (P, V), Tr::Diff>
+) -> VecCollection<'scope, Tr::Time, (P, V), BatchDiff<Tr>>
 where
-    Tr: for<'a> TraceReader<
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    for<'a> BatchCursor<Tr>: Cursor<
+        Time = Tr::Time,
         ValOwn = V,
-        Time: std::hash::Hash,
-        Diff: Monoid+Multiply<Output = Tr::Diff>+ExchangeData+Semigroup<Tr::DiffGat<'a>>,
-    >+Clone+'static,
-    Tr::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
+        Diff: Monoid+Multiply<Output = BatchDiff<Tr>>+ExchangeData+Semigroup<BatchDiffGat<'a, Tr>>,
+    >,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
     K: Hashable + Default + Ord + 'static,
     F: Fn(&P)->K+Clone+'static,
     P: ExchangeData,
@@ -32,7 +33,7 @@ where
         prefixes,
         arrangement,
         move |p: &P, k: &mut K | { *k = key_selector(p); },
-        move |prefix, diff, value, sum| ((prefix.clone(), Tr::owned_val(value)), diff.clone().multiply(sum)),
+        move |prefix, diff, value, sum| ((prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)), diff.clone().multiply(sum)),
         Default::default(),
         Default::default(),
         Default::default(),
@@ -45,17 +46,18 @@ 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<'scope, Tr, K, F, P, V>(
-    prefixes: VecCollection<'scope, Tr::Time, P, Tr::Diff>,
+    prefixes: VecCollection<'scope, Tr::Time, P, BatchDiff<Tr>>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
-) -> VecCollection<'scope, Tr::Time, (P, V), Tr::Diff>
+) -> VecCollection<'scope, Tr::Time, (P, V), BatchDiff<Tr>>
 where
-    Tr: for<'a> TraceReader<
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    for<'a> BatchCursor<Tr>: Cursor<
+        Time = Tr::Time,
         ValOwn = V,
-        Time: std::hash::Hash,
-        Diff : Semigroup<Tr::DiffGat<'a>>+Monoid+Multiply<Output = Tr::Diff>+ExchangeData,
-    >+Clone+'static,
-    Tr::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
+        Diff : Semigroup<BatchDiffGat<'a, Tr>>+Monoid+Multiply<Output = BatchDiff<Tr>>+ExchangeData,
+    >,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
     K: Hashable + Default + Ord + 'static,
     F: Fn(&P)->K+Clone+'static,
     P: ExchangeData,
@@ -65,7 +67,7 @@ where
         prefixes,
         arrangement,
         move |p: &P, k: &mut K| { *k = key_selector(p); },
-        move |prefix, diff, value, _sum| ((prefix.clone(), Tr::owned_val(value)), diff.clone()),
+        move |prefix, diff, value, _sum| ((prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)), diff.clone()),
         Default::default(),
         Default::default(),
         Default::default(),
diff --git a/dogsdogsdogs/src/operators/validate.rs b/dogsdogsdogs/src/operators/validate.rs
index 565bcf510..262957f71 100644
--- a/dogsdogsdogs/src/operators/validate.rs
+++ b/dogsdogsdogs/src/operators/validate.rs
@@ -3,7 +3,7 @@ use std::hash::Hash;
 use differential_dataflow::{ExchangeData, VecCollection};
 use differential_dataflow::difference::{Semigroup, Monoid, Multiply};
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::TraceReader;
+use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, Cursor, Navigable, TraceReader};
 
 /// Proposes extensions to a stream of prefixes.
 ///
@@ -11,16 +11,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<'scope, K, V, Tr, F, P>(
-    extensions: VecCollection<'scope, Tr::Time, (P, V), Tr::Diff>,
+    extensions: VecCollection<'scope, Tr::Time, (P, V), BatchDiff<Tr>>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
-) -> VecCollection<'scope, Tr::Time, (P, V), Tr::Diff>
+) -> VecCollection<'scope, Tr::Time, (P, V), BatchDiff<Tr>>
 where
-    Tr: for<'a> TraceReader<
-        Time: std::hash::Hash,
-        Diff : Semigroup<Tr::DiffGat<'a>>+Monoid+Multiply<Output = Tr::Diff>+ExchangeData,
-    >+Clone+'static,
-    Tr::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=(K,V)>,
+    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    for<'a> BatchCursor<Tr>: Cursor<
+        Time = Tr::Time,
+        Diff : Semigroup<BatchDiffGat<'a, Tr>>+Monoid+Multiply<Output = BatchDiff<Tr>>+ExchangeData,
+    >,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=(K,V)>,
     K: Ord+Hash+Clone+Default + 'static,
     V: ExchangeData+Hash+Default,
     F: Fn(&P)->K+Clone+'static,