diff --git a/container/src/lib.rs b/container/src/lib.rs
index 839f80167..fa97aff69 100644
--- a/container/src/lib.rs
+++ b/container/src/lib.rs
@@ -4,46 +4,32 @@
 
 use std::collections::VecDeque;
 
-/// A container transferring data through dataflow edges
+/// A type representing progress, with an update count.
 ///
-/// 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`).
+/// It describes its update count (`count()`).
 ///
-/// 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`.)
+/// We require [`Default`] for convenience purposes.
 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
     ///
     /// 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
+    /// imposed on the [`LengthPreservingContainerBuilder`] trait, whose implementors
     /// must preserve the number of items.
-    fn len(&self) -> usize;
-
-    /// Determine if the container contains any elements, corresponding to `len() == 0`.
-    fn is_empty(&self) -> bool {
-        self.len() == 0
-    }
+    fn count(&self) -> usize;
 
     /// Remove all contents from `self` while retaining allocated memory.
     /// After calling `clear`, `is_empty` must return `true` and `len` 0.
     fn clear(&mut self);
+}
+
+/// A container that can reveal its contents through iterating by reference and draining.
+pub trait IterContainer: Container {
+    /// 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;
 
     /// Iterator type when reading from the container.
     type Iter<'a>: Iterator<Item=Self::ItemRef<'a>> where Self: 'a;
@@ -116,8 +102,9 @@ pub trait ContainerBuilder: Default + 'static {
     /// Partitions `container` among `builders`, using the function `index` to direct items.
     fn partition<I>(container: &mut Self::Container, builders: &mut [Self], mut index: I)
     where
-        Self: for<'a> PushInto<<Self::Container as Container>::Item<'a>>,
-        I: for<'a> FnMut(&<Self::Container as Container>::Item<'a>) -> usize,
+        Self: for<'a> PushInto<<Self::Container as IterContainer>::Item<'a>>,
+        I: for<'a> FnMut(&<Self::Container as IterContainer>::Item<'a>) -> usize,
+        Self::Container: IterContainer,
     {
         for datum in container.drain() {
             let index = index(&datum);
@@ -142,6 +129,33 @@ pub trait ContainerBuilder: Default + 'static {
 /// If you have any questions about this trait you are best off not implementing it.
 pub trait LengthPreservingContainerBuilder : ContainerBuilder { }
 
+/// A container builder that never produces any outputs, and can be used to pass through data in
+/// operators.
+#[derive(Debug, Clone)]
+pub struct PassthroughContainerBuilder<C>(std::marker::PhantomData<C>);
+
+impl<C> Default for PassthroughContainerBuilder<C> {
+    #[inline(always)]
+    fn default() -> Self {
+        PassthroughContainerBuilder(std::marker::PhantomData)
+    }
+}
+
+impl<C: Container + Clone + 'static> ContainerBuilder for PassthroughContainerBuilder<C>
+{
+    type Container = C;
+
+    #[inline(always)]
+    fn extract(&mut self) -> Option<&mut Self::Container> {
+        None
+    }
+
+    #[inline(always)]
+    fn finish(&mut self) -> Option<&mut Self::Container> {
+        None
+    }
+}
+
 /// A default container builder that uses length and preferred capacity to chunk data.
 ///
 /// Maintains a single empty allocation between [`Self::push_into`] and [`Self::extract`], but not
@@ -165,7 +179,7 @@ impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuil
         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() {
@@ -189,7 +203,7 @@ impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuild
 
     #[inline]
     fn finish(&mut self) -> Option<&mut C> {
-        if !self.current.is_empty() {
+        if self.current.count() > 0 {
             self.pending.push_back(std::mem::take(&mut self.current));
         }
         self.empty = self.pending.pop_front();
@@ -197,22 +211,16 @@ impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuild
     }
 }
 
-impl<C: Container + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
+impl<C: Container + SizableContainer + Clone + 'static> LengthPreservingContainerBuilder for CapacityContainerBuilder<C> { }
 
 impl<T> Container for Vec<T> {
+    #[inline(always)] fn count(&self) -> usize { Vec::len(self) }
+    #[inline(always)] fn clear(&mut self) { Vec::clear(self) }
+}
+
+impl<T> IterContainer 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) }
-
     type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;
 
     fn iter(&self) -> Self::Iter<'_> {
@@ -268,21 +276,11 @@ mod rc {
     use std::ops::Deref;
     use std::rc::Rc;
 
-    use crate::Container;
+    use crate::{Container, IterContainer};
 
     impl<T: Container> Container 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) {
+        #[inline(always)] fn count(&self) -> usize { self.as_ref().count() }
+        #[inline(always)] fn clear(&mut self) {
             // Try to reuse the allocation if possible
             if let Some(inner) = Rc::get_mut(self) {
                 inner.clear();
@@ -290,7 +288,12 @@ mod rc {
                 *self = Self::default();
             }
         }
+    }
+
 
+    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;
         type Iter<'a> = T::Iter<'a> where Self: 'a;
 
         fn iter(&self) -> Self::Iter<'_> {
@@ -309,21 +312,11 @@ mod arc {
     use std::ops::Deref;
     use std::sync::Arc;
 
-    use crate::Container;
-
-    impl<T: Container> Container 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()
-        }
+    use crate::{Container, IterContainer};
 
-        fn clear(&mut self) {
+    impl<T: Container> Container for std::sync::Arc<T> {
+        #[inline(always)] fn count(&self) -> usize { self.as_ref().count() }
+        #[inline(always)] fn clear(&mut self) {
             // Try to reuse the allocation if possible
             if let Some(inner) = Arc::get_mut(self) {
                 inner.clear();
@@ -331,7 +324,11 @@ mod arc {
                 *self = Self::default();
             }
         }
+    }
 
+    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;
         type Iter<'a> = T::Iter<'a> where Self: 'a;
 
         fn iter(&self) -> Self::Iter<'_> {
diff --git a/timely/examples/columnar.rs b/timely/examples/columnar.rs
index d1869ee0a..5026fd9ca 100644
--- a/timely/examples/columnar.rs
+++ b/timely/examples/columnar.rs
@@ -2,7 +2,7 @@
 
 use {
     std::collections::HashMap,
-    timely::{Container, container::CapacityContainerBuilder},
+    timely::container::{CapacityContainerBuilder, IterContainer},
     timely::dataflow::channels::pact::{ExchangeCore, Pipeline},
     timely::dataflow::InputHandleCore,
     timely::dataflow::operators::{Inspect, Operator, Probe},
@@ -165,9 +165,10 @@ mod container {
         }
     }
 
-    impl<C: columnar::ContainerBytes> timely::Container for Column<C> {
-        fn len(&self) -> usize { self.borrow().len() }
+    impl<C: columnar::ContainerBytes> timely::container::Container for Column<C> {
+        #[inline(always)] fn count(&self) -> usize { self.borrow().len() }
         // This sets `self` to be an empty `Typed` variant, appropriate for pushing into.
+        #[inline(always)]
         fn clear(&mut self) {
             match self {
                 Column::Typed(t) => t.clear(),
@@ -175,7 +176,8 @@ mod container {
                 Column::Align(_) => *self = Column::Typed(Default::default()),
             }
         }
-
+    }
+    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() }
diff --git a/timely/src/dataflow/channels/pact.rs b/timely/src/dataflow/channels/pact.rs
index 55404015c..e7f761048 100644
--- a/timely/src/dataflow/channels/pact.rs
+++ b/timely/src/dataflow/channels/pact.rs
@@ -10,7 +10,7 @@
 use std::{fmt::{self, Debug}, marker::PhantomData};
 use std::rc::Rc;
 
-use crate::{Container, container::{ContainerBuilder, LengthPreservingContainerBuilder, SizableContainer, CapacityContainerBuilder, PushInto}};
+use crate::container::{ContainerBuilder, LengthPreservingContainerBuilder, IterContainer, Container, SizableContainer, CapacityContainerBuilder, PushInto};
 use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
 use crate::communication::{Push, Pull};
 use crate::dataflow::channels::pushers::Exchange as ExchangePusher;
@@ -52,8 +52,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: LengthPreservingContainerBuilder<Container: IterContainer>,
+    for<'a> F: FnMut(&<CB::Container as IterContainer>::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
     pub fn new_core(func: F) -> ExchangeCore<CB, F> {
@@ -66,7 +66,7 @@ where
 
 impl<C, F> ExchangeCore<CapacityContainerBuilder<C>, F>
 where
-    C: SizableContainer,
+    C: SizableContainer + IterContainer,
     for<'a> F: FnMut(&C::Item<'a>)->u64
 {
     /// Allocates a new `Exchange` pact from a distribution function.
@@ -81,10 +81,10 @@ where
 // Exchange uses a `Box<Pushable>` because it cannot know what type of pushable will return from the allocator.
 impl<T: Timestamp, CB, H: 'static> ParallelizationContract<T, CB::Container> for ExchangeCore<CB, H>
 where
-    CB: ContainerBuilder,
-    CB: for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
+    CB: ContainerBuilder<Container: IterContainer>,
+    CB: for<'a> PushInto<<CB::Container as IterContainer>::Item<'a>>,
     CB::Container: Data + Send + crate::dataflow::channels::ContainerBytes,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as IterContainer>::Item<'a>) -> u64
 {
     type Pusher = ExchangePusher<T, CB, LogPusher<T, CB::Container, Box<dyn Push<Message<T, CB::Container>>>>, H>;
     type Puller = LogPuller<T, CB::Container, Box<dyn Pull<Message<T, CB::Container>>>>;
@@ -147,7 +147,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(),
+                    length: bundle.data.count(),
                 })
             }
         }
@@ -194,7 +194,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(),
+                    length: bundle.data.count(),
                 });
             }
         }
diff --git a/timely/src/dataflow/channels/pullers/counter.rs b/timely/src/dataflow/channels/pullers/counter.rs
index 6f71848a4..4693dfef1 100644
--- a/timely/src/dataflow/channels/pullers/counter.rs
+++ b/timely/src/dataflow/channels/pullers/counter.rs
@@ -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(),
+                len: message.data.count(),
             };
             Some((guard, message))
         }
diff --git a/timely/src/dataflow/channels/pushers/buffer.rs b/timely/src/dataflow/channels/pushers/buffer.rs
index be579936c..fc6be3055 100644
--- a/timely/src/dataflow/channels/pushers/buffer.rs
+++ b/timely/src/dataflow/channels/pushers/buffer.rs
@@ -109,7 +109,7 @@ impl<T, CB: ContainerBuilder, P: Push<Message<T, CB::Container>>> Buffer<T, CB,
     // buffer always requires a container builder. We could expose the buffer's underlying pusher
     // directly, but this would bypass the buffer's time tracking.
     fn give_container(&mut self, container: &mut CB::Container) {
-        if !container.is_empty() {
+        if container.count() > 0 {
             self.flush();
             let time = self.time.as_ref().unwrap().clone();
             Message::push_at(container, time, &mut self.pusher);
diff --git a/timely/src/dataflow/channels/pushers/counter.rs b/timely/src/dataflow/channels/pushers/counter.rs
index a6a16ce5d..a1f39fbba 100644
--- a/timely/src/dataflow/channels/pushers/counter.rs
+++ b/timely/src/dataflow/channels/pushers/counter.rs
@@ -21,7 +21,7 @@ impl<T: Timestamp, C: Container, P> Push<Message<T, C>> for Counter<T, C, P> whe
     #[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.count() as i64);
         }
 
         // 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 fec26cae7..a495f9476 100644
--- a/timely/src/dataflow/channels/pushers/exchange.rs
+++ b/timely/src/dataflow/channels/pushers/exchange.rs
@@ -1,17 +1,17 @@
 //! The exchange pattern distributes pushed data between many target pushees.
 
 use crate::communication::Push;
-use crate::container::{ContainerBuilder, PushInto};
+use crate::container::{ContainerBuilder, IterContainer, PushInto};
 use crate::dataflow::channels::Message;
-use crate::{Container, Data};
+use crate::Data;
 
 // TODO : Software write combining
 /// Distributes records among target pushees according to a distribution function.
 pub struct Exchange<T, CB, P, H>
 where
-    CB: ContainerBuilder,
+    CB: ContainerBuilder<Container: IterContainer>,
     P: Push<Message<T, CB::Container>>,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as IterContainer>::Item<'a>) -> u64
 {
     pushers: Vec<P>,
     builders: Vec<CB>,
@@ -21,9 +21,9 @@ where
 
 impl<T: Clone, CB, P, H>  Exchange<T, CB, P, H>
 where
-    CB: ContainerBuilder,
+    CB: ContainerBuilder<Container: IterContainer>,
     P: Push<Message<T, CB::Container>>,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as IterContainer>::Item<'a>) -> u64
 {
     /// Allocates a new `Exchange` from a supplied set of pushers and a distribution function.
     pub fn new(pushers: Vec<P>, key: H) -> Exchange<T, CB, P, H> {
@@ -50,10 +50,10 @@ where
 
 impl<T: Eq+Data, CB, P, H> Push<Message<T, CB::Container>> for Exchange<T, CB, P, H>
 where
-    CB: ContainerBuilder,
-    CB: for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
+    CB: ContainerBuilder<Container: IterContainer>,
+    CB: for<'a> PushInto<<CB::Container as IterContainer>::Item<'a>>,
     P: Push<Message<T, CB::Container>>,
-    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
+    for<'a> H: FnMut(&<CB::Container as IterContainer>::Item<'a>) -> u64
 {
     #[inline(never)]
     fn push(&mut self, message: &mut Option<Message<T, CB::Container>>) {
diff --git a/timely/src/dataflow/operators/branch.rs b/timely/src/dataflow/operators/branch.rs
index 924bd196a..3798adc83 100644
--- a/timely/src/dataflow/operators/branch.rs
+++ b/timely/src/dataflow/operators/branch.rs
@@ -4,6 +4,7 @@ use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
 use crate::dataflow::{Scope, Stream, StreamCore};
 use crate::{Container, Data};
+use crate::container::PassthroughContainerBuilder;
 
 /// Extension trait for `Stream`.
 pub trait Branch<S: Scope, D: Data> {
@@ -99,8 +100,8 @@ impl<S: Scope, C: Container + Data> BranchWhen<S::Timestamp> for StreamCore<S, C
         builder.set_notify(false);
 
         let mut input = builder.new_input(self, Pipeline);
-        let (mut output1, stream1) = builder.new_output();
-        let (mut output2, stream2) = builder.new_output();
+        let (mut output1, stream1) = builder.new_output::<PassthroughContainerBuilder<_>>();
+        let (mut output2, stream2) = builder.new_output::<PassthroughContainerBuilder<_>>();
 
         builder.build(move |_| {
 
@@ -110,9 +111,9 @@ impl<S: Scope, C: Container + Data> BranchWhen<S::Timestamp> for StreamCore<S, C
 
                 input.for_each(|time, data| {
                     let mut out = if condition(time.time()) {
-                        output2_handle.session(&time)
+                        output2_handle.session_with_builder(&time)
                     } else {
-                        output1_handle.session(&time)
+                        output1_handle.session_with_builder(&time)
                     };
                     out.give_container(data);
                 });
diff --git a/timely/src/dataflow/operators/core/capture/extract.rs b/timely/src/dataflow/operators/core/capture/extract.rs
index fcf024f31..49b05d533 100644
--- a/timely/src/dataflow/operators/core/capture/extract.rs
+++ b/timely/src/dataflow/operators/core/capture/extract.rs
@@ -1,7 +1,7 @@
 //! Traits and types for extracting captured timely dataflow streams.
 
 use super::Event;
-use crate::{container::{SizableContainer, PushInto}};
+use crate::{container::{IterContainer, PushInto}};
 
 /// Supports extracting a sequence of timestamp and data.
 pub trait Extract<T, C> {
@@ -48,7 +48,7 @@ pub trait Extract<T, C> {
     fn extract(self) -> Vec<(T, C)>;
 }
 
-impl<T: Ord, C: SizableContainer> Extract<T, C> for ::std::sync::mpsc::Receiver<Event<T, C>>
+impl<T: Ord, C: IterContainer> 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,
@@ -71,11 +71,9 @@ where
             to_sort.sort();
             let mut sorted = C::default();
             for datum in to_sort.into_iter() {
-                sorted.push(datum);
-            }
-            if !sorted.is_empty() {
-                result.push((time, sorted));
+                sorted.push_into(datum);
             }
+            result.push((time, sorted));
         }
         result
     }
diff --git a/timely/src/dataflow/operators/core/capture/replay.rs b/timely/src/dataflow/operators/core/capture/replay.rs
index 49878b887..0d6081bf4 100644
--- a/timely/src/dataflow/operators/core/capture/replay.rs
+++ b/timely/src/dataflow/operators/core/capture/replay.rs
@@ -38,6 +38,7 @@
 //! allowing the replay to occur in a timely dataflow computation with more or fewer workers
 //! than that in which the stream was captured.
 
+use crate::container::PassthroughContainerBuilder;
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::Buffer as PushBuffer;
@@ -76,7 +77,7 @@ where
 
         let (targets, stream) = builder.new_output();
 
-        let mut output = PushBuffer::new(PushCounter::new(targets));
+        let mut output = PushBuffer::<_,PassthroughContainerBuilder<_>,_>::new(PushCounter::new(targets));
         let mut event_streams = self.into_iter().collect::<Vec<_>>();
         let mut started = false;
         let mut allocation: C = Default::default();
@@ -100,14 +101,14 @@ where
                                 progress.internals[0].extend(vec.into_iter());
                             },
                             Owned(Event::Messages(time, mut data)) => {
-                                output.session(&time).give_container(&mut data);
+                                output.session_with_builder(&time).give_container(&mut data);
                             }
                             Borrowed(Event::Progress(vec)) => {
                                 progress.internals[0].extend(vec.iter().cloned());
                             },
                             Borrowed(Event::Messages(time, data)) => {
                                 allocation.clone_from(data);
-                                output.session(time).give_container(&mut allocation);
+                                output.session_with_builder(time).give_container(&mut allocation);
                             }
                         }
                     }
diff --git a/timely/src/dataflow/operators/core/concat.rs b/timely/src/dataflow/operators/core/concat.rs
index 24fcef532..36fc1c71c 100644
--- a/timely/src/dataflow/operators/core/concat.rs
+++ b/timely/src/dataflow/operators/core/concat.rs
@@ -2,6 +2,7 @@
 
 
 use crate::{Container, Data};
+use crate::container::PassthroughContainerBuilder;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::{StreamCore, Scope};
 
@@ -77,7 +78,7 @@ impl<G: Scope, C: Container + Data> Concatenate<G, C> for G {
         let mut handles = sources.into_iter().map(|s| builder.new_input(&s, Pipeline)).collect::<Vec<_>>();
 
         // create one output handle for the concatenated results.
-        let (mut output, result) = builder.new_output();
+        let (mut output, result) = builder.new_output::<PassthroughContainerBuilder<_>>();
 
         // build an operator that plays out all input data.
         builder.build(move |_capability| {
@@ -86,7 +87,7 @@ impl<G: Scope, C: Container + Data> Concatenate<G, C> for G {
                 let mut output = output.activate();
                 for handle in handles.iter_mut() {
                     handle.for_each(|time, data| {
-                        output.session(&time).give_container(data);
+                        output.session_with_builder(&time).give_container(data);
                     })
                 }
             }
diff --git a/timely/src/dataflow/operators/core/enterleave.rs b/timely/src/dataflow/operators/core/enterleave.rs
index bc221c4a6..eedca6963 100644
--- a/timely/src/dataflow/operators/core/enterleave.rs
+++ b/timely/src/dataflow/operators/core/enterleave.rs
@@ -218,7 +218,7 @@ where
                 source: self.index,
                 target: self.index,
                 seq_no: self.counter,
-                length: bundle.data.len(),
+                length: bundle.data.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..bc04c6d88 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::{IterContainer, 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: IterContainer> {
     /// Exchange records between workers.
     ///
     /// The closure supplied should map a reference to a record to a `u64`,
@@ -30,7 +30,7 @@ pub trait Exchange<C: Container> {
 
 impl<G: Scope, C> Exchange<C> for StreamCore<G, C>
 where
-    C: SizableContainer + ExchangeData + crate::dataflow::channels::ContainerBytes,
+    C: SizableContainer + IterContainer + ExchangeData + crate::dataflow::channels::ContainerBytes,
     C: for<'a> PushInto<C::Item<'a>>,
 
 {
diff --git a/timely/src/dataflow/operators/core/feedback.rs b/timely/src/dataflow/operators/core/feedback.rs
index 61bd5c196..229dc6939 100644
--- a/timely/src/dataflow/operators/core/feedback.rs
+++ b/timely/src/dataflow/operators/core/feedback.rs
@@ -1,7 +1,7 @@
 //! Create cycles in a timely dataflow graph.
 
 use crate::{Container, Data};
-use crate::container::CapacityContainerBuilder;
+use crate::container::PassthroughContainerBuilder;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::channels::pushers::Tee;
 use crate::dataflow::operators::generic::OutputWrapper;
@@ -122,7 +122,7 @@ impl<G: Scope, C: Container + Data> ConnectLoop<G, C> for StreamCore<G, C> {
                 if let Some(new_time) = summary.results_in(cap.time()) {
                     let new_cap = cap.delayed(&new_time);
                     output
-                        .session(&new_cap)
+                        .session_with_builder(&new_cap)
                         .give_container(data);
                 }
             });
@@ -135,5 +135,5 @@ impl<G: Scope, C: Container + Data> ConnectLoop<G, C> for StreamCore<G, C> {
 pub struct Handle<G: Scope, C: Container + Data> {
     builder: OperatorBuilder<G>,
     summary: <G::Timestamp as Timestamp>::Summary,
-    output: OutputWrapper<G::Timestamp, CapacityContainerBuilder<C>, Tee<G::Timestamp, C>>,
+    output: OutputWrapper<G::Timestamp, PassthroughContainerBuilder<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..3e1901d74 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::container::{IterContainer, SizableContainer, PushInto};
 use crate::Data;
 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: IterContainer> {
     /// 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: SizableContainer + IterContainer + Data> 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 c5f974432..533047b0f 100644
--- a/timely/src/dataflow/operators/core/input.rs
+++ b/timely/src/dataflow/operators/core/input.rs
@@ -3,7 +3,7 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 
-use crate::container::{CapacityContainerBuilder, ContainerBuilder, PushInto};
+use crate::container::{CapacityContainerBuilder, ContainerBuilder, SizableContainer, PushInto};
 
 use crate::scheduling::{Schedule, Activator};
 
@@ -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: SizableContainer + Data>(&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: SizableContainer + Data>(&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: SizableContainer + Data> Handle<T, CapacityContainerBuilder<C>> {
     /// Allocates a new input handle, from which one can create timely streams.
     ///
     /// # Examples
@@ -440,7 +440,7 @@ impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
     /// });
     /// ```
     pub fn send_batch(&mut self, buffer: &mut CB::Container) {
-        if !buffer.is_empty() {
+        if !buffer.count() > 0 {
             // flush buffered elements to ensure local fifo.
             self.flush();
             Self::send_container(buffer, &mut self.buffer, &mut self.pushers, &self.now_at);
diff --git a/timely/src/dataflow/operators/core/inspect.rs b/timely/src/dataflow/operators/core/inspect.rs
index b8c41f97b..305e700ff 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::{IterContainer, PassthroughContainerBuilder};
 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: IterContainer + Data> 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,
@@ -127,7 +128,7 @@ impl<G: Scope, C: Container + Data> InspectCore<G, C> for StreamCore<G, C> {
     {
         use crate::progress::timestamp::Timestamp;
         let mut frontier = crate::progress::Antichain::from_elem(G::Timestamp::minimum());
-        self.unary_frontier(Pipeline, "InspectBatch", move |_,_| move |input, output| {
+        self.unary_frontier::<PassthroughContainerBuilder<_>,_,_,_>(Pipeline, "InspectBatch", move |_,_| move |input, output| {
             if input.frontier.frontier() != frontier.borrow() {
                 frontier.clear();
                 frontier.extend(input.frontier.frontier().iter().cloned());
@@ -135,7 +136,7 @@ impl<G: Scope, C: Container + Data> InspectCore<G, C> for StreamCore<G, C> {
             }
             input.for_each(|time, data| {
                 func(Ok((&time, &*data)));
-                output.session(&time).give_container(data);
+                output.session_with_builder(&time).give_container(data);
             });
         })
     }
diff --git a/timely/src/dataflow/operators/core/map.rs b/timely/src/dataflow/operators/core/map.rs
index 8af70e4a4..adfd1063c 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::container::{IterContainer, SizableContainer, PushInto};
 use crate::Data;
 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: IterContainer> {
     /// 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: SizableContainer + IterContainer + PushInto<D2> + Data,
         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: SizableContainer + IterContainer + PushInto<I::Item> + Data,
         L: FnMut(C::Item<'_>)->I + 'static,
     ;
 }
 
-impl<S: Scope, C: Container + Data> Map<S, C> for StreamCore<S, C> {
+impl<S: Scope, C: IterContainer + Data> 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: SizableContainer + IterContainer + PushInto<I::Item> + Data,
         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..e8d1aa835 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::container::{IterContainer, SizableContainer, PushInto};
 use crate::Data;
 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: IterContainer> {
     /// 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
@@ -39,7 +39,7 @@ pub trait OkErr<S: Scope, C: Container> {
     ;
 }
 
-impl<S: Scope, C: Container + Data> OkErr<S, C> for StreamCore<S, C> {
+impl<S: Scope, C: IterContainer + Data> OkErr<S, C> for StreamCore<S, C> {
     fn ok_err<C1, D1, C2, D2, L>(
         &self,
         mut logic: L,
diff --git a/timely/src/dataflow/operators/core/partition.rs b/timely/src/dataflow/operators/core/partition.rs
index 754e9182c..6517ab94a 100644
--- a/timely/src/dataflow/operators/core/partition.rs
+++ b/timely/src/dataflow/operators/core/partition.rs
@@ -1,6 +1,6 @@
 //! Partition a stream of records into multiple streams.
 
-use timely_container::{Container, ContainerBuilder, PushInto};
+use timely_container::{IterContainer, ContainerBuilder, PushInto};
 
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
@@ -8,7 +8,7 @@ use crate::dataflow::{Scope, StreamCore};
 use crate::Data;
 
 /// Partition a stream of records into multiple streams.
-pub trait Partition<G: Scope, C: Container> {
+pub trait Partition<G: Scope, C: IterContainer> {
     /// Produces `parts` output streams, containing records produced and assigned by `route`.
     ///
     /// # Examples
@@ -34,7 +34,7 @@ pub trait Partition<G: Scope, C: Container> {
         F: FnMut(C::Item<'_>) -> (u64, D2) + 'static;
 }
 
-impl<G: Scope, C: Container + Data> Partition<G, C> for StreamCore<G, C> {
+impl<G: Scope, C: IterContainer + Data> 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>,
diff --git a/timely/src/dataflow/operators/core/probe.rs b/timely/src/dataflow/operators/core/probe.rs
index cb3c96b34..997b3922e 100644
--- a/timely/src/dataflow/operators/core/probe.rs
+++ b/timely/src/dataflow/operators/core/probe.rs
@@ -14,6 +14,7 @@ use crate::dataflow::operators::generic::builder_raw::OperatorBuilder;
 
 use crate::dataflow::{StreamCore, Scope};
 use crate::{Container, Data};
+use crate::container::PassthroughContainerBuilder;
 
 /// Monitors progress at a `Stream`.
 pub trait Probe<G: Scope, C: Container> {
@@ -92,7 +93,7 @@ impl<G: Scope, C: Container + Data> Probe<G, C> for StreamCore<G, C> {
         let mut builder = OperatorBuilder::new("Probe".to_owned(), self.scope());
         let mut input = PullCounter::new(builder.new_input(self, Pipeline));
         let (tee, stream) = builder.new_output();
-        let mut output = PushBuffer::new(PushCounter::new(tee));
+        let mut output = PushBuffer::<_, PassthroughContainerBuilder<_>, _>::new(PushCounter::new(tee));
 
         let shared_frontier = Rc::downgrade(&handle.frontier);
         let mut started = false;
@@ -115,7 +116,7 @@ impl<G: Scope, C: Container + Data> Probe<G, C> for StreamCore<G, C> {
                 while let Some(message) = input.next() {
                     let time = &message.time;
                     let data = &mut message.data;
-                    output.session(time).give_container(data);
+                    output.session_with_builder(time).give_container(data);
                 }
                 output.cease();
 
diff --git a/timely/src/dataflow/operators/core/rc.rs b/timely/src/dataflow/operators/core/rc.rs
index fdc68b9d4..392637d22 100644
--- a/timely/src/dataflow/operators/core/rc.rs
+++ b/timely/src/dataflow/operators/core/rc.rs
@@ -4,6 +4,7 @@ use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::Operator;
 use crate::dataflow::{Scope, StreamCore};
 use crate::{Container, Data};
+use crate::container::PassthroughContainerBuilder;
 use std::rc::Rc;
 
 /// Convert a stream into a stream of shared containers
@@ -26,11 +27,11 @@ pub trait SharedStream<S: Scope, C: Container> {
 
 impl<S: Scope, C: Container + Data> SharedStream<S, C> for StreamCore<S, C> {
     fn shared(&self) -> StreamCore<S, Rc<C>> {
-        self.unary(Pipeline, "Shared", move |_, _| {
+        self.unary::<PassthroughContainerBuilder<_>,_,_,_>(Pipeline, "Shared", move |_, _| {
             move |input, output| {
                 input.for_each(|time, data| {
                     output
-                        .session(&time)
+                        .session_with_builder(&time)
                         .give_container(&mut Rc::new(std::mem::take(data)));
                 });
             }
diff --git a/timely/src/dataflow/operators/core/reclock.rs b/timely/src/dataflow/operators/core/reclock.rs
index e74bedbc4..d592669f6 100644
--- a/timely/src/dataflow/operators/core/reclock.rs
+++ b/timely/src/dataflow/operators/core/reclock.rs
@@ -1,6 +1,7 @@
 //! Extension methods for `Stream` based on record-by-record transformation.
 
 use crate::{Container, Data};
+use crate::container::PassthroughContainerBuilder;
 use crate::order::PartialOrder;
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::channels::pact::Pipeline;
@@ -53,7 +54,7 @@ impl<S: Scope, C: Container + Data> Reclock<S> for StreamCore<S, C> {
 
         let mut stash = vec![];
 
-        self.binary_notify(clock, Pipeline, Pipeline, "Reclock", vec![], move |input1, input2, output, notificator| {
+        self.binary_notify::<_,PassthroughContainerBuilder<_>,_,_,_>(clock, Pipeline, Pipeline, "Reclock", vec![], move |input1, input2, output, notificator| {
 
             // stash each data input with its timestamp.
             input1.for_each(|cap, data| {
@@ -67,7 +68,7 @@ impl<S: Scope, C: Container + Data> Reclock<S> for StreamCore<S, C> {
 
             // each time with complete stash can be flushed.
             notificator.for_each(|cap,_,_| {
-                let mut session = output.session(&cap);
+                let mut session = output.session_with_builder(&cap);
                 for &mut (ref t, ref mut data) in &mut stash {
                     if t.less_equal(cap.time()) {
                         session.give_container(data);
diff --git a/timely/src/dataflow/operators/core/unordered_input.rs b/timely/src/dataflow/operators/core/unordered_input.rs
index 77c6ba81b..72d9c6ad4 100644
--- a/timely/src/dataflow/operators/core/unordered_input.rs
+++ b/timely/src/dataflow/operators/core/unordered_input.rs
@@ -2,8 +2,8 @@
 
 use std::rc::Rc;
 use std::cell::RefCell;
-use crate::{Container, Data};
-use crate::container::{ContainerBuilder, CapacityContainerBuilder};
+use crate::Data;
+use crate::container::{ContainerBuilder, CapacityContainerBuilder, SizableContainer};
 
 use crate::scheduling::{Schedule, ActivateOnDrop};
 
@@ -164,7 +164,7 @@ impl<T: Timestamp, CB: ContainerBuilder> UnorderedHandle<T, CB> {
     }
 }
 
-impl<T: Timestamp, C: Container + Data> UnorderedHandle<T, CapacityContainerBuilder<C>> {
+impl<T: Timestamp, C: SizableContainer + Data> 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 6a289e95e..09cd20733 100644
--- a/timely/src/dataflow/operators/generic/handles.rs
+++ b/timely/src/dataflow/operators/generic/handles.rs
@@ -16,7 +16,7 @@ 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::{ContainerBuilder, CapacityContainerBuilder};
+use crate::container::{ContainerBuilder, CapacityContainerBuilder, SizableContainer};
 use crate::logging::TimelyLogger as Logger;
 
 use crate::dataflow::operators::InputCapability;
@@ -235,7 +235,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: SizableContainer + Data, 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 1c51f7ecd..4b17a891c 100644
--- a/timely/src/dataflow/operators/generic/operator.rs
+++ b/timely/src/dataflow/operators/generic/operator.rs
@@ -13,7 +13,7 @@ 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::{ContainerBuilder, CapacityContainerBuilder};
+use crate::container::{ContainerBuilder, PassthroughContainerBuilder};
 
 /// Methods to construct generic streaming and blocking operators.
 pub trait Operator<G: Scope, C1: Container> {
@@ -583,7 +583,13 @@ where
 /// });
 /// ```
 pub fn empty<G: Scope, C: Container + Data>(scope: &G) -> StreamCore<G, C> {
-    source::<_, CapacityContainerBuilder<C>, _, _>(scope, "Empty", |_capability, _info| |_output| {
+    let mut builder = OperatorBuilder::new("Empty".to_owned(), scope.clone());
+
+    let (_output, stream) = builder.new_output::<PassthroughContainerBuilder<C>>();
+    builder.set_notify(false);
+    builder.build(|_caps| |_frontier| {
         // drop capability, do nothing
-    })
+    });
+
+    stream
 }