From 5fc079fdbadbb254c6983a53d5cdd9e37855a553 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Thu, 28 May 2026 16:10:22 -0400
Subject: [PATCH] Separate chunker from batcher

The chunker was part of the batcher and responsible for transforming input
data into the batcher's chunk format. Hence, the batcher needed to be aware
of its input types, although it would not otherwise use this information.

Drop the `Input` associated type and `push_container` method from the
`Batcher` trait; batchers now accept already-chunked input via
`PushInto<Self::Output>`. The vec `MergeBatcher` loses its `Input` and `C`
(chunker) type parameters, and the columnar `MergeBatcher` loses its internal
`TrieChunker`. Both now expose `PushInto` that inserts a chunk directly as a
chain.

Chunking moves into `arrange_core`, which gains a `Chu: ContainerBuilder`
type parameter so callers supply a chunker that maps the stream's input
container into the batcher's output container. The operator drives the
chunker (push, extract, and a `finish` drain before sealing) where the
batcher previously did.

The `Arrange` trait constrains `Ba::Output = C` and hardcodes
`ContainerChunker<C>` internally, so `.arrange::<Ba, Bu, Tr>()` callsites for
`Vec`-based collections are unchanged. Callers needing a cross-container
chunker (columnar layouts, interactive, spill) drop to `arrange_core`
directly and pass an explicit `ValChunker`.

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .../examples/columnar/main.rs                 |  9 ++-
 .../examples/columnar_spill.rs                | 19 ++++--
 differential-dataflow/examples/spines.rs      |  6 +-
 differential-dataflow/src/collection.rs       | 17 +++---
 .../src/columnar/arrangement/mod.rs           |  2 +
 differential-dataflow/src/columnar/batcher.rs | 41 +++++--------
 differential-dataflow/src/columnar/mod.rs     |  2 +-
 .../src/operators/arrange/arrangement.rs      | 33 ++++++++--
 .../trace/implementations/merge_batcher.rs    | 61 ++++++-------------
 .../src/trace/implementations/mod.rs          |  1 +
 .../src/trace/implementations/ord_neu.rs      |  5 +-
 differential-dataflow/src/trace/mod.rs        | 15 ++---
 differential-dataflow/tests/trace.rs          |  3 +-
 interactive/examples/ddir_col.rs              |  5 +-
 14 files changed, 110 insertions(+), 109 deletions(-)

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")
                 }
             }
         }