From 16f55f8ac8fbd6ef9e051fbf789adbf92fa4d360 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Tue, 9 Apr 2024 12:00:52 -0400
Subject: [PATCH 1/4] Extract batcher input to assoc type, arrange_core
 freestanding

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>
---
 .github/workflows/deploy.yml                  |   2 +-
 .github/workflows/test.yml                    |   4 +-
 interactive/src/command.rs                    |   6 +-
 interactive/src/logging.rs                    |   6 +-
 src/operators/arrange/arrangement.rs          | 290 +++++++++---------
 src/operators/consolidate.rs                  |   2 +-
 src/trace/implementations/merge_batcher.rs    |   1 +
 .../implementations/merge_batcher_col.rs      |   1 +
 src/trace/mod.rs                              |   4 +-
 9 files changed, 165 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/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..617dfca2b 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>,
     ;
 }
@@ -561,164 +561,174 @@ where
     R: Semigroup,
 {
     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<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp>,
-        Tr::Builder: Builder<Item = ((K,V),G::Timestamp,R), Time = G::Timestamp, Output = Tr::Batch>,
+        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>,
     {
-        // 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| {
-
-                    // 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.
+        arrange_core(&self.inner, pact, name)
+    }
+}
+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 +740,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 20cf803e69bf59837fe6412e8a4dcfb682630f54 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Tue, 9 Apr 2024 12:07:40 -0400
Subject: [PATCH 2/4] Fix warnings

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>
---
 doop/src/main.rs                              | 8 ++++----
 experiments/src/bin/graphs-interactive-alt.rs | 4 +++-
 experiments/src/bin/graphs-interactive-neu.rs | 4 +++-
 3 files changed, 10 insertions(+), 6 deletions(-)

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
+}

From a749a63a9d7f3ccd407406cfc5ff82e20d316e64 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Wed, 10 Apr 2024 13:48:57 -0400
Subject: [PATCH 3/4] documentation

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>
---
 src/operators/arrange/arrangement.rs | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index 617dfca2b..f5514467c 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -571,6 +571,12 @@ where
         arrange_core(&self.inner, pact, name)
     }
 }
+
+/// 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,

From d44d7e2400b5f6e0997b354d1b24be8b5defa4fa Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Wed, 10 Apr 2024 13:50:24 -0400
Subject: [PATCH 4/4] formatting

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>
---
 src/operators/arrange/arrangement.rs | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index f5514467c..653ed1e87 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -561,12 +561,12 @@ where
     R: Semigroup,
 {
     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>,
+    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>,
     {
         arrange_core(&self.inner, pact, name)
     }