diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs
index 582e5ce98..f26d435b2 100644
--- a/differential-dataflow/examples/columnar/main.rs
+++ b/differential-dataflow/examples/columnar/main.rs
@@ -125,13 +125,15 @@ mod reachability {
             let edges_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 };
             let reach_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 };
 
-            let edges_arr = arrange_core::<_,
+            let edges_arr = arrange_core::<_, _,
+                ValChunker<(Node, Node, IterTime, Diff)>,
                 ValBatcher<Node, Node, IterTime, Diff>,
                 ValBuilder<Node, Node, IterTime, Diff>,
                 ValSpine<Node, Node, IterTime, Diff>,
             >(edges_inner.inner, edges_pact, "Edges");
 
-            let reach_arr = arrange_core::<_,
+            let reach_arr = arrange_core::<_, _,
+                ValChunker<(Node, (), IterTime, Diff)>,
                 ValBatcher<Node, (), IterTime, Diff>,
                 ValBuilder<Node, (), IterTime, Diff>,
                 ValSpine<Node, (), IterTime, Diff>,
@@ -155,7 +157,8 @@ mod reachability {
 
             // Arrange for reduce.
             let combined_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 };
-            let combined_arr = arrange_core::<_,
+            let combined_arr = arrange_core::<_, _,
+                ValChunker<(Node, (), IterTime, Diff)>,
                 ValBatcher<Node, (), IterTime, Diff>,
                 ValBuilder<Node, (), IterTime, Diff>,
                 ValSpine<Node, (), IterTime, Diff>,
diff --git a/differential-dataflow/examples/columnar_spill.rs b/differential-dataflow/examples/columnar_spill.rs
index 96776121f..5203ed97b 100644
--- a/differential-dataflow/examples/columnar_spill.rs
+++ b/differential-dataflow/examples/columnar_spill.rs
@@ -63,7 +63,7 @@ fn reset_stats() {
 use columnar::Push;
 use columnar::bytes::stash::Stash;
 
-use differential_dataflow::columnar::{RecordedUpdates, ValBuilder, ValColBuilder, ValSpine};
+use differential_dataflow::columnar::{ValBuilder, ValChunker, ValColBuilder, ValSpine};
 use differential_dataflow::columnar::batcher::MergeBatcher;
 use differential_dataflow::columnar::layout::ColumnarUpdate as Update;
 use differential_dataflow::columnar::spill::{Entry, Fetch, Spill, SpillPolicy};
@@ -75,6 +75,7 @@ use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::probe::{Handle as ProbeHandle, Probe};
 use timely::dataflow::operators::Input;
 use timely::dataflow::InputHandle;
+use timely::container::PushInto;
 use timely::progress::frontier::AntichainRef;
 use timely::progress::{frontier::Antichain, Timestamp};
 
@@ -323,7 +324,6 @@ where
     R: columnar::Columnar + 'static,
     (K, V, T, R): Update<Time = T> + 'static,
 {
-    type Input = RecordedUpdates<(K, V, T, R)>;
     type Time = T;
     type Output = UpdatesTyped<(K, V, T, R)>;
 
@@ -343,10 +343,6 @@ where
         Self(inner)
     }
 
-    fn push_container(&mut self, container: &mut Self::Input) {
-        self.0.push_container(container);
-    }
-
     fn seal<B: Builder<Input = Self::Output, Time = Self::Time>>(
         &mut self,
         upper: Antichain<T>,
@@ -359,6 +355,15 @@ where
     }
 }
 
+impl<K, V, T, R> PushInto<UpdatesTyped<(K, V, T, R)>> for SpillBatcher<K, V, T, R>
+where
+    (K, V, T, R): Update,
+{
+    fn push_into(&mut self, chunk: UpdatesTyped<(K, V, T, R)>) {
+        self.0.push_into(chunk);
+    }
+}
+
 type TestUpdate = (u64, u64, u64, i64);
 
 fn make_chunk(updates: &[(u64, u64, u64, i64)]) -> UpdatesTyped<TestUpdate> {
@@ -537,6 +542,8 @@ fn run_timely_dataflow(
             let stream = scope.input_from(&mut input);
             let arranged = arrange_core::<
                 _,
+                _,
+                ValChunker<(u64, u64, u64, i64)>,
                 SpillBatcher<u64, u64, u64, i64>,
                 ValBuilder<u64, u64, u64, i64>,
                 ValSpine<u64, u64, u64, i64>,
diff --git a/differential-dataflow/examples/spines.rs b/differential-dataflow/examples/spines.rs
index c8df1db10..e5c153a3c 100644
--- a/differential-dataflow/examples/spines.rs
+++ b/differential-dataflow/examples/spines.rs
@@ -108,7 +108,7 @@ fn main() {
                 },
                 "col" => {
                     use timely::dataflow::operators::Input as _;
-                    use differential_dataflow::columnar::{ValBatcher, ValBuilder, ValPact, ValSpine};
+                    use differential_dataflow::columnar::{ValBatcher, ValBuilder, ValChunker, ValPact, ValSpine};
                     use differential_dataflow::operators::arrange::arrangement::arrange_core;
 
                     fn string_hash(s: columnar::Ref<'_, String>) -> u64 {
@@ -124,10 +124,10 @@ fn main() {
                     let data_stream = scope.input_from(&mut data_input);
                     let keys_stream = scope.input_from(&mut keys_input);
 
-                    let data = arrange_core::<_, ValBatcher<String,(),u64,i64>, ValBuilder<String,(),u64,i64>, ValSpine<String,(),u64,i64>>(
+                    let data = arrange_core::<_, _, ValChunker<(String,(),u64,i64)>, ValBatcher<String,(),u64,i64>, ValBuilder<String,(),u64,i64>, ValSpine<String,(),u64,i64>>(
                         data_stream, ValPact { hashfunc: |k: columnar::Ref<'_, String>| string_hash(k) }, "DataArrange",
                     );
-                    let keys = arrange_core::<_, ValBatcher<String,(),u64,i64>, ValBuilder<String,(),u64,i64>, ValSpine<String,(),u64,i64>>(
+                    let keys = arrange_core::<_, _, ValChunker<(String,(),u64,i64)>, ValBatcher<String,(),u64,i64>, ValBuilder<String,(),u64,i64>, ValSpine<String,(),u64,i64>>(
                         keys_stream, ValPact { hashfunc: |k: columnar::Ref<'_, String>| string_hash(k) }, "KeysArrange",
                     );
                     keys.join_core(data, |_k, (), ()| Option::<()>::None)
diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs
index 16755c27e..a319f5c0b 100644
--- a/differential-dataflow/src/collection.rs
+++ b/differential-dataflow/src/collection.rs
@@ -961,9 +961,9 @@ pub mod vec {
         /// and provide the function `reify` to produce owned keys and values..
         pub fn consolidate_named<Ba, Bu, Tr, F>(self, name: &str, reify: F) -> Self
         where
-            Ba: crate::trace::Batcher<Input=Vec<((D,()),T,R)>, Time=T> + 'static,
+            Ba: crate::trace::Batcher<Output=Vec<((D, ()), T, R)>, Time=T> + 'static,
             Tr: for<'a> crate::trace::Trace<Time=T,Diff=R>+'static,
-            Bu: crate::trace::Builder<Time=Tr::Time, Input=Ba::Output, Output=Tr::Batch>,
+            Bu: crate::trace::Builder<Time=Tr::Time, Input=Vec<((D, ()), T, R)>, Output=Tr::Batch>,
             F: Fn(Tr::Key<'_>, Tr::Val<'_>) -> D + 'static,
         {
             use crate::operators::arrange::arrangement::Arrange;
@@ -1018,6 +1018,7 @@ pub mod vec {
 
     use crate::trace::implementations::{ValSpine, ValBatcher, ValBuilder};
     use crate::trace::implementations::{KeySpine, KeyBatcher, KeyBuilder};
+    use crate::trace::implementations::ContainerChunker;
     use crate::operators::arrange::Arrange;
 
     impl<'scope, T, K, V, R> Arrange<'scope, T, Vec<((K, V), T, R)>> for Collection<'scope, T, (K, V), R>
@@ -1029,12 +1030,12 @@ pub mod vec {
     {
         fn arrange_named<Ba, Bu, Tr>(self, name: &str) -> Arranged<'scope, TraceAgent<Tr>>
         where
-            Ba: crate::trace::Batcher<Input=Vec<((K, V), T, R)>, Time=T> + 'static,
-            Bu: crate::trace::Builder<Time=T, Input=Ba::Output, Output = Tr::Batch>,
+            Ba: crate::trace::Batcher<Output=Vec<((K, V), T, R)>, Time=T> + 'static,
+            Bu: crate::trace::Builder<Time=T, Input=Vec<((K, V), T, R)>, Output = Tr::Batch>,
             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::<_, Ba, Bu, _>(self.inner, exchange, name)
+            crate::operators::arrange::arrangement::arrange_core::<_, _, ContainerChunker<Vec<((K, V), T, R)>>, Ba, Bu, _>(self.inner, exchange, name)
         }
     }
 
@@ -1044,12 +1045,12 @@ pub mod vec {
     {
         fn arrange_named<Ba, Bu, Tr>(self, name: &str) -> Arranged<'scope, TraceAgent<Tr>>
         where
-            Ba: crate::trace::Batcher<Input=Vec<((K,()),T,R)>, Time=T> + 'static,
-            Bu: crate::trace::Builder<Time=T, Input=Ba::Output, Output = Tr::Batch>,
+            Ba: crate::trace::Batcher<Output=Vec<((K, ()), T, R)>, Time=T> + 'static,
+            Bu: crate::trace::Builder<Time=T, Input=Vec<((K, ()), T, R)>, Output = Tr::Batch>,
             Tr: crate::trace::Trace<Time=T> + 'static,
         {
             let exchange = timely::dataflow::channels::pact::Exchange::new(move |update: &((K,()),T,R)| (update.0).0.hashed().into());
-            crate::operators::arrange::arrangement::arrange_core::<_,Ba,Bu,_>(self.map(|k| (k, ())).inner, exchange, name)
+            crate::operators::arrange::arrangement::arrange_core::<_, _, ContainerChunker<Vec<((K, ()), T, R)>>, Ba, Bu, _>(self.map(|k| (k, ())).inner, exchange, name)
         }
     }
 
diff --git a/differential-dataflow/src/columnar/arrangement/mod.rs b/differential-dataflow/src/columnar/arrangement/mod.rs
index 801ecbe0a..1b71ff4ae 100644
--- a/differential-dataflow/src/columnar/arrangement/mod.rs
+++ b/differential-dataflow/src/columnar/arrangement/mod.rs
@@ -21,6 +21,8 @@ pub mod trie_merger;
 pub type ValSpine<K, V, T, R> = Spine<Rc<OrdValBatch<ColumnarLayout<(K,V,T,R)>>>>;
 /// A batcher for columnar storage.
 pub type ValBatcher<K, V, T, R> = super::batcher::MergeBatcher<(K,V,T,R)>;
+/// A chunker that maps `RecordedUpdates<U>` streams into the batcher's `UpdatesTyped<U>` chunks.
+pub type ValChunker<U> = TrieChunker<U>;
 /// A builder for columnar storage.
 pub type ValBuilder<K, V, T, R> = RcBuilder<builder::ValMirror<(K,V,T,R)>>;
 
diff --git a/differential-dataflow/src/columnar/batcher.rs b/differential-dataflow/src/columnar/batcher.rs
index 085c44012..fff0cfe8e 100644
--- a/differential-dataflow/src/columnar/batcher.rs
+++ b/differential-dataflow/src/columnar/batcher.rs
@@ -1,26 +1,27 @@
-//! A `Batcher` for `RecordedUpdates<U>` streams that consolidates input via
-//! `TrieChunker` and merges sorted chains via the free functions in `trie_merger`.
+//! A `Batcher` for columnar streams that merges sorted chains via the free
+//! functions in `trie_merger`.
+//!
+//! Callers feed already-chunked, sorted-and-consolidated `UpdatesTyped<U>` into
+//! the batcher via [`PushInto`]; forming such chunks from `RecordedUpdates<U>`
+//! is the responsibility of the surrounding dataflow operator's chunker
+//! (`TrieChunker`).
 
 use std::collections::VecDeque;
 
 use timely::progress::frontier::AntichainRef;
 use timely::progress::{frontier::Antichain, Timestamp};
-use timely::container::{ContainerBuilder, PushInto};
+use timely::container::PushInto;
 
 use crate::logging::Logger;
 use crate::trace::{Batcher, Builder, Description};
 
 use super::layout::ColumnarUpdate as Update;
 use super::updates::UpdatesTyped;
-use super::RecordedUpdates;
-use super::arrangement::TrieChunker;
 use super::arrangement::trie_merger;
 use super::spill::{Entry, SpillPolicy};
 
-/// Creates batches from `RecordedUpdates<U>` streams.
+/// Creates batches from chunks of sorted, consolidated columnar updates.
 pub struct MergeBatcher<U: Update> {
-    /// Transforms input streams to chunks of sorted, consolidated data.
-    chunker: TrieChunker<U>,
     /// A sequence of power-of-two length chains of sorted, consolidated entries.
     /// Each entry is either an in-memory chunk or a handle to a paged-out chunk.
     chains: Vec<VecDeque<Entry<UpdatesTyped<U>>>>,
@@ -34,13 +35,11 @@ pub struct MergeBatcher<U: Update> {
 }
 
 impl<U: Update<Time: Timestamp>> Batcher for MergeBatcher<U> {
-    type Input = RecordedUpdates<U>;
     type Time = U::Time;
     type Output = UpdatesTyped<U>;
 
     fn new(_logger: Option<Logger>, _operator_id: usize) -> Self {
         Self {
-            chunker: TrieChunker::default(),
             chains: Vec::new(),
             frontier: Antichain::new(),
             lower: Antichain::from_elem(U::Time::minimum()),
@@ -48,27 +47,11 @@ impl<U: Update<Time: Timestamp>> Batcher for MergeBatcher<U> {
         }
     }
 
-    /// Push a container of data into this merge batcher. Updates the internal chain structure if
-    /// needed.
-    fn push_container(&mut self, container: &mut RecordedUpdates<U>) {
-        self.chunker.push_into(container);
-        while let Some(chunk) = self.chunker.extract() {
-            let chunk = std::mem::take(chunk);
-            self.insert_chain(VecDeque::from([Entry::Typed(chunk)]));
-        }
-    }
-
     // Sealing a batch 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`,
     // which we call `lower`, by assumption that after sealing a batcher we receive no more
     // updates with times not greater or equal to `upper`.
     fn seal<B: Builder<Input = Self::Output, Time = Self::Time>>(&mut self, upper: Antichain<U::Time>) -> B::Output {
-        // Finish
-        while let Some(chunk) = self.chunker.finish() {
-            let chunk = std::mem::take(chunk);
-            self.insert_chain(VecDeque::from([Entry::Typed(chunk)]));
-        }
-
         // Merge all remaining chains into a single chain.
         while self.chains.len() > 1 {
             let list1 = self.chains.pop().unwrap();
@@ -122,6 +105,12 @@ impl<U: Update<Time: Timestamp>> Batcher for MergeBatcher<U> {
     }
 }
 
+impl<U: Update> PushInto<UpdatesTyped<U>> for MergeBatcher<U> {
+    fn push_into(&mut self, chunk: UpdatesTyped<U>) {
+        self.insert_chain(VecDeque::from([Entry::Typed(chunk)]));
+    }
+}
+
 impl<U: Update> MergeBatcher<U> {
     /// Install a spill policy. Consulted after each chain insert.
     pub fn set_spill_policy(&mut self, policy: Box<dyn SpillPolicy<UpdatesTyped<U>>>) {
diff --git a/differential-dataflow/src/columnar/mod.rs b/differential-dataflow/src/columnar/mod.rs
index a0a104a18..6ead4fe80 100644
--- a/differential-dataflow/src/columnar/mod.rs
+++ b/differential-dataflow/src/columnar/mod.rs
@@ -42,7 +42,7 @@ pub mod spill;
 pub use updates::UpdatesTyped;
 pub use builder::ValBuilder as ValColBuilder;
 pub use exchange::ValPact;
-pub use arrangement::{ValBatcher, ValBuilder, ValSpine};
+pub use arrangement::{ValBatcher, ValBuilder, ValChunker, ValSpine};
 
 /// Target size for update batches, in number of updates.
 pub const LINK_TARGET: usize = 64 * 1024;
diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index 42f3d7b43..857444eee 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -24,6 +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::dataflow::operators::Capability;
 
 use crate::{Data, VecCollection, AsCollection};
@@ -303,9 +304,13 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
 /// A type that can be arranged as if a collection of updates.
 pub trait Arrange<'scope, T: Timestamp+Lattice, C> : Sized {
     /// Arranges updates into a shared trace.
+    ///
+    /// The batcher's output container must equal the stream container `C`; the default
+    /// chunker only consolidates same-type containers. For chunker setups that convert
+    /// between container types (e.g. columnar layouts), call [`arrange_core`] directly.
     fn arrange<Ba, Bu, Tr>(self) -> Arranged<'scope, TraceAgent<Tr>>
     where
-        Ba: Batcher<Input=C, Time=T> + 'static,
+        Ba: Batcher<Output=C, Time=T> + 'static,
         Bu: Builder<Time=T, Input=Ba::Output, Output = Tr::Batch>,
         Tr: Trace<Time=T> + 'static,
     {
@@ -313,9 +318,11 @@ pub trait Arrange<'scope, T: Timestamp+Lattice, C> : Sized {
     }
 
     /// Arranges updates into a shared trace, with a supplied name.
+    ///
+    /// See [`Arrange::arrange`] for constraints on the batcher's output container.
     fn arrange_named<Ba, Bu, Tr>(self, name: &str) -> Arranged<'scope, TraceAgent<Tr>>
     where
-        Ba: Batcher<Input=C, Time=T> + 'static,
+        Ba: Batcher<Output=C, Time=T> + 'static,
         Bu: Builder<Time=T, Input=Ba::Output, Output = Tr::Batch>,
         Tr: Trace<Time=T> + 'static,
     ;
@@ -326,10 +333,12 @@ pub trait Arrange<'scope, T: Timestamp+Lattice, C> : Sized {
 /// 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, Ba, Bu, Tr>(stream: Stream<'scope, Tr::Time, Ba::Input>, pact: P, name: &str) -> Arranged<'scope, TraceAgent<Tr>>
+pub fn arrange_core<'scope, P, C, Chu, Ba, Bu, Tr>(stream: Stream<'scope, Tr::Time, C>, pact: P, name: &str) -> Arranged<'scope, TraceAgent<Tr>>
 where
-    P: ParallelizationContract<Tr::Time, Ba::Input>,
-    Ba: Batcher<Time=Tr::Time,Input: Container> + 'static,
+    C: Container + Clone + 'static,
+    P: ParallelizationContract<Tr::Time, C>,
+    Chu: ContainerBuilder<Container=Ba::Output> + for<'a> PushInto<&'a mut C> + 'static,
+    Ba: Batcher<Time=Tr::Time> + 'static,
     Bu: Builder<Time=Tr::Time, Input=Ba::Output, Output = Tr::Batch>,
     Tr: Trace+'static,
 {
@@ -379,6 +388,8 @@ 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.
@@ -387,7 +398,10 @@ where
 
             input.for_each(|cap, data| {
                 capabilities.insert(cap.retain(0));
-                batcher.push_container(data);
+                chunker.push_into(data);
+                while let Some(chunk) = chunker.extract() {
+                    batcher.push_into(std::mem::take(chunk));
+                }
             });
 
             // The frontier may have advanced by multiple elements, which is an issue because
@@ -402,6 +416,13 @@ 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.push_into(std::mem::take(chunk));
+                }
+
                 // There are two cases to handle with some care:
                 //
                 // 1. If any held capabilities are not in advance of the new input frontier,
diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs
index ccb2ac943..6f86f503a 100644
--- a/differential-dataflow/src/trace/implementations/merge_batcher.rs
+++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs
@@ -1,31 +1,22 @@
 //! A `Batcher` implementation based on merge sort.
 //!
-//! The `MergeBatcher` requires support from two types, a "chunker" and a "merger".
-//! The chunker receives input batches and consolidates them, producing sorted output
-//! "chunks" that are fully consolidated (no adjacent updates can be accumulated).
-//! The merger implements the [`Merger`] trait, and provides hooks for manipulating
-//! sorted "chains" of chunks as needed by the merge batcher: merging chunks and also
-//! splitting them apart based on time.
+//! The `MergeBatcher` requires a "merger" that implements the [`Merger`] trait, which provides
+//! hooks for manipulating sorted "chains" of chunks as needed by the merge batcher: merging
+//! chunks and also splitting them apart based on time.
 //!
-//! Implementations of `MergeBatcher` can be instantiated through the choice of both
-//! the chunker and the merger, provided their respective output and input types align.
-
-use std::marker::PhantomData;
+//! Callers feed already-chunked, sorted-and-consolidated input into the batcher via [`PushInto`].
+//! Forming such chunks from raw data is the responsibility of the caller (typically a chunker
+//! living in the surrounding dataflow operator).
 
 use timely::progress::frontier::AntichainRef;
 use timely::progress::{frontier::Antichain, Timestamp};
-use timely::container::{ContainerBuilder, PushInto};
+use timely::container::PushInto;
 
 use crate::logging::{BatcherEvent, Logger};
 use crate::trace::{Batcher, Builder, Description};
 
-/// Creates batches from containers of unordered tuples.
-///
-/// To implement `Batcher`, the container builder `C` must accept `&mut Input` as inputs,
-/// and must produce outputs of type `M::Chunk`.
-pub struct MergeBatcher<Input, C, M: Merger> {
-    /// Transforms input streams to chunks of sorted, consolidated data.
-    chunker: C,
+/// Creates batches from chunks of sorted, consolidated tuples.
+pub struct MergeBatcher<M: Merger> {
     /// A sequence of power-of-two length lists of sorted, consolidated containers.
     ///
     /// Do not push/pop directly but use the corresponding functions ([`Self::chain_push`]/[`Self::chain_pop`]).
@@ -42,16 +33,12 @@ pub struct MergeBatcher<Input, C, M: Merger> {
     logger: Option<Logger>,
     /// Timely operator ID.
     operator_id: usize,
-    /// The `Input` type needs to be called out as the type of container accepted, but it is not otherwise present.
-    _marker: PhantomData<Input>,
 }
 
-impl<Input, C, M> Batcher for MergeBatcher<Input, C, M>
+impl<M> Batcher for MergeBatcher<M>
 where
-    C: ContainerBuilder<Container=M::Chunk> + for<'a> PushInto<&'a mut Input>,
     M: Merger<Time: Timestamp>,
 {
-    type Input = Input;
     type Time = M::Time;
     type Output = M::Chunk;
 
@@ -59,23 +46,11 @@ where
         Self {
             logger,
             operator_id,
-            chunker: C::default(),
             merger: M::default(),
             chains: Vec::new(),
             stash: Vec::new(),
             frontier: Antichain::new(),
             lower: Antichain::from_elem(M::Time::minimum()),
-            _marker: PhantomData,
-        }
-    }
-
-    /// Push a container of data into this merge batcher. Updates the internal chain structure if
-    /// needed.
-    fn push_container(&mut self, container: &mut Input) {
-        self.chunker.push_into(container);
-        while let Some(chunk) = self.chunker.extract() {
-            let chunk = std::mem::take(chunk);
-            self.insert_chain(vec![chunk]);
         }
     }
 
@@ -84,12 +59,6 @@ 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`.
     fn seal<B: Builder<Input = Self::Output, Time = Self::Time>>(&mut self, upper: Antichain<M::Time>) -> B::Output {
-        // Finish
-        while let Some(chunk) = self.chunker.finish() {
-            let chunk = std::mem::take(chunk);
-            self.insert_chain(vec![chunk]);
-        }
-
         // Merge all remaining chains into a single chain.
         while self.chains.len() > 1 {
             let list1 = self.chain_pop().unwrap();
@@ -125,7 +94,13 @@ where
     }
 }
 
-impl<Input, C, M: Merger> MergeBatcher<Input, C, M> {
+impl<M: Merger> PushInto<M::Chunk> for MergeBatcher<M> {
+    fn push_into(&mut self, chunk: M::Chunk) {
+        self.insert_chain(vec![chunk]);
+    }
+}
+
+impl<M: Merger> MergeBatcher<M> {
     /// Insert a chain and maintain chain properties: Chains are geometrically sized and ordered
     /// by decreasing length.
     fn insert_chain(&mut self, chain: Vec<M::Chunk>) {
@@ -189,7 +164,7 @@ impl<Input, C, M: Merger> MergeBatcher<Input, C, M> {
     }
 }
 
-impl<Input, C, M: Merger> Drop for MergeBatcher<Input, C, M> {
+impl<M: Merger> Drop for MergeBatcher<M> {
     fn drop(&mut self) {
         // Cleanup chain to retract accounting information.
         while self.chain_pop().is_some() {}
diff --git a/differential-dataflow/src/trace/implementations/mod.rs b/differential-dataflow/src/trace/implementations/mod.rs
index 2b14659f1..46bb27e90 100644
--- a/differential-dataflow/src/trace/implementations/mod.rs
+++ b/differential-dataflow/src/trace/implementations/mod.rs
@@ -45,6 +45,7 @@ pub mod ord_neu;
 pub mod chunker;
 
 // Opinionated takes on default spines.
+pub use self::chunker::ContainerChunker;
 pub use self::ord_neu::OrdValSpine as ValSpine;
 pub use self::ord_neu::OrdValBatcher as ValBatcher;
 pub use self::ord_neu::RcOrdValBuilder as ValBuilder;
diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs
index 37a863811..309ef6694 100644
--- a/differential-dataflow/src/trace/implementations/ord_neu.rs
+++ b/differential-dataflow/src/trace/implementations/ord_neu.rs
@@ -10,7 +10,6 @@
 
 use std::rc::Rc;
 
-use crate::trace::implementations::chunker::ContainerChunker;
 use crate::trace::implementations::spine_fueled::Spine;
 use crate::trace::implementations::merge_batcher::MergeBatcher;
 use crate::trace::implementations::merge_batcher::vec::VecMerger;
@@ -24,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<Vec<((K,V),T,R)>, ContainerChunker<Vec<((K,V),T,R)>>, VecMerger<(K, V), T, R>>;
+pub type OrdValBatcher<K, V, T, R> = MergeBatcher<VecMerger<(K, V), T, R>>;
 /// A builder using ordered lists.
 pub type RcOrdValBuilder<K, V, T, R> = RcBuilder<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<Vec<((K,()),T,R)>, ContainerChunker<Vec<((K,()),T,R)>>, VecMerger<(K, ()), T, R>>;
+pub type OrdKeyBatcher<K, T, R> = MergeBatcher<VecMerger<(K, ()), T, R>>;
 /// A builder for ordered lists.
 pub type RcOrdKeyBuilder<K, T, R> = RcBuilder<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 582a6da62..7ab47c98d 100644
--- a/differential-dataflow/src/trace/mod.rs
+++ b/differential-dataflow/src/trace/mod.rs
@@ -12,6 +12,7 @@ pub mod description;
 pub mod implementations;
 pub mod wrappers;
 
+use timely::container::PushInto;
 use timely::progress::{Antichain, frontier::AntichainRef};
 use timely::progress::Timestamp;
 
@@ -297,17 +298,17 @@ pub trait Batch : BatchReader + Sized {
 }
 
 /// Functionality for collecting and batching updates.
-pub trait Batcher {
-    /// Type pushed into the batcher.
-    type Input;
-    /// Type produced by the batcher.
-    type Output;
+///
+/// Accepts containers of type `Output` via [`PushInto`] and produces output batches of the same
+/// type. Callers are responsible for converting raw input data into `Output` containers (e.g.
+/// using a chunker) before pushing into the batcher.
+pub trait Batcher: PushInto<Self::Output> {
+    /// Type produced by the batcher, and also the type it consumes.
+    type Output: Default;
     /// Times at which batches are formed.
     type Time: Timestamp;
     /// Allocates a new empty batcher.
     fn new(logger: Option<Logger>, operator_id: usize) -> Self;
-    /// Adds an unordered container of elements to the batcher.
-    fn push_container(&mut self, batch: &mut Self::Input);
     /// Returns all updates not greater or equal to an element of `upper`.
     fn seal<B: Builder<Input=Self::Output, Time=Self::Time>>(&mut self, upper: Antichain<Self::Time>) -> B::Output;
     /// Returns the lower envelope of contained update times.
diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs
index 54f111a7d..75911e5bc 100644
--- a/differential-dataflow/tests/trace.rs
+++ b/differential-dataflow/tests/trace.rs
@@ -1,3 +1,4 @@
+use timely::container::PushInto;
 use timely::dataflow::operators::generic::OperatorInfo;
 use timely::progress::{Antichain, frontier::AntichainRef};
 
@@ -14,7 +15,7 @@ fn get_trace() -> ValSpine<u64, u64, usize, i64> {
     {
         let mut batcher = ValBatcher::<u64,u64,usize,i64>::new(None, 0);
 
-        batcher.push_container(&mut vec![
+        batcher.push_into(vec![
             ((1, 2), 0, 1),
             ((2, 3), 1, 1),
             ((2, 3), 2, -1),
diff --git a/interactive/examples/ddir_col.rs b/interactive/examples/ddir_col.rs
index 1c6eca11c..8be9ad863 100644
--- a/interactive/examples/ddir_col.rs
+++ b/interactive/examples/ddir_col.rs
@@ -84,6 +84,7 @@ mod columnar {
     pub type ColValSpine<K, V, T, R> = ValSpine<K, V, T, R>;
     pub type ColValBatcher<K, V, T, R> = ValBatcher<K, V, T, R>;
     pub type ColValBuilder<K, V, T, R> = ValBuilder<K, V, T, R>;
+    pub type ColValChunker<U> = ValChunker<U>;
 }
 
 mod render {
@@ -127,8 +128,8 @@ mod render {
                 Rendered::Arrangement(a) => a.clone(),
                 Rendered::Collection(c) => {
                     use differential_dataflow::operators::arrange::arrangement::arrange_core;
-                    use super::columnar::ColValBatcher;
-                    arrange_core::<_, ColValBatcher<Row,Row,Time,Diff>, ColValBuilder<Row,Row,Time,Diff>, ColValSpine<Row,Row,Time,Diff>>(c.inner.clone(), timely::dataflow::channels::pact::Pipeline, "Arrange")
+                    use super::columnar::{ColValBatcher, ColValChunker};
+                    arrange_core::<_, _, ColValChunker<(Row,Row,Time,Diff)>, ColValBatcher<Row,Row,Time,Diff>, ColValBuilder<Row,Row,Time,Diff>, ColValSpine<Row,Row,Time,Diff>>(c.inner.clone(), timely::dataflow::channels::pact::Pipeline, "Arrange")
                 }
             }
         }