From d4e20a01c8071da1e356a428e3753d36faa508cf Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 21 Mar 2024 12:53:23 -0400
Subject: [PATCH 1/7] Update to track TD changes (#467)

---
 src/operators/arrange/arrangement.rs | 6 +++---
 src/operators/join.rs                | 2 +-
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index 031e587b9..fe65a702d 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -541,7 +541,7 @@ where
     /// be consistently by key (though this is the most common).
     fn arrange_core<P, Tr>(&self, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
-        P: ParallelizationContract<G::Timestamp, ((K,V),G::Timestamp,R)>,
+        P: ParallelizationContract<G::Timestamp, Vec<((K,V),G::Timestamp,R)>>,
         K: Clone,
         V: Clone,
         R: Clone,
@@ -562,7 +562,7 @@ where
 {
     fn arrange_core<P, Tr>(&self, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
-        P: ParallelizationContract<G::Timestamp, ((K,V),G::Timestamp,R)>,
+        P: ParallelizationContract<G::Timestamp, Vec<((K,V),G::Timestamp,R)>>,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
         Tr::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
@@ -727,7 +727,7 @@ where
 {
     fn arrange_core<P, Tr>(&self, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
-        P: ParallelizationContract<G::Timestamp, ((K,()),G::Timestamp,R)>,
+        P: ParallelizationContract<G::Timestamp, Vec<((K,()),G::Timestamp,R)>>,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
         Tr::Batcher: Batcher<Item = ((K,()),G::Timestamp,R), Time = G::Timestamp>,
diff --git a/src/operators/join.rs b/src/operators/join.rs
index ee1c28724..a31e682a5 100644
--- a/src/operators/join.rs
+++ b/src/operators/join.rs
@@ -617,7 +617,7 @@ where
 
     /// Process keys until at least `fuel` output tuples produced, or the work is exhausted.
     #[inline(never)]
-    fn work<L, I>(&mut self, output: &mut OutputHandle<T, (D, T, R), Tee<T, (D, T, R)>>, mut logic: L, fuel: &mut usize)
+    fn work<L, I>(&mut self, output: &mut OutputHandle<T, (D, T, R), Tee<T, Vec<(D, T, R)>>>, mut logic: L, fuel: &mut usize)
     where 
         I: IntoIterator<Item=(D, T, R)>,
         L: for<'a> FnMut(C1::Key<'a>, C1::Val<'a>, C2::Val<'a>, &T, &C1::Diff, &C2::Diff)->I,

From 84d0d4d6440971bfd6fefba112f631c98506dfc4 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 21 Mar 2024 15:34:52 -0400
Subject: [PATCH 2/7] Name Handle's container type (#468)

---
 src/operators/iterate.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/operators/iterate.rs b/src/operators/iterate.rs
index 8a60da4bb..6d0e27144 100644
--- a/src/operators/iterate.rs
+++ b/src/operators/iterate.rs
@@ -154,7 +154,7 @@ impl<G: Scope, D: Ord+Data+Debug, R: Semigroup> Iterate<G, D, R> for G {
 pub struct Variable<G: Scope, D: Data, R: Abelian>
 where G::Timestamp: Lattice {
     collection: Collection<G, D, R>,
-    feedback: Handle<G, (D, G::Timestamp, R)>,
+    feedback: Handle<G, Vec<(D, G::Timestamp, R)>>,
     source: Option<Collection<G, D, R>>,
     step: <G::Timestamp as Timestamp>::Summary,
 }
@@ -225,7 +225,7 @@ impl<G: Scope, D: Data, R: Abelian> Deref for Variable<G, D, R> where G::Timesta
 pub struct SemigroupVariable<G: Scope, D: Data, R: Semigroup>
 where G::Timestamp: Lattice {
     collection: Collection<G, D, R>,
-    feedback: Handle<G, (D, G::Timestamp, R)>,
+    feedback: Handle<G, Vec<(D, G::Timestamp, R)>>,
     step: <G::Timestamp as Timestamp>::Summary,
 }
 

From dfc0062b9ae34c6515568b84b215d2d2744413d8 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 21 Mar 2024 19:13:02 -0400
Subject: [PATCH 3/7] Prevent enter_at from using enter_at (#469)

---
 examples/accumulate.rs |  9 ++++++++-
 src/collection.rs      | 10 +++-------
 2 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/examples/accumulate.rs b/examples/accumulate.rs
index 4693431e4..0fa792926 100644
--- a/examples/accumulate.rs
+++ b/examples/accumulate.rs
@@ -16,7 +16,14 @@ fn main() {
 
         let mut input = worker.dataflow::<(), _, _>(|scope| {
             let (input, data) = scope.new_collection::<_, isize>();
-            data.consolidate();
+
+            use timely::dataflow::Scope;
+            scope.iterative::<u32,_,_>(|inner| {
+                data.enter_at(inner, |_| 0)
+                    .consolidate()
+                    .leave()
+            });
+
             input
         });
 
diff --git a/src/collection.rs b/src/collection.rs
index aeb6aad38..2e82c7e62 100644
--- a/src/collection.rs
+++ b/src/collection.rs
@@ -322,20 +322,16 @@ impl<G: Scope, D: Data, R: Semigroup> Collection<G, D, R> where G::Timestamp: Da
     ///     data.assert_eq(&result);
     /// });
     /// ```
-    pub fn enter_at<'a, T, F>(&self, child: &Iterative<'a, G, T>, initial: F) -> Collection<Iterative<'a, G, T>, D, R>
+    pub fn enter_at<'a, T, F>(&self, child: &Iterative<'a, G, T>, mut initial: F) -> Collection<Iterative<'a, G, T>, D, R>
     where
         T: Timestamp+Hash,
         F: FnMut(&D) -> T + Clone + 'static,
         G::Timestamp: Hash,
     {
-
-        let mut initial1 = initial.clone();
-        let mut initial2 = initial.clone();
-
         self.inner
-            .enter_at(child, move |x| initial1(&x.0))
+            .enter(child)
             .map(move |(data, time, diff)| {
-                let new_time = Product::new(time, initial2(&data));
+                let new_time = Product::new(time, initial(&data));
                 (data, new_time, diff)
             })
             .as_collection()

From 1dee988cc90badf79bb6b1d7dad9d9e91e8e90a2 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 23 Mar 2024 03:30:42 -0400
Subject: [PATCH 4/7] Track TD Capture changes (#470)

---
 tests/import.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/import.rs b/tests/import.rs
index 8176f14af..1a15d3d7a 100644
--- a/tests/import.rs
+++ b/tests/import.rs
@@ -9,7 +9,7 @@ use differential_dataflow::operators::reduce::Reduce;
 use differential_dataflow::trace::TraceReader;
 use itertools::Itertools;
 
-type Result = std::sync::mpsc::Receiver<timely::dataflow::operators::capture::Event<usize, ((u64, i64), usize, i64)>>;
+type Result = std::sync::mpsc::Receiver<timely::dataflow::operators::capture::Event<usize, Vec<((u64, i64), usize, i64)>>>;
 
 fn run_test<T>(test: T, expected: Vec<(usize, Vec<((u64, i64), i64)>)>) -> ()
         where T: FnOnce(Vec<Vec<((u64, u64), i64)>>)-> Result + ::std::panic::UnwindSafe

From 51966ad62680bd7c674e786dd56d1c8daec8d9d7 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Fri, 12 Apr 2024 14:11:07 -0400
Subject: [PATCH 5/7] WIP: Extract batcher input to assoc type, arrange_core
 freestanding (#471)

* Extract batcher input to assoc type, arrange_core freestanding

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

* Fix warnings

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

* documentation

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

* formatting

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

---------

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>
---
 .github/workflows/deploy.yml                  |   2 +-
 .github/workflows/test.yml                    |   4 +-
 doop/src/main.rs                              |   8 +-
 experiments/src/bin/graphs-interactive-alt.rs |   4 +-
 experiments/src/bin/graphs-interactive-neu.rs |   4 +-
 interactive/src/command.rs                    |   6 +-
 interactive/src/logging.rs                    |   6 +-
 src/operators/arrange/arrangement.rs          | 284 +++++++++---------
 src/operators/consolidate.rs                  |   2 +-
 src/trace/implementations/merge_batcher.rs    |   1 +
 .../implementations/merge_batcher_col.rs      |   1 +
 src/trace/mod.rs                              |   4 +-
 12 files changed, 175 insertions(+), 151 deletions(-)

diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml
index babc088a8..41c27604b 100644
--- a/.github/workflows/deploy.yml
+++ b/.github/workflows/deploy.yml
@@ -9,7 +9,7 @@ jobs:
   deploy:
     runs-on: ubuntu-22.04
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - run: cargo install mdbook --version 0.4.31
       - run: cd mdbook && mdbook build
       - uses: JamesIves/github-pages-deploy-action@v4
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index e4643725d..e3c9b6087 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -16,7 +16,7 @@ jobs:
         toolchain:
           - stable
           - 1.72
-    name: cargo test on ${{ matrix.os }}
+    name: cargo test on ${{ matrix.os }}, rust ${{ matrix.toolchain }}
     runs-on: ${{ matrix.os }}-latest
     steps:
       - uses: actions/checkout@v4
@@ -24,7 +24,7 @@ jobs:
         with:
           toolchain: ${{ matrix.toolchain }}
       - name: Cargo test
-        run: cargo test
+        run: cargo test --workspace --all-targets
 
   # Check formatting with rustfmt
   mdbook:
diff --git a/doop/src/main.rs b/doop/src/main.rs
index 6628a7e30..d6ad19c19 100644
--- a/doop/src/main.rs
+++ b/doop/src/main.rs
@@ -1,4 +1,4 @@
-#![allow(non_snake_case)]
+#![allow(non_snake_case, dead_code)]
 
 use std::collections::HashMap;
 use std::rc::Rc;
@@ -145,7 +145,7 @@ fn load<'a>(filename: &str, interner: Rc<RefCell<StringInterner>>) -> impl Itera
         })
 }
 
-fn load1<'a>(index: usize, prefix: &str, filename: &str, interner: Rc<RefCell<StringInterner>>) -> impl Iterator<Item=((Symbol), Time, Diff)>+'a {
+fn load1<'a>(index: usize, prefix: &str, filename: &str, interner: Rc<RefCell<StringInterner>>) -> impl Iterator<Item=(Symbol, Time, Diff)>+'a {
     read_file(&format!("{}{}", prefix, filename))
         .filter(move |_| index == 0)
         .map(move |line| {
@@ -791,7 +791,7 @@ fn main() {
                 let SupertypeOf = SupertypeOf.enter(scope);
 
                 // Required by all
-                let mut Reachable = Relation::<_,(Method)>::new(scope);
+                let mut Reachable = Relation::<_,Method>::new(scope);
 
                 // NOTE: Common subexpression.
                 let Reachable_Invocation =
@@ -805,7 +805,7 @@ fn main() {
                 // let Reachable = ReachableFinal.clone();
 
                 // Class initialization
-                let mut InitializedClass = Relation::<_,(Type)>::new(scope);
+                let mut InitializedClass = Relation::<_,Type>::new(scope);
 
                 // ClassInitializer(?type, ?method) :- basic.MethodImplemented("<clinit>", "void()", ?type, ?method).
                 let temp1 = interner.borrow_mut().intern("<clinit>");
diff --git a/experiments/src/bin/graphs-interactive-alt.rs b/experiments/src/bin/graphs-interactive-alt.rs
index 55c8a8cf6..18dc0923f 100644
--- a/experiments/src/bin/graphs-interactive-alt.rs
+++ b/experiments/src/bin/graphs-interactive-alt.rs
@@ -1,3 +1,5 @@
+#![allow(dead_code)]
+
 use rand::{Rng, SeedableRng, StdRng};
 
 use timely::dataflow::*;
@@ -389,4 +391,4 @@ where G::Timestamp: Lattice {
                 .concat(&prop)
                 .reduce(|_, s, t| { t.push((*s[0].0, 1)); })
         })
-}
\ No newline at end of file
+}
diff --git a/experiments/src/bin/graphs-interactive-neu.rs b/experiments/src/bin/graphs-interactive-neu.rs
index a01f23d97..6210113ed 100644
--- a/experiments/src/bin/graphs-interactive-neu.rs
+++ b/experiments/src/bin/graphs-interactive-neu.rs
@@ -1,3 +1,5 @@
+#![allow(dead_code)]
+
 use rand::{Rng, SeedableRng, StdRng};
 
 use timely::dataflow::*;
@@ -392,4 +394,4 @@ where G::Timestamp: Lattice+Ord {
 
         reached.leave()
     })
-}
\ No newline at end of file
+}
diff --git a/interactive/src/command.rs b/interactive/src/command.rs
index c31963afe..d795ec13a 100644
--- a/interactive/src/command.rs
+++ b/interactive/src/command.rs
@@ -151,7 +151,7 @@ where
                                 println!("\tTimely logging connection {} of {}", index, number);
                                 let socket = listener.incoming().next().unwrap().unwrap();
                                 socket.set_nonblocking(true).expect("failed to set nonblocking");
-                                streams.push(EventReader::<Duration, (Duration, usize, TimelyEvent),_>::new(socket));
+                                streams.push(EventReader::<Duration, Vec<(Duration, usize, TimelyEvent)>,_>::new(socket));
                             }
 
                             println!("\tAll logging connections established");
@@ -174,7 +174,7 @@ where
                             for _ in 0 .. number {
                                 let socket = listener.incoming().next().unwrap().unwrap();
                                 socket.set_nonblocking(true).expect("failed to set nonblocking");
-                                streams.push(EventReader::<Duration, (Duration, usize, DifferentialEvent),_>::new(socket));
+                                streams.push(EventReader::<Duration, Vec<(Duration, usize, DifferentialEvent)>,_>::new(socket));
                             }
                         }
                         crate::logging::publish_differential_logging(manager, worker, granularity, &name_as, streams);
@@ -195,4 +195,4 @@ where
     pub fn serialize_into<W: Write>(&self, writer: W) {
         bincode::serialize_into(writer, self).expect("bincode: serialization failed");
     }
-}
\ No newline at end of file
+}
diff --git a/interactive/src/logging.rs b/interactive/src/logging.rs
index cfc524660..be07fa4b6 100644
--- a/interactive/src/logging.rs
+++ b/interactive/src/logging.rs
@@ -30,7 +30,7 @@ where
     V: ExchangeData+Hash+LoggingValue+Datum,
     A: Allocate,
     I : IntoIterator,
-    <I as IntoIterator>::Item: EventIterator<Duration, (Duration, usize, TimelyEvent)>+'static
+    <I as IntoIterator>::Item: EventIterator<Duration, Vec<(Duration, usize, TimelyEvent)>>+'static
 {
     let (operates, channels, schedule, messages, shutdown, park, text) =
     worker.dataflow(move |scope| {
@@ -217,7 +217,7 @@ where
     V: ExchangeData+Hash+LoggingValue+Datum,
     A: Allocate,
     I : IntoIterator,
-    <I as IntoIterator>::Item: EventIterator<Duration, (Duration, usize, DifferentialEvent)>+'static
+    <I as IntoIterator>::Item: EventIterator<Duration, Vec<(Duration, usize, DifferentialEvent)>>+'static
 {
     let (merge,batch) =
     worker.dataflow(move |scope| {
@@ -280,4 +280,4 @@ where
 
     manager.traces.set_unkeyed(&Plan::Source(format!("logs/{}/differential/arrange/batch", name)), &batch);
     manager.traces.set_unkeyed(&Plan::Source(format!("logs/{}/differential/arrange/merge", name)), &merge);
-}
\ No newline at end of file
+}
diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index fe65a702d..653ed1e87 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -19,7 +19,7 @@
 
 use timely::dataflow::operators::{Enter, Map};
 use timely::order::PartialOrder;
-use timely::dataflow::{Scope, Stream};
+use timely::dataflow::{Scope, Stream, StreamCore};
 use timely::dataflow::operators::generic::Operator;
 use timely::dataflow::channels::pact::{ParallelizationContract, Pipeline, Exchange};
 use timely::progress::Timestamp;
@@ -511,7 +511,7 @@ where
         V: ExchangeData,
         R: ExchangeData,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
         Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         self.arrange_named("Arrange")
@@ -527,7 +527,7 @@ where
         V: ExchangeData,
         R: ExchangeData,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
         Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         let exchange = Exchange::new(move |update: &((K,V),G::Timestamp,R)| (update.0).0.hashed().into());
@@ -547,7 +547,7 @@ where
         R: Clone,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
         Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     ;
 }
@@ -565,160 +565,176 @@ where
         P: ParallelizationContract<G::Timestamp, Vec<((K,V),G::Timestamp,R)>>,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
+        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
         Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
-        // The `Arrange` operator is tasked with reacting to an advancing input
-        // frontier by producing the sequence of batches whose lower and upper
-        // bounds are those frontiers, containing updates at times greater or
-        // equal to lower and not greater or equal to upper.
-        //
-        // The operator uses its batch type's `Batcher`, which accepts update
-        // triples and responds to requests to "seal" batches (presented as new
-        // upper frontiers).
-        //
-        // Each sealed batch is presented to the trace, and if at all possible
-        // transmitted along the outgoing channel. Empty batches may not have
-        // a corresponding capability, as they are only retained for actual data
-        // held by the batcher, which may prevents the operator from sending an
-        // empty batch.
-
-        let mut reader: Option<TraceAgent<Tr>> = None;
-
-        // fabricate a data-parallel operator using the `unary_notify` pattern.
-        let stream = {
-
-            let reader = &mut reader;
-
-            self.inner.unary_frontier(pact, name, move |_capability, info| {
-
-                // Acquire a logger for arrange events.
-                let logger = {
-                    let scope = self.scope();
-                    let register = scope.log_register();
-                    register.get::<crate::logging::DifferentialEvent>("differential/arrange")
-                };
-
-                // Where we will deposit received updates, and from which we extract batches.
-                let mut batcher = Tr::Batcher::new(logger.clone(), info.global_id);
-
-                // Capabilities for the lower envelope of updates in `batcher`.
-                let mut capabilities = Antichain::<Capability<G::Timestamp>>::new();
-
-                let activator = Some(self.scope().activator_for(&info.address[..]));
-                let mut empty_trace = Tr::new(info.clone(), logger.clone(), activator);
-                // If there is default exertion logic set, install it.
-                if let Some(exert_logic) = self.inner.scope().config().get::<trace::ExertionLogic>("differential/default_exert_logic").cloned() {
-                    empty_trace.set_exert_logic(exert_logic);
-                }
-
-                let (reader_local, mut writer) = TraceAgent::new(empty_trace, info, logger);
-
-                *reader = Some(reader_local);
-
-                // Initialize to the minimal input frontier.
-                let mut prev_frontier = Antichain::from_elem(<G::Timestamp as Timestamp>::minimum());
-
-                move |input, output| {
+        arrange_core(&self.inner, pact, name)
+    }
+}
 
-                    // As we receive data, we need to (i) stash the data and (ii) keep *enough* capabilities.
-                    // We don't have to keep all capabilities, but we need to be able to form output messages
-                    // when we realize that time intervals are complete.
+/// Arranges a stream of  updates by a key, configured with a name and a parallelization contract.
+///
+/// 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).
+fn arrange_core<G, P, Tr>(stream: &StreamCore<G, <Tr::Batcher as Batcher>::Input>, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
+where
+    G: Scope,
+    G::Timestamp: Lattice,
+    P: ParallelizationContract<G::Timestamp, <Tr::Batcher as Batcher>::Input>,
+    Tr: Trace<Time=G::Timestamp>+'static,
+    Tr::Batch: Batch,
+    Tr::Batcher: Batcher<Time = G::Timestamp>,
+    <Tr::Batcher as Batcher>::Input: timely::Container,
+    Tr::Builder: Builder<Time = G::Timestamp, Output = Tr::Batch>,
+{
+    // The `Arrange` operator is tasked with reacting to an advancing input
+    // frontier by producing the sequence of batches whose lower and upper
+    // bounds are those frontiers, containing updates at times greater or
+    // equal to lower and not greater or equal to upper.
+    //
+    // The operator uses its batch type's `Batcher`, which accepts update
+    // triples and responds to requests to "seal" batches (presented as new
+    // upper frontiers).
+    //
+    // Each sealed batch is presented to the trace, and if at all possible
+    // transmitted along the outgoing channel. Empty batches may not have
+    // a corresponding capability, as they are only retained for actual data
+    // held by the batcher, which may prevents the operator from sending an
+    // empty batch.
+
+    let mut reader: Option<TraceAgent<Tr>> = None;
+
+    // fabricate a data-parallel operator using the `unary_notify` pattern.
+    let reader_ref = &mut reader;
+    let scope = stream.scope();
+
+    let stream = stream.unary_frontier(pact, name, move |_capability, info| {
+
+        // Acquire a logger for arrange events.
+        let logger = {
+            let register = scope.log_register();
+            register.get::<crate::logging::DifferentialEvent>("differential/arrange")
+        };
 
-                    input.for_each(|cap, data| {
-                        capabilities.insert(cap.retain());
-                        batcher.push_batch(data);
-                    });
+        // Where we will deposit received updates, and from which we extract batches.
+        let mut batcher = Tr::Batcher::new(logger.clone(), info.global_id);
 
-                    // The frontier may have advanced by multiple elements, which is an issue because
-                    // timely dataflow currently only allows one capability per message. This means we
-                    // must pretend to process the frontier advances one element at a time, batching
-                    // and sending smaller bites than we might have otherwise done.
+        // Capabilities for the lower envelope of updates in `batcher`.
+        let mut capabilities = Antichain::<Capability<G::Timestamp>>::new();
 
-                    // Assert that the frontier never regresses.
-                    assert!(PartialOrder::less_equal(&prev_frontier.borrow(), &input.frontier().frontier()));
+        let activator = Some(scope.activator_for(&info.address[..]));
+        let mut empty_trace = Tr::new(info.clone(), logger.clone(), activator);
+        // If there is default exertion logic set, install it.
+        if let Some(exert_logic) = scope.config().get::<trace::ExertionLogic>("differential/default_exert_logic").cloned() {
+            empty_trace.set_exert_logic(exert_logic);
+        }
 
-                    // Test to see if strict progress has occurred, which happens whenever the new
-                    // 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() != input.frontier().frontier() {
-                        // There are two cases to handle with some care:
-                        //
-                        // 1. If any held capabilities are not in advance of the new input frontier,
-                        //    we must carve out updates now in advance of the new input frontier and
-                        //    transmit them as batches, which requires appropriate *single* capabilites;
-                        //    Until timely dataflow supports multiple capabilities on messages, at least.
-                        //
-                        // 2. If there are no held capabilities in advance of the new input frontier,
-                        //    then there are no updates not in advance of the new input frontier and
-                        //    we can simply create an empty input batch with the new upper frontier
-                        //    and feed this to the trace agent (but not along the timely output).
+        let (reader_local, mut writer) = TraceAgent::new(empty_trace, info, logger);
 
-                        // If there is at least one capability not in advance of the input frontier ...
-                        if capabilities.elements().iter().any(|c| !input.frontier().less_equal(c.time())) {
+        *reader_ref = Some(reader_local);
 
-                            let mut upper = Antichain::new();   // re-used allocation for sealing batches.
+        // Initialize to the minimal input frontier.
+        let mut prev_frontier = Antichain::from_elem(<G::Timestamp as Timestamp>::minimum());
 
-                            // For each capability not in advance of the input frontier ...
-                            for (index, capability) in capabilities.elements().iter().enumerate() {
+        move |input, output| {
 
-                                if !input.frontier().less_equal(capability.time()) {
+            // As we receive data, we need to (i) stash the data and (ii) keep *enough* capabilities.
+            // We don't have to keep all capabilities, but we need to be able to form output messages
+            // when we realize that time intervals are complete.
 
-                                    // Assemble the upper bound on times we can commit with this capabilities.
-                                    // We must respect the input frontier, and *subsequent* capabilities, as
-                                    // we are pretending to retire the capability changes one by one.
-                                    upper.clear();
-                                    for time in input.frontier().frontier().iter() {
-                                        upper.insert(time.clone());
-                                    }
-                                    for other_capability in &capabilities.elements()[(index + 1) .. ] {
-                                        upper.insert(other_capability.time().clone());
-                                    }
+            input.for_each(|cap, data| {
+                capabilities.insert(cap.retain());
+                batcher.push_batch(data);
+            });
 
-                                    // Extract updates not in advance of `upper`.
-                                    let batch = batcher.seal::<Tr::Builder>(upper.clone());
+            // The frontier may have advanced by multiple elements, which is an issue because
+            // timely dataflow currently only allows one capability per message. This means we
+            // must pretend to process the frontier advances one element at a time, batching
+            // and sending smaller bites than we might have otherwise done.
+
+            // Assert that the frontier never regresses.
+            assert!(PartialOrder::less_equal(&prev_frontier.borrow(), &input.frontier().frontier()));
+
+            // Test to see if strict progress has occurred, which happens whenever the new
+            // 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() != input.frontier().frontier() {
+                // There are two cases to handle with some care:
+                //
+                // 1. If any held capabilities are not in advance of the new input frontier,
+                //    we must carve out updates now in advance of the new input frontier and
+                //    transmit them as batches, which requires appropriate *single* capabilites;
+                //    Until timely dataflow supports multiple capabilities on messages, at least.
+                //
+                // 2. If there are no held capabilities in advance of the new input frontier,
+                //    then there are no updates not in advance of the new input frontier and
+                //    we can simply create an empty input batch with the new upper frontier
+                //    and feed this to the trace agent (but not along the timely output).
+
+                // If there is at least one capability not in advance of the input frontier ...
+                if capabilities.elements().iter().any(|c| !input.frontier().less_equal(c.time())) {
+
+                    let mut upper = Antichain::new();   // re-used allocation for sealing batches.
+
+                    // For each capability not in advance of the input frontier ...
+                    for (index, capability) in capabilities.elements().iter().enumerate() {
+
+                        if !input.frontier().less_equal(capability.time()) {
+
+                            // Assemble the upper bound on times we can commit with this capabilities.
+                            // We must respect the input frontier, and *subsequent* capabilities, as
+                            // we are pretending to retire the capability changes one by one.
+                            upper.clear();
+                            for time in input.frontier().frontier().iter() {
+                                upper.insert(time.clone());
+                            }
+                            for other_capability in &capabilities.elements()[(index + 1) .. ] {
+                                upper.insert(other_capability.time().clone());
+                            }
 
-                                    writer.insert(batch.clone(), Some(capability.time().clone()));
+                            // Extract updates not in advance of `upper`.
+                            let batch = batcher.seal::<Tr::Builder>(upper.clone());
 
-                                    // send the batch to downstream consumers, empty or not.
-                                    output.session(&capabilities.elements()[index]).give(batch);
-                                }
-                            }
+                            writer.insert(batch.clone(), Some(capability.time().clone()));
 
-                            // Having extracted and sent batches between each capability and the input frontier,
-                            // we should 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.
+                            // send the batch to downstream consumers, empty or not.
+                            output.session(&capabilities.elements()[index]).give(batch);
+                        }
+                    }
 
-                            let mut new_capabilities = Antichain::new();
-                            for time in batcher.frontier().iter() {
-                                if let Some(capability) = capabilities.elements().iter().find(|c| c.time().less_equal(time)) {
-                                    new_capabilities.insert(capability.delayed(time));
-                                }
-                                else {
-                                    panic!("failed to find capability");
-                                }
-                            }
+                    // Having extracted and sent batches between each capability and the input frontier,
+                    // we should 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.
 
-                            capabilities = new_capabilities;
+                    let mut new_capabilities = Antichain::new();
+                    for time in batcher.frontier().iter() {
+                        if let Some(capability) = capabilities.elements().iter().find(|c| c.time().less_equal(time)) {
+                            new_capabilities.insert(capability.delayed(time));
                         }
                         else {
-                            // Announce progress updates, even without data.
-                            let _batch = batcher.seal::<Tr::Builder>(input.frontier().frontier().to_owned());
-                            writer.seal(input.frontier().frontier().to_owned());
+                            panic!("failed to find capability");
                         }
-
-                        prev_frontier.clear();
-                        prev_frontier.extend(input.frontier().frontier().iter().cloned());
                     }
 
-                    writer.exert();
+                    capabilities = new_capabilities;
+                }
+                else {
+                    // Announce progress updates, even without data.
+                    let _batch = batcher.seal::<Tr::Builder>(input.frontier().frontier().to_owned());
+                    writer.seal(input.frontier().frontier().to_owned());
                 }
-            })
-        };
 
-        Arranged { stream, trace: reader.unwrap() }
-    }
+                prev_frontier.clear();
+                prev_frontier.extend(input.frontier().frontier().iter().cloned());
+            }
+
+            writer.exert();
+        }
+    });
+
+    Arranged { stream, trace: reader.unwrap() }
 }
 
 impl<G: Scope, K: ExchangeData+Hashable, R: ExchangeData+Semigroup> Arrange<G, K, (), R> for Collection<G, K, R>
@@ -730,7 +746,7 @@ where
         P: ParallelizationContract<G::Timestamp, Vec<((K,()),G::Timestamp,R)>>,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Item = ((K,()),G::Timestamp,R), Time = G::Timestamp>,
+        Tr::Batcher: Batcher<Input=Vec<((K,()),G::Timestamp,R)>, Item = ((K,()),G::Timestamp,R), Time = G::Timestamp>,
         Tr::Builder: Builder<Item = ((K,()),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
     {
         self.map(|k| (k, ()))
diff --git a/src/operators/consolidate.rs b/src/operators/consolidate.rs
index 5339a54aa..52fb8919f 100644
--- a/src/operators/consolidate.rs
+++ b/src/operators/consolidate.rs
@@ -53,7 +53,7 @@ where
     where
         Tr: crate::trace::Trace<KeyOwned = D,ValOwned = (),Time=G::Timestamp,Diff=R>+'static,
         Tr::Batch: crate::trace::Batch,
-        Tr::Batcher: Batcher<Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
+        Tr::Batcher: Batcher<Input=Vec<((D,()),G::Timestamp,R)>, Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
         Tr::Builder: Builder<Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
     {
         use crate::operators::arrange::arrangement::Arrange;
diff --git a/src/trace/implementations/merge_batcher.rs b/src/trace/implementations/merge_batcher.rs
index 680aa5306..673fedce6 100644
--- a/src/trace/implementations/merge_batcher.rs
+++ b/src/trace/implementations/merge_batcher.rs
@@ -25,6 +25,7 @@ where
     T: Timestamp,
     D: Semigroup,
 {
+    type Input = Vec<Self::Item>;
     type Item = ((K,V),T,D);
     type Time = T;
 
diff --git a/src/trace/implementations/merge_batcher_col.rs b/src/trace/implementations/merge_batcher_col.rs
index 9c2faab12..7989eece8 100644
--- a/src/trace/implementations/merge_batcher_col.rs
+++ b/src/trace/implementations/merge_batcher_col.rs
@@ -31,6 +31,7 @@ where
     T: Columnation + Timestamp + 'static,
     D: Columnation + Semigroup + 'static,
 {
+    type Input = Vec<Self::Item>;
     type Item = ((K,V),T,D);
     type Time = T;
 
diff --git a/src/trace/mod.rs b/src/trace/mod.rs
index 136654689..d0d6791e3 100644
--- a/src/trace/mod.rs
+++ b/src/trace/mod.rs
@@ -307,13 +307,15 @@ pub trait Batch : BatchReader where Self: ::std::marker::Sized {
 /// Functionality for collecting and batching updates.
 pub trait Batcher {
     /// Type of update pushed into the batcher.
+    type Input;
+    /// Type of update pushed into the builder.
     type Item;
     /// Times at which batches are formed.
     type Time: Timestamp;
     /// Allocates a new empty batcher.
     fn new(logger: Option<Logger<DifferentialEvent, WorkerIdentifier>>, operator_id: usize) -> Self;
     /// Adds an unordered batch of elements to the batcher.
-    fn push_batch(&mut self, batch: RefOrMut<Vec<Self::Item>>);
+    fn push_batch(&mut self, batch: RefOrMut<Self::Input>);
     /// Returns all updates not greater or equal to an element of `upper`.
     fn seal<B: Builder<Item=Self::Item, Time=Self::Time>>(&mut self, upper: Antichain<Self::Time>) -> B::Output;
     /// Returns the lower envelope of contained update times.

From 37c505d2c607cc37cd9e6a913ebd06fa6196647c Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 12 Apr 2024 17:04:36 -0400
Subject: [PATCH 6/7] `Arrange` cleanup (#472)

* Remove Item constraint from arrange_core

* Remove redundant trait constraints

* Replace K,V,R Arrange type parameters by C
---
 src/operators/arrange/arrangement.rs | 89 ++++++++++++----------------
 1 file changed, 38 insertions(+), 51 deletions(-)

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index 653ed1e87..7995b4397 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -489,90 +489,70 @@ where
     }
 }
 
-/// A type that can be arranged as if a collection of updates shaped as `((K,V),G::Timestamp,R)`.
-///
-/// This trait is primarily implemented by `Collection<G,(K,V),R>`.
-///
-/// The resulting arrangements may not present as `((K,V),T,R)`, as their output types are unconstrained.
-/// This allows e.g. for `Vec<u8>` inputs to present as `&[u8]` when read, but that relationship is not
-/// constrained by this trait.
-pub trait Arrange<G, K, V, R>
+/// A type that can be arranged as if a collection of updates.
+pub trait Arrange<G, C>
 where
     G: Scope,
     G::Timestamp: Lattice,
 {
-    /// Arranges a stream of `(Key, Val)` updates by `Key`.
-    ///
-    /// This operator arranges a stream of values into a shared trace, whose contents it maintains.
+    /// Arranges updates into a shared trace.
     fn arrange<Tr>(&self) -> Arranged<G, TraceAgent<Tr>>
     where
         Tr: Trace<Time=G::Timestamp> + 'static,
-        K: ExchangeData + Hashable,
-        V: ExchangeData,
-        R: ExchangeData,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
-        Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
+        Tr::Batcher: Batcher<Input=C>,
     {
         self.arrange_named("Arrange")
     }
 
-    /// Arranges a stream of `(Key, Val)` updates by `Key`, and presents with a `name` argument.
-    ///
-    /// This operator arranges a stream of values into a shared trace, whose contents it maintains.
+    /// Arranges updates into a shared trace, with a supplied name.
     fn arrange_named<Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
         Tr: Trace<Time=G::Timestamp> + 'static,
-        K: ExchangeData + Hashable,
-        V: ExchangeData,
-        R: ExchangeData,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
-        Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
-    {
-        let exchange = Exchange::new(move |update: &((K,V),G::Timestamp,R)| (update.0).0.hashed().into());
-        self.arrange_core(exchange, name)
-    }
+        Tr::Batcher: Batcher<Input=C>,
+    ;
 
-    /// Arranges a stream of `(Key, Val)` updates by `Key`, configured with a name and a parallelization contract.
-    ///
-    /// 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).
+    /// Arranges updates into a shared trace, using a supplied parallelization contract, with a supplied name.
     fn arrange_core<P, Tr>(&self, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
-        P: ParallelizationContract<G::Timestamp, Vec<((K,V),G::Timestamp,R)>>,
-        K: Clone,
-        V: Clone,
-        R: Clone,
+        P: ParallelizationContract<G::Timestamp, C>,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
-        Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
+        Tr::Batcher: Batcher<Input=C>,
     ;
 }
 
-impl<G, K, V, R> Arrange<G, K, V, R> for Collection<G, (K, V), R>
+impl<G, K, V, R> Arrange<G, Vec<((K, V), G::Timestamp, R)>> for Collection<G, (K, V), R>
 where
     G: Scope,
     G::Timestamp: Lattice,
-    K: Clone + 'static,
-    V: Clone + 'static,
-    R: Semigroup,
+    K: ExchangeData + Hashable,
+    V: ExchangeData,
+    R: ExchangeData + Semigroup,
 {
+    fn arrange_named<Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
+    where
+        Tr: Trace<Time=G::Timestamp> + 'static,
+        Tr::Batch: Batch,
+        Tr::Batcher: Batcher<Input=Vec<((K, V), G::Timestamp, R)>>,
+    {
+        let exchange = Exchange::new(move |update: &((K,V),G::Timestamp,R)| (update.0).0.hashed().into());
+        self.arrange_core(exchange, name)
+    }
+
     fn arrange_core<P, Tr>(&self, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
         P: ParallelizationContract<G::Timestamp, Vec<((K,V),G::Timestamp,R)>>,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
-        Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
+        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>>,
     {
         arrange_core(&self.inner, pact, name)
     }
 }
 
-/// Arranges a stream of  updates by a key, configured with a name and a parallelization contract.
+/// Arranges a stream of updates by a key, configured with a name and a parallelization contract.
 ///
 /// 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
@@ -584,9 +564,7 @@ where
     P: ParallelizationContract<G::Timestamp, <Tr::Batcher as Batcher>::Input>,
     Tr: Trace<Time=G::Timestamp>+'static,
     Tr::Batch: Batch,
-    Tr::Batcher: Batcher<Time = G::Timestamp>,
     <Tr::Batcher as Batcher>::Input: timely::Container,
-    Tr::Builder: Builder<Time = G::Timestamp, Output = Tr::Batch>,
 {
     // The `Arrange` operator is tasked with reacting to an advancing input
     // frontier by producing the sequence of batches whose lower and upper
@@ -737,17 +715,26 @@ where
     Arranged { stream, trace: reader.unwrap() }
 }
 
-impl<G: Scope, K: ExchangeData+Hashable, R: ExchangeData+Semigroup> Arrange<G, K, (), R> for Collection<G, K, R>
+impl<G: Scope, K: ExchangeData+Hashable, R: ExchangeData+Semigroup> Arrange<G, Vec<((K, ()), G::Timestamp, R)>> for Collection<G, K, R>
 where
     G::Timestamp: Lattice+Ord,
 {
+    fn arrange_named<Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
+    where
+        Tr: Trace<Time=G::Timestamp> + 'static,
+        Tr::Batch: Batch,
+        Tr::Batcher: Batcher<Input=Vec<((K, ()), G::Timestamp, R)>>,
+    {
+        let exchange = Exchange::new(move |update: &((K,()),G::Timestamp,R)| (update.0).0.hashed().into());
+        self.arrange_core(exchange, name)
+    }
+
     fn arrange_core<P, Tr>(&self, pact: P, name: &str) -> Arranged<G, TraceAgent<Tr>>
     where
         P: ParallelizationContract<G::Timestamp, Vec<((K,()),G::Timestamp,R)>>,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=Vec<((K,()),G::Timestamp,R)>, Item = ((K,()),G::Timestamp,R), Time = G::Timestamp>,
-        Tr::Builder: Builder<Item = ((K,()),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
+        Tr::Batcher: Batcher<Input=Vec<((K,()),G::Timestamp,R)>>,
     {
         self.map(|k| (k, ()))
             .arrange_core(pact, name)

From c564e8f19a6ee8d140864af66a29f050309c84ac Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 13 Apr 2024 18:47:11 -0400
Subject: [PATCH 7/7] Further clean-up (#473)

---
 src/operators/arrange/arrangement.rs          |  4 +--
 src/operators/arrange/upsert.rs               |  2 +-
 src/operators/consolidate.rs                  |  5 ++-
 src/operators/reduce.rs                       |  8 ++---
 src/trace/implementations/merge_batcher.rs    |  8 ++---
 .../implementations/merge_batcher_col.rs      |  8 ++---
 src/trace/implementations/ord_neu.rs          | 12 +++----
 src/trace/implementations/rhh.rs              |  6 ++--
 src/trace/implementations/spine_fueled.rs     |  2 +-
 src/trace/mod.rs                              | 32 ++++++++-----------
 10 files changed, 41 insertions(+), 46 deletions(-)

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index 7995b4397..9856e38e0 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -444,7 +444,7 @@ where
         T2::ValOwned: Data,
         T2::Diff: Abelian,
         T2::Batch: Batch,
-        T2::Builder: Builder<Output=T2::Batch, Item = ((T1::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
+        T2::Builder: Builder<Input = ((T1::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
         L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(<T2::Cursor as Cursor>::ValOwned, T2::Diff)>)+'static,
     {
         self.reduce_core::<_,T2>(name, move |key, input, output, change| {
@@ -462,7 +462,7 @@ where
         T2: for<'a> Trace<Key<'a>=T1::Key<'a>, Time=T1::Time>+'static,
         T2::ValOwned: Data,
         T2::Batch: Batch,
-        T2::Builder: Builder<Output=T2::Batch, Item = ((T1::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
+        T2::Builder: Builder<Input = ((T1::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
         L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(<T2::Cursor as Cursor>::ValOwned,T2::Diff)>, &mut Vec<(<T2::Cursor as Cursor>::ValOwned, T2::Diff)>)+'static,
     {
         use crate::operators::reduce::reduce_trace;
diff --git a/src/operators/arrange/upsert.rs b/src/operators/arrange/upsert.rs
index e9d1744ff..4ca9dc9b3 100644
--- a/src/operators/arrange/upsert.rs
+++ b/src/operators/arrange/upsert.rs
@@ -136,7 +136,7 @@ where
     Tr::ValOwned: ExchangeData,
     Tr::Time: TotalOrder+ExchangeData,
     Tr::Batch: Batch,
-    Tr::Builder: Builder<Item = ((Tr::KeyOwned, Tr::ValOwned), Tr::Time, Tr::Diff)>,
+    Tr::Builder: Builder<Input = ((Tr::KeyOwned, Tr::ValOwned), Tr::Time, Tr::Diff)>,
 {
     let mut reader: Option<TraceAgent<Tr>> = None;
 
diff --git a/src/operators/consolidate.rs b/src/operators/consolidate.rs
index 52fb8919f..b67c1c4b8 100644
--- a/src/operators/consolidate.rs
+++ b/src/operators/consolidate.rs
@@ -13,7 +13,7 @@ use crate::difference::Semigroup;
 
 use crate::Data;
 use crate::lattice::Lattice;
-use crate::trace::{Batcher, Builder};
+use crate::trace::Batcher;
 
 /// Methods which require data be arrangeable.
 impl<G, D, R> Collection<G, D, R>
@@ -53,8 +53,7 @@ where
     where
         Tr: crate::trace::Trace<KeyOwned = D,ValOwned = (),Time=G::Timestamp,Diff=R>+'static,
         Tr::Batch: crate::trace::Batch,
-        Tr::Batcher: Batcher<Input=Vec<((D,()),G::Timestamp,R)>, Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
-        Tr::Builder: Builder<Item = ((D,()),G::Timestamp,R), Time = G::Timestamp>,
+        Tr::Batcher: Batcher<Input=Vec<((D,()),G::Timestamp,R)>>,
     {
         use crate::operators::arrange::arrangement::Arrange;
         use crate::trace::cursor::MyTrait;
diff --git a/src/operators/reduce.rs b/src/operators/reduce.rs
index 7f322c00e..37d98d8ca 100644
--- a/src/operators/reduce.rs
+++ b/src/operators/reduce.rs
@@ -251,7 +251,7 @@ pub trait ReduceCore<G: Scope, K: ToOwned + ?Sized, V: ToOwned + ?Sized, R: Semi
             T2::ValOwned: Data,
             T2::Diff: Abelian,
             T2::Batch: Batch,
-            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, T2::ValOwned), T2::Time, T2::Diff)>,
+            T2::Builder: Builder<Input = ((K::Owned, T2::ValOwned), T2::Time, T2::Diff)>,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(<T2::Cursor as Cursor>::ValOwned, T2::Diff)>)+'static,
         {
             self.reduce_core::<_,T2>(name, move |key, input, output, change| {
@@ -273,7 +273,7 @@ pub trait ReduceCore<G: Scope, K: ToOwned + ?Sized, V: ToOwned + ?Sized, R: Semi
             T2: for<'a> Trace<Key<'a>=&'a K, Time=G::Timestamp>+'static,
             T2::ValOwned: Data,
             T2::Batch: Batch,
-            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, T2::ValOwned), T2::Time, T2::Diff)>,
+            T2::Builder: Builder<Input = ((K::Owned, T2::ValOwned), T2::Time, T2::Diff)>,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(<T2::Cursor as Cursor>::ValOwned,T2::Diff)>, &mut Vec<(<T2::Cursor as Cursor>::ValOwned, T2::Diff)>)+'static,
             ;
 }
@@ -292,7 +292,7 @@ where
             T2::ValOwned: Data,
             T2: for<'a> Trace<Key<'a>=&'a K, Time=G::Timestamp>+'static,
             T2::Batch: Batch,
-            T2::Builder: Builder<Output=T2::Batch, Item = ((K::Owned, T2::ValOwned), T2::Time, T2::Diff)>,
+            T2::Builder: Builder<Input = ((K::Owned, T2::ValOwned), T2::Time, T2::Diff)>,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(<T2::Cursor as Cursor>::ValOwned,T2::Diff)>, &mut Vec<(<T2::Cursor as Cursor>::ValOwned, T2::Diff)>)+'static,
     {
         self.arrange_by_key_named(&format!("Arrange: {}", name))
@@ -310,7 +310,7 @@ where
     T2: for<'a> Trace<Key<'a>=T1::Key<'a>, Time=T1::Time> + 'static,
     T2::ValOwned: Data,
     T2::Batch: Batch,
-    T2::Builder: Builder<Output=T2::Batch, Item = ((T1::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
+    T2::Builder: Builder<Input = ((T1::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
     L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(T2::ValOwned,T2::Diff)>, &mut Vec<(T2::ValOwned, T2::Diff)>)+'static,
 {
     let mut result_trace = None;
diff --git a/src/trace/implementations/merge_batcher.rs b/src/trace/implementations/merge_batcher.rs
index 673fedce6..2cf27ed1b 100644
--- a/src/trace/implementations/merge_batcher.rs
+++ b/src/trace/implementations/merge_batcher.rs
@@ -25,8 +25,8 @@ where
     T: Timestamp,
     D: Semigroup,
 {
-    type Input = Vec<Self::Item>;
-    type Item = ((K,V),T,D);
+    type Input = Vec<((K,V),T,D)>;
+    type Output = ((K,V),T,D);
     type Time = T;
 
     fn new(logger: Option<Logger<DifferentialEvent, WorkerIdentifier>>, operator_id: usize) -> Self {
@@ -38,7 +38,7 @@ where
     }
 
     #[inline(never)]
-    fn push_batch(&mut self, batch: RefOrMut<Vec<Self::Item>>) {
+    fn push_batch(&mut self, batch: RefOrMut<Self::Input>) {
         // `batch` is either a shared reference or an owned allocations.
         match batch {
             RefOrMut::Ref(reference) => {
@@ -59,7 +59,7 @@ where
     // which we call `lower`, by assumption that after sealing a batcher we receive no more
     // updates with times not greater or equal to `upper`.
     #[inline(never)]
-    fn seal<B: Builder<Item=Self::Item, Time=Self::Time>>(&mut self, upper: Antichain<T>) -> B::Output {
+    fn seal<B: Builder<Input=Self::Output, Time=Self::Time>>(&mut self, upper: Antichain<T>) -> B::Output {
 
         let mut merged = Vec::new();
         self.sorter.finish_into(&mut merged);
diff --git a/src/trace/implementations/merge_batcher_col.rs b/src/trace/implementations/merge_batcher_col.rs
index 7989eece8..21416170d 100644
--- a/src/trace/implementations/merge_batcher_col.rs
+++ b/src/trace/implementations/merge_batcher_col.rs
@@ -31,8 +31,8 @@ where
     T: Columnation + Timestamp + 'static,
     D: Columnation + Semigroup + 'static,
 {
-    type Input = Vec<Self::Item>;
-    type Item = ((K,V),T,D);
+    type Input = Vec<((K,V),T,D)>;
+    type Output = ((K,V),T,D);
     type Time = T;
 
     fn new(logger: Option<Logger<DifferentialEvent, WorkerIdentifier>>, operator_id: usize) -> Self {
@@ -44,7 +44,7 @@ where
     }
 
     #[inline]
-    fn push_batch(&mut self, batch: RefOrMut<Vec<Self::Item>>) {
+    fn push_batch(&mut self, batch: RefOrMut<Self::Input>) {
         // `batch` is either a shared reference or an owned allocations.
         match batch {
             RefOrMut::Ref(reference) => {
@@ -63,7 +63,7 @@ where
     // which we call `lower`, by assumption that after sealing a batcher we receive no more
     // updates with times not greater or equal to `upper`.
     #[inline]
-    fn seal<B: Builder<Item=Self::Item, Time=Self::Time>>(&mut self, upper: Antichain<T>) -> B::Output {
+    fn seal<B: Builder<Input=Self::Output, Time=Self::Time>>(&mut self, upper: Antichain<T>) -> B::Output {
 
         let mut merged = Default::default();
         self.sorter.finish_into(&mut merged);
diff --git a/src/trace/implementations/ord_neu.rs b/src/trace/implementations/ord_neu.rs
index bf4dd9544..ce0c6532a 100644
--- a/src/trace/implementations/ord_neu.rs
+++ b/src/trace/implementations/ord_neu.rs
@@ -540,7 +540,7 @@ mod val_batch {
 
     impl<L: Layout> Builder for OrdValBuilder<L> {
 
-        type Item = ((<L::Target as Update>::Key, <L::Target as Update>::Val), <L::Target as Update>::Time, <L::Target as Update>::Diff);
+        type Input = ((<L::Target as Update>::Key, <L::Target as Update>::Val), <L::Target as Update>::Time, <L::Target as Update>::Diff);
         type Time = <L::Target as Update>::Time;
         type Output = OrdValBatch<L>;
 
@@ -560,7 +560,7 @@ mod val_batch {
         }
 
         #[inline]
-        fn push(&mut self, ((key, val), time, diff): Self::Item) {
+        fn push(&mut self, ((key, val), time, diff): Self::Input) {
 
             // Perhaps this is a continuation of an already received key.
             if self.result.keys.last().map(|k| k.equals(&key)).unwrap_or(false) {
@@ -586,7 +586,7 @@ mod val_batch {
         }
 
         #[inline]
-        fn copy(&mut self, ((key, val), time, diff): &Self::Item) {
+        fn copy(&mut self, ((key, val), time, diff): &Self::Input) {
 
             // Perhaps this is a continuation of an already received key.
             if self.result.keys.last().map(|k| k.equals(key)).unwrap_or(false) {
@@ -1006,7 +1006,7 @@ mod key_batch {
 
     impl<L: Layout> Builder for OrdKeyBuilder<L> {
 
-        type Item = ((<L::Target as Update>::Key, ()), <L::Target as Update>::Time, <L::Target as Update>::Diff);
+        type Input = ((<L::Target as Update>::Key, ()), <L::Target as Update>::Time, <L::Target as Update>::Diff);
         type Time = <L::Target as Update>::Time;
         type Output = OrdKeyBatch<L>;
 
@@ -1024,7 +1024,7 @@ mod key_batch {
         }
 
         #[inline]
-        fn push(&mut self, ((key, ()), time, diff): Self::Item) {
+        fn push(&mut self, ((key, ()), time, diff): Self::Input) {
 
             // Perhaps this is a continuation of an already received key.
             if self.result.keys.last().map(|k| k.equals(&key)).unwrap_or(false) {
@@ -1040,7 +1040,7 @@ mod key_batch {
         }
 
         #[inline]
-        fn copy(&mut self, ((key, ()), time, diff): &Self::Item) {
+        fn copy(&mut self, ((key, ()), time, diff): &Self::Input) {
 
             // Perhaps this is a continuation of an already received key.
             if self.result.keys.last().map(|k| k.equals(key)).unwrap_or(false) {
diff --git a/src/trace/implementations/rhh.rs b/src/trace/implementations/rhh.rs
index 94ed3b95b..f29d39a8b 100644
--- a/src/trace/implementations/rhh.rs
+++ b/src/trace/implementations/rhh.rs
@@ -731,7 +731,7 @@ mod val_batch {
         <L::Target as Update>::Key: Default + HashOrdered,
         // RhhValBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
     {
-        type Item = ((<L::Target as Update>::Key, <L::Target as Update>::Val), <L::Target as Update>::Time, <L::Target as Update>::Diff);
+        type Input = ((<L::Target as Update>::Key, <L::Target as Update>::Val), <L::Target as Update>::Time, <L::Target as Update>::Diff);
         type Time = <L::Target as Update>::Time;
         type Output = RhhValBatch<L>;
 
@@ -763,7 +763,7 @@ mod val_batch {
         }
 
         #[inline]
-        fn push(&mut self, ((key, val), time, diff): Self::Item) {
+        fn push(&mut self, ((key, val), time, diff): Self::Input) {
 
             // Perhaps this is a continuation of an already received key.
             if self.result.keys.last().map(|k| k.equals(&key)).unwrap_or(false) {
@@ -790,7 +790,7 @@ mod val_batch {
         }
 
         #[inline]
-        fn copy(&mut self, ((key, val), time, diff): &Self::Item) {
+        fn copy(&mut self, ((key, val), time, diff): &Self::Input) {
 
             // Perhaps this is a continuation of an already received key.
             if self.result.keys.last().map(|k| k.equals(key)).unwrap_or(false) {
diff --git a/src/trace/implementations/spine_fueled.rs b/src/trace/implementations/spine_fueled.rs
index fd3af3b76..c616e2b7d 100644
--- a/src/trace/implementations/spine_fueled.rs
+++ b/src/trace/implementations/spine_fueled.rs
@@ -255,7 +255,7 @@ impl<B, BA, BU> Trace for Spine<B, BA, BU>
 where
     B: Batch+Clone+'static,
     BA: Batcher<Time = B::Time>,
-    BU: Builder<Item=BA::Item, Time=BA::Time, Output = B>,
+    BU: Builder<Input=BA::Output, Time=BA::Time, Output = B>,
 {
     /// A type used to assemble batches from disordered updates.
     type Batcher = BA;
diff --git a/src/trace/mod.rs b/src/trace/mod.rs
index d0d6791e3..145c2990f 100644
--- a/src/trace/mod.rs
+++ b/src/trace/mod.rs
@@ -213,7 +213,7 @@ where <Self as TraceReader>::Batch: Batch {
     /// A type used to assemble batches from disordered updates.
     type Batcher: Batcher<Time = Self::Time>;
     /// A type used to assemble batches from ordered update sequences.
-    type Builder: Builder<Item=<Self::Batcher as Batcher>::Item, Time=Self::Time, Output = Self::Batch>;
+    type Builder: Builder<Input=<Self::Batcher as Batcher>::Output, Time=Self::Time, Output = Self::Batch>;
 
     /// Allocates a new empty trace.
     fn new(
@@ -306,10 +306,10 @@ pub trait Batch : BatchReader where Self: ::std::marker::Sized {
 
 /// Functionality for collecting and batching updates.
 pub trait Batcher {
-    /// Type of update pushed into the batcher.
+    /// Type pushed into the batcher.
     type Input;
-    /// Type of update pushed into the builder.
-    type Item;
+    /// Type produced by the batcher.
+    type Output;
     /// Times at which batches are formed.
     type Time: Timestamp;
     /// Allocates a new empty batcher.
@@ -317,7 +317,7 @@ pub trait Batcher {
     /// Adds an unordered batch of elements to the batcher.
     fn push_batch(&mut self, batch: RefOrMut<Self::Input>);
     /// Returns all updates not greater or equal to an element of `upper`.
-    fn seal<B: Builder<Item=Self::Item, Time=Self::Time>>(&mut self, upper: Antichain<Self::Time>) -> B::Output;
+    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.
     fn frontier(&mut self) -> timely::progress::frontier::AntichainRef<Self::Time>;
 }
@@ -325,7 +325,7 @@ pub trait Batcher {
 /// Functionality for building batches from ordered update sequences.
 pub trait Builder: Sized {
     /// Input item type.
-    type Item;
+    type Input;
     /// Timestamp type.
     type Time: Timestamp;
     /// Output batch type.
@@ -344,15 +344,11 @@ pub trait Builder: Sized {
     ///
     /// The default implementation uses `self.copy` with references to the owned arguments.
     /// One should override it if the builder can take advantage of owned arguments.
-    fn push(&mut self, element: Self::Item) {
+    fn push(&mut self, element: Self::Input) {
         self.copy(&element);
     }
     /// Adds an element to the batch.
-    fn copy(&mut self, element: &Self::Item);
-    /// Adds an ordered sequence of elements to the batch.
-    fn extend<I: Iterator<Item=Self::Item>>(&mut self, iter: I) {
-        for item in iter { self.push(item); }
-    }
+    fn copy(&mut self, element: &Self::Input);
     /// Completes building and returns the batch.
     fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Self::Output;
 }
@@ -460,12 +456,12 @@ pub mod rc_blanket_impls {
 
     /// Functionality for building batches from ordered update sequences.
     impl<B: Builder> Builder for RcBuilder<B> {
-        type Item = B::Item;
+        type Input = B::Input;
         type Time = B::Time;
         type Output = Rc<B::Output>;
         fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self { RcBuilder { builder: B::with_capacity(keys, vals, upds) } }
-        fn push(&mut self, element: Self::Item) { self.builder.push(element) }
-        fn copy(&mut self, element: &Self::Item) { self.builder.copy(element) }
+        fn push(&mut self, element: Self::Input) { self.builder.push(element) }
+        fn copy(&mut self, element: &Self::Input) { self.builder.copy(element) }
         fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Rc<B::Output> { Rc::new(self.builder.done(lower, upper, since)) }
     }
 
@@ -569,12 +565,12 @@ pub mod abomonated_blanket_impls {
     where
         B::Output: Abomonation,
     {
-        type Item = B::Item;
+        type Input = B::Input;
         type Time = B::Time;
         type Output = Abomonated<B::Output, Vec<u8>>;
         fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self { AbomonatedBuilder { builder: B::with_capacity(keys, vals, upds) } }
-        fn push(&mut self, element: Self::Item) { self.builder.push(element) }
-        fn copy(&mut self, element: &Self::Item) { self.builder.copy(element) }
+        fn push(&mut self, element: Self::Input) { self.builder.push(element) }
+        fn copy(&mut self, element: &Self::Input) { self.builder.copy(element) }
         fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Self::Output {
             let batch = self.builder.done(lower, upper, since);
             let mut bytes = Vec::with_capacity(measure(&batch));