diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs
index d566c2424..d85d572f5 100644
--- a/differential-dataflow/examples/columnar/main.rs
+++ b/differential-dataflow/examples/columnar/main.rs
@@ -98,7 +98,7 @@ mod reachability {
     use differential_dataflow::operators::arrange::arrangement::arrange_core;
     use differential_dataflow::operators::join::join_traces;
 
-    use differential_dataflow::columnar::trace::{Batcher as ValBatcher, Builder as ValBuilder, Chunker as ValChunker, Spine as ValSpine};
+    use differential_dataflow::columnar::trace::{Batcher as ValBatcher, Builder as ValBuilder, Spine as ValSpine};
     use differential_dataflow::columnar::collection::{Builder as ValColBuilder, Pact as ValPact, RecordedUpdates, as_recorded_updates};
 
     type Node = u32;
@@ -128,16 +128,12 @@ mod reachability {
             let reach_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 };
 
             let edges_arr = arrange_core::<_, _,
-                ValChunker<(Node, Node, IterTime, Diff)>,
                 _,
-                ValBuilder<Node, Node, IterTime, Diff>,
                 ValSpine<Node, Node, IterTime, Diff>,
             >(edges_inner.inner, edges_pact, "Edges", ValBatcher::new);
 
             let reach_arr = arrange_core::<_, _,
-                ValChunker<(Node, (), IterTime, Diff)>,
                 _,
-                ValBuilder<Node, (), IterTime, Diff>,
                 ValSpine<Node, (), IterTime, Diff>,
             >(reach.inner, reach_pact, "Reach", ValBatcher::new);
 
@@ -161,9 +157,7 @@ mod reachability {
             // Arrange for reduce.
             let combined_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 };
             let combined_arr = arrange_core::<_, _,
-                ValChunker<(Node, (), IterTime, Diff)>,
                 _,
-                ValBuilder<Node, (), IterTime, Diff>,
                 ValSpine<Node, (), IterTime, Diff>,
             >(combined.inner, combined_pact, "Combined", ValBatcher::new);
 
diff --git a/differential-dataflow/examples/columnar_spill.rs b/differential-dataflow/examples/columnar_spill.rs
index c5619a08b..611b8602a 100644
--- a/differential-dataflow/examples/columnar_spill.rs
+++ b/differential-dataflow/examples/columnar_spill.rs
@@ -50,7 +50,7 @@ fn reset_stats() {
     BYTES_COMPRESSED.store(0, Ordering::Relaxed);
 }
 
-use differential_dataflow::columnar::trace::{Batcher as ValBatcher, Builder as ValBuilder, Chunker as ValChunker, Spine as ValSpine};
+use differential_dataflow::columnar::trace::{Batcher as ValBatcher, Spine as ValSpine};
 use differential_dataflow::columnar::collection::Builder as ValColBuilder;
 use differential_dataflow::columnar::trace::spill::{self, BytesSource, BytesStore, SpillStats};
 use differential_dataflow::columnar::updates::{Updates, UpdatesTyped};
@@ -238,9 +238,7 @@ fn run_timely_dataflow(times: u64, keys_per_time: u64, workers: usize, sample_se
             let stream = scope.input_from(&mut input);
             let arranged = arrange_core::<
                 _, _,
-                ValChunker<(u64, u64, u64, i64)>,
                 _,
-                ValBuilder<u64, u64, u64, i64>,
                 ValSpine<u64, u64, u64, i64>,
             >(stream, Pipeline, "ColumnarSpillArrange", ValBatcher::new);
             arranged.stream.probe_with(&mut probe);
diff --git a/differential-dataflow/examples/spines.rs b/differential-dataflow/examples/spines.rs
index e7bdb630f..f6606afdd 100644
--- a/differential-dataflow/examples/spines.rs
+++ b/differential-dataflow/examples/spines.rs
@@ -55,21 +55,21 @@ fn main() {
 
             match mode.as_str() {
                 "key" => {
-                    use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatcher, VecOrdKeyBuilder, OrdKeySpine};
+                    use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatcher, OrdKeySpine};
                     let (data_input, data) = scope.new_collection::<String, isize>();
                     let (keys_input, keys) = scope.new_collection::<String, isize>();
-                    let data = data.map(|k| (k, ())).arrange::<_, VecOrdKeyBuilder<String,_,isize>, OrdKeySpine<String,_,isize>>(OrdKeyBatcher::new);
-                    let keys = keys.map(|k| (k, ())).arrange::<_, VecOrdKeyBuilder<String,_,isize>, OrdKeySpine<String,_,isize>>(OrdKeyBatcher::new);
+                    let data = data.map(|k| (k, ())).arrange::<_, OrdKeySpine<String,_,isize>>(OrdKeyBatcher::new);
+                    let keys = keys.map(|k| (k, ())).arrange::<_, OrdKeySpine<String,_,isize>>(OrdKeyBatcher::new);
                     keys.join_core(data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                     Workload { data_input, keys_input }
                 },
                 "val" => {
-                    use differential_dataflow::trace::implementations::ord_neu::{OrdValBatcher, VecOrdValBuilder, OrdValSpine};
+                    use differential_dataflow::trace::implementations::ord_neu::{OrdValBatcher, OrdValSpine};
                     let (data_input, data) = scope.new_collection::<String, isize>();
                     let (keys_input, keys) = scope.new_collection::<String, isize>();
-                    let data = data.map(|x| (x, ())).arrange::<_, VecOrdValBuilder<String,(),_,isize>, OrdValSpine<String,(),_,isize>>(OrdValBatcher::new);
-                    let keys = keys.map(|x| (x, ())).arrange::<_, VecOrdValBuilder<String,(),_,isize>, OrdValSpine<String,(),_,isize>>(OrdValBatcher::new);
+                    let data = data.map(|x| (x, ())).arrange::<_, OrdValSpine<String,(),_,isize>>(OrdValBatcher::new);
+                    let keys = keys.map(|x| (x, ())).arrange::<_, OrdValSpine<String,(),_,isize>>(OrdValBatcher::new);
                     keys.join_core(data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                     Workload { data_input, keys_input }
@@ -79,7 +79,8 @@ fn main() {
                     // the same generic `Chunk` harness as `vec` via a
                     // `ContainerChunker<ColChunk>`.
                     use differential_dataflow::Hashable;
-                    use differential_dataflow::columnar::trace::{Batcher, Builder, Spine, ColChunk};
+                    use differential_dataflow::columnar::trace::{Spine, ColChunk};
+                    use differential_dataflow::trace::chunk::ChunkBatcher;
                     use differential_dataflow::trace::implementations::chunker::ContainerChunker;
                     use differential_dataflow::operators::arrange::arrangement::arrange_core;
                     use timely::dataflow::channels::pact::Exchange;
@@ -89,13 +90,12 @@ fn main() {
                     let data = data.map(|x| (x, ()));
                     let keys = keys.map(|x| (x, ()));
 
-                    type Ba = Batcher<String, (), u64, isize>;
-                    type Bu = Builder<String, (), u64, isize>;
-                    type Sp = Spine<String, (), u64, isize>;
                     type Chu = ContainerChunker<ColChunk<(String, (), u64, isize)>>;
+                    type Ba = ChunkBatcher<Chu, ColChunk<(String, (), u64, isize)>>;
+                    type Sp = Spine<String, (), u64, isize>;
                     let exchange = || Exchange::new(|u: &((String, ()), u64, isize)| (u.0).0.hashed().into());
-                    let data = arrange_core::<_, _, Chu, Ba, Bu, Sp>(data.inner, exchange(), "DataArrange", Ba::new);
-                    let keys = arrange_core::<_, _, Chu, Ba, Bu, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new);
+                    let data = arrange_core::<_, _, Ba, Sp>(data.inner, exchange(), "DataArrange", Ba::new);
+                    let keys = arrange_core::<_, _, Ba, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new);
                     // `ColChunk`'s cursor yields `Val = columnar::Ref<()> = ()`, not `&()`.
                     keys.join_core(data, |_k, _, _| Option::<()>::None)
                         .probe_with(&mut probe);
@@ -106,8 +106,7 @@ fn main() {
                     // insert allocates a `String`) but arranged through the `Chunk`
                     // harness via a `ContainerChunker<VecChunk>`.
                     use differential_dataflow::Hashable;
-                    use differential_dataflow::trace::chunk::vec::{ChunkBatcher, ChunkBuilder, ChunkSpine, VecChunk};
-                    use differential_dataflow::trace::implementations::chunker::ContainerChunker;
+                    use differential_dataflow::trace::chunk::vec::{ChunkBatcher, ChunkSpine};
                     use differential_dataflow::operators::arrange::arrangement::arrange_core;
                     use timely::dataflow::channels::pact::Exchange;
 
@@ -117,12 +116,10 @@ fn main() {
                     let keys = keys.map(|x| (x, ()));
 
                     type Ba = ChunkBatcher<String, (), u64, isize>;
-                    type Bu = ChunkBuilder<String, (), u64, isize>;
                     type Sp = ChunkSpine<String, (), u64, isize>;
-                    type Chu = ContainerChunker<VecChunk<String, (), u64, isize>>;
                     let exchange = || Exchange::new(|u: &((String, ()), u64, isize)| (u.0).0.hashed().into());
-                    let data = arrange_core::<_, _, Chu, Ba, Bu, Sp>(data.inner, exchange(), "DataArrange", Ba::new);
-                    let keys = arrange_core::<_, _, Chu, Ba, Bu, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new);
+                    let data = arrange_core::<_, _, Ba, Sp>(data.inner, exchange(), "DataArrange", Ba::new);
+                    let keys = arrange_core::<_, _, Ba, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new);
                     keys.join_core(data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                     Workload { data_input, keys_input }
diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs
index a77b05267..abe41b3fc 100644
--- a/differential-dataflow/src/collection.rs
+++ b/differential-dataflow/src/collection.rs
@@ -957,22 +957,21 @@ pub mod vec {
         /// });
         /// ```
         pub fn consolidate(self) -> Self {
-            use crate::trace::implementations::{KeyBatcher, KeyBuilder, KeySpine};
-            self.consolidate_named::<_,KeyBuilder<_,_,_>, KeySpine<_,_,_>,_>("Consolidate", KeyBatcher::new, |key,&()| key.clone())
+            use crate::trace::implementations::{KeyBatcher, KeySpine};
+            self.consolidate_named::<_, KeySpine<_,_,_>,_>("Consolidate", KeyBatcher::new, |key,&()| key.clone())
         }
 
         /// As `consolidate` but with the ability to name the operator, specify the trace type,
         /// and provide the function `reify` to produce owned keys and values..
-        pub fn consolidate_named<Ba, Bu, Tr, F>(self, name: &str, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba, reify: F) -> Self
+        pub fn consolidate_named<Ba, Tr, F>(self, name: &str, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba, reify: F) -> Self
         where
-            Ba: crate::trace::Batcher<T, Vec<((D, ()), T, R)>, Vec<Vec<((D, ()), T, R)>>> + 'static,
+            Ba: crate::trace::Batcher<Vec<((D, ()), T, R)>, Time = T, Output: Into<Tr::Batch>> + '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: Into<Tr::Batch>>,
             F: Fn(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D + 'static,
         {
             self.map(|k| (k, ()))
-                .arrange_named::<Ba, Bu, Tr>(name, batcher)
+                .arrange_named::<Ba, Tr>(name, batcher)
                 .as_collection(reify)
         }
 
@@ -1020,9 +1019,8 @@ pub mod vec {
         }
     }
 
-    use crate::trace::implementations::{ValSpine, ValBatcher, ValBuilder};
-    use crate::trace::implementations::{KeySpine, KeyBatcher, KeyBuilder};
-    use crate::trace::implementations::ContainerChunker;
+    use crate::trace::implementations::{ValSpine, ValBatcher};
+    use crate::trace::implementations::{KeySpine, KeyBatcher};
     impl<'scope, T, K, V, R> Collection<'scope, T, (K, V), R>
     where
         T: Timestamp + Lattice,
@@ -1036,24 +1034,22 @@ pub mod vec {
         /// same-type containers. For chunker setups that convert between container types (e.g.
         /// columnar layouts), call [`arrange_core`](crate::operators::arrange::arrangement::arrange_core)
         /// directly.
-        pub fn arrange<Ba, Bu, Tr>(self, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba) -> Arranged<'scope, TraceAgent<Tr>>
+        pub fn arrange<Ba, Tr>(self, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba) -> Arranged<'scope, TraceAgent<Tr>>
         where
-            Ba: crate::trace::Batcher<T, Vec<((K, V), T, R)>, Vec<Bu::Input>> + 'static,
-            Bu: crate::trace::Builder<Time=T, Output: Into<Tr::Batch>>,
+            Ba: crate::trace::Batcher<Vec<((K, V), T, R)>, Time = T, Output: Into<Tr::Batch>> + 'static,
             Tr: crate::trace::Trace<Time=T> + 'static,
         {
-            self.arrange_named::<Ba, Bu, Tr>("Arrange", batcher)
+            self.arrange_named::<Ba, Tr>("Arrange", batcher)
         }
 
         /// As [`Collection::arrange`] but with the ability to name the operator.
-        pub fn arrange_named<Ba, Bu, Tr>(self, name: &str, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba) -> Arranged<'scope, TraceAgent<Tr>>
+        pub fn arrange_named<Ba, Tr>(self, name: &str, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba) -> Arranged<'scope, TraceAgent<Tr>>
         where
-            Ba: crate::trace::Batcher<T, Vec<((K, V), T, R)>, Vec<Bu::Input>> + 'static,
-            Bu: crate::trace::Builder<Time=T, Output: Into<Tr::Batch>>,
+            Ba: crate::trace::Batcher<Vec<((K, V), T, R)>, Time = T, Output: Into<Tr::Batch>> + 'static,
             Tr: crate::trace::Trace<Time=T> + 'static,
         {
             let exchange = timely::dataflow::channels::pact::Exchange::new(move |update: &((K,V),T,R)| (update.0).0.hashed().into());
-            crate::operators::arrange::arrangement::arrange_core::<_, _, ContainerChunker<Vec<((K, V), T, R)>>, Ba, Bu, _>(self.inner, exchange, name, batcher)
+            crate::operators::arrange::arrangement::arrange_core::<_, _, Ba, _>(self.inner, exchange, name, batcher)
         }
     }
 
@@ -1072,7 +1068,7 @@ pub mod vec {
 
         /// As `arrange_by_key` but with the ability to name the arrangement.
         pub fn arrange_by_key_named(self, name: &str) -> Arranged<'scope, TraceAgent<ValSpine<K, V, T, R>>> {
-            self.arrange_named::<_,ValBuilder<_,_,_,_>,_>(name, ValBatcher::new)
+            self.arrange_named::<_,_>(name, ValBatcher::new)
         }
     }
 
@@ -1092,7 +1088,7 @@ pub mod vec {
         /// As `arrange_by_self` but with the ability to name the arrangement.
         pub fn arrange_by_self_named(self, name: &str) -> Arranged<'scope, TraceAgent<KeySpine<K, T, R>>> {
             self.map(|k| (k, ()))
-                .arrange_named::<_,KeyBuilder<_,_,_>,_>(name, KeyBatcher::new)
+                .arrange_named::<_,_>(name, KeyBatcher::new)
         }
     }
 
diff --git a/differential-dataflow/src/columnar/trace/mod.rs b/differential-dataflow/src/columnar/trace/mod.rs
index eb2e35740..5b6a987e3 100644
--- a/differential-dataflow/src/columnar/trace/mod.rs
+++ b/differential-dataflow/src/columnar/trace/mod.rs
@@ -21,7 +21,7 @@ pub use chunk::ColChunk;
 /// The columnar trace: a spine of `Rc`-shared [`ColChunk`] batches.
 pub type Spine<K, V, T, R> = crate::trace::chunk::ChunkSpine<ColChunk<(K, V, T, R)>>;
 /// The columnar merge batcher (the chunk harness's `MergeBatcher` over `ColChunk`).
-pub type Batcher<K, V, T, R> = crate::trace::chunk::ChunkBatcher<ColChunk<(K, V, T, R)>>;
+pub type Batcher<K, V, T, R> = crate::trace::chunk::ChunkBatcher<Chunker<(K, V, T, R)>, ColChunk<(K, V, T, R)>>;
 /// The columnar batch builder.
 pub type Builder<K, V, T, R> = crate::trace::chunk::ChunkBuilder<ColChunk<(K, V, T, R)>>;
 /// The input chunker: melds `RecordedUpdates<U>` streams into `ColChunk<U>` batches.
diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index 3d1d9264e..c553aeac5 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -24,7 +24,7 @@ use timely::dataflow::operators::generic::Operator;
 use timely::dataflow::channels::pact::{ParallelizationContract, Pipeline};
 use timely::progress::Timestamp;
 use timely::progress::Antichain;
-use timely::container::{ContainerBuilder, PushInto};
+use timely::container::PushInto;
 use timely::dataflow::operators::{Capability, CapabilitySet};
 use timely::progress::Stamp;
 
@@ -303,7 +303,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
 /// 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<'scope, P, C, Chu, Ba, Bu, Tr>(
+pub fn arrange_core<'scope, P, C, Ba, Tr>(
     stream: Stream<'scope, Tr::Time, C>,
     pact: P,
     name: &str,
@@ -312,9 +312,7 @@ pub fn arrange_core<'scope, P, C, Chu, Ba, Bu, Tr>(
 where
     C: Container + Clone + 'static,
     P: ParallelizationContract<Tr::Time, C>,
-    Chu: ContainerBuilder + for<'a> PushInto<&'a mut C> + 'static,
-    Ba: Batcher<Tr::Time, Chu::Container, Vec<Bu::Input>> + 'static,
-    Bu: Builder<Time=Tr::Time, Output: Into<Tr::Batch>>,
+    Ba: Batcher<C, Time = Tr::Time, Output: Into<Tr::Batch>> + 'static,
     Tr: Trace+'static,
 {
     // The `Arrange` operator is tasked with reacting to an advancing input
@@ -363,8 +361,6 @@ where
         // Initialize to the minimal input frontier.
         let mut prev_frontier = Antichain::from_elem(Tr::Time::minimum());
 
-        let mut chunker = Chu::default();
-
         move |(input, frontier), output| {
 
             // As we receive data, we need to (i) stash the data and (ii) keep *enough* capabilities.
@@ -375,10 +371,7 @@ where
                 for capability in cap.retain_stamp(0).iter() {
                     capabilities.insert(capability.clone());
                 }
-                chunker.push_into(data);
-                while let Some(chunk) = chunker.extract() {
-                    batcher.insert(chunk);
-                }
+                batcher.insert(data);
             });
 
             // The frontier may have advanced by multiple elements, which is an issue because
@@ -393,13 +386,6 @@ where
             // frontier isn't equal to the previous. It is only in this case that we have any
             // data processing to do.
             if prev_frontier.borrow() != frontier.frontier() {
-                // Flush any data the chunker is still accumulating into the batcher before we
-                // seal. The batcher only sees chunks the chunker has emitted; without this drain
-                // a partial final chunk would never reach the batcher.
-                while let Some(chunk) = chunker.finish() {
-                    batcher.insert(chunk);
-                }
-
                 // There are two cases to handle with some care:
                 //
                 // 1. If any held capabilities are not in advance of the new input frontier,
@@ -433,9 +419,9 @@ where
                         frontier.frontier().to_owned(),
                         Antichain::from_elem(Tr::Time::minimum()),
                     );
-                    let (chain, retained) = batcher.extract(frontier.frontier());
+                    let (extracted, retained) = batcher.extract(frontier.frontier());
 
-                    let batch = trace::Span::new(description, chain.and_then(|mut chain| Bu::seal(&mut chain)).map(Into::into));
+                    let batch = trace::Span::new(description, extracted.map(Into::into));
 
                     let stamp = retired.iter().map(|c| c.time().clone()).collect::<Stamp<_>>();
                     writer.insert(batch.clone(), stamp);
@@ -444,11 +430,10 @@ where
                     output.session(&retired).give(batch);
 
                     // Having extracted and sent the batch of updates not in advance of the input
-                    // frontier, we should downgrade all capabilities to match the batcher's lower
-                    // update frontier.
+                    // frontier, we downgrade all capabilities to match the batcher's lower update
+                    // frontier.
                     // This may involve discarding capabilities, which is fine as any new updates arrive
                     // in messages with new capabilities.
-
                     let mut new_capabilities = Antichain::new();
                     for time in retained.iter() {
                         if let Some(capability) = capabilities.elements().iter().find(|c| c.time().less_equal(time)) {
@@ -458,7 +443,6 @@ where
                             panic!("failed to find capability");
                         }
                     }
-
                     capabilities = new_capabilities;
                 }
                 else {
diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs
index b1aecc244..ccc126b37 100644
--- a/differential-dataflow/src/trace/chunk/mod.rs
+++ b/differential-dataflow/src/trace/chunk/mod.rs
@@ -277,7 +277,7 @@ where
 /// using [`Chunk::extract`]. The batcher consolidates equal `(data, time)` updates
 /// but does *not* advance times — time advancement is advance's job, handled later in
 /// the trace. Both settle their output, since the batcher's chains want to be graded.
-pub type ChunkBatcher<C> = crate::trace::implementations::merge_batcher::MergeBatcher<ChunkMerger<C>>;
+pub type ChunkBatcher<Chu, C> = crate::trace::implementations::merge_batcher::MergeBatcher<Chu, ChunkMerger<C>, ChunkBuilder<C>>;
 
 /// A spine of `Rc`-shared [`ChunkBatch`]s of type `C`: the trace type for `arrange`.
 pub type ChunkSpine<C> = crate::trace::implementations::spine_fueled::Spine<std::rc::Rc<ChunkBatch<C>>>;
diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs
index 315a96138..df9c6c51e 100644
--- a/differential-dataflow/src/trace/chunk/vec.rs
+++ b/differential-dataflow/src/trace/chunk/vec.rs
@@ -53,7 +53,7 @@ impl<K, V, T, R> Default for VecChunk<K, V, T, R> {
 pub type ChunkSpine<K, V, T, R> = super::ChunkSpine<VecChunk<K, V, T, R>>;
 /// Merge batcher over `VecChunk`s; a `ContainerChunker<VecChunk>` at the
 /// `arrange_core` callsite forms the chunks it merges (via the container traits below).
-pub type ChunkBatcher<K, V, T, R> = super::ChunkBatcher<VecChunk<K, V, T, R>>;
+pub type ChunkBatcher<K, V, T, R> = super::ChunkBatcher<crate::trace::implementations::ContainerChunker<VecChunk<K, V, T, R>>, VecChunk<K, V, T, R>>;
 /// Batch builder.
 pub type ChunkBuilder<K, V, T, R> = super::ChunkBuilder<VecChunk<K, V, T, R>>;
 
diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs
index 1606a15cc..b668d32d5 100644
--- a/differential-dataflow/src/trace/implementations/merge_batcher.rs
+++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs
@@ -4,18 +4,25 @@
 //! hooks for manipulating sorted "chains" of chunks as needed by the merge batcher: merging
 //! chunks and also splitting them apart based on time.
 //!
-//! Callers feed already-chunked, sorted-and-consolidated input into the batcher via [`Batcher::insert`].
-//! Forming such chunks from raw data is the responsibility of the caller (typically a chunker
-//! living in the surrounding dataflow operator).
+//! Raw input containers are fed to the batcher via [`Batcher::insert`], which chunks them with
+//! its `Chu` before merging: forming sorted, consolidated chunks is the first stage of the
+//! batcher's own work rather than something a caller arranges.
 
+use timely::container::{ContainerBuilder, PushInto};
 use timely::progress::frontier::AntichainRef;
 use timely::progress::{frontier::Antichain, Timestamp};
 
 use crate::logging::{BatcherEvent, Logger};
-use crate::trace::Batcher;
+use crate::trace::{Batcher, Builder};
 
 /// Creates batches from chunks of sorted, consolidated tuples.
-pub struct MergeBatcher<M: Merger> {
+///
+/// Chunking input is `Chu`'s business, merging chunks is `M`'s, and building the extracted chain
+/// into a batch is `Bu`'s; the batcher's own work is the geometric ladder of chains and the
+/// carve-by-frontier.
+pub struct MergeBatcher<Chu, M: Merger, Bu> {
+    /// Melds input containers into sorted, consolidated chunks.
+    chunker: Chu,
     /// Sorted, consolidated chains, each paired with its cached summed update count.
     ///
     /// The cached count is the chain's *merge weight*: the geometric ladder weighs
@@ -34,21 +41,37 @@ pub struct MergeBatcher<M: Merger> {
     logger: Option<Logger>,
     /// Timely operator ID.
     operator_id: usize,
+    /// Seals each extracted chain into a batch.
+    builder: std::marker::PhantomData<Bu>,
 }
 
-impl<M> Batcher<M::Time, M::Chunk, Vec<M::Chunk>> for MergeBatcher<M>
+impl<C, Chu, M, Bu> Batcher<C> for MergeBatcher<Chu, M, Bu>
 where
     M: Merger<Time: Timestamp>,
+    Chu: ContainerBuilder<Container = M::Chunk> + for<'a> PushInto<&'a mut C>,
+    Bu: Builder<Input = M::Chunk>,
 {
-    fn insert(&mut self, chunk: &mut M::Chunk) {
-        self.insert_chain(vec![std::mem::take(chunk)]);
+    type Time = M::Time;
+    type Output = Bu::Output;
+
+    fn insert(&mut self, container: &mut C) {
+        self.chunker.push_into(container);
+        while let Some(chunk) = self.chunker.extract().map(std::mem::take) {
+            self.insert_chain(vec![chunk]);
+        }
     }
 
     // Extraction means finding those updates with times not greater or equal to any time in
     // `upper`. All updates must have time greater or equal to the previously used `upper`, by
     // assumption that after extracting from a batcher we receive no more updates with times not
     // greater or equal to `upper`.
-    fn extract<'a>(&'a mut self, upper: AntichainRef<'_, M::Time>) -> (Option<Vec<M::Chunk>>, AntichainRef<'a, M::Time>) {
+    fn extract<'a>(&'a mut self, upper: AntichainRef<'_, M::Time>) -> (Option<Bu::Output>, AntichainRef<'a, M::Time>) {
+        // Flush whatever the chunker is still accumulating: a partial final chunk would
+        // otherwise never reach the merge ladder.
+        while let Some(chunk) = self.chunker.finish().map(std::mem::take) {
+            self.insert_chain(vec![chunk]);
+        }
+
         // Merge all remaining chains into a single chain.
         while self.chains.len() > 1 {
             let list1 = self.chain_pop().unwrap();
@@ -71,12 +94,11 @@ where
 
         self.stash.clear();
 
-        let readied = if readied.is_empty() { None } else { Some(readied) };
-        (readied, self.frontier.borrow())
+        (Bu::seal(&mut readied), self.frontier.borrow())
     }
 }
 
-impl<M: Merger> MergeBatcher<M> {
+impl<Chu: Default, M: Merger, Bu> MergeBatcher<Chu, M, Bu> {
     /// Allocates a new empty batcher.
     ///
     /// The logger and operator identifier are used to report the batcher's memory footprint,
@@ -86,12 +108,16 @@ impl<M: Merger> MergeBatcher<M> {
             logger,
             operator_id,
             merger: M::default(),
+            chunker: Chu::default(),
             chains: Vec::new(),
             stash: Vec::new(),
             frontier: Antichain::new(),
+            builder: std::marker::PhantomData,
         }
     }
+}
 
+impl<Chu, M: Merger, Bu> MergeBatcher<Chu, M, Bu> {
     /// Insert a chain and maintain chain properties: Chains are geometrically sized
     /// (by summed updates) and ordered by decreasing update weight.
     fn insert_chain(&mut self, chain: Vec<M::Chunk>) {
@@ -166,7 +192,7 @@ impl<M: Merger> MergeBatcher<M> {
     }
 }
 
-impl<M: Merger> Drop for MergeBatcher<M> {
+impl<Chu, M: Merger, Bu> Drop for MergeBatcher<Chu, M, Bu> {
     fn drop(&mut self) {
         // Cleanup chain to retract accounting information.
         while self.chain_pop().is_some() {}
@@ -375,8 +401,11 @@ mod test {
     use crate::trace::Batcher;
     use super::MergeBatcher;
     use super::vec::VecMerger;
+    use crate::trace::implementations::ord_neu::VecOrdKeyBuilder;
+    use crate::trace::implementations::chunker::ContainerChunker;
 
-    type Bt = MergeBatcher<VecMerger<u64, u64, i64>>;
+    type In = Vec<((u64, ()), u64, i64)>;
+    type Bt = MergeBatcher<ContainerChunker<In>, VecMerger<(u64, ()), u64, i64>, VecOrdKeyBuilder<u64, u64, i64>>;
 
     /// The sealed frontier must reflect the POST-CONSOLIDATION set of distinct kept times:
     /// two chains carry cancelling updates at a kept time (`t=5`), plus a survivor at a later
@@ -386,9 +415,9 @@ mod test {
     #[test]
     fn frontier_is_post_consolidation() {
         let mut b = Bt::new(None, 0);
-        b.chain_push(vec![vec![(100u64, 5u64, 1i64), (200u64, 7u64, 1i64)]]);
-        b.chain_push(vec![vec![(100u64, 5u64, -1i64)]]);
-        let (_, retained) = b.extract(Antichain::from_elem(3).borrow());
+        b.chain_push(vec![vec![((100u64, ()), 5u64, 1i64), ((200u64, ()), 7u64, 1i64)]]);
+        b.chain_push(vec![vec![((100u64, ()), 5u64, -1i64)]]);
+        let (_, retained) = Batcher::<In>::extract(&mut b, Antichain::from_elem(3).borrow());
         let got: Vec<u64> = retained.iter().cloned().collect();
         assert_eq!(got, vec![7u64],
             "frontier held a capability at t=5, which consolidates to zero (got {got:?})");
@@ -398,9 +427,9 @@ mod test {
     #[test]
     fn frontier_survivor_minimum() {
         let mut b = Bt::new(None, 0);
-        b.chain_push(vec![vec![(100u64, 5u64, 1i64)]]);
-        b.chain_push(vec![vec![(200u64, 7u64, 1i64)]]);
-        let (_, retained) = b.extract(Antichain::from_elem(3).borrow());
+        b.chain_push(vec![vec![((100u64, ()), 5u64, 1i64)]]);
+        b.chain_push(vec![vec![((200u64, ()), 7u64, 1i64)]]);
+        let (_, retained) = Batcher::<In>::extract(&mut b, Antichain::from_elem(3).borrow());
         let got: Vec<u64> = retained.iter().cloned().collect();
         assert_eq!(got, vec![5u64]);
     }
diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs
index 81e396b9e..89ed2d15f 100644
--- a/differential-dataflow/src/trace/implementations/ord_neu.rs
+++ b/differential-dataflow/src/trace/implementations/ord_neu.rs
@@ -11,6 +11,7 @@
 use std::rc::Rc;
 
 use crate::trace::implementations::spine_fueled::Spine;
+use crate::trace::implementations::chunker::ContainerChunker;
 use crate::trace::implementations::merge_batcher::MergeBatcher;
 use crate::trace::implementations::merge_batcher::vec::VecMerger;
 
@@ -22,14 +23,14 @@ 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)>>>>;
 /// A batcher using ordered lists.
-pub type OrdValBatcher<K, V, T, R> = MergeBatcher<VecMerger<(K, V), T, R>>;
+pub type OrdValBatcher<K, V, T, R> = MergeBatcher<ContainerChunker<Vec<((K, V), T, R)>>, VecMerger<(K, V), T, R>, VecOrdValBuilder<K, V, T, R>>;
 /// A builder using ordered lists.
 pub type VecOrdValBuilder<K, V, T, R> = OrdValBuilder<Vector<((K,V),T,R)>, Vec<((K,V),T,R)>>;
 
 /// A trace implementation using a spine of ordered lists.
 pub type OrdKeySpine<K, T, R> = Spine<Rc<OrdKeyBatch<Vector<((K,()),T,R)>>>>;
 /// A batcher for ordered lists.
-pub type OrdKeyBatcher<K, T, R> = MergeBatcher<VecMerger<(K, ()), T, R>>;
+pub type OrdKeyBatcher<K, T, R> = MergeBatcher<ContainerChunker<Vec<((K, ()), T, R)>>, VecMerger<(K, ()), T, R>, VecOrdKeyBuilder<K, T, R>>;
 /// A builder for ordered lists.
 pub type VecOrdKeyBuilder<K, T, R> = OrdKeyBuilder<Vector<((K,()),T,R)>, Vec<((K,()),T,R)>>;
 
diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs
index 3c114d282..75a82c4a2 100644
--- a/differential-dataflow/src/trace/mod.rs
+++ b/differential-dataflow/src/trace/mod.rs
@@ -228,13 +228,18 @@ pub trait Trace : TraceReader {
 /// A type capable of accepting containers of updates, and carving them out by time as batches.
 ///
 /// Updates are accepted as `C0`, the containers that arrive on the dataflow edge, and released as
-/// `C1`, whatever the consumer means by a batch. The two need not agree: an implementor staging
-/// updates in a form of its own can release that form directly. A consumer whose batch is a
-/// sequence of chunks names a sequence as `C1`.
+/// `Output`, whatever the implementor means by a batch. The two need not agree: an implementor
+/// staging updates in a form of its own can release that form directly, and one whose batch is a
+/// sequence of chunks names a sequence as its output.
 ///
 /// The implementor determines the meaning of extraction by a frontier; it is not required to be by
 /// antichain partial order.
-pub trait Batcher<T, C0, C1> {
+pub trait Batcher<C0> {
+    /// The timestamps by which updates are carved out.
+    type Time;
+    /// The batches released by extraction.
+    type Output;
+
     /// Takes the updates in `container`, leaving it in an undefined state.
     ///
     /// The implementor decides whether to claim the container's allocation or to drain it and
@@ -249,7 +254,7 @@ pub trait Batcher<T, C0, C1> {
     /// The reported lower bound should accurately reflect the times of all accepted updates that
     /// have not been extracted. Over approximation can result in stalling dataflows, and under
     /// approximation is simply incorrect.
-    fn extract<'a>(&'a mut self, upper: AntichainRef<'_, T>) -> (Option<C1>, AntichainRef<'a, T>);
+    fn extract<'a>(&'a mut self, upper: AntichainRef<'_, Self::Time>) -> (Option<Self::Output>, AntichainRef<'a, Self::Time>);
 }
 
 /// Functionality for building batches from ordered update sequences.
diff --git a/differential-dataflow/tests/columnar.rs b/differential-dataflow/tests/columnar.rs
index 5001d01be..5f7fa642a 100644
--- a/differential-dataflow/tests/columnar.rs
+++ b/differential-dataflow/tests/columnar.rs
@@ -5,7 +5,7 @@ use timely::dataflow::{InputHandle, ProbeHandle};
 use timely::Config;
 
 use differential_dataflow::columnar::collection;
-use differential_dataflow::columnar::trace::{Batcher, Builder, Chunker, Spine};
+use differential_dataflow::columnar::trace::{Batcher, Spine};
 use differential_dataflow::operators::arrange::arrangement::arrange_core;
 
 type Upd = (u64, (), u64, i64);
@@ -29,9 +29,7 @@ fn arrange_reaches_one(config: Config, diffs: &'static [i64]) -> bool {
             arrange_core::<
                 _,
                 _,
-                Chunker<Upd>,
                 _,
-                Builder<u64, (), u64, i64>,
                 Spine<u64, (), u64, i64>,
             >(stream, pact, "Arrange", Batcher::new)
             .stream
diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs
index a71fca87d..8618bf1e5 100644
--- a/differential-dataflow/tests/int_proxy.rs
+++ b/differential-dataflow/tests/int_proxy.rs
@@ -24,11 +24,10 @@ use differential_dataflow::operators::int_proxy::vec_backend::VecReduceBackend;
 use differential_dataflow::operators::iterate::Iterate;
 use differential_dataflow::operators::reduce::{reduce_with_tactic, ReduceTactic};
 use differential_dataflow::trace::chunk::vec::{
-    ChunkBatcher as VChunkBatcher, ChunkBuilder as VChunkBuilder, ChunkSpine as VChunkSpine, VecChunk,
+    ChunkBatcher as VChunkBatcher, ChunkSpine as VChunkSpine, VecChunk,
 };
 use differential_dataflow::trace::chunk::ChunkBatch;
 use differential_dataflow::trace::cursor::Cursor;
-use differential_dataflow::trace::implementations::ContainerChunker;
 use differential_dataflow::trace::{Navigable};
 use differential_dataflow::AsCollection;
 
@@ -139,7 +138,7 @@ fn proxy_matches_mainline(updates: Vec<((u64, u64), u64, i64)>, window: usize) {
         worker.dataflow::<u64, _, _>(|scope| {
             let coll = updates.clone().to_stream(scope).as_collection();
             let hashed = coll.clone().map(|(k, v)| (k.hashed(), (k, v)));
-            let arr = arrange_core::<Pipeline, Vec<((u64, (u64, u64)), u64, i64)>, ContainerChunker<VecChunk<u64, (u64, u64), u64, i64>>, _, VChunkBuilder<u64, (u64, u64), u64, i64>, VChunkSpine<u64, (u64, u64), u64, i64>>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new);
+            let arr = arrange_core::<Pipeline, Vec<((u64, (u64, u64)), u64, i64)>, _, VChunkSpine<u64, (u64, u64), u64, i64>>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new);
             reduce_with_tactic::<_, VChunkSpine<u64, (u64, u64), u64, i64>, _>(arr, "VecReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, window)))
                 .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1))
                 .inspect(move |(d, t, r)| ts.lock().unwrap().push((*d, *t, *r)));
@@ -202,7 +201,7 @@ fn reduce_string_values_matches_mainline() {
         worker.dataflow::<u64, _, _>(|scope| {
             let coll = updates.clone().to_stream(scope).as_collection();
             let hashed = coll.clone().map(|(k, v): (u64, String)| (k.hashed(), (k, v)));
-            let arr = arrange_core::<Pipeline, Vec<((u64, (u64, String)), u64, i64)>, ContainerChunker<VecChunk<u64, (u64, String), u64, i64>>, _, VChunkBuilder<u64, (u64, String), u64, i64>, VChunkSpine<u64, (u64, String), u64, i64>>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new);
+            let arr = arrange_core::<Pipeline, Vec<((u64, (u64, String)), u64, i64)>, _, VChunkSpine<u64, (u64, String), u64, i64>>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new);
             reduce_with_tactic::<_, VChunkSpine<u64, (u64, String), u64, i64>, _>(arr, "VecReduceStr", ProxyReduceTactic::new(VecReduceBackend::with_window(
                 |_k: &u64, input: &[(String, i64)], current: &mut Vec<(String, i64)>, updates: &mut Vec<(String, i64)>| {
                     if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| v.clone()).max() { updates.push((m, 1)); }
@@ -236,7 +235,7 @@ fn reduce_inside_iterate() {
             let input = updates.clone().to_stream(scope).as_collection();
             let result = input.iterate(|_scope, inner| {
                 let hashed = inner.map(|(k, v)| (k.hashed(), (k, v)));
-                let arr = arrange_core::<Pipeline, Vec<((u64, (u64, u64)), Product<u64, u64>, i64)>, ContainerChunker<VecChunk<u64, (u64, u64), Product<u64, u64>, i64>>, _, VChunkBuilder<u64, (u64, u64), Product<u64, u64>, i64>, VChunkSpine<u64, (u64, u64), Product<u64, u64>, i64>>(hashed.inner, Pipeline, "ArrIter", VChunkBatcher::new);
+                let arr = arrange_core::<Pipeline, Vec<((u64, (u64, u64)), Product<u64, u64>, i64)>, _, VChunkSpine<u64, (u64, u64), Product<u64, u64>, i64>>(hashed.inner, Pipeline, "ArrIter", VChunkBatcher::new);
                 reduce_with_tactic::<_, VChunkSpine<u64, (u64, u64), Product<u64, u64>, i64>, _>(arr, "IterReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1)))
                     .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1))
             });
@@ -311,7 +310,7 @@ fn reduce_collision_inside_iterate() {
             let input = updates.clone().to_stream(scope).as_collection();
             let result = input.iterate(|_scope, inner| {
                 let hashed = inner.map(|(k, v)| (k % 2, (k, v)));
-                let arr = arrange_core::<Pipeline, Vec<((u64, (u64, u64)), Product<u64, u64>, i64)>, ContainerChunker<VecChunk<u64, (u64, u64), Product<u64, u64>, i64>>, _, VChunkBuilder<u64, (u64, u64), Product<u64, u64>, i64>, VChunkSpine<u64, (u64, u64), Product<u64, u64>, i64>>(hashed.inner, Pipeline, "ArrCollide", VChunkBatcher::new);
+                let arr = arrange_core::<Pipeline, Vec<((u64, (u64, u64)), Product<u64, u64>, i64)>, _, VChunkSpine<u64, (u64, u64), Product<u64, u64>, i64>>(hashed.inner, Pipeline, "ArrCollide", VChunkBatcher::new);
                 reduce_with_tactic::<_, VChunkSpine<u64, (u64, u64), Product<u64, u64>, i64>, _>(arr, "CollideReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1)))
                     .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1))
             });
diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs
index c3942300b..094d5361e 100644
--- a/differential-dataflow/tests/int_proxy_bench.rs
+++ b/differential-dataflow/tests/int_proxy_bench.rs
@@ -34,10 +34,9 @@ use differential_dataflow::operators::iterate::Iterate;
 use differential_dataflow::operators::reduce::reduce_with_tactic;
 use differential_dataflow::trace::chunk::vec::{
     ChunkBatcher as VChunkBatcher, ChunkBuilder as VChunkBuilder, ChunkSpine as VChunkSpine,
-    VecChunk, VecChunkCursor,
+    VecChunkCursor,
 };
 use differential_dataflow::trace::cursor::Cursor;
-use differential_dataflow::trace::implementations::ContainerChunker;
 
 #[derive(Clone, Copy, PartialEq, Eq, Debug)]
 enum Mode {
@@ -56,9 +55,7 @@ macro_rules! harrange {
         arrange_core::<
             Pipeline,
             Vec<((u64, (u64, u64)), $t, isize)>,
-            ContainerChunker<VecChunk<u64, (u64, u64), $t, isize>>,
             _,
-            VChunkBuilder<u64, (u64, u64), $t, isize>,
             VChunkSpine<u64, (u64, u64), $t, isize>,
         >(hashed.inner, Pipeline, $name, VChunkBatcher::new)
     }};
@@ -147,9 +144,7 @@ fn run_wide(mode: Mode) -> f64 {
                 Mode::CursorSame => {
                     let hashed = coll.map(|(k, v)| (k.hashed(), (k, v)));
                     let arr = arrange_core::<Pipeline, Vec<((u64, (u64, String)), u64, isize)>,
-                        ContainerChunker<VecChunk<u64, (u64, String), u64, isize>>,
                         _,
-                        VChunkBuilder<u64, (u64, String), u64, isize>,
                         VChunkSpine<u64, (u64, String), u64, isize>>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new);
                     arr.reduce_core::<_, VChunkBuilder<u64, (u64, String), u64, isize>,
                         VChunkSpine<u64, (u64, String), u64, isize>,
@@ -171,9 +166,7 @@ fn run_wide(mode: Mode) -> f64 {
                 Mode::Proxy => {
                     let hashed = coll.map(|(k, v)| (k.hashed(), (k, v)));
                     let arr = arrange_core::<Pipeline, Vec<((u64, (u64, String)), u64, isize)>,
-                        ContainerChunker<VecChunk<u64, (u64, String), u64, isize>>,
                         _,
-                        VChunkBuilder<u64, (u64, String), u64, isize>,
                         VChunkSpine<u64, (u64, String), u64, isize>>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new);
                     reduce_with_tactic::<_, VChunkSpine<u64, (u64, String), u64, isize>, _>(
                         arr, "ProxyWide",
diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs
index 7a9aea117..1fd9129f5 100644
--- a/differential-dataflow/tests/trace.rs
+++ b/differential-dataflow/tests/trace.rs
@@ -1,12 +1,11 @@
 use timely::dataflow::operators::generic::OperatorInfo;
 use timely::progress::{Antichain, frontier::AntichainRef};
 
-use differential_dataflow::trace::implementations::{ValBatcher, ValBuilder, ValSpine};
-use differential_dataflow::trace::{Description, Span, Trace, TraceReader, Batcher, Builder};
+use differential_dataflow::trace::implementations::{ValBatcher, ValSpine};
+use differential_dataflow::trace::{Description, Span, Trace, TraceReader, Batcher};
 use differential_dataflow::trace::cursor::{Cursor, cursor_list};
 
 type IntegerTrace = ValSpine<u64, u64, usize, i64>;
-type IntegerBuilder = ValBuilder<u64, u64, usize, i64>;
 
 fn get_trace() -> ValSpine<u64, u64, usize, i64> {
     let op_info = OperatorInfo::new(0, 0, [].into());
@@ -14,19 +13,19 @@ fn get_trace() -> ValSpine<u64, u64, usize, i64> {
     {
         let mut batcher = ValBatcher::<u64,u64,usize,i64>::new(None, 0);
 
-        batcher.insert(&mut vec![
+        let mut input: Vec<((u64, u64), usize, i64)> = vec![
             ((1, 2), 0, 1),
             ((2, 3), 1, 1),
             ((2, 3), 2, -1),
-        ]);
+        ];
+        batcher.insert(&mut input);
 
         let batch_ts = &[1, 2, 3];
         let mut lower = Antichain::from_elem(0);
         for i in batch_ts {
             let upper = Antichain::from_elem(*i);
-            let (chain, _retained) = batcher.extract(upper.borrow());
+            let (batch, _retained) = Batcher::<Vec<((u64, u64), usize, i64)>>::extract(&mut batcher, upper.borrow());
             let description = Description::new(lower, upper.clone(), Antichain::from_elem(0));
-            let batch = chain.and_then(|mut chain| IntegerBuilder::seal(&mut chain));
             trace.insert(Span::new(description, batch.map(Into::into)));
             lower = upper;
         }
diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs
index 77cd816b4..588a6f571 100644
--- a/dogsdogsdogs/src/operators/half_join.rs
+++ b/dogsdogsdogs/src/operators/half_join.rs
@@ -77,7 +77,7 @@ where
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
     P: ParallelizationContract<Tr::Time, CIn>,
     Y: Fn(std::time::Instant, usize) -> bool + 'static,
-    Bat: Batcher<Tr::Time, CIn, Vec<CMid>> + 'static,
+    Bat: Batcher<CIn, Time = Tr::Time, Output = Vec<CMid>> + 'static,
     Tac: HalfJoinTactic<Tr::Time, Tr::Batch, CMid, C> + 'static,
     CIn: Container,
     C: Container + 'static,
@@ -484,7 +484,10 @@ pub mod cursors {
         }
     }
 
-    impl<D: Ord, T: Timestamp, R: Semigroup> Batcher<T, Vec<(D, T, R)>, Vec<Vec<(D, T, R)>>> for BlobList<D, T, R> {
+    impl<D: Ord, T: Timestamp, R: Semigroup> Batcher<Vec<(D, T, R)>> for BlobList<D, T, R> {
+        type Time = T;
+        type Output = Vec<Vec<(D, T, R)>>;
+
         fn insert(&mut self, container: &mut Vec<(D, T, R)>) {
             self.stage.extend(container.drain(..).map(|(d,t,r)| (t,d,r)));
         }
diff --git a/experiments/src/bin/deals.rs b/experiments/src/bin/deals.rs
index a6502baf4..dfbec4768 100644
--- a/experiments/src/bin/deals.rs
+++ b/experiments/src/bin/deals.rs
@@ -4,7 +4,7 @@ use differential_dataflow::input::Input;
 use differential_dataflow::VecCollection;
 use differential_dataflow::operators::*;
 
-use differential_dataflow::trace::implementations::{ValSpine, ValBatcher, ValBuilder};
+use differential_dataflow::trace::implementations::{ValSpine, ValBatcher};
 use differential_dataflow::operators::arrange::TraceAgent;
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::operators::iterate::Variable;
@@ -39,7 +39,7 @@ fn main() {
             let (input, graph) = scope.new_collection();
 
             // each edge should exist in both directions.
-            let graph = graph.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
+            let graph = graph.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
 
             match program.as_str() {
                 "tc"    => tc(graph.clone()).filter(move |_| inspect).map(|_| ()).consolidate().inspect(|x| println!("tc count: {:?}", x)).probe(),
@@ -93,7 +93,7 @@ fn tc<'s, T: timely::progress::Timestamp + Lattice + Default + timely::order::Em
             let result =
             inner_collection
                 .map(|(x,y)| (y,x))
-                .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                 .join_core(edges.clone(), |_y,&x,&z| Some((x, z)))
                 .concat(edges.as_collection(|&k,&v| (k,v)))
                 .arrange_by_self()
@@ -121,9 +121,9 @@ fn sg<'s, T: timely::progress::Timestamp + Lattice + Default + timely::order::Em
 
             let result =
             inner_collection
-                .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                 .join_core(edges.clone(), |_,&x,&z| Some((x, z)))
-                .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                 .join_core(edges, |_,&x,&z| Some((x, z)))
                 .concat(peers)
                 .arrange_by_self()
diff --git a/experiments/src/bin/graspan1.rs b/experiments/src/bin/graspan1.rs
index 28e96c818..443b64e07 100644
--- a/experiments/src/bin/graspan1.rs
+++ b/experiments/src/bin/graspan1.rs
@@ -5,7 +5,7 @@ use timely::order::Product;
 
 use differential_dataflow::difference::Present;
 use differential_dataflow::input::Input;
-use differential_dataflow::trace::implementations::{ValBatcher, ValBuilder, ValSpine};
+use differential_dataflow::trace::implementations::{ValBatcher, ValSpine};
 use differential_dataflow::operators::*;
 use differential_dataflow::operators::iterate::Variable;
 
@@ -29,7 +29,7 @@ fn main() {
             let (n_handle, nodes) = scope.new_collection();
             let (e_handle, edges) = scope.new_collection();
 
-            let edges = edges.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
+            let edges = edges.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
 
             // a N c  <-  a N b && b E c
             // N(a,c) <-  N(a,b), E(b, c)
@@ -46,7 +46,7 @@ fn main() {
                 labels_collection.join_core(edges, |_b, a, c| Some((*c, *a)))
                       .concat(nodes)
                       .map(|k| (k, ()))
-                      .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                      .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                     //   .distinct_total_core::<Diff>();
                       .threshold_semigroup(|_,_,x: Option<&Present>| if x.is_none() { Some(Present) } else { None });
 
diff --git a/experiments/src/bin/graspan2.rs b/experiments/src/bin/graspan2.rs
index d087c7a7b..40d20eba1 100644
--- a/experiments/src/bin/graspan2.rs
+++ b/experiments/src/bin/graspan2.rs
@@ -8,7 +8,7 @@ use differential_dataflow::operators::iterate::Variable;
 use differential_dataflow::VecCollection;
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::*;
-use differential_dataflow::trace::implementations::{ValSpine, ValBatcher, ValBuilder};
+use differential_dataflow::trace::implementations::{ValSpine, ValBatcher};
 use differential_dataflow::difference::Present;
 
 type Node = u32;
@@ -45,7 +45,7 @@ fn unoptimized() {
                 .flat_map(|(a,b)| vec![a,b])
                 .concat(dereference.clone().flat_map(|(a,b)| vec![a,b]));
 
-            let dereference = dereference.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
+            let dereference = dereference.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
 
             let (value_flow, memory_alias, value_alias) =
             scope
@@ -58,14 +58,14 @@ fn unoptimized() {
                     let (value_flow, value_flow_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1));
                     let (memory_alias, memory_alias_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1));
 
-                    let value_flow_arranged = value_flow_collection.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
-                    let memory_alias_arranged = memory_alias_collection.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
+                    let value_flow_arranged = value_flow_collection.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
+                    let memory_alias_arranged = memory_alias_collection.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
 
                     // 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.clone().join_core(value_flow_arranged.clone(), |_,&a,&b| Some((a,b)));
                     let value_alias_next = value_flow_arranged.clone().join_core(memory_alias_arranged.clone(), |_,&a,&b| Some((b,a)))
-                                                              .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                                                              .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                                                               .join_core(value_flow_arranged.clone(), |_,&a,&b| Some((a,b)))
                                                               .concat(value_alias_next);
 
@@ -75,10 +75,10 @@ fn unoptimized() {
                     let value_flow_next =
                     assignment.clone()
                         .map(|(a,b)| (b,a))
-                        .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                        .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                         .join_core(memory_alias_arranged, |_,&a,&b| Some((b,a)))
                         .concat(assignment.map(|(a,b)| (b,a)))
-                        .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                        .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                         .join_core(value_flow_arranged, |_,&a,&b| Some((a,b)))
                         .concat(nodes.map(|n| (n,n)));
 
@@ -93,7 +93,7 @@ fn unoptimized() {
                     let memory_alias_next: VecCollection<_,_,Present> =
                     value_alias_next.clone()
                         .join_core(dereference.clone(), |_x,&y,&a| Some((y,a)))
-                        .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                        .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                         .join_core(dereference, |_y,&a,&b| Some((a,b)));
 
                     let memory_alias_next: VecCollection<_,_,Present>  =
@@ -170,7 +170,7 @@ fn optimized() {
                 .flat_map(|(a,b)| vec![a,b])
                 .concat(dereference.clone().flat_map(|(a,b)| vec![a,b]));
 
-            let dereference = dereference.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
+            let dereference = dereference.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
 
             let (value_flow, memory_alias) =
             scope
@@ -183,8 +183,8 @@ fn optimized() {
                     let (value_flow, value_flow_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1));
                     let (memory_alias, memory_alias_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1));
 
-                    let value_flow_arranged = value_flow_collection.clone().arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
-                    let memory_alias_arranged = memory_alias_collection.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
+                    let value_flow_arranged = value_flow_collection.clone().arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
+                    let memory_alias_arranged = memory_alias_collection.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
 
                     // VF(a,a) <-
                     // VF(a,b) <- A(a,x),VF(x,b)
@@ -192,10 +192,10 @@ fn optimized() {
                     let value_flow_next =
                     assignment.clone()
                         .map(|(a,b)| (b,a))
-                        .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                        .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                         .join_core(memory_alias_arranged.clone(), |_,&a,&b| Some((b,a)))
                         .concat(assignment.map(|(a,b)| (b,a)))
-                        .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                        .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                         .join_core(value_flow_arranged, |_,&a,&b| Some((a,b)))
                         .concat(nodes.map(|n| (n,n)))
                         .arrange_by_self()
@@ -207,9 +207,9 @@ fn optimized() {
                     let value_flow_deref =
                     value_flow_collection
                         .map(|(a,b)| (b,a))
-                        .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                        .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                         .join_core(dereference, |_x,&a,&b| Some((a,b)))
-                        .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new);
+                        .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new);
 
                     // MA(a,b) <- VFD(x,a),VFD(y,b)
                     // MA(a,b) <- VFD(x,a),MA(x,y),VFD(y,b)
@@ -220,7 +220,7 @@ fn optimized() {
                     let memory_alias_next =
                     memory_alias_arranged
                         .join_core(value_flow_deref.clone(), |_x,&y,&a| Some((y,a)))
-                        .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new)
+                        .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new)
                         .join_core(value_flow_deref, |_y,&a,&b| Some((a,b)))
                         .concat(memory_alias_next)
                         .arrange_by_self()
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index d3cedb101..81cdaeff8 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -18,7 +18,7 @@ use differential_dataflow::operators::join::join_with_tactic;
 use differential_dataflow::operators::reduce::reduce_with_tactic;
 use differential_dataflow::operators::arrange::arrangement::arrange_core;
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
-use differential_dataflow::trace::chunk::{Chunk, ChunkBatcher, ChunkBuilder};
+use differential_dataflow::trace::chunk::{Chunk, ChunkBatcher};
 
 use corgi::arrange::gather;
 use corgi::Value as CValue;
@@ -295,11 +295,11 @@ impl Backend for CorgiBackend {
         // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s
         // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher +
         // builder. No columns→rows→columns round-trip at the arrangement boundary.
-        arrange_core::<_, CC, CorgiChunker<Time, Diff>, _, ChunkBuilder<CorgiChunk<Time, Diff>>, CTrace>(
+        arrange_core::<_, CC, _, CTrace>(
             c.inner,
             Pipeline,
             "CorgiArrange",
-            ChunkBatcher::new,
+            ChunkBatcher::<CorgiChunker<Time, Diff>, _>::new,
         )
     }