From f3c37dcfc0e5bb9982980f40f1cd95810eae69d0 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Thu, 5 Dec 2024 15:52:17 -0500
Subject: [PATCH] Remove batcher from Trace

There is no inherent reason a trace must know about a batcher, other than
convenience. This change factors the batcher out of the trace and moves
it to an explicit type parameter where necessary.

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>
---
 examples/spines.rs                        | 30 ++++++++--------
 experiments/src/bin/deals.rs              | 14 ++++----
 experiments/src/bin/graspan1.rs           |  6 ++--
 experiments/src/bin/graspan2.rs           | 40 ++++++++++-----------
 src/operators/arrange/arrangement.rs      | 42 +++++++++++++----------
 src/operators/consolidate.rs              | 13 +++----
 src/operators/reduce.rs                   |  2 +-
 src/trace/implementations/mod.rs          |  2 ++
 src/trace/implementations/ord_neu.rs      | 36 ++++++++++++-------
 src/trace/implementations/rhh.rs          |  8 +++--
 src/trace/implementations/spine_fueled.rs | 26 +++++---------
 src/trace/mod.rs                          |  4 +--
 tests/bfs.rs                              | 13 +++----
 tests/trace.rs                            |  4 +--
 14 files changed, 127 insertions(+), 113 deletions(-)

diff --git a/examples/spines.rs b/examples/spines.rs
index 7fc726783..e4cd50bea 100644
--- a/examples/spines.rs
+++ b/examples/spines.rs
@@ -28,46 +28,46 @@ fn main() {
 
             match mode.as_str() {
                 "new" => {
-                    use differential_dataflow::trace::implementations::ord_neu::ColKeySpine;
-                    let data = data.arrange::<ColKeySpine<_,_,_>>();
-                    let keys = keys.arrange::<ColKeySpine<_,_,_>>();
+                    use differential_dataflow::trace::implementations::ord_neu::{ColKeyBatcher, ColKeySpine};
+                    let data = data.arrange::<ColKeyBatcher<_,_,_>, ColKeySpine<_,_,_>>();
+                    let keys = keys.arrange::<ColKeyBatcher<_,_,_>, ColKeySpine<_,_,_>>();
                     keys.join_core(&data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                 },
                 "old" => {
-                    use differential_dataflow::trace::implementations::ord_neu::OrdKeySpine;
-                    let data = data.arrange::<OrdKeySpine<_,_,_>>();
-                    let keys = keys.arrange::<OrdKeySpine<_,_,_>>();
+                    use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatcher, OrdKeySpine};
+                    let data = data.arrange::<OrdKeyBatcher<_,_,_>, OrdKeySpine<_,_,_>>();
+                    let keys = keys.arrange::<OrdKeyBatcher<_,_,_>, OrdKeySpine<_,_,_>>();
                     keys.join_core(&data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                 },
                 "rhh" => {
-                    use differential_dataflow::trace::implementations::rhh::{HashWrapper, VecSpine};
-                    let data = data.map(|x| HashWrapper { inner: x }).arrange::<VecSpine<_,(),_,_>>();
-                    let keys = keys.map(|x| HashWrapper { inner: x }).arrange::<VecSpine<_,(),_,_>>();
+                    use differential_dataflow::trace::implementations::rhh::{HashWrapper, VecBatcher, VecSpine};
+                    let data = data.map(|x| HashWrapper { inner: x }).arrange::<VecBatcher<_,(),_,_>, VecSpine<_,(),_,_>>();
+                    let keys = keys.map(|x| HashWrapper { inner: x }).arrange::<VecBatcher<_,(),_,_>, VecSpine<_,(),_,_>>();
                     keys.join_core(&data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                 },
                 "slc" => {
 
-                    use differential_dataflow::trace::implementations::ord_neu::PreferredSpine;
+                    use differential_dataflow::trace::implementations::ord_neu::{PreferredBatcher, PreferredSpine};
 
                     let data =
                     data.map(|x| (x.clone().into_bytes(), x.into_bytes()))
-                        .arrange::<PreferredSpine<[u8],[u8],_,_>>()
+                        .arrange::<PreferredBatcher<[u8],[u8],_,_>, PreferredSpine<[u8],[u8],_,_>>()
                         .reduce_abelian::<_, _, _, PreferredSpine<[u8],(),_,_>>("distinct", |_,_,output| output.push(((), 1)));
                     let keys =
                     keys.map(|x| (x.clone().into_bytes(), 7))
-                        .arrange::<PreferredSpine<[u8],u8,_,_>>()
+                        .arrange::<PreferredBatcher<[u8],u8,_,_>, PreferredSpine<[u8],u8,_,_>>()
                         .reduce_abelian::<_, _, _, PreferredSpine<[u8],(),_,_>>("distinct", |_,_,output| output.push(((), 1)));
 
                     keys.join_core(&data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                 },
                 "flat" => {
-                    use differential_dataflow::trace::implementations::ord_neu::FlatKeySpineDefault;
-                    let data = data.arrange::<FlatKeySpineDefault<String,usize,isize, _>>();
-                    let keys = keys.arrange::<FlatKeySpineDefault<String,usize,isize,_>>();
+                    use differential_dataflow::trace::implementations::ord_neu::{FlatKeyBatcherDefault, FlatKeySpineDefault};
+                    let data = data.arrange::<FlatKeyBatcherDefault<String,usize,isize,_>, FlatKeySpineDefault<String,usize,isize>>();
+                    let keys = keys.arrange::<FlatKeyBatcherDefault<String,usize,isize,_>, FlatKeySpineDefault<String,usize,isize>>();
                     keys.join_core(&data, |_k, (), ()| Option::<()>::None)
                         .probe_with(&mut probe);
                 }
diff --git a/experiments/src/bin/deals.rs b/experiments/src/bin/deals.rs
index d162ca457..ab7871ccb 100644
--- a/experiments/src/bin/deals.rs
+++ b/experiments/src/bin/deals.rs
@@ -6,7 +6,7 @@ use differential_dataflow::input::Input;
 use differential_dataflow::Collection;
 use differential_dataflow::operators::*;
 
-use differential_dataflow::trace::implementations::{ValSpine, KeySpine};
+use differential_dataflow::trace::implementations::{ValSpine, KeySpine, KeyBatcher, ValBatcher};
 use differential_dataflow::operators::arrange::TraceAgent;
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::operators::arrange::Arrange;
@@ -41,7 +41,7 @@ fn main() {
             let (input, graph) = scope.new_collection();
 
             // each edge should exist in both directions.
-            let graph = graph.arrange::<ValSpine<_,_,_,_>>();
+            let graph = graph.arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
 
             match program.as_str() {
                 "tc"    => tc(&graph).filter(move |_| inspect).map(|_| ()).consolidate().inspect(|x| println!("tc count: {:?}", x)).probe(),
@@ -94,10 +94,10 @@ fn tc<G: Scope<Timestamp=()>>(edges: &EdgeArranged<G, Node, Node, Present>) -> C
             let result =
             inner
                 .map(|(x,y)| (y,x))
-                .arrange::<ValSpine<_,_,_,_>>()
+                .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                 .join_core(&edges, |_y,&x,&z| Some((x, z)))
                 .concat(&edges.as_collection(|&k,&v| (k,v)))
-                .arrange::<KeySpine<_,_,_>>()
+                .arrange::<KeyBatcher<_,_,_>, KeySpine<_,_,_>>()
                 .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None })
                 ;
 
@@ -121,12 +121,12 @@ fn sg<G: Scope<Timestamp=()>>(edges: &EdgeArranged<G, Node, Node, Present>) -> C
 
             let result =
             inner
-                .arrange::<ValSpine<_,_,_,_>>()
+                .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                 .join_core(&edges, |_,&x,&z| Some((x, z)))
-                .arrange::<ValSpine<_,_,_,_>>()
+                .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                 .join_core(&edges, |_,&x,&z| Some((x, z)))
                 .concat(&peers)
-                .arrange::<KeySpine<_,_,_>>()
+                .arrange::<KeyBatcher<_,_,_>, KeySpine<_,_,_>>()
                 .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None })
                 ;
 
diff --git a/experiments/src/bin/graspan1.rs b/experiments/src/bin/graspan1.rs
index baa50bd00..41266c4b4 100644
--- a/experiments/src/bin/graspan1.rs
+++ b/experiments/src/bin/graspan1.rs
@@ -6,7 +6,7 @@ use timely::order::Product;
 
 use differential_dataflow::difference::Present;
 use differential_dataflow::input::Input;
-use differential_dataflow::trace::implementations::ValSpine;
+use differential_dataflow::trace::implementations::{ValBatcher, ValSpine};
 use differential_dataflow::operators::*;
 use differential_dataflow::operators::arrange::Arrange;
 use differential_dataflow::operators::iterate::SemigroupVariable;
@@ -31,7 +31,7 @@ fn main() {
             let (n_handle, nodes) = scope.new_collection();
             let (e_handle, edges) = scope.new_collection();
 
-            let edges = edges.arrange::<ValSpine<_,_,_,_>>();
+            let edges = edges.arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
 
             // a N c  <-  a N b && b E c
             // N(a,c) <-  N(a,b), E(b, c)
@@ -46,7 +46,7 @@ fn main() {
                 let next =
                 labels.join_core(&edges, |_b, a, c| Some((*c, *a)))
                       .concat(&nodes)
-                      .arrange::<ValSpine<_,_,_,_>>()
+                      .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                     //   .distinct_total_core::<Diff>();
                       .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None });
 
diff --git a/experiments/src/bin/graspan2.rs b/experiments/src/bin/graspan2.rs
index 637936aa4..cde55b71c 100644
--- a/experiments/src/bin/graspan2.rs
+++ b/experiments/src/bin/graspan2.rs
@@ -10,7 +10,7 @@ use differential_dataflow::Collection;
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::*;
 use differential_dataflow::operators::arrange::Arrange;
-use differential_dataflow::trace::implementations::{ValSpine, KeySpine};
+use differential_dataflow::trace::implementations::{ValSpine, KeySpine, ValBatcher, KeyBatcher};
 use differential_dataflow::difference::Present;
 
 type Node = u32;
@@ -47,7 +47,7 @@ fn unoptimized() {
                 .flat_map(|(a,b)| vec![a,b])
                 .concat(&dereference.flat_map(|(a,b)| vec![a,b]));
 
-            let dereference = dereference.arrange::<ValSpine<_,_,_,_>>();
+            let dereference = dereference.arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
 
             let (value_flow, memory_alias, value_alias) =
             scope
@@ -60,14 +60,14 @@ fn unoptimized() {
                     let value_flow = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
                     let memory_alias = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
 
-                    let value_flow_arranged = value_flow.arrange::<ValSpine<_,_,_,_>>();
-                    let memory_alias_arranged = memory_alias.arrange::<ValSpine<_,_,_,_>>();
+                    let value_flow_arranged = value_flow.arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
+                    let memory_alias_arranged = memory_alias.arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
 
                     // VA(a,b) <- VF(x,a),VF(x,b)
                     // VA(a,b) <- VF(x,a),MA(x,y),VF(y,b)
                     let value_alias_next = value_flow_arranged.join_core(&value_flow_arranged, |_,&a,&b| Some((a,b)));
                     let value_alias_next = value_flow_arranged.join_core(&memory_alias_arranged, |_,&a,&b| Some((b,a)))
-                                                              .arrange::<ValSpine<_,_,_,_>>()
+                                                              .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                                                               .join_core(&value_flow_arranged, |_,&a,&b| Some((a,b)))
                                                               .concat(&value_alias_next);
 
@@ -77,16 +77,16 @@ fn unoptimized() {
                     let value_flow_next =
                     assignment
                         .map(|(a,b)| (b,a))
-                        .arrange::<ValSpine<_,_,_,_>>()
+                        .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                         .join_core(&memory_alias_arranged, |_,&a,&b| Some((b,a)))
                         .concat(&assignment.map(|(a,b)| (b,a)))
-                        .arrange::<ValSpine<_,_,_,_>>()
+                        .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                         .join_core(&value_flow_arranged, |_,&a,&b| Some((a,b)))
                         .concat(&nodes.map(|n| (n,n)));
 
                     let value_flow_next =
                     value_flow_next
-                        .arrange::<KeySpine<_,_,_>>()
+                        .arrange::<ValBatcher<_,_,_,_>, KeySpine<_,_,_>>()
                         // .distinct_total_core::<Diff>()
                         .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None })
                         ;
@@ -95,12 +95,12 @@ fn unoptimized() {
                     let memory_alias_next: Collection<_,_,Present> =
                     value_alias_next
                         .join_core(&dereference, |_x,&y,&a| Some((y,a)))
-                        .arrange::<ValSpine<_,_,_,_>>()
+                        .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                         .join_core(&dereference, |_y,&a,&b| Some((a,b)));
 
                     let memory_alias_next: Collection<_,_,Present>  =
                     memory_alias_next
-                        .arrange::<KeySpine<_,_,_>>()
+                        .arrange::<KeyBatcher<_,_,_>, KeySpine<_,_,_>>()
                         // .distinct_total_core::<Diff>()
                         .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None })
                         ;
@@ -172,7 +172,7 @@ fn optimized() {
                 .flat_map(|(a,b)| vec![a,b])
                 .concat(&dereference.flat_map(|(a,b)| vec![a,b]));
 
-            let dereference = dereference.arrange::<ValSpine<_,_,_,_>>();
+            let dereference = dereference.arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
 
             let (value_flow, memory_alias) =
             scope
@@ -185,8 +185,8 @@ fn optimized() {
                     let value_flow = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
                     let memory_alias = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
 
-                    let value_flow_arranged = value_flow.arrange::<ValSpine<_,_,_,_>>();
-                    let memory_alias_arranged = memory_alias.arrange::<ValSpine<_,_,_,_>>();
+                    let value_flow_arranged = value_flow.arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
+                    let memory_alias_arranged = memory_alias.arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
 
                     // VF(a,a) <-
                     // VF(a,b) <- A(a,x),VF(x,b)
@@ -194,13 +194,13 @@ fn optimized() {
                     let value_flow_next =
                     assignment
                         .map(|(a,b)| (b,a))
-                        .arrange::<ValSpine<_,_,_,_>>()
+                        .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                         .join_core(&memory_alias_arranged, |_,&a,&b| Some((b,a)))
                         .concat(&assignment.map(|(a,b)| (b,a)))
-                        .arrange::<ValSpine<_,_,_,_>>()
+                        .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                         .join_core(&value_flow_arranged, |_,&a,&b| Some((a,b)))
                         .concat(&nodes.map(|n| (n,n)))
-                        .arrange::<KeySpine<_,_,_>>()
+                        .arrange::<KeyBatcher<_,_,_>, KeySpine<_,_,_>>()
                         // .distinct_total_core::<Diff>()
                         .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None })
                         ;
@@ -209,9 +209,9 @@ fn optimized() {
                     let value_flow_deref =
                     value_flow
                         .map(|(a,b)| (b,a))
-                        .arrange::<ValSpine<_,_,_,_>>()
+                        .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                         .join_core(&dereference, |_x,&a,&b| Some((a,b)))
-                        .arrange::<ValSpine<_,_,_,_>>();
+                        .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>();
 
                     // MA(a,b) <- VFD(x,a),VFD(y,b)
                     // MA(a,b) <- VFD(x,a),MA(x,y),VFD(y,b)
@@ -222,10 +222,10 @@ fn optimized() {
                     let memory_alias_next =
                     memory_alias_arranged
                         .join_core(&value_flow_deref, |_x,&y,&a| Some((y,a)))
-                        .arrange::<ValSpine<_,_,_,_>>()
+                        .arrange::<ValBatcher<_,_,_,_>, ValSpine<_,_,_,_>>()
                         .join_core(&value_flow_deref, |_y,&a,&b| Some((a,b)))
                         .concat(&memory_alias_next)
-                        .arrange::<KeySpine<_,_,_>>()
+                        .arrange::<KeyBatcher<_,_,_>, KeySpine<_,_,_>>()
                         // .distinct_total_core::<Diff>()
                         .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None })
                         ;
diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index fd9e7fd08..c0615d8c0 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -30,7 +30,7 @@ use crate::{Data, ExchangeData, Collection, AsCollection, Hashable};
 use crate::difference::Semigroup;
 use crate::lattice::Lattice;
 use crate::trace::{self, Trace, TraceReader, Batch, BatchReader, Batcher, Builder, Cursor};
-use crate::trace::implementations::{KeySpine, ValSpine};
+use crate::trace::implementations::{KeyBatcher, KeySpine, ValBatcher, ValSpine};
 
 use trace::wrappers::enter::{TraceEnter, BatchEnter,};
 use trace::wrappers::enter_at::TraceEnter as TraceEnterAt;
@@ -353,21 +353,23 @@ where
     G::Timestamp: Lattice,
 {
     /// Arranges updates into a shared trace.
-    fn arrange<Tr>(&self) -> Arranged<G, TraceAgent<Tr>>
+    fn arrange<Ba, Tr>(&self) -> Arranged<G, TraceAgent<Tr>>
     where
+        Ba: Batcher<Input=C, Time=G::Timestamp> + 'static,
         Tr: Trace<Time=G::Timestamp> + 'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=C>,
+        Tr::Builder: Builder<Input=Ba::Output>,
     {
-        self.arrange_named("Arrange")
+        self.arrange_named::<Ba, Tr>("Arrange")
     }
 
     /// Arranges updates into a shared trace, with a supplied name.
-    fn arrange_named<Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
+    fn arrange_named<Ba, Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
+        Ba: Batcher<Input=C, Time=G::Timestamp> + 'static,
         Tr: Trace<Time=G::Timestamp> + 'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=C>,
+        Tr::Builder: Builder<Input=Ba::Output>,
     ;
 }
 
@@ -379,14 +381,15 @@ where
     V: ExchangeData,
     R: ExchangeData + Semigroup,
 {
-    fn arrange_named<Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
+    fn arrange_named<Ba, Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
+        Ba: Batcher<Input=Vec<((K, V), G::Timestamp, R)>, Time=G::Timestamp> + 'static,
         Tr: Trace<Time=G::Timestamp> + 'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=Vec<((K, V), G::Timestamp, R)>>,
+        Tr::Builder: Builder<Input=Ba::Output>,
     {
         let exchange = Exchange::new(move |update: &((K,V),G::Timestamp,R)| (update.0).0.hashed().into());
-        arrange_core(&self.inner, exchange, name)
+        arrange_core::<_, _, Ba, _>(&self.inner, exchange, name)
     }
 }
 
@@ -395,14 +398,16 @@ where
 /// This operator arranges a stream of values into a shared trace, whose contents it maintains.
 /// It uses the supplied parallelization contract to distribute the data, which does not need to
 /// be consistently by key (though this is the most common).
-pub fn arrange_core<G, P, Tr>(stream: &StreamCore<G, <Tr::Batcher as Batcher>::Input>, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
+pub fn arrange_core<G, P, Ba, Tr>(stream: &StreamCore<G, Ba::Input>, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
 where
     G: Scope,
     G::Timestamp: Lattice,
-    P: ParallelizationContract<G::Timestamp, <Tr::Batcher as Batcher>::Input>,
+    P: ParallelizationContract<G::Timestamp, Ba::Input>,
+    Ba: Batcher<Time=G::Timestamp> + 'static,
+    Ba::Input: Container,
     Tr: Trace<Time=G::Timestamp>+'static,
     Tr::Batch: Batch,
-    <Tr::Batcher as Batcher>::Input: timely::Container,
+    Tr::Builder: Builder<Input=Ba::Output>,
 {
     // The `Arrange` operator is tasked with reacting to an advancing input
     // frontier by producing the sequence of batches whose lower and upper
@@ -434,7 +439,7 @@ where
         };
 
         // Where we will deposit received updates, and from which we extract batches.
-        let mut batcher = Tr::Batcher::new(logger.clone(), info.global_id);
+        let mut batcher = Ba::new(logger.clone(), info.global_id);
 
         // Capabilities for the lower envelope of updates in `batcher`.
         let mut capabilities = Antichain::<Capability<G::Timestamp>>::new();
@@ -557,14 +562,15 @@ impl<G: Scope, K: ExchangeData+Hashable, R: ExchangeData+Semigroup> Arrange<G, V
 where
     G::Timestamp: Lattice+Ord,
 {
-    fn arrange_named<Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
+    fn arrange_named<Ba, Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
+        Ba: Batcher<Input=Vec<((K,()),G::Timestamp,R)>, Time=G::Timestamp> + 'static,
         Tr: Trace<Time=G::Timestamp> + 'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=Vec<((K, ()), G::Timestamp, R)>>,
+        Tr::Builder: Builder<Input=Ba::Output>,
     {
         let exchange = Exchange::new(move |update: &((K,()),G::Timestamp,R)| (update.0).0.hashed().into());
-        arrange_core(&self.map(|k| (k, ())).inner, exchange, name)
+        arrange_core::<_,_,Ba,_>(&self.map(|k| (k, ())).inner, exchange, name)
     }
 }
 
@@ -595,7 +601,7 @@ where
     }
 
     fn arrange_by_key_named(&self, name: &str) -> Arranged<G, TraceAgent<ValSpine<K, V, G::Timestamp, R>>> {
-        self.arrange_named(name)
+        self.arrange_named::<ValBatcher<_,_,_,_>, _>(name)
     }
 }
 
@@ -630,6 +636,6 @@ where
 
     fn arrange_by_self_named(&self, name: &str) -> Arranged<G, TraceAgent<KeySpine<K, G::Timestamp, R>>> {
         self.map(|k| (k, ()))
-            .arrange_named(name)
+            .arrange_named::<KeyBatcher<_,_,_>, _>(name)
     }
 }
diff --git a/src/operators/consolidate.rs b/src/operators/consolidate.rs
index 41824bd6f..e3a551246 100644
--- a/src/operators/consolidate.rs
+++ b/src/operators/consolidate.rs
@@ -16,7 +16,7 @@ use crate::difference::Semigroup;
 
 use crate::Data;
 use crate::lattice::Lattice;
-use crate::trace::Batcher;
+use crate::trace::{Batcher, Builder};
 
 /// Methods which require data be arrangeable.
 impl<G, D, R> Collection<G, D, R>
@@ -47,21 +47,22 @@ where
     /// });
     /// ```
     pub fn consolidate(&self) -> Self {
-        use crate::trace::implementations::KeySpine;
-        self.consolidate_named::<KeySpine<_,_,_>>("Consolidate")
+        use crate::trace::implementations::{KeyBatcher, KeySpine};
+        self.consolidate_named::<KeyBatcher<_, _, _>, KeySpine<_,_,_>>("Consolidate")
     }
 
     /// As `consolidate` but with the ability to name the operator and specify the trace type.
-    pub fn consolidate_named<Tr>(&self, name: &str) -> Self
+    pub fn consolidate_named<Ba, Tr>(&self, name: &str) -> Self
     where
+        Ba: Batcher<Input=Vec<((D,()),G::Timestamp,R)>, Time=G::Timestamp> + 'static,
         Tr: crate::trace::Trace<Time=G::Timestamp,Diff=R>+'static,
         for<'a> Tr::Key<'a>: IntoOwned<'a, Owned = D>,
         Tr::Batch: crate::trace::Batch,
-        Tr::Batcher: Batcher<Input=Vec<((D,()),G::Timestamp,R)>>,
+        Tr::Builder: Builder<Input=Ba::Output>,
     {
         use crate::operators::arrange::arrangement::Arrange;
         self.map(|k| (k, ()))
-            .arrange_named::<Tr>(name)
+            .arrange_named::<Ba, Tr>(name)
             .as_collection(|d, _| d.into_owned())
     }
 
diff --git a/src/operators/reduce.rs b/src/operators/reduce.rs
index 6b1017130..2818340ec 100644
--- a/src/operators/reduce.rs
+++ b/src/operators/reduce.rs
@@ -460,7 +460,7 @@ where
                             builders.push(T2::Builder::new());
                         }
 
-                        let mut buffer = <<T2 as Trace>::Batcher as crate::trace::Batcher>::Output::default();
+                        let mut buffer = <T2::Builder as Builder>::Input::default();
 
                         // cursors for navigating input and output traces.
                         let (mut source_cursor, source_storage): (T1::Cursor, _) = source_trace.cursor_through(lower_limit.borrow()).expect("failed to acquire source cursor");
diff --git a/src/trace/implementations/mod.rs b/src/trace/implementations/mod.rs
index 168f5427c..b7414b986 100644
--- a/src/trace/implementations/mod.rs
+++ b/src/trace/implementations/mod.rs
@@ -50,7 +50,9 @@ pub mod chunker;
 
 // Opinionated takes on default spines.
 pub use self::ord_neu::OrdValSpine as ValSpine;
+pub use self::ord_neu::OrdValBatcher as ValBatcher;
 pub use self::ord_neu::OrdKeySpine as KeySpine;
+pub use self::ord_neu::OrdKeyBatcher as KeyBatcher;
 
 use std::borrow::{ToOwned};
 use std::convert::TryInto;
diff --git a/src/trace/implementations/ord_neu.rs b/src/trace/implementations/ord_neu.rs
index f870a6277..35c2941dc 100644
--- a/src/trace/implementations/ord_neu.rs
+++ b/src/trace/implementations/ord_neu.rs
@@ -28,69 +28,79 @@ pub use self::key_batch::{OrdKeyBatch, OrdKeyBuilder};
 /// A trace implementation using a spine of ordered lists.
 pub type OrdValSpine<K, V, T, R> = Spine<
     Rc<OrdValBatch<Vector<((K,V),T,R)>>>,
-    MergeBatcher<Vec<((K,V),T,R)>, VecChunker<((K,V),T,R)>, VecMerger<((K, V), T, R)>, T>,
     RcBuilder<OrdValBuilder<Vector<((K,V),T,R)>, Vec<((K,V),T,R)>>>,
 >;
+/// A batcher using ordered lists.
+pub type OrdValBatcher<K, V, T, R> = MergeBatcher<Vec<((K,V),T,R)>, VecChunker<((K,V),T,R)>, VecMerger<((K, V), T, R)>, T>;
+
 // /// A trace implementation for empty values using a spine of ordered lists.
 // pub type OrdKeySpine<K, T, R> = Spine<Rc<OrdKeyBatch<Vector<((K,()),T,R)>>>>;
 
 /// A trace implementation backed by columnar storage.
 pub type ColValSpine<K, V, T, R> = Spine<
     Rc<OrdValBatch<TStack<((K,V),T,R)>>>,
-    MergeBatcher<Vec<((K,V),T,R)>, ColumnationChunker<((K,V),T,R)>, ColumnationMerger<((K,V),T,R)>, T>,
     RcBuilder<OrdValBuilder<TStack<((K,V),T,R)>, TimelyStack<((K,V),T,R)>>>,
 >;
+/// A batcher for columnar storage.
+pub type ColValBatcher<K, V, T, R> = MergeBatcher<Vec<((K,V),T,R)>, ColumnationChunker<((K,V),T,R)>, ColumnationMerger<((K,V),T,R)>, T>;
 
 /// A trace implementation backed by flatcontainer storage.
-pub type FlatValSpine<L, R, C> = Spine<
+pub type FlatValSpine<L, R> = Spine<
     Rc<OrdValBatch<L>>,
-    MergeBatcher<C, ContainerChunker<FlatStack<R>>, FlatcontainerMerger<R>, <R as MergerChunk>::TimeOwned>,
     RcBuilder<OrdValBuilder<L, FlatStack<R>>>,
 >;
+/// A batcher for flatcontainer storage.
+pub type FlatValBatcher<R, C> = MergeBatcher<C, ContainerChunker<FlatStack<R>>, FlatcontainerMerger<R>, <R as MergerChunk>::TimeOwned>;
 
 /// A trace implementation backed by flatcontainer storage, using [`FlatLayout`] as the layout.
-pub type FlatValSpineDefault<K, V, T, R, C> = FlatValSpine<
+pub type FlatValSpineDefault<K, V, T, R> = FlatValSpine<
     FlatLayout<<K as RegionPreference>::Region, <V as RegionPreference>::Region, <T as RegionPreference>::Region, <R as RegionPreference>::Region>,
     TupleABCRegion<TupleABRegion<<K as RegionPreference>::Region, <V as RegionPreference>::Region>, <T as RegionPreference>::Region, <R as RegionPreference>::Region>,
-    C,
 >;
+/// A batcher for flatcontainer storage, using [`FlatLayout`] as the layout.
+pub type FlatValBatcherDefault<K, V, T, R, C> = FlatValBatcher<TupleABCRegion<TupleABRegion<<K as RegionPreference>::Region, <V as RegionPreference>::Region>, <T as RegionPreference>::Region, <R as RegionPreference>::Region>, C>;
 
 /// A trace implementation using a spine of ordered lists.
 pub type OrdKeySpine<K, T, R> = Spine<
     Rc<OrdKeyBatch<Vector<((K,()),T,R)>>>,
-    MergeBatcher<Vec<((K,()),T,R)>, VecChunker<((K,()),T,R)>, VecMerger<((K, ()), T, R)>, T>,
     RcBuilder<OrdKeyBuilder<Vector<((K,()),T,R)>, Vec<((K,()),T,R)>>>,
 >;
+/// A batcher for ordered lists.
+pub type OrdKeyBatcher<K, T, R> = MergeBatcher<Vec<((K,()),T,R)>, VecChunker<((K,()),T,R)>, VecMerger<((K, ()), T, R)>, T>;
 // /// A trace implementation for empty values using a spine of ordered lists.
 // pub type OrdKeySpine<K, T, R> = Spine<Rc<OrdKeyBatch<Vector<((K,()),T,R)>>>>;
 
 /// A trace implementation backed by columnar storage.
 pub type ColKeySpine<K, T, R> = Spine<
     Rc<OrdKeyBatch<TStack<((K,()),T,R)>>>,
-    MergeBatcher<Vec<((K,()),T,R)>, ColumnationChunker<((K,()),T,R)>, ColumnationMerger<((K,()),T,R)>, T>,
     RcBuilder<OrdKeyBuilder<TStack<((K,()),T,R)>, TimelyStack<((K,()),T,R)>>>,
 >;
+/// A batcher for columnar storage
+pub type ColKeyBatcher<K, T, R> = MergeBatcher<Vec<((K,()),T,R)>, ColumnationChunker<((K,()),T,R)>, ColumnationMerger<((K,()),T,R)>, T>;
 
 /// A trace implementation backed by flatcontainer storage.
-pub type FlatKeySpine<L, R, C> = Spine<
+pub type FlatKeySpine<L, R> = Spine<
     Rc<OrdKeyBatch<L>>,
-    MergeBatcher<C, ContainerChunker<FlatStack<R>>, FlatcontainerMerger<R>, <R as MergerChunk>::TimeOwned>,
     RcBuilder<OrdKeyBuilder<L, FlatStack<R>>>,
 >;
+/// A batcher for flatcontainer storage.
+pub type FlatKeyBatcher<R, C> = MergeBatcher<C, ContainerChunker<FlatStack<R>>, FlatcontainerMerger<R>, <R as MergerChunk>::TimeOwned>;
 
 /// A trace implementation backed by flatcontainer storage, using [`FlatLayout`] as the layout.
-pub type FlatKeySpineDefault<K,T,R, C> = FlatKeySpine<
+pub type FlatKeySpineDefault<K,T,R> = FlatKeySpine<
     FlatLayout<<K as RegionPreference>::Region, <() as RegionPreference>::Region, <T as RegionPreference>::Region, <R as RegionPreference>::Region>,
     TupleABCRegion<TupleABRegion<<K as RegionPreference>::Region, <() as RegionPreference>::Region>, <T as RegionPreference>::Region, <R as RegionPreference>::Region>,
-    C,
 >;
+/// A batcher for flatcontainer storage, using [`FlatLayout`] as the layout.
+pub type FlatKeyBatcherDefault<K, T, R, C> = FlatValBatcher<TupleABCRegion<TupleABRegion<<K as RegionPreference>::Region, <() as RegionPreference>::Region>, <T as RegionPreference>::Region, <R as RegionPreference>::Region>, C>;
 
 /// A trace implementation backed by columnar storage.
 pub type PreferredSpine<K, V, T, R> = Spine<
     Rc<OrdValBatch<Preferred<K,V,T,R>>>,
-    MergeBatcher<Vec<((<K as ToOwned>::Owned,<V as ToOwned>::Owned),T,R)>, ColumnationChunker<((<K as ToOwned>::Owned,<V as ToOwned>::Owned),T,R)>, ColumnationMerger<((<K as ToOwned>::Owned,<V as ToOwned>::Owned),T,R)>,T>,
     RcBuilder<OrdValBuilder<Preferred<K,V,T,R>, TimelyStack<((<K as ToOwned>::Owned,<V as ToOwned>::Owned),T,R)>>>,
 >;
+/// A batcher for columnar storage.
+pub type PreferredBatcher<K, V, T, R> = MergeBatcher<Vec<((<K as ToOwned>::Owned,<V as ToOwned>::Owned),T,R)>, ColumnationChunker<((<K as ToOwned>::Owned,<V as ToOwned>::Owned),T,R)>, ColumnationMerger<((<K as ToOwned>::Owned,<V as ToOwned>::Owned),T,R)>,T>;
 
 
 // /// A trace implementation backed by columnar storage.
diff --git a/src/trace/implementations/rhh.rs b/src/trace/implementations/rhh.rs
index 01f314650..a07b2dc67 100644
--- a/src/trace/implementations/rhh.rs
+++ b/src/trace/implementations/rhh.rs
@@ -25,18 +25,22 @@ use self::val_batch::{RhhValBatch, RhhValBuilder};
 /// A trace implementation using a spine of ordered lists.
 pub type VecSpine<K, V, T, R> = Spine<
     Rc<RhhValBatch<Vector<((K,V),T,R)>>>,
-    MergeBatcher<Vec<((K,V),T,R)>, VecChunker<((K,V),T,R)>, VecMerger<((K, V), T, R)>, T>,
     RcBuilder<RhhValBuilder<Vector<((K,V),T,R)>, Vec<((K,V),T,R)>>>,
 >;
+/// A batcher for ordered lists.
+pub type VecBatcher<K,V,T,R> = MergeBatcher<Vec<((K,V),T,R)>, VecChunker<((K,V),T,R)>, VecMerger<((K, V), T, R)>, T>;
+
 // /// A trace implementation for empty values using a spine of ordered lists.
 // pub type OrdKeySpine<K, T, R> = Spine<Rc<OrdKeyBatch<Vector<((K,()),T,R)>>>>;
 
 /// A trace implementation backed by columnar storage.
 pub type ColSpine<K, V, T, R> = Spine<
     Rc<RhhValBatch<TStack<((K,V),T,R)>>>,
-    MergeBatcher<Vec<((K,V),T,R)>, ColumnationChunker<((K,V),T,R)>, ColumnationMerger<((K,V),T,R)>, T>,
     RcBuilder<RhhValBuilder<TStack<((K,V),T,R)>, TimelyStack<((K,V),T,R)>>>,
 >;
+/// A batcher for columnar storage.
+pub type ColBatcher<K,V,T,R> = MergeBatcher<Vec<((K,V),T,R)>, ColumnationChunker<((K,V),T,R)>, ColumnationMerger<((K,V),T,R)>, T>;
+
 // /// A trace implementation backed by columnar storage.
 // pub type ColKeySpine<K, T, R> = Spine<Rc<OrdKeyBatch<TStack<((K,()),T,R)>>>>;
 
diff --git a/src/trace/implementations/spine_fueled.rs b/src/trace/implementations/spine_fueled.rs
index e6e014aa6..deefbaa61 100644
--- a/src/trace/implementations/spine_fueled.rs
+++ b/src/trace/implementations/spine_fueled.rs
@@ -70,7 +70,7 @@
 
 
 use crate::logging::Logger;
-use crate::trace::{Batch, Batcher, Builder, BatchReader, Trace, TraceReader, ExertionLogic};
+use crate::trace::{Batch, Builder, BatchReader, Trace, TraceReader, ExertionLogic};
 use crate::trace::cursor::CursorList;
 use crate::trace::Merger;
 
@@ -83,12 +83,7 @@ use ::timely::order::PartialOrder;
 /// A spine maintains a small number of immutable collections of update tuples, merging the collections when
 /// two have similar sizes. In this way, it allows the addition of more tuples, which may then be merged with
 /// other immutable collections.
-pub struct Spine<B: Batch, BA, BU>
-where
-    // Intended constraints:
-    // BA: Batcher<Time = B::Time>,
-    // BU: Builder<Item=BA::Item, Time=BA::Time, Output = B>,
-{
+pub struct Spine<B: Batch, BU> {
     operator: OperatorInfo,
     logger: Option<Logger>,
     logical_frontier: Antichain<B::Time>,   // Times after which the trace must accumulate correctly.
@@ -102,10 +97,10 @@ where
     exert_logic_param: Vec<(usize, usize, usize)>,
     /// Logic to indicate whether and how many records we should introduce in the absence of actual updates.
     exert_logic: Option<ExertionLogic>,
-    phantom: std::marker::PhantomData<(BA, BU)>,
+    phantom: std::marker::PhantomData<BU>,
 }
 
-impl<B, BA, BU> TraceReader for Spine<B, BA, BU>
+impl<B, BU> TraceReader for Spine<B, BU>
 where
     B: Batch+Clone+'static,
 {
@@ -251,14 +246,11 @@ where
 
 // A trace implementation for any key type that can be borrowed from or converted into `Key`.
 // TODO: Almost all this implementation seems to be generic with respect to the trace and batch types.
-impl<B, BA, BU> Trace for Spine<B, BA, BU>
+impl<B, BU> Trace for Spine<B, BU>
 where
     B: Batch+Clone+'static,
-    BA: Batcher<Time = B::Time>,
-    BU: Builder<Input=BA::Output, Time=BA::Time, Output = B>,
+    BU: Builder<Output=B, Time=B::Time>,
 {
-    /// A type used to assemble batches from disordered updates.
-    type Batcher = BA;
     /// A type used to assemble batches from ordered update sequences.
     type Builder = BU;
 
@@ -332,14 +324,14 @@ where
 }
 
 // Drop implementation allows us to log batch drops, to zero out maintained totals.
-impl<B: Batch, BA, BU> Drop for Spine<B, BA, BU> {
+impl<B: Batch, BU> Drop for Spine<B, BU> {
     fn drop(&mut self) {
         self.drop_batches();
     }
 }
 
 
-impl<B: Batch, BA, BU> Spine<B, BA, BU> {
+impl<B: Batch, BU> Spine<B, BU> {
     /// Drops and logs batches. Used in `set_logical_compaction` and drop.
     fn drop_batches(&mut self) {
         if let Some(logger) = &self.logger {
@@ -380,7 +372,7 @@ impl<B: Batch, BA, BU> Spine<B, BA, BU> {
     }
 }
 
-impl<B: Batch, BA, BU> Spine<B, BA, BU> {
+impl<B: Batch, BU> Spine<B, BU> {
     /// Determine the amount of effort we should exert in the absence of updates.
     ///
     /// This method prepares an iterator over batches, including the level, count, and length of each layer.
diff --git a/src/trace/mod.rs b/src/trace/mod.rs
index a7b68ef8d..07a6c937c 100644
--- a/src/trace/mod.rs
+++ b/src/trace/mod.rs
@@ -209,10 +209,8 @@ pub trait TraceReader {
 pub trait Trace : TraceReader
 where <Self as TraceReader>::Batch: Batch {
 
-    /// A type used to assemble batches from disordered updates.
-    type Batcher: Batcher<Time = Self::Time>;
     /// A type used to assemble batches from ordered update sequences.
-    type Builder: Builder<Input=<Self::Batcher as Batcher>::Output, Time=Self::Time, Output = Self::Batch>;
+    type Builder: Builder<Time=Self::Time, Output = Self::Batch>;
 
     /// Allocates a new empty trace.
     fn new(
diff --git a/tests/bfs.rs b/tests/bfs.rs
index 58471947b..43d08c4eb 100644
--- a/tests/bfs.rs
+++ b/tests/bfs.rs
@@ -17,7 +17,7 @@ use differential_dataflow::Collection;
 use differential_dataflow::operators::*;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arrange;
-use differential_dataflow::trace::implementations::ord_neu::{FlatKeySpineDefault, FlatValSpineDefault};
+use differential_dataflow::trace::implementations::ord_neu::{FlatKeyBatcherDefault, FlatKeySpineDefault, FlatValBatcherDefault, FlatValSpineDefault};
 
 type Node = usize;
 type Edge = (Node, Node);
@@ -246,8 +246,8 @@ fn bfs_differential_flat(
             let (edge_input, edges) = scope.new_collection();
 
             let c = bfs_flat(&edges, &roots).map(|(_, dist)| (dist, ()));
-            let arranged = c.arrange::<FlatKeySpineDefault<usize, usize, isize, Vec<((usize, ()), _, _)>>>();
-            type T2 = FlatValSpineDefault<usize, isize, usize, isize, Vec<((usize, isize), usize, isize)>>;
+            let arranged = c.arrange::<FlatKeyBatcherDefault<usize,usize,isize,Vec<((usize,()),_,_)>>, FlatKeySpineDefault<usize, usize, isize>>();
+            type T2 = FlatValSpineDefault<usize, isize, usize, isize>;
             let reduced = arranged.reduce_abelian::<_, _, _, T2>("Count", |_k, s, t| {
                 t.push((s[0].1.clone(), isize::from(1i8)))
             });
@@ -315,9 +315,10 @@ where
         let edges = edges.enter(&inner.scope());
         let nodes = nodes.enter(&inner.scope());
 
-        type Spine<K, V, T, R = isize> = FlatValSpineDefault<K, V, T, R, Vec<((K, V), T, R)>>;
-        let arranged1 = inner.arrange::<Spine<Node, Node, Product<G::Timestamp, _>>>();
-        let arranged2 = edges.arrange::<Spine<Node, Node, Product<G::Timestamp, _>>>();
+        type Batcher<K, V, T, R = isize> = FlatValBatcherDefault<K, V, T, R, Vec<((K,V),T,R)>>;
+        type Spine<K, V, T, R = isize> = FlatValSpineDefault<K, V, T, R>;
+        let arranged1 = inner.arrange::<Batcher<Node, Node, Product<G::Timestamp, u64>>, Spine<Node, Node, Product<G::Timestamp, u64>>>();
+        let arranged2 = edges.arrange::<Batcher<Node, Node, Product<G::Timestamp, u64>>, Spine<Node, Node, Product<G::Timestamp, u64>>>();
         arranged1
             .join_core(&arranged2, move |_k, l, d| Some((d, l + 1)))
             .concat(&nodes)
diff --git a/tests/trace.rs b/tests/trace.rs
index 10c42cb79..addf72cfb 100644
--- a/tests/trace.rs
+++ b/tests/trace.rs
@@ -1,7 +1,7 @@
 use timely::dataflow::operators::generic::OperatorInfo;
 use timely::progress::{Antichain, frontier::AntichainRef};
 
-use differential_dataflow::trace::implementations::ValSpine;
+use differential_dataflow::trace::implementations::{ValBatcher, ValSpine};
 use differential_dataflow::trace::{Trace, TraceReader, Batcher};
 use differential_dataflow::trace::cursor::Cursor;
 
@@ -12,7 +12,7 @@ fn get_trace() -> ValSpine<u64, u64, usize, i64> {
     let op_info = OperatorInfo::new(0, 0, [].into());
     let mut trace = IntegerTrace::new(op_info, None, None);
     {
-        let mut batcher = <IntegerTrace as Trace>::Batcher::new(None, 0);
+        let mut batcher = ValBatcher::<u64,u64,usize,i64>::new(None, 0);
 
         batcher.push_container(&mut vec![
             ((1, 2), 0, 1),