diff --git a/container/src/lib.rs b/container/src/lib.rs
index d1cd022ad..eb5e98852 100644
--- a/container/src/lib.rs
+++ b/container/src/lib.rs
@@ -23,7 +23,7 @@ pub trait Container: Default + Clone + 'static {
     /// The type of elements when reading non-destructively from the container.
     type ItemRef<'a> where Self: 'a;
 
-    /// The type of elements when draining the continer.
+    /// The type of elements when draining the container.
     type Item<'a> where Self: 'a;
 
     /// Push `item` into self
@@ -75,6 +75,9 @@ pub trait PushInto<T> {
 /// chunked into individual containers, but is free to change the data representation to
 /// better fit the properties of the container.
 ///
+/// Types implementing this trait should provide appropriate [`PushInto`] implementations such
+/// that users can push the expected item types.
+///
 /// The owner extracts data in two ways. The opportunistic [`Self::extract`] method returns
 /// any ready data, but doesn't need to produce partial outputs. In contrast, [`Self::finish`]
 /// needs to produce all outputs, even partial ones.
@@ -83,23 +86,15 @@ pub trait PushInto<T> {
 /// to ensure that it preserves the intended information.
 ///
 /// The trait does not prescribe any specific ordering guarantees, and each implementation can
-/// decide to represent a `push`/`push_container` order for `extract` and `finish`, or not.
-// TODO: Consider adding `push_iterator` to receive an iterator of data.
+/// 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;
-    /// Add an item to a container.
-    ///
-    /// The restriction to [`SizeableContainer`] only exists so that types
-    /// relying on [`CapacityContainerBuilder`] only need to constrain their container
-    /// to [`Container`] instead of [`SizableContainer`], which otherwise would be a pervasive
-    /// requirement.
-    fn push<T>(&mut self, item: T) where Self::Container: SizableContainer + PushInto<T>;
-    /// Push a pre-built container.
-    fn push_container(&mut self, container: &mut Self::Container);
-    /// Extract assembled containers, potentially leaving unfinished data behind.
+    /// Extract assembled containers, potentially leaving unfinished data behind. Should
+    /// be called repeatedly until it returns `None`.
     fn extract(&mut self) -> Option<&mut Self::Container>;
-    /// Extract assembled containers and any unfinished data.
+    /// Extract assembled containers and any unfinished data. Should
+    /// be called repeatedly until it returns `None`.
     fn finish(&mut self) -> Option<&mut Self::Container>;
 }
 
@@ -126,11 +121,8 @@ pub trait SizableContainer: Container {
     fn reserve(&mut self, additional: usize);
 }
 
-impl<C: Container> ContainerBuilder for CapacityContainerBuilder<C> {
-    type Container = C;
-
-    #[inline]
-    fn push<T>(&mut self, item: T) where C: SizableContainer + PushInto<T> {
+impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuilder<C> {
+    fn push_into(&mut self, item: T) {
         if self.current.capacity() == 0 {
             self.current = self.empty.take().unwrap_or_default();
             // Discard any non-uniform capacity container.
@@ -153,9 +145,35 @@ impl<C: Container> ContainerBuilder for CapacityContainerBuilder<C> {
             self.pending.push_back(std::mem::take(&mut self.current));
         }
     }
+}
+
+impl<C: Container> ContainerBuilder for CapacityContainerBuilder<C> {
+    type Container = C;
+
+    #[inline]
+    fn extract(&mut self) -> Option<&mut C> {
+        if let Some(container) = self.pending.pop_front() {
+            self.empty = Some(container);
+            self.empty.as_mut()
+        } else {
+            None
+        }
+    }
 
     #[inline]
-    fn push_container(&mut self, container: &mut Self::Container) {
+    fn finish(&mut self) -> Option<&mut C> {
+        if self.current.len() > 0 {
+            self.pending.push_back(std::mem::take(&mut self.current));
+        }
+        self.extract()
+    }
+}
+
+impl<C: Container> CapacityContainerBuilder<C> {
+    /// Push a pre-formed container at this builder. This exists to maintain
+    /// API compatibility.
+    #[inline]
+    pub fn push_container(&mut self, container: &mut C) {
         if !container.is_empty() {
             // Flush to maintain FIFO ordering.
             if self.current.len() > 0 {
@@ -170,26 +188,58 @@ impl<C: Container> ContainerBuilder for CapacityContainerBuilder<C> {
             self.pending.push_back(std::mem::replace(container, empty));
         }
     }
+}
+
+/// A container builder that absorbs entire containers. Maintains FIFO order.
+pub struct BufferingContainerBuilder<C> {
+    /// Container that we're extracting.
+    current: Option<C>,
+    /// Completed containers pending to be sent.
+    pending: VecDeque<C>,
+}
 
+impl<C> Default for BufferingContainerBuilder<C> {
     #[inline]
-    fn extract(&mut self) -> Option<&mut C> {
+    fn default() -> Self {
+        Self {
+            current: None,
+            pending: VecDeque::default(),
+        }
+    }
+}
+
+impl<C: Container> ContainerBuilder for BufferingContainerBuilder<C> {
+    type Container = C;
+
+    #[inline]
+    fn extract(&mut self) -> Option<&mut Self::Container> {
         if let Some(container) = self.pending.pop_front() {
-            self.empty = Some(container);
-            self.empty.as_mut()
+            self.current = Some(container);
+            self.current.as_mut()
         } else {
             None
         }
     }
 
     #[inline]
-    fn finish(&mut self) -> Option<&mut C> {
-        if self.current.len() > 0 {
-            self.pending.push_back(std::mem::take(&mut self.current));
-        }
+    fn finish(&mut self) -> Option<&mut Self::Container> {
         self.extract()
     }
 }
 
+impl<C: Container> PushInto<&mut C> for BufferingContainerBuilder<C> {
+    #[inline]
+    fn push_into(&mut self, item: &mut C) {
+        if !item.is_empty() {
+            // Grab the last returned container, or a default one, to pass back to the caller
+            let mut empty = self.current.take().unwrap_or_default();
+            empty.clear();
+            let container = std::mem::replace(item, empty);
+            self.pending.push_back(container);
+        }
+    }
+}
+
 impl<T: Clone + 'static> Container for Vec<T> {
     type ItemRef<'a> = &'a T where T: 'a;
     type Item<'a> = T where T: 'a;
@@ -347,7 +397,7 @@ pub trait PushPartitioned: SizableContainer {
         F: FnMut(usize, &mut Self);
 }
 
-impl<T: SizableContainer> PushPartitioned for T where for<'a> T: PushInto<T::Item<'a>> {
+impl<C: SizableContainer> PushPartitioned for C where for<'a> C: PushInto<C::Item<'a>> {
     fn push_partitioned<I, F>(&mut self, buffers: &mut [Self], mut index: I, mut flush: F)
     where
         for<'a> I: FnMut(&Self::Item<'a>) -> usize,
@@ -383,7 +433,7 @@ pub mod buffer {
     /// The maximum buffer capacity in elements. Returns a number between [BUFFER_SIZE_BYTES]
     /// and 1, inclusively.
     pub const fn default_capacity<T>() -> usize {
-        let size = ::std::mem::size_of::<T>();
+        let size = std::mem::size_of::<T>();
         if size == 0 {
             BUFFER_SIZE_BYTES
         } else if size <= BUFFER_SIZE_BYTES {
diff --git a/timely/src/dataflow/channels/pushers/buffer.rs b/timely/src/dataflow/channels/pushers/buffer.rs
index ae878358f..d48d44772 100644
--- a/timely/src/dataflow/channels/pushers/buffer.rs
+++ b/timely/src/dataflow/channels/pushers/buffer.rs
@@ -2,7 +2,7 @@
 //! with the performance of batched sends.
 
 use crate::communication::Push;
-use crate::container::{ContainerBuilder, CapacityContainerBuilder, SizableContainer, PushInto};
+use crate::container::{ContainerBuilder, CapacityContainerBuilder, PushInto};
 use crate::dataflow::channels::{Bundle, Message};
 use crate::dataflow::operators::Capability;
 use crate::progress::Timestamp;
@@ -56,6 +56,16 @@ impl<T, C: Container, P: Push<Bundle<T, C>>> Buffer<T, CapacityContainerBuilder<
     pub fn autoflush_session(&mut self, cap: Capability<T>) -> AutoflushSession<T, CapacityContainerBuilder<C>, P> where T: Timestamp {
         self.autoflush_session_with_builder(cap)
     }
+
+    /// Gives an entire container at the current time. Only provided for
+    /// buffers of [`CapacityContainerBuilder`]s. Other container builders
+    /// should use [`PushInto`] instead.
+    fn give_container(&mut self, container: &mut C) {
+        if !container.is_empty() {
+            self.builder.push_container(container);
+            self.extract();
+        }
+    }
 }
 
 impl<T, CB: ContainerBuilder, P: Push<Bundle<T, CB::Container>>> Buffer<T, CB, P> where T: Eq+Clone {
@@ -101,25 +111,18 @@ impl<T, CB: ContainerBuilder, P: Push<Bundle<T, CB::Container>>> Buffer<T, CB, P
             Message::push_at(container, time, &mut self.pusher);
         }
     }
-
-    /// Gives an entire container at the current time.
-    fn give_container(&mut self, container: &mut CB::Container) {
-        if !container.is_empty() {
-            self.builder.push_container(container);
-            self.extract();
-        }
-    }
 }
 
-impl<T, CB: ContainerBuilder, P: Push<Bundle<T, CB::Container>>> Buffer<T, CB, P>
+impl<T, CB, P> Buffer<T, CB, P>
 where
     T: Eq+Clone,
-    CB::Container: SizableContainer,
+    CB: ContainerBuilder,
+    P: Push<Bundle<T, CB::Container>>
 {
     // Push a single item into the builder. Internal method for use by `Session`.
     #[inline]
-    fn give<D>(&mut self, data: D) where CB::Container: PushInto<D> {
-        self.builder.push(data);
+    fn give<D>(&mut self, data: D) where CB: PushInto<D> {
+        self.builder.push_into(data);
         self.extract();
     }
 }
@@ -133,33 +136,34 @@ pub struct Session<'a, T, CB, P> {
     buffer: &'a mut Buffer<T, CB, P>,
 }
 
-impl<'a, T, CB, P> Session<'a, T, CB, P>
+impl<'a, T, C: Container, P> Session<'a, T, CapacityContainerBuilder<C>, P>
 where
     T: Eq + Clone + 'a,
-    CB: ContainerBuilder + 'a,
-    P: Push<Bundle<T, CB::Container>> + 'a
+    P: Push<Bundle<T, C>> + 'a,
 {
-    /// Provide a container at the time specified by the [Session].
-    pub fn give_container(&mut self, container: &mut CB::Container) {
+    /// Provide a container at the time specified by the [Session]. Only provided for
+    /// buffers of [`CapacityContainerBuilder`]s. Other container builders
+    /// should use [`PushInto`] instead.
+    pub fn give_container(&mut self, container: &mut C) {
         self.buffer.give_container(container)
     }
+}
 
+impl<'a, T, CB, P> Session<'a, T, CB, P>
+where
+    T: Eq + Clone + 'a,
+    CB: ContainerBuilder + 'a,
+    P: Push<Bundle<T, CB::Container>> + 'a
+{
     /// Access the builder. Immutable access to prevent races with flushing
     /// the underlying buffer.
     pub fn builder(&self) -> &CB {
         self.buffer.builder()
     }
-}
 
-impl<'a, T, CB, P: Push<Bundle<T, CB::Container>>+'a> Session<'a, T, CB, P>
-where
-    T: Eq + Clone + 'a,
-    CB: ContainerBuilder + 'a,
-    CB::Container: SizableContainer,
-{
     /// Provides one record at the time specified by the `Session`.
     #[inline]
-    pub fn give<D>(&mut self, data: D) where CB::Container: PushInto<D> {
+    pub fn give<D>(&mut self, data: D) where CB: PushInto<D> {
         self.buffer.give(data);
     }
 
@@ -168,7 +172,7 @@ where
     pub fn give_iterator<I>(&mut self, iter: I)
     where
         I: Iterator,
-        CB::Container: PushInto<I::Item>,
+        CB: PushInto<I::Item>,
     {
         for item in iter {
             self.give(item);
@@ -197,15 +201,19 @@ where
 {
     /// Transmits a single record.
     #[inline]
-    pub fn give<D>(&mut self, data: D) where CB::Container: SizableContainer + PushInto<D> {
+    pub fn give<D>(&mut self, data: D)
+    where
+        CB: PushInto<D>,
+    {
         self.buffer.give(data);
     }
+
     /// Transmits records produced by an iterator.
     #[inline]
     pub fn give_iterator<I, D>(&mut self, iter: I)
     where
         I: Iterator<Item=D>,
-        CB::Container: SizableContainer + PushInto<D>,
+        CB: PushInto<D>,
     {
         for item in iter {
             self.give(item);
diff --git a/timely/src/dataflow/operators/core/input.rs b/timely/src/dataflow/operators/core/input.rs
index 40d42cf19..a9a24fdbe 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::{SizableContainer, PushInto};
+use crate::container::{CapacityContainerBuilder, ContainerBuilder, PushInto};
 
 use crate::scheduling::{Schedule, Activator};
 
@@ -60,7 +60,44 @@ pub trait Input : Scope {
     ///     }
     /// });
     /// ```
-    fn new_input<C: Container>(&mut self) -> (Handle<<Self as ScopeParent>::Timestamp, 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.
+    ///
+    /// The `new_input` method returns a pair `(Handle, StreamCore)` where the [StreamCore] can be used
+    /// immediately for timely dataflow construction, and the `Handle` is later used to introduce
+    /// data into the timely dataflow computation.
+    ///
+    /// The `Handle` also provides a means to indicate
+    /// to timely dataflow that the input has advanced beyond certain timestamps, allowing timely
+    /// to issue progress notifications.
+    ///
+    /// # Examples
+    /// ```
+    /// use std::rc::Rc;
+    /// use timely::*;
+    /// use timely::dataflow::operators::core::{Input, Inspect};
+    /// use timely::container::BufferingContainerBuilder;
+    ///
+    /// // construct and execute a timely dataflow
+    /// timely::execute(Config::thread(), |worker| {
+    ///
+    ///     // add an input and base computation off of it
+    ///     let mut input = worker.dataflow(|scope| {
+    ///         let (input, stream) = scope.new_input_with_builder::<BufferingContainerBuilder<Rc<Vec<_>>>>();
+    ///         stream.inspect(|x| println!("hello {:?}", x));
+    ///         input
+    ///     });
+    ///
+    ///     // introduce input, advance computation
+    ///     for round in 0..10 {
+    ///         input.send(&mut Rc::new(vec![round]));
+    ///         input.advance_to(round + 1);
+    ///         worker.step();
+    ///     }
+    /// });
+    /// ```
+    fn new_input_with_builder<CB: ContainerBuilder>(&mut self) -> (Handle<<Self as ScopeParent>::Timestamp, CB>, StreamCore<Self, CB::Container>);
 
     /// Create a new stream from a supplied interactive handle.
     ///
@@ -93,19 +130,25 @@ pub trait Input : Scope {
     ///     }
     /// });
     /// ```
-    fn input_from<C: Container>(&mut self, handle: &mut Handle<<Self as ScopeParent>::Timestamp, C>) -> StreamCore<Self, C>;
+    fn input_from<CB: ContainerBuilder>(&mut self, handle: &mut Handle<<Self as ScopeParent>::Timestamp, CB>) -> StreamCore<Self, CB::Container>;
 }
 
 use crate::order::TotalOrder;
 impl<G: Scope> Input for G where <G as ScopeParent>::Timestamp: TotalOrder {
-    fn new_input<C: Container>(&mut self) -> (Handle<<G as ScopeParent>::Timestamp, 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)
     }
 
-    fn input_from<C: Container>(&mut self, handle: &mut Handle<<G as ScopeParent>::Timestamp, C>) -> StreamCore<G, C> {
-        let (output, registrar) = Tee::<<G as ScopeParent>::Timestamp, C>::new();
+    fn new_input_with_builder<CB: ContainerBuilder>(&mut self) -> (Handle<<G as ScopeParent>::Timestamp, CB>, StreamCore<G, CB::Container>) {
+        let mut handle = Handle::new_with_builder();
+        let stream = self.input_from(&mut handle);
+        (handle, stream)
+    }
+
+    fn input_from<CB: ContainerBuilder>(&mut self, handle: &mut Handle<<G as ScopeParent>::Timestamp, CB>) -> StreamCore<G, CB::Container> {
+        let (output, registrar) = Tee::<<G as ScopeParent>::Timestamp, CB::Container>::new();
         let counter = Counter::new(output);
         let produced = counter.produced().clone();
 
@@ -174,16 +217,16 @@ impl<T:Timestamp> Operate<T> for Operator<T> {
 
 /// A handle to an input `StreamCore`, used to introduce data to a timely dataflow computation.
 #[derive(Debug)]
-pub struct Handle<T: Timestamp, C: Container> {
+pub struct Handle<T: Timestamp, CB: ContainerBuilder> {
     activate: Vec<Activator>,
     progress: Vec<Rc<RefCell<ChangeBatch<T>>>>,
-    pushers: Vec<Counter<T, C, Tee<T, C>>>,
-    buffer1: C,
-    buffer2: C,
+    pushers: Vec<Counter<T, CB::Container, Tee<T, CB::Container>>>,
+    builder: CB,
+    buffer: CB::Container,
     now_at: T,
 }
 
-impl<T: Timestamp, C: Container> Handle<T, C> {
+impl<T: Timestamp, C: Container> Handle<T, CapacityContainerBuilder<C>> {
     /// Allocates a new input handle, from which one can create timely streams.
     ///
     /// # Examples
@@ -216,8 +259,48 @@ impl<T: Timestamp, C: Container> Handle<T, C> {
             activate: Vec::new(),
             progress: Vec::new(),
             pushers: Vec::new(),
-            buffer1: Default::default(),
-            buffer2: Default::default(),
+            builder: CapacityContainerBuilder::default(),
+            buffer: Default::default(),
+            now_at: T::minimum(),
+        }
+    }
+}
+
+impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
+    /// Allocates a new input handle, from which one can create timely streams.
+    ///
+    /// # Examples
+    /// ```
+    /// use timely::*;
+    /// use timely::dataflow::operators::core::{Input, Inspect};
+    /// use timely::dataflow::operators::core::input::Handle;
+    ///
+    /// // construct and execute a timely dataflow
+    /// timely::execute(Config::thread(), |worker| {
+    ///
+    ///     // add an input and base computation off of it
+    ///     let mut input = Handle::new();
+    ///     worker.dataflow(|scope| {
+    ///         scope.input_from(&mut input)
+    ///              .container::<Vec<_>>()
+    ///              .inspect(|x| println!("hello {:?}", x));
+    ///     });
+    ///
+    ///     // introduce input, advance computation
+    ///     for round in 0..10 {
+    ///         input.send(round);
+    ///         input.advance_to(round + 1);
+    ///         worker.step();
+    ///     }
+    /// });
+    /// ```
+    pub fn new_with_builder() -> Self {
+        Self {
+            activate: Vec::new(),
+            progress: Vec::new(),
+            pushers: Vec::new(),
+            builder: CB::default(),
+            buffer: Default::default(),
             now_at: T::minimum(),
         }
     }
@@ -249,21 +332,21 @@ impl<T: Timestamp, C: Container> Handle<T, C> {
     ///     }
     /// });
     /// ```
-    pub fn to_stream<G: Scope>(&mut self, scope: &mut G) -> StreamCore<G, C>
+    pub fn to_stream<G>(&mut self, scope: &mut G) -> StreamCore<G, CB::Container>
     where
         T: TotalOrder,
-        G: ScopeParent<Timestamp=T>,
+        G: Scope<Timestamp=T>,
     {
         scope.input_from(self)
     }
 
     fn register(
         &mut self,
-        pusher: Counter<T, C, Tee<T, C>>,
+        pusher: Counter<T, CB::Container, Tee<T, CB::Container>>,
         progress: Rc<RefCell<ChangeBatch<T>>>,
     ) {
         // flush current contents, so new registrant does not see existing data.
-        if !self.buffer1.is_empty() { self.flush(); }
+        self.flush();
 
         // we need to produce an appropriate update to the capabilities for `progress`, in case a
         // user has decided to drive the handle around a bit before registering it.
@@ -274,26 +357,48 @@ impl<T: Timestamp, C: Container> Handle<T, C> {
         self.pushers.push(pusher);
     }
 
-    // flushes our buffer at each of the destinations. there can be more than one; clone if needed.
-    #[inline(never)]
+    /// Extract all ready contents from the builder and distribute to downstream operators.
+    #[inline]
+    fn extract(&mut self) {
+        while let Some(container) = self.builder.extract() {
+            Self::flush_container(container, &mut self.buffer, &mut self.pushers, &self.now_at);
+        }
+    }
+
+    /// Flush all contents and distribute to downstream operators.
+    #[inline]
     fn flush(&mut self) {
-        for index in 0 .. self.pushers.len() {
-            if index < self.pushers.len() - 1 {
-                self.buffer2.clone_from(&self.buffer1);
-                Message::push_at(&mut self.buffer2, self.now_at.clone(), &mut self.pushers[index]);
-                debug_assert!(self.buffer2.is_empty());
+        while let Some(container) = self.builder.finish() {
+            Self::flush_container(container, &mut self.buffer, &mut self.pushers, &self.now_at);
+        }
+    }
+
+    /// flushes our buffer 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.
+    #[inline(never)]
+    fn flush_container(
+        container: &mut CB::Container,
+        buffer: &mut CB::Container,
+        pushers: &mut [Counter<T, CB::Container, Tee<T, CB::Container>>],
+        now_at: &T
+    ) {
+        for index in 0 .. pushers.len() {
+            if index < pushers.len() - 1 {
+                buffer.clone_from(container);
+                Message::push_at(buffer, now_at.clone(), &mut pushers[index]);
+                debug_assert!(buffer.is_empty());
             }
             else {
-                Message::push_at(&mut self.buffer1, self.now_at.clone(), &mut self.pushers[index]);
-                debug_assert!(self.buffer1.is_empty());
+                Message::push_at(container, now_at.clone(), &mut pushers[index]);
+                debug_assert!(container.is_empty());
             }
         }
-        self.buffer1.clear();
+        container.clear();
     }
 
     // closes the current epoch, flushing if needed, shutting if needed, and updating the frontier.
     fn close_epoch(&mut self) {
-        if !self.buffer1.is_empty() { self.flush(); }
+        self.flush();
         for pusher in self.pushers.iter_mut() {
             pusher.done();
         }
@@ -334,25 +439,11 @@ impl<T: Timestamp, C: Container> Handle<T, C> {
     ///     }
     /// });
     /// ```
-    pub fn send_batch(&mut self, buffer: &mut C) {
-
+    pub fn send_batch(&mut self, buffer: &mut CB::Container) {
         if !buffer.is_empty() {
             // flush buffered elements to ensure local fifo.
-            if !self.buffer1.is_empty() { self.flush(); }
-
-            // push buffer (or clone of buffer) at each destination.
-            for index in 0 .. self.pushers.len() {
-                if index < self.pushers.len() - 1 {
-                    self.buffer2.clone_from(&buffer);
-                    Message::push_at(&mut self.buffer2, self.now_at.clone(), &mut self.pushers[index]);
-                    assert!(self.buffer2.is_empty());
-                }
-                else {
-                    Message::push_at(buffer, self.now_at.clone(), &mut self.pushers[index]);
-                    assert!(buffer.is_empty());
-                }
-            }
-            buffer.clear();
+            self.flush();
+            Self::flush_container(buffer, &mut self.buffer, &mut self.pushers, &self.now_at);
         }
     }
 
@@ -390,8 +481,19 @@ impl<T: Timestamp, C: Container> Handle<T, C> {
     }
 }
 
-impl<T: Timestamp, C: SizableContainer> Handle<T, C> {
+impl<T, CB, D> PushInto<D> for Handle<T, CB>
+where
+    T: Timestamp,
+    CB: ContainerBuilder + PushInto<D>,
+{
     #[inline]
+    fn push_into(&mut self, item: D) {
+        self.builder.push_into(item);
+        self.extract();
+    }
+}
+
+impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
     /// Sends one record into the corresponding timely dataflow `Stream`, at the current epoch.
     ///
     /// # Examples
@@ -419,21 +521,19 @@ impl<T: Timestamp, C: SizableContainer> Handle<T, C> {
     ///     }
     /// });
     /// ```
-    pub fn send<D>(&mut self, data: D) where C: PushInto<D> {
-        self.buffer1.push(data);
-        if self.buffer1.len() == self.buffer1.capacity() {
-            self.flush();
-        }
+    #[inline]
+    pub fn send<D>(&mut self, data: D) where CB: PushInto<D> {
+        self.push_into(data)
     }
 }
 
-impl<T: Timestamp, C: Container> Default for Handle<T, C> {
+impl<T: Timestamp, CB: ContainerBuilder> Default for Handle<T, CB> {
     fn default() -> Self {
-        Self::new()
+        Self::new_with_builder()
     }
 }
 
-impl<T:Timestamp, C: Container> Drop for Handle<T, C> {
+impl<T:Timestamp, CB: ContainerBuilder> Drop for Handle<T, CB> {
     fn drop(&mut self) {
         self.close_epoch();
     }
diff --git a/timely/src/dataflow/operators/core/mod.rs b/timely/src/dataflow/operators/core/mod.rs
index 19cba0afd..5074014bd 100644
--- a/timely/src/dataflow/operators/core/mod.rs
+++ b/timely/src/dataflow/operators/core/mod.rs
@@ -28,6 +28,6 @@ pub use inspect::{Inspect, InspectCore};
 pub use map::Map;
 pub use ok_err::OkErr;
 pub use probe::Probe;
-pub use to_stream::ToStream;
+pub use to_stream::{ToStream, ToStreamBuilder};
 pub use reclock::Reclock;
 pub use unordered_input::{UnorderedInput, UnorderedHandle};
diff --git a/timely/src/dataflow/operators/core/rc.rs b/timely/src/dataflow/operators/core/rc.rs
index c00108a7c..34403f39c 100644
--- a/timely/src/dataflow/operators/core/rc.rs
+++ b/timely/src/dataflow/operators/core/rc.rs
@@ -1,5 +1,6 @@
 //! Shared containers
 
+use crate::container::BufferingContainerBuilder;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::Operator;
 use crate::dataflow::{Scope, StreamCore};
@@ -27,13 +28,13 @@ pub trait SharedStream<S: Scope, C: Container> {
 impl<S: Scope, C: Container> SharedStream<S, C> for StreamCore<S, C> {
     fn shared(&self) -> StreamCore<S, Rc<C>> {
         let mut container = Default::default();
-        self.unary(Pipeline, "Shared", move |_, _| {
+        self.unary::<BufferingContainerBuilder<_>, _, _, _>(Pipeline, "Shared", move |_, _| {
             move |input, output| {
                 input.for_each(|time, data| {
                     data.swap(&mut container);
                     output
-                        .session(&time)
-                        .give_container(&mut Rc::new(std::mem::take(&mut container)));
+                        .session_with_builder(&time)
+                        .give(&mut Rc::new(std::mem::take(&mut container)));
                 });
             }
         })
diff --git a/timely/src/dataflow/operators/core/to_stream.rs b/timely/src/dataflow/operators/core/to_stream.rs
index dcb9aa684..c37092ea5 100644
--- a/timely/src/dataflow/operators/core/to_stream.rs
+++ b/timely/src/dataflow/operators/core/to_stream.rs
@@ -1,35 +1,40 @@
 //! Conversion to the `StreamCore` type from iterators.
 
-use crate::container::{SizableContainer, PushInto};
+use crate::container::{CapacityContainerBuilder, ContainerBuilder, SizableContainer, PushInto};
 use crate::Container;
 use crate::dataflow::operators::generic::operator::source;
 use crate::dataflow::{StreamCore, Scope};
 
-/// Converts to a timely [StreamCore].
-pub trait ToStream<C: Container> {
-    /// Converts to a timely [StreamCore].
+/// Converts to a timely [StreamCore], using a container builder.
+pub trait ToStreamBuilder<CB: ContainerBuilder> {
+    /// Converts to a timely [StreamCore], using the supplied container builder type.
     ///
     /// # Examples
     ///
     /// ```
-    /// use timely::dataflow::operators::core::{ToStream, Capture};
+    /// use timely::dataflow::operators::core::{ToStreamBuilder, Capture};
     /// use timely::dataflow::operators::core::capture::Extract;
+    /// use timely::container::CapacityContainerBuilder;
     ///
     /// let (data1, data2) = timely::example(|scope| {
-    ///     let data1 = (0..3).to_stream(scope).container::<Vec<_>>().capture();
-    ///     let data2 = vec![0,1,2].to_stream(scope).container::<Vec<_>>().capture();
+    ///     let data1 = ToStreamBuilder::<CapacityContainerBuilder<_>>::to_stream_with_builder(0..3, scope)
+    ///         .container::<Vec<_>>()
+    ///         .capture();
+    ///     let data2 = ToStreamBuilder::<CapacityContainerBuilder<_>>::to_stream_with_builder(vec![0,1,2], scope)
+    ///         .container::<Vec<_>>()
+    ///         .capture();
     ///     (data1, data2)
     /// });
     ///
     /// assert_eq!(data1.extract(), data2.extract());
     /// ```
-    fn to_stream<S: Scope>(self, scope: &mut S) -> StreamCore<S, C>;
+    fn to_stream_with_builder<S: Scope>(self, scope: &mut S) -> StreamCore<S, CB::Container>;
 }
 
-impl<C: 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> {
+impl<CB: ContainerBuilder, I: IntoIterator+'static> ToStreamBuilder<CB> for I where CB: PushInto<I::Item> {
+    fn to_stream_with_builder<S: Scope>(self, scope: &mut S) -> StreamCore<S, CB::Container> {
 
-        source(scope, "ToStream", |capability, info| {
+        source::<_, CB, _, _>(scope, "ToStreamBuilder", |capability, info| {
 
             // Acquire an activator, so that the operator can rescheduled itself.
             let activator = scope.activator_for(&info.address[..]);
@@ -40,7 +45,7 @@ impl<C: SizableContainer, I: IntoIterator+'static> ToStream<C> for I where C: Pu
             move |output| {
 
                 if let Some(element) = iterator.next() {
-                    let mut session = output.session(capability.as_ref().unwrap());
+                    let mut session = output.session_with_builder(capability.as_ref().unwrap());
                     session.give(element);
                     let n = 256 * crate::container::buffer::default_capacity::<I::Item>();
                     for element in iterator.by_ref().take(n - 1) {
@@ -55,3 +60,31 @@ impl<C: SizableContainer, I: IntoIterator+'static> ToStream<C> for I where C: Pu
         })
     }
 }
+
+/// Converts to a timely [StreamCore]. Equivalent to [`ToStreamBuilder`] but
+/// uses a [`CapacityContainerBuilder`].
+pub trait ToStream<C: Container> {
+    /// Converts to a timely [StreamCore].
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use timely::dataflow::operators::core::{ToStream, Capture};
+    /// use timely::dataflow::operators::core::capture::Extract;
+    ///
+    /// let (data1, data2) = timely::example(|scope| {
+    ///     let data1 = (0..3).to_stream(scope).container::<Vec<_>>().capture();
+    ///     let data2 = vec![0,1,2].to_stream(scope).container::<Vec<_>>().capture();
+    ///     (data1, data2)
+    /// });
+    ///
+    /// assert_eq!(data1.extract(), data2.extract());
+    /// ```
+    fn to_stream<S: Scope>(self, scope: &mut S) -> StreamCore<S, C>;
+}
+
+impl<C: 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/input.rs b/timely/src/dataflow/operators/input.rs
index d1e30280b..30d243ab4 100644
--- a/timely/src/dataflow/operators/input.rs
+++ b/timely/src/dataflow/operators/input.rs
@@ -1,6 +1,7 @@
 //! Create new `Streams` connected to external inputs.
 
 use crate::Data;
+use crate::container::CapacityContainerBuilder;
 use crate::dataflow::{Stream, ScopeParent, Scope};
 use crate::dataflow::operators::core::{Input as InputCore};
 
@@ -93,4 +94,4 @@ impl<G: Scope> Input for G where <G as ScopeParent>::Timestamp: TotalOrder {
 }
 
 /// A handle to an input `Stream`, used to introduce data to a timely dataflow computation.
-pub type Handle<T, D> = crate::dataflow::operators::core::input::Handle<T, Vec<D>>;
+pub type Handle<T, D> = crate::dataflow::operators::core::input::Handle<T, CapacityContainerBuilder<Vec<D>>>;