diff --git a/Cargo.lock b/Cargo.lock
index e5d1410829e7a..959e93fee731a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1973,7 +1973,7 @@ checksum = "0e25ea47919b1560c4e3b7fe0aaab9becf5b84a10325ddf7db0f0ba5e1026499"
 [[package]]
 name = "differential-dataflow"
 version = "0.12.0"
-source = "git+https://github.com/MaterializeInc/differential-dataflow.git#07ed40c3c44e6fc43efa5decdaa6e57ba9ce7fac"
+source = "git+https://github.com/MaterializeInc/differential-dataflow.git#a3f0f31d280b41254fd98fe51ef614622addd949"
 dependencies = [
  "abomonation",
  "abomonation_derive",
@@ -2029,7 +2029,7 @@ checksum = "923dea538cea0aa3025e8685b20d6ee21ef99c4f77e954a30febbaac5ec73a97"
 [[package]]
 name = "dogsdogsdogs"
 version = "0.1.0"
-source = "git+https://github.com/MaterializeInc/differential-dataflow.git#07ed40c3c44e6fc43efa5decdaa6e57ba9ce7fac"
+source = "git+https://github.com/MaterializeInc/differential-dataflow.git#a3f0f31d280b41254fd98fe51ef614622addd949"
 dependencies = [
  "abomonation",
  "abomonation_derive",
@@ -9346,7 +9346,7 @@ dependencies = [
 [[package]]
 name = "timely"
 version = "0.12.0"
-source = "git+https://github.com/MaterializeInc/timely-dataflow.git#dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+source = "git+https://github.com/MaterializeInc/timely-dataflow.git#89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 dependencies = [
  "abomonation",
  "abomonation_derive",
@@ -9363,12 +9363,12 @@ dependencies = [
 [[package]]
 name = "timely_bytes"
 version = "0.12.0"
-source = "git+https://github.com/MaterializeInc/timely-dataflow.git#dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+source = "git+https://github.com/MaterializeInc/timely-dataflow.git#89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 
 [[package]]
 name = "timely_communication"
 version = "0.12.0"
-source = "git+https://github.com/MaterializeInc/timely-dataflow.git#dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+source = "git+https://github.com/MaterializeInc/timely-dataflow.git#89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 dependencies = [
  "abomonation",
  "abomonation_derive",
@@ -9384,7 +9384,7 @@ dependencies = [
 [[package]]
 name = "timely_container"
 version = "0.12.0"
-source = "git+https://github.com/MaterializeInc/timely-dataflow.git#dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+source = "git+https://github.com/MaterializeInc/timely-dataflow.git#89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 dependencies = [
  "columnation",
  "flatcontainer",
@@ -9394,7 +9394,7 @@ dependencies = [
 [[package]]
 name = "timely_logging"
 version = "0.12.0"
-source = "git+https://github.com/MaterializeInc/timely-dataflow.git#dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+source = "git+https://github.com/MaterializeInc/timely-dataflow.git#89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 
 [[package]]
 name = "tiny-keccak"
diff --git a/misc/cargo-vet/audits.toml b/misc/cargo-vet/audits.toml
index 57aa47a393ff4..3757357f5f93b 100644
--- a/misc/cargo-vet/audits.toml
+++ b/misc/cargo-vet/audits.toml
@@ -144,7 +144,7 @@ notes = "Reads and writes files under a prefix controlled by the caller."
 [[audits.differential-dataflow]]
 who = "Moritz Hoffmann <mh@materialize.com>"
 criteria = "safe-to-deploy"
-version = "0.12.0@git:07ed40c3c44e6fc43efa5decdaa6e57ba9ce7fac"
+version = "0.12.0@git:a3f0f31d280b41254fd98fe51ef614622addd949"
 
 [[audits.domain]]
 who = "Matt Jibson <matt.jibson@gmail.com>"
@@ -480,22 +480,22 @@ version = "0.17.0"
 [[audits.timely]]
 who = "Moritz Hoffmann <mh@materialize.com>"
 criteria = "safe-to-deploy"
-version = "0.12.0@git:dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+version = "0.12.0@git:89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 
 [[audits.timely_bytes]]
 who = "Moritz Hoffmann <mh@materialize.com>"
 criteria = "safe-to-deploy"
-version = "0.12.0@git:dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+version = "0.12.0@git:89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 
 [[audits.timely_communication]]
 who = "Moritz Hoffmann <mh@materialize.com>"
 criteria = "safe-to-deploy"
-version = "0.12.0@git:dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+version = "0.12.0@git:89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 
 [[audits.timely_logging]]
 who = "Moritz Hoffmann <mh@materialize.com>"
 criteria = "safe-to-deploy"
-version = "0.12.0@git:dfb4721ac698a3a26dcecf7826d8a3f4fd23952f"
+version = "0.12.0@git:89bcb738c573f2b2b23ac3ec71d858c5afc54bb0"
 
 [[audits.tokio-postgres]]
 who = "Roshan Jobanputra <roshan@materialize.com>"
diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs
index 896612a954c49..232a2895e757f 100644
--- a/src/compute/src/compute_state.rs
+++ b/src/compute/src/compute_state.rs
@@ -1137,7 +1137,6 @@ impl IndexPeek {
         for<'a> Tr::Key<'a>: ToDatumIter,
         for<'a> Tr::Val<'a>: ToDatumIter,
         Tr::KeyOwned: Columnation + Data + FromDatumIter + ToDatumIter,
-        Tr::ValOwned: Columnation + Data + ToDatumIter,
     {
         let max_result_size = usize::cast_from(max_result_size);
         let count_byte_size = std::mem::size_of::<NonZeroUsize>();
diff --git a/src/compute/src/extensions/reduce.rs b/src/compute/src/extensions/reduce.rs
index 74a1adf1ca4dc..643df0651f405 100644
--- a/src/compute/src/extensions/reduce.rs
+++ b/src/compute/src/extensions/reduce.rs
@@ -28,15 +28,20 @@ where
     G::Timestamp: Lattice + Ord,
 {
     /// Applies `reduce` to arranged data, and returns an arrangement of output data.
-    fn mz_reduce_abelian<L, T2>(&self, name: &str, logic: L) -> Arranged<G, TraceAgent<T2>>
+    fn mz_reduce_abelian<L, V, F, T2>(
+        &self,
+        name: &str,
+        from: F,
+        logic: L,
+    ) -> Arranged<G, TraceAgent<T2>>
     where
         T2: for<'a> Trace<Key<'a> = T1::Key<'a>, Time = G::Timestamp> + 'static,
-        T2::ValOwned: Data,
+        V: Data,
+        F: Fn(T2::Val<'_>) -> V + 'static,
         T2::Diff: Abelian,
         T2::Batch: Batch,
-        T2::Builder: Builder<Input = ((T1::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
-        L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(T2::ValOwned, T2::Diff)>)
-            + 'static,
+        T2::Builder: Builder<Input = ((T1::KeyOwned, V), T2::Time, T2::Diff)>,
+        L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(V, T2::Diff)>) + 'static,
         Arranged<G, TraceAgent<T2>>: ArrangementSize;
 }
 
@@ -48,20 +53,26 @@ where
     T1::Diff: Semigroup,
 {
     /// Applies `reduce` to arranged data, and returns an arrangement of output data.
-    fn mz_reduce_abelian<L, T2>(&self, name: &str, logic: L) -> Arranged<G, TraceAgent<T2>>
+    fn mz_reduce_abelian<L, V, F, T2>(
+        &self,
+        name: &str,
+        from: F,
+        logic: L,
+    ) -> Arranged<G, TraceAgent<T2>>
     where
         T2: for<'a> Trace<Key<'a> = T1::Key<'a>, Time = G::Timestamp> + 'static,
-        T2::ValOwned: Data,
+        V: Data,
+        F: Fn(T2::Val<'_>) -> V + 'static,
         T2::Diff: Abelian,
         T2::Batch: Batch,
-        T2::Builder: Builder<Input = ((T1::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
-        L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(T2::ValOwned, T2::Diff)>)
-            + 'static,
+        T2::Builder: Builder<Input = ((T1::KeyOwned, V), T2::Time, T2::Diff)>,
+        L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(V, T2::Diff)>) + 'static,
         Arranged<G, TraceAgent<T2>>: ArrangementSize,
     {
         // Allow access to `reduce_abelian` since we're within Mz's wrapper and force arrangement size logging.
         #[allow(clippy::disallowed_methods)]
-        Arranged::<_, _>::reduce_abelian::<_, T2>(self, name, logic).log_arrangement_size()
+        Arranged::<_, _>::reduce_abelian::<_, _, _, T2>(self, name, from, logic)
+            .log_arrangement_size()
     }
 }
 
@@ -76,10 +87,12 @@ where
     /// are produced as a result. The method is useful for reductions that need to present different
     /// output views on the same input data. An example is producing an error-free reduction output
     /// along with a separate error output indicating when the error-free output is valid.
-    fn reduce_pair<L1, T1, L2, T2>(
+    fn reduce_pair<L1, V1, F1, T1, L2, V2, F2, T2>(
         &self,
         name1: &str,
         name2: &str,
+        from1: F1,
+        from2: F2,
         logic1: L1,
         logic2: L2,
     ) -> (Arranged<G, TraceAgent<T1>>, Arranged<G, TraceAgent<T2>>)
@@ -87,21 +100,21 @@ where
         T1: Trace
             + for<'a> TraceReader<Key<'a> = Tr::Key<'a>, KeyOwned = Tr::KeyOwned, Time = G::Timestamp>
             + 'static,
-        T1::ValOwned: Data,
         T1::Diff: Abelian,
         T1::Batch: Batch,
-        T1::Builder: Builder<Input = ((T1::KeyOwned, T1::ValOwned), T1::Time, T1::Diff)>,
-        L1: FnMut(Tr::Key<'_>, &[(Tr::Val<'_>, Tr::Diff)], &mut Vec<(T1::ValOwned, T1::Diff)>)
-            + 'static,
+        T1::Builder: Builder<Input = ((T1::KeyOwned, V1), T1::Time, T1::Diff)>,
+        L1: FnMut(Tr::Key<'_>, &[(Tr::Val<'_>, Tr::Diff)], &mut Vec<(V1, T1::Diff)>) + 'static,
+        V1: Data,
+        F1: Fn(T1::Val<'_>) -> V1 + 'static,
         T2: Trace
             + for<'a> TraceReader<Key<'a> = Tr::Key<'a>, KeyOwned = Tr::KeyOwned, Time = G::Timestamp>
             + 'static,
-        T2::ValOwned: Data,
         T2::Diff: Abelian,
         T2::Batch: Batch,
-        T2::Builder: Builder<Input = ((T2::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
-        L2: FnMut(Tr::Key<'_>, &[(Tr::Val<'_>, Tr::Diff)], &mut Vec<(T2::ValOwned, T2::Diff)>)
-            + 'static,
+        T2::Builder: Builder<Input = ((T2::KeyOwned, V2), T2::Time, T2::Diff)>,
+        L2: FnMut(Tr::Key<'_>, &[(Tr::Val<'_>, Tr::Diff)], &mut Vec<(V2, T2::Diff)>) + 'static,
+        V2: Data,
+        F2: Fn(T2::Val<'_>) -> V2 + 'static,
         Arranged<G, TraceAgent<T1>>: ArrangementSize,
         Arranged<G, TraceAgent<T2>>: ArrangementSize;
 }
@@ -112,10 +125,12 @@ where
     Tr: TraceReader<Time = G::Timestamp> + Clone + 'static,
     Tr::Diff: Semigroup,
 {
-    fn reduce_pair<L1, T1, L2, T2>(
+    fn reduce_pair<L1, V1, F1, T1, L2, V2, F2, T2>(
         &self,
         name1: &str,
         name2: &str,
+        from1: F1,
+        from2: F2,
         logic1: L1,
         logic2: L2,
     ) -> (Arranged<G, TraceAgent<T1>>, Arranged<G, TraceAgent<T2>>)
@@ -123,26 +138,26 @@ where
         T1: Trace
             + for<'a> TraceReader<Key<'a> = Tr::Key<'a>, KeyOwned = Tr::KeyOwned, Time = G::Timestamp>
             + 'static,
-        T1::ValOwned: Data,
         T1::Diff: Abelian,
         T1::Batch: Batch,
-        T1::Builder: Builder<Input = ((T1::KeyOwned, T1::ValOwned), T1::Time, T1::Diff)>,
-        L1: FnMut(Tr::Key<'_>, &[(Tr::Val<'_>, Tr::Diff)], &mut Vec<(T1::ValOwned, T1::Diff)>)
-            + 'static,
+        T1::Builder: Builder<Input = ((T1::KeyOwned, V1), T1::Time, T1::Diff)>,
+        L1: FnMut(Tr::Key<'_>, &[(Tr::Val<'_>, Tr::Diff)], &mut Vec<(V1, T1::Diff)>) + 'static,
+        V1: Data,
+        F1: Fn(T1::Val<'_>) -> V1 + 'static,
         T2: Trace
             + for<'a> TraceReader<Key<'a> = Tr::Key<'a>, KeyOwned = Tr::KeyOwned, Time = G::Timestamp>
             + 'static,
-        T2::ValOwned: Data,
         T2::Diff: Abelian,
         T2::Batch: Batch,
-        T2::Builder: Builder<Input = ((T2::KeyOwned, T2::ValOwned), T2::Time, T2::Diff)>,
-        L2: FnMut(Tr::Key<'_>, &[(Tr::Val<'_>, Tr::Diff)], &mut Vec<(T2::ValOwned, T2::Diff)>)
-            + 'static,
+        T2::Builder: Builder<Input = ((T2::KeyOwned, V2), T2::Time, T2::Diff)>,
+        L2: FnMut(Tr::Key<'_>, &[(Tr::Val<'_>, Tr::Diff)], &mut Vec<(V2, T2::Diff)>) + 'static,
+        V2: Data,
+        F2: Fn(T2::Val<'_>) -> V2 + 'static,
         Arranged<G, TraceAgent<T1>>: ArrangementSize,
         Arranged<G, TraceAgent<T2>>: ArrangementSize,
     {
-        let arranged1 = self.mz_reduce_abelian::<L1, T1>(name1, logic1);
-        let arranged2 = self.mz_reduce_abelian::<L2, T2>(name2, logic2);
+        let arranged1 = self.mz_reduce_abelian::<L1, _, _, T1>(name1, from1, logic1);
+        let arranged2 = self.mz_reduce_abelian::<L2, _, _, T2>(name2, from2, logic2);
         (arranged1, arranged2)
     }
 }
diff --git a/src/compute/src/logging/compute.rs b/src/compute/src/logging/compute.rs
index 71ecda4ac22e8..7160c3eeade5d 100644
--- a/src/compute/src/logging/compute.rs
+++ b/src/compute/src/logging/compute.rs
@@ -22,6 +22,7 @@ use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, GlobalId, Timestamp};
 use mz_timely_util::replay::MzReplay;
 use timely::communication::Allocate;
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::channels::pushers::buffer::Session;
 use timely::dataflow::channels::pushers::{Counter, Tee};
@@ -473,7 +474,8 @@ struct ArrangementSizeState {
 
 type Update<D> = (D, Timestamp, Diff);
 type Pusher<D> = Counter<Timestamp, Vec<Update<D>>, Tee<Timestamp, Vec<Update<D>>>>;
-type OutputSession<'a, D> = Session<'a, Timestamp, Vec<Update<D>>, Pusher<D>>;
+type OutputSession<'a, D> =
+    Session<'a, Timestamp, CapacityContainerBuilder<Vec<Update<D>>>, Pusher<D>>;
 
 /// Bundled output sessions used by the demux operator.
 struct DemuxOutput<'a> {
@@ -978,7 +980,7 @@ where
                         let diff = buffer.iter().map(|(_d, _t, r)| r).sum();
                         logger.log(ComputeEvent::ErrorCount { export_id, diff });
 
-                        output.session(&cap).give_vec(&mut buffer);
+                        output.session(&cap).give_container(&mut buffer);
                     });
                 }
             })
@@ -1001,7 +1003,7 @@ where
                     let diff = buffer.iter().map(sum_batch_diffs).sum();
                     logger.log(ComputeEvent::ErrorCount { export_id, diff });
 
-                    output.session(&cap).give_vec(&mut buffer);
+                    output.session(&cap).give_container(&mut buffer);
                 });
             }
         })
diff --git a/src/compute/src/logging/differential.rs b/src/compute/src/logging/differential.rs
index 7d6728213e148..1f0daf0c50535 100644
--- a/src/compute/src/logging/differential.rs
+++ b/src/compute/src/logging/differential.rs
@@ -15,18 +15,19 @@ use std::rc::Rc;
 use std::time::Duration;
 
 use differential_dataflow::collection::AsCollection;
+use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::logging::{
     BatchEvent, BatcherEvent, DifferentialEvent, DropEvent, MergeEvent, TraceShare,
 };
 use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, Timestamp};
-use mz_timely_util::buffer::ConsolidateBuffer;
 use mz_timely_util::replay::MzReplay;
 use timely::communication::Allocate;
 use timely::dataflow::channels::pact::Pipeline;
-use timely::dataflow::channels::pushers::Tee;
+use timely::dataflow::channels::pushers::buffer::Session;
+use timely::dataflow::channels::pushers::{Counter, Tee};
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
-use timely::dataflow::operators::{Filter, InputCapability};
+use timely::dataflow::operators::Filter;
 use timely::dataflow::Stream;
 
 use crate::extensions::arrange::MzArrange;
@@ -91,17 +92,17 @@ pub(super) fn construct<A: Allocate>(
                 let mut batcher_capacity = batcher_capacity_out.activate();
                 let mut batcher_allocations = batcher_allocations_out.activate();
 
-                let mut output_buffers = DemuxOutput {
-                    batches: ConsolidateBuffer::new(&mut batches, 0),
-                    records: ConsolidateBuffer::new(&mut records, 1),
-                    sharing: ConsolidateBuffer::new(&mut sharing, 2),
-                    batcher_records: ConsolidateBuffer::new(&mut batcher_records, 3),
-                    batcher_size: ConsolidateBuffer::new(&mut batcher_size, 4),
-                    batcher_capacity: ConsolidateBuffer::new(&mut batcher_capacity, 5),
-                    batcher_allocations: ConsolidateBuffer::new(&mut batcher_allocations, 6),
-                };
-
                 input.for_each(|cap, data| {
+                    let mut output_buffers = DemuxOutput {
+                        batches: batches.session_with_builder(&cap),
+                        records: records.session_with_builder(&cap),
+                        sharing: sharing.session_with_builder(&cap),
+                        batcher_records: batcher_records.session_with_builder(&cap),
+                        batcher_size: batcher_size.session_with_builder(&cap),
+                        batcher_capacity: batcher_capacity.session_with_builder(&cap),
+                        batcher_allocations: batcher_allocations.session_with_builder(&cap),
+                    };
+
                     data.swap(&mut demux_buffer);
 
                     for (time, logger_id, event) in demux_buffer.drain(..) {
@@ -115,7 +116,6 @@ pub(super) fn construct<A: Allocate>(
                             output: &mut output_buffers,
                             logging_interval_ms,
                             time,
-                            cap: &cap,
                             shared_state: &mut shared_state.borrow_mut(),
                         }
                         .handle(event);
@@ -187,18 +187,20 @@ pub(super) fn construct<A: Allocate>(
     })
 }
 
-type Pusher<D> = Tee<Timestamp, Vec<(D, Timestamp, Diff)>>;
-type OutputBuffer<'a, 'b, D> = ConsolidateBuffer<'a, 'b, Timestamp, D, Diff, Pusher<D>>;
+type Pusher<D> =
+    Counter<Timestamp, Vec<(D, Timestamp, Diff)>, Tee<Timestamp, Vec<(D, Timestamp, Diff)>>>;
+type OutputSession<'a, D> =
+    Session<'a, Timestamp, ConsolidatingContainerBuilder<Vec<(D, Timestamp, Diff)>>, Pusher<D>>;
 
 /// Bundled output buffers used by the demux operator.
-struct DemuxOutput<'a, 'b> {
-    batches: OutputBuffer<'a, 'b, (usize, ())>,
-    records: OutputBuffer<'a, 'b, (usize, ())>,
-    sharing: OutputBuffer<'a, 'b, (usize, ())>,
-    batcher_records: OutputBuffer<'a, 'b, (usize, ())>,
-    batcher_size: OutputBuffer<'a, 'b, (usize, ())>,
-    batcher_capacity: OutputBuffer<'a, 'b, (usize, ())>,
-    batcher_allocations: OutputBuffer<'a, 'b, (usize, ())>,
+struct DemuxOutput<'a> {
+    batches: OutputSession<'a, (usize, ())>,
+    records: OutputSession<'a, (usize, ())>,
+    sharing: OutputSession<'a, (usize, ())>,
+    batcher_records: OutputSession<'a, (usize, ())>,
+    batcher_size: OutputSession<'a, (usize, ())>,
+    batcher_capacity: OutputSession<'a, (usize, ())>,
+    batcher_allocations: OutputSession<'a, (usize, ())>,
 }
 
 /// State maintained by the demux operator.
@@ -209,22 +211,20 @@ struct DemuxState {
 }
 
 /// Event handler of the demux operator.
-struct DemuxHandler<'a, 'b, 'c> {
+struct DemuxHandler<'a, 'b> {
     /// State kept by the demux operator
     state: &'a mut DemuxState,
     /// Demux output buffers.
-    output: &'a mut DemuxOutput<'b, 'c>,
+    output: &'a mut DemuxOutput<'b>,
     /// The logging interval specifying the time granularity for the updates.
     logging_interval_ms: u128,
     /// The current event time.
     time: Duration,
-    /// A capability usable for emitting outputs.
-    cap: &'a InputCapability<Timestamp>,
     /// State shared across log receivers.
     shared_state: &'a mut SharedLoggingState,
 }
 
-impl DemuxHandler<'_, '_, '_> {
+impl DemuxHandler<'_, '_> {
     /// Return the timestamp associated with the current event, based on the event time and the
     /// logging interval.
     fn ts(&self) -> Timestamp {
@@ -251,10 +251,10 @@ impl DemuxHandler<'_, '_, '_> {
     fn handle_batch(&mut self, event: BatchEvent) {
         let ts = self.ts();
         let op = event.operator;
-        self.output.batches.give(self.cap, ((op, ()), ts, 1));
+        self.output.batches.give(((op, ()), ts, 1));
 
         let diff = Diff::try_from(event.length).expect("must fit");
-        self.output.records.give(self.cap, ((op, ()), ts, diff));
+        self.output.records.give(((op, ()), ts, diff));
         self.notify_arrangement_size(op);
     }
 
@@ -263,12 +263,12 @@ impl DemuxHandler<'_, '_, '_> {
 
         let ts = self.ts();
         let op = event.operator;
-        self.output.batches.give(self.cap, ((op, ()), ts, -1));
+        self.output.batches.give(((op, ()), ts, -1));
 
         let diff = Diff::try_from(done).expect("must fit")
             - Diff::try_from(event.length1 + event.length2).expect("must fit");
         if diff != 0 {
-            self.output.records.give(self.cap, ((op, ()), ts, diff));
+            self.output.records.give(((op, ()), ts, diff));
         }
         self.notify_arrangement_size(op);
     }
@@ -276,11 +276,11 @@ impl DemuxHandler<'_, '_, '_> {
     fn handle_drop(&mut self, event: DropEvent) {
         let ts = self.ts();
         let op = event.operator;
-        self.output.batches.give(self.cap, ((op, ()), ts, -1));
+        self.output.batches.give(((op, ()), ts, -1));
 
         let diff = -Diff::try_from(event.length).expect("must fit");
         if diff != 0 {
-            self.output.records.give(self.cap, ((op, ()), ts, diff));
+            self.output.records.give(((op, ()), ts, diff));
         }
         self.notify_arrangement_size(op);
     }
@@ -290,7 +290,7 @@ impl DemuxHandler<'_, '_, '_> {
         let op = event.operator;
         let diff = Diff::cast_from(event.diff);
         debug_assert_ne!(diff, 0);
-        self.output.sharing.give(self.cap, ((op, ()), ts, diff));
+        self.output.sharing.give(((op, ()), ts, diff));
 
         if let Some(logger) = &mut self.shared_state.compute_logger {
             let sharing = self.state.sharing.entry(op).or_default();
@@ -313,16 +313,14 @@ impl DemuxHandler<'_, '_, '_> {
         let allocations_diff = Diff::cast_from(event.allocations_diff);
         self.output
             .batcher_records
-            .give(self.cap, ((op, ()), ts, records_diff));
-        self.output
-            .batcher_size
-            .give(self.cap, ((op, ()), ts, size_diff));
+            .give(((op, ()), ts, records_diff));
+        self.output.batcher_size.give(((op, ()), ts, size_diff));
         self.output
             .batcher_capacity
-            .give(self.cap, ((op, ()), ts, capacity_diff));
+            .give(((op, ()), ts, capacity_diff));
         self.output
             .batcher_allocations
-            .give(self.cap, ((op, ()), ts, allocations_diff));
+            .give(((op, ()), ts, allocations_diff));
     }
 
     fn notify_arrangement_size(&self, operator: usize) {
diff --git a/src/compute/src/logging/reachability.rs b/src/compute/src/logging/reachability.rs
index 0b98da7f769c8..bd0139f855a58 100644
--- a/src/compute/src/logging/reachability.rs
+++ b/src/compute/src/logging/reachability.rs
@@ -13,13 +13,13 @@ use std::collections::BTreeMap;
 use std::convert::TryInto;
 use std::rc::Rc;
 
+use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::AsCollection;
 use mz_compute_client::logging::LoggingConfig;
 use mz_expr::{permutation_for_arrangement, MirScalarExpr};
 use mz_ore::cast::CastFrom;
 use mz_ore::iter::IteratorExt;
 use mz_repr::{Datum, Diff, RowArena, SharedRow, Timestamp};
-use mz_timely_util::buffer::ConsolidateBuffer;
 use mz_timely_util::replay::MzReplay;
 use timely::communication::Allocate;
 use timely::dataflow::operators::Filter;
@@ -76,15 +76,15 @@ pub(super) fn construct<A: Allocate>(
         use timely::dataflow::channels::pact::Pipeline;
         let mut input = flatten.new_input(&logs, Pipeline);
 
-        let (mut updates_out, updates) = flatten.new_output();
+        let (mut updates_out, updates) = flatten.new_output::<ConsolidatingContainerBuilder<_>>();
 
         let mut buffer = Vec::new();
         flatten.build(move |_capability| {
             move |_frontiers| {
                 let mut updates = updates_out.activate();
-                let mut updates_session = ConsolidateBuffer::new(&mut updates, 0);
 
                 input.for_each(|cap, data| {
+                    let mut updates_session = updates.session_with_builder(&cap);
                     data.swap(&mut buffer);
 
                     for (time, _worker, (addr, massaged)) in buffer.drain(..) {
@@ -93,7 +93,7 @@ pub(super) fn construct<A: Allocate>(
                             .expect("must fit");
                         for (source, port, update_type, ts, diff) in massaged {
                             let datum = (update_type, addr.clone(), source, port, ts);
-                            updates_session.give(&cap, ((datum, ()), time_ms, diff));
+                            updates_session.give(((datum, ()), time_ms, diff));
                         }
                     }
                 });
diff --git a/src/compute/src/logging/timely.rs b/src/compute/src/logging/timely.rs
index fb9a2556b1bc6..d27a8103164d6 100644
--- a/src/compute/src/logging/timely.rs
+++ b/src/compute/src/logging/timely.rs
@@ -15,18 +15,19 @@ use std::rc::Rc;
 use std::time::Duration;
 
 use differential_dataflow::collection::AsCollection;
+use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use mz_compute_client::logging::LoggingConfig;
 use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, Timestamp};
-use mz_timely_util::buffer::ConsolidateBuffer;
 use mz_timely_util::replay::MzReplay;
 use serde::{Deserialize, Serialize};
 use timely::communication::Allocate;
 use timely::container::columnation::{Columnation, CopyRegion};
 use timely::dataflow::channels::pact::Pipeline;
-use timely::dataflow::channels::pushers::Tee;
+use timely::dataflow::channels::pushers::buffer::Session;
+use timely::dataflow::channels::pushers::{Counter, Tee};
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
-use timely::dataflow::operators::{Filter, InputCapability};
+use timely::dataflow::operators::Filter;
 use timely::logging::{
     ChannelsEvent, MessagesEvent, OperatesEvent, ParkEvent, ScheduleEvent, ShutdownEvent,
     TimelyEvent,
@@ -100,20 +101,20 @@ pub(super) fn construct<A: Allocate>(
                 let mut schedules_duration = schedules_duration_out.activate();
                 let mut schedules_histogram = schedules_histogram_out.activate();
 
-                let mut output_buffers = DemuxOutput {
-                    operates: ConsolidateBuffer::new(&mut operates, 0),
-                    channels: ConsolidateBuffer::new(&mut channels, 1),
-                    addresses: ConsolidateBuffer::new(&mut addresses, 2),
-                    parks: ConsolidateBuffer::new(&mut parks, 3),
-                    messages_sent: ConsolidateBuffer::new(&mut messages_sent, 4),
-                    messages_received: ConsolidateBuffer::new(&mut messages_received, 5),
-                    schedules_duration: ConsolidateBuffer::new(&mut schedules_duration, 6),
-                    schedules_histogram: ConsolidateBuffer::new(&mut schedules_histogram, 7),
-                    batches_sent: ConsolidateBuffer::new(&mut batches_sent, 8),
-                    batches_received: ConsolidateBuffer::new(&mut batches_received, 9),
-                };
-
                 input.for_each(|cap, data| {
+                    let mut output_buffers = DemuxOutput {
+                        operates: operates.session_with_builder(&cap),
+                        channels: channels.session_with_builder(&cap),
+                        addresses: addresses.session_with_builder(&cap),
+                        parks: parks.session_with_builder(&cap),
+                        messages_sent: messages_sent.session_with_builder(&cap),
+                        messages_received: messages_received.session_with_builder(&cap),
+                        schedules_duration: schedules_duration.session_with_builder(&cap),
+                        schedules_histogram: schedules_histogram.session_with_builder(&cap),
+                        batches_sent: batches_sent.session_with_builder(&cap),
+                        batches_received: batches_received.session_with_builder(&cap),
+                    };
+
                     data.swap(&mut demux_buffer);
 
                     for (time, logger_id, event) in demux_buffer.drain(..) {
@@ -135,7 +136,6 @@ pub(super) fn construct<A: Allocate>(
                             logging_interval_ms,
                             peers,
                             time,
-                            cap: &cap,
                         }
                         .handle(event);
                     }
@@ -352,25 +352,27 @@ struct MessageCount {
     records: i64,
 }
 
-type Pusher<D> = Tee<Timestamp, Vec<(D, Timestamp, Diff)>>;
-type OutputBuffer<'a, 'b, D> = ConsolidateBuffer<'a, 'b, Timestamp, D, Diff, Pusher<D>>;
+type Pusher<D> =
+    Counter<Timestamp, Vec<(D, Timestamp, Diff)>, Tee<Timestamp, Vec<(D, Timestamp, Diff)>>>;
+type OutputSession<'a, D> =
+    Session<'a, Timestamp, ConsolidatingContainerBuilder<Vec<(D, Timestamp, Diff)>>, Pusher<D>>;
 
 /// Bundled output buffers used by the demux operator.
 //
 // We use tuples rather than dedicated `*Datum` structs for `operates` and `addresses` to avoid
 // having to manually implement `Columnation`. If `Columnation` could be `#[derive]`ed, that
 // wouldn't be an issue.
-struct DemuxOutput<'a, 'b> {
-    operates: OutputBuffer<'a, 'b, (usize, String)>,
-    channels: OutputBuffer<'a, 'b, (ChannelDatum, ())>,
-    addresses: OutputBuffer<'a, 'b, (usize, Vec<usize>)>,
-    parks: OutputBuffer<'a, 'b, (ParkDatum, ())>,
-    batches_sent: OutputBuffer<'a, 'b, (MessageDatum, ())>,
-    batches_received: OutputBuffer<'a, 'b, (MessageDatum, ())>,
-    messages_sent: OutputBuffer<'a, 'b, (MessageDatum, ())>,
-    messages_received: OutputBuffer<'a, 'b, (MessageDatum, ())>,
-    schedules_duration: OutputBuffer<'a, 'b, (usize, ())>,
-    schedules_histogram: OutputBuffer<'a, 'b, (ScheduleHistogramDatum, ())>,
+struct DemuxOutput<'a> {
+    operates: OutputSession<'a, (usize, String)>,
+    channels: OutputSession<'a, (ChannelDatum, ())>,
+    addresses: OutputSession<'a, (usize, Vec<usize>)>,
+    parks: OutputSession<'a, (ParkDatum, ())>,
+    batches_sent: OutputSession<'a, (MessageDatum, ())>,
+    batches_received: OutputSession<'a, (MessageDatum, ())>,
+    messages_sent: OutputSession<'a, (MessageDatum, ())>,
+    messages_received: OutputSession<'a, (MessageDatum, ())>,
+    schedules_duration: OutputSession<'a, (usize, ())>,
+    schedules_histogram: OutputSession<'a, (ScheduleHistogramDatum, ())>,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
@@ -415,24 +417,22 @@ impl Columnation for ScheduleHistogramDatum {
 }
 
 /// Event handler of the demux operator.
-struct DemuxHandler<'a, 'b, 'c> {
+struct DemuxHandler<'a, 'b> {
     /// State kept by the demux operator.
     state: &'a mut DemuxState,
     /// State shared across log receivers.
     shared_state: &'a mut SharedLoggingState,
     /// Demux output buffers.
-    output: &'a mut DemuxOutput<'b, 'c>,
+    output: &'a mut DemuxOutput<'b>,
     /// The logging interval specifying the time granularity for the updates.
     logging_interval_ms: u128,
     /// The number of timely workers.
     peers: usize,
     /// The current event time.
     time: Duration,
-    /// A capability usable for emitting outputs.
-    cap: &'a InputCapability<Timestamp>,
 }
 
-impl DemuxHandler<'_, '_, '_> {
+impl DemuxHandler<'_, '_> {
     /// Return the timestamp associated with the current event, based on the event time and the
     /// logging interval.
     fn ts(&self) -> Timestamp {
@@ -460,10 +460,10 @@ impl DemuxHandler<'_, '_, '_> {
     fn handle_operates(&mut self, event: OperatesEvent) {
         let ts = self.ts();
         let datum = (event.id, event.name.clone());
-        self.output.operates.give(self.cap, (datum, ts, 1));
+        self.output.operates.give((datum, ts, 1));
 
         let datum = (event.id, event.addr.clone());
-        self.output.addresses.give(self.cap, (datum, ts, 1));
+        self.output.addresses.give((datum, ts, 1));
 
         self.state.operators.insert(event.id, event);
     }
@@ -475,10 +475,10 @@ impl DemuxHandler<'_, '_, '_> {
             source: event.source,
             target: event.target,
         };
-        self.output.channels.give(self.cap, ((datum, ()), ts, 1));
+        self.output.channels.give(((datum, ()), ts, 1));
 
         let datum = (event.id, event.scope_addr.clone());
-        self.output.addresses.give(self.cap, (datum, ts, 1));
+        self.output.addresses.give((datum, ts, 1));
 
         let dataflow_index = event.scope_addr[0];
         self.state
@@ -502,7 +502,7 @@ impl DemuxHandler<'_, '_, '_> {
         // Retract operator information.
         let ts = self.ts();
         let datum = (operator.id, operator.name);
-        self.output.operates.give(self.cap, (datum, ts, -1));
+        self.output.operates.give((datum, ts, -1));
 
         // Retract schedules information for the operator
         if let Some(schedules) = self.state.schedules_data.remove(&event.id) {
@@ -513,7 +513,7 @@ impl DemuxHandler<'_, '_, '_> {
             {
                 self.output
                     .schedules_duration
-                    .give(self.cap, ((event.id, ()), ts, -elapsed_ns));
+                    .give(((event.id, ()), ts, -elapsed_ns));
 
                 let datum = ScheduleHistogramDatum {
                     operator: event.id,
@@ -522,7 +522,7 @@ impl DemuxHandler<'_, '_, '_> {
                 let diff = Diff::cast_from(-count);
                 self.output
                     .schedules_histogram
-                    .give(self.cap, ((datum, ()), ts, diff));
+                    .give(((datum, ()), ts, diff));
             }
         }
 
@@ -532,7 +532,7 @@ impl DemuxHandler<'_, '_, '_> {
         }
 
         let datum = (operator.id, operator.addr);
-        self.output.addresses.give(self.cap, (datum, ts, -1));
+        self.output.addresses.give((datum, ts, -1));
     }
 
     fn handle_dataflow_shutdown(&mut self, dataflow_index: usize) {
@@ -554,10 +554,10 @@ impl DemuxHandler<'_, '_, '_> {
                 source: channel.source,
                 target: channel.target,
             };
-            self.output.channels.give(self.cap, ((datum, ()), ts, -1));
+            self.output.channels.give(((datum, ()), ts, -1));
 
             let datum = (channel.id, channel.scope_addr);
-            self.output.addresses.give(self.cap, (datum, ts, -1));
+            self.output.addresses.give((datum, ts, -1));
 
             // Retract messages logged for this channel.
             if let Some(sent) = self.state.messages_sent.remove(&channel.id) {
@@ -568,10 +568,10 @@ impl DemuxHandler<'_, '_, '_> {
                     };
                     self.output
                         .messages_sent
-                        .give(self.cap, ((datum, ()), ts, -count.records));
+                        .give(((datum, ()), ts, -count.records));
                     self.output
                         .batches_sent
-                        .give(self.cap, ((datum, ()), ts, -count.batches));
+                        .give(((datum, ()), ts, -count.batches));
                 }
             }
             if let Some(received) = self.state.messages_received.remove(&channel.id) {
@@ -582,10 +582,10 @@ impl DemuxHandler<'_, '_, '_> {
                     };
                     self.output
                         .messages_received
-                        .give(self.cap, ((datum, ()), ts, -count.records));
+                        .give(((datum, ()), ts, -count.records));
                     self.output
                         .batches_received
-                        .give(self.cap, ((datum, ()), ts, -count.batches));
+                        .give(((datum, ()), ts, -count.batches));
                 }
             }
         }
@@ -618,7 +618,7 @@ impl DemuxHandler<'_, '_, '_> {
                     duration_pow,
                     requested_pow,
                 };
-                self.output.parks.give(self.cap, ((datum, ()), ts, 1));
+                self.output.parks.give(((datum, ()), ts, 1));
             }
         }
     }
@@ -632,12 +632,8 @@ impl DemuxHandler<'_, '_, '_> {
                 channel: event.channel,
                 worker: event.target,
             };
-            self.output
-                .messages_sent
-                .give(self.cap, ((datum, ()), ts, count));
-            self.output
-                .batches_sent
-                .give(self.cap, ((datum, ()), ts, 1));
+            self.output.messages_sent.give(((datum, ()), ts, count));
+            self.output.batches_sent.give(((datum, ()), ts, 1));
 
             let sent_counts = self
                 .state
@@ -651,12 +647,8 @@ impl DemuxHandler<'_, '_, '_> {
                 channel: event.channel,
                 worker: event.source,
             };
-            self.output
-                .messages_received
-                .give(self.cap, ((datum, ()), ts, count));
-            self.output
-                .batches_received
-                .give(self.cap, ((datum, ()), ts, 1));
+            self.output.messages_received.give(((datum, ()), ts, count));
+            self.output.batches_received.give(((datum, ()), ts, 1));
 
             let received_counts = self
                 .state
@@ -690,15 +682,13 @@ impl DemuxHandler<'_, '_, '_> {
                 let datum = event.id;
                 self.output
                     .schedules_duration
-                    .give(self.cap, ((datum, ()), ts, elapsed_diff));
+                    .give(((datum, ()), ts, elapsed_diff));
 
                 let datum = ScheduleHistogramDatum {
                     operator: event.id,
                     duration_pow: elapsed_pow,
                 };
-                self.output
-                    .schedules_histogram
-                    .give(self.cap, ((datum, ()), ts, 1));
+                self.output.schedules_histogram.give(((datum, ()), ts, 1));
 
                 // Record count and elapsed time for later retraction.
                 let index = usize::cast_from(elapsed_pow.trailing_zeros());
diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs
index 54fb848f4a5b2..ee6bc5c4a1062 100644
--- a/src/compute/src/render.rs
+++ b/src/compute/src/render.rs
@@ -112,9 +112,9 @@ use std::task::Poll;
 
 use differential_dataflow::dynamic::pointstamp::PointStamp;
 use differential_dataflow::lattice::Lattice;
-use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
-use differential_dataflow::trace::{Batch, Batcher, Trace, TraceReader};
-use differential_dataflow::{AsCollection, Collection, Data, ExchangeData, Hashable};
+use differential_dataflow::operators::arrange::Arranged;
+use differential_dataflow::trace::TraceReader;
+use differential_dataflow::{AsCollection, Collection, Data};
 use futures::channel::oneshot;
 use futures::FutureExt;
 use mz_compute_types::dataflows::{BuildDesc, DataflowDescription, IndexDesc};
@@ -122,7 +122,7 @@ use mz_compute_types::plan::flat_plan::{FlatPlan, FlatPlanNode};
 use mz_compute_types::plan::LirId;
 use mz_expr::{EvalError, Id};
 use mz_persist_client::operators::shard_source::SnapshotMode;
-use mz_repr::{Datum, Diff, GlobalId, Row, SharedRow};
+use mz_repr::{Datum, GlobalId, Row, SharedRow};
 use mz_storage_operators::persist_source;
 use mz_storage_types::controller::CollectionMetadata;
 use mz_storage_types::errors::DataflowError;
@@ -143,7 +143,7 @@ use timely::PartialOrder;
 
 use crate::arrangement::manager::TraceBundle;
 use crate::compute_state::ComputeState;
-use crate::extensions::arrange::{ArrangementSize, KeyCollection, MzArrange};
+use crate::extensions::arrange::{KeyCollection, MzArrange};
 use crate::extensions::reduce::MzReduce;
 use crate::logging::compute::LogDataflowErrors;
 use crate::render::context::{
@@ -603,33 +603,15 @@ where
     ) -> MzArrangement<G> {
         match oks {
             MzArrangement::RowRow(inner) => {
-                let oks = self.rearrange_iterative(inner, name);
+                use differential_dataflow::trace::cursor::MyTrait;
+                let oks = inner
+                    .as_collection(|k, v| (k.into_owned(), v.into_owned()))
+                    .leave()
+                    .mz_arrange(name);
                 MzArrangement::RowRow(oks)
             }
         }
     }
-
-    /// Rearranges an arrangement coming from an iterative scope into an arrangement
-    /// in the outer timestamp scope.
-    fn rearrange_iterative<Tr1, Tr2>(
-        &self,
-        oks: Arranged<Child<'g, G, T>, TraceAgent<Tr1>>,
-        name: &str,
-    ) -> Arranged<G, TraceAgent<Tr2>>
-    where
-        Tr1: TraceReader<Time = T, Diff = Diff>,
-        Tr1::KeyOwned: Columnation + ExchangeData + Hashable,
-        Tr1::ValOwned: Columnation + ExchangeData,
-        Tr2: Trace + TraceReader<Time = G::Timestamp, Diff = Diff> + 'static,
-        Tr2::Batch: Batch,
-        Tr2::Batcher: Batcher<Input = Vec<((Tr1::KeyOwned, Tr1::ValOwned), G::Timestamp, Diff)>>,
-        Arranged<G, TraceAgent<Tr2>>: ArrangementSize,
-    {
-        use differential_dataflow::trace::cursor::MyTrait;
-        oks.as_collection(|k, v| (k.into_owned(), v.into_owned()))
-            .leave()
-            .mz_arrange(name)
-    }
 }
 
 impl<G> Context<G>
@@ -717,8 +699,9 @@ where
                 let err: KeyCollection<_, _, _> = err.into();
                 let mut errs = err
                     .mz_arrange::<ErrSpine<_, _>>("Arrange recursive err")
-                    .mz_reduce_abelian::<_, ErrSpine<_, _>>(
+                    .mz_reduce_abelian::<_, _, _, ErrSpine<_, _>>(
                         "Distinct recursive err",
+                        Clone::clone,
                         move |_k, _s, t| t.push(((), 1)),
                     )
                     .as_collection(|k, _| k.clone());
@@ -1049,7 +1032,7 @@ where
                 // Pass through inputs.
                 input.for_each(|cap, data| {
                     data.swap(&mut buffer);
-                    output.session(&cap).give_vec(&mut buffer);
+                    output.session(&cap).give_container(&mut buffer);
                 });
 
                 if hydrated {
@@ -1308,7 +1291,7 @@ where
                     let cap = early_cap.as_ref().expect("early_cap can't be dropped yet");
                     output.session(cap)
                 };
-                session.give_vec(&mut buffer);
+                session.give_container(&mut buffer);
             });
 
             let frontier = input.frontier().frontier();
diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs
index 596f6ff32466b..71f3b88109057 100644
--- a/src/compute/src/render/context.rs
+++ b/src/compute/src/render/context.rs
@@ -1043,7 +1043,6 @@ impl<C> PendingWork<C>
 where
     C: Cursor,
     C::KeyOwned: PartialEq + Sized,
-    C::ValOwned: Sized,
     C::Time: Timestamp,
 {
     /// Create a new bundle of pending work, from the capability, cursor, and backing storage.
diff --git a/src/compute/src/render/flat_map.rs b/src/compute/src/render/flat_map.rs
index 3395bd7e46892..8e7ca3ddbc031 100644
--- a/src/compute/src/render/flat_map.rs
+++ b/src/compute/src/render/flat_map.rs
@@ -7,15 +7,15 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use mz_expr::MfpPlan;
 use mz_expr::{MapFilterProject, MirScalarExpr, TableFunc};
 use mz_repr::{DatumVec, RowArena, SharedRow};
 use mz_repr::{Diff, Row, Timestamp};
-use mz_timely_util::buffer::ConsolidateBuffer;
 use mz_timely_util::operator::StreamExt;
 use timely::dataflow::channels::pact::Pipeline;
-use timely::dataflow::channels::pushers::Tee;
-use timely::dataflow::operators::InputCapability;
+use timely::dataflow::channels::pushers::buffer::Session;
+use timely::dataflow::channels::pushers::{Counter, Tee};
 use timely::dataflow::Scope;
 use timely::progress::Antichain;
 
@@ -43,9 +43,6 @@ where
         let stream = ok_collection.inner;
         let (oks, errs) = stream.unary_fallible(Pipeline, "FlatMapStage", move |_, _| {
             Box::new(move |input, ok_output, err_output| {
-                let mut ok_session = ConsolidateBuffer::new(ok_output, 0);
-                let mut err_session = ConsolidateBuffer::new(err_output, 1);
-
                 let mut datums = DatumVec::new();
                 let mut datums_mfp = DatumVec::new();
 
@@ -55,6 +52,9 @@ where
                 input.for_each(|cap, data| {
                     data.swap(&mut storage);
 
+                    let mut ok_session = ok_output.session_with_builder(&cap);
+                    let mut err_session = err_output.session_with_builder(&cap);
+
                     'input: for (input_row, time, diff) in storage.drain(..) {
                         let temp_storage = RowArena::new();
 
@@ -67,14 +67,14 @@ where
                         let args = match args {
                             Ok(args) => args,
                             Err(e) => {
-                                err_session.give(&cap, (e.into(), time, diff));
+                                err_session.give((e.into(), time, diff));
                                 continue 'input;
                             }
                         };
                         let mut extensions = match func.eval(&args, &temp_storage) {
                             Ok(exts) => exts.fuse(),
                             Err(e) => {
-                                err_session.give(&cap, (e.into(), time, diff));
+                                err_session.give((e.into(), time, diff));
                                 continue 'input;
                             }
                         };
@@ -85,7 +85,6 @@ where
                             table_func_output.extend((&mut extensions).take(1023));
                             // We could consolidate `table_func_output`, but it seems unlikely to be productive.
                             drain_through_mfp(
-                                &cap,
                                 &input_row,
                                 &time,
                                 &diff,
@@ -115,7 +114,6 @@ where
 ///
 /// The method decodes `input_row`, and should be amortized across non-trivial `extensions`.
 fn drain_through_mfp<T>(
-    cap: &InputCapability<T>,
     input_row: &Row,
     input_time: &T,
     input_diff: &Diff,
@@ -123,12 +121,15 @@ fn drain_through_mfp<T>(
     extensions: &[(Row, Diff)],
     mfp_plan: &MfpPlan,
     until: &Antichain<Timestamp>,
-    ok_output: &mut ConsolidateBuffer<T, Row, Diff, Tee<T, Vec<(Row, T, Diff)>>>,
-    err_output: &mut ConsolidateBuffer<
+    ok_output: &mut Session<
+        T,
+        ConsolidatingContainerBuilder<Vec<(Row, T, Diff)>>,
+        Counter<T, Vec<(Row, T, Diff)>, Tee<T, Vec<(Row, T, Diff)>>>,
+    >,
+    err_output: &mut Session<
         T,
-        DataflowError,
-        Diff,
-        Tee<T, Vec<(DataflowError, T, Diff)>>,
+        ConsolidatingContainerBuilder<Vec<(DataflowError, T, Diff)>>,
+        Counter<T, Vec<(DataflowError, T, Diff)>, Tee<T, Vec<(DataflowError, T, Diff)>>>,
     >,
 ) where
     T: crate::render::RenderTimestamp,
@@ -163,13 +164,13 @@ fn drain_through_mfp<T>(
                     // Copy the whole time, and re-populate event time.
                     let mut time = input_time.clone();
                     *time.event_time_mut() = event_time;
-                    ok_output.give(cap, (row, time, diff));
+                    ok_output.give((row, time, diff));
                 }
                 Err((err, event_time, diff)) => {
                     // Copy the whole time, and re-populate event time.
                     let mut time = input_time.clone();
                     *time.event_time_mut() = event_time;
-                    err_output.give(cap, (err, time, diff));
+                    err_output.give((err, time, diff));
                 }
             };
         }
diff --git a/src/compute/src/render/reduce.rs b/src/compute/src/render/reduce.rs
index 211fdfa73f1a9..3586426a7f455 100644
--- a/src/compute/src/render/reduce.rs
+++ b/src/compute/src/render/reduce.rs
@@ -21,7 +21,7 @@ use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
 use differential_dataflow::trace::cursor::MyTrait;
 use differential_dataflow::trace::{Batch, Batcher, Builder, Trace, TraceReader};
-use differential_dataflow::{Collection, ExchangeData};
+use differential_dataflow::Collection;
 use mz_compute_types::plan::reduce::{
     reduction_type, AccumulablePlan, BasicPlan, BucketedPlan, HierarchicalPlan, KeyValPlan,
     MonotonicPlan, ReducePlan, ReductionType,
@@ -337,9 +337,11 @@ where
         let aggregate_types_err = aggregate_types.clone();
         let (oks, errs) = differential_dataflow::collection::concatenate(scope, to_concat)
             .mz_arrange::<RowValSpine<_, _, _>>("Arrange ReduceCollation")
-            .reduce_pair::<_, RowRowSpine<_, _>, _, RowErrSpine<_, _>>(
+            .reduce_pair::<_, _, _, RowRowSpine<_, _>, _, _, _, RowErrSpine<_, _>>(
                 "ReduceCollation",
                 "ReduceCollation Errors",
+                |v| v.into_owned(),
+                |v| v.into_owned(),
                 {
                     move |key, input, output| {
                         // The inputs are pairs of a reduction type, and a row consisting of densely
@@ -499,7 +501,7 @@ where
     /// Build the dataflow to compute the set of distinct keys.
     fn build_distinct<T1, T2, S>(
         &self,
-        collection: Collection<S, (Row, T1::ValOwned), Diff>,
+        collection: Collection<S, (Row, Row), Diff>,
         tag: &str,
         mfp_after: Option<SafeMfpPlan>,
     ) -> (
@@ -508,22 +510,22 @@ where
     )
     where
         S: Scope<Timestamp = G::Timestamp>,
-        T1: Trace<KeyOwned = Row, Time = G::Timestamp, Diff = Diff> + 'static,
+        for<'a> T1: Trace<KeyOwned = Row, Val<'a> = DatumSeq<'a>, Time = G::Timestamp, Diff = Diff>
+            + 'static,
         T1::Batch: Batch,
-        T1::Batcher: Batcher<Input = Vec<((Row, T1::ValOwned), G::Timestamp, Diff)>>,
-        T1::Builder: Builder<Input = ((Row, T1::ValOwned), G::Timestamp, Diff)>,
+        T1::Batcher: Batcher<Input = Vec<((Row, Row), G::Timestamp, Diff)>>,
+        T1::Builder: Builder<Input = ((Row, Row), G::Timestamp, Diff)>,
         for<'a> T1::Key<'a>: std::fmt::Debug + ToDatumIter,
-        T1::ValOwned: Columnation + ExchangeData + Default,
         Arranged<S, TraceAgent<T1>>: ArrangementSize,
         T2: for<'a> Trace<
                 Key<'a> = T1::Key<'a>,
+                Val<'a> = &'a DataflowError,
                 KeyOwned = Row,
-                ValOwned = DataflowError,
                 Time = G::Timestamp,
                 Diff = Diff,
             > + 'static,
         T2::Batch: Batch,
-        T2::Builder: Builder<Input = ((Row, T2::ValOwned), G::Timestamp, Diff)>,
+        T2::Builder: Builder<Input = ((Row, DataflowError), G::Timestamp, Diff)>,
         Arranged<S, TraceAgent<T2>>: ArrangementSize,
     {
         let error_logger = self.error_logger();
@@ -541,9 +543,11 @@ where
 
         let (output, errors) = collection
             .mz_arrange::<T1>(&input_name)
-            .reduce_pair::<_, T1, _, T2>(
+            .reduce_pair::<_, Row, _, T1, _, _, _, T2>(
                 &output_name,
                 "DistinctByErrorCheck",
+                |v| v.into_owned(),
+                |v| v.into_owned(),
                 move |key, _input, output| {
                     let temp_storage = RowArena::new();
                     let mut datums_local = datums1.borrow();
@@ -561,10 +565,10 @@ where
                         // We're pushing a unit value here because the key is implicitly added by the
                         // arrangement, and the permutation logic takes care of using the key part of the
                         // output.
-                        output.push((T1::ValOwned::default(), 1));
+                        output.push((Row::default(), 1));
                     }
                 },
-                move |key, input: &[(_, Diff)], output| {
+                move |key, input: &[(_, Diff)], output: &mut Vec<(DataflowError, _)>| {
                     for (_, count) in input.iter() {
                         if count.is_positive() {
                             continue;
@@ -586,7 +590,7 @@ where
                     }
                 },
             );
-        (output, errors.as_collection(|_k, v| v.into_owned()))
+        (output, errors.as_collection(|_k, v| v.clone()))
     }
 
     /// Build the dataflow to compute and arrange multiple non-accumulable,
@@ -643,25 +647,29 @@ where
             differential_dataflow::collection::concatenate(&mut input.scope(), to_collect)
                 .mz_arrange::<RowValSpine<_, _, _>>("Arranged ReduceFuseBasic input");
 
-        let output = arranged.mz_reduce_abelian::<_, RowRowSpine<_, _>>("ReduceFuseBasic", {
-            move |key, input, output| {
-                let temp_storage = RowArena::new();
-                let datum_iter = key.to_datum_iter();
-                let mut datums_local = datums1.borrow();
-                datums_local.extend(datum_iter);
-                let key_len = datums_local.len();
+        let output = arranged.mz_reduce_abelian::<_, _, _, RowRowSpine<_, _>>(
+            "ReduceFuseBasic",
+            |v| v.into_owned(),
+            {
+                move |key, input, output| {
+                    let temp_storage = RowArena::new();
+                    let datum_iter = key.to_datum_iter();
+                    let mut datums_local = datums1.borrow();
+                    datums_local.extend(datum_iter);
+                    let key_len = datums_local.len();
 
-                for ((_, row), _) in input.iter() {
-                    datums_local.push(row.unpack_first());
-                }
+                    for ((_, row), _) in input.iter() {
+                        datums_local.push(row.unpack_first());
+                    }
 
-                if let Some(row) =
-                    evaluate_mfp_after(&mfp_after1, &mut datums_local, &temp_storage, key_len)
-                {
-                    output.push((row, 1));
+                    if let Some(row) =
+                        evaluate_mfp_after(&mfp_after1, &mut datums_local, &temp_storage, key_len)
+                    {
+                        output.push((row, 1));
+                    }
                 }
-            }
-        });
+            },
+        );
         // If `mfp_after` can error, then we need to render a paired reduction
         // to scan for these potential errors. Note that we cannot directly use
         // `mz_timely_util::reduce::ReduceExt::reduce_pair` here because we only
@@ -669,8 +677,9 @@ where
         let validation_errs = err_output.expect("expected to validate in at least one aggregate");
         if let Some(mfp) = mfp_after2 {
             let mfp_errs = arranged
-                .mz_reduce_abelian::<_, RowErrSpine<_, _>>(
+                .mz_reduce_abelian::<_, _, _, RowErrSpine<_, _>>(
                     "ReduceFuseBasic Error Check",
+                    |v| v.into_owned(),
                     move |key, input, output| {
                         // Since negative accumulations are checked in at least one component
                         // aggregate, we only need to look for MFP errors here.
@@ -738,7 +747,7 @@ where
             let keyed = partial.map(move |(key, val)| pairer.merge(&key, &val));
             if validating {
                 let (oks, errs) = self
-                    .build_reduce_inaccumulable_distinct::<_, RowValSpine<Result<(), String>, _, _>>(keyed, None)
+                    .build_reduce_inaccumulable_distinct::<_, _, _, RowValSpine<Result<(), String>, _, _>>(keyed, None, |v| v.into_owned())
                     .as_collection(|k, v| (k.into_owned(), v.into_owned()))
                     .map_fallible("Demux Errors", move |(key_val, result)| match result {
                         Ok(()) => Ok(pairer.split(&key_val)),
@@ -748,9 +757,10 @@ where
                 partial = oks;
             } else {
                 partial = self
-                    .build_reduce_inaccumulable_distinct::<_, RowSpine<_, _>>(
+                    .build_reduce_inaccumulable_distinct::<_, _, _, RowSpine<_, _>>(
                         keyed,
                         Some(" [val: empty]"),
+                        Clone::clone,
                     )
                     .as_collection(move |key_val_iter, _| pairer.split(key_val_iter));
             }
@@ -764,39 +774,43 @@ where
         let func2 = func.clone();
 
         let arranged = partial.mz_arrange::<RowRowSpine<_, _>>("Arranged ReduceInaccumulable");
-        let oks = arranged.mz_reduce_abelian::<_, RowRowSpine<_, _>>("ReduceInaccumulable", {
-            move |key, source, target| {
-                // We respect the multiplicity here (unlike in hierarchical aggregation)
-                // because we don't know that the aggregation method is not sensitive
-                // to the number of records.
-                let iter = source.iter().flat_map(|(v, w)| {
-                    // Note that in the non-positive case, this is wrong, but harmless because
-                    // our other reduction will produce an error.
-                    let count = usize::try_from(*w).unwrap_or(0);
-                    std::iter::repeat(v.to_datum_iter().next().unwrap()).take(count)
-                });
+        let oks = arranged.mz_reduce_abelian::<_, _, _, RowRowSpine<_, _>>(
+            "ReduceInaccumulable",
+            |v| v.into_owned(),
+            {
+                move |key, source, target| {
+                    // We respect the multiplicity here (unlike in hierarchical aggregation)
+                    // because we don't know that the aggregation method is not sensitive
+                    // to the number of records.
+                    let iter = source.iter().flat_map(|(v, w)| {
+                        // Note that in the non-positive case, this is wrong, but harmless because
+                        // our other reduction will produce an error.
+                        let count = usize::try_from(*w).unwrap_or(0);
+                        std::iter::repeat(v.to_datum_iter().next().unwrap()).take(count)
+                    });
 
-                let temp_storage = RowArena::new();
-                let datum_iter = key.to_datum_iter();
-                let mut datums_local = datums1.borrow();
-                datums_local.extend(datum_iter);
-                let key_len = datums_local.len();
-                datums_local.push(
-                    // Note that this is not necessarily a window aggregation, in which case
-                    // `eval_fast_window_agg` delegates to the normal `eval`.
-                    func.eval_fast_window_agg::<_, window_agg_helpers::OneByOneAggrImpls>(
-                        iter,
-                        &temp_storage,
-                    ),
-                );
+                    let temp_storage = RowArena::new();
+                    let datum_iter = key.to_datum_iter();
+                    let mut datums_local = datums1.borrow();
+                    datums_local.extend(datum_iter);
+                    let key_len = datums_local.len();
+                    datums_local.push(
+                        // Note that this is not necessarily a window aggregation, in which case
+                        // `eval_fast_window_agg` delegates to the normal `eval`.
+                        func.eval_fast_window_agg::<_, window_agg_helpers::OneByOneAggrImpls>(
+                            iter,
+                            &temp_storage,
+                        ),
+                    );
 
-                if let Some(row) =
-                    evaluate_mfp_after(&mfp_after1, &mut datums_local, &temp_storage, key_len)
-                {
-                    target.push((row, 1));
+                    if let Some(row) =
+                        evaluate_mfp_after(&mfp_after1, &mut datums_local, &temp_storage, key_len)
+                    {
+                        target.push((row, 1));
+                    }
                 }
-            }
-        });
+            },
+        );
 
         // Note that we would prefer to use `mz_timely_util::reduce::ReduceExt::reduce_pair` here, but
         // we then wouldn't be able to do this error check conditionally.  See its documentation for the
@@ -806,8 +820,9 @@ where
             let error_logger = self.error_logger();
 
             let errs = arranged
-                .mz_reduce_abelian::<_, RowErrSpine<_, _>>(
+                .mz_reduce_abelian::<_, _, _, RowErrSpine<_, _>>(
                     "ReduceInaccumulable Error Check",
+                    |v| v.into_owned(),
                     move |key, source, target| {
                         // Negative counts would be surprising, but until we are 100% certain we won't
                         // see them, we should report when we do. We may want to bake even more info
@@ -861,19 +876,21 @@ where
         (oks, err_output)
     }
 
-    fn build_reduce_inaccumulable_distinct<S, Tr>(
+    fn build_reduce_inaccumulable_distinct<S, V, F, Tr>(
         &self,
         input: Collection<S, Row, Diff>,
         name_tag: Option<&str>,
+        from: F,
     ) -> Arranged<S, TraceAgent<Tr>>
     where
         S: Scope<Timestamp = G::Timestamp>,
-        Tr::ValOwned: MaybeValidatingRow<(), String>,
+        V: MaybeValidatingRow<(), String>,
+        F: Fn(Tr::Val<'_>) -> V + 'static,
         Tr: Trace
             + for<'a> TraceReader<Key<'a> = DatumSeq<'a>, Time = G::Timestamp, Diff = Diff>
             + 'static,
         Tr::Batch: Batch,
-        Tr::Builder: Builder<Input = ((Row, Tr::ValOwned), G::Timestamp, Diff)>,
+        Tr::Builder: Builder<Input = ((Row, V), G::Timestamp, Diff)>,
         Arranged<S, TraceAgent<Tr>>: ArrangementSize,
     {
         let error_logger = self.error_logger();
@@ -886,8 +903,8 @@ where
         let input: KeyCollection<_, _, _> = input.into();
         input
             .mz_arrange::<RowSpine<_, _>>("Arranged ReduceInaccumulable Distinct [val: empty]")
-            .mz_reduce_abelian::<_, Tr>(&output_name, move |_, source, t| {
-                if let Some(err) = Tr::ValOwned::into_error() {
+            .mz_reduce_abelian::<_, _, _, Tr>(&output_name, from, move |_, source, t| {
+                if let Some(err) = V::into_error() {
                     for (value, count) in source.iter() {
                         if count.is_positive() {
                             continue;
@@ -899,7 +916,7 @@ where
                         return;
                     }
                 }
-                t.push((Tr::ValOwned::ok(()), 1))
+                t.push((V::ok(()), 1))
             })
     }
 
@@ -967,8 +984,9 @@ where
                 // stream or produce correct data in the output stream.
                 let negated_output = if err_output.is_none() {
                     let (oks, errs) = self
-                        .build_bucketed_negated_output::<_, RowValSpine<Result<Row, Row>, _, _>>(
+                        .build_bucketed_negated_output::<_,_,_, RowValSpine<Result<Row, Row>, _, _>>(
                             &input,
+                            |v| v.into_owned(),
                             aggr_funcs.clone(),
                         )
                         .as_collection(|k, v| (SharedRow::pack(k), v.clone()))
@@ -990,8 +1008,9 @@ where
                     err_output = Some(errs.leave_region());
                     oks
                 } else {
-                    self.build_bucketed_negated_output::<_, RowRowSpine<_, _>>(
+                    self.build_bucketed_negated_output::<_,_,_, RowRowSpine<_, _>>(
                         &input,
+                        |v| v.into_owned(),
                         aggr_funcs.clone(),
                     )
                     .as_collection(|k, v| {
@@ -1035,8 +1054,9 @@ where
             let must_validate = err_output.is_none();
             if must_validate || mfp_after2.is_some() {
                 let errs = arranged
-                    .mz_reduce_abelian::<_, RowErrSpine<_, _>>(
+                    .mz_reduce_abelian::<_,_,_, RowErrSpine<_, _>>(
                         "ReduceMinsMaxes Error Check",
+                        |v| v.into_owned(),
                         move |key, source, target| {
                             // Negative counts would be surprising, but until we are 100% certain we wont
                             // see them, we should report when we do. We may want to bake even more info
@@ -1088,7 +1108,9 @@ where
                 }
             }
             arranged
-                .mz_reduce_abelian::<_, RowRowSpine<_, _>>("ReduceMinsMaxes", {
+                .mz_reduce_abelian::<_,_,_, RowRowSpine<_, _>>("ReduceMinsMaxes",
+                                                               |v| v.into_owned(),
+
                     move |key, source, target| {
                         let temp_storage = RowArena::new();
                         let datum_iter = key.to_datum_iter();
@@ -1115,7 +1137,7 @@ where
                             target.push((row, 1));
                         }
                     }
-                })
+                )
                 .leave_region()
         });
         (
@@ -1135,19 +1157,21 @@ where
     /// `validating` indicates whether we want this stage to perform error detection
     /// for invalid accumulations. Once a stage is clean of such errors, subsequent
     /// stages can skip validation.
-    fn build_bucketed_negated_output<S, Tr>(
+    fn build_bucketed_negated_output<S, V, F, Tr>(
         &self,
         input: &Collection<S, (Row, Row), Diff>,
+        from: F,
         aggrs: Vec<AggregateFunc>,
     ) -> Arranged<S, TraceAgent<Tr>>
     where
         S: Scope<Timestamp = G::Timestamp>,
-        Tr::ValOwned: MaybeValidatingRow<Row, Row>,
+        V: MaybeValidatingRow<Row, Row>,
+        F: Fn(Tr::Val<'_>) -> V + 'static,
         Tr: Trace
             + for<'a> TraceReader<Key<'a> = DatumSeq<'a>, Time = G::Timestamp, Diff = Diff>
             + 'static,
         Tr::Batch: Batch,
-        Tr::Builder: Builder<Input = ((Row, Tr::ValOwned), G::Timestamp, Diff)>,
+        Tr::Builder: Builder<Input = ((Row, V), G::Timestamp, Diff)>,
         Arranged<S, TraceAgent<Tr>>: ArrangementSize,
     {
         let error_logger = self.error_logger();
@@ -1156,13 +1180,14 @@ where
         let arranged_input =
             input.mz_arrange::<RowRowSpine<_, _>>("Arranged MinsMaxesHierarchical input");
 
-        arranged_input.mz_reduce_abelian::<_, Tr>(
+        arranged_input.mz_reduce_abelian::<_, _, _, Tr>(
             "Reduced Fallibly MinsMaxesHierarchical",
+            from,
             move |key, source, target| {
                 let binding = SharedRow::get();
                 let mut row_builder = binding.borrow_mut();
                 let mut row_packer = row_builder.packer();
-                if let Some(err) = Tr::ValOwned::into_error() {
+                if let Some(err) = V::into_error() {
                     // Should negative accumulations reach us, we should loudly complain.
                     for (value, count) in source.iter() {
                         if count.is_positive() {
@@ -1194,11 +1219,11 @@ where
                 // of the multiplicity of the final result in the input, we only want to have one copy
                 // in the output.
                 target.reserve(source.len().saturating_add(1));
-                target.push((Tr::ValOwned::ok(row_builder.clone()), -1));
+                target.push((V::ok(row_builder.clone()), -1));
                 for (values, cnt) in source.iter() {
                     row_packer = row_builder.packer();
                     row_packer.extend(*values);
-                    target.push((Tr::ValOwned::ok(row_builder.clone()), *cnt));
+                    target.push((V::ok(row_builder.clone()), *cnt));
                 }
             },
         )
@@ -1271,25 +1296,29 @@ where
         let partial: KeyCollection<_, _, _> = partial.into();
         let arranged = partial
             .mz_arrange::<RowSpine<_, Vec<ReductionMonoid>>>("ArrangeMonotonic [val: empty]");
-        let output = arranged.mz_reduce_abelian::<_, RowRowSpine<_, _>>("ReduceMonotonic", {
-            move |key, input, output| {
-                let temp_storage = RowArena::new();
-                let datum_iter = key.to_datum_iter();
-                let mut datums_local = datums1.borrow();
-                datums_local.extend(datum_iter);
-                let key_len = datums_local.len();
-                let accum = &input[0].1;
-                for monoid in accum.iter() {
-                    datums_local.extend(monoid.finalize().iter());
-                }
+        let output = arranged.mz_reduce_abelian::<_, _, _, RowRowSpine<_, _>>(
+            "ReduceMonotonic",
+            |v| v.into_owned(),
+            {
+                move |key, input, output| {
+                    let temp_storage = RowArena::new();
+                    let datum_iter = key.to_datum_iter();
+                    let mut datums_local = datums1.borrow();
+                    datums_local.extend(datum_iter);
+                    let key_len = datums_local.len();
+                    let accum = &input[0].1;
+                    for monoid in accum.iter() {
+                        datums_local.extend(monoid.finalize().iter());
+                    }
 
-                if let Some(row) =
-                    evaluate_mfp_after(&mfp_after1, &mut datums_local, &temp_storage, key_len)
-                {
-                    output.push((row, 1));
+                    if let Some(row) =
+                        evaluate_mfp_after(&mfp_after1, &mut datums_local, &temp_storage, key_len)
+                    {
+                        output.push((row, 1));
+                    }
                 }
-            }
-        });
+            },
+        );
 
         // If `mfp_after` can error, then we need to render a paired reduction
         // to scan for these potential errors. Note that we cannot directly use
@@ -1297,7 +1326,9 @@ where
         // conditionally render the second component of the reduction pair.
         if let Some(mfp) = mfp_after2 {
             let mfp_errs = arranged
-                .mz_reduce_abelian::<_, RowErrSpine<_, _>>("ReduceMonotonic Error Check", {
+                .mz_reduce_abelian::<_, _, _, RowErrSpine<_, _>>(
+                    "ReduceMonotonic Error Check",
+                    |v| v.into_owned(),
                     move |key, input, output| {
                         let temp_storage = RowArena::new();
                         let datum_iter = key.to_datum_iter();
@@ -1311,8 +1342,8 @@ where
                         {
                             output.push((e.into(), 1));
                         }
-                    }
-                })
+                    },
+                )
                 .as_collection(|_k, v| v.clone());
             (output, validation_errs.concat(&mfp_errs))
         } else {
@@ -1409,8 +1440,9 @@ where
                     (pairer.merge(&key, std::iter::once(value)), ())
                 })
                 .mz_arrange::<RowSpine<_, _>>("Arranged Accumulable Distinct [val: empty]")
-                .mz_reduce_abelian::<_, RowSpine<_, _>>(
+                .mz_reduce_abelian::<_, _, _, RowSpine<_, _>>(
                     "Reduced Accumulable Distinct [val: empty]",
+                    |v| v.into_owned(),
                     move |_k, _s, t| t.push(((), 1)),
                 )
                 .as_collection(move |key_val_iter, _| pairer.split(key_val_iter))
@@ -1445,9 +1477,11 @@ where
         let err_full_aggrs = full_aggrs.clone();
         let (arranged_output, arranged_errs) = collection
             .mz_arrange::<RowSpine<_, (Vec<Accum>, Diff)>>("ArrangeAccumulable [val: empty]")
-            .reduce_pair::<_, RowRowSpine<_, _>, _, RowErrSpine<_, _>>(
+            .reduce_pair::<_, _, _, RowRowSpine<_, _>, _, _, _, RowErrSpine<_, _>>(
                 "ReduceAccumulable",
                 "AccumulableErrorCheck",
+                |v| v.into_owned(),
+                |v| v.into_owned(),
                 {
                     move |key, input, output| {
                         let (ref accums, total) = input[0].1;
diff --git a/src/compute/src/render/threshold.rs b/src/compute/src/render/threshold.rs
index 27f19d7e66e3b..e33a63cf483b8 100644
--- a/src/compute/src/render/threshold.rs
+++ b/src/compute/src/render/threshold.rs
@@ -13,6 +13,7 @@
 
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
+use differential_dataflow::trace::cursor::MyTrait;
 use differential_dataflow::trace::{Batch, Builder, Trace, TraceReader};
 use differential_dataflow::Data;
 use mz_compute_types::plan::threshold::{BasicThresholdPlan, ThresholdPlan};
@@ -28,35 +29,37 @@ use crate::extensions::reduce::MzReduce;
 use crate::render::context::{
     ArrangementFlavor, CollectionBundle, Context, MzArrangement, MzArrangementImport,
 };
+use crate::row_spine::DatumSeq;
 
 /// Shared function to compute an arrangement of values matching `logic`.
-fn threshold_arrangement<G, T1, T2, L>(
+fn threshold_arrangement<G, V, F, T1, T2, L>(
     arrangement: &Arranged<G, T1>,
     name: &str,
+    from: F,
     logic: L,
 ) -> Arranged<G, TraceAgent<T2>>
 where
     G: Scope,
     G::Timestamp: Lattice + Columnation,
+    F: Fn(T2::Val<'_>) -> V + Clone + 'static,
+    V: Data,
     T1: TraceReader<Time = G::Timestamp, Diff = Diff> + Clone + 'static,
     T1::KeyOwned: Columnation + Data,
     T2: for<'a> Trace<
             Key<'a> = T1::Key<'a>,
-            ValOwned = T1::ValOwned,
+            Val<'a> = T1::Val<'a>,
             Time = G::Timestamp,
             Diff = Diff,
         > + 'static,
-    T2::ValOwned: Columnation + Data,
     T2::Batch: Batch,
-    T2::Builder: Builder<Input = ((T1::KeyOwned, T2::ValOwned), G::Timestamp, Diff)>,
+    T2::Builder: Builder<Input = ((T1::KeyOwned, V), G::Timestamp, Diff)>,
     L: Fn(&Diff) -> bool + 'static,
     Arranged<G, TraceAgent<T2>>: ArrangementSize,
 {
-    arrangement.mz_reduce_abelian(name, move |_key, s, t| {
+    arrangement.mz_reduce_abelian(name, from.clone(), move |_key, s, t| {
         for (record, count) in s.iter() {
             if logic(count) {
-                use differential_dataflow::trace::cursor::MyTrait;
-                t.push(((*record).into_owned(), *count));
+                t.push((from(*record), *count));
             }
         }
     })
@@ -76,7 +79,7 @@ where
 {
     match oks {
         MzArrangement::RowRow(inner) => {
-            let oks = threshold_arrangement(inner, name, logic);
+            let oks = threshold_arrangement(inner, name, |v: DatumSeq| v.into_owned(), logic);
             MzArrangement::RowRow(oks)
         }
     }
@@ -96,7 +99,7 @@ where
 {
     match oks {
         MzArrangementImport::RowRow(inner) => {
-            let oks = threshold_arrangement(inner, name, logic);
+            let oks = threshold_arrangement(inner, name, |v: DatumSeq| v.into_owned(), logic);
             MzArrangement::RowRow(oks)
         }
     }
diff --git a/src/compute/src/render/top_k.rs b/src/compute/src/render/top_k.rs
index 0d19bbb6b0b72..cb0a70b1c2fef 100644
--- a/src/compute/src/render/top_k.rs
+++ b/src/compute/src/render/top_k.rs
@@ -18,6 +18,7 @@ use std::rc::Rc;
 use differential_dataflow::hashable::Hashable;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
+use differential_dataflow::trace::cursor::MyTrait;
 use differential_dataflow::trace::{Batch, Builder, Trace, TraceReader};
 use differential_dataflow::{AsCollection, Collection};
 use mz_compute_types::plan::top_k::{
@@ -363,8 +364,13 @@ where
             (hash_key, row)
         });
         let (oks, errs) = if validating {
-            let stage = build_topk_negated_stage::<S, RowValSpine<Result<Row, Row>, _, _>>(
-                &input, order_key, offset, limit, arity,
+            let stage = build_topk_negated_stage::<S, _, _, RowValSpine<Result<Row, Row>, _, _>>(
+                &input,
+                |v| v.into_owned(),
+                order_key,
+                offset,
+                limit,
+                arity,
             )
             .as_collection(|k, v| (SharedRow::pack(k), v.clone()));
 
@@ -384,8 +390,13 @@ where
             (oks, Some(errs))
         } else {
             (
-                build_topk_negated_stage::<S, RowRowSpine<_, _>>(
-                    &input, order_key, offset, limit, arity,
+                build_topk_negated_stage::<S, _, _, RowRowSpine<_, _>>(
+                    &input,
+                    |v| v.into_owned(),
+                    order_key,
+                    offset,
+                    limit,
+                    arity,
                 )
                 .as_collection(|k, v| {
                     let binding = SharedRow::get();
@@ -459,20 +470,22 @@ where
             .into();
         let result = partial
             .mz_arrange::<RowSpine<_, _>>("Arranged MonotonicTop1 partial [val: empty]")
-            .mz_reduce_abelian::<_, RowRowSpine<_, _>>("MonotonicTop1", {
+            .mz_reduce_abelian::<_, _, _, RowRowSpine<_, _>>(
+                "MonotonicTop1",
+                |v| v.into_owned(),
                 move |_key, input, output| {
                     let accum: &monoids::Top1Monoid = &input[0].1;
                     output.push((accum.row.clone(), 1));
-                }
-            });
+                },
+            );
         // TODO(#7331): Here we discard the arranged output.
-        use differential_dataflow::trace::cursor::MyTrait;
         (result.as_collection(|_k, v| v.into_owned()), errs)
     }
 }
 
-fn build_topk_negated_stage<G, Tr>(
+fn build_topk_negated_stage<G, V, F, Tr>(
     input: &Collection<G, (Row, Row), Diff>,
+    from: F,
     order_key: Vec<mz_expr::ColumnOrder>,
     offset: usize,
     limit: Option<mz_expr::MirScalarExpr>,
@@ -481,12 +494,13 @@ fn build_topk_negated_stage<G, Tr>(
 where
     G: Scope,
     G::Timestamp: Lattice + Columnation,
-    Tr::ValOwned: MaybeValidatingRow<Row, Row>,
+    F: Fn(Tr::Val<'_>) -> V + 'static,
+    V: MaybeValidatingRow<Row, Row>,
     Tr: Trace
         + for<'a> TraceReader<Key<'a> = DatumSeq<'a>, Time = G::Timestamp, Diff = Diff>
         + 'static,
     Tr::Batch: Batch,
-    Tr::Builder: Builder<Input = ((Row, Tr::ValOwned), G::Timestamp, Diff)>,
+    Tr::Builder: Builder<Input = ((Row, V), G::Timestamp, Diff)>,
     Arranged<G, TraceAgent<Tr>>: ArrangementSize,
 {
     let mut datum_vec = mz_repr::DatumVec::new();
@@ -497,8 +511,8 @@ where
     // built-in view mz_internal.mz_expected_group_size_advice.
     input
         .mz_arrange::<RowRowSpine<_, _>>("Arranged TopK input")
-        .mz_reduce_abelian::<_, Tr>("Reduced TopK input", {
-            move |mut hash_key, source, target: &mut Vec<(Tr::ValOwned, Diff)>| {
+        .mz_reduce_abelian::<_, _, _, Tr>("Reduced TopK input", from, {
+            move |mut hash_key, source, target: &mut Vec<(V, Diff)>| {
                 // Unpack the limit, either into an integer literal or an expression to evaluate.
                 let limit: Option<i64> = limit.as_ref().map(|l| {
                     if let Some(l) = l.as_literal_int64() {
@@ -521,7 +535,7 @@ where
                     }
                 });
 
-                if let Some(err) = Tr::ValOwned::into_error() {
+                if let Some(err) = V::into_error() {
                     for (datums, diff) in source.iter() {
                         if diff.is_positive() {
                             continue;
@@ -549,10 +563,7 @@ where
                 // dependencies on the user-provided (potentially unbounded) limit.
                 target.reserve(source.len());
                 for (datums, diff) in source.iter() {
-                    target.push((
-                        Tr::ValOwned::ok(row_builder.pack_using(*datums)),
-                        diff.clone(),
-                    ));
+                    target.push((V::ok(row_builder.pack_using(*datums)), diff.clone()));
                 }
                 // local copies that may count down to zero.
                 let mut offset = offset;
@@ -603,7 +614,7 @@ where
                         // Emit retractions for the elements actually part of
                         // the set of TopK elements.
                         row_builder.packer().extend(datums);
-                        target.push((Tr::ValOwned::ok(row_builder.clone()), -diff));
+                        target.push((V::ok(row_builder.clone()), -diff));
                     }
                 }
             }
diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs
index a73eb6b9259f6..bf35126476650 100644
--- a/src/compute/src/server.rs
+++ b/src/compute/src/server.rs
@@ -344,6 +344,7 @@ impl<'w, A: Allocate + 'static> Worker<'w, A> {
             let cmd_queue = Rc::clone(&cmd_queue);
 
             move |scope| {
+                let mut container = Default::default();
                 source(scope, "CmdSource", |capability, info| {
                     // Send activator for this operator back
                     let activator = scope.sync_activator_for(&info.address[..]);
@@ -398,21 +399,18 @@ impl<'w, A: Allocate + 'static> Worker<'w, A> {
                         }
                     }
                 })
-                .unary_frontier::<Vec<()>, _, _, _>(
+                .sink(
                     Exchange::new(|(idx, _)| u64::cast_from(*idx)),
                     "CmdReceiver",
-                    |_, _| {
-                        let mut container = Default::default();
-                        move |input, _| {
-                            let mut queue = cmd_queue.borrow_mut();
-                            if input.frontier().is_empty() {
-                                queue.push_back(Err(TryRecvError::Disconnected))
-                            }
-                            while let Some((_, data)) = input.next() {
-                                data.swap(&mut container);
-                                for (_, cmd) in container.drain(..) {
-                                    queue.push_back(Ok(cmd));
-                                }
+                    move |input| {
+                        let mut queue = cmd_queue.borrow_mut();
+                        if input.frontier().is_empty() {
+                            queue.push_back(Err(TryRecvError::Disconnected))
+                        }
+                        while let Some((_, data)) = input.next() {
+                            data.swap(&mut container);
+                            for (_, cmd) in container.drain(..) {
+                                queue.push_back(Ok(cmd));
                             }
                         }
                     },
diff --git a/src/compute/src/sink/persist_sink.rs b/src/compute/src/sink/persist_sink.rs
index b424f8198179a..5d41d1af8cee0 100644
--- a/src/compute/src/sink/persist_sink.rs
+++ b/src/compute/src/sink/persist_sink.rs
@@ -29,6 +29,7 @@ use mz_storage_types::errors::DataflowError;
 use mz_storage_types::sources::SourceData;
 use mz_timely_util::builder_async::{Event, OperatorBuilder as AsyncOperatorBuilder};
 use serde::{Deserialize, Serialize};
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::{Exchange, Pipeline};
 use timely::dataflow::operators::{
     Broadcast, Capability, CapabilitySet, ConnectLoop, Feedback, Inspect,
@@ -297,9 +298,11 @@ where
     let mut mint_op =
         AsyncOperatorBuilder::new(format!("{} mint_batch_descriptions", operator_name), scope);
 
-    let (mut output, output_stream) = mint_op.new_output();
-    let (mut desired_oks_output, desired_oks_output_stream) = mint_op.new_output();
-    let (mut desired_errs_output, desired_errs_output_stream) = mint_op.new_output();
+    let (mut output, output_stream) = mint_op.new_output::<CapacityContainerBuilder<_>>();
+    let (mut desired_oks_output, desired_oks_output_stream) =
+        mint_op.new_output::<CapacityContainerBuilder<_>>();
+    let (mut desired_errs_output, desired_errs_output_stream) =
+        mint_op.new_output::<CapacityContainerBuilder<_>>();
 
     // The `desired` inputs drive both the description output and their respective passthrough
     // output.
@@ -952,7 +955,7 @@ where
     // We never output anything, but we update our capabilities based on the
     // persist frontier we know about. So someone can listen on our output
     // frontier and learn about the persist frontier advancing.
-    let (mut _output, output_stream) = append_op.new_output();
+    let (mut _output, output_stream) = append_op.new_output::<CapacityContainerBuilder<_>>();
 
     let hashed_id = sink_id.hashed();
     let active_worker = usize::cast_from(hashed_id) % scope.peers() == scope.index();
diff --git a/src/compute/src/sink/refresh.rs b/src/compute/src/sink/refresh.rs
index b1d0e2bc3ec07..4a849919bbe87 100644
--- a/src/compute/src/sink/refresh.rs
+++ b/src/compute/src/sink/refresh.rs
@@ -7,11 +7,11 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::{AsCollection, Collection, Data};
 use mz_expr::refresh_schedule::RefreshSchedule;
 use mz_ore::soft_panic_or_log;
 use mz_repr::{Diff, Timestamp};
-use mz_timely_util::buffer::ConsolidateBuffer;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
 use timely::dataflow::Scope;
@@ -35,7 +35,7 @@ where
     // like to round up frontiers as well as data: as soon as our input frontier passes a refresh
     // time, we'll round it up to the next refresh time.
     let mut builder = OperatorBuilder::new("apply_refresh".to_string(), coll.scope());
-    let (mut output_buf, output_stream) = builder.new_output();
+    let (mut output_buf, output_stream) = builder.new_output::<ConsolidatingContainerBuilder<_>>();
     let mut input = builder.new_input_connection(&coll.inner, Pipeline, vec![Antichain::new()]);
     builder.build(move |capabilities| {
         // This capability directly controls this operator's output frontier (because we have
@@ -46,7 +46,6 @@ where
         let mut buffer = Vec::new();
         move |frontiers| {
             let mut output_handle_core = output_buf.activate();
-            let mut output_buf = ConsolidateBuffer::new(&mut output_handle_core, 0);
             input.for_each(|input_cap, data| {
                 // Note that we can't use `input_cap` to get an output session because we might have
                 // advanced our output frontier already beyond the frontier of this capability.
@@ -62,6 +61,7 @@ where
                     );
                     return;
                 };
+                let mut output_buf = output_handle_core.session_with_builder(&capability);
 
                 data.swap(&mut buffer);
                 let mut cached_ts: Option<Timestamp> = None;
@@ -82,7 +82,7 @@ where
                     };
                     match rounded_up_data_ts {
                         Some(rounded_up_ts) => {
-                            output_buf.give_at(capability, (d, rounded_up_ts, r));
+                            output_buf.give((d, rounded_up_ts, r));
                         }
                         None => {
                             // This record is after the last refresh, which is not possible because
diff --git a/src/compute/src/typedefs.rs b/src/compute/src/typedefs.rs
index df0ceb56846df..8540a3adbbfd4 100644
--- a/src/compute/src/typedefs.rs
+++ b/src/compute/src/typedefs.rs
@@ -13,7 +13,8 @@
 
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::operators::arrange::TraceAgent;
-use differential_dataflow::trace::implementations::merge_batcher_col::ColumnatedMergeBatcher;
+use differential_dataflow::trace::implementations::merge_batcher::MergeBatcher;
+use differential_dataflow::trace::implementations::merge_batcher_col::ColumnationMerger;
 use differential_dataflow::trace::wrappers::enter::TraceEnter;
 use differential_dataflow::trace::wrappers::frontier::TraceFrontier;
 use mz_repr::Diff;
@@ -107,5 +108,5 @@ pub type KeyErrSpine<K, T, R> = ColValSpine<K, DataflowError, T, R>;
 pub type RowErrSpine<T, R> = RowValSpine<DataflowError, T, R>;
 
 // Batchers for consolidation
-pub type KeyBatcher<K, T, D> = ColumnatedMergeBatcher<K, (), T, D>;
-pub type KeyValBatcher<K, V, T, D> = ColumnatedMergeBatcher<K, V, T, D>;
+pub type KeyBatcher<K, T, D> = MergeBatcher<ColumnationMerger<((K, ()), T, D)>, T>;
+pub type KeyValBatcher<K, V, T, D> = MergeBatcher<ColumnationMerger<((K, V), T, D)>, T>;
diff --git a/src/persist-txn/src/operator.rs b/src/persist-txn/src/operator.rs
index fd906c29e1f88..3048a1817197b 100644
--- a/src/persist-txn/src/operator.rs
+++ b/src/persist-txn/src/operator.rs
@@ -31,6 +31,7 @@ use mz_timely_util::builder_async::{
     AsyncInputHandle, Event as AsyncEvent, InputConnection,
     OperatorBuilder as AsyncOperatorBuilder, PressOnDropButton,
 };
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::capture::Event;
 use timely::dataflow::operators::{Broadcast, Capture, Leave, Map, Probe};
@@ -312,7 +313,8 @@ where
         passthrough.scope().peers(),
         data_id.to_string(),
     );
-    let (mut passthrough_output, passthrough_stream) = builder.new_output();
+    let (mut passthrough_output, passthrough_stream) =
+        builder.new_output::<CapacityContainerBuilder<_>>();
     let mut remap_input = builder.new_disconnected_input(&remap, Pipeline);
     let mut passthrough_input = builder.new_disconnected_input(&passthrough, Pipeline);
 
diff --git a/src/storage-operators/src/persist_source.rs b/src/storage-operators/src/persist_source.rs
index 822c35d0f0489..4745d74be7496 100644
--- a/src/storage-operators/src/persist_source.rs
+++ b/src/storage-operators/src/persist_source.rs
@@ -9,6 +9,7 @@
 
 //! A source that reads from an a persist shard.
 
+use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use std::convert::Infallible;
 use std::fmt::Debug;
 use std::future::Future;
@@ -35,7 +36,6 @@ use mz_storage_types::controller::{CollectionMetadata, TxnsCodecRow};
 use mz_storage_types::errors::DataflowError;
 use mz_storage_types::sources::SourceData;
 use mz_storage_types::stats::RelationPartStats;
-use mz_timely_util::buffer::ConsolidateBuffer;
 use mz_timely_util::builder_async::{
     Event, OperatorBuilder as AsyncOperatorBuilder, PressOnDropButton,
 };
@@ -45,6 +45,7 @@ use timely::communication::Push;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::channels::Bundle;
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
+use timely::dataflow::operators::generic::OutputHandleCore;
 use timely::dataflow::operators::{Capability, Leave, OkErr};
 use timely::dataflow::operators::{CapabilitySet, ConnectLoop, Feedback};
 use timely::dataflow::scopes::Child;
@@ -398,7 +399,8 @@ where
     let operator_info = builder.operator_info();
 
     let mut fetched_input = builder.new_input(fetched, Pipeline);
-    let (mut updates_output, updates_stream) = builder.new_output();
+    let (mut updates_output, updates_stream) =
+        builder.new_output::<ConsolidatingContainerBuilder<_>>();
 
     // Re-used state for processing and building rows.
     let mut datum_vec = mz_repr::DatumVec::new();
@@ -436,7 +438,6 @@ where
             let mut work = 0;
             let start_time = Instant::now();
             let mut output = updates_output.activate();
-            let mut handle = ConsolidateBuffer::new(&mut output, 0);
             while !pending_work.is_empty() && !yield_fn(start_time, work) {
                 let done = pending_work.front_mut().unwrap().do_work(
                     &mut work,
@@ -447,7 +448,7 @@ where
                     map_filter_project.as_ref(),
                     &mut datum_vec,
                     &mut row_builder,
-                    &mut handle,
+                    &mut output,
                 );
                 if done {
                     pending_work.pop_front();
@@ -503,10 +504,16 @@ impl PendingWork {
         map_filter_project: Option<&MfpPlan>,
         datum_vec: &mut DatumVec,
         row_builder: &mut Row,
-        output: &mut ConsolidateBuffer<
+        output: &mut OutputHandleCore<
+            '_,
             (mz_repr::Timestamp, Subtime),
-            Result<Row, DataflowError>,
-            Diff,
+            ConsolidatingContainerBuilder<
+                Vec<(
+                    Result<Row, DataflowError>,
+                    (mz_repr::Timestamp, Subtime),
+                    Diff,
+                )>,
+            >,
             P,
         >,
     ) -> bool
@@ -523,6 +530,7 @@ impl PendingWork {
         >,
         YFn: Fn(Instant, usize) -> bool,
     {
+        let mut session = output.session_with_builder(&self.capability);
         let fetched_part = self.part.part_mut();
         let is_filter_pushdown_audit = fetched_part.is_filter_pushdown_audit();
         let mut row_buf = None;
@@ -576,8 +584,7 @@ impl PendingWork {
                                     if !until.less_equal(&time) {
                                         let mut emit_time = *self.capability.time();
                                         emit_time.0 = time;
-                                        output
-                                            .give_at(&self.capability, (Ok(row), emit_time, diff));
+                                        session.give((Ok(row), emit_time, diff));
                                         *work += 1;
                                     }
                                 }
@@ -586,8 +593,7 @@ impl PendingWork {
                                     if !until.less_equal(&time) {
                                         let mut emit_time = *self.capability.time();
                                         emit_time.0 = time;
-                                        output
-                                            .give_at(&self.capability, (Err(err), emit_time, diff));
+                                        session.give((Err(err), emit_time, diff));
                                         *work += 1;
                                     }
                                 }
@@ -601,14 +607,14 @@ impl PendingWork {
                     } else {
                         let mut emit_time = *self.capability.time();
                         emit_time.0 = time;
-                        output.give_at(&self.capability, (Ok(row), emit_time, diff));
+                        session.give((Ok(row), emit_time, diff));
                         *work += 1;
                     }
                 }
                 (Ok(SourceData(Err(err))), Ok(())) => {
                     let mut emit_time = *self.capability.time();
                     emit_time.0 = time;
-                    output.give_at(&self.capability, (Err(err), emit_time, diff));
+                    session.give((Err(err), emit_time, diff));
                     *work += 1;
                 }
                 // TODO(petrosagg): error handling
@@ -930,6 +936,7 @@ where
 
 #[cfg(test)]
 mod tests {
+    use timely::container::CapacityContainerBuilder;
     use timely::dataflow::operators::{Enter, Probe};
     use tokio::sync::mpsc::unbounded_channel;
     use tokio::sync::oneshot;
@@ -1354,7 +1361,8 @@ mod tests {
     ) -> (Stream<G, Part>, oneshot::Sender<()>) {
         let (finalizer_tx, finalizer_rx) = oneshot::channel();
         let mut iterator = AsyncOperatorBuilder::new("iterator".to_string(), scope);
-        let (mut output_handle, output) = iterator.new_output::<Vec<Part>>();
+        let (mut output_handle, output) =
+            iterator.new_output::<CapacityContainerBuilder<Vec<Part>>>();
 
         iterator.build(|mut caps| async move {
             let mut capability = Some(caps.pop().unwrap());
@@ -1391,7 +1399,8 @@ mod tests {
     ) -> UnboundedSender<()> {
         let (tx, mut rx) = unbounded_channel::<()>();
         let mut consumer = AsyncOperatorBuilder::new("consumer".to_string(), scope);
-        let (output_handle, output) = consumer.new_output::<Vec<std::convert::Infallible>>();
+        let (output_handle, output) =
+            consumer.new_output::<CapacityContainerBuilder<Vec<std::convert::Infallible>>>();
         let mut input = consumer.new_input_for(input, Pipeline, &output_handle);
 
         consumer.build(|_caps| async move {
diff --git a/src/storage/examples/upsert_open_loop.rs b/src/storage/examples/upsert_open_loop.rs
index c6d1484185663..3e745bb59f06a 100644
--- a/src/storage/examples/upsert_open_loop.rs
+++ b/src/storage/examples/upsert_open_loop.rs
@@ -115,6 +115,7 @@ use mz_ore::task;
 use mz_persist::indexed::columnar::ColumnarRecords;
 use mz_timely_util::builder_async::{Event as AsyncEvent, OperatorBuilder as AsyncOperatorBuilder};
 use mz_timely_util::probe::{Handle, ProbeNotify};
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Exchange;
 use timely::dataflow::operators::Operator;
 use timely::dataflow::{Scope, Stream};
@@ -577,87 +578,80 @@ async fn run_benchmark(
                 let active_worker = chosen_worker == worker_id;
 
                 let rocks_options = Arc::clone(rocks_options);
-                let _output: Stream<_, ()> = upsert_stream.unary_frontier(
+                let mut num_additions = 0;
+                let mut num_retractions = 0;
+                let mut buffer = Vec::new();
+
+                let mut frontier = Antichain::from_elem(0);
+                let mut max_lag = 0;
+                upsert_stream.sink(
                     Exchange::new(move |_| u64::cast_from(chosen_worker)),
                     &format!("source-{source_id}-counter"),
-                    move |_caps, _info| {
-                        let mut num_additions = 0;
-                        let mut num_retractions = 0;
-                        let mut buffer = Vec::new();
-
-                        let mut frontier = Antichain::from_elem(0);
-                        let mut max_lag = 0;
-
-                        move |input, _output| {
-                            if !active_worker {
-                                return;
-                            }
-                            input.for_each(|_time, data| {
-                                data.swap(&mut buffer);
-                                for (_k, _v, diff) in buffer.drain(..) {
-                                    if diff == 1 {
-                                        num_additions += 1;
-                                    } else if diff == -1 {
-                                        num_retractions += 1;
-                                    } else {
-                                        panic!("unexpected record diff: {diff}");
-                                    }
+                    move |input| {
+                        if !active_worker {
+                            return;
+                        }
+                        input.for_each(|_time, data| {
+                            data.swap(&mut buffer);
+                            for (_k, _v, diff) in buffer.drain(..) {
+                                if diff == 1 {
+                                    num_additions += 1;
+                                } else if diff == -1 {
+                                    num_retractions += 1;
+                                } else {
+                                    panic!("unexpected record diff: {diff}");
                                 }
-                            });
+                            }
+                        });
 
-                            if input.frontier().is_empty() {
-                                assert_eq!(num_records_total, num_additions);
-                                info!(
-                                    "Processing source {source_id} finished \
+                        if input.frontier().is_empty() {
+                            assert_eq!(num_records_total, num_additions);
+                            info!(
+                                "Processing source {source_id} finished \
                                     after {} ms and processed {num_additions} additions and \
                                     {num_retractions} retractions",
-                                    start.elapsed().as_millis(),
-                                );
-                            } else if PartialOrder::less_than(
-                                &frontier.borrow(),
-                                &input.frontier().frontier(),
-                            ) {
-                                frontier = input.frontier().frontier().to_owned();
-                                let data_timestamp = frontier.clone().into_option().unwrap();
-                                let elapsed = start.elapsed();
-
-                                #[allow(clippy::as_conversions)]
-                                {
-                                    let lag = elapsed.as_millis() as u64 - data_timestamp;
-                                    max_lag = std::cmp::max(max_lag, lag);
-                                    let elapsed_seconds = elapsed.as_secs();
-                                    let key_mb_read = (num_additions * args.key_record_size_bytes)
-                                        as f64
-                                        / MIB as f64;
-                                    let mb_read = (num_additions
-                                        * (args.key_record_size_bytes
-                                            + args.value_record_size_bytes))
-                                        as f64
-                                        / MIB as f64;
-                                    let key_throughput = key_mb_read / elapsed_seconds as f64;
-                                    let throughput = mb_read / elapsed_seconds as f64;
-
-                                    let rocksdb_stats = if args.rocksdb_print_stats {
-                                        calculate_rocksdb_stats(
-                                            Some(&rocks_options),
-                                            elapsed_seconds,
-                                        )
+                                start.elapsed().as_millis(),
+                            );
+                        } else if PartialOrder::less_than(
+                            &frontier.borrow(),
+                            &input.frontier().frontier(),
+                        ) {
+                            frontier = input.frontier().frontier().to_owned();
+                            let data_timestamp = frontier.clone().into_option().unwrap();
+                            let elapsed = start.elapsed();
+
+                            #[allow(clippy::as_conversions)]
+                            {
+                                let lag = elapsed.as_millis() as u64 - data_timestamp;
+                                max_lag = std::cmp::max(max_lag, lag);
+                                let elapsed_seconds = elapsed.as_secs();
+                                let key_mb_read = (num_additions * args.key_record_size_bytes)
+                                    as f64
+                                    / MIB as f64;
+                                let mb_read = (num_additions
+                                    * (args.key_record_size_bytes + args.value_record_size_bytes))
+                                    as f64
+                                    / MIB as f64;
+                                let key_throughput = key_mb_read / elapsed_seconds as f64;
+                                let throughput = mb_read / elapsed_seconds as f64;
+
+                                let rocksdb_stats = if args.rocksdb_print_stats {
+                                    calculate_rocksdb_stats(Some(&rocks_options), elapsed_seconds)
                                         .unwrap_or_else(String::new)
-                                    } else {
-                                        "".to_string()
-                                    };
+                                } else {
+                                    "".to_string()
+                                };
 
-                                    info!(
-                                        "After {} ms, source {source_id} has read {num_additions} \
+                                info!(
+                                    "After {} ms, source {source_id} has read {num_additions} \
                                     records (throughput {:.3} MiB/s, key throughput {:.3} MiB/s). \
                                     Max processing lag {max_lag}ms, \
                                     most recent processing lag {lag}ms.{}",
-                                        elapsed.as_millis(),
-                                        throughput,
-                                        key_throughput,
-                                        rocksdb_stats
-                                    );
-                                }
+                                    elapsed.as_millis(),
+                                    throughput,
+                                    key_throughput,
+                                    rocksdb_stats
+                                );
                             }
                         }
                     },
@@ -792,7 +786,7 @@ where
 
     let mut source_op = AsyncOperatorBuilder::new(format!("source-{source_id}"), scope);
 
-    let (mut output, output_stream) = source_op.new_output();
+    let (mut output, output_stream) = source_op.new_output::<CapacityContainerBuilder<_>>();
 
     let _shutdown_button = source_op.build(move |mut capabilities| async move {
         if !active_worker {
diff --git a/src/storage/src/decode.rs b/src/storage/src/decode.rs
index bde3a1504f38c..67b655a6b9207 100644
--- a/src/storage/src/decode.rs
+++ b/src/storage/src/decode.rs
@@ -30,6 +30,7 @@ use mz_timely_util::builder_async::{
     Event as AsyncEvent, OperatorBuilder as AsyncOperatorBuilder, PressOnDropButton,
 };
 use regex::Regex;
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Exchange;
 use timely::dataflow::operators::{Map, Operator};
 use timely::dataflow::{Scope, Stream};
@@ -440,7 +441,7 @@ pub fn render_decode_delimited<G: Scope, FromTime: Timestamp>(
 
     let mut builder = AsyncOperatorBuilder::new(op_name, input.scope());
 
-    let (mut output_handle, output) = builder.new_output();
+    let (mut output_handle, output) = builder.new_output::<CapacityContainerBuilder<_>>();
     let mut input = builder.new_input_for(&input.inner, Exchange::new(dist), &output_handle);
 
     let (_, transient_errors) = builder.build_fallible(move |caps| {
diff --git a/src/storage/src/render/persist_sink.rs b/src/storage/src/render/persist_sink.rs
index 9f00016e60b1d..bc3558a4c2d46 100644
--- a/src/storage/src/render/persist_sink.rs
+++ b/src/storage/src/render/persist_sink.rs
@@ -113,6 +113,7 @@ use mz_timely_util::builder_async::{
     Event, OperatorBuilder as AsyncOperatorBuilder, PressOnDropButton,
 };
 use serde::{Deserialize, Serialize};
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::{Exchange, Pipeline};
 use timely::dataflow::operators::{Broadcast, Capability, CapabilitySet, Inspect};
 use timely::dataflow::{Scope, Stream};
@@ -411,7 +412,7 @@ where
     );
 
     let (mut output, output_stream) = mint_op.new_output();
-    let (mut data_output, data_output_stream) = mint_op.new_output();
+    let (mut data_output, data_output_stream) = mint_op.new_output::<CapacityContainerBuilder<_>>();
 
     // The description and the data-passthrough outputs are both driven by this input, so
     // they use a standard input connection.
@@ -592,7 +593,7 @@ where
     let mut write_op =
         AsyncOperatorBuilder::new(format!("{} write_batches", operator_name), scope.clone());
 
-    let (mut output, output_stream) = write_op.new_output();
+    let (mut output, output_stream) = write_op.new_output::<CapacityContainerBuilder<_>>();
 
     let mut descriptions_input =
         write_op.new_input_for(&batch_descriptions.broadcast(), Pipeline, &output);
@@ -962,7 +963,7 @@ where
         .clone();
 
     // An output whose frontier tracks the last successful compare and append of this operator
-    let (_upper_output, upper_stream) = append_op.new_output();
+    let (_upper_output, upper_stream) = append_op.new_output::<CapacityContainerBuilder<_>>();
 
     // This operator accepts the batch descriptions and tokens that represent
     // written batches. Written batches get appended to persist when we learn
diff --git a/src/storage/src/source/generator/key_value.rs b/src/storage/src/source/generator/key_value.rs
index 54bf26e48703b..02ba64cb2c44f 100644
--- a/src/storage/src/source/generator/key_value.rs
+++ b/src/storage/src/source/generator/key_value.rs
@@ -18,6 +18,7 @@ use mz_storage_types::sources::{MzOffset, SourceTimestamp};
 use mz_timely_util::builder_async::{OperatorBuilder as AsyncOperatorBuilder, PressOnDropButton};
 use rand::rngs::StdRng;
 use rand::{RngCore, SeedableRng};
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::operators::{Concat, ToStream};
 use timely::dataflow::{Scope, Stream};
 use timely::progress::Antichain;
@@ -44,9 +45,9 @@ pub fn render<G: Scope<Timestamp = MzOffset>>(
 
     let mut builder = AsyncOperatorBuilder::new(config.name.clone(), scope.clone());
 
-    let (mut data_output, stream) = builder.new_output();
-    let (_progress_output, progress_stream) = builder.new_output();
-    let (mut stats_output, stats_stream) = builder.new_output();
+    let (mut data_output, stream) = builder.new_output::<CapacityContainerBuilder<_>>();
+    let (_progress_output, progress_stream) = builder.new_output::<CapacityContainerBuilder<_>>();
+    let (mut stats_output, stats_stream) = builder.new_output::<CapacityContainerBuilder<_>>();
 
     let button = builder.build(move |caps| async move {
         let [mut cap, mut progress_cap, stats_cap]: [_; 3] = caps.try_into().unwrap();
diff --git a/src/storage/src/source/kafka.rs b/src/storage/src/source/kafka.rs
index db33fd741f4bb..1b655695bb9a9 100644
--- a/src/storage/src/source/kafka.rs
+++ b/src/storage/src/source/kafka.rs
@@ -41,6 +41,7 @@ use rdkafka::message::{BorrowedMessage, Headers};
 use rdkafka::statistics::Statistics;
 use rdkafka::topic_partition_list::Offset;
 use rdkafka::{ClientContext, Message, TopicPartitionList};
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::operators::Capability;
 use timely::dataflow::{Scope, Stream};
 use timely::progress::Antichain;
@@ -165,7 +166,8 @@ impl SourceRender for KafkaSourceConnection {
         let mut builder = AsyncOperatorBuilder::new(config.name.clone(), scope.clone());
 
         let (mut data_output, stream) = builder.new_output();
-        let (_progress_output, progress_stream) = builder.new_output();
+        let (_progress_output, progress_stream) =
+            builder.new_output::<CapacityContainerBuilder<_>>();
         let (mut health_output, health_stream) = builder.new_output();
         let (mut stats_output, stats_stream) = builder.new_output();
 
diff --git a/src/storage/src/source/mysql.rs b/src/storage/src/source/mysql.rs
index 312303105eb1b..800ab7eb653ef 100644
--- a/src/storage/src/source/mysql.rs
+++ b/src/storage/src/source/mysql.rs
@@ -59,6 +59,7 @@ use std::rc::Rc;
 use differential_dataflow::Collection;
 use mz_storage_types::sources::SourceExport;
 use serde::{Deserialize, Serialize};
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pushers::Tee;
 use timely::dataflow::operators::{CapabilitySet, Concat, Map, ToStream};
 use timely::dataflow::{Scope, Stream};
@@ -312,13 +313,13 @@ async fn return_definite_error(
     outputs: &[usize],
     data_handle: &mut AsyncOutputHandle<
         GtidPartition,
-        Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>,
+        CapacityContainerBuilder<Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>>,
         Tee<GtidPartition, Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>>,
     >,
     data_cap_set: &CapabilitySet<GtidPartition>,
     definite_error_handle: &mut AsyncOutputHandle<
         GtidPartition,
-        Vec<ReplicationError>,
+        CapacityContainerBuilder<Vec<ReplicationError>>,
         Tee<GtidPartition, Vec<ReplicationError>>,
     >,
     definite_error_cap_set: &CapabilitySet<GtidPartition>,
diff --git a/src/storage/src/source/mysql/replication.rs b/src/storage/src/source/mysql/replication.rs
index 96022adc49a87..9a9b7b8c66393 100644
--- a/src/storage/src/source/mysql/replication.rs
+++ b/src/storage/src/source/mysql/replication.rs
@@ -52,6 +52,7 @@ use mysql_async::prelude::Queryable;
 use mysql_async::{BinlogStream, BinlogStreamRequest, GnoInterval, Sid};
 use mz_ore::future::InTask;
 use mz_ssh_util::tunnel_manager::ManagedSshTunnelHandle;
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Exchange;
 use timely::dataflow::operators::{Concat, Map};
 use timely::dataflow::{Scope, Stream};
@@ -115,10 +116,11 @@ pub(crate) fn render<G: Scope<Timestamp = GtidPartition>>(
     let mut builder = AsyncOperatorBuilder::new(op_name, scope);
 
     let repl_reader_id = u64::cast_from(config.responsible_worker(REPL_READER));
-    let (mut data_output, data_stream) = builder.new_output();
-    let (upper_output, upper_stream) = builder.new_output();
+    let (mut data_output, data_stream) = builder.new_output::<CapacityContainerBuilder<_>>();
+    let (upper_output, upper_stream) = builder.new_output::<CapacityContainerBuilder<_>>();
     // Captures DefiniteErrors that affect the entire source, including all subsources
-    let (mut definite_error_handle, definite_errors) = builder.new_output();
+    let (mut definite_error_handle, definite_errors) =
+        builder.new_output::<CapacityContainerBuilder<_>>();
     let mut rewind_input = builder.new_input_for_many(
         rewind_stream,
         Exchange::new(move |_| repl_reader_id),
diff --git a/src/storage/src/source/mysql/replication/context.rs b/src/storage/src/source/mysql/replication/context.rs
index a9330eb48d002..3ec2d000a27ad 100644
--- a/src/storage/src/source/mysql/replication/context.rs
+++ b/src/storage/src/source/mysql/replication/context.rs
@@ -11,6 +11,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use std::pin::Pin;
 
 use mysql_async::BinlogStream;
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pushers::Tee;
 use timely::dataflow::operators::{Capability, CapabilitySet};
 use timely::progress::Antichain;
@@ -37,7 +38,7 @@ pub(super) struct ReplContext<'a> {
     pub(super) ignore_columns: &'a Vec<MySqlColumnRef>,
     pub(super) data_output: &'a mut AsyncOutputHandle<
         GtidPartition,
-        Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>,
+        CapacityContainerBuilder<Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>>,
         Tee<GtidPartition, Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>>,
     >,
     pub(super) data_cap_set: &'a mut CapabilitySet<GtidPartition>,
@@ -58,7 +59,9 @@ impl<'a> ReplContext<'a> {
         ignore_columns: &'a Vec<MySqlColumnRef>,
         data_output: &'a mut AsyncOutputHandle<
             GtidPartition,
-            Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>,
+            CapacityContainerBuilder<
+                Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>,
+            >,
             Tee<GtidPartition, Vec<((usize, Result<Row, DefiniteError>), GtidPartition, i64)>>,
         >,
         data_cap_set: &'a mut CapabilitySet<GtidPartition>,
diff --git a/src/storage/src/source/postgres/replication.rs b/src/storage/src/source/postgres/replication.rs
index 83078f8ea1d19..6752694caf3a0 100644
--- a/src/storage/src/source/postgres/replication.rs
+++ b/src/storage/src/source/postgres/replication.rs
@@ -102,6 +102,7 @@ use postgres_protocol::message::backend::{
     LogicalReplicationMessage, ReplicationMessage, TupleData,
 };
 use serde::{Deserialize, Serialize};
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Exchange;
 use timely::dataflow::channels::pushers::Tee;
 use timely::dataflow::operators::Capability;
@@ -165,11 +166,12 @@ pub(crate) fn render<G: Scope<Timestamp = MzOffset>>(
     let mut builder = AsyncOperatorBuilder::new(op_name, scope.clone());
 
     let slot_reader = u64::cast_from(config.responsible_worker("slot"));
-    let (mut data_output, data_stream) = builder.new_output();
-    let (upper_output, upper_stream) = builder.new_output();
-    let (mut definite_error_handle, definite_errors) = builder.new_output();
+    let (mut data_output, data_stream) = builder.new_output::<CapacityContainerBuilder<_>>();
+    let (upper_output, upper_stream) = builder.new_output::<CapacityContainerBuilder<_>>();
+    let (mut definite_error_handle, definite_errors) =
+        builder.new_output::<CapacityContainerBuilder<_>>();
 
-    let (mut stats_output, stats_stream) = builder.new_output();
+    let (mut stats_output, stats_stream) = builder.new_output::<CapacityContainerBuilder<_>>();
 
     let mut rewind_input = builder.new_input_for_many(
         rewind_stream,
@@ -446,7 +448,7 @@ async fn raw_stream<'a>(
     uppers: impl futures::Stream<Item = Antichain<MzOffset>> + 'a,
     stats_output: &'a mut AsyncOutputHandle<
         MzOffset,
-        Vec<ProgressStatisticsUpdate>,
+        CapacityContainerBuilder<Vec<ProgressStatisticsUpdate>>,
         Tee<MzOffset, Vec<ProgressStatisticsUpdate>>,
     >,
     stats_cap: &'a Capability<MzOffset>,
diff --git a/src/storage/src/source/source_reader_pipeline.rs b/src/storage/src/source/source_reader_pipeline.rs
index 4a21b4cda4fc1..9b1c2fe61a4cd 100644
--- a/src/storage/src/source/source_reader_pipeline.rs
+++ b/src/storage/src/source/source_reader_pipeline.rs
@@ -57,6 +57,7 @@ use mz_timely_util::builder_async::{
 };
 use mz_timely_util::capture::UnboundedTokioCapture;
 use mz_timely_util::operator::StreamExt as _;
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::capture::capture::Capture;
 use timely::dataflow::operators::capture::Event;
@@ -293,14 +294,14 @@ where
     let name = format!("SourceGenericStats({})", source_id);
     let mut builder = AsyncOperatorBuilder::new(name, scope.clone());
 
-    let (mut data_output, data) = builder.new_output();
-    let (progress_output, derived_progress) = builder.new_output();
+    let (mut data_output, data) = builder.new_output::<CapacityContainerBuilder<_>>();
+    let (progress_output, derived_progress) = builder.new_output::<CapacityContainerBuilder<_>>();
     let mut data_input = builder.new_input_for_many(
         &input_data.inner,
         Pipeline,
         [&data_output, &progress_output],
     );
-    let (mut health_output, derived_health) = builder.new_output();
+    let (mut health_output, derived_health) = builder.new_output::<CapacityContainerBuilder<_>>();
 
     builder.build(move |mut caps| async move {
         let health_cap = caps.pop().unwrap();
@@ -453,7 +454,7 @@ where
 
     let operator_name = format!("remap({})", id);
     let mut remap_op = AsyncOperatorBuilder::new(operator_name, scope.clone());
-    let (mut remap_output, remap_stream) = remap_op.new_output();
+    let (mut remap_output, remap_stream) = remap_op.new_output::<CapacityContainerBuilder<_>>();
 
     let button = remap_op.build(move |capabilities| async move {
         if !active_worker {
diff --git a/src/storage/src/upsert.rs b/src/storage/src/upsert.rs
index c718754a182be..c956c9bf868be 100644
--- a/src/storage/src/upsert.rs
+++ b/src/storage/src/upsert.rs
@@ -141,6 +141,7 @@ impl<H: Digest> Hasher for DigestHasher<H> {
 }
 
 use std::convert::Infallible;
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Pipeline;
 
 use self::types::ValueMetadata;
@@ -915,7 +916,7 @@ impl<G: Scope> UpsertErrorEmitter<G>
     for (
         &mut AsyncOutputHandle<
             <G as ScopeParent>::Timestamp,
-            Vec<(OutputIndex, HealthStatusUpdate)>,
+            CapacityContainerBuilder<Vec<(OutputIndex, HealthStatusUpdate)>>,
             Tee<<G as ScopeParent>::Timestamp, Vec<(OutputIndex, HealthStatusUpdate)>>,
         >,
         &Capability<<G as ScopeParent>::Timestamp>,
@@ -932,7 +933,7 @@ async fn process_upsert_state_error<G: Scope>(
     e: anyhow::Error,
     health_output: &mut AsyncOutputHandle<
         <G as ScopeParent>::Timestamp,
-        Vec<(OutputIndex, HealthStatusUpdate)>,
+        CapacityContainerBuilder<Vec<(OutputIndex, HealthStatusUpdate)>>,
         Tee<<G as ScopeParent>::Timestamp, Vec<(OutputIndex, HealthStatusUpdate)>>,
     >,
     health_cap: &Capability<<G as ScopeParent>::Timestamp>,
diff --git a/src/timely-util/src/buffer.rs b/src/timely-util/src/buffer.rs
deleted file mode 100644
index 96b0bf0315cc9..0000000000000
--- a/src/timely-util/src/buffer.rs
+++ /dev/null
@@ -1,165 +0,0 @@
-// Copyright Materialize, Inc. and contributors. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License in the LICENSE file at the
-// root of this repository, or online at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use differential_dataflow::consolidation::consolidate_updates;
-use differential_dataflow::difference::Semigroup;
-use differential_dataflow::Data;
-use timely::communication::Push;
-use timely::dataflow::channels::Bundle;
-use timely::dataflow::operators::generic::OutputHandle;
-use timely::dataflow::operators::{Capability, InputCapability};
-use timely::progress::Timestamp;
-
-/// A buffer that consolidates updates
-///
-/// The buffer implements a wrapper around [OutputHandle] consolidating elements pushed to it. It is
-/// backed by a capacity-limited buffer, which means that compaction only occurs within the
-/// dimensions of the buffer, i.e. the number of unique keys is less than half of the buffer's
-/// capacity.
-///
-/// A cap is retained whenever the current time changes to be able to flush on drop or when the time
-/// changes again.
-///
-/// The buffer is filled with updates until it reaches its capacity. At this point, the updates are
-/// consolidated to free up space. This process repeats until the consolidation recovered less than
-/// half of the buffer's capacity, at which point the buffer will be shipped.
-///
-/// The buffer retains a capability to send data on flush. It will flush all data once dropped, if
-/// time changes, or if the buffer capacity is reached.
-pub struct ConsolidateBuffer<'a, 'b, T, D: Data, R: Semigroup, P>
-where
-    P: Push<Bundle<T, Vec<(D, T, R)>>> + 'a,
-    T: Data + Timestamp + 'a,
-    D: 'a,
-{
-    // a buffer for records, to send at self.cap
-    // Invariant: Buffer only contains data if cap is Some.
-    buffer: Vec<(D, T, R)>,
-    output_handle: &'b mut OutputHandle<'a, T, (D, T, R), P>,
-    cap: Option<Capability<T>>,
-    port: usize,
-    previous_len: usize,
-}
-
-impl<'a, 'b, T, D: Data, R: Semigroup, P> ConsolidateBuffer<'a, 'b, T, D, R, P>
-where
-    T: Data + Timestamp + 'a,
-    P: Push<Bundle<T, Vec<(D, T, R)>>> + 'a,
-{
-    /// Create a new [ConsolidateBuffer], wrapping the provided session.
-    ///
-    /// * `output_handle`: The output to send data to.
-    /// * 'port': The output port to retain capabilities for.
-    pub fn new(output_handle: &'b mut OutputHandle<'a, T, (D, T, R), P>, port: usize) -> Self {
-        Self {
-            output_handle,
-            port,
-            cap: None,
-            buffer: Vec::with_capacity(::timely::container::buffer::default_capacity::<(D, T, R)>()),
-            previous_len: 0,
-        }
-    }
-
-    #[inline]
-    /// Provides an iterator of elements to the buffer
-    pub fn give_iterator<I: Iterator<Item = (D, T, R)>>(
-        &mut self,
-        cap: &InputCapability<T>,
-        iter: I,
-    ) {
-        for item in iter {
-            self.give(cap, item);
-        }
-    }
-
-    /// Give an element to the buffer
-    pub fn give(&mut self, cap: &InputCapability<T>, data: (D, T, R)) {
-        // Retain a cap for the current time, which will be used on flush.
-        if self.cap.as_ref().map_or(true, |t| t.time() != cap.time()) {
-            // Flush on capability change
-            self.flush();
-            // Retain capability for the specified output port.
-            self.cap = Some(cap.delayed_for_output(cap.time(), self.port));
-        }
-        self.give_internal(data);
-    }
-
-    /// Give an element to the buffer, using a pre-fabricated capability. Note that the capability
-    /// must be valid for the associated output.
-    pub fn give_at(&mut self, cap: &Capability<T>, data: (D, T, R)) {
-        // Retain a cap for the current time, which will be used on flush.
-        if self.cap.as_ref().map_or(true, |t| t.time() != cap.time()) {
-            // Flush on capability change
-            self.flush();
-            // Retain capability.
-            self.cap = Some(cap.clone());
-        }
-        self.give_internal(data);
-    }
-
-    /// Give an element and possibly flush the buffer. Note that this needs to have access
-    /// to a capability, which the public functions ensure.
-    fn give_internal(&mut self, data: (D, T, R)) {
-        self.buffer.push(data);
-
-        // Limit, if possible, the lifetime of the allocations for data
-        // and consolidate smaller buffers if we're in the lucky case
-        // of a small domain for D
-        if self.buffer.len() >= 2 * self.previous_len {
-            // Consolidate while the consolidation frees at least half the buffer
-            consolidate_updates(&mut self.buffer);
-            if self.buffer.len() > self.buffer.capacity() / 2 {
-                self.flush();
-            } else {
-                self.previous_len = self.buffer.len();
-            }
-            // At this point, it is an invariant across give calls that self.previous_len
-            // will be in the interval [0, self.buffer.capacity() / 2]. So, we will enter
-            // this if-statement block again when self.buffer.len() == self.buffer.capacity()
-            // or earlier. If consolidation is not effective to keep self.buffer.len()
-            // below half capacity, then flushing when more than half-full will
-            // maintain the invariant.
-        }
-    }
-
-    /// Flush the internal buffer to the underlying session
-    pub fn flush(&mut self) {
-        if let Some(cap) = &self.cap {
-            self.output_handle.session(cap).give_vec(&mut self.buffer);
-
-            // Ensure that the capacity is at least equal to the default in case
-            // it was reduced by give_vec. Note that we cannot rely here on give_vec
-            // returning us a buffer with zero capacity.
-            if self.buffer.capacity() < ::timely::container::buffer::default_capacity::<(D, T, R)>()
-            {
-                let to_reserve = ::timely::container::buffer::default_capacity::<(D, T, R)>()
-                    - self.buffer.capacity();
-                self.buffer.reserve_exact(to_reserve);
-            }
-            self.previous_len = 0;
-        }
-    }
-}
-
-impl<'a, 'b, T, D: Data, R: Semigroup, P> Drop for ConsolidateBuffer<'a, 'b, T, D, R, P>
-where
-    P: Push<Bundle<T, Vec<(D, T, R)>>> + 'a,
-    T: Data + Timestamp + 'a,
-    D: 'a,
-{
-    fn drop(&mut self) {
-        self.flush();
-    }
-}
diff --git a/src/timely-util/src/builder_async.rs b/src/timely-util/src/builder_async.rs
index 6622f8ed4a359..f66f4bca67cd0 100644
--- a/src/timely-util/src/builder_async.rs
+++ b/src/timely-util/src/builder_async.rs
@@ -20,6 +20,7 @@ use std::task::{Context, Poll, Waker};
 
 use futures_util::task::ArcWake;
 use timely::communication::{Message, Pull, Push};
+use timely::container::{CapacityContainerBuilder, ContainerBuilder};
 use timely::dataflow::channels::pact::ParallelizationContract;
 use timely::dataflow::channels::pushers::buffer::Session;
 use timely::dataflow::channels::pushers::counter::Counter as PushCounter;
@@ -191,58 +192,62 @@ pub enum Event<T: Timestamp, C, D> {
 
 // TODO: delete and use CapabilityTrait instead once TimelyDataflow/timely-dataflow#512 gets merged
 pub trait CapabilityTrait<T: Timestamp> {
-    fn session<'a, D, P>(
+    fn session<'a, CB, P>(
         &'a self,
-        handle: &'a mut OutputHandleCore<'_, T, D, P>,
-    ) -> Session<'a, T, D, PushCounter<T, D, P>>
+        handle: &'a mut OutputHandleCore<'_, T, CB, P>,
+    ) -> Session<'a, T, CB, PushCounter<T, CB::Container, P>>
     where
-        D: Container,
-        P: Push<Bundle<T, D>>;
+        CB: ContainerBuilder,
+        P: Push<Bundle<T, CB::Container>>;
 }
 
 impl<T: Timestamp> CapabilityTrait<T> for InputCapability<T> {
     #[inline]
-    fn session<'a, D, P>(
+    fn session<'a, CB, P>(
         &'a self,
-        handle: &'a mut OutputHandleCore<'_, T, D, P>,
-    ) -> Session<'a, T, D, PushCounter<T, D, P>>
+        handle: &'a mut OutputHandleCore<'_, T, CB, P>,
+    ) -> Session<'a, T, CB, PushCounter<T, CB::Container, P>>
     where
-        D: Container,
-        P: Push<Bundle<T, D>>,
+        CB: ContainerBuilder,
+        P: Push<Bundle<T, CB::Container>>,
     {
-        handle.session(self)
+        handle.session_with_builder(self)
     }
 }
 
 impl<T: Timestamp> CapabilityTrait<T> for Capability<T> {
     #[inline]
-    fn session<'a, D, P>(
+    fn session<'a, CB, P>(
         &'a self,
-        handle: &'a mut OutputHandleCore<'_, T, D, P>,
-    ) -> Session<'a, T, D, PushCounter<T, D, P>>
+        handle: &'a mut OutputHandleCore<'_, T, CB, P>,
+    ) -> Session<'a, T, CB, PushCounter<T, CB::Container, P>>
     where
-        D: Container,
-        P: Push<Bundle<T, D>>,
+        CB: ContainerBuilder,
+        P: Push<Bundle<T, CB::Container>>,
     {
-        handle.session(self)
+        handle.session_with_builder(self)
     }
 }
 
-pub struct AsyncOutputHandle<T: Timestamp, D: Container, P: Push<Bundle<T, D>> + 'static> {
+pub struct AsyncOutputHandle<
+    T: Timestamp,
+    CB: ContainerBuilder,
+    P: Push<Bundle<T, CB::Container>> + 'static,
+> {
     // The field order is important here as the handle is borrowing from the wrapper. See also the
     // safety argument in the constructor
-    handle: Rc<RefCell<OutputHandleCore<'static, T, D, P>>>,
-    wrapper: Rc<Pin<Box<OutputWrapper<T, D, P>>>>,
+    handle: Rc<RefCell<OutputHandleCore<'static, T, CB, P>>>,
+    wrapper: Rc<Pin<Box<OutputWrapper<T, CB, P>>>>,
     index: usize,
 }
 
-impl<T, D, P> AsyncOutputHandle<T, D, P>
+impl<T, CB, P> AsyncOutputHandle<T, CB, P>
 where
     T: Timestamp,
-    D: Container,
-    P: Push<Bundle<T, D>> + 'static,
+    CB: ContainerBuilder,
+    P: Push<Bundle<T, CB::Container>> + 'static,
 {
-    fn new(wrapper: OutputWrapper<T, D, P>, index: usize) -> Self {
+    fn new(wrapper: OutputWrapper<T, CB, P>, index: usize) -> Self {
         let mut wrapper = Rc::new(Box::pin(wrapper));
         // SAFETY:
         // get_unchecked_mut is safe because we are not moving the wrapper
@@ -257,7 +262,7 @@ where
                 .as_mut()
                 .get_unchecked_mut()
                 .activate();
-            std::mem::transmute::<OutputHandleCore<'_, T, D, P>, OutputHandleCore<'static, T, D, P>>(
+            std::mem::transmute::<OutputHandleCore<'_, T, CB, P>, OutputHandleCore<'static, T, CB, P>>(
                 handle,
             )
         };
@@ -270,7 +275,10 @@ where
 
     #[allow(clippy::unused_async)]
     #[inline]
-    pub async fn give_container<C: CapabilityTrait<T>>(&mut self, cap: &C, container: &mut D) {
+    pub async fn give_container<C>(&mut self, cap: &C, container: &mut CB::Container)
+    where
+        C: CapabilityTrait<T>,
+    {
         let mut handle = self.handle.borrow_mut();
         cap.session(&mut handle).give_container(container);
     }
@@ -280,7 +288,7 @@ where
     }
 }
 
-impl<'a, T, D, P> AsyncOutputHandle<T, Vec<D>, P>
+impl<'a, T, D, P> AsyncOutputHandle<T, CapacityContainerBuilder<Vec<D>>, P>
 where
     T: Timestamp,
     D: Data,
@@ -293,8 +301,8 @@ where
     }
 }
 
-impl<T: Timestamp, D: Container, P: Push<Bundle<T, D>> + 'static> Clone
-    for AsyncOutputHandle<T, D, P>
+impl<T: Timestamp, CB: ContainerBuilder, P: Push<Bundle<T, CB::Container>> + 'static> Clone
+    for AsyncOutputHandle<T, CB, P>
 {
     fn clone(&self) -> Self {
         Self {
@@ -378,7 +386,9 @@ pub trait OutputIndex {
     fn index(&self) -> usize;
 }
 
-impl<T: Timestamp, D: Container> OutputIndex for AsyncOutputHandle<T, D, Tee<T, D>> {
+impl<T: Timestamp, CB: ContainerBuilder> OutputIndex
+    for AsyncOutputHandle<T, CB, Tee<T, CB::Container>>
+{
     fn index(&self) -> usize {
         self.index
     }
@@ -491,11 +501,11 @@ impl<G: Scope> OperatorBuilder<G> {
     }
 
     /// Adds a new output, returning the output handle and stream.
-    pub fn new_output<D: Container>(
+    pub fn new_output<CB: ContainerBuilder>(
         &mut self,
     ) -> (
-        AsyncOutputHandle<G::Timestamp, D, Tee<G::Timestamp, D>>,
-        StreamCore<G, D>,
+        AsyncOutputHandle<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>,
+        StreamCore<G, CB::Container>,
     ) {
         let index = self.builder.shape().outputs();
 
@@ -791,7 +801,8 @@ mod test {
             let index = worker.index();
             let tokens = worker.dataflow::<u64, _, _>(move |scope| {
                 let mut producer = OperatorBuilder::new("producer".to_string(), scope.clone());
-                let (_output, output_stream) = producer.new_output::<Vec<usize>>();
+                let (_output, output_stream) =
+                    producer.new_output::<CapacityContainerBuilder<Vec<usize>>>();
                 let producer_button = producer.build(move |mut capabilities| async move {
                     let mut cap = capabilities.pop().unwrap();
                     if index != 0 {
diff --git a/src/timely-util/src/lib.rs b/src/timely-util/src/lib.rs
index a5e047d3e1e07..c0264bb629159 100644
--- a/src/timely-util/src/lib.rs
+++ b/src/timely-util/src/lib.rs
@@ -17,7 +17,6 @@
 
 pub mod activator;
 pub mod antichain;
-pub mod buffer;
 pub mod builder_async;
 pub mod capture;
 pub mod event;
diff --git a/src/timely-util/src/operator.rs b/src/timely-util/src/operator.rs
index ae33d1c5f91b1..a0e30ecdc7204 100644
--- a/src/timely-util/src/operator.rs
+++ b/src/timely-util/src/operator.rs
@@ -13,21 +13,22 @@ use std::future::Future;
 use std::hash::{BuildHasher, Hash, Hasher};
 use std::rc::Weak;
 
+use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::difference::{Multiply, Semigroup};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::trace::{Batcher, Builder};
 use differential_dataflow::{AsCollection, Collection, Hashable};
+use timely::container::{CapacityContainerBuilder, ContainerBuilder};
 use timely::dataflow::channels::pact::{Exchange, ParallelizationContract, Pipeline};
 use timely::dataflow::channels::pushers::Tee;
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder as OperatorBuilderRc;
 use timely::dataflow::operators::generic::operator::{self, Operator};
-use timely::dataflow::operators::generic::{InputHandle, OperatorInfo, OutputHandle};
+use timely::dataflow::operators::generic::{InputHandle, OperatorInfo, OutputHandleCore};
 use timely::dataflow::operators::Capability;
 use timely::dataflow::{Scope, Stream, StreamCore};
 use timely::progress::{Antichain, Timestamp};
-use timely::{Container, Data, ExchangeData, PartialOrder};
+use timely::{Data, ExchangeData, PartialOrder};
 
-use crate::buffer::ConsolidateBuffer;
 use crate::builder_async::{
     AsyncInputHandle, AsyncOutputHandle, ConnectedToOne, Disconnected,
     OperatorBuilder as OperatorBuilderAsync,
@@ -49,23 +50,23 @@ where
     /// streams, where the first output stream represents successful
     /// computations and the second output stream represents failed
     /// computations.
-    fn unary_fallible<D2, E, B, P>(
+    fn unary_fallible<DCB, ECB, B, P>(
         &self,
         pact: P,
         name: &str,
         constructor: B,
-    ) -> (Stream<G, D2>, Stream<G, E>)
+    ) -> (StreamCore<G, DCB::Container>, StreamCore<G, ECB::Container>)
     where
-        D2: Data,
-        E: Data,
+        DCB: ContainerBuilder,
+        ECB: ContainerBuilder,
         B: FnOnce(
             Capability<G::Timestamp>,
             OperatorInfo,
         ) -> Box<
             dyn FnMut(
                     &mut InputHandle<G::Timestamp, D1, P::Puller>,
-                    &mut OutputHandle<G::Timestamp, D2, Tee<G::Timestamp, Vec<D2>>>,
-                    &mut OutputHandle<G::Timestamp, E, Tee<G::Timestamp, Vec<E>>>,
+                    &mut OutputHandleCore<G::Timestamp, DCB, Tee<G::Timestamp, DCB::Container>>,
+                    &mut OutputHandleCore<G::Timestamp, ECB, Tee<G::Timestamp, ECB::Container>>,
                 ) + 'static,
         >,
         P: ParallelizationContract<G::Timestamp, Vec<D1>>;
@@ -80,7 +81,11 @@ where
             Capability<G::Timestamp>,
             OperatorInfo,
             AsyncInputHandle<G::Timestamp, Vec<D1>, ConnectedToOne>,
-            AsyncOutputHandle<G::Timestamp, Vec<D2>, Tee<G::Timestamp, Vec<D2>>>,
+            AsyncOutputHandle<
+                G::Timestamp,
+                CapacityContainerBuilder<Vec<D2>>,
+                Tee<G::Timestamp, Vec<D2>>,
+            >,
         ) -> BFut,
         BFut: Future + 'static,
         P: ParallelizationContract<G::Timestamp, Vec<D1>>;
@@ -104,7 +109,11 @@ where
             OperatorInfo,
             AsyncInputHandle<G::Timestamp, Vec<D1>, ConnectedToOne>,
             AsyncInputHandle<G::Timestamp, Vec<D2>, ConnectedToOne>,
-            AsyncOutputHandle<G::Timestamp, Vec<D3>, Tee<G::Timestamp, Vec<D3>>>,
+            AsyncOutputHandle<
+                G::Timestamp,
+                CapacityContainerBuilder<Vec<D3>>,
+                Tee<G::Timestamp, Vec<D3>>,
+            >,
         ) -> BFut,
         BFut: Future + 'static,
         P1: ParallelizationContract<G::Timestamp, Vec<D1>>,
@@ -257,23 +266,23 @@ where
     D1: Data,
     G: Scope,
 {
-    fn unary_fallible<D2, E, B, P>(
+    fn unary_fallible<DCB, ECB, B, P>(
         &self,
         pact: P,
         name: &str,
         constructor: B,
-    ) -> (Stream<G, D2>, Stream<G, E>)
+    ) -> (StreamCore<G, DCB::Container>, StreamCore<G, ECB::Container>)
     where
-        D2: Data,
-        E: Data,
+        DCB: ContainerBuilder,
+        ECB: ContainerBuilder,
         B: FnOnce(
             Capability<G::Timestamp>,
             OperatorInfo,
         ) -> Box<
             dyn FnMut(
                     &mut InputHandle<G::Timestamp, D1, P::Puller>,
-                    &mut OutputHandle<G::Timestamp, D2, Tee<G::Timestamp, Vec<D2>>>,
-                    &mut OutputHandle<G::Timestamp, E, Tee<G::Timestamp, Vec<E>>>,
+                    &mut OutputHandleCore<G::Timestamp, DCB, Tee<G::Timestamp, DCB::Container>>,
+                    &mut OutputHandleCore<G::Timestamp, ECB, Tee<G::Timestamp, ECB::Container>>,
                 ) + 'static,
         >,
         P: ParallelizationContract<G::Timestamp, Vec<D1>>,
@@ -308,7 +317,11 @@ where
             Capability<G::Timestamp>,
             OperatorInfo,
             AsyncInputHandle<G::Timestamp, Vec<D1>, ConnectedToOne>,
-            AsyncOutputHandle<G::Timestamp, Vec<D2>, Tee<G::Timestamp, Vec<D2>>>,
+            AsyncOutputHandle<
+                G::Timestamp,
+                CapacityContainerBuilder<Vec<D2>>,
+                Tee<G::Timestamp, Vec<D2>>,
+            >,
         ) -> BFut,
         BFut: Future + 'static,
         P: ParallelizationContract<G::Timestamp, Vec<D1>>,
@@ -344,7 +357,11 @@ where
             OperatorInfo,
             AsyncInputHandle<G::Timestamp, Vec<D1>, ConnectedToOne>,
             AsyncInputHandle<G::Timestamp, Vec<D2>, ConnectedToOne>,
-            AsyncOutputHandle<G::Timestamp, Vec<D3>, Tee<G::Timestamp, Vec<D3>>>,
+            AsyncOutputHandle<
+                G::Timestamp,
+                CapacityContainerBuilder<Vec<D3>>,
+                Tee<G::Timestamp, Vec<D3>>,
+            >,
         ) -> BFut,
         BFut: Future + 'static,
         P1: ParallelizationContract<G::Timestamp, Vec<D1>>,
@@ -457,7 +474,7 @@ where
             move |input, output| {
                 input.for_each(|time, data| {
                     data.swap(&mut vector);
-                    output.session(&time).give_vec(&mut vector);
+                    output.session(&time).give_container(&mut vector);
                 });
             }
         })
@@ -504,22 +521,24 @@ where
         G::Timestamp: Lattice,
     {
         self.inner
-            .unary(Pipeline, "ExplodeOne", move |_, _| {
-                let mut buffer = Vec::new();
-                move |input, output| {
-                    let mut out = ConsolidateBuffer::new(output, 0);
-                    input.for_each(|time, data| {
-                        data.swap(&mut buffer);
-                        out.give_iterator(
-                            &time,
-                            buffer.drain(..).map(|(x, t, d)| {
-                                let (x, d2) = logic(x);
-                                (x, t, d2.multiply(&d))
-                            }),
-                        );
-                    });
-                }
-            })
+            .unary::<ConsolidatingContainerBuilder<_>, _, _, _>(
+                Pipeline,
+                "ExplodeOne",
+                move |_, _| {
+                    let mut buffer = Vec::new();
+                    move |input, output| {
+                        input.for_each(|time, data| {
+                            data.swap(&mut buffer);
+                            output
+                                .session_with_builder(&time)
+                                .give_iterator(buffer.drain(..).map(|(x, t, d)| {
+                                    let (x, d2) = logic(x);
+                                    (x, t, d2.multiply(&d))
+                                }));
+                        });
+                    }
+                },
+            )
             .as_collection()
     }
 
@@ -629,17 +648,17 @@ where
 /// The source is defined by a name, and a constructor which takes a default capability and an
 /// output handle to a future. The future is then repeatedly scheduled, and is expected to
 /// eventually send data and downgrade and release capabilities.
-pub fn source_async<G: Scope, C, B, BFut>(
+pub fn source_async<G: Scope, CB, B, BFut>(
     scope: &G,
     name: String,
     constructor: B,
-) -> StreamCore<G, C>
+) -> StreamCore<G, CB::Container>
 where
-    C: Container,
+    CB: ContainerBuilder,
     B: FnOnce(
         Capability<G::Timestamp>,
         OperatorInfo,
-        AsyncOutputHandle<G::Timestamp, C, Tee<G::Timestamp, C>>,
+        AsyncOutputHandle<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>,
     ) -> BFut,
     BFut: Future + 'static,
 {
@@ -699,7 +718,7 @@ where
             move |input, output| {
                 input.for_each(|cap, data| {
                     capabilities.insert(cap.retain());
-                    batcher.push_batch(data);
+                    batcher.push_container(data);
                 });
 
                 if prev_frontier.borrow() != input.frontier().frontier() {
diff --git a/src/timely-util/src/probe.rs b/src/timely-util/src/probe.rs
index 2d959258dbb7f..0cae9e64e05ee 100644
--- a/src/timely-util/src/probe.rs
+++ b/src/timely-util/src/probe.rs
@@ -17,6 +17,7 @@ use std::cell::RefCell;
 use std::convert::Infallible;
 use std::rc::Rc;
 
+use timely::container::CapacityContainerBuilder;
 use timely::dataflow::operators::{CapabilitySet, InspectCore};
 use timely::dataflow::{Scope, Stream, StreamCore};
 use timely::progress::frontier::{Antichain, AntichainRef, MutableAntichain};
@@ -163,7 +164,7 @@ where
     T: Timestamp,
 {
     let mut builder = AsyncOperatorBuilder::new(name, scope);
-    let (_output, output_stream) = builder.new_output();
+    let (_output, output_stream) = builder.new_output::<CapacityContainerBuilder<_>>();
 
     builder.build(move |capabilities| async move {
         let mut cap_set = CapabilitySet::from(capabilities);
diff --git a/src/timely-util/src/replay.rs b/src/timely-util/src/replay.rs
index c6424180ef29b..c21da5c4413af 100644
--- a/src/timely-util/src/replay.rs
+++ b/src/timely-util/src/replay.rs
@@ -125,7 +125,7 @@ where
                             }
                             Event::Messages(time, data) => {
                                 buffer.extend_from_slice(data);
-                                output.session(time).give_vec(&mut buffer);
+                                output.session(time).give_container(&mut buffer);
                             }
                         }
                     }