From 95792c53a9d4f884d9c77a30a142b5c0fc90aa2c Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Tue, 28 Nov 2023 20:14:14 -0500
Subject: [PATCH] Remove Container: Clone + 'static

Remove the requirement that all Container implementations are Clone and
'static. This makes implementing types simpler that depend on Container,
and requires us to explicitly mark various places as `Data` such that they
comply with Timely's type requirements.

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>
---
 container/src/columnation.rs                  |  4 +-
 container/src/lib.rs                          |  7 ++-
 timely/examples/rc.rs                         |  4 +-
 timely/src/dataflow/channels/pact.rs          | 14 +++---
 .../src/dataflow/channels/pushers/exchange.rs |  2 +-
 timely/src/dataflow/channels/pushers/tee.rs   |  2 +-
 timely/src/dataflow/operators/branch.rs       |  2 +-
 .../src/dataflow/operators/capture/capture.rs |  8 ++--
 .../src/dataflow/operators/capture/replay.rs  |  4 +-
 timely/src/dataflow/operators/concat.rs       |  8 ++--
 timely/src/dataflow/operators/enterleave.rs   |  6 +--
 timely/src/dataflow/operators/feedback.rs     | 14 +++---
 .../dataflow/operators/generic/builder_rc.rs  |  6 +--
 .../dataflow/operators/generic/operator.rs    | 44 +++++++++----------
 timely/src/dataflow/operators/input.rs        | 14 +++---
 timely/src/dataflow/operators/inspect.rs      |  4 +-
 timely/src/dataflow/operators/probe.rs        |  4 +-
 timely/src/dataflow/operators/rc.rs           |  4 +-
 timely/src/dataflow/operators/reclock.rs      |  8 ++--
 timely/src/dataflow/operators/to_stream.rs    |  2 +-
 .../src/dataflow/operators/unordered_input.rs |  8 ++--
 timely/src/synchronization/sequence.rs        |  4 +-
 22 files changed, 86 insertions(+), 87 deletions(-)

diff --git a/container/src/columnation.rs b/container/src/columnation.rs
index 54a00ca4e3..e131e7548f 100644
--- a/container/src/columnation.rs
+++ b/container/src/columnation.rs
@@ -295,7 +295,7 @@ mod container {
 
     use crate::columnation::{Columnation, TimelyStack};
 
-    impl<T: Columnation + 'static> Container for TimelyStack<T> {
+    impl<T: Columnation> Container for TimelyStack<T> {
         type Item = T;
 
         fn len(&self) -> usize {
@@ -315,7 +315,7 @@ mod container {
         }
     }
 
-    impl<T: Columnation + 'static> PushPartitioned for TimelyStack<T> {
+    impl<T: Columnation> PushPartitioned for TimelyStack<T> {
         fn push_partitioned<I, F>(&mut self, buffers: &mut [Self], mut index: I, mut flush: F)
         where
             I: FnMut(&Self::Item) -> usize,
diff --git a/container/src/lib.rs b/container/src/lib.rs
index eb9b3ce77d..69358524c7 100644
--- a/container/src/lib.rs
+++ b/container/src/lib.rs
@@ -15,8 +15,7 @@ pub mod columnation;
 /// We require the container to be cloneable to enable efficient copies when providing references
 /// of containers to operators. Care must be taken that the type's `clone_from` implementation
 /// is efficient (which is not necessarily the case when deriving `Clone`.)
-/// TODO: Don't require `Container: Clone`
-pub trait Container: Default + Clone + 'static {
+pub trait Container: Default {
     /// The type of elements this container holds.
     type Item;
 
@@ -40,7 +39,7 @@ pub trait Container: Default + Clone + 'static {
     fn clear(&mut self);
 }
 
-impl<T: Clone + 'static> Container for Vec<T> {
+impl<T> Container for Vec<T> {
     type Item = T;
 
     fn len(&self) -> usize {
@@ -132,7 +131,7 @@ pub trait PushPartitioned: Container {
         F: FnMut(usize, &mut Self);
 }
 
-impl<T: Clone + 'static> PushPartitioned for Vec<T> {
+impl<T> PushPartitioned for Vec<T> {
     fn push_partitioned<I, F>(&mut self, buffers: &mut [Self], mut index: I, mut flush: F)
     where
         I: FnMut(&Self::Item) -> usize,
diff --git a/timely/examples/rc.rs b/timely/examples/rc.rs
index 18f6fe972e..f79ef74729 100644
--- a/timely/examples/rc.rs
+++ b/timely/examples/rc.rs
@@ -8,7 +8,7 @@ use abomonation::Abomonation;
 
 #[derive(Debug, Clone)]
 pub struct Test {
-    field: Rc<usize>,
+    _field: Rc<usize>,
 }
 
 impl Abomonation for Test {
@@ -32,7 +32,7 @@ fn main() {
 
         // introduce data and watch!
         for round in 0..10 {
-            input.send(Test { field: Rc::new(round) } );
+            input.send(Test { _field: Rc::new(round) } );
             input.advance_to(round + 1);
             worker.step_while(|| probe.less_than(input.time()));
         }
diff --git a/timely/src/dataflow/channels/pact.rs b/timely/src/dataflow/channels/pact.rs
index 976023a19f..969bac91fa 100644
--- a/timely/src/dataflow/channels/pact.rs
+++ b/timely/src/dataflow/channels/pact.rs
@@ -10,9 +10,9 @@
 use std::{fmt::{self, Debug}, marker::PhantomData};
 use timely_container::PushPartitioned;
 
-use crate::communication::{Push, Pull, Data};
+use crate::communication::{Push, Pull};
 use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
-use crate::Container;
+use crate::{Container, ExchangeData};
 
 use crate::worker::AsWorker;
 use crate::dataflow::channels::pushers::Exchange as ExchangePusher;
@@ -33,14 +33,14 @@ pub trait ParallelizationContractCore<T, D> {
 
 /// A `ParallelizationContractCore` specialized for `Vec` containers
 /// TODO: Use trait aliases once stable.
-pub trait ParallelizationContract<T, D: Clone>: ParallelizationContractCore<T, Vec<D>> { }
-impl<T, D: Clone, P: ParallelizationContractCore<T, Vec<D>>> ParallelizationContract<T, D> for P { }
+pub trait ParallelizationContract<T, D>: ParallelizationContractCore<T, Vec<D>> { }
+impl<T, D, P: ParallelizationContractCore<T, Vec<D>>> ParallelizationContract<T, D> for P { }
 
 /// A direct connection
 #[derive(Debug)]
 pub struct Pipeline;
 
-impl<T: 'static, D: Container> ParallelizationContractCore<T, D> for Pipeline {
+impl<T: 'static, D: Container + 'static> ParallelizationContractCore<T, D> for Pipeline {
     type Pusher = LogPusher<T, D, ThreadPusher<BundleCore<T, D>>>;
     type Puller = LogPuller<T, D, ThreadPuller<BundleCore<T, D>>>;
     fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: &[usize], logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
@@ -69,9 +69,9 @@ impl<C, D, F: FnMut(&D)->u64+'static> ExchangeCore<C, D, F> {
 }
 
 // Exchange uses a `Box<Pushable>` because it cannot know what type of pushable will return from the allocator.
-impl<T: Timestamp, C, D: Data+Clone, F: FnMut(&D)->u64+'static> ParallelizationContractCore<T, C> for ExchangeCore<C, D, F>
+impl<T: Timestamp, C, D: ExchangeData, F: FnMut(&D)->u64+'static> ParallelizationContractCore<T, C> for ExchangeCore<C, D, F>
 where
-    C: Data + Container + PushPartitioned<Item=D>,
+    C: ExchangeData + Container + PushPartitioned<Item=D>,
 {
     type Pusher = ExchangePusher<T, C, D, LogPusher<T, C, Box<dyn Push<BundleCore<T, C>>>>, F>;
     type Puller = LogPuller<T, C, Box<dyn Pull<BundleCore<T, C>>>>;
diff --git a/timely/src/dataflow/channels/pushers/exchange.rs b/timely/src/dataflow/channels/pushers/exchange.rs
index 9ea271d310..f8f0b2f673 100644
--- a/timely/src/dataflow/channels/pushers/exchange.rs
+++ b/timely/src/dataflow/channels/pushers/exchange.rs
@@ -40,7 +40,7 @@ impl<T: Clone, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&D) ->
     }
 }
 
-impl<T: Eq+Data, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&D) -> u64> Push<BundleCore<T, C>> for Exchange<T, C, D, P, H>
+impl<T: Eq+Data, C: Container + Data, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&D) -> u64> Push<BundleCore<T, C>> for Exchange<T, C, D, P, H>
 where
     C: PushPartitioned<Item=D>
 {
diff --git a/timely/src/dataflow/channels/pushers/tee.rs b/timely/src/dataflow/channels/pushers/tee.rs
index 1ec05e4e75..b15aa9a6f8 100644
--- a/timely/src/dataflow/channels/pushers/tee.rs
+++ b/timely/src/dataflow/channels/pushers/tee.rs
@@ -20,7 +20,7 @@ pub struct TeeCore<T, D> {
 /// [TeeCore] specialized to `Vec`-based container.
 pub type Tee<T, D> = TeeCore<T, Vec<D>>;
 
-impl<T: Data, D: Container> Push<BundleCore<T, D>> for TeeCore<T, D> {
+impl<T: Data, D: Container + Data> Push<BundleCore<T, D>> for TeeCore<T, D> {
     #[inline]
     fn push(&mut self, message: &mut Option<BundleCore<T, D>>) {
         let mut pushers = self.shared.borrow_mut();
diff --git a/timely/src/dataflow/operators/branch.rs b/timely/src/dataflow/operators/branch.rs
index 70e087abde..b3dacac8bb 100644
--- a/timely/src/dataflow/operators/branch.rs
+++ b/timely/src/dataflow/operators/branch.rs
@@ -94,7 +94,7 @@ pub trait BranchWhen<T>: Sized {
     fn branch_when(&self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self);
 }
 
-impl<S: Scope, C: Container> BranchWhen<S::Timestamp> for StreamCore<S, C> {
+impl<S: Scope, C: Container + Data> BranchWhen<S::Timestamp> for StreamCore<S, C> {
     fn branch_when(&self, condition: impl Fn(&S::Timestamp) -> bool + 'static) -> (Self, Self) {
         let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());
 
diff --git a/timely/src/dataflow/operators/capture/capture.rs b/timely/src/dataflow/operators/capture/capture.rs
index c62b95417b..1720ef8214 100644
--- a/timely/src/dataflow/operators/capture/capture.rs
+++ b/timely/src/dataflow/operators/capture/capture.rs
@@ -10,14 +10,14 @@ use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::channels::pullers::Counter as PullCounter;
 use crate::dataflow::operators::generic::builder_raw::OperatorBuilder;
 
-use crate::Container;
+use crate::{Container, Data};
 use crate::progress::ChangeBatch;
 use crate::progress::Timestamp;
 
 use super::{EventCore, EventPusherCore};
 
 /// Capture a stream of timestamped data for later replay.
-pub trait Capture<T: Timestamp, D: Container> {
+pub trait Capture<T: Timestamp, D: Container + Data> {
     /// Captures a stream of timestamped data for later replay.
     ///
     /// # Examples
@@ -113,7 +113,7 @@ pub trait Capture<T: Timestamp, D: Container> {
     }
 }
 
-impl<S: Scope, D: Container> Capture<S::Timestamp, D> for StreamCore<S, D> {
+impl<S: Scope, D: Container + Data> Capture<S::Timestamp, D> for StreamCore<S, D> {
     fn capture_into<P: EventPusherCore<S::Timestamp, D>+'static>(&self, mut event_pusher: P) {
 
         let mut builder = OperatorBuilder::new("Capture".to_owned(), self.scope());
@@ -142,7 +142,7 @@ impl<S: Scope, D: Container> Capture<S::Timestamp, D> for StreamCore<S, D> {
                         RefOrMut::Ref(reference) => (&reference.time, RefOrMut::Ref(&reference.data)),
                         RefOrMut::Mut(reference) => (&reference.time, RefOrMut::Mut(&mut reference.data)),
                     };
-                    let vector = data.replace(Default::default());
+                    let vector = data.take();
                     event_pusher.push(EventCore::Messages(time.clone(), vector));
                 }
                 input.consumed().borrow_mut().drain_into(&mut progress.consumeds[0]);
diff --git a/timely/src/dataflow/operators/capture/replay.rs b/timely/src/dataflow/operators/capture/replay.rs
index 2cc3254196..44a5462ecd 100644
--- a/timely/src/dataflow/operators/capture/replay.rs
+++ b/timely/src/dataflow/operators/capture/replay.rs
@@ -46,7 +46,7 @@ use crate::progress::Timestamp;
 
 use super::EventCore;
 use super::event::EventIteratorCore;
-use crate::Container;
+use crate::{Container, Data};
 
 /// Replay a capture stream into a scope with the same timestamp.
 pub trait Replay<T: Timestamp, C> : Sized {
@@ -62,7 +62,7 @@ pub trait Replay<T: Timestamp, C> : Sized {
     fn replay_core<S: Scope<Timestamp=T>>(self, scope: &mut S, period: Option<std::time::Duration>) -> StreamCore<S, C>;
 }
 
-impl<T: Timestamp, C: Container, I> Replay<T, C> for I
+impl<T: Timestamp, C: Container + Data, I> Replay<T, C> for I
 where I : IntoIterator,
       <I as IntoIterator>::Item: EventIteratorCore<T, C>+'static {
     fn replay_core<S: Scope<Timestamp=T>>(self, scope: &mut S, period: Option<std::time::Duration>) -> StreamCore<S, C>{
diff --git a/timely/src/dataflow/operators/concat.rs b/timely/src/dataflow/operators/concat.rs
index 449609f5bd..e9f1179eb7 100644
--- a/timely/src/dataflow/operators/concat.rs
+++ b/timely/src/dataflow/operators/concat.rs
@@ -1,7 +1,7 @@
 //! Merges the contents of multiple streams.
 
 
-use crate::Container;
+use crate::{Container, Data};
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{StreamCore, Scope};
 
@@ -23,7 +23,7 @@ pub trait Concat<G: Scope, D: Container> {
     fn concat(&self, _: &StreamCore<G, D>) -> StreamCore<G, D>;
 }
 
-impl<G: Scope, D: Container> Concat<G, D> for StreamCore<G, D> {
+impl<G: Scope, D: Container + Data> Concat<G, D> for StreamCore<G, D> {
     fn concat(&self, other: &StreamCore<G, D>) -> StreamCore<G, D> {
         self.scope().concatenate([self.clone(), other.clone()])
     }
@@ -52,7 +52,7 @@ pub trait Concatenate<G: Scope, D: Container> {
         I: IntoIterator<Item=StreamCore<G, D>>;
 }
 
-impl<G: Scope, D: Container> Concatenate<G, D> for StreamCore<G, D> {
+impl<G: Scope, D: Container + Data> Concatenate<G, D> for StreamCore<G, D> {
     fn concatenate<I>(&self, sources: I) -> StreamCore<G, D>
     where
         I: IntoIterator<Item=StreamCore<G, D>>
@@ -62,7 +62,7 @@ impl<G: Scope, D: Container> Concatenate<G, D> for StreamCore<G, D> {
     }
 }
 
-impl<G: Scope, D: Container> Concatenate<G, D> for G {
+impl<G: Scope, D: Container + Data> Concatenate<G, D> for G {
     fn concatenate<I>(&self, sources: I) -> StreamCore<G, D>
     where
         I: IntoIterator<Item=StreamCore<G, D>>
diff --git a/timely/src/dataflow/operators/enterleave.rs b/timely/src/dataflow/operators/enterleave.rs
index 24c8fa12eb..42e521ab7a 100644
--- a/timely/src/dataflow/operators/enterleave.rs
+++ b/timely/src/dataflow/operators/enterleave.rs
@@ -133,7 +133,7 @@ pub trait Leave<G: Scope, D: Container> {
     fn leave(&self) -> StreamCore<G, D>;
 }
 
-impl<'a, G: Scope, D: Clone+Container, T: Timestamp+Refines<G::Timestamp>> Leave<G, D> for StreamCore<Child<'a, G, T>, D> {
+impl<'a, G: Scope, D: Container + Data, T: Timestamp+Refines<G::Timestamp>> Leave<G, D> for StreamCore<Child<'a, G, T>, D> {
     fn leave(&self) -> StreamCore<G, D> {
 
         let scope = self.scope();
@@ -160,14 +160,14 @@ impl<'a, G: Scope, D: Clone+Container, T: Timestamp+Refines<G::Timestamp>> Leave
 }
 
 
-struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> {
+struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container + Data> {
     targets: CounterCore<TInner, TData, TeeCore<TInner, TData>>,
     phantom: ::std::marker::PhantomData<TOuter>,
     activator: crate::scheduling::Activator,
     active: bool,
 }
 
-impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> Push<BundleCore<TOuter, TData>> for IngressNub<TOuter, TInner, TData> {
+impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container + Data> Push<BundleCore<TOuter, TData>> for IngressNub<TOuter, TInner, TData> {
     fn push(&mut self, element: &mut Option<BundleCore<TOuter, TData>>) {
         if let Some(message) = element {
             let outer_message = message.as_mut();
diff --git a/timely/src/dataflow/operators/feedback.rs b/timely/src/dataflow/operators/feedback.rs
index a7eb90c658..76bf98a557 100644
--- a/timely/src/dataflow/operators/feedback.rs
+++ b/timely/src/dataflow/operators/feedback.rs
@@ -59,7 +59,7 @@ pub trait Feedback<G: Scope> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn feedback_core<D: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, D>, StreamCore<G, D>);
+    fn feedback_core<D: Container + Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, D>, StreamCore<G, D>);
 }
 
 /// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
@@ -87,7 +87,7 @@ pub trait LoopVariable<'a, G: Scope, T: Timestamp> {
     ///     });
     /// });
     /// ```
-    fn loop_variable<D: Container>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, D>, StreamCore<Iterative<'a, G, T>, D>);
+    fn loop_variable<D: Container + Data>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, D>, StreamCore<Iterative<'a, G, T>, D>);
 }
 
 impl<G: Scope> Feedback<G> for G {
@@ -95,7 +95,7 @@ impl<G: Scope> Feedback<G> for G {
         self.feedback_core(summary)
     }
 
-    fn feedback_core<D: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, D>, StreamCore<G, D>) {
+    fn feedback_core<D: Container + Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (HandleCore<G, D>, StreamCore<G, D>) {
 
         let mut builder = OperatorBuilder::new("Feedback".to_owned(), self.clone());
         let (output, stream) = builder.new_output();
@@ -105,13 +105,13 @@ impl<G: Scope> Feedback<G> for G {
 }
 
 impl<'a, G: Scope, T: Timestamp> LoopVariable<'a, G, T> for Iterative<'a, G, T> {
-    fn loop_variable<D: Container>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, D>, StreamCore<Iterative<'a, G, T>, D>) {
+    fn loop_variable<D: Container + Data>(&mut self, summary: T::Summary) -> (HandleCore<Iterative<'a, G, T>, D>, StreamCore<Iterative<'a, G, T>, D>) {
         self.feedback_core(Product::new(Default::default(), summary))
     }
 }
 
 /// Connect a `Stream` to the input of a loop variable.
-pub trait ConnectLoop<G: Scope, D: Container> {
+pub trait ConnectLoop<G: Scope, D: Container + Data> {
     /// Connect a `Stream` to be the input of a loop variable.
     ///
     /// # Examples
@@ -132,7 +132,7 @@ pub trait ConnectLoop<G: Scope, D: Container> {
     fn connect_loop(&self, _: HandleCore<G, D>);
 }
 
-impl<G: Scope, D: Container> ConnectLoop<G, D> for StreamCore<G, D> {
+impl<G: Scope, D: Container + Data> ConnectLoop<G, D> for StreamCore<G, D> {
     fn connect_loop(&self, helper: HandleCore<G, D>) {
 
         let mut builder = helper.builder;
@@ -159,7 +159,7 @@ impl<G: Scope, D: Container> ConnectLoop<G, D> for StreamCore<G, D> {
 
 /// A handle used to bind the source of a loop variable.
 #[derive(Debug)]
-pub struct HandleCore<G: Scope, D: Container> {
+pub struct HandleCore<G: Scope, D: Container + Data> {
     builder: OperatorBuilder<G>,
     summary: <G::Timestamp as Timestamp>::Summary,
     output: OutputWrapper<G::Timestamp, D, TeeCore<G::Timestamp, D>>,
diff --git a/timely/src/dataflow/operators/generic/builder_rc.rs b/timely/src/dataflow/operators/generic/builder_rc.rs
index ca80e31828..a2a06ab571 100644
--- a/timely/src/dataflow/operators/generic/builder_rc.rs
+++ b/timely/src/dataflow/operators/generic/builder_rc.rs
@@ -8,7 +8,7 @@ use crate::progress::{ChangeBatch, Timestamp};
 use crate::progress::operate::SharedProgress;
 use crate::progress::frontier::{Antichain, MutableAntichain};
 
-use crate::Container;
+use crate::{Container, Data};
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::channels::pushers::TeeCore;
 use crate::dataflow::channels::pushers::CounterCore as PushCounter;
@@ -92,7 +92,7 @@ impl<G: Scope> OperatorBuilder<G> {
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
-    pub fn new_output<D: Container>(&mut self) -> (OutputWrapper<G::Timestamp, D, TeeCore<G::Timestamp, D>>, StreamCore<G, D>) {
+    pub fn new_output<D: Container + Data>(&mut self) -> (OutputWrapper<G::Timestamp, D, TeeCore<G::Timestamp, D>>, StreamCore<G, D>) {
         let connection = vec![Antichain::from_elem(Default::default()); self.builder.shape().inputs()];
         self.new_output_connection(connection)
     }
@@ -105,7 +105,7 @@ impl<G: Scope> OperatorBuilder<G> {
     ///
     /// Commonly the connections are either the unit summary, indicating the same timestamp might be produced as output, or an empty
     /// antichain indicating that there is no connection from the input to the output.
-    pub fn new_output_connection<D: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (OutputWrapper<G::Timestamp, D, TeeCore<G::Timestamp, D>>, StreamCore<G, D>) {
+    pub fn new_output_connection<D: Container + Data>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (OutputWrapper<G::Timestamp, D, TeeCore<G::Timestamp, D>>, StreamCore<G, D>) {
 
         let (tee, stream) = self.builder.new_output_connection(connection.clone());
 
diff --git a/timely/src/dataflow/operators/generic/operator.rs b/timely/src/dataflow/operators/generic/operator.rs
index afb7a25d11..05c8e7db38 100644
--- a/timely/src/dataflow/operators/generic/operator.rs
+++ b/timely/src/dataflow/operators/generic/operator.rs
@@ -12,7 +12,7 @@ use crate::dataflow::{Scope, StreamCore};
 use super::builder_rc::OperatorBuilder;
 use crate::dataflow::operators::generic::OperatorInfo;
 use crate::dataflow::operators::generic::notificator::{Notificator, FrontierNotificator};
-use crate::Container;
+use crate::{Container, Data};
 
 /// Methods to construct generic streaming and blocking operators.
 pub trait Operator<G: Scope, D1: Container> {
@@ -57,7 +57,7 @@ pub trait Operator<G: Scope, D1: Container> {
     /// ```
     fn unary_frontier<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, D2>
     where
-        D2: Container,
+        D2: Container + Data,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P::Puller>,
                  &mut OutputHandleCore<G::Timestamp, D2, TeeCore<G::Timestamp, D2>>)+'static,
@@ -92,7 +92,7 @@ pub trait Operator<G: Scope, D1: Container> {
     ///     });
     /// }
     /// ```
-    fn unary_notify<D2: Container,
+    fn unary_notify<D2: Container + Data,
             L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>,
                      &mut OutputHandleCore<G::Timestamp, D2, TeeCore<G::Timestamp, D2>>,
                      &mut Notificator<G::Timestamp>)+'static,
@@ -129,7 +129,7 @@ pub trait Operator<G: Scope, D1: Container> {
     /// ```
     fn unary<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, D2>
     where
-        D2: Container,
+        D2: Container + Data,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>,
                  &mut OutputHandleCore<G::Timestamp, D2, TeeCore<G::Timestamp, D2>>)+'static,
@@ -187,8 +187,8 @@ pub trait Operator<G: Scope, D1: Container> {
     /// ```
     fn binary_frontier<D2, D3, B, L, P1, P2>(&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, D3>
     where
-        D2: Container,
-        D3: Container,
+        D2: Container + Data,
+        D3: Container + Data,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P1::Puller>,
                  &mut FrontieredInputHandleCore<G::Timestamp, D2, P2::Puller>,
@@ -241,8 +241,8 @@ pub trait Operator<G: Scope, D1: Container> {
     ///    }
     /// }).unwrap();
     /// ```
-    fn binary_notify<D2: Container,
-              D3: Container,
+    fn binary_notify<D2: Container + Data,
+              D3: Container + Data,
               L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>,
                        &mut InputHandleCore<G::Timestamp, D2, P2::Puller>,
                        &mut OutputHandleCore<G::Timestamp, D3, TeeCore<G::Timestamp, D3>>,
@@ -287,8 +287,8 @@ pub trait Operator<G: Scope, D1: Container> {
     /// ```
     fn binary<D2, D3, B, L, P1, P2>(&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, D3>
     where
-        D2: Container,
-        D3: Container,
+        D2: Container + Data,
+        D3: Container + Data,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>,
                  &mut InputHandleCore<G::Timestamp, D2, P2::Puller>,
@@ -325,11 +325,11 @@ pub trait Operator<G: Scope, D1: Container> {
         P: ParallelizationContractCore<G::Timestamp, D1>;
 }
 
-impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
+impl<G: Scope, D1: Container + Data> Operator<G, D1> for StreamCore<G, D1> {
 
     fn unary_frontier<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, D2>
     where
-        D2: Container,
+        D2: Container + Data,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P::Puller>,
                  &mut OutputHandleCore<G::Timestamp, D2, TeeCore<G::Timestamp, D2>>)+'static,
@@ -355,7 +355,7 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
         stream
     }
 
-    fn unary_notify<D2: Container,
+    fn unary_notify<D2: Container + Data,
             L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>,
                      &mut OutputHandleCore<G::Timestamp, D2, TeeCore<G::Timestamp, D2>>,
                      &mut Notificator<G::Timestamp>)+'static,
@@ -379,7 +379,7 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
 
     fn unary<D2, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, D2>
     where
-        D2: Container,
+        D2: Container + Data,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>,
                  &mut OutputHandleCore<G::Timestamp, D2, TeeCore<G::Timestamp, D2>>)+'static,
@@ -407,8 +407,8 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
 
     fn binary_frontier<D2, D3, B, L, P1, P2>(&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, D3>
     where
-        D2: Container,
-        D3: Container,
+        D2: Container + Data,
+        D3: Container + Data,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, D1, P1::Puller>,
                  &mut FrontieredInputHandleCore<G::Timestamp, D2, P2::Puller>,
@@ -438,8 +438,8 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
         stream
     }
 
-    fn binary_notify<D2: Container,
-              D3: Container,
+    fn binary_notify<D2: Container + Data,
+              D3: Container + Data,
               L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>,
                        &mut InputHandleCore<G::Timestamp, D2, P2::Puller>,
                        &mut OutputHandleCore<G::Timestamp, D3, TeeCore<G::Timestamp, D3>>,
@@ -467,8 +467,8 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
 
     fn binary<D2, D3, B, L, P1, P2>(&self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, D3>
     where
-        D2: Container,
-        D3: Container,
+        D2: Container + Data,
+        D3: Container + Data,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>,
                  &mut InputHandleCore<G::Timestamp, D2, P2::Puller>,
@@ -557,7 +557,7 @@ impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1> {
 /// ```
 pub fn source<G: Scope, D, B, L>(scope: &G, name: &str, constructor: B) -> StreamCore<G, D>
 where
-    D: Container,
+    D: Container + Data,
     B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
     L: FnMut(&mut OutputHandleCore<G::Timestamp, D, TeeCore<G::Timestamp, D>>)+'static {
 
@@ -599,7 +599,7 @@ where
 ///
 /// });
 /// ```
-pub fn empty<G: Scope, D: Container>(scope: &G) -> StreamCore<G, D> {
+pub fn empty<G: Scope, D: Container + Data>(scope: &G) -> StreamCore<G, D> {
     source(scope, "Empty", |_capability, _info| |_output| {
         // drop capability, do nothing
     })
diff --git a/timely/src/dataflow/operators/input.rs b/timely/src/dataflow/operators/input.rs
index 72719c5f85..a9d928b03b 100644
--- a/timely/src/dataflow/operators/input.rs
+++ b/timely/src/dataflow/operators/input.rs
@@ -93,7 +93,7 @@ pub trait Input : Scope {
     ///     }
     /// });
     /// ```
-    fn new_input_core<D: Container>(&mut self) -> (HandleCore<<Self as ScopeParent>::Timestamp, D>, StreamCore<Self, D>);
+    fn new_input_core<D: Container + Data>(&mut self) -> (HandleCore<<Self as ScopeParent>::Timestamp, D>, StreamCore<Self, D>);
 
     /// Create a new stream from a supplied interactive handle.
     ///
@@ -157,7 +157,7 @@ pub trait Input : Scope {
     ///     }
     /// });
     /// ```
-    fn input_from_core<D: Container>(&mut self, handle: &mut HandleCore<<Self as ScopeParent>::Timestamp, D>) -> StreamCore<Self, D>;
+    fn input_from_core<D: Container + Data>(&mut self, handle: &mut HandleCore<<Self as ScopeParent>::Timestamp, D>) -> StreamCore<Self, D>;
 }
 
 use crate::order::TotalOrder;
@@ -170,13 +170,13 @@ impl<G: Scope> Input for G where <G as ScopeParent>::Timestamp: TotalOrder {
         self.input_from_core(handle)
     }
 
-    fn new_input_core<D: Container>(&mut self) -> (HandleCore<<G as ScopeParent>::Timestamp, D>, StreamCore<G, D>) {
+    fn new_input_core<D: Container + Data>(&mut self) -> (HandleCore<<G as ScopeParent>::Timestamp, D>, StreamCore<G, D>) {
         let mut handle = HandleCore::new();
         let stream = self.input_from_core(&mut handle);
         (handle, stream)
     }
 
-    fn input_from_core<D: Container>(&mut self, handle: &mut HandleCore<<G as ScopeParent>::Timestamp, D>) -> StreamCore<G, D> {
+    fn input_from_core<D: Container + Data + std::clone::Clone>(&mut self, handle: &mut HandleCore<<G as ScopeParent>::Timestamp, D>) -> StreamCore<G, D> {
         let (output, registrar) = TeeCore::<<G as ScopeParent>::Timestamp, D>::new();
         let counter = CounterCore::new(output);
         let produced = counter.produced().clone();
@@ -246,7 +246,7 @@ impl<T:Timestamp> Operate<T> for Operator<T> {
 
 /// A handle to an input `Stream`, used to introduce data to a timely dataflow computation.
 #[derive(Debug)]
-pub struct HandleCore<T: Timestamp, C: Container> {
+pub struct HandleCore<T: Timestamp, C: Container + Data> {
     activate: Vec<Activator>,
     progress: Vec<Rc<RefCell<ChangeBatch<T>>>>,
     pushers: Vec<CounterCore<T, C, TeeCore<T, C>>>,
@@ -258,7 +258,7 @@ pub struct HandleCore<T: Timestamp, C: Container> {
 /// A handle specialized to vector-based containers.
 pub type Handle<T, D> = HandleCore<T, Vec<D>>;
 
-impl<T: Timestamp, D: Container> HandleCore<T, D> {
+impl<T: Timestamp, D: Container + Data> HandleCore<T, D> {
     /// Allocates a new input handle, from which one can create timely streams.
     ///
     /// # Examples
@@ -506,7 +506,7 @@ impl<T: Timestamp, D: Data> Default for Handle<T, D> {
     }
 }
 
-impl<T:Timestamp, C: Container> Drop for HandleCore<T, C> {
+impl<T:Timestamp, C: Container + Data> Drop for HandleCore<T, C> {
     fn drop(&mut self) {
         self.close_epoch();
     }
diff --git a/timely/src/dataflow/operators/inspect.rs b/timely/src/dataflow/operators/inspect.rs
index 6e26856d18..fcbe72923b 100644
--- a/timely/src/dataflow/operators/inspect.rs
+++ b/timely/src/dataflow/operators/inspect.rs
@@ -99,7 +99,7 @@ impl<G: Scope, D: Data+Columnation> Inspect<G, TimelyStack<D>> for StreamCore<G,
     }
 }
 
-impl<G: Scope, C: Container> Inspect<G, Rc<C>> for StreamCore<G, Rc<C>>
+impl<G: Scope, C: Container + Data> Inspect<G, Rc<C>> for StreamCore<G, Rc<C>>
     where C: AsRef<[C::Item]>
 {
     fn inspect_core<F>(&self, mut func: F) -> Self where F: FnMut(Result<(&G::Timestamp, &[C::Item]), &[G::Timestamp]>) + 'static {
@@ -131,7 +131,7 @@ pub trait InspectCore<G: Scope, C: Container> {
     fn inspect_container<F>(&self, func: F) -> StreamCore<G, C> where F: FnMut(Result<(&G::Timestamp, &C), &[G::Timestamp]>)+'static;
 }
 
-impl<G: Scope, C: Container> InspectCore<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container + Data> InspectCore<G, C> for StreamCore<G, C> {
 
     fn inspect_container<F>(&self, mut func: F) -> StreamCore<G, C>
         where F: FnMut(Result<(&G::Timestamp, &C), &[G::Timestamp]>)+'static
diff --git a/timely/src/dataflow/operators/probe.rs b/timely/src/dataflow/operators/probe.rs
index 7c5a8567ec..2208680bd2 100644
--- a/timely/src/dataflow/operators/probe.rs
+++ b/timely/src/dataflow/operators/probe.rs
@@ -13,7 +13,7 @@ use crate::dataflow::operators::generic::builder_raw::OperatorBuilder;
 
 
 use crate::dataflow::{StreamCore, Scope};
-use crate::Container;
+use crate::{Container, Data};
 
 /// Monitors progress at a `Stream`.
 pub trait Probe<G: Scope, D: Container> {
@@ -79,7 +79,7 @@ pub trait Probe<G: Scope, D: Container> {
     fn probe_with(&self, handle: &mut Handle<G::Timestamp>) -> StreamCore<G, D>;
 }
 
-impl<G: Scope, D: Container> Probe<G, D> for StreamCore<G, D> {
+impl<G: Scope, D: Container + Data> Probe<G, D> for StreamCore<G, D> {
     fn probe(&self) -> Handle<G::Timestamp> {
 
         // the frontier is shared state; scope updates, handle reads.
diff --git a/timely/src/dataflow/operators/rc.rs b/timely/src/dataflow/operators/rc.rs
index eaae55093e..e2be883814 100644
--- a/timely/src/dataflow/operators/rc.rs
+++ b/timely/src/dataflow/operators/rc.rs
@@ -3,7 +3,7 @@
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::Operator;
 use crate::dataflow::{Scope, StreamCore};
-use crate::Container;
+use crate::{Container, Data};
 use std::rc::Rc;
 
 /// Convert a stream into a stream of shared containers
@@ -24,7 +24,7 @@ pub trait SharedStream<S: Scope, C: Container> {
     fn shared(&self) -> StreamCore<S, Rc<C>>;
 }
 
-impl<S: Scope, C: Container> SharedStream<S, C> for StreamCore<S, C> {
+impl<S: Scope, C: Container + Data> SharedStream<S, C> for StreamCore<S, C> {
     fn shared(&self) -> StreamCore<S, Rc<C>> {
         let mut container = Default::default();
         self.unary(Pipeline, "Shared", move |_, _| {
diff --git a/timely/src/dataflow/operators/reclock.rs b/timely/src/dataflow/operators/reclock.rs
index b656e0aaf8..20562c9988 100644
--- a/timely/src/dataflow/operators/reclock.rs
+++ b/timely/src/dataflow/operators/reclock.rs
@@ -1,6 +1,6 @@
 //! Extension methods for `Stream` based on record-by-record transformation.
 
-use crate::Container;
+use crate::{Container, Data};
 use crate::order::PartialOrder;
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::channels::pact::Pipeline;
@@ -45,11 +45,11 @@ pub trait Reclock<S: Scope> {
     /// assert_eq!(extracted[1], (5, vec![4,5]));
     /// assert_eq!(extracted[2], (8, vec![6,7,8]));
     /// ```
-    fn reclock<TC: Container<Item=()>>(&self, clock: &StreamCore<S, TC>) -> Self;
+    fn reclock<TC: Container<Item=()> + Data>(&self, clock: &StreamCore<S, TC>) -> Self;
 }
 
-impl<S: Scope, C: Container> Reclock<S> for StreamCore<S, C> {
-    fn reclock<TC: Container<Item=()>>(&self, clock: &StreamCore<S, TC>) -> StreamCore<S, C> {
+impl<S: Scope, C: Container + Data> Reclock<S> for StreamCore<S, C> {
+    fn reclock<TC: Container<Item=()> + Data>(&self, clock: &StreamCore<S, TC>) -> StreamCore<S, C> {
 
         let mut stash = vec![];
 
diff --git a/timely/src/dataflow/operators/to_stream.rs b/timely/src/dataflow/operators/to_stream.rs
index 597ec9163c..6d37e9fab5 100644
--- a/timely/src/dataflow/operators/to_stream.rs
+++ b/timely/src/dataflow/operators/to_stream.rs
@@ -83,7 +83,7 @@ pub trait ToStreamCore<T: Timestamp, C: Container> {
     fn to_stream_core<S: Scope<Timestamp=T>>(self, scope: &mut S) -> StreamCore<S, C>;
 }
 
-impl<T: Timestamp, I: IntoIterator+'static> ToStreamCore<T, I::Item> for I where I::Item: Container {
+impl<T: Timestamp, I: IntoIterator+'static> ToStreamCore<T, I::Item> for I where I::Item: Container + Data {
     fn to_stream_core<S: Scope<Timestamp=T>>(self, scope: &mut S) -> StreamCore<S, I::Item> {
 
         source(scope, "ToStreamCore", |capability, info| {
diff --git a/timely/src/dataflow/operators/unordered_input.rs b/timely/src/dataflow/operators/unordered_input.rs
index c7e6002341..805d54c085 100644
--- a/timely/src/dataflow/operators/unordered_input.rs
+++ b/timely/src/dataflow/operators/unordered_input.rs
@@ -142,12 +142,12 @@ pub trait UnorderedInputCore<G: Scope> {
     ///     assert_eq!(extract[i], (i, vec![i]));
     /// }
     /// ```
-    fn new_unordered_input_core<D: Container>(&mut self) -> ((UnorderedHandleCore<G::Timestamp, D>, ActivateCapability<G::Timestamp>), StreamCore<G, D>);
+    fn new_unordered_input_core<D: Container + Data>(&mut self) -> ((UnorderedHandleCore<G::Timestamp, D>, ActivateCapability<G::Timestamp>), StreamCore<G, D>);
 }
 
 
 impl<G: Scope> UnorderedInputCore<G> for G {
-    fn new_unordered_input_core<D: Container>(&mut self) -> ((UnorderedHandleCore<G::Timestamp, D>, ActivateCapability<G::Timestamp>), StreamCore<G, D>) {
+    fn new_unordered_input_core<D: Container + Data>(&mut self) -> ((UnorderedHandleCore<G::Timestamp, D>, ActivateCapability<G::Timestamp>), StreamCore<G, D>) {
 
         let (output, registrar) = TeeCore::<G::Timestamp, D>::new();
         let internal = Rc::new(RefCell::new(ChangeBatch::new()));
@@ -215,11 +215,11 @@ impl<T:Timestamp> Operate<T> for UnorderedOperator<T> {
 
 /// A handle to an input [StreamCore], used to introduce data to a timely dataflow computation.
 #[derive(Debug)]
-pub struct UnorderedHandleCore<T: Timestamp, D: Container> {
+pub struct UnorderedHandleCore<T: Timestamp, D: Container + Data> {
     buffer: PushBuffer<T, D, PushCounter<T, D, TeeCore<T, D>>>,
 }
 
-impl<T: Timestamp, D: Container> UnorderedHandleCore<T, D> {
+impl<T: Timestamp, D: Container + Data> UnorderedHandleCore<T, D> {
     fn new(pusher: PushCounter<T, D, TeeCore<T, D>>) -> UnorderedHandleCore<T, D> {
         UnorderedHandleCore {
             buffer: PushBuffer::new(pusher),
diff --git a/timely/src/synchronization/sequence.rs b/timely/src/synchronization/sequence.rs
index 7525d33892..320a3dcb87 100644
--- a/timely/src/synchronization/sequence.rs
+++ b/timely/src/synchronization/sequence.rs
@@ -157,7 +157,7 @@ impl<T: ExchangeData> Sequencer<T> {
                         }
 
                         let mut activator_borrow = activator_source.borrow_mut();
-                        let mut activator = activator_borrow.as_mut().unwrap();
+                        let activator = activator_borrow.as_mut().unwrap();
 
                         if let Some(t) = activator.catchup_until {
                             if capability.time().less_than(&t) {
@@ -190,7 +190,7 @@ impl<T: ExchangeData> Sequencer<T> {
 
                     if let Some(last) = recvd.last() {
                         let mut activator_borrow = activator_sink.borrow_mut();
-                        let mut activator = activator_borrow.as_mut().unwrap();
+                        let activator = activator_borrow.as_mut().unwrap();
 
                         activator.catchup_until = Some((last.0).0);
                         activator.activate();