From eafc8f12d623af01cec8bbc841c5ffc914b64623 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 12 Apr 2024 14:15:59 -0400
Subject: [PATCH 1/3] Remove Item constraint from arrange_core

---
 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 653ed1e87..47bc2fc42 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -547,8 +547,8 @@ where
         R: Clone,
         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)>, Time = G::Timestamp>,
+        Tr::Builder: Builder<Time = G::Timestamp, Output = Tr::Batch>,
     ;
 }
 
@@ -565,8 +565,8 @@ 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)>, Time = G::Timestamp>,
+        Tr::Builder: Builder<Time = G::Timestamp, Output = Tr::Batch>,
     {
         arrange_core(&self.inner, pact, name)
     }
@@ -746,8 +746,8 @@ 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)>, Time = G::Timestamp>,
+        Tr::Builder: Builder<Time = G::Timestamp, Output = Tr::Batch>,
     {
         self.map(|k| (k, ()))
             .arrange_core(pact, name)

From 206f2977253049de48b0549e732703215a531f2b Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 12 Apr 2024 14:28:27 -0400
Subject: [PATCH 2/3] Remove redundant trait constraints

---
 src/operators/arrange/arrangement.rs | 17 +++++------------
 1 file changed, 5 insertions(+), 12 deletions(-)

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index 47bc2fc42..011c0769a 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -511,8 +511,7 @@ where
         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=Vec<((K,V),G::Timestamp,R)>>,
     {
         self.arrange_named("Arrange")
     }
@@ -527,8 +526,7 @@ where
         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=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)
@@ -547,8 +545,7 @@ where
         R: Clone,
         Tr: Trace<Time=G::Timestamp>+'static,
         Tr::Batch: Batch,
-        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>, Time = G::Timestamp>,
-        Tr::Builder: Builder<Time = G::Timestamp, Output = Tr::Batch>,
+        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>>,
     ;
 }
 
@@ -565,8 +562,7 @@ 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)>, Time = G::Timestamp>,
-        Tr::Builder: Builder<Time = G::Timestamp, Output = Tr::Batch>,
+        Tr::Batcher: Batcher<Input=Vec<((K,V),G::Timestamp,R)>>,
     {
         arrange_core(&self.inner, pact, name)
     }
@@ -584,9 +580,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
@@ -746,8 +740,7 @@ 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)>, Time = G::Timestamp>,
-        Tr::Builder: Builder<Time = G::Timestamp, Output = Tr::Batch>,
+        Tr::Batcher: Batcher<Input=Vec<((K,()),G::Timestamp,R)>>,
     {
         self.map(|k| (k, ()))
             .arrange_core(pact, name)

From 21272224139c7a393dd9e230e11a60c554be50e1 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 12 Apr 2024 14:54:46 -0400
Subject: [PATCH 3/3] Replace K,V,R Arrange type parameters by C

---
 src/operators/arrange/arrangement.rs | 78 +++++++++++++---------------
 1 file changed, 36 insertions(+), 42 deletions(-)

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index 011c0769a..7995b4397 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -489,74 +489,58 @@ 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)>>,
+        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)>>,
-    {
-        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)>>,
+        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)>>,
@@ -568,7 +552,7 @@ where
     }
 }
 
-/// 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
@@ -731,10 +715,20 @@ 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)>>,