From 186b94a19f84faab0e8956fcf1b15cdd1ab26927 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Mon, 9 Jan 2023 20:52:07 -0500
Subject: [PATCH] Loosen trait bounds on Arrange for Collection

Allows to call arrange_core on a collection that cannot be exchanged.

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

diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
index 34988adb7..2d4318e43 100644
--- a/src/operators/arrange/arrangement.rs
+++ b/src/operators/arrange/arrangement.rs
@@ -495,10 +495,31 @@ impl<G, K, V, R> Arrange<G, K, V, R> for Collection<G, (K, V), R>
 where
     G: Scope,
     G::Timestamp: Lattice+Ord,
-    K: ExchangeData+Hashable,
-    V: ExchangeData,
-    R: Semigroup+ExchangeData,
+    K: Data,
+    V: Data,
+    R: Semigroup,
 {
+    fn arrange<Tr>(&self) -> Arranged<G, TraceAgent<Tr>>
+    where
+        K: ExchangeData + Hashable,
+        V: ExchangeData,
+        R: ExchangeData,
+        Tr: Trace + TraceReader<Key=K, Val=V, Time=G::Timestamp, R=R> + 'static, Tr::Batch: Batch
+    {
+        self.arrange_named("Arrange")
+    }
+
+    fn arrange_named<Tr>(&self, name: &str) -> Arranged<G, TraceAgent<Tr>>
+    where
+        K: ExchangeData + Hashable,
+        V: ExchangeData,
+        R: ExchangeData,
+        Tr: Trace + TraceReader<Key=K, Val=V, Time=G::Timestamp, R=R> + 'static, Tr::Batch: Batch
+    {
+        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, ((K,V),G::Timestamp,R)>,