From 1bd6c4ec9d98343b89d49e9a4ad28483355c9231 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 22 Nov 2023 14:34:54 -0500
Subject: [PATCH] Allow Batcher::seal to be generic in Builder

---
 src/operators/arrange/arrangement.rs          | 34 +++----
 src/operators/arrange/upsert.rs               |  2 +-
 src/operators/consolidate.rs                  |  6 +-
 src/operators/reduce.rs                       |  8 +-
 src/trace/implementations/merge_batcher.rs    | 40 +++-----
 .../implementations/merge_batcher_col.rs      | 51 +++++-----
 src/trace/implementations/ord.rs              | 44 +++++----
 src/trace/implementations/ord_neu.rs          | 19 ++--
 src/trace/mod.rs                              | 98 +++++++------------
 tests/trace.rs                                |  6 +-
 10 files changed, 143 insertions(+), 165 deletions(-)

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index ae929dbdd..ca6986328 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -29,7 +29,7 @@ use timely::dataflow::operators::Capability;
 use ::{Data, ExchangeData, Collection, AsCollection, Hashable};
 use ::difference::Semigroup;
 use lattice::Lattice;
-use trace::{self, Trace, TraceReader, Batch, BatchReader, Batcher, Cursor};
+use trace::{self, Trace, TraceReader, Batch, BatchReader, Batcher, Builder, Cursor};
 use trace::implementations::{KeySpine, ValSpine};
 
 use trace::wrappers::enter::{TraceEnter, BatchEnter,};
@@ -456,8 +456,8 @@ where
         R: ExchangeData,
         Tr: Trace+TraceReader<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Item = ((K,V),G::Timestamp,R)>,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         self.arrange_named("Arrange")
     }
@@ -474,8 +474,8 @@ where
         R: ExchangeData,
         Tr: Trace+TraceReader<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Item = ((K,V),G::Timestamp,R)>,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         let exchange = Exchange::new(move |update: &((K,V),G::Timestamp,R)| (update.0).0.hashed().into());
         self.arrange_core(exchange, name)
@@ -491,8 +491,8 @@ where
         P: ParallelizationContract<G::Timestamp, ((K,V),G::Timestamp,R)>,
         Tr: Trace+TraceReader<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Item = ((K,V),G::Timestamp,R)>,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     ;
 }
 
@@ -510,8 +510,8 @@ where
         V: ExchangeData,
         R: ExchangeData,
         Tr: Trace + TraceReader<Time=G::Timestamp> + 'static, Tr::Batch: Batch,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Item = ((K,V),G::Timestamp,R)>,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         self.arrange_named("Arrange")
     }
@@ -522,8 +522,8 @@ where
         V: ExchangeData,
         R: ExchangeData,
         Tr: Trace + TraceReader<Time=G::Timestamp> + 'static, Tr::Batch: Batch,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Item = ((K,V),G::Timestamp,R)>,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         let exchange = Exchange::new(move |update: &((K,V),G::Timestamp,R)| (update.0).0.hashed().into());
         self.arrange_core(exchange, name)
@@ -534,8 +534,8 @@ where
         P: ParallelizationContract<G::Timestamp, ((K,V),G::Timestamp,R)>,
         Tr: Trace+TraceReader<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Item = ((K,V),G::Timestamp,R)>,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         // The `Arrange` operator is tasked with reacting to an advancing input
         // frontier by producing the sequence of batches whose lower and upper
@@ -645,7 +645,7 @@ where
                                     }
 
                                     // Extract updates not in advance of `upper`.
-                                    let batch = batcher.seal(upper.clone());
+                                    let batch = batcher.seal::<<Tr::Batch as Batch>::Builder>(upper.clone());
 
                                     writer.insert(batch.clone(), Some(capability.time().clone()));
 
@@ -673,7 +673,7 @@ where
                         }
                         else {
                             // Announce progress updates, even without data.
-                            let _batch = batcher.seal(input.frontier().frontier().to_owned());
+                            let _batch = batcher.seal::<<Tr::Batch as Batch>::Builder>(input.frontier().frontier().to_owned());
                             writer.seal(input.frontier().frontier().to_owned());
                         }
 
@@ -699,8 +699,8 @@ where
         P: ParallelizationContract<G::Timestamp, ((K,()),G::Timestamp,R)>,
         Tr: Trace+TraceReader<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Item = ((K,()),G::Timestamp,R)>,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Batcher: Batcher<Item = ((K,()),G::Timestamp,R), Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Builder: Builder<Item = ((K,()),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         self.map(|k| (k, ()))
             .arrange_core(pact, name)
diff --git a/src/operators/arrange/upsert.rs b/src/operators/arrange/upsert.rs
index 242203b20..bdfa23fb2 100644
--- a/src/operators/arrange/upsert.rs
+++ b/src/operators/arrange/upsert.rs
@@ -146,7 +146,7 @@ where
     Tr::Val: ExchangeData,
     Tr: Trace+TraceReader<Time=G::Timestamp,R=isize>+'static,
     Tr::Batch: Batch,
-    <Tr::Batch as Batch>::Builder: Builder<Tr::Batch, Item = ((Tr::Key, Tr::Val), Tr::Time, Tr::R)>,
+    <Tr::Batch as Batch>::Builder: Builder<Item = ((Tr::Key, Tr::Val), Tr::Time, Tr::R)>,
 {
     let mut reader: Option<TraceAgent<Tr>> = None;
 
diff --git a/src/operators/consolidate.rs b/src/operators/consolidate.rs
index 2ace0a06c..52776d781 100644
--- a/src/operators/consolidate.rs
+++ b/src/operators/consolidate.rs
@@ -13,7 +13,7 @@ use ::difference::Semigroup;
 
 use Data;
 use lattice::Lattice;
-use trace::{Batch, Batcher};
+use trace::{Batch, Batcher, Builder};
 
 /// Methods which require data be arrangeable.
 impl<G, D, R> Collection<G, D, R>
@@ -58,8 +58,8 @@ where
     where
         Tr: crate::trace::Trace+crate::trace::TraceReader<Key=D,Val=(),Time=G::Timestamp,R=R>+'static,
         Tr::Batch: crate::trace::Batch,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Item = ((D,()),G::Timestamp,R)>,
-        <Tr::Batch as Batch>::Batcher: Batcher<Tr::Batch, Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Batcher: Batcher<Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
+        <Tr::Batch as Batch>::Builder: Builder<Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
     {
         use operators::arrange::arrangement::Arrange;
         self.map(|k| (k, ()))
diff --git a/src/operators/reduce.rs b/src/operators/reduce.rs
index 3a8ce9bc1..7185e83da 100644
--- a/src/operators/reduce.rs
+++ b/src/operators/reduce.rs
@@ -276,7 +276,7 @@ pub trait ReduceCore<G: Scope, K: Data, V: Data, R: Semigroup> where G::Timestam
             T2::Val: Data,
             T2::R: Abelian,
             T2::Batch: Batch,
-            <T2::Batch as Batch>::Builder: Builder<T2::Batch, Item = ((T2::Key, T2::Val), T2::Time, T2::R)>,
+            <T2::Batch as Batch>::Builder: Builder<Output=T2::Batch, Item = ((T2::Key, T2::Val), T2::Time, T2::R)>,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val, T2::R)>)+'static,
         {
             self.reduce_core::<_,T2>(name, move |key, input, output, change| {
@@ -299,7 +299,7 @@ pub trait ReduceCore<G: Scope, K: Data, V: Data, R: Semigroup> where G::Timestam
             T2::Val: Data,
             T2::R: Semigroup,
             T2::Batch: Batch,
-            <T2::Batch as Batch>::Builder: Builder<T2::Batch, Item = ((T2::Key, T2::Val), T2::Time, T2::R)>,
+            <T2::Batch as Batch>::Builder: Builder<Output=T2::Batch, Item = ((T2::Key, T2::Val), T2::Time, T2::R)>,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::R)>, &mut Vec<(T2::Val,T2::R)>)+'static
             ;
 }
@@ -318,7 +318,7 @@ where
             T2::R: Semigroup,
             T2: Trace+TraceReader<Key=K, Time=G::Timestamp>+'static,
             T2::Batch: Batch,
-            <T2::Batch as Batch>::Builder: Builder<T2::Batch, Item = ((T2::Key, T2::Val), T2::Time, T2::R)>,
+            <T2::Batch as Batch>::Builder: Builder<Output=T2::Batch, Item = ((T2::Key, T2::Val), T2::Time, T2::R)>,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::R)>, &mut Vec<(T2::Val, T2::R)>)+'static
     {
         self.arrange_by_key_named(&format!("Arrange: {}", name))
@@ -337,7 +337,7 @@ where
             T2::Val: Data,
             T2::R: Semigroup,
             T2::Batch: Batch,
-            <T2::Batch as Batch>::Builder: Builder<T2::Batch, Item = ((T2::Key, T2::Val), T2::Time, T2::R)>,
+            <T2::Batch as Batch>::Builder: Builder<Output=T2::Batch, Item = ((T2::Key, T2::Val), T2::Time, T2::R)>,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::R)>, &mut Vec<(T2::Val, T2::R)>)+'static {
 
         let mut result_trace = None;
diff --git a/src/trace/implementations/merge_batcher.rs b/src/trace/implementations/merge_batcher.rs
index ae367f3a0..9daec42b3 100644
--- a/src/trace/implementations/merge_batcher.rs
+++ b/src/trace/implementations/merge_batcher.rs
@@ -7,34 +7,26 @@ use timely::progress::frontier::Antichain;
 
 use ::difference::Semigroup;
 
-use lattice::Lattice;
-use trace::{Batch, Batcher, Builder};
+use trace::{Batcher, Builder};
+use trace::implementations::Update;
 
 /// Creates batches from unordered tuples.
-pub struct MergeBatcher<B: Batch> where B::Key: Ord, B::Val: Ord, B::Time: Ord, B::R: Semigroup {
-    sorter: MergeSorter<(B::Key, B::Val), B::Time, B::R>,
-    lower: Antichain<B::Time>,
-    frontier: Antichain<B::Time>,
-    phantom: ::std::marker::PhantomData<B>,
+pub struct MergeBatcher<U: Update> {
+    sorter: MergeSorter<(U::Key, U::Val), U::Time, U::Diff>,
+    lower: Antichain<U::Time>,
+    frontier: Antichain<U::Time>,
+    phantom: ::std::marker::PhantomData<U>,
 }
 
-impl<B> Batcher<B> for MergeBatcher<B>
-where
-    B: Batch,
-    B::Key: Ord+Clone,
-    B::Val: Ord+Clone,
-    B::Time: Lattice+timely::progress::Timestamp+Ord+Clone,
-    B::R: Semigroup,
-    B::Builder: Builder<B, Item = ((B::Key, B::Val), B::Time, B::R)>,
-{
-    type Item = ((B::Key,B::Val),B::Time,B::R);
-    type Time = B::Time;
+impl<U: Update> Batcher for MergeBatcher<U> {
+    type Item = ((U::Key,U::Val),U::Time,U::Diff);
+    type Time = U::Time;
 
     fn new() -> Self {
         MergeBatcher {
             sorter: MergeSorter::new(),
             frontier: Antichain::new(),
-            lower: Antichain::from_elem(<B::Time as timely::progress::Timestamp>::minimum()),
+            lower: Antichain::from_elem(<U::Time as timely::progress::Timestamp>::minimum()),
             phantom: ::std::marker::PhantomData,
         }
     }
@@ -61,9 +53,9 @@ where
     // which we call `lower`, by assumption that after sealing a batcher we receive no more
     // updates with times not greater or equal to `upper`.
     #[inline(never)]
-    fn seal(&mut self, upper: Antichain<B::Time>) -> B {
+    fn seal<B: Builder<Item=Self::Item, Time=Self::Time>>(&mut self, upper: Antichain<U::Time>) -> B::Output {
 
-        let mut builder = B::Builder::new();
+        let mut builder = B::new();
 
         let mut merged = Vec::new();
         self.sorter.finish_into(&mut merged);
@@ -109,18 +101,18 @@ where
         let mut buffer = Vec::new();
         self.sorter.push(&mut buffer);
         // We recycle buffers with allocations (capacity, and not zero-sized).
-        while buffer.capacity() > 0 && std::mem::size_of::<((B::Key,B::Val),B::Time,B::R)>() > 0 {
+        while buffer.capacity() > 0 && std::mem::size_of::<((U::Key,U::Val),U::Time,U::Diff)>() > 0 {
             buffer = Vec::new();
             self.sorter.push(&mut buffer);
         }
 
-        let seal = builder.done(self.lower.clone(), upper.clone(), Antichain::from_elem(<B::Time as timely::progress::Timestamp>::minimum()));
+        let seal = builder.done(self.lower.clone(), upper.clone(), Antichain::from_elem(<U::Time as timely::progress::Timestamp>::minimum()));
         self.lower = upper;
         seal
     }
 
     // the frontier of elements remaining after the most recent call to `self.seal`.
-    fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<B::Time> {
+    fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<U::Time> {
         self.frontier.borrow()
     }
 }
diff --git a/src/trace/implementations/merge_batcher_col.rs b/src/trace/implementations/merge_batcher_col.rs
index 8a316e32f..3b1d5eaec 100644
--- a/src/trace/implementations/merge_batcher_col.rs
+++ b/src/trace/implementations/merge_batcher_col.rs
@@ -8,39 +8,38 @@ use timely::progress::frontier::Antichain;
 
 use ::difference::Semigroup;
 
-use lattice::Lattice;
-use trace::{Batch, Batcher, Builder};
+use trace::{Batcher, Builder};
+use trace::implementations::Update;
 
 /// Creates batches from unordered tuples.
-pub struct ColumnatedMergeBatcher<B: Batch>
-    where
-        B::Key: Ord+Clone+Columnation,
-        B::Val: Ord+Clone+Columnation,
-        B::Time: Lattice+timely::progress::Timestamp+Ord+Clone+Columnation,
-        B::R: Semigroup+Columnation,
+pub struct ColumnatedMergeBatcher<U: Update>
+where
+    U::Key: Columnation,
+    U::Val: Columnation,
+    U::Time: Columnation,
+    U::Diff: Columnation,
 {
-    sorter: MergeSorterColumnation<(B::Key, B::Val), B::Time, B::R>,
-    lower: Antichain<B::Time>,
-    frontier: Antichain<B::Time>,
-    phantom: PhantomData<B>,
+    sorter: MergeSorterColumnation<(U::Key, U::Val), U::Time, U::Diff>,
+    lower: Antichain<U::Time>,
+    frontier: Antichain<U::Time>,
+    phantom: PhantomData<U>,
 }
 
-impl<B: Batch> Batcher<B> for ColumnatedMergeBatcher<B>
-    where
-        B::Key: Ord+Clone+Columnation+'static,
-        B::Val: Ord+Clone+Columnation+'static,
-        B::Time: Lattice+timely::progress::Timestamp+Ord+Clone+Columnation+'static,
-        B::R: Semigroup+Columnation+'static,
-        B::Builder: Builder<B, Item = ((B::Key, B::Val), B::Time, B::R)>,
+impl<U: Update> Batcher for ColumnatedMergeBatcher<U>
+where
+    U::Key: Columnation + 'static,
+    U::Val: Columnation + 'static,
+    U::Time: Columnation + 'static,
+    U::Diff: Columnation + 'static,
 {
-    type Item = ((B::Key,B::Val),B::Time,B::R);
-    type Time = B::Time;
+    type Item = ((U::Key,U::Val),U::Time,U::Diff);
+    type Time = U::Time;
 
     fn new() -> Self {
         ColumnatedMergeBatcher {
             sorter: MergeSorterColumnation::new(),
             frontier: Antichain::new(),
-            lower: Antichain::from_elem(<B::Time as timely::progress::Timestamp>::minimum()),
+            lower: Antichain::from_elem(<U::Time as timely::progress::Timestamp>::minimum()),
             phantom: PhantomData,
         }
     }
@@ -65,9 +64,9 @@ impl<B: Batch> Batcher<B> for ColumnatedMergeBatcher<B>
     // which we call `lower`, by assumption that after sealing a batcher we receive no more
     // updates with times not greater or equal to `upper`.
     #[inline]
-    fn seal(&mut self, upper: Antichain<B::Time>) -> B {
+    fn seal<B: Builder<Item=Self::Item, Time=Self::Time>>(&mut self, upper: Antichain<U::Time>) -> B::Output {
 
-        let mut builder = B::Builder::new();
+        let mut builder = B::new();
 
         let mut merged = Default::default();
         self.sorter.finish_into(&mut merged);
@@ -106,13 +105,13 @@ impl<B: Batch> Batcher<B> for ColumnatedMergeBatcher<B>
         // Drain buffers (fast reclamation).
         self.sorter.clear_stash();
 
-        let seal = builder.done(self.lower.clone(), upper.clone(), Antichain::from_elem(<B::Time as timely::progress::Timestamp>::minimum()));
+        let seal = builder.done(self.lower.clone(), upper.clone(), Antichain::from_elem(<U::Time as timely::progress::Timestamp>::minimum()));
         self.lower = upper;
         seal
     }
 
     // the frontier of elements remaining after the most recent call to `self.seal`.
-    fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<B::Time> {
+    fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<U::Time> {
         self.frontier.borrow()
     }
 }
diff --git a/src/trace/implementations/ord.rs b/src/trace/implementations/ord.rs
index 94b124b0a..3ac2946a3 100644
--- a/src/trace/implementations/ord.rs
+++ b/src/trace/implementations/ord.rs
@@ -94,8 +94,8 @@ impl<L: Layout, C> BatchReader for OrdValBatch<L, C> {
 
 impl<L: Layout> Batch for OrdValBatch<L, Vec<L::Target>>
 {
-    type Batcher = MergeBatcher<Self>;
-    type Builder = OrdValBuilder<L>;
+    type Batcher = MergeBatcher<L::Target>;
+    type Builder = OrdValBuilder<L, Vec<L::Target>>;
     type Merger = OrdValMerger<L>;
 
     fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef<<L::Target as Update>::Time>) -> Self::Merger {
@@ -111,8 +111,8 @@ where
     Self::Time: Columnation + 'static,
     Self::R: Columnation + 'static,
 {
-    type Batcher = ColumnatedMergeBatcher<Self>;
-    type Builder = OrdValBuilder<L>;
+    type Batcher = ColumnatedMergeBatcher<L::Target>;
+    type Builder = OrdValBuilder<L, TimelyStack<L::Target>>;
     type Merger = OrdValMerger<L>;
 
     fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef<<L::Target as Update>::Time>) -> Self::Merger {
@@ -337,26 +337,30 @@ impl<L: Layout, C> Cursor<OrdValBatch<L, C>> for OrdValCursor<L, C> {
 }
 
 /// A builder for creating layers from unsorted update tuples.
-pub struct OrdValBuilder<L: Layout> {
+pub struct OrdValBuilder<L: Layout, C> {
     builder: KVTDBuilder<L>,
+    phantom: PhantomData<C>,
 }
 
 
-impl<L: Layout, C> Builder<OrdValBatch<L, C>> for OrdValBuilder<L>
+impl<L: Layout, C> Builder for OrdValBuilder<L, C>
 where
     OrdValBatch<L, C>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, R=<L::Target as Update>::Diff>
 {
     type Item = ((<L::Target as Update>::Key, <L::Target as Update>::Val), <L::Target as Update>::Time, <L::Target as Update>::Diff);
     type Time = <L::Target as Update>::Time;
+    type Output = OrdValBatch<L, C>;
 
     fn new() -> Self {
         OrdValBuilder {
-            builder: <KVTDBuilder<L>>::new()
+            builder: <KVTDBuilder<L>>::new(),
+            phantom: std::marker::PhantomData,
         }
     }
     fn with_capacity(cap: usize) -> Self {
         OrdValBuilder {
-            builder: <KVTDBuilder<L> as TupleBuilder>::with_capacity(cap)
+            builder: <KVTDBuilder<L> as TupleBuilder>::with_capacity(cap),
+            phantom: std::marker::PhantomData,
         }
     }
 
@@ -411,9 +415,9 @@ impl<L: Layout, C> BatchReader for OrdKeyBatch<L, C> {
     fn description(&self) -> &Description<<L::Target as Update>::Time> { &self.desc }
 }
 
-impl<L: Layout> Batch for OrdKeyBatch<L, Vec<L::Target>> {
-    type Batcher = MergeBatcher<Self>;
-    type Builder = OrdKeyBuilder<L>;
+impl<L: Layout> Batch for OrdKeyBatch<L, Vec<L::Target>> where L::Target: Update<Val = ()> {
+    type Batcher = MergeBatcher<L::Target>;
+    type Builder = OrdKeyBuilder<L, Vec<L::Target>>;
     type Merger = OrdKeyMerger<L>;
 
     fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef<<L::Target as Update>::Time>) -> Self::Merger {
@@ -423,13 +427,13 @@ impl<L: Layout> Batch for OrdKeyBatch<L, Vec<L::Target>> {
 
 impl<L: Layout> Batch for OrdKeyBatch<L, TimelyStack<L::Target>>
 where
-    <L as Layout>::Target: Columnation + 'static,
+    <L as Layout>::Target: Update<Val = ()> + Columnation + 'static,
     Self::Key: Columnation + 'static,
     Self::Time: Columnation + 'static,
     Self::R: Columnation + 'static,
 {
-    type Batcher = ColumnatedMergeBatcher<Self>;
-    type Builder = OrdKeyBuilder<L>;
+    type Batcher = ColumnatedMergeBatcher<L::Target>;
+    type Builder = OrdKeyBuilder<L, TimelyStack<L::Target>>;
     type Merger = OrdKeyMerger<L>;
 
     fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef<<L::Target as Update>::Time>) -> Self::Merger {
@@ -636,26 +640,30 @@ impl<L: Layout, C> Cursor<OrdKeyBatch<L, C>> for OrdKeyCursor<L, C> {
 
 
 /// A builder for creating layers from unsorted update tuples.
-pub struct OrdKeyBuilder<L: Layout> {
+pub struct OrdKeyBuilder<L: Layout, C> {
     builder: KTDBuilder<L>,
+    phantom: std::marker::PhantomData<C>
 }
 
-impl<L: Layout, C> Builder<OrdKeyBatch<L, C>> for OrdKeyBuilder<L>
+impl<L: Layout, C> Builder for OrdKeyBuilder<L, C>
 where
     OrdKeyBatch<L, C>: Batch<Key=<L::Target as Update>::Key, Val=(), Time=<L::Target as Update>::Time, R=<L::Target as Update>::Diff>
 {
     type Item = ((<L::Target as Update>::Key, ()), <L::Target as Update>::Time, <L::Target as Update>::Diff);
     type Time = <L::Target as Update>::Time;
+    type Output = OrdKeyBatch<L, C>;
 
     fn new() -> Self {
         OrdKeyBuilder {
-            builder: <KTDBuilder<L>>::new()
+            builder: <KTDBuilder<L>>::new(),
+            phantom: std::marker::PhantomData,
         }
     }
 
     fn with_capacity(cap: usize) -> Self {
         OrdKeyBuilder {
-            builder: <KTDBuilder<L> as TupleBuilder>::with_capacity(cap)
+            builder: <KTDBuilder<L> as TupleBuilder>::with_capacity(cap),
+            phantom: std::marker::PhantomData,
         }
     }
 
diff --git a/src/trace/implementations/ord_neu.rs b/src/trace/implementations/ord_neu.rs
index d8dbe5cf4..694af897c 100644
--- a/src/trace/implementations/ord_neu.rs
+++ b/src/trace/implementations/ord_neu.rs
@@ -128,8 +128,8 @@ mod val_batch {
     }
 
     impl<L: Layout> Batch for OrdValBatch<L, Vec<L::Target>> {
-        type Batcher = MergeBatcher<Self>;
-        type Builder = OrdValBuilder<L>;
+        type Batcher = MergeBatcher<L::Target>;
+        type Builder = OrdValBuilder<L, Vec<L::Target>>;
         type Merger = OrdValMerger<L>;
 
         fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef<<L::Target as Update>::Time>) -> Self::Merger {
@@ -145,8 +145,8 @@ mod val_batch {
         Self::Time: Columnation + 'static,
         Self::R: Columnation + 'static,
     {
-        type Batcher = ColumnatedMergeBatcher<Self>;
-        type Builder = OrdValBuilder<L>;
+        type Batcher = ColumnatedMergeBatcher<L::Target>;
+        type Builder = OrdValBuilder<L, TimelyStack<L::Target>>;
         type Merger = OrdValMerger<L>;
 
         fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef<<L::Target as Update>::Time>) -> Self::Merger {
@@ -479,7 +479,7 @@ mod val_batch {
     }
 
     /// A builder for creating layers from unsorted update tuples.
-    pub struct OrdValBuilder<L: Layout> {
+    pub struct OrdValBuilder<L: Layout, C> {
         result: OrdValStorage<L>,
         singleton: Option<(<L::Target as Update>::Time, <L::Target as Update>::Diff)>,
         /// Counts the number of singleton optimizations we performed.
@@ -487,9 +487,11 @@ mod val_batch {
         /// This number allows us to correctly gauge the total number of updates reflected in a batch,
         /// even though `updates.len()` may be much shorter than this amount.
         singletons: usize,
+        /// Phantom marker for Rust happiness.
+        pub phantom: PhantomData<C>,
     }
 
-    impl<L: Layout> OrdValBuilder<L> {
+    impl<L: Layout, C> OrdValBuilder<L, C> {
         /// Pushes a single update, which may set `self.singleton` rather than push.
         ///
         /// This operation is meant to be equivalent to `self.results.updates.push((time, diff))`.
@@ -517,13 +519,13 @@ mod val_batch {
         }
     }
 
-    impl<L: Layout, C> Builder<OrdValBatch<L, C>> for OrdValBuilder<L>
+    impl<L: Layout, C> Builder for OrdValBuilder<L, C>
     where
         OrdValBatch<L, C>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, R=<L::Target as Update>::Diff>
     {
-
         type Item = ((<L::Target as Update>::Key, <L::Target as Update>::Val), <L::Target as Update>::Time, <L::Target as Update>::Diff);
         type Time = <L::Target as Update>::Time;
+        type Output = OrdValBatch<L, C>;
 
         fn new() -> Self { Self::with_capacity(0) }
         fn with_capacity(cap: usize) -> Self {
@@ -538,6 +540,7 @@ mod val_batch {
                 },
                 singleton: None,
                 singletons: 0,
+                phantom: std::marker::PhantomData,
             }
         }
 
diff --git a/src/trace/mod.rs b/src/trace/mod.rs
index a94db6c11..0f5bffaa3 100644
--- a/src/trace/mod.rs
+++ b/src/trace/mod.rs
@@ -280,9 +280,9 @@ where
 /// An immutable collection of updates.
 pub trait Batch : BatchReader where Self: ::std::marker::Sized {
     /// A type used to assemble batches from disordered updates.
-    type Batcher: Batcher<Self, Time = Self::Time>;
+    type Batcher: Batcher<Time = Self::Time>;
     /// A type used to assemble batches from ordered update sequences.
-    type Builder: Builder<Self, Time = Self::Time>;
+    type Builder: Builder<Item=<Self::Batcher as Batcher>::Item, Time=Self::Time, Output = Self>;
     /// A type used to progressively merge batches.
     type Merger: Merger<Self>;
 
@@ -301,7 +301,7 @@ pub trait Batch : BatchReader where Self: ::std::marker::Sized {
 }
 
 /// Functionality for collecting and batching updates.
-pub trait Batcher<Output> {
+pub trait Batcher {
     /// Type of update pushed into the batcher.
     type Item;
     /// Times at which batches are formed.
@@ -311,17 +311,19 @@ pub trait Batcher<Output> {
     /// Adds an unordered batch of elements to the batcher.
     fn push_batch(&mut self, batch: RefOrMut<Vec<Self::Item>>);
     /// Returns all updates not greater or equal to an element of `upper`.
-    fn seal(&mut self, upper: Antichain<Self::Time>) -> Output;
+    fn seal<B: Builder<Item=Self::Item, Time=Self::Time>>(&mut self, upper: Antichain<Self::Time>) -> B::Output;
     /// Returns the lower envelope of contained update times.
     fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<Self::Time>;
 }
 
 /// Functionality for building batches from ordered update sequences.
-pub trait Builder<Output> {
-    /// Type of update pushed into the builder.
+pub trait Builder {
+    /// Input item type.
     type Item;
-    /// Times at which batches are formed.
+    /// Timestamp type.
     type Time: Timestamp;
+    /// Output batch type.
+    type Output;
     
     /// Allocates an empty builder.
     fn new() -> Self;
@@ -341,7 +343,7 @@ pub trait Builder<Output> {
         for item in iter { self.push(item); }
     }
     /// Completes building and returns the batch.
-    fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Output;
+    fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Self::Output;
 }
 
 /// Represents a merge in progress.
@@ -367,10 +369,9 @@ pub trait Merger<Output: Batch> {
 pub mod rc_blanket_impls {
 
     use std::rc::Rc;
-    use timely::communication::message::RefOrMut;
 
     use timely::progress::{Antichain, frontier::AntichainRef};
-    use super::{Batch, BatchReader, Batcher, Builder, Merger, Cursor, Description};
+    use super::{Batch, BatchReader, Builder, Merger, Cursor, Description};
 
     impl<B: BatchReader> BatchReader for Rc<B> {
         type Key = B::Key;
@@ -434,36 +435,24 @@ pub mod rc_blanket_impls {
 
     /// An immutable collection of updates.
     impl<B: Batch> Batch for Rc<B> {
-        type Batcher = RcBatcher<B>;
-        type Builder = RcBuilder<B>;
+        type Batcher = B::Batcher;
+        type Builder = RcBuilder<B::Builder>;
         type Merger = RcMerger<B>;
     }
 
-    /// Wrapper type for batching reference counted batches.
-    pub struct RcBatcher<B:Batch> { batcher: B::Batcher }
-
-    /// Functionality for collecting and batching updates.
-    impl<B:Batch> Batcher<Rc<B>> for RcBatcher<B> {
-        type Item = <<B as Batch>::Batcher as Batcher<B>>::Item;
-        type Time = <<B as Batch>::Batcher as Batcher<B>>::Time;
-        fn new() -> Self { RcBatcher { batcher: <B::Batcher as Batcher<B>>::new() } }
-        fn push_batch(&mut self, batch: RefOrMut<Vec<Self::Item>>) { self.batcher.push_batch(batch) }
-        fn seal(&mut self, upper: Antichain<Self::Time>) -> Rc<B> { Rc::new(self.batcher.seal(upper)) }
-        fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<Self::Time> { self.batcher.frontier() }
-    }
-
     /// Wrapper type for building reference counted batches.
-    pub struct RcBuilder<B:Batch> { builder: B::Builder }
+    pub struct RcBuilder<B: Builder> { builder: B }
 
     /// Functionality for building batches from ordered update sequences.
-    impl<B:Batch> Builder<Rc<B>> for RcBuilder<B> {
-        type Item = <<B as Batch>::Builder as Builder<B>>::Item;
-        type Time = <<B as Batch>::Builder as Builder<B>>::Time;
-        fn new() -> Self { RcBuilder { builder: <B::Builder as Builder<B>>::new() } }
-        fn with_capacity(cap: usize) -> Self { RcBuilder { builder: <B::Builder as Builder<B>>::with_capacity(cap) } }
+    impl<B: Builder> Builder for RcBuilder<B> {
+        type Item = B::Item;
+        type Time = B::Time;
+        type Output = Rc<B::Output>;
+        fn new() -> Self { RcBuilder { builder: B::new() } }
+        fn with_capacity(cap: usize) -> Self { RcBuilder { builder: B::with_capacity(cap) } }
         fn push(&mut self, element: Self::Item) { self.builder.push(element) }
         fn copy(&mut self, element: &Self::Item) { self.builder.copy(element) }
-        fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Rc<B> { Rc::new(self.builder.done(lower, upper, since)) }
+        fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Rc<B::Output> { Rc::new(self.builder.done(lower, upper, since)) }
     }
 
     /// Wrapper type for merging reference counted batches.
@@ -485,10 +474,9 @@ pub mod abomonated_blanket_impls {
 
     use abomonation::{Abomonation, measure};
     use abomonation::abomonated::Abomonated;
-    use timely::communication::message::RefOrMut;
     use timely::progress::{Antichain, frontier::AntichainRef};
 
-    use super::{Batch, BatchReader, Batcher, Builder, Merger, Cursor, Description};
+    use super::{Batch, BatchReader, Builder, Merger, Cursor, Description};
 
     impl<B: BatchReader+Abomonation> BatchReader for Abomonated<B, Vec<u8>> {
 
@@ -553,45 +541,31 @@ pub mod abomonated_blanket_impls {
 
     /// An immutable collection of updates.
     impl<B: Batch+Abomonation> Batch for Abomonated<B, Vec<u8>> {
-        type Batcher = AbomonatedBatcher<B>;
-        type Builder = AbomonatedBuilder<B>;
+        type Batcher = B::Batcher;
+        type Builder = AbomonatedBuilder<B::Builder>;
         type Merger = AbomonatedMerger<B>;
     }
 
-    /// Wrapper type for batching reference counted batches.
-    pub struct AbomonatedBatcher<B:Batch> { batcher: B::Batcher }
-
-    /// Functionality for collecting and batching updates.
-    impl<B:Batch+Abomonation> Batcher<Abomonated<B,Vec<u8>>> for AbomonatedBatcher<B> {
-        type Item = <<B as Batch>::Batcher as Batcher<B>>::Item;
-        type Time = <<B as Batch>::Batcher as Batcher<B>>::Time;
-        fn new() -> Self { AbomonatedBatcher { batcher: <B::Batcher as Batcher<B>>::new() } }
-        fn push_batch(&mut self, batch: RefOrMut<Vec<Self::Item>>) { self.batcher.push_batch(batch) }
-        fn seal(&mut self, upper: Antichain<Self::Time>) -> Abomonated<B, Vec<u8>> {
-            let batch = self.batcher.seal(upper);
-            let mut bytes = Vec::with_capacity(measure(&batch));
-            unsafe { abomonation::encode(&batch, &mut bytes).unwrap() };
-            unsafe { Abomonated::<B,_>::new(bytes).unwrap() }
-        }
-        fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<Self::Time> { self.batcher.frontier() }
-    }
-
     /// Wrapper type for building reference counted batches.
-    pub struct AbomonatedBuilder<B:Batch> { builder: B::Builder }
+    pub struct AbomonatedBuilder<B: Builder> { builder: B }
 
     /// Functionality for building batches from ordered update sequences.
-    impl<B:Batch+Abomonation> Builder<Abomonated<B,Vec<u8>>> for AbomonatedBuilder<B> {
-        type Item = <<B as Batch>::Builder as Builder<B>>::Item;
-        type Time = <<B as Batch>::Builder as Builder<B>>::Time;
-        fn new() -> Self { AbomonatedBuilder { builder: <B::Builder as Builder<B>>::new() } }
-        fn with_capacity(cap: usize) -> Self { AbomonatedBuilder { builder: <B::Builder as Builder<B>>::with_capacity(cap) } }
+    impl<B: Builder> Builder for AbomonatedBuilder<B>
+    where
+        B::Output: Abomonation,
+    {
+        type Item = B::Item;
+        type Time = B::Time;
+        type Output = Abomonated<B::Output, Vec<u8>>;
+        fn new() -> Self { AbomonatedBuilder { builder: B::new() } }
+        fn with_capacity(cap: usize) -> Self { AbomonatedBuilder { builder: B::with_capacity(cap) } }
         fn push(&mut self, element: Self::Item) { self.builder.push(element) }
         fn copy(&mut self, element: &Self::Item) { self.builder.copy(element) }
-        fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Abomonated<B, Vec<u8>> {
+        fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Self::Output {
             let batch = self.builder.done(lower, upper, since);
             let mut bytes = Vec::with_capacity(measure(&batch));
             unsafe { abomonation::encode(&batch, &mut bytes).unwrap() };
-            unsafe { Abomonated::<B,_>::new(bytes).unwrap() }
+            unsafe { Abomonated::<B::Output,_>::new(bytes).unwrap() }
         }
     }
 
diff --git a/tests/trace.rs b/tests/trace.rs
index d29999af2..73e8269ba 100644
--- a/tests/trace.rs
+++ b/tests/trace.rs
@@ -9,12 +9,14 @@ use differential_dataflow::trace::{Trace, TraceReader, Batch, Batcher};
 use differential_dataflow::trace::cursor::Cursor;
 
 type IntegerTrace = ValSpine<u64, u64, usize, i64>;
+type IntegerBatch = <IntegerTrace as TraceReader>::Batch;
+type IntegerBuilder = <IntegerBatch as Batch>::Builder;
 
 fn get_trace() -> ValSpine<u64, u64, usize, i64> {
     let op_info = OperatorInfo::new(0, 0, &[]);
     let mut trace = IntegerTrace::new(op_info, None, None);
     {
-        let mut batcher = <<IntegerTrace as TraceReader>::Batch as Batch>::Batcher::new();
+        let mut batcher = <IntegerBatch as Batch>::Batcher::new();
 
         use timely::communication::message::RefOrMut;
         batcher.push_batch(RefOrMut::Mut(&mut vec![
@@ -24,7 +26,7 @@ fn get_trace() -> ValSpine<u64, u64, usize, i64> {
         ]));
 
         let batch_ts = &[1, 2, 3];
-        let batches = batch_ts.iter().map(move |i| batcher.seal(Antichain::from_elem(*i)));
+        let batches = batch_ts.iter().map(move |i| batcher.seal::<IntegerBuilder>(Antichain::from_elem(*i)));
         for b in batches {
             trace.insert(b);
         }