diff --git a/container/src/lib.rs b/container/src/lib.rs
index 2ff6589fe..7f8447d9d 100644
--- a/container/src/lib.rs
+++ b/container/src/lib.rs
@@ -4,63 +4,55 @@
 
 use std::collections::VecDeque;
 
-/// A container transferring data through dataflow edges
+/// An type containing a number of records accounted for by progress tracking.
 ///
-/// A container stores a number of elements and thus is able to describe it length (`len()`) and
-/// whether it is empty (`is_empty()`). It supports removing all elements (`clear`).
-///
-/// A container must implement default. The default implementation is not required to allocate
-/// memory for variable-length components.
-///
-/// 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`.)
-pub trait Container: Default {
-    /// The type of elements when reading non-destructively from the container.
-    type ItemRef<'a> where Self: 'a;
-
-    /// The type of elements when draining the container.
-    type Item<'a> where Self: 'a;
-
-    /// Push `item` into self
-    #[inline]
-    fn push<T>(&mut self, item: T) where Self: PushInto<T> {
-        self.push_into(item)
-    }
-
-    /// The number of elements in this container
+/// The object stores a number of updates and thus is able to describe it count
+/// (`update_count()`) and whether it is empty (`is_empty()`). It is empty if the
+/// update count is zero.
+pub trait Accountable {
+    /// The number of records
     ///
     /// This number is used in progress tracking to confirm the receipt of some number
     /// of outstanding records, and it is highly load bearing. The main restriction is
     /// imposed on the `LengthPreservingContainerBuilder` trait, whose implementors
-    /// must preserve the number of items.
-    fn len(&self) -> usize;
+    /// must preserve the number of records.
+    fn record_count(&self) -> i64;
 
-    /// Determine if the container contains any elements, corresponding to `len() == 0`.
-    fn is_empty(&self) -> bool {
-        self.len() == 0
-    }
-
-    /// Remove all contents from `self` while retaining allocated memory.
-    /// After calling `clear`, `is_empty` must return `true` and `len` 0.
-    fn clear(&mut self);
+    /// Determine if this contains any updates, corresponding to `update_count() == 0`.
+    /// It is a correctness error for this to by anything other than `self.record_count() == 0`.
+    #[inline] fn is_empty(&self) -> bool { self.record_count() == 0 }
+}
 
+/// A container that allows iteration morally equivalent to [`IntoIterator`].
+///
+/// Iterating the container presents items in an implementation-specific order.
+/// The container's contents are not changed.
+pub trait IterContainer {
+    /// The type of elements when reading non-destructively from the container.
+    type ItemRef<'a> where Self: 'a;
     /// Iterator type when reading from the container.
     type Iter<'a>: Iterator<Item=Self::ItemRef<'a>> where Self: 'a;
-
     /// Returns an iterator that reads the contents of this container.
     fn iter(&self) -> Self::Iter<'_>;
+}
 
+/// A container that can drain itself.
+///
+/// Draining the container presents items in an implementation-specific order.
+/// The container is in an undefined state after calling [`drain`]. Dropping
+/// the iterator also leaves the container in an undefined state.
+pub trait DrainContainer {
+    /// The type of elements when draining the container.
+    type Item<'a> where Self: 'a;
     /// Iterator type when draining the container.
     type DrainIter<'a>: Iterator<Item=Self::Item<'a>> where Self: 'a;
-
     /// Returns an iterator that drains the contents of this container.
     /// Drain leaves the container in an undefined state.
     fn drain(&mut self) -> Self::DrainIter<'_>;
 }
 
 /// A container that can be sized and reveals its capacity.
-pub trait SizableContainer: Container {
+pub trait SizableContainer {
     /// Indicates that the container is "full" and should be shipped.
     fn at_capacity(&self) -> bool;
     /// Restores `self` to its desired capacity, if it has one.
@@ -68,7 +60,10 @@ pub trait SizableContainer: Container {
     /// The `stash` argument is available, and may have the intended capacity.
     /// However, it may be non-empty, and may be of the wrong capacity. The
     /// method should guard against these cases.
-    fn ensure_capacity(&mut self, stash: &mut Option<Self>);
+    ///
+    /// Assume that the `stash` is in an undefined state, and properly clear it
+    /// before re-using it.
+    fn ensure_capacity(&mut self, stash: &mut Option<Self>) where Self: Sized;
 }
 
 /// A container that can absorb items of a specific type.
@@ -95,6 +90,9 @@ pub trait PushInto<T> {
 /// [`Self::finish`]. Implementations can recycle buffers, but should ensure that they clear
 /// any remaining elements.
 ///
+/// Implementations are allowed to re-use the contents of the mutable references left by the caller,
+/// but they should ensure that they clear the contents before doing so.
+///
 /// For example, a consolidating builder can aggregate differences in-place, but it has
 /// to ensure that it preserves the intended information.
 ///
@@ -102,7 +100,9 @@ pub trait PushInto<T> {
 /// decide to represent a push order for `extract` and `finish`, or not.
 pub trait ContainerBuilder: Default + 'static {
     /// The container type we're building.
-    type Container: Container + Clone + 'static;
+    // The container is `Clone` because `Tee` requires it, otherwise we need to repeat it
+    // all over Timely. `'static` because we don't want lifetimes everywhere.
+    type Container: Accountable + Default + Clone + 'static;
     /// Extract assembled containers, potentially leaving unfinished data behind. Can
     /// be called repeatedly, for example while the caller can send data.
     ///
@@ -124,8 +124,8 @@ pub trait ContainerBuilder: Default + 'static {
 
 /// A wrapper trait indicating that the container building will preserve the number of records.
 ///
-/// Specifically, the sum of lengths of all extracted and finished containers must equal the
-/// number of times that `push_into` is called on the container builder.
+/// Specifically, the sum of record counts of all extracted and finished containers must equal the
+/// number of accounted records that are pushed into the container builder.
 /// If you have any questions about this trait you are best off not implementing it.
 pub trait LengthPreservingContainerBuilder : ContainerBuilder { }
 
@@ -145,14 +145,14 @@ pub struct CapacityContainerBuilder<C>{
     pending: VecDeque<C>,
 }
 
-impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuilder<C> {
+impl<T, C: SizableContainer + Default + PushInto<T>> PushInto<T> for CapacityContainerBuilder<C> {
     #[inline]
     fn push_into(&mut self, item: T) {
         // Ensure capacity
         self.current.ensure_capacity(&mut self.empty);
 
         // Push item
-        self.current.push(item);
+        self.current.push_into(item);
 
         // Maybe flush
         if self.current.at_capacity() {
@@ -161,7 +161,7 @@ impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuil
     }
 }
 
-impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
+impl<C: Accountable + Default + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
     type Container = C;
 
     #[inline]
@@ -184,31 +184,25 @@ impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuild
     }
 }
 
-impl<C: Container + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
+impl<C: Accountable + SizableContainer + Default + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
 
-impl<T> Container for Vec<T> {
-    type ItemRef<'a> = &'a T where T: 'a;
-    type Item<'a> = T where T: 'a;
-
-    fn len(&self) -> usize {
-        Vec::len(self)
-    }
-
-    fn is_empty(&self) -> bool {
-        Vec::is_empty(self)
-    }
-
-    fn clear(&mut self) { Vec::clear(self) }
+impl<T> Accountable for Vec<T> {
+    #[inline] fn record_count(&self) -> i64 { i64::try_from(Vec::len(self)).unwrap() }
+    #[inline] fn is_empty(&self) -> bool { Vec::is_empty(self) }
+}
 
+impl<T> IterContainer for Vec<T> {
+    type ItemRef<'a> = &'a T where T: 'a;
     type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;
-
-    fn iter(&self) -> Self::Iter<'_> {
+    #[inline] fn iter(&self) -> Self::Iter<'_> {
         self.as_slice().iter()
     }
+}
 
+impl<T> DrainContainer for Vec<T> {
+    type Item<'a> = T where T: 'a;
     type DrainIter<'a> = std::vec::Drain<'a, T> where Self: 'a;
-
-    fn drain(&mut self) -> Self::DrainIter<'_> {
+    #[inline] fn drain(&mut self) -> Self::DrainIter<'_> {
         self.drain(..)
     }
 }
@@ -255,40 +249,21 @@ mod rc {
     use std::ops::Deref;
     use std::rc::Rc;
 
-    use crate::Container;
+    use crate::{IterContainer, DrainContainer};
 
-    impl<T: Container> Container for Rc<T> {
+    impl<T: crate::Accountable> crate::Accountable for Rc<T> {
+        #[inline] fn record_count(&self) -> i64 { std::ops::Deref::deref(self).record_count() }
+        #[inline] fn is_empty(&self) -> bool { std::ops::Deref::deref(self).is_empty() }
+    }
+    impl<T: IterContainer> IterContainer for Rc<T> {
         type ItemRef<'a> = T::ItemRef<'a> where Self: 'a;
-        type Item<'a> = T::ItemRef<'a> where Self: 'a;
-
-        fn len(&self) -> usize {
-            std::ops::Deref::deref(self).len()
-        }
-
-        fn is_empty(&self) -> bool {
-            std::ops::Deref::deref(self).is_empty()
-        }
-
-        fn clear(&mut self) {
-            // Try to reuse the allocation if possible
-            if let Some(inner) = Rc::get_mut(self) {
-                inner.clear();
-            } else {
-                *self = Self::default();
-            }
-        }
-
         type Iter<'a> = T::Iter<'a> where Self: 'a;
-
-        fn iter(&self) -> Self::Iter<'_> {
-            self.deref().iter()
-        }
-
+        #[inline] fn iter(&self) -> Self::Iter<'_> { self.deref().iter() }
+    }
+    impl<T: IterContainer> DrainContainer for Rc<T> {
+        type Item<'a> = T::ItemRef<'a> where Self: 'a;
         type DrainIter<'a> = T::Iter<'a> where Self: 'a;
-
-        fn drain(&mut self) -> Self::DrainIter<'_> {
-            self.iter()
-        }
+        #[inline] fn drain(&mut self) -> Self::DrainIter<'_> { self.iter() }
     }
 }
 
@@ -296,40 +271,21 @@ mod arc {
     use std::ops::Deref;
     use std::sync::Arc;
 
-    use crate::Container;
+    use crate::{IterContainer, DrainContainer};
 
-    impl<T: Container> Container for Arc<T> {
+    impl<T: crate::Accountable> crate::Accountable for Arc<T> {
+        #[inline] fn record_count(&self) -> i64 { std::ops::Deref::deref(self).record_count() }
+        #[inline] fn is_empty(&self) -> bool { std::ops::Deref::deref(self).is_empty() }
+    }
+    impl<T: IterContainer> IterContainer for Arc<T> {
         type ItemRef<'a> = T::ItemRef<'a> where Self: 'a;
-        type Item<'a> = T::ItemRef<'a> where Self: 'a;
-
-        fn len(&self) -> usize {
-            std::ops::Deref::deref(self).len()
-        }
-
-        fn is_empty(&self) -> bool {
-            std::ops::Deref::deref(self).is_empty()
-        }
-
-        fn clear(&mut self) {
-            // Try to reuse the allocation if possible
-            if let Some(inner) = Arc::get_mut(self) {
-                inner.clear();
-            } else {
-                *self = Self::default();
-            }
-        }
-
         type Iter<'a> = T::Iter<'a> where Self: 'a;
-
-        fn iter(&self) -> Self::Iter<'_> {
-            self.deref().iter()
-        }
-
+        #[inline] fn iter(&self) -> Self::Iter<'_> { self.deref().iter() }
+    }
+    impl<T: IterContainer> DrainContainer for Arc<T> {
+        type Item<'a> = T::ItemRef<'a> where Self: 'a;
         type DrainIter<'a> = T::Iter<'a> where Self: 'a;
-
-        fn drain(&mut self) -> Self::DrainIter<'_> {
-            self.iter()
-        }
+        #[inline] fn drain(&mut self) -> Self::DrainIter<'_> { self.iter() }
     }
 }
 
diff --git a/timely/examples/columnar.rs b/timely/examples/columnar.rs
index b190cd64e..d858aac79 100644
--- a/timely/examples/columnar.rs
+++ b/timely/examples/columnar.rs
@@ -1,13 +1,12 @@
 //! Wordcount based on the `columnar` crate.
 
-use {
-    std::collections::HashMap,
-    timely::{Container, container::CapacityContainerBuilder},
-    timely::dataflow::channels::pact::{ExchangeCore, Pipeline},
-    timely::dataflow::InputHandleCore,
-    timely::dataflow::operators::{Inspect, Operator, Probe},
-    timely::dataflow::ProbeHandle,
-};
+use std::collections::HashMap;
+
+use timely::container::{IterContainer, CapacityContainerBuilder};
+use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
+use timely::dataflow::InputHandleCore;
+use timely::dataflow::operators::{Inspect, Operator, Probe};
+use timely::dataflow::ProbeHandle;
 
 // Creates `WordCountContainer` and `WordCountReference` structs,
 // as well as various implementations relating them to `WordCount`.
@@ -177,21 +176,16 @@ mod container {
         }
     }
 
-    impl<C: columnar::ContainerBytes> timely::Container for Column<C> {
-        fn len(&self) -> usize { self.borrow().len() }
-        // This sets `self` to be an empty `Typed` variant, appropriate for pushing into.
-        fn clear(&mut self) {
-            match self {
-                Column::Typed(t) => t.clear(),
-                Column::Bytes(_) => *self = Column::Typed(Default::default()),
-                Column::Align(_) => *self = Column::Typed(Default::default()),
-            }
-        }
-
+    impl<C: columnar::ContainerBytes> timely::Accountable for Column<C> {
+        #[inline] fn record_count(&self) -> i64 { i64::try_from(self.borrow().len()).unwrap() }
+        #[inline] fn is_empty(&self) -> bool { self.borrow().is_empty() }
+    }
+    impl<C: columnar::ContainerBytes> timely::container::IterContainer for Column<C> {
         type ItemRef<'a> = C::Ref<'a>;
         type Iter<'a> = IterOwn<C::Borrowed<'a>>;
         fn iter<'a>(&'a self) -> Self::Iter<'a> { self.borrow().into_index_iter() }
-
+    }
+    impl<C: columnar::ContainerBytes> timely::container::DrainContainer for Column<C> {
         type Item<'a> = C::Ref<'a>;
         type DrainIter<'a> = IterOwn<C::Borrowed<'a>>;
         fn drain<'a>(&'a mut self) -> Self::DrainIter<'a> { self.borrow().into_index_iter() }
diff --git a/timely/src/dataflow/channels/mod.rs b/timely/src/dataflow/channels/mod.rs
index 6a351753b..36fb44bb1 100644
--- a/timely/src/dataflow/channels/mod.rs
+++ b/timely/src/dataflow/channels/mod.rs
@@ -39,7 +39,7 @@ impl<T, C: Container> Message<T, C> {
     }
 
     /// Forms a message, and pushes contents at `pusher`. Replaces `buffer` with what the pusher
-    /// leaves in place, or the container's default element. The buffer is cleared.
+    /// leaves in place, or the container's default element. The buffer is left in an undefined state.
     #[inline]
     pub fn push_at<P: Push<Message<T, C>>>(buffer: &mut C, time: T, pusher: &mut P) {
 
@@ -51,7 +51,6 @@ impl<T, C: Container> Message<T, C> {
 
         if let Some(message) = bundle {
             *buffer = message.data;
-            buffer.clear();
         }
     }
 }
diff --git a/timely/src/dataflow/channels/pact.rs b/timely/src/dataflow/channels/pact.rs
index bb99cd277..96aaed605 100644
--- a/timely/src/dataflow/channels/pact.rs
+++ b/timely/src/dataflow/channels/pact.rs
@@ -10,7 +10,8 @@
 use std::{fmt::{self, Debug}, marker::PhantomData};
 use std::rc::Rc;
 
-use crate::{Container, container::{ContainerBuilder, LengthPreservingContainerBuilder, SizableContainer, CapacityContainerBuilder, PushInto}};
+use crate::Accountable;
+use crate::container::{ContainerBuilder, DrainContainer, LengthPreservingContainerBuilder, SizableContainer, CapacityContainerBuilder, PushInto};
 use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
 use crate::communication::{Push, Pull};
 use crate::dataflow::channels::pushers::Exchange as ExchangePusher;
@@ -19,7 +20,6 @@ use crate::dataflow::channels::Message;
 use crate::logging::{TimelyLogger as Logger, MessagesEvent};
 use crate::progress::Timestamp;
 use crate::worker::AsWorker;
-use crate::Data;
 
 /// A `ParallelizationContract` allocates paired `Push` and `Pull` implementors.
 pub trait ParallelizationContract<T, C> {
@@ -35,7 +35,7 @@ pub trait ParallelizationContract<T, C> {
 #[derive(Debug)]
 pub struct Pipeline;
 
-impl<T: 'static, C: Container + 'static> ParallelizationContract<T, C> for Pipeline {
+impl<T: 'static, C: Accountable + 'static> ParallelizationContract<T, C> for Pipeline {
     type Pusher = LogPusher<ThreadPusher<Message<T, C>>>;
     type Puller = LogPuller<ThreadPuller<Message<T, C>>>;
     fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: Rc<[usize]>, logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
@@ -54,7 +54,8 @@ pub type Exchange<D, F> = ExchangeCore<CapacityContainerBuilder<Vec<D>>, F>;
 impl<CB, F> ExchangeCore<CB, F>
 where
     CB: LengthPreservingContainerBuilder,
-    for<'a> F: FnMut(&<CB::Container as Container>::Item<'a>)->u64
+    CB::Container: DrainContainer,
+    for<'a> F: FnMut(&<CB::Container as DrainContainer>::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
     pub fn new_core(func: F) -> ExchangeCore<CB, F> {
@@ -67,7 +68,7 @@ where
 
 impl<C, F> ExchangeCore<CapacityContainerBuilder<C>, F>
 where
-    C: SizableContainer,
+    C: SizableContainer + DrainContainer,
     for<'a> F: FnMut(&C::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
@@ -82,9 +83,9 @@ where
 // Exchange uses a `Box<Pushable>` because it cannot know what type of pushable will return from the allocator.
 impl<T: Timestamp, CB, H> ParallelizationContract<T, CB::Container> for ExchangeCore<CB, H>
 where
-    CB: ContainerBuilder + for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
-    CB::Container: Data + Send + crate::dataflow::channels::ContainerBytes,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64 + 'static,
+    CB: ContainerBuilder<Container: DrainContainer> + for<'a> PushInto<<CB::Container as DrainContainer>::Item<'a>>,
+    CB::Container: Send + crate::dataflow::channels::ContainerBytes,
+    for<'a> H: FnMut(&<CB::Container as DrainContainer>::Item<'a>) -> u64 + 'static,
 {
     type Pusher = ExchangePusher<
         T,
@@ -132,7 +133,7 @@ impl<P> LogPusher<P> {
     }
 }
 
-impl<T, C: Container, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<P> {
+impl<T, C: Accountable, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<P> {
     #[inline]
     fn push(&mut self, pair: &mut Option<Message<T, C>>) {
         if let Some(bundle) = pair {
@@ -150,7 +151,7 @@ impl<T, C: Container, P: Push<Message<T, C>>> Push<Message<T, C>> for LogPusher<
                     source: self.source,
                     target: self.target,
                     seq_no: self.counter - 1,
-                    length: bundle.data.len(),
+                    record_count: bundle.data.record_count(),
                 })
             }
         }
@@ -180,7 +181,7 @@ impl<P> LogPuller<P> {
     }
 }
 
-impl<T, C: Container, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<P> {
+impl<T, C: Accountable, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<P> {
     #[inline]
     fn pull(&mut self) -> &mut Option<Message<T, C>> {
         let result = self.puller.pull();
@@ -195,7 +196,7 @@ impl<T, C: Container, P: Pull<Message<T, C>>> Pull<Message<T, C>> for LogPuller<
                     source: bundle.from,
                     target,
                     seq_no: bundle.seq,
-                    length: bundle.data.len(),
+                    record_count: bundle.data.record_count(),
                 });
             }
         }
diff --git a/timely/src/dataflow/channels/pullers/counter.rs b/timely/src/dataflow/channels/pullers/counter.rs
index 6f71848a4..5903b2e4c 100644
--- a/timely/src/dataflow/channels/pullers/counter.rs
+++ b/timely/src/dataflow/channels/pullers/counter.rs
@@ -6,10 +6,10 @@ use std::cell::RefCell;
 use crate::dataflow::channels::Message;
 use crate::progress::ChangeBatch;
 use crate::communication::Pull;
-use crate::Container;
+use crate::Accountable;
 
 /// A wrapper which accounts records pulled past in a shared count map.
-pub struct Counter<T: Ord+Clone+'static, C, P: Pull<Message<T, C>>> {
+pub struct Counter<T, C, P> {
     pullable: P,
     consumed: Rc<RefCell<ChangeBatch<T>>>,
     phantom: ::std::marker::PhantomData<C>,
@@ -19,7 +19,7 @@ pub struct Counter<T: Ord+Clone+'static, C, P: Pull<Message<T, C>>> {
 pub struct ConsumedGuard<T: Ord + Clone + 'static> {
     consumed: Rc<RefCell<ChangeBatch<T>>>,
     time: Option<T>,
-    len: usize,
+    record_count: i64,
 }
 
 impl<T:Ord+Clone+'static> ConsumedGuard<T> {
@@ -32,11 +32,11 @@ impl<T:Ord+Clone+'static> Drop for ConsumedGuard<T> {
     fn drop(&mut self) {
         // SAFETY: we're in a Drop impl, so this runs at most once
         let time = self.time.take().unwrap();
-        self.consumed.borrow_mut().update(time, self.len as i64);
+        self.consumed.borrow_mut().update(time, self.record_count);
     }
 }
 
-impl<T:Ord+Clone+'static, C: Container, P: Pull<Message<T, C>>> Counter<T, C, P> {
+impl<T:Ord+Clone+'static, C: Accountable, P: Pull<Message<T, C>>> Counter<T, C, P> {
     /// Retrieves the next timestamp and batch of data.
     #[inline]
     pub fn next(&mut self) -> Option<&mut Message<T, C>> {
@@ -49,7 +49,7 @@ impl<T:Ord+Clone+'static, C: Container, P: Pull<Message<T, C>>> Counter<T, C, P>
             let guard = ConsumedGuard {
                 consumed: Rc::clone(&self.consumed),
                 time: Some(message.time.clone()),
-                len: message.data.len(),
+                record_count: message.data.record_count(),
             };
             Some((guard, message))
         }
diff --git a/timely/src/dataflow/channels/pushers/buffer.rs b/timely/src/dataflow/channels/pushers/buffer.rs
index be579936c..c99d76060 100644
--- a/timely/src/dataflow/channels/pushers/buffer.rs
+++ b/timely/src/dataflow/channels/pushers/buffer.rs
@@ -6,7 +6,7 @@ use crate::container::{ContainerBuilder, CapacityContainerBuilder, PushInto};
 use crate::dataflow::channels::Message;
 use crate::dataflow::operators::Capability;
 use crate::progress::Timestamp;
-use crate::{Container, Data};
+use crate::{Container, Accountable};
 
 /// Buffers data sent at the same time, for efficient communication.
 ///
@@ -44,7 +44,7 @@ impl<T, CB: Default, P> Buffer<T, CB, P> {
     }
 }
 
-impl<T, C: Container + Data, P: Push<Message<T, C>>> Buffer<T, CapacityContainerBuilder<C>, P> where T: Eq+Clone {
+impl<T, C: Container, P: Push<Message<T, C>>> Buffer<T, CapacityContainerBuilder<C>, P> where T: Eq+Clone {
     /// Returns a `Session`, which accepts data to send at the associated time
     #[inline]
     pub fn session(&mut self, time: &T) -> Session<'_, T, CapacityContainerBuilder<C>, P> {
diff --git a/timely/src/dataflow/channels/pushers/counter.rs b/timely/src/dataflow/channels/pushers/counter.rs
index a6a16ce5d..67d39318d 100644
--- a/timely/src/dataflow/channels/pushers/counter.rs
+++ b/timely/src/dataflow/channels/pushers/counter.rs
@@ -7,7 +7,7 @@ use std::cell::RefCell;
 use crate::progress::{ChangeBatch, Timestamp};
 use crate::dataflow::channels::Message;
 use crate::communication::Push;
-use crate::Container;
+use crate::Accountable;
 
 /// A wrapper which updates shared `produced` based on the number of records pushed.
 #[derive(Debug)]
@@ -17,11 +17,11 @@ pub struct Counter<T, C, P: Push<Message<T, C>>> {
     phantom: PhantomData<C>,
 }
 
-impl<T: Timestamp, C: Container, P> Push<Message<T, C>> for Counter<T, C, P> where P: Push<Message<T, C>> {
+impl<T: Timestamp, C: Accountable, P> Push<Message<T, C>> for Counter<T, C, P> where P: Push<Message<T, C>> {
     #[inline]
     fn push(&mut self, message: &mut Option<Message<T, C>>) {
         if let Some(message) = message {
-            self.produced.borrow_mut().update(message.time.clone(), message.data.len() as i64);
+            self.produced.borrow_mut().update(message.time.clone(), message.data.record_count());
         }
 
         // only propagate `None` if dirty (indicates flush)
diff --git a/timely/src/dataflow/channels/pushers/exchange.rs b/timely/src/dataflow/channels/pushers/exchange.rs
index e72085b4f..a9ea9b4fa 100644
--- a/timely/src/dataflow/channels/pushers/exchange.rs
+++ b/timely/src/dataflow/channels/pushers/exchange.rs
@@ -1,9 +1,8 @@
 //! The exchange pattern distributes pushed data between many target pushees.
 
 use crate::communication::Push;
-use crate::container::{ContainerBuilder, PushInto};
+use crate::container::{ContainerBuilder, DrainContainer, PushInto};
 use crate::dataflow::channels::Message;
-use crate::Container;
 
 /// Distribute containers to several pushers.
 ///
@@ -43,8 +42,8 @@ impl<CB: Default, H> DrainContainerDistributor<CB, H> {
 
 impl<CB, H> Distributor<CB::Container> for DrainContainerDistributor<CB, H>
 where
-    CB: ContainerBuilder + for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64,
+    CB: ContainerBuilder<Container: DrainContainer> + for<'a> PushInto<<CB::Container as DrainContainer>::Item<'a>>,
+    for<'a> H: FnMut(&<CB::Container as DrainContainer>::Item<'a>) -> u64,
 {
     fn partition<T: Clone, P: Push<Message<T, CB::Container>>>(&mut self, container: &mut CB::Container, time: &T, pushers: &mut [P]) {
         debug_assert_eq!(self.builders.len(), pushers.len());
diff --git a/timely/src/dataflow/channels/pushers/tee.rs b/timely/src/dataflow/channels/pushers/tee.rs
index 49d9e17d0..df2e2090d 100644
--- a/timely/src/dataflow/channels/pushers/tee.rs
+++ b/timely/src/dataflow/channels/pushers/tee.rs
@@ -17,7 +17,7 @@ pub struct Tee<T, C> {
     shared: PushList<T, C>,
 }
 
-impl<T: Data, C: Container + Data> Push<Message<T, C>> for Tee<T, C> {
+impl<T: Data, C: Container> Push<Message<T, C>> for Tee<T, C> {
     #[inline]
     fn push(&mut self, message: &mut Option<Message<T, C>>) {
         let mut pushers = self.shared.borrow_mut();
@@ -32,14 +32,14 @@ impl<T: Data, C: Container + Data> Push<Message<T, C>> for Tee<T, C> {
                 pushers[index-1].push(&mut None);
             }
         }
-        if pushers.len() > 0 {
+        if !pushers.is_empty() {
             let last = pushers.len() - 1;
             pushers[last].push(message);
         }
     }
 }
 
-impl<T, C: Container> Tee<T, C> {
+impl<T, C: Default> Tee<T, C> {
     /// Allocates a new pair of `Tee` and `TeeHelper`.
     pub fn new() -> (Tee<T, C>, TeeHelper<T, C>) {
         let shared = Rc::new(RefCell::new(Vec::new()));
@@ -52,7 +52,7 @@ impl<T, C: Container> Tee<T, C> {
     }
 }
 
-impl<T, C: Container> Clone for Tee<T, C> {
+impl<T, C: Default> Clone for Tee<T, C> {
     fn clone(&self) -> Self {
         Self {
             buffer: Default::default(),
diff --git a/timely/src/dataflow/operators/branch.rs b/timely/src/dataflow/operators/branch.rs
index 924bd196a..dad88be2b 100644
--- a/timely/src/dataflow/operators/branch.rs
+++ b/timely/src/dataflow/operators/branch.rs
@@ -93,7 +93,7 @@ pub trait BranchWhen<T>: Sized {
     fn branch_when(&self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self);
 }
 
-impl<S: Scope, C: Container + Data> BranchWhen<S::Timestamp> for StreamCore<S, C> {
+impl<S: Scope, C: Container> 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());
         builder.set_notify(false);
diff --git a/timely/src/dataflow/operators/core/capture/capture.rs b/timely/src/dataflow/operators/core/capture/capture.rs
index 410fe486e..4d8e3053e 100644
--- a/timely/src/dataflow/operators/core/capture/capture.rs
+++ b/timely/src/dataflow/operators/core/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, Data};
+use crate::Container;
 use crate::progress::ChangeBatch;
 use crate::progress::Timestamp;
 
 use super::{Event, EventPusher};
 
 /// Capture a stream of timestamped data for later replay.
-pub trait Capture<T: Timestamp, C: Container + Data> {
+pub trait Capture<T: Timestamp, C: Container> {
     /// Captures a stream of timestamped data for later replay.
     ///
     /// # Examples
@@ -113,7 +113,7 @@ pub trait Capture<T: Timestamp, C: Container + Data> {
     }
 }
 
-impl<S: Scope, C: Container + Data> Capture<S::Timestamp, C> for StreamCore<S, C> {
+impl<S: Scope, C: Container> Capture<S::Timestamp, C> for StreamCore<S, C> {
     fn capture_into<P: EventPusher<S::Timestamp, C>+'static>(&self, mut event_pusher: P) {
 
         let mut builder = OperatorBuilder::new("Capture".to_owned(), self.scope());
diff --git a/timely/src/dataflow/operators/core/capture/extract.rs b/timely/src/dataflow/operators/core/capture/extract.rs
index fcf024f31..c0af57d37 100644
--- a/timely/src/dataflow/operators/core/capture/extract.rs
+++ b/timely/src/dataflow/operators/core/capture/extract.rs
@@ -1,7 +1,8 @@
 //! Traits and types for extracting captured timely dataflow streams.
 
 use super::Event;
-use crate::{container::{SizableContainer, PushInto}};
+use crate::Container;
+use crate::container::{SizableContainer, DrainContainer, PushInto};
 
 /// Supports extracting a sequence of timestamp and data.
 pub trait Extract<T, C> {
@@ -50,8 +51,7 @@ pub trait Extract<T, C> {
 
 impl<T: Ord, C: SizableContainer> Extract<T, C> for ::std::sync::mpsc::Receiver<Event<T, C>>
 where
-    for<'a> C: PushInto<C::Item<'a>>,
-    for<'a> C::Item<'a>: Ord,
+    for<'a> C: Container + DrainContainer<Item<'a>: Ord> + PushInto<C::Item<'a>>,
 {
     fn extract(self) -> Vec<(T, C)> {
         let mut staged = std::collections::BTreeMap::new();
@@ -71,7 +71,7 @@ where
             to_sort.sort();
             let mut sorted = C::default();
             for datum in to_sort.into_iter() {
-                sorted.push(datum);
+                sorted.push_into(datum);
             }
             if !sorted.is_empty() {
                 result.push((time, sorted));
diff --git a/timely/src/dataflow/operators/core/capture/replay.rs b/timely/src/dataflow/operators/core/capture/replay.rs
index 49878b887..41ff50d12 100644
--- a/timely/src/dataflow/operators/core/capture/replay.rs
+++ b/timely/src/dataflow/operators/core/capture/replay.rs
@@ -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 + Clone + 'static, I> Replay<T, C> for I
+impl<T: Timestamp, C: Container, I> Replay<T, C> for I
 where
     I : IntoIterator,
     <I as IntoIterator>::Item: EventIterator<T, C>+'static,
diff --git a/timely/src/dataflow/operators/core/concat.rs b/timely/src/dataflow/operators/core/concat.rs
index 24fcef532..f377d3626 100644
--- a/timely/src/dataflow/operators/core/concat.rs
+++ b/timely/src/dataflow/operators/core/concat.rs
@@ -1,12 +1,11 @@
 //! Merges the contents of multiple streams.
 
-
-use crate::{Container, Data};
+use crate::Container;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{StreamCore, Scope};
 
 /// Merge the contents of two streams.
-pub trait Concat<G: Scope, C: Container> {
+pub trait Concat<G: Scope, C> {
     /// Merge the contents of two streams.
     ///
     /// # Examples
@@ -23,14 +22,14 @@ pub trait Concat<G: Scope, C: Container> {
     fn concat(&self, _: &StreamCore<G, C>) -> StreamCore<G, C>;
 }
 
-impl<G: Scope, C: Container + Data> Concat<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container> Concat<G, C> for StreamCore<G, C> {
     fn concat(&self, other: &StreamCore<G, C>) -> StreamCore<G, C> {
         self.scope().concatenate([self.clone(), other.clone()])
     }
 }
 
 /// Merge the contents of multiple streams.
-pub trait Concatenate<G: Scope, C: Container> {
+pub trait Concatenate<G: Scope, C> {
     /// Merge the contents of multiple streams.
     ///
     /// # Examples
@@ -52,7 +51,7 @@ pub trait Concatenate<G: Scope, C: Container> {
         I: IntoIterator<Item=StreamCore<G, C>>;
 }
 
-impl<G: Scope, C: Container + Data> Concatenate<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container> Concatenate<G, C> for StreamCore<G, C> {
     fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
     where
         I: IntoIterator<Item=StreamCore<G, C>>
@@ -62,7 +61,7 @@ impl<G: Scope, C: Container + Data> Concatenate<G, C> for StreamCore<G, C> {
     }
 }
 
-impl<G: Scope, C: Container + Data> Concatenate<G, C> for G {
+impl<G: Scope, C: Container> Concatenate<G, C> for G {
     fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
     where
         I: IntoIterator<Item=StreamCore<G, C>>
diff --git a/timely/src/dataflow/operators/core/enterleave.rs b/timely/src/dataflow/operators/core/enterleave.rs
index bc221c4a6..c0889498e 100644
--- a/timely/src/dataflow/operators/core/enterleave.rs
+++ b/timely/src/dataflow/operators/core/enterleave.rs
@@ -26,7 +26,7 @@ use crate::logging::{TimelyLogger, MessagesEvent};
 use crate::progress::Timestamp;
 use crate::progress::timestamp::Refines;
 use crate::progress::{Source, Target};
-use crate::{Container, Data};
+use crate::{Accountable, Container};
 use crate::communication::Push;
 use crate::dataflow::channels::pushers::{Counter, Tee};
 use crate::dataflow::channels::Message;
@@ -35,7 +35,7 @@ use crate::dataflow::{StreamCore, Scope};
 use crate::dataflow::scopes::Child;
 
 /// Extension trait to move a `Stream` into a child of its current `Scope`.
-pub trait Enter<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container> {
+pub trait Enter<G: Scope, T: Timestamp+Refines<G::Timestamp>, C> {
     /// Moves the `Stream` argument into a child of its current `Scope`.
     ///
     /// # Examples
@@ -53,7 +53,7 @@ pub trait Enter<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container> {
     fn enter<'a>(&self, _: &Child<'a, G, T>) -> StreamCore<Child<'a, G, T>, C>;
 }
 
-impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Data+Container> Enter<G, T, C> for StreamCore<G, C> {
+impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container> Enter<G, T, C> for StreamCore<G, C> {
     fn enter<'a>(&self, scope: &Child<'a, G, T>) -> StreamCore<Child<'a, G, T>, C> {
 
         use crate::scheduling::Scheduler;
@@ -85,7 +85,7 @@ impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Data+Container> Enter<G, T
 }
 
 /// Extension trait to move a `Stream` to the parent of its current `Scope`.
-pub trait Leave<G: Scope, C: Container> {
+pub trait Leave<G: Scope, C> {
     /// Moves a `Stream` to the parent of its current `Scope`.
     ///
     /// # Examples
@@ -103,7 +103,7 @@ pub trait Leave<G: Scope, C: Container> {
     fn leave(&self) -> StreamCore<G, C>;
 }
 
-impl<G: Scope, C: Container + Data, T: Timestamp+Refines<G::Timestamp>> Leave<G, C> for StreamCore<Child<'_, G, T>, C> {
+impl<G: Scope, C: Container, T: Timestamp+Refines<G::Timestamp>> Leave<G, C> for StreamCore<Child<'_, G, T>, C> {
     fn leave(&self) -> StreamCore<G, C> {
 
         let scope = self.scope();
@@ -130,14 +130,14 @@ impl<G: Scope, C: Container + Data, T: Timestamp+Refines<G::Timestamp>> Leave<G,
 }
 
 
-struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container + Data> {
+struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> {
     targets: Counter<TInner, TContainer, Tee<TInner, TContainer>>,
     phantom: ::std::marker::PhantomData<TOuter>,
     activator: crate::scheduling::Activator,
     active: bool,
 }
 
-impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container + Data> Push<Message<TOuter, TContainer>> for IngressNub<TOuter, TInner, TContainer> {
+impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> Push<Message<TOuter, TContainer>> for IngressNub<TOuter, TInner, TContainer> {
     fn push(&mut self, element: &mut Option<Message<TOuter, TContainer>>) {
         if let Some(outer_message) = element {
             let data = ::std::mem::take(&mut outer_message.data);
@@ -159,13 +159,13 @@ impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container
 }
 
 
-struct EgressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Data> {
+struct EgressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer> {
     targets: Tee<TOuter, TContainer>,
     phantom: PhantomData<TInner>,
 }
 
 impl<TOuter, TInner, TContainer: Container> Push<Message<TInner, TContainer>> for EgressNub<TOuter, TInner, TContainer>
-where TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Data {
+where TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, {
     fn push(&mut self, message: &mut Option<Message<TInner, TContainer>>) {
         if let Some(inner_message) = message {
             let data = ::std::mem::take(&mut inner_message.data);
@@ -207,7 +207,7 @@ impl<P> LogPusher<P> {
 
 impl<T, C, P> Push<Message<T, C>> for LogPusher<P>
 where
-    C: Container,
+    C: Accountable,
     P: Push<Message<T, C>>,
 {
     fn push(&mut self, element: &mut Option<Message<T, C>>) {
@@ -218,7 +218,7 @@ where
                 source: self.index,
                 target: self.index,
                 seq_no: self.counter,
-                length: bundle.data.len(),
+                record_count: bundle.data.record_count(),
             };
             let recv_event = MessagesEvent {
                 is_send: false,
diff --git a/timely/src/dataflow/operators/core/exchange.rs b/timely/src/dataflow/operators/core/exchange.rs
index bb021a5d5..92c9d78f5 100644
--- a/timely/src/dataflow/operators/core/exchange.rs
+++ b/timely/src/dataflow/operators/core/exchange.rs
@@ -1,13 +1,13 @@
 //! Exchange records between workers.
 
-use crate::ExchangeData;
-use crate::container::{Container, SizableContainer, PushInto};
+use crate::Container;
+use crate::container::{DrainContainer, SizableContainer, PushInto};
 use crate::dataflow::channels::pact::ExchangeCore;
 use crate::dataflow::operators::generic::operator::Operator;
 use crate::dataflow::{Scope, StreamCore};
 
 /// Exchange records between workers.
-pub trait Exchange<C: Container> {
+pub trait Exchange<C: DrainContainer> {
     /// Exchange records between workers.
     ///
     /// The closure supplied should map a reference to a record to a `u64`,
@@ -30,9 +30,12 @@ pub trait Exchange<C: Container> {
 
 impl<G: Scope, C> Exchange<C> for StreamCore<G, C>
 where
-    C: SizableContainer + ExchangeData + crate::dataflow::channels::ContainerBytes,
-    C: for<'a> PushInto<C::Item<'a>>,
-
+    C: Container
+        + SizableContainer
+        + DrainContainer
+        + Send
+        + crate::dataflow::channels::ContainerBytes
+        + for<'a> PushInto<C::Item<'a>>,
 {
     fn exchange<F>(&self, route: F) -> StreamCore<G, C>
     where
diff --git a/timely/src/dataflow/operators/core/feedback.rs b/timely/src/dataflow/operators/core/feedback.rs
index 61bd5c196..067ea3db8 100644
--- a/timely/src/dataflow/operators/core/feedback.rs
+++ b/timely/src/dataflow/operators/core/feedback.rs
@@ -1,6 +1,6 @@
 //! Create cycles in a timely dataflow graph.
 
-use crate::{Container, Data};
+use crate::Container;
 use crate::container::CapacityContainerBuilder;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::channels::pushers::Tee;
@@ -36,7 +36,7 @@ pub trait Feedback<G: Scope> {
     ///            .connect_loop(handle);
     /// });
     /// ```
-    fn feedback<C: Container + Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>);
+    fn feedback<C: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>);
 }
 
 /// Creates a `StreamCore` and a `Handle` to later bind the source of that `StreamCore`.
@@ -64,12 +64,12 @@ pub trait LoopVariable<'a, G: Scope, T: Timestamp> {
     ///     });
     /// });
     /// ```
-    fn loop_variable<C: Container + Data>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>);
+    fn loop_variable<C: Container>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>);
 }
 
 impl<G: Scope> Feedback<G> for G {
 
-    fn feedback<C: Container + Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>) {
+    fn feedback<C: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>) {
 
         let mut builder = OperatorBuilder::new("Feedback".to_owned(), self.clone());
         builder.set_notify(false);
@@ -80,13 +80,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<C: Container + Data>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>) {
+    fn loop_variable<C: Container>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>) {
         self.feedback(Product::new(Default::default(), summary))
     }
 }
 
 /// Connect a `Stream` to the input of a loop variable.
-pub trait ConnectLoop<G: Scope, C: Container + Data> {
+pub trait ConnectLoop<G: Scope, C: Container> {
     /// Connect a `Stream` to be the input of a loop variable.
     ///
     /// # Examples
@@ -107,7 +107,7 @@ pub trait ConnectLoop<G: Scope, C: Container + Data> {
     fn connect_loop(&self, handle: Handle<G, C>);
 }
 
-impl<G: Scope, C: Container + Data> ConnectLoop<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container> ConnectLoop<G, C> for StreamCore<G, C> {
     fn connect_loop(&self, handle: Handle<G, C>) {
 
         let mut builder = handle.builder;
@@ -132,7 +132,7 @@ impl<G: Scope, C: Container + Data> ConnectLoop<G, C> for StreamCore<G, C> {
 
 /// A handle used to bind the source of a loop variable.
 #[derive(Debug)]
-pub struct Handle<G: Scope, C: Container + Data> {
+pub struct Handle<G: Scope, C: Container> {
     builder: OperatorBuilder<G>,
     summary: <G::Timestamp as Timestamp>::Summary,
     output: OutputWrapper<G::Timestamp, CapacityContainerBuilder<C>, Tee<G::Timestamp, C>>,
diff --git a/timely/src/dataflow/operators/core/filter.rs b/timely/src/dataflow/operators/core/filter.rs
index 80ef564b0..f82a76339 100644
--- a/timely/src/dataflow/operators/core/filter.rs
+++ b/timely/src/dataflow/operators/core/filter.rs
@@ -1,12 +1,12 @@
 //! Filters a stream by a predicate.
-use crate::container::{Container, SizableContainer, PushInto};
-use crate::Data;
+use crate::container::{DrainContainer, SizableContainer, PushInto};
+use crate::Container;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::operators::generic::operator::Operator;
 
 /// Extension trait for filtering.
-pub trait Filter<C: Container> {
+pub trait Filter<C: DrainContainer> {
     /// Returns a new instance of `self` containing only records satisfying `predicate`.
     ///
     /// # Examples
@@ -23,16 +23,14 @@ pub trait Filter<C: Container> {
     fn filter<P: FnMut(&C::Item<'_>)->bool+'static>(&self, predicate: P) -> Self;
 }
 
-impl<G: Scope, C: SizableContainer + Data> Filter<C> for StreamCore<G, C>
+impl<G: Scope, C: Container + SizableContainer + DrainContainer> Filter<C> for StreamCore<G, C>
 where
     for<'a> C: PushInto<C::Item<'a>>
 {
     fn filter<P: FnMut(&C::Item<'_>)->bool+'static>(&self, mut predicate: P) -> StreamCore<G, C> {
         self.unary(Pipeline, "Filter", move |_,_| move |input, output| {
             input.for_each(|time, data| {
-                if !data.is_empty() {
-                    output.session(&time).give_iterator(data.drain().filter(&mut predicate));
-                }
+                output.session(&time).give_iterator(data.drain().filter(&mut predicate));
             });
         })
     }
diff --git a/timely/src/dataflow/operators/core/input.rs b/timely/src/dataflow/operators/core/input.rs
index 5088f6eba..8db72755e 100644
--- a/timely/src/dataflow/operators/core/input.rs
+++ b/timely/src/dataflow/operators/core/input.rs
@@ -10,7 +10,7 @@ use crate::scheduling::{Schedule, Activator};
 use crate::progress::{Operate, operate::SharedProgress, Timestamp, ChangeBatch};
 use crate::progress::Source;
 use crate::progress::operate::Connectivity;
-use crate::{Container, Data};
+use crate::{Accountable, Container};
 use crate::communication::Push;
 use crate::dataflow::{Scope, ScopeParent, StreamCore};
 use crate::dataflow::channels::pushers::{Tee, Counter};
@@ -59,7 +59,7 @@ pub trait Input : Scope {
     ///     }
     /// });
     /// ```
-    fn new_input<C: Container + Data>(&mut self) -> (Handle<<Self as ScopeParent>::Timestamp, CapacityContainerBuilder<C>>, StreamCore<Self, C>);
+    fn new_input<C: Container>(&mut self) -> (Handle<<Self as ScopeParent>::Timestamp, CapacityContainerBuilder<C>>, StreamCore<Self, C>);
 
     /// Create a new [StreamCore] and [Handle] through which to supply input.
     ///
@@ -134,7 +134,7 @@ pub trait Input : Scope {
 
 use crate::order::TotalOrder;
 impl<G: Scope> Input for G where <G as ScopeParent>::Timestamp: TotalOrder {
-    fn new_input<C: Container + Data>(&mut self) -> (Handle<<G as ScopeParent>::Timestamp, CapacityContainerBuilder<C>>, StreamCore<G, C>) {
+    fn new_input<C: Container>(&mut self) -> (Handle<<G as ScopeParent>::Timestamp, CapacityContainerBuilder<C>>, StreamCore<G, C>) {
         let mut handle = Handle::new();
         let stream = self.input_from(&mut handle);
         (handle, stream)
@@ -224,7 +224,7 @@ pub struct Handle<T: Timestamp, CB: ContainerBuilder> {
     now_at: T,
 }
 
-impl<T: Timestamp, C: Container + Data> Handle<T, CapacityContainerBuilder<C>> {
+impl<T: Timestamp, C: Container> Handle<T, CapacityContainerBuilder<C>> {
     /// Allocates a new input handle, from which one can create timely streams.
     ///
     /// # Examples
@@ -374,7 +374,7 @@ impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
 
     /// Sends a container at each of the destinations. There can be more than one; clone if needed.
     /// Does not take `self` because `flush` and `extract` borrow `self` mutably.
-    /// Clears the container.
+    /// Leaves the container in an undefined state.
     // TODO: Find a better name for this function.
     #[inline]
     fn send_container(
@@ -392,7 +392,6 @@ impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
                 Message::push_at(container, now_at.clone(), &mut pushers[index]);
             }
         }
-        container.clear();
     }
 
     /// Closes the current epoch, flushing if needed, shutting if needed, and updating the frontier.
diff --git a/timely/src/dataflow/operators/core/inspect.rs b/timely/src/dataflow/operators/core/inspect.rs
index b8c41f97b..6491b9d54 100644
--- a/timely/src/dataflow/operators/core/inspect.rs
+++ b/timely/src/dataflow/operators/core/inspect.rs
@@ -1,12 +1,13 @@
 //! Extension trait and implementation for observing and action on streamed data.
 
-use crate::{Container, Data};
+use crate::Container;
+use crate::container::IterContainer;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::operators::generic::Operator;
 
 /// Methods to inspect records and batches of records on a stream.
-pub trait Inspect<G: Scope, C: Container>: InspectCore<G, C> + Sized {
+pub trait Inspect<G: Scope, C: IterContainer>: InspectCore<G, C> + Sized {
     /// Runs a supplied closure on each observed data element.
     ///
     /// # Examples
@@ -90,14 +91,14 @@ pub trait Inspect<G: Scope, C: Container>: InspectCore<G, C> + Sized {
     fn inspect_core<F>(&self, func: F) -> Self where F: FnMut(Result<(&G::Timestamp, &C), &[G::Timestamp]>)+'static;
 }
 
-impl<G: Scope, C: Container + Data> Inspect<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container + IterContainer> Inspect<G, C> for StreamCore<G, C> {
     fn inspect_core<F>(&self, func: F) -> Self where F: FnMut(Result<(&G::Timestamp, &C), &[G::Timestamp]>) + 'static {
         self.inspect_container(func)
     }
 }
 
 /// Inspect containers
-pub trait InspectCore<G: Scope, C: Container> {
+pub trait InspectCore<G: Scope, C> {
     /// Runs a supplied closure on each observed container, and each frontier advancement.
     ///
     /// Rust's `Result` type is used to distinguish the events, with `Ok` for time and data,
@@ -120,7 +121,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 + Data> InspectCore<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container> 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/core/map.rs b/timely/src/dataflow/operators/core/map.rs
index 8af70e4a4..140eeba62 100644
--- a/timely/src/dataflow/operators/core/map.rs
+++ b/timely/src/dataflow/operators/core/map.rs
@@ -1,13 +1,13 @@
 //! Extension methods for `StreamCore` based on record-by-record transformation.
 
-use crate::container::{Container, SizableContainer, PushInto};
-use crate::Data;
+use crate::container::{DrainContainer, SizableContainer, PushInto};
+use crate::Container;
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::generic::operator::Operator;
 
 /// Extension trait for `Stream`.
-pub trait Map<S: Scope, C: Container> {
+pub trait Map<S: Scope, C: DrainContainer> {
     /// Consumes each element of the stream and yields a new element.
     ///
     /// # Examples
@@ -24,7 +24,7 @@ pub trait Map<S: Scope, C: Container> {
     /// ```
     fn map<C2, D2, L>(&self, mut logic: L) -> StreamCore<S, C2>
     where
-        C2: SizableContainer + PushInto<D2> + Data,
+        C2: Container + SizableContainer + PushInto<D2>,
         L: FnMut(C::Item<'_>)->D2 + 'static,
     {
         self.flat_map(move |x| std::iter::once(logic(x)))
@@ -46,19 +46,19 @@ pub trait Map<S: Scope, C: Container> {
     fn flat_map<C2, I, L>(&self, logic: L) -> StreamCore<S, C2>
     where
         I: IntoIterator,
-        C2: SizableContainer + PushInto<I::Item> + Data,
+        C2: Container + SizableContainer + PushInto<I::Item>,
         L: FnMut(C::Item<'_>)->I + 'static,
     ;
 }
 
-impl<S: Scope, C: Container + Data> Map<S, C> for StreamCore<S, C> {
+impl<S: Scope, C: Container + DrainContainer> Map<S, C> for StreamCore<S, C> {
     // TODO : This would be more robust if it captured an iterator and then pulled an appropriate
     // TODO : number of elements from the iterator. This would allow iterators that produce many
     // TODO : records without taking arbitrarily long and arbitrarily much memory.
     fn flat_map<C2, I, L>(&self, mut logic: L) -> StreamCore<S, C2>
     where
         I: IntoIterator,
-        C2: SizableContainer + PushInto<I::Item> + Data,
+        C2: Container + SizableContainer + PushInto<I::Item>,
         L: FnMut(C::Item<'_>)->I + 'static,
     {
         self.unary(Pipeline, "FlatMap", move |_,_| move |input, output| {
diff --git a/timely/src/dataflow/operators/core/ok_err.rs b/timely/src/dataflow/operators/core/ok_err.rs
index 77fd1f527..339b51576 100644
--- a/timely/src/dataflow/operators/core/ok_err.rs
+++ b/timely/src/dataflow/operators/core/ok_err.rs
@@ -1,13 +1,13 @@
 //! Operators that separate one stream into two streams based on some condition
 
-use crate::container::{Container, SizableContainer, PushInto};
-use crate::Data;
+use crate::Container;
+use crate::container::{DrainContainer, SizableContainer, PushInto};
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
 use crate::dataflow::{Scope, StreamCore};
 
 /// Extension trait for `Stream`.
-pub trait OkErr<S: Scope, C: Container> {
+pub trait OkErr<S: Scope, C: DrainContainer> {
     /// Takes one input stream and splits it into two output streams.
     /// For each record, the supplied closure is called with the data.
     /// If it returns `Ok(x)`, then `x` will be sent
@@ -33,20 +33,20 @@ pub trait OkErr<S: Scope, C: Container> {
         logic: L,
     ) -> (StreamCore<S, C1>, StreamCore<S, C2>)
     where
-        C1: SizableContainer + PushInto<D1> + Data,
-        C2: SizableContainer + PushInto<D2> + Data,
+        C1: Container + SizableContainer + PushInto<D1>,
+        C2: Container + SizableContainer + PushInto<D2>,
         L: FnMut(C::Item<'_>) -> Result<D1,D2>+'static
     ;
 }
 
-impl<S: Scope, C: Container + Data> OkErr<S, C> for StreamCore<S, C> {
+impl<S: Scope, C: Container + DrainContainer> OkErr<S, C> for StreamCore<S, C> {
     fn ok_err<C1, D1, C2, D2, L>(
         &self,
         mut logic: L,
     ) -> (StreamCore<S, C1>, StreamCore<S, C2>)
     where
-        C1: SizableContainer + PushInto<D1> + Data,
-        C2: SizableContainer + PushInto<D2> + Data,
+        C1: Container + SizableContainer + PushInto<D1>,
+        C2: Container + SizableContainer + PushInto<D2>,
         L: FnMut(C::Item<'_>) -> Result<D1,D2>+'static
     {
         let mut builder = OperatorBuilder::new("OkErr".to_owned(), self.scope());
diff --git a/timely/src/dataflow/operators/core/partition.rs b/timely/src/dataflow/operators/core/partition.rs
index 58205364b..7602b4892 100644
--- a/timely/src/dataflow/operators/core/partition.rs
+++ b/timely/src/dataflow/operators/core/partition.rs
@@ -1,15 +1,14 @@
 //! Partition a stream of records into multiple streams.
 
-use timely_container::{Container, ContainerBuilder, PushInto};
-
+use crate::container::{DrainContainer, ContainerBuilder, PushInto};
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
 use crate::dataflow::operators::InputCapability;
 use crate::dataflow::{Scope, StreamCore};
-use crate::Data;
+use crate::Container;
 
 /// Partition a stream of records into multiple streams.
-pub trait Partition<G: Scope, C: Container> {
+pub trait Partition<G: Scope, C: DrainContainer> {
     /// Produces `parts` output streams, containing records produced and assigned by `route`.
     ///
     /// # Examples
@@ -31,15 +30,13 @@ pub trait Partition<G: Scope, C: Container> {
     fn partition<CB, D2, F>(&self, parts: u64, route: F) -> Vec<StreamCore<G, CB::Container>>
     where
         CB: ContainerBuilder + PushInto<D2>,
-        CB::Container: Data,
         F: FnMut(C::Item<'_>) -> (u64, D2) + 'static;
 }
 
-impl<G: Scope, C: Container + Data> Partition<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container + DrainContainer> Partition<G, C> for StreamCore<G, C> {
     fn partition<CB, D2, F>(&self, parts: u64, mut route: F) -> Vec<StreamCore<G, CB::Container>>
     where
         CB: ContainerBuilder + PushInto<D2>,
-        CB::Container: Data,
         F: FnMut(C::Item<'_>) -> (u64, D2) + 'static,
     {
         let mut builder = OperatorBuilder::new("Partition".to_owned(), self.scope());
diff --git a/timely/src/dataflow/operators/core/probe.rs b/timely/src/dataflow/operators/core/probe.rs
index cb3c96b34..324061991 100644
--- a/timely/src/dataflow/operators/core/probe.rs
+++ b/timely/src/dataflow/operators/core/probe.rs
@@ -13,7 +13,7 @@ use crate::dataflow::operators::generic::builder_raw::OperatorBuilder;
 
 
 use crate::dataflow::{StreamCore, Scope};
-use crate::{Container, Data};
+use crate::Container;
 
 /// Monitors progress at a `Stream`.
 pub trait Probe<G: Scope, C: Container> {
@@ -79,7 +79,7 @@ pub trait Probe<G: Scope, C: Container> {
     fn probe_with(&self, handle: &Handle<G::Timestamp>) -> StreamCore<G, C>;
 }
 
-impl<G: Scope, C: Container + Data> Probe<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: Container> Probe<G, C> for StreamCore<G, C> {
     fn probe(&self) -> Handle<G::Timestamp> {
 
         // the frontier is shared state; scope updates, handle reads.
diff --git a/timely/src/dataflow/operators/core/rc.rs b/timely/src/dataflow/operators/core/rc.rs
index fdc68b9d4..e9c622bc7 100644
--- a/timely/src/dataflow/operators/core/rc.rs
+++ b/timely/src/dataflow/operators/core/rc.rs
@@ -3,11 +3,11 @@
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::Operator;
 use crate::dataflow::{Scope, StreamCore};
-use crate::{Container, Data};
+use crate::Container;
 use std::rc::Rc;
 
 /// Convert a stream into a stream of shared containers
-pub trait SharedStream<S: Scope, C: Container> {
+pub trait SharedStream<S: Scope, C> {
     /// Convert a stream into a stream of shared data
     ///
     /// # Examples
@@ -24,7 +24,7 @@ pub trait SharedStream<S: Scope, C: Container> {
     fn shared(&self) -> StreamCore<S, Rc<C>>;
 }
 
-impl<S: Scope, C: Container + Data> SharedStream<S, C> for StreamCore<S, C> {
+impl<S: Scope, C: Container> SharedStream<S, C> for StreamCore<S, C> {
     fn shared(&self) -> StreamCore<S, Rc<C>> {
         self.unary(Pipeline, "Shared", move |_, _| {
             move |input, output| {
diff --git a/timely/src/dataflow/operators/core/reclock.rs b/timely/src/dataflow/operators/core/reclock.rs
index e74bedbc4..86ec5ae69 100644
--- a/timely/src/dataflow/operators/core/reclock.rs
+++ b/timely/src/dataflow/operators/core/reclock.rs
@@ -1,6 +1,6 @@
 //! Extension methods for `Stream` based on record-by-record transformation.
 
-use crate::{Container, Data};
+use crate::Container;
 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 + Data>(&self, clock: &StreamCore<S, TC>) -> Self;
+    fn reclock<TC: Container>(&self, clock: &StreamCore<S, TC>) -> Self;
 }
 
-impl<S: Scope, C: Container + Data> Reclock<S> for StreamCore<S, C> {
-    fn reclock<TC: Container + Data>(&self, clock: &StreamCore<S, TC>) -> StreamCore<S, C> {
+impl<S: Scope, C: Container> Reclock<S> for StreamCore<S, C> {
+    fn reclock<TC: Container>(&self, clock: &StreamCore<S, TC>) -> StreamCore<S, C> {
 
         let mut stash = vec![];
 
diff --git a/timely/src/dataflow/operators/core/to_stream.rs b/timely/src/dataflow/operators/core/to_stream.rs
index cfc6f429b..4d2dea525 100644
--- a/timely/src/dataflow/operators/core/to_stream.rs
+++ b/timely/src/dataflow/operators/core/to_stream.rs
@@ -1,7 +1,7 @@
 //! Conversion to the `StreamCore` type from iterators.
 
 use crate::container::{CapacityContainerBuilder, ContainerBuilder, SizableContainer, PushInto};
-use crate::{Container, Data};
+use crate::Container;
 use crate::dataflow::operators::generic::operator::source;
 use crate::dataflow::{StreamCore, Scope};
 
@@ -61,7 +61,7 @@ impl<CB: ContainerBuilder, I: IntoIterator+'static> ToStreamBuilder<CB> for I wh
 
 /// Converts to a timely [StreamCore]. Equivalent to [`ToStreamBuilder`] but
 /// uses a [`CapacityContainerBuilder`].
-pub trait ToStream<C: Container> {
+pub trait ToStream<C> {
     /// Converts to a timely [StreamCore].
     ///
     /// # Examples
@@ -81,7 +81,7 @@ pub trait ToStream<C: Container> {
     fn to_stream<S: Scope>(self, scope: &mut S) -> StreamCore<S, C>;
 }
 
-impl<C: SizableContainer + Data, I: IntoIterator+'static> ToStream<C> for I where C: PushInto<I::Item> {
+impl<C: Container + SizableContainer, I: IntoIterator+'static> ToStream<C> for I where C: PushInto<I::Item> {
     fn to_stream<S: Scope>(self, scope: &mut S) -> StreamCore<S, C> {
         ToStreamBuilder::<CapacityContainerBuilder<C>>::to_stream_with_builder(self, scope)
     }
diff --git a/timely/src/dataflow/operators/core/unordered_input.rs b/timely/src/dataflow/operators/core/unordered_input.rs
index 77c6ba81b..703a461d3 100644
--- a/timely/src/dataflow/operators/core/unordered_input.rs
+++ b/timely/src/dataflow/operators/core/unordered_input.rs
@@ -2,7 +2,8 @@
 
 use std::rc::Rc;
 use std::cell::RefCell;
-use crate::{Container, Data};
+
+use crate::Container;
 use crate::container::{ContainerBuilder, CapacityContainerBuilder};
 
 use crate::scheduling::{Schedule, ActivateOnDrop};
@@ -164,7 +165,7 @@ impl<T: Timestamp, CB: ContainerBuilder> UnorderedHandle<T, CB> {
     }
 }
 
-impl<T: Timestamp, C: Container + Data> UnorderedHandle<T, CapacityContainerBuilder<C>> {
+impl<T: Timestamp, C: Container> UnorderedHandle<T, CapacityContainerBuilder<C>> {
     /// Allocates a new automatically flushing session based on the supplied capability.
     #[inline]
     pub fn session(&mut self, cap: ActivateCapability<T>) -> ActivateOnDrop<AutoflushSession<'_, T, CapacityContainerBuilder<C>, Counter<T, C, Tee<T, C>>>> {
diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs
index dc9db01ce..a10ff2f40 100644
--- a/timely/src/dataflow/operators/generic/handles.rs
+++ b/timely/src/dataflow/operators/generic/handles.rs
@@ -15,14 +15,14 @@ use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::{Buffer, Session};
 use crate::dataflow::channels::Message;
 use crate::communication::{Push, Pull};
-use crate::{Container, Data};
+use crate::{Container, Accountable};
 use crate::container::{ContainerBuilder, CapacityContainerBuilder};
 
 use crate::dataflow::operators::InputCapability;
 use crate::dataflow::operators::capability::CapabilityTrait;
 
 /// Handle to an operator's input stream.
-pub struct InputHandleCore<T: Timestamp, C: Container, P: Pull<Message<T, C>>> {
+pub struct InputHandleCore<T: Timestamp, C, P: Pull<Message<T, C>>> {
     pull_counter: PullCounter<T, C, P>,
     internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>,
     /// Timestamp summaries from this input to each output.
@@ -36,7 +36,7 @@ pub struct InputHandleCore<T: Timestamp, C: Container, P: Pull<Message<T, C>>> {
 pub type InputHandle<T, D, P> = InputHandleCore<T, Vec<D>, P>;
 
 /// Handle to an operator's input stream and frontier.
-pub struct FrontieredInputHandleCore<'a, T: Timestamp, C: Container+'a, P: Pull<Message<T, C>>+'a> {
+pub struct FrontieredInputHandleCore<'a, T: Timestamp, C: 'a, P: Pull<Message<T, C>>+'a> {
     /// The underlying input handle.
     pub handle: &'a mut InputHandleCore<T, C, P>,
     /// The frontier as reported by timely progress tracking.
@@ -46,7 +46,7 @@ pub struct FrontieredInputHandleCore<'a, T: Timestamp, C: Container+'a, P: Pull<
 /// Handle to an operator's input stream and frontier, specialized to vectors.
 pub type FrontieredInputHandle<'a, T, D, P> = FrontieredInputHandleCore<'a, T, Vec<D>, P>;
 
-impl<T: Timestamp, C: Container, P: Pull<Message<T, C>>> InputHandleCore<T, C, P> {
+impl<T: Timestamp, C: Accountable, P: Pull<Message<T, C>>> InputHandleCore<T, C, P> {
 
     /// Reads the next input buffer (at some timestamp `t`) and a corresponding capability for `t`.
     /// The timestamp `t` of the input buffer can be retrieved by invoking `.time()` on the capability.
@@ -87,7 +87,7 @@ impl<T: Timestamp, C: Container, P: Pull<Message<T, C>>> InputHandleCore<T, C, P
 
 }
 
-impl<'a, T: Timestamp, C: Container, P: Pull<Message<T, C>>+'a> FrontieredInputHandleCore<'a, T, C, P> {
+impl<'a, T: Timestamp, C: Accountable, P: Pull<Message<T, C>>+'a> FrontieredInputHandleCore<'a, T, C, P> {
     /// Allocate a new frontiered input handle.
     pub fn new(handle: &'a mut InputHandleCore<T, C, P>, frontier: &'a MutableAntichain<T>) -> Self {
         FrontieredInputHandleCore {
@@ -134,13 +134,13 @@ impl<'a, T: Timestamp, C: Container, P: Pull<Message<T, C>>+'a> FrontieredInputH
     }
 }
 
-pub fn _access_pull_counter<T: Timestamp, C: Container, P: Pull<Message<T, C>>>(input: &mut InputHandleCore<T, C, P>) -> &mut PullCounter<T, C, P> {
+pub fn _access_pull_counter<T: Timestamp, C: Accountable, P: Pull<Message<T, C>>>(input: &mut InputHandleCore<T, C, P>) -> &mut PullCounter<T, C, P> {
     &mut input.pull_counter
 }
 
 /// Constructs an input handle.
 /// Declared separately so that it can be kept private when `InputHandle` is re-exported.
-pub fn new_input_handle<T: Timestamp, C: Container, P: Pull<Message<T, C>>>(
+pub fn new_input_handle<T: Timestamp, C: Accountable, P: Pull<Message<T, C>>>(
     pull_counter: PullCounter<T, C, P>,
     internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>,
     summaries: Rc<RefCell<PortConnectivity<T::Summary>>>,
@@ -231,7 +231,7 @@ impl<'a, T: Timestamp, CB: ContainerBuilder, P: Push<Message<T, CB::Container>>>
     }
 }
 
-impl<'a, T: Timestamp, C: Container + Data, P: Push<Message<T, C>>> OutputHandleCore<'a, T, CapacityContainerBuilder<C>, P> {
+impl<'a, T: Timestamp, C: Container, P: Push<Message<T, C>>> OutputHandleCore<'a, T, CapacityContainerBuilder<C>, P> {
     /// Obtains a session that can send data at the timestamp associated with capability `cap`.
     ///
     /// In order to send data at a future timestamp, obtain a capability for the new timestamp
diff --git a/timely/src/dataflow/operators/generic/operator.rs b/timely/src/dataflow/operators/generic/operator.rs
index 5f76bc59c..79f6076e4 100644
--- a/timely/src/dataflow/operators/generic/operator.rs
+++ b/timely/src/dataflow/operators/generic/operator.rs
@@ -12,11 +12,11 @@ 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, Data};
+use crate::Container;
 use crate::container::{ContainerBuilder, CapacityContainerBuilder};
 
 /// Methods to construct generic streaming and blocking operators.
-pub trait Operator<G: Scope, C1: Container> {
+pub trait Operator<G: Scope, C1> {
     /// Creates a new dataflow operator that partitions its input stream by a parallelization
     /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
     /// `logic` can read from the input stream, write to the output stream, and inspect the frontier at the input.
@@ -177,7 +177,7 @@ pub trait Operator<G: Scope, C1: Container> {
     /// ```
     fn binary_frontier<C2, CB, B, L, P1, P2>(&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, CB::Container>
     where
-        C2: Container + Data,
+        C2: Container,
         CB: ContainerBuilder,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, C1, P1::Puller>,
@@ -227,7 +227,7 @@ pub trait Operator<G: Scope, C1: Container> {
     ///    }
     /// }).unwrap();
     /// ```
-    fn binary_notify<C2: Container + Data,
+    fn binary_notify<C2: Container,
               CB: ContainerBuilder,
               L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P1::Puller>,
                        &mut InputHandleCore<G::Timestamp, C2, P2::Puller>,
@@ -269,7 +269,7 @@ pub trait Operator<G: Scope, C1: Container> {
     /// ```
     fn binary<C2, CB, B, L, P1, P2>(&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, CB::Container>
     where
-        C2: Container + Data,
+        C2: Container,
         CB: ContainerBuilder,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P1::Puller>,
@@ -307,7 +307,7 @@ pub trait Operator<G: Scope, C1: Container> {
         P: ParallelizationContract<G::Timestamp, C1>;
 }
 
-impl<G: Scope, C1: Container + Data> Operator<G, C1> for StreamCore<G, C1> {
+impl<G: Scope, C1: Container> Operator<G, C1> for StreamCore<G, C1> {
 
     fn unary_frontier<CB, B, L, P>(&self, pact: P, name: &str, constructor: B) -> StreamCore<G, CB::Container>
     where
@@ -388,7 +388,7 @@ impl<G: Scope, C1: Container + Data> Operator<G, C1> for StreamCore<G, C1> {
 
     fn binary_frontier<C2, CB, B, L, P1, P2>(&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, CB::Container>
     where
-        C2: Container + Data,
+        C2: Container,
         CB: ContainerBuilder,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut FrontieredInputHandleCore<G::Timestamp, C1, P1::Puller>,
@@ -419,7 +419,7 @@ impl<G: Scope, C1: Container + Data> Operator<G, C1> for StreamCore<G, C1> {
         stream
     }
 
-    fn binary_notify<C2: Container + Data,
+    fn binary_notify<C2: Container,
               CB: ContainerBuilder,
               L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P1::Puller>,
                        &mut InputHandleCore<G::Timestamp, C2, P2::Puller>,
@@ -447,7 +447,7 @@ impl<G: Scope, C1: Container + Data> Operator<G, C1> for StreamCore<G, C1> {
 
     fn binary<C2, CB, B, L, P1, P2>(&self, other: &StreamCore<G, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> StreamCore<G, CB::Container>
     where
-        C2: Container + Data,
+        C2: Container,
         CB: ContainerBuilder,
         B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L,
         L: FnMut(&mut InputHandleCore<G::Timestamp, C1, P1::Puller>,
@@ -580,7 +580,7 @@ where
 ///
 /// });
 /// ```
-pub fn empty<G: Scope, C: Container + Data>(scope: &G) -> StreamCore<G, C> {
+pub fn empty<G: Scope, C: Container>(scope: &G) -> StreamCore<G, C> {
     source::<_, CapacityContainerBuilder<C>, _, _>(scope, "Empty", |_capability, _info| |_output| {
         // drop capability, do nothing
     })
diff --git a/timely/src/dataflow/stream.rs b/timely/src/dataflow/stream.rs
index 7959c842a..b5bb93a01 100644
--- a/timely/src/dataflow/stream.rs
+++ b/timely/src/dataflow/stream.rs
@@ -11,7 +11,6 @@ use crate::dataflow::Scope;
 use crate::dataflow::channels::pushers::tee::TeeHelper;
 use crate::dataflow::channels::Message;
 use std::fmt::{self, Debug};
-use crate::Container;
 
 // use dataflow::scopes::root::loggers::CHANNELS_Q;
 
@@ -47,7 +46,7 @@ impl<S: Scope, C> Clone for StreamCore<S, C> {
 /// A stream batching data in vectors.
 pub type Stream<S, D> = StreamCore<S, Vec<D>>;
 
-impl<S: Scope, C: Container> StreamCore<S, C> {
+impl<S: Scope, C> StreamCore<S, C> {
     /// Connects the stream to a destination.
     ///
     /// The destination is described both by a `Target`, for progress tracking information, and a `P: Push` where the
@@ -76,7 +75,7 @@ impl<S: Scope, C: Container> StreamCore<S, C> {
     pub fn scope(&self) -> S { self.scope.clone() }
 
     /// Allows the assertion of a container type, for the benefit of type inference.
-    pub fn container<D: Container>(self) -> StreamCore<S, D> where Self: AsStream<S, D> { self.as_stream() }
+    pub fn container<C2>(self) -> StreamCore<S, C2> where Self: AsStream<S, C2> { self.as_stream() }
 }
 
 /// A type that can be translated to a [StreamCore].
diff --git a/timely/src/lib.rs b/timely/src/lib.rs
index 557b7c7e1..20f3f995e 100644
--- a/timely/src/lib.rs
+++ b/timely/src/lib.rs
@@ -65,7 +65,7 @@ pub use timely_communication::Config as CommunicationConfig;
 pub use worker::Config as WorkerConfig;
 pub use execute::Config as Config;
 
-pub use timely_container::Container;
+pub use timely_container::Accountable;
 /// Re-export of the `timely_container` crate.
 pub mod container {
     pub use timely_container::*;
@@ -104,6 +104,12 @@ pub mod scheduling;
 pub trait Data: Clone+'static { }
 impl<T: Clone+'static> Data for T { }
 
+/// A composite trait for types usable as containers in timely dataflow.
+///
+/// The `Container` trait is necessary for all containers in timely dataflow channels.
+pub trait Container: Accountable + Default + Clone + 'static { }
+impl<C: Accountable + Default + Clone + 'static> Container for C { }
+
 /// A composite trait for types usable on exchange channels in timely dataflow.
 ///
 /// The `ExchangeData` trait extends `Data` with any requirements imposed by the `timely_communication`
diff --git a/timely/src/logging.rs b/timely/src/logging.rs
index 2c2439319..9122f6da4 100644
--- a/timely/src/logging.rs
+++ b/timely/src/logging.rs
@@ -135,7 +135,7 @@ pub struct MessagesEvent {
     /// Message sequence number.
     pub seq_no: usize,
     /// Number of typed records in the message.
-    pub length: usize,
+    pub record_count: i64,
 }
 
 /// Records the starting and stopping of an operator.