diff --git a/Cargo.lock b/Cargo.lock
index 2657322314fa2..afb4089d16dac 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1436,9 +1436,9 @@ dependencies = [
 
 [[package]]
 name = "bytemuck"
-version = "1.20.0"
+version = "1.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b37c88a63ffd85d15b406896cc343916d7cf57838a847b3a6f2ca5d39a5695a"
+checksum = "ef657dfab802224e671f5818e9a4935f9b1957ed18e58292690cc39e7a4092a3"
 
 [[package]]
 name = "byteorder"
@@ -1754,9 +1754,9 @@ checksum = "5b63caa9aa9397e2d9480a9b13673856c78d8ac123288526c37d7839f2a86990"
 
 [[package]]
 name = "columnar"
-version = "0.2.0"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d754936b0b004c01c338b3d81c39e0a81d2ead5b6ee9fa64bfa140e6c430b80d"
+checksum = "d58a4c12223e2d2140bbf4be9fb38b3a612804230c91388dfa4e56a8a6464bf3"
 dependencies = [
  "bincode",
  "bytemuck",
@@ -1768,9 +1768,9 @@ dependencies = [
 
 [[package]]
 name = "columnar_derive"
-version = "0.2.0"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "174af3249fb00e9845597cb3a8259f05ff62b4060650b6af6adb4e05df277f7a"
+checksum = "087b1ac5c4ecad28348b6a9957e3dbd44ac7d041d267370acfdbbfa66b514c7d"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -5113,6 +5113,7 @@ dependencies = [
  "anyhow",
  "async-stream",
  "bytesize",
+ "columnar",
  "core_affinity",
  "crossbeam-channel",
  "dec",
@@ -5199,6 +5200,7 @@ dependencies = [
 name = "mz-compute-types"
 version = "0.0.0"
 dependencies = [
+ "columnar",
  "columnation",
  "differential-dataflow",
  "itertools 0.12.1",
@@ -7309,6 +7311,8 @@ version = "0.0.0"
 dependencies = [
  "ahash",
  "bincode",
+ "bytemuck",
+ "columnar",
  "columnation",
  "differential-dataflow",
  "either",
@@ -10369,9 +10373,9 @@ dependencies = [
 
 [[package]]
 name = "timely"
-version = "0.16.0"
+version = "0.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b134806db44c8452ba2a9b18e6112f3d5e0a0185c6e58524269a08ca2891149c"
+checksum = "0ace21eb2a22c1b80b0b9b5be0627eb0f95a128a5751e866aefc90a85e3007d3"
 dependencies = [
  "bincode",
  "byteorder",
@@ -10394,9 +10398,9 @@ checksum = "99223f9594ab7d4dd36b55b1d7eb9bd2cd205f4304b5cc5381d5cdd417ec06f2"
 
 [[package]]
 name = "timely_communication"
-version = "0.16.1"
+version = "0.16.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f00d5c7f7bbb132a32220890240e5d951fc9582f82fb94452c011a8033c39f7a"
+checksum = "de7b73d69bd229dae4ea989beb7e48e652c0a576774134e6fed59e29f2a6f52c"
 dependencies = [
  "byteorder",
  "columnar",
@@ -10410,9 +10414,9 @@ dependencies = [
 
 [[package]]
 name = "timely_container"
-version = "0.13.1"
+version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92b763064e76ba4f650dcdb035d82bcaad09986971fab002276e0b4cb10b3aa5"
+checksum = "b8bb051808b458a14146900968293da6166b1e6df5dab7c4d1d163f4734b7431"
 dependencies = [
  "columnation",
  "flatcontainer",
@@ -10421,9 +10425,9 @@ dependencies = [
 
 [[package]]
 name = "timely_logging"
-version = "0.13.1"
+version = "0.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d8300263e23bebac4ec9fbbeee65954a25402037609b7716d20191ed35829b14"
+checksum = "e6760517ac4ffa29d00841b0fccf513ece5da896f159045a43970fdb228726d9"
 dependencies = [
  "timely_container",
 ]
diff --git a/src/compute-types/Cargo.toml b/src/compute-types/Cargo.toml
index a26797e591820..a82db2a67b378 100644
--- a/src/compute-types/Cargo.toml
+++ b/src/compute-types/Cargo.toml
@@ -10,6 +10,7 @@ publish = false
 workspace = true
 
 [dependencies]
+columnar = "0.2.2"
 columnation = "0.1.0"
 differential-dataflow = "0.13.3"
 itertools = "0.12.1"
diff --git a/src/compute-types/src/plan.rs b/src/compute-types/src/plan.rs
index 88260444c32f7..abd1eee3afb17 100644
--- a/src/compute-types/src/plan.rs
+++ b/src/compute-types/src/plan.rs
@@ -14,6 +14,7 @@
 use std::collections::{BTreeMap, BTreeSet};
 use std::num::NonZeroU64;
 
+use columnar::Columnar;
 use mz_expr::{
     CollectionPlan, EvalError, Id, LetRecLimit, LocalId, MapFilterProject, MirScalarExpr,
     OptimizedMirRelationExpr, TableFunc,
@@ -161,14 +162,12 @@ impl AvailableCollections {
 }
 
 /// An identifier for an LIR node.
-///
-/// LirIds start at 1, not 0, which let's us get a better struct packing in `ComputeEvent::LirMapping`.
-#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
-pub struct LirId(NonZeroU64);
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize, Columnar)]
+pub struct LirId(u64);
 
 impl LirId {
     fn as_u64(&self) -> u64 {
-        self.0.into()
+        self.0
     }
 }
 
@@ -186,11 +185,11 @@ impl std::fmt::Display for LirId {
 
 impl RustType<u64> for LirId {
     fn into_proto(&self) -> u64 {
-        u64::from(self.0)
+        self.0
     }
 
     fn from_proto(proto: u64) -> Result<Self, mz_proto::TryFromProtoError> {
-        Ok(Self(proto.try_into()?))
+        Ok(Self(proto))
     }
 }
 
@@ -513,7 +512,7 @@ impl Arbitrary for LirId {
     type Parameters = ();
 
     fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
-        let lir_id = NonZeroU64::arbitrary();
+        let lir_id = u64::arbitrary();
         lir_id.prop_map(LirId).boxed()
     }
 }
@@ -1132,16 +1131,6 @@ mod tests {
 
     use super::*;
 
-    #[mz_ore::test]
-    fn test_option_lirid_fits_in_usize() {
-        let option_lirid_size = std::mem::size_of::<Option<LirId>>();
-        let usize_size = std::mem::size_of::<usize>();
-        assert!(
-            option_lirid_size <= usize_size,
-            "Option<LirId> (size {option_lirid_size}) should fit in usize (size {usize_size})"
-        );
-    }
-
     proptest! {
         #![proptest_config(ProptestConfig::with_cases(10))]
         #[mz_ore::test]
diff --git a/src/compute-types/src/plan/lowering.rs b/src/compute-types/src/plan/lowering.rs
index 2fa49f3585d59..7b9d2d1a693c3 100644
--- a/src/compute-types/src/plan/lowering.rs
+++ b/src/compute-types/src/plan/lowering.rs
@@ -43,7 +43,7 @@ impl Context {
     pub fn new(debug_name: String, features: &OptimizerFeatures) -> Self {
         Self {
             arrangements: Default::default(),
-            next_lir_id: LirId(std::num::NonZero::<u64>::MIN),
+            next_lir_id: LirId(1),
             debug_info: LirDebugInfo {
                 debug_name,
                 id: GlobalId::Transient(0),
diff --git a/src/compute/Cargo.toml b/src/compute/Cargo.toml
index 5f459f9edeb53..3f9ba4d140a3d 100644
--- a/src/compute/Cargo.toml
+++ b/src/compute/Cargo.toml
@@ -13,6 +13,7 @@ workspace = true
 anyhow = "1.0.66"
 async-stream = "0.3.3"
 bytesize = "1.1.0"
+columnar = "0.2.2"
 crossbeam-channel = "0.5.8"
 dec = { version = "0.4.8", features = ["serde"] }
 differential-dataflow = "0.13.3"
diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs
index 70aa3b9f419dc..1e419838c64a2 100644
--- a/src/compute/src/compute_state.rs
+++ b/src/compute/src/compute_state.rs
@@ -597,7 +597,7 @@ impl<'a, A: Allocate + 'static> ActiveComputeState<'a, A> {
 
         // Log the receipt of the peek.
         if let Some(logger) = self.compute_state.compute_logger.as_mut() {
-            logger.log(pending.as_log_event(true));
+            logger.log(&pending.as_log_event(true));
         }
 
         self.process_peek(&mut Antichain::new(), pending);
@@ -891,7 +891,7 @@ impl<'a, A: Allocate + 'static> ActiveComputeState<'a, A> {
 
         // Log responding to the peek request.
         if let Some(logger) = self.compute_state.compute_logger.as_mut() {
-            logger.log(log_event);
+            logger.log(&log_event);
         }
     }
 
diff --git a/src/compute/src/extensions/arrange.rs b/src/compute/src/extensions/arrange.rs
index 6409d44244d7d..555bc69a0271e 100644
--- a/src/compute/src/extensions/arrange.rs
+++ b/src/compute/src/extensions/arrange.rs
@@ -263,10 +263,10 @@ where
         .stream
         .unary(Pipeline, "ArrangementSize", |_cap, info| {
             let address = info.address;
-            logger.log(ComputeEvent::ArrangementHeapSizeOperator(
+            logger.log(&ComputeEvent::ArrangementHeapSizeOperator(
                 ArrangementHeapSizeOperator {
                     operator_id,
-                    address,
+                    address: address.to_vec(),
                 },
             ));
             move |input, output| {
@@ -281,7 +281,7 @@ where
 
                 let size = size.try_into().expect("must fit");
                 if size != old_size {
-                    logger.log(ComputeEvent::ArrangementHeapSize(ArrangementHeapSize {
+                    logger.log(&ComputeEvent::ArrangementHeapSize(ArrangementHeapSize {
                         operator_id,
                         delta_size: size - old_size,
                     }));
@@ -289,7 +289,7 @@ where
 
                 let capacity = capacity.try_into().expect("must fit");
                 if capacity != old_capacity {
-                    logger.log(ComputeEvent::ArrangementHeapCapacity(
+                    logger.log(&ComputeEvent::ArrangementHeapCapacity(
                         ArrangementHeapCapacity {
                             operator_id,
                             delta_capacity: capacity - old_capacity,
@@ -299,7 +299,7 @@ where
 
                 let allocations = allocations.try_into().expect("must fit");
                 if allocations != old_allocations {
-                    logger.log(ComputeEvent::ArrangementHeapAllocations(
+                    logger.log(&ComputeEvent::ArrangementHeapAllocations(
                         ArrangementHeapAllocations {
                             operator_id,
                             delta_allocations: allocations - old_allocations,
diff --git a/src/compute/src/logging.rs b/src/compute/src/logging.rs
index 1aafb9847697e..17c9d04279a49 100644
--- a/src/compute/src/logging.rs
+++ b/src/compute/src/logging.rs
@@ -21,14 +21,22 @@ use std::marker::PhantomData;
 use std::rc::Rc;
 use std::time::Duration;
 
+use ::timely::container::ContainerBuilder;
+use ::timely::dataflow::channels::pact::Pipeline;
+use ::timely::dataflow::channels::pushers::buffer::Session;
+use ::timely::dataflow::channels::pushers::{Counter, Tee};
 use ::timely::dataflow::operators::capture::{Event, EventLink, EventPusher};
+use ::timely::dataflow::operators::Operator;
+use ::timely::dataflow::StreamCore;
 use ::timely::progress::Timestamp as TimelyTimestamp;
 use ::timely::scheduling::Activator;
 use ::timely::Container;
+use differential_dataflow::trace::Batcher;
 use mz_compute_client::logging::{ComputeLog, DifferentialLog, LogVariant, TimelyLog};
 use mz_expr::{permutation_for_arrangement, MirScalarExpr};
-use mz_repr::{Datum, Diff, Row, RowPacker, SharedRow, Timestamp};
+use mz_repr::{Datum, Diff, Row, RowPacker, RowRef, Timestamp};
 use mz_timely_util::activator::RcActivator;
+use mz_timely_util::operator::consolidate_pact;
 
 use crate::logging::compute::Logger as ComputeLogger;
 use crate::typedefs::RowRowAgent;
@@ -64,30 +72,35 @@ where
             _marker: PhantomData,
         }
     }
+}
+
+impl<C, P> BatchLogger<C, P>
+where
+    P: EventPusher<Timestamp, C>,
+    C: Container,
+{
+    /// Publishes a batch of logged events.
+    fn publish_batch(&mut self, data: C) {
+        self.event_pusher.push(Event::Messages(self.time_ms, data));
+    }
 
-    /// Indicate progress up to a specific `time`.
-    fn report_progress(&mut self, time: &Duration) {
+    /// Indicate progress up to `time`, advances the capability.
+    ///
+    /// Returns `true` if the capability was advanced.
+    fn report_progress(&mut self, time: Duration) -> bool {
         let time_ms = ((time.as_millis() / self.interval_ms) + 1) * self.interval_ms;
         let new_time_ms: Timestamp = time_ms.try_into().expect("must fit");
         if self.time_ms < new_time_ms {
             self.event_pusher
                 .push(Event::Progress(vec![(new_time_ms, 1), (self.time_ms, -1)]));
             self.time_ms = new_time_ms;
+            true
+        } else {
+            false
         }
     }
 }
 
-impl<C, P> BatchLogger<C, P>
-where
-    P: EventPusher<Timestamp, C>,
-{
-    /// Publishes a batch of logged events and advances the capability.
-    fn publish_batch(&mut self, time: &Duration, data: C) {
-        self.event_pusher.push(Event::Messages(self.time_ms, data));
-        self.report_progress(time);
-    }
-}
-
 impl<C, P> Drop for BatchLogger<C, P>
 where
     P: EventPusher<Timestamp, C>,
@@ -132,6 +145,8 @@ struct SharedLoggingState {
 pub(crate) struct PermutedRowPacker {
     key: Vec<usize>,
     value: Vec<usize>,
+    key_row: Row,
+    value_row: Row,
 }
 
 impl PermutedRowPacker {
@@ -146,32 +161,45 @@ impl PermutedRowPacker {
                 .collect::<Vec<_>>(),
             variant.desc().arity(),
         );
-        Self { key, value }
+        Self {
+            key,
+            value,
+            key_row: Row::default(),
+            value_row: Row::default(),
+        }
     }
 
     /// Pack a slice of datums suitable for the key columns in the log variant.
-    pub(crate) fn pack_slice(&self, datums: &[Datum]) -> (Row, Row) {
+    pub(crate) fn pack_slice(&mut self, datums: &[Datum]) -> (&RowRef, &RowRef) {
         self.pack_by_index(|packer, index| packer.push(datums[index]))
     }
 
-    /// Pack using a callback suitable for the key columns in the log variant.
-    pub(crate) fn pack_by_index<F: Fn(&mut RowPacker, usize)>(&self, logic: F) -> (Row, Row) {
-        let binding = SharedRow::get();
-        let mut row_builder = binding.borrow_mut();
+    /// Pack a slice of datums suitable for the key columns in the log variant, returning owned
+    /// rows.
+    ///
+    /// This is equivalent to calling [`PermutedRowPacker::pack_slice`] and then calling `to_owned`
+    /// on the returned rows.
+    pub(crate) fn pack_slice_owned(&mut self, datums: &[Datum]) -> (Row, Row) {
+        let (key, value) = self.pack_slice(datums);
+        (key.to_owned(), value.to_owned())
+    }
 
-        let mut packer = row_builder.packer();
+    /// Pack using a callback suitable for the key columns in the log variant.
+    pub(crate) fn pack_by_index<F: Fn(&mut RowPacker, usize)>(
+        &mut self,
+        logic: F,
+    ) -> (&RowRef, &RowRef) {
+        let mut packer = self.key_row.packer();
         for index in &self.key {
             logic(&mut packer, *index);
         }
-        let key_row = row_builder.clone();
 
-        let mut packer = row_builder.packer();
+        let mut packer = self.value_row.packer();
         for index in &self.value {
             logic(&mut packer, *index);
         }
-        let value_row = row_builder.clone();
 
-        (key_row, value_row)
+        (&self.key_row, &self.value_row)
     }
 }
 
@@ -184,3 +212,49 @@ struct LogCollection {
     /// Index of the dataflow exporting this collection.
     dataflow_index: usize,
 }
+
+pub(super) type Pusher<C> = Counter<Timestamp, C, Tee<Timestamp, C>>;
+pub(super) type OutputSession<'a, CB> =
+    Session<'a, Timestamp, CB, Pusher<<CB as ContainerBuilder>::Container>>;
+
+/// A single-purpose function to consolidate and pack updates for log collection.
+///
+/// The function first consolidates worker-local updates using the [`Pipeline`] pact, then converts
+/// the updates into `(Row, Row)` pairs using the provided logic function. It is crucial that the
+/// data is not exchanged between workers, as the consolidation would not function as desired
+/// otherwise.
+pub(super) fn consolidate_and_pack<G, B, CB, L, F>(
+    input: &StreamCore<G, B::Input>,
+    log: L,
+    mut logic: F,
+) -> StreamCore<G, CB::Container>
+where
+    G: ::timely::dataflow::Scope<Timestamp = Timestamp>,
+    B: Batcher<Time = G::Timestamp> + 'static,
+    B::Input: Container + Clone + 'static,
+    B::Output: Container + Clone + 'static,
+    CB: ContainerBuilder,
+    L: Into<LogVariant>,
+    F: for<'a> FnMut(
+            <B::Output as Container>::Item<'a>,
+            &mut PermutedRowPacker,
+            &mut OutputSession<CB>,
+        ) + 'static,
+{
+    let log = log.into();
+    // TODO: Use something other than the debug representation of the log variant as a name.
+    let c_name = &format!("Consolidate {log:?}");
+    let u_name = &format!("ToRow {log:?}");
+    let mut packer = PermutedRowPacker::new(log);
+    let consolidated = consolidate_pact::<B, _, _>(input, Pipeline, c_name);
+    consolidated.unary::<CB, _, _, _>(Pipeline, u_name, |_, _| {
+        move |input, output| {
+            while let Some((time, data)) = input.next() {
+                let mut session = output.session_with_builder(&time);
+                for item in data.drain() {
+                    logic(item, &mut packer, &mut session);
+                }
+            }
+        }
+    })
+}
diff --git a/src/compute/src/logging/compute.rs b/src/compute/src/logging/compute.rs
index cd3e3633d620e..049f952b2f50a 100644
--- a/src/compute/src/logging/compute.rs
+++ b/src/compute/src/logging/compute.rs
@@ -15,20 +15,23 @@ use std::fmt::{Display, Write};
 use std::rc::Rc;
 use std::time::{Duration, Instant};
 
+use columnar::{Columnar, Index};
 use differential_dataflow::collection::AsCollection;
 use differential_dataflow::trace::{BatchReader, Cursor};
 use differential_dataflow::Collection;
 use mz_compute_types::plan::LirId;
 use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, GlobalId, Timestamp};
+use mz_timely_util::containers::{Column, ColumnBuilder, ProvidedBuilder};
 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};
+use timely::dataflow::operators::core::Map;
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
-use timely::dataflow::operators::{Filter, Operator};
+use timely::dataflow::operators::Operator;
 use timely::dataflow::{Scope, Stream};
 use timely::scheduling::Scheduler;
 use timely::worker::Worker;
@@ -45,10 +48,10 @@ use crate::typedefs::RowRowSpine;
 
 /// Type alias for a logger of compute events.
 pub type Logger = timely::logging_core::Logger<ComputeEventBuilder>;
-pub type ComputeEventBuilder = CapacityContainerBuilder<Vec<(Duration, ComputeEvent)>>;
+pub type ComputeEventBuilder = ColumnBuilder<(Duration, ComputeEvent)>;
 
 /// A dataflow exports a global ID.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct Export {
     /// Identifier of the export.
     pub export_id: GlobalId,
@@ -57,14 +60,14 @@ pub struct Export {
 }
 
 /// The export for a global id was dropped.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct ExportDropped {
     /// Identifier of the export.
     pub export_id: GlobalId,
 }
 
 /// A peek event with a [`Peek`], a [`PeekType`], and an installation status.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct PeekEvent {
     /// The data for the peek itself.
     pub peek: Peek,
@@ -76,7 +79,7 @@ pub struct PeekEvent {
 }
 
 /// Frontier change event.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct Frontier {
     pub export_id: GlobalId,
     pub time: Timestamp,
@@ -84,7 +87,7 @@ pub struct Frontier {
 }
 
 /// An import frontier change.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct ImportFrontier {
     pub import_id: GlobalId,
     pub export_id: GlobalId,
@@ -93,7 +96,7 @@ pub struct ImportFrontier {
 }
 
 /// A change in an arrangement's heap size.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct ArrangementHeapSize {
     /// Operator index
     pub operator_id: usize,
@@ -102,7 +105,7 @@ pub struct ArrangementHeapSize {
 }
 
 /// A change in an arrangement's heap capacity.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct ArrangementHeapCapacity {
     /// Operator index
     pub operator_id: usize,
@@ -111,7 +114,7 @@ pub struct ArrangementHeapCapacity {
 }
 
 /// A change in an arrangement's heap allocation count.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct ArrangementHeapAllocations {
     /// Operator index
     pub operator_id: usize,
@@ -120,30 +123,30 @@ pub struct ArrangementHeapAllocations {
 }
 
 /// Announcing an operator that manages an arrangement.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct ArrangementHeapSizeOperator {
     /// Operator index
     pub operator_id: usize,
     /// The address of the operator.
-    pub address: Rc<[usize]>,
+    pub address: Vec<usize>,
 }
 
 /// Drop event for an operator managing an arrangement.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct ArrangementHeapSizeOperatorDrop {
     /// Operator index
     pub operator_id: usize,
 }
 
 /// Dataflow shutdown event.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct DataflowShutdown {
     /// Timely worker index of the dataflow.
     pub dataflow_index: usize,
 }
 
 /// Error count update event.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct ErrorCount {
     /// Identifier of the export.
     pub export_id: GlobalId,
@@ -152,13 +155,13 @@ pub struct ErrorCount {
 }
 
 /// An export is hydrated.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct Hydration {
     pub export_id: GlobalId,
 }
 
 /// Announce a mapping of an LIR operator to a dataflow operator for a global ID.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct LirMapping {
     /// The `GlobalId` in which the LIR operator is rendered.
     ///
@@ -169,11 +172,11 @@ pub struct LirMapping {
     pub global_id: GlobalId,
     /// The actual mapping.
     /// Represented this way to reduce the size of `ComputeEvent`.
-    pub mapping: Box<[(LirId, LirMetadata)]>,
+    pub mapping: Vec<(LirId, LirMetadata)>,
 }
 
 /// Announce that a dataflow supports a specific global ID.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct DataflowGlobal {
     /// The identifier of the dataflow.
     pub dataflow_index: usize,
@@ -182,7 +185,7 @@ pub struct DataflowGlobal {
 }
 
 /// A logged compute event.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub enum ComputeEvent {
     /// A dataflow export was created.
     Export(Export),
@@ -218,7 +221,7 @@ pub enum ComputeEvent {
 }
 
 /// A peek type distinguishing between index and persist peeks.
-#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug)]
+#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug, Columnar)]
 pub enum PeekType {
     /// A peek against an index.
     Index,
@@ -237,31 +240,30 @@ impl PeekType {
 }
 
 /// A logged peek event.
-#[derive(Debug, Clone, PartialOrd, PartialEq)]
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct Peek {
     /// The identifier of the view the peek targets.
     id: GlobalId,
     /// The logical timestamp requested.
     time: Timestamp,
     /// The ID of the peek.
-    uuid: Uuid,
+    uuid: uuid::Bytes,
 }
 
 impl Peek {
     /// Create a new peek from its arguments.
     pub fn new(id: GlobalId, time: Timestamp, uuid: Uuid) -> Self {
+        let uuid = uuid.into_bytes();
         Self { id, time, uuid }
     }
 }
 
 /// Metadata for LIR operators.
-#[derive(Clone, Debug, PartialEq, PartialOrd)]
+#[derive(Clone, Debug, PartialEq, PartialOrd, Columnar)]
 pub struct LirMetadata {
     /// The LIR operator, as a string (see `FlatPlanNode::humanize`).
-    operator: Box<str>,
+    operator: String,
     /// The LIR identifier of the parent (if any).
-    /// Since `LirId`s are strictly positive, Rust can steal the low bit.
-    /// (See `test_option_lirid_fits_in_usize`.)
     parent_lir_id: Option<LirId>,
     /// How nested the operator is (for nice indentation).
     nesting: u8,
@@ -273,7 +275,7 @@ pub struct LirMetadata {
 impl LirMetadata {
     /// Construct a new LIR metadata object.
     pub fn new(
-        operator: Box<str>,
+        operator: String,
         parent_lir_id: Option<LirId>,
         nesting: u8,
         operator_span: (usize, usize),
@@ -296,7 +298,7 @@ impl LirMetadata {
 pub(super) fn construct<A: Allocate + 'static>(
     worker: &mut timely::worker::Worker<A>,
     config: &mz_compute_client::logging::LoggingConfig,
-    event_queue: EventQueue<Vec<(Duration, ComputeEvent)>>,
+    event_queue: EventQueue<Column<(Duration, ComputeEvent)>>,
     shared_state: Rc<RefCell<SharedLoggingState>>,
 ) -> BTreeMap<LogVariant, LogCollection> {
     let logging_interval_ms = std::cmp::max(1, config.interval.as_millis());
@@ -305,20 +307,20 @@ pub(super) fn construct<A: Allocate + 'static>(
     let dataflow_index = worker.next_dataflow_index();
 
     worker.dataflow_named("Dataflow: compute logging", move |scope| {
-        let (mut logs, token) = Some(event_queue.link)
-            .mz_replay::<_, CapacityContainerBuilder<_>, _>(
-                scope,
-                "compute logs",
-                config.interval,
-                event_queue.activator,
-                |mut session, data| session.give_iterator(data.iter()),
-            );
-
-        // If logging is disabled, we still need to install the indexes, but we can leave them
-        // empty. We do so by immediately filtering all logs events.
-        if !config.enable_logging {
-            logs = logs.filter(|_| false);
-        }
+        let enable_logging = config.enable_logging;
+        let (logs, token) = Some(event_queue.link).mz_replay::<_, ProvidedBuilder<_>, _>(
+            scope,
+            "compute logs",
+            config.interval,
+            event_queue.activator,
+            move |mut session, data| {
+                // If logging is disabled, we still need to install the indexes, but we can leave them
+                // empty. We do so by immediately filtering all logs events.
+                if enable_logging {
+                    session.give_container(&mut data.clone())
+                }
+            },
+        );
 
         // Build a demux operator that splits the replayed event stream up into the separate
         // logging streams.
@@ -369,14 +371,15 @@ pub(super) fn construct<A: Allocate + 'static>(
                         arrangement_heap_allocations: arrangement_heap_allocations.session(&cap),
                         error_count: error_count.session(&cap),
                         hydration_time: hydration_time.session(&cap),
-                        lir_mapping: lir_mapping.session(&cap),
+                        lir_mapping: lir_mapping.session_with_builder(&cap),
                         dataflow_global_ids: dataflow_global_ids.session(&cap),
                     };
 
-                    for (time, event) in data.drain(..) {
+                    let shared_state = &mut shared_state.borrow_mut();
+                    for (time, event) in data.drain() {
                         DemuxHandler {
                             state: &mut demux_state,
-                            shared_state: &mut shared_state.borrow_mut(),
+                            shared_state,
                             output: &mut output_sessions,
                             logging_interval_ms,
                             time,
@@ -388,34 +391,34 @@ pub(super) fn construct<A: Allocate + 'static>(
         });
 
         // Encode the contents of each logging stream into its expected `Row` format.
-        let packer = PermutedRowPacker::new(ComputeLog::DataflowCurrent);
+        let mut packer = PermutedRowPacker::new(ComputeLog::DataflowCurrent);
         let dataflow_current = export.as_collection().map({
             let mut scratch = String::new();
             move |datum| {
-                packer.pack_slice(&[
+                packer.pack_slice_owned(&[
                     make_string_datum(datum.export_id, &mut scratch),
                     Datum::UInt64(u64::cast_from(worker_id)),
                     Datum::UInt64(u64::cast_from(datum.dataflow_index)),
                 ])
             }
         });
-        let packer = PermutedRowPacker::new(ComputeLog::FrontierCurrent);
+        let mut packer = PermutedRowPacker::new(ComputeLog::FrontierCurrent);
         let frontier_current = frontier.as_collection().map({
             let mut scratch = String::new();
             move |datum| {
-                packer.pack_slice(&[
+                packer.pack_slice_owned(&[
                     make_string_datum(datum.export_id, &mut scratch),
                     Datum::UInt64(u64::cast_from(worker_id)),
                     Datum::MzTimestamp(datum.time),
                 ])
             }
         });
-        let packer = PermutedRowPacker::new(ComputeLog::ImportFrontierCurrent);
+        let mut packer = PermutedRowPacker::new(ComputeLog::ImportFrontierCurrent);
         let import_frontier_current = import_frontier.as_collection().map({
             let mut scratch1 = String::new();
             let mut scratch2 = String::new();
             move |datum| {
-                packer.pack_slice(&[
+                packer.pack_slice_owned(&[
                     make_string_datum(datum.export_id, &mut scratch1),
                     make_string_datum(datum.import_id, &mut scratch2),
                     Datum::UInt64(u64::cast_from(worker_id)),
@@ -423,12 +426,12 @@ pub(super) fn construct<A: Allocate + 'static>(
                 ])
             }
         });
-        let packer = PermutedRowPacker::new(ComputeLog::PeekCurrent);
+        let mut packer = PermutedRowPacker::new(ComputeLog::PeekCurrent);
         let peek_current = peek.as_collection().map({
             let mut scratch = String::new();
             move |PeekDatum { peek, peek_type }| {
-                packer.pack_slice(&[
-                    Datum::Uuid(peek.uuid),
+                packer.pack_slice_owned(&[
+                    Datum::Uuid(Uuid::from_bytes(peek.uuid)),
                     Datum::UInt64(u64::cast_from(worker_id)),
                     make_string_datum(peek.id, &mut scratch),
                     Datum::String(peek_type.name()),
@@ -436,20 +439,20 @@ pub(super) fn construct<A: Allocate + 'static>(
                 ])
             }
         });
-        let packer = PermutedRowPacker::new(ComputeLog::PeekDuration);
+        let mut packer = PermutedRowPacker::new(ComputeLog::PeekDuration);
         let peek_duration =
             peek_duration
                 .as_collection()
                 .map(move |PeekDurationDatum { peek_type, bucket }| {
-                    packer.pack_slice(&[
+                    packer.pack_slice_owned(&[
                         Datum::UInt64(u64::cast_from(worker_id)),
                         Datum::String(peek_type.name()),
                         Datum::UInt64(bucket.try_into().expect("bucket too big")),
                     ])
                 });
-        let packer = PermutedRowPacker::new(ComputeLog::ShutdownDuration);
+        let mut packer = PermutedRowPacker::new(ComputeLog::ShutdownDuration);
         let shutdown_duration = shutdown_duration.as_collection().map(move |bucket| {
-            packer.pack_slice(&[
+            packer.pack_slice_owned(&[
                 Datum::UInt64(u64::cast_from(worker_id)),
                 Datum::UInt64(bucket.try_into().expect("bucket too big")),
             ])
@@ -457,7 +460,7 @@ pub(super) fn construct<A: Allocate + 'static>(
 
         let arrangement_heap_datum_to_row =
             move |packer: &mut PermutedRowPacker, ArrangementHeapDatum { operator_id }| {
-                packer.pack_slice(&[
+                packer.pack_slice_owned(&[
                     Datum::UInt64(operator_id.try_into().expect("operator_id too big")),
                     Datum::UInt64(u64::cast_from(worker_id)),
                 ])
@@ -478,11 +481,11 @@ pub(super) fn construct<A: Allocate + 'static>(
             .as_collection()
             .map(move |d| arrangement_heap_datum_to_row(&mut packer, d));
 
-        let packer = PermutedRowPacker::new(ComputeLog::ErrorCount);
+        let mut packer = PermutedRowPacker::new(ComputeLog::ErrorCount);
         let error_count = error_count.as_collection().map({
             let mut scratch = String::new();
             move |datum| {
-                packer.pack_slice(&[
+                packer.pack_slice_owned(&[
                     make_string_datum(datum.export_id, &mut scratch),
                     Datum::UInt64(u64::cast_from(worker_id)),
                     Datum::Int64(datum.count),
@@ -490,11 +493,11 @@ pub(super) fn construct<A: Allocate + 'static>(
             }
         });
 
-        let packer = PermutedRowPacker::new(ComputeLog::HydrationTime);
+        let mut packer = PermutedRowPacker::new(ComputeLog::HydrationTime);
         let hydration_time = hydration_time.as_collection().map({
             let mut scratch = String::new();
             move |datum| {
-                packer.pack_slice(&[
+                packer.pack_slice_owned(&[
                     make_string_datum(datum.export_id, &mut scratch),
                     Datum::UInt64(u64::cast_from(worker_id)),
                     Datum::from(datum.time_ns),
@@ -502,32 +505,33 @@ pub(super) fn construct<A: Allocate + 'static>(
             }
         });
 
-        let packer = PermutedRowPacker::new(ComputeLog::LirMapping);
-        let lir_mapping = lir_mapping.as_collection().map({
-            let mut scratch1 = String::new();
-            let mut scratch2 = String::new();
-            move |datum| {
-                packer.pack_slice(&[
-                    make_string_datum(datum.global_id, &mut scratch1),
-                    Datum::UInt64(datum.lir_id.into()),
+        let mut scratch1 = String::new();
+        let mut scratch2 = String::new();
+        let mut packer = PermutedRowPacker::new(ComputeLog::LirMapping);
+        let lir_mapping = lir_mapping
+            .map(move |(datum, time, diff)| {
+                let row = packer.pack_slice_owned(&[
+                    make_string_datum(GlobalId::into_owned(datum.global_id), &mut scratch1),
+                    Datum::UInt64(<LirId as Columnar>::into_owned(datum.lir_id).into()),
                     Datum::UInt64(u64::cast_from(worker_id)),
-                    make_string_datum(&datum.operator, &mut scratch2),
+                    make_string_datum(datum.operator, &mut scratch2),
                     datum
                         .parent_lir_id
-                        .map(|lir_id| Datum::UInt64(lir_id.into()))
+                        .map(|lir_id| Datum::UInt64(LirId::into_owned(lir_id).into()))
                         .unwrap_or_else(|| Datum::Null),
-                    Datum::UInt16(u16::cast_from(datum.nesting)),
+                    Datum::UInt16(u16::cast_from(*datum.nesting)),
                     Datum::UInt64(u64::cast_from(datum.operator_span.0)),
                     Datum::UInt64(u64::cast_from(datum.operator_span.1)),
-                ])
-            }
-        });
+                ]);
+                (row, Timestamp::into_owned(time), *diff)
+            })
+            .as_collection();
 
-        let packer = PermutedRowPacker::new(ComputeLog::DataflowGlobal);
+        let mut packer = PermutedRowPacker::new(ComputeLog::DataflowGlobal);
         let dataflow_global_ids = dataflow_global_ids.as_collection().map({
             let mut scratch = String::new();
             move |datum| {
-                packer.pack_slice(&[
+                packer.pack_slice_owned(&[
                     Datum::UInt64(u64::cast_from(datum.dataflow_index)),
                     Datum::UInt64(u64::cast_from(worker_id)),
                     make_string_datum(datum.global_id, &mut scratch),
@@ -665,9 +669,14 @@ struct ArrangementSizeState {
 type Update<D> = (D, Timestamp, Diff);
 /// A pusher for updates of value `D` for vector-based containers.
 type Pusher<D> = Counter<Timestamp, Vec<Update<D>>, Tee<Timestamp, Vec<Update<D>>>>;
+/// A pusher for updates of value `D` for columnar containers.
+type PusherColumnar<D> = Counter<Timestamp, Column<Update<D>>, Tee<Timestamp, Column<Update<D>>>>;
 /// An output session for vector-based containers of updates `D`, using a capacity container builder.
 type OutputSession<'a, D> =
     Session<'a, Timestamp, CapacityContainerBuilder<Vec<Update<D>>>, Pusher<D>>;
+/// An output session for columnar containers of updates `D`, using a column builder.
+type OutputSessionColumnar<'a, D> =
+    Session<'a, Timestamp, ColumnBuilder<Update<D>>, PusherColumnar<D>>;
 
 /// Bundled output sessions used by the demux operator.
 struct DemuxOutput<'a> {
@@ -682,7 +691,7 @@ struct DemuxOutput<'a> {
     arrangement_heap_allocations: OutputSession<'a, ArrangementHeapDatum>,
     hydration_time: OutputSession<'a, HydrationTimeDatum>,
     error_count: OutputSession<'a, ErrorCountDatum>,
-    lir_mapping: OutputSession<'a, LirMappingDatum>,
+    lir_mapping: OutputSessionColumnar<'a, LirMappingDatum>,
     dataflow_global_ids: OutputSession<'a, DataflowGlobalDatum>,
 }
 
@@ -717,7 +726,7 @@ struct PeekDurationDatum {
     bucket: u128,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Copy)]
 struct ArrangementHeapDatum {
     operator_id: usize,
 }
@@ -737,11 +746,11 @@ struct ErrorCountDatum {
     count: i64,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Columnar)]
 struct LirMappingDatum {
     global_id: GlobalId,
     lir_id: LirId,
-    operator: Box<str>,
+    operator: String,
     parent_lir_id: Option<LirId>,
     nesting: u8,
     operator_span: (usize, usize),
@@ -778,54 +787,38 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
     }
 
     /// Handle the given compute event.
-    fn handle(&mut self, event: ComputeEvent) {
+    fn handle(&mut self, event: <ComputeEvent as Columnar>::Ref<'_>) {
+        use ComputeEventReference::*;
         match event {
-            ComputeEvent::Export(export) => self.handle_export(export),
-            ComputeEvent::ExportDropped(export_dropped) => {
-                self.handle_export_dropped(export_dropped)
-            }
-            ComputeEvent::Peek(
-                peek @ PeekEvent {
-                    installed: true, ..
-                },
-            ) => self.handle_peek_install(peek),
-            ComputeEvent::Peek(
-                peek @ PeekEvent {
-                    installed: false, ..
-                },
-            ) => self.handle_peek_retire(peek),
-            ComputeEvent::Frontier(frontier) => self.handle_frontier(frontier),
-            ComputeEvent::ImportFrontier(import_frontier) => {
-                self.handle_import_frontier(import_frontier)
-            }
-            ComputeEvent::ArrangementHeapSize(inner) => self.handle_arrangement_heap_size(inner),
-            ComputeEvent::ArrangementHeapCapacity(inner) => {
-                self.handle_arrangement_heap_capacity(inner)
-            }
-            ComputeEvent::ArrangementHeapAllocations(inner) => {
-                self.handle_arrangement_heap_allocations(inner)
-            }
-            ComputeEvent::ArrangementHeapSizeOperator(inner) => {
-                self.handle_arrangement_heap_size_operator(inner)
-            }
-            ComputeEvent::ArrangementHeapSizeOperatorDrop(inner) => {
+            Export(export) => self.handle_export(export),
+            ExportDropped(export_dropped) => self.handle_export_dropped(export_dropped),
+            Peek(peek) if peek.installed => self.handle_peek_install(peek),
+            Peek(peek) => self.handle_peek_retire(peek),
+            Frontier(frontier) => self.handle_frontier(frontier),
+            ImportFrontier(import_frontier) => self.handle_import_frontier(import_frontier),
+            ArrangementHeapSize(inner) => self.handle_arrangement_heap_size(inner),
+            ArrangementHeapCapacity(inner) => self.handle_arrangement_heap_capacity(inner),
+            ArrangementHeapAllocations(inner) => self.handle_arrangement_heap_allocations(inner),
+            ArrangementHeapSizeOperator(inner) => self.handle_arrangement_heap_size_operator(inner),
+            ArrangementHeapSizeOperatorDrop(inner) => {
                 self.handle_arrangement_heap_size_operator_dropped(inner)
             }
-            ComputeEvent::DataflowShutdown(shutdown) => self.handle_dataflow_shutdown(shutdown),
-            ComputeEvent::ErrorCount(error_count) => self.handle_error_count(error_count),
-            ComputeEvent::Hydration(hydration) => self.handle_hydration(hydration),
-            ComputeEvent::LirMapping(mapping) => self.handle_lir_mapping(mapping),
-            ComputeEvent::DataflowGlobal(global) => self.handle_dataflow_global(global),
+            DataflowShutdown(shutdown) => self.handle_dataflow_shutdown(shutdown),
+            ErrorCount(error_count) => self.handle_error_count(error_count),
+            Hydration(hydration) => self.handle_hydration(hydration),
+            LirMapping(mapping) => self.handle_lir_mapping(mapping),
+            DataflowGlobal(global) => self.handle_dataflow_global(global),
         }
     }
 
     fn handle_export(
         &mut self,
-        Export {
+        ExportReference {
             export_id,
             dataflow_index,
-        }: Export,
+        }: <Export as Columnar>::Ref<'_>,
     ) {
+        let export_id = Columnar::into_owned(export_id);
         let ts = self.ts();
         let datum = ExportDatum {
             export_id,
@@ -855,7 +848,11 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
         self.output.hydration_time.give((datum, ts, 1));
     }
 
-    fn handle_export_dropped(&mut self, ExportDropped { export_id }: ExportDropped) {
+    fn handle_export_dropped(
+        &mut self,
+        ExportDroppedReference { export_id }: <ExportDropped as Columnar>::Ref<'_>,
+    ) {
+        let export_id = Columnar::into_owned(export_id);
         let Some(export) = self.state.exports.remove(&export_id) else {
             error!(%export_id, "missing exports entry at time of export drop");
             return;
@@ -917,7 +914,10 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
         }
     }
 
-    fn handle_dataflow_shutdown(&mut self, DataflowShutdown { dataflow_index }: DataflowShutdown) {
+    fn handle_dataflow_shutdown(
+        &mut self,
+        DataflowShutdownReference { dataflow_index }: <DataflowShutdown as Columnar>::Ref<'_>,
+    ) {
         let ts = self.ts();
 
         if let Some(start) = self.state.dataflow_drop_times.remove(&dataflow_index) {
@@ -963,15 +963,19 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
                             nesting,
                             operator_span,
                         };
-                        self.output.lir_mapping.give((datum, ts, -1));
+                        self.output.lir_mapping.give(&(datum, ts, -1));
                     }
                 }
             }
         }
     }
 
-    fn handle_error_count(&mut self, ErrorCount { export_id, diff }: ErrorCount) {
+    fn handle_error_count(
+        &mut self,
+        ErrorCountReference { export_id, diff }: <ErrorCount as Columnar>::Ref<'_>,
+    ) {
         let ts = self.ts();
+        let export_id = Columnar::into_owned(export_id);
 
         let Some(export) = self.state.exports.get_mut(&export_id) else {
             // The export might have already been dropped, in which case we are no longer
@@ -1000,8 +1004,12 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
         export.error_count = new_count;
     }
 
-    fn handle_hydration(&mut self, Hydration { export_id }: Hydration) {
+    fn handle_hydration(
+        &mut self,
+        HydrationReference { export_id }: <Hydration as Columnar>::Ref<'_>,
+    ) {
         let ts = self.ts();
+        let export_id = Columnar::into_owned(export_id);
 
         let Some(export) = self.state.exports.get_mut(&export_id) else {
             error!(%export_id, "hydration event for unknown export");
@@ -1032,13 +1040,14 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
 
     fn handle_peek_install(
         &mut self,
-        PeekEvent {
+        PeekEventReference {
             peek,
             peek_type,
             installed: _,
-        }: PeekEvent,
+        }: <PeekEvent as Columnar>::Ref<'_>,
     ) {
-        let uuid = peek.uuid;
+        let peek = Peek::into_owned(peek);
+        let uuid = Uuid::from_bytes(peek.uuid);
         let ts = self.ts();
         self.output
             .peek
@@ -1052,13 +1061,14 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
 
     fn handle_peek_retire(
         &mut self,
-        PeekEvent {
+        PeekEventReference {
             peek,
             peek_type,
             installed: _,
-        }: PeekEvent,
+        }: <PeekEvent as Columnar>::Ref<'_>,
     ) {
-        let uuid = peek.uuid;
+        let peek = Peek::into_owned(peek);
+        let uuid = Uuid::from_bytes(peek.uuid);
         let ts = self.ts();
         self.output
             .peek
@@ -1077,43 +1087,50 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
 
     fn handle_frontier(
         &mut self,
-        Frontier {
+        FrontierReference {
             export_id,
             time,
             diff,
-        }: Frontier,
+        }: <Frontier as Columnar>::Ref<'_>,
     ) {
-        let diff = i64::from(diff);
+        let export_id = Columnar::into_owned(export_id);
+        let diff = i64::from(*diff);
         let ts = self.ts();
+        let time = Columnar::into_owned(time);
         let datum = FrontierDatum { export_id, time };
         self.output.frontier.give((datum, ts, diff));
     }
 
     fn handle_import_frontier(
         &mut self,
-        ImportFrontier {
+        ImportFrontierReference {
             import_id,
             export_id,
             time,
             diff,
-        }: ImportFrontier,
+        }: <ImportFrontier as Columnar>::Ref<'_>,
     ) {
+        let import_id = Columnar::into_owned(import_id);
+        let export_id = Columnar::into_owned(export_id);
         let ts = self.ts();
+        let time = Columnar::into_owned(time);
         let datum = ImportFrontierDatum {
             export_id,
             import_id,
             time,
         };
-        self.output.import_frontier.give((datum, ts, diff.into()));
+        self.output
+            .import_frontier
+            .give((datum, ts, (*diff).into()));
     }
 
     /// Update the allocation size for an arrangement.
     fn handle_arrangement_heap_size(
         &mut self,
-        ArrangementHeapSize {
+        ArrangementHeapSizeReference {
             operator_id,
             delta_size,
-        }: ArrangementHeapSize,
+        }: <ArrangementHeapSize as Columnar>::Ref<'_>,
     ) {
         let ts = self.ts();
         let Some(state) = self.state.arrangement_size.get_mut(&operator_id) else {
@@ -1131,10 +1148,10 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
     /// Update the allocation capacity for an arrangement.
     fn handle_arrangement_heap_capacity(
         &mut self,
-        ArrangementHeapCapacity {
+        ArrangementHeapCapacityReference {
             operator_id,
             delta_capacity,
-        }: ArrangementHeapCapacity,
+        }: <ArrangementHeapCapacity as Columnar>::Ref<'_>,
     ) {
         let ts = self.ts();
         let Some(state) = self.state.arrangement_size.get_mut(&operator_id) else {
@@ -1152,10 +1169,10 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
     /// Update the allocation count for an arrangement.
     fn handle_arrangement_heap_allocations(
         &mut self,
-        ArrangementHeapAllocations {
+        ArrangementHeapAllocationsReference {
             operator_id,
             delta_allocations,
-        }: ArrangementHeapAllocations,
+        }: <ArrangementHeapAllocations as Columnar>::Ref<'_>,
     ) {
         let ts = self.ts();
         let Some(state) = self.state.arrangement_size.get_mut(&operator_id) else {
@@ -1174,12 +1191,15 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
     /// Indicate that a new arrangement exists, start maintaining the heap size state.
     fn handle_arrangement_heap_size_operator(
         &mut self,
-        ArrangementHeapSizeOperator {
+        ArrangementHeapSizeOperatorReference {
             operator_id,
             address,
-        }: ArrangementHeapSizeOperator,
+        }: <ArrangementHeapSizeOperator as Columnar>::Ref<'_>,
     ) {
-        let activator = self.state.worker.activator_for(address);
+        let activator = self
+            .state
+            .worker
+            .activator_for(address.into_iter().collect());
         let existing = self
             .state
             .arrangement_size
@@ -1199,21 +1219,19 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
     /// Indicate that an arrangement has been dropped and we can cleanup the heap size state.
     fn handle_arrangement_heap_size_operator_dropped(
         &mut self,
-        ArrangementHeapSizeOperatorDrop { operator_id }: ArrangementHeapSizeOperatorDrop,
+        ArrangementHeapSizeOperatorDropReference { operator_id }: <ArrangementHeapSizeOperatorDrop as Columnar>::Ref<'_>,
     ) {
         if let Some(state) = self.state.arrangement_size.remove(&operator_id) {
             let ts = self.ts();
             let datum = ArrangementHeapDatum { operator_id };
 
             let diff = -Diff::cast_from(state.size);
-            self.output
-                .arrangement_heap_size
-                .give((datum.clone(), ts, diff));
+            self.output.arrangement_heap_size.give((datum, ts, diff));
 
             let diff = -Diff::cast_from(state.capacity);
             self.output
                 .arrangement_heap_capacity
-                .give((datum.clone(), ts, diff));
+                .give((datum, ts, diff));
 
             let diff = -Diff::cast_from(state.count);
             self.output
@@ -1226,33 +1244,29 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
     }
 
     /// Indicate that a new LIR operator exists; record the dataflow address it maps to.
-    fn handle_lir_mapping(&mut self, LirMapping { global_id, mapping }: LirMapping) {
+    fn handle_lir_mapping(
+        &mut self,
+        LirMappingReference { global_id, mapping }: <LirMapping as Columnar>::Ref<'_>,
+    ) {
+        let global_id = Columnar::into_owned(global_id);
         // record the state (for the later drop)
+        let mappings = || mapping.into_iter().map(Columnar::into_owned);
         self.state
             .lir_mapping
             .entry(global_id)
-            .and_modify(|existing_mapping| existing_mapping.extend(mapping.iter().cloned()))
-            .or_insert_with(|| mapping.iter().cloned().collect::<BTreeMap<_, _>>());
+            .and_modify(|existing_mapping| existing_mapping.extend(mappings()))
+            .or_insert_with(|| mappings().collect());
 
         // send the datum out
         let ts = self.ts();
-        for (
-            lir_id,
-            LirMetadata {
-                operator,
-                parent_lir_id,
-                nesting,
-                operator_span,
-            },
-        ) in mapping
-        {
-            let datum = LirMappingDatum {
+        for (lir_id, meta) in mapping.into_iter() {
+            let datum = LirMappingDatumReference {
                 global_id,
                 lir_id,
-                operator,
-                parent_lir_id,
-                nesting,
-                operator_span,
+                operator: meta.operator,
+                parent_lir_id: meta.parent_lir_id,
+                nesting: meta.nesting,
+                operator_span: meta.operator_span,
             };
             self.output.lir_mapping.give((datum, ts, 1));
         }
@@ -1260,21 +1274,22 @@ impl<A: Allocate> DemuxHandler<'_, '_, A> {
 
     fn handle_dataflow_global(
         &mut self,
-        DataflowGlobal {
+        DataflowGlobalReference {
             dataflow_index,
             global_id,
-        }: DataflowGlobal,
+        }: <DataflowGlobal as Columnar>::Ref<'_>,
     ) {
+        let global_id = Columnar::into_owned(global_id);
         self.state
             .dataflow_global_ids
             .entry(dataflow_index)
             .and_modify(|globals| {
                 // NB BTreeSet::insert() returns `false` when the element was already in the set
-                if !globals.insert(global_id.clone()) {
+                if !globals.insert(global_id) {
                     error!(%dataflow_index, %global_id, "dataflow mapping already knew about this GlobalId");
                 }
             })
-            .or_insert_with(|| BTreeSet::from([global_id.clone()]));
+            .or_insert_with(|| BTreeSet::from([global_id]));
 
         let ts = self.ts();
         let datum = DataflowGlobalDatum {
@@ -1305,7 +1320,7 @@ impl CollectionLogging {
         dataflow_index: usize,
         import_ids: impl Iterator<Item = GlobalId>,
     ) -> Self {
-        logger.log(ComputeEvent::Export(Export {
+        logger.log(&ComputeEvent::Export(Export {
             export_id,
             dataflow_index,
         }));
@@ -1346,7 +1361,7 @@ impl CollectionLogging {
                     diff: 1,
                 })
             });
-            let events = retraction.into_iter().chain(insertion);
+            let events = retraction.as_ref().into_iter().chain(insertion.as_ref());
             self.logger.log_many(events);
         }
     }
@@ -1377,14 +1392,14 @@ impl CollectionLogging {
                     diff: 1,
                 })
             });
-            let events = retraction.into_iter().chain(insertion);
+            let events = retraction.as_ref().into_iter().chain(insertion.as_ref());
             self.logger.log_many(events);
         }
     }
 
     /// Set the collection as hydrated.
     pub fn set_hydrated(&self) {
-        self.logger.log(ComputeEvent::Hydration(Hydration {
+        self.logger.log(&ComputeEvent::Hydration(Hydration {
             export_id: self.export_id,
         }));
     }
@@ -1400,7 +1415,7 @@ impl Drop for CollectionLogging {
             self.set_import_frontier(import_id, None);
         }
 
-        self.logger.log(ComputeEvent::ExportDropped(ExportDropped {
+        self.logger.log(&ComputeEvent::ExportDropped(ExportDropped {
             export_id: self.export_id,
         }));
     }
@@ -1423,7 +1438,7 @@ where
                 move |input, output| {
                     input.for_each(|cap, data| {
                         let diff = data.iter().map(|(_d, _t, r)| r).sum();
-                        logger.log(ComputeEvent::ErrorCount(ErrorCount { export_id, diff }));
+                        logger.log(&ComputeEvent::ErrorCount(ErrorCount { export_id, diff }));
 
                         output.session(&cap).give_container(data);
                     });
@@ -1443,7 +1458,7 @@ where
             move |input, output| {
                 input.for_each(|cap, data| {
                     let diff = data.iter().map(sum_batch_diffs).sum();
-                    logger.log(ComputeEvent::ErrorCount(ErrorCount { export_id, diff }));
+                    logger.log(&ComputeEvent::ErrorCount(ErrorCount { export_id, diff }));
 
                     output.session(&cap).give_container(data);
                 });
diff --git a/src/compute/src/logging/differential.rs b/src/compute/src/logging/differential.rs
index 27149e7e7a362..42ae6e508bdec 100644
--- a/src/compute/src/logging/differential.rs
+++ b/src/compute/src/logging/differential.rs
@@ -18,26 +18,26 @@ use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::logging::{
     BatchEvent, BatcherEvent, DifferentialEvent, DropEvent, MergeEvent, TraceShare,
 };
-use differential_dataflow::AsCollection;
 use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, Timestamp};
-use mz_timely_util::operator::consolidate_pact;
+use mz_timely_util::containers::{columnar_exchange, Col2ValBatcher, ColumnBuilder};
 use mz_timely_util::replay::MzReplay;
 use timely::communication::Allocate;
 use timely::container::CapacityContainerBuilder;
-use timely::dataflow::channels::pact::Pipeline;
+use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
 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;
 use timely::dataflow::Stream;
 
-use crate::extensions::arrange::MzArrange;
+use crate::extensions::arrange::MzArrangeCore;
 use crate::logging::compute::{ArrangementHeapSizeOperatorDrop, ComputeEvent};
 use crate::logging::{
-    DifferentialLog, EventQueue, LogCollection, LogVariant, PermutedRowPacker, SharedLoggingState,
+    consolidate_and_pack, DifferentialLog, EventQueue, LogCollection, LogVariant,
+    SharedLoggingState,
 };
-use crate::row_spine::{RowRowBatcher, RowRowBuilder};
+use crate::row_spine::RowRowBuilder;
 use crate::typedefs::{KeyBatcher, RowRowSpine};
 
 /// Constructs the logging dataflow for differential logs.
@@ -121,36 +121,30 @@ pub(super) fn construct<A: Allocate>(
             }
         });
 
-        let stream_to_collection = |input: Stream<_, ((usize, ()), Timestamp, Diff)>, log, name| {
-            let packer = PermutedRowPacker::new(log);
-            consolidate_pact::<KeyBatcher<_, _, _>, _, _, _, _>(
-                &input,
-                Pipeline,
-                &format!("Consolidate Differential {name}"),
+        // We're lucky and the differential logs all have the same stream format, so just implement
+        // the call once.
+        let stream_to_collection = |input: &Stream<_, ((usize, ()), Timestamp, Diff)>, log| {
+            consolidate_and_pack::<_, KeyBatcher<_, _, _>, ColumnBuilder<_>, _, _>(
+                input,
+                log,
+                move |((op, ()), time, diff), packer, session| {
+                    let data = packer.pack_slice(&[
+                        Datum::UInt64(u64::cast_from(*op)),
+                        Datum::UInt64(u64::cast_from(worker_id)),
+                    ]);
+                    session.give((data, *time, *diff))
+                },
             )
-            .as_collection()
-            .map(move |(op, ())| {
-                packer.pack_slice(&[
-                    Datum::UInt64(u64::cast_from(op)),
-                    Datum::UInt64(u64::cast_from(worker_id)),
-                ])
-            })
         };
 
         // Encode the contents of each logging stream into its expected `Row` format.
-        let arrangement_batches = stream_to_collection(batches, ArrangementBatches, "batches");
-        let arrangement_records = stream_to_collection(records, ArrangementRecords, "records");
-        let sharing = stream_to_collection(sharing, Sharing, "sharing");
-        let batcher_records =
-            stream_to_collection(batcher_records, BatcherRecords, "batcher records");
-        let batcher_size = stream_to_collection(batcher_size, BatcherSize, "batcher size");
-        let batcher_capacity =
-            stream_to_collection(batcher_capacity, BatcherCapacity, "batcher capacity");
-        let batcher_allocations = stream_to_collection(
-            batcher_allocations,
-            BatcherAllocations,
-            "batcher allocations",
-        );
+        let arrangement_batches = stream_to_collection(&batches, ArrangementBatches);
+        let arrangement_records = stream_to_collection(&records, ArrangementRecords);
+        let sharing = stream_to_collection(&sharing, Sharing);
+        let batcher_records = stream_to_collection(&batcher_records, BatcherRecords);
+        let batcher_size = stream_to_collection(&batcher_size, BatcherSize);
+        let batcher_capacity = stream_to_collection(&batcher_capacity, BatcherCapacity);
+        let batcher_allocations = stream_to_collection(&batcher_allocations, BatcherAllocations);
 
         use DifferentialLog::*;
         let logs = [
@@ -169,7 +163,8 @@ pub(super) fn construct<A: Allocate>(
             let variant = LogVariant::Differential(variant);
             if config.index_logs.contains_key(&variant) {
                 let trace = collection
-                    .mz_arrange::<RowRowBatcher<_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                        ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<mz_repr::Row, mz_repr::Row, Timestamp, mz_repr::Diff>),
                         &format!("Arrange {variant:?}"),
                     )
                     .trace;
@@ -298,7 +293,7 @@ impl DemuxHandler<'_, '_> {
                 .expect("under/overflow");
             if *sharing == 0 {
                 self.state.sharing.remove(&operator_id);
-                logger.log(ComputeEvent::ArrangementHeapSizeOperatorDrop(
+                logger.log(&ComputeEvent::ArrangementHeapSizeOperatorDrop(
                     ArrangementHeapSizeOperatorDrop { operator_id },
                 ));
             }
diff --git a/src/compute/src/logging/initialize.rs b/src/compute/src/logging/initialize.rs
index 5c6d2a37e9bba..9c4a51fc7adf0 100644
--- a/src/compute/src/logging/initialize.rs
+++ b/src/compute/src/logging/initialize.rs
@@ -14,14 +14,13 @@ use differential_dataflow::dynamic::pointstamp::PointStamp;
 use differential_dataflow::logging::{DifferentialEvent, DifferentialEventBuilder};
 use differential_dataflow::Collection;
 use mz_compute_client::logging::{LogVariant, LoggingConfig};
-use mz_ore::flatcontainer::{MzRegionPreference, OwnedRegionOpinion};
 use mz_repr::{Diff, Timestamp};
 use mz_storage_operators::persist_source::Subtime;
 use mz_storage_types::errors::DataflowError;
+use mz_timely_util::containers::{Column, ColumnBuilder};
 use mz_timely_util::operator::CollectionExt;
 use timely::communication::Allocate;
-use timely::container::flatcontainer::FlatStack;
-use timely::container::ContainerBuilder;
+use timely::container::{ContainerBuilder, PushInto};
 use timely::logging::{ProgressEventTimestamp, TimelyEvent, TimelyEventBuilder};
 use timely::logging_core::Logger;
 use timely::order::Product;
@@ -82,14 +81,10 @@ pub fn initialize<A: Allocate + 'static>(
     (logger, traces)
 }
 
-/// Type to specify the region for holding reachability events. Only intended to be interpreted
-/// as [`MzRegionPreference`].
-type ReachabilityEventRegionPreference = (
-    OwnedRegionOpinion<Vec<usize>>,
-    OwnedRegionOpinion<Vec<(usize, usize, bool, Option<Timestamp>, Diff)>>,
+pub(super) type ReachabilityEvent = (
+    Vec<usize>,
+    Vec<(usize, usize, bool, Option<Timestamp>, Diff)>,
 );
-pub(super) type ReachabilityEventRegion =
-    <(Duration, ReachabilityEventRegionPreference) as MzRegionPreference>::Region;
 
 struct LoggingContext<'a, A: Allocate> {
     worker: &'a mut timely::worker::Worker<A>,
@@ -98,9 +93,9 @@ struct LoggingContext<'a, A: Allocate> {
     now: Instant,
     start_offset: Duration,
     t_event_queue: EventQueue<Vec<(Duration, TimelyEvent)>>,
-    r_event_queue: EventQueue<FlatStack<ReachabilityEventRegion>>,
+    r_event_queue: EventQueue<Column<(Duration, ReachabilityEvent)>>,
     d_event_queue: EventQueue<Vec<(Duration, DifferentialEvent)>>,
-    c_event_queue: EventQueue<Vec<(Duration, ComputeEvent)>>,
+    c_event_queue: EventQueue<Column<(Duration, ComputeEvent)>>,
     shared_state: Rc<RefCell<SharedLoggingState>>,
 }
 
@@ -174,24 +169,29 @@ impl<A: Allocate + 'static> LoggingContext<'_, A> {
         event_queue: EventQueue<CB::Container>,
     ) -> Logger<CB> {
         let mut logger = BatchLogger::new(event_queue.link, self.interval_ms);
-        Logger::new(self.now, self.start_offset, move |time, data| {
-            logger.publish_batch(time, std::mem::take(data));
-            event_queue.activator.activate();
-        })
+        Logger::new(
+            self.now,
+            self.start_offset,
+            move |time, data: &mut Option<CB::Container>| {
+                if let Some(data) = data.take() {
+                    logger.publish_batch(data);
+                } else if logger.report_progress(*time) {
+                    event_queue.activator.activate();
+                }
+            },
+        )
     }
 
     fn reachability_logger(&self) -> Logger<TrackerEventBuilder> {
         let event_queue = self.r_event_queue.clone();
-        let mut logger = BatchLogger::<FlatStack<ReachabilityEventRegion>, _>::new(
-            event_queue.link,
-            self.interval_ms,
-        );
-        Logger::new(
-            self.now,
-            self.start_offset,
-            move |time, data: &mut Vec<_>| {
-                let mut massaged = Vec::new();
-                let mut container = FlatStack::default();
+
+        let mut logger = BatchLogger::new(event_queue.link, self.interval_ms);
+        let mut massaged = Vec::new();
+        let mut builder = ColumnBuilder::default();
+
+        let action = move |batch_time: &Duration, data: &mut Option<Vec<_>>| {
+            if let Some(data) = data {
+                // Handle data
                 for (time, event) in data.drain(..) {
                     match event {
                         TrackerEvent::SourceUpdate(update) => {
@@ -203,7 +203,7 @@ impl<A: Allocate + 'static> LoggingContext<'_, A> {
                                 },
                             ));
 
-                            container.copy((time, (update.tracker_id, &massaged)));
+                            builder.push_into((time, (&update.tracker_id, &massaged)));
                             massaged.clear();
                         }
                         TrackerEvent::TargetUpdate(update) => {
@@ -215,16 +215,27 @@ impl<A: Allocate + 'static> LoggingContext<'_, A> {
                                 },
                             ));
 
-                            container.copy((time, (update.tracker_id, &massaged)));
+                            builder.push_into((time, (&update.tracker_id, &massaged)));
                             massaged.clear();
                         }
                     }
+                    while let Some(container) = builder.extract() {
+                        logger.publish_batch(std::mem::take(container));
+                    }
                 }
-                logger.publish_batch(time, container);
-                logger.report_progress(time);
-                event_queue.activator.activate();
-            },
-        )
+            } else {
+                // Handle a flush
+                while let Some(container) = builder.finish() {
+                    logger.publish_batch(std::mem::take(container));
+                }
+
+                if logger.report_progress(*batch_time) {
+                    event_queue.activator.activate();
+                }
+            }
+        };
+
+        Logger::new(self.now, self.start_offset, action)
     }
 }
 
diff --git a/src/compute/src/logging/reachability.rs b/src/compute/src/logging/reachability.rs
index 6fd790c559a48..079ea5972e332 100644
--- a/src/compute/src/logging/reachability.rs
+++ b/src/compute/src/logging/reachability.rs
@@ -12,27 +12,23 @@
 use std::collections::BTreeMap;
 use std::convert::TryInto;
 use std::rc::Rc;
+use std::time::Duration;
 
-use differential_dataflow::AsCollection;
+use columnar::Index;
 use mz_compute_client::logging::LoggingConfig;
-use mz_expr::{permutation_for_arrangement, MirScalarExpr};
 use mz_ore::cast::CastFrom;
-use mz_ore::flatcontainer::{MzRegionPreference, OwnedRegionOpinion};
-use mz_ore::iter::IteratorExt;
-use mz_repr::{Datum, Diff, RowArena, SharedRow, Timestamp};
-use mz_timely_util::operator::consolidate_pact;
+use mz_repr::{Datum, Diff, Row, Timestamp};
+use mz_timely_util::containers::{columnar_exchange, Col2ValBatcher, Column, ColumnBuilder};
 use mz_timely_util::replay::MzReplay;
 use timely::communication::Allocate;
-use timely::container::flatcontainer::FlatStack;
-use timely::container::CapacityContainerBuilder;
-use timely::dataflow::channels::pact::Pipeline;
-use timely::dataflow::operators::core::Map;
+use timely::dataflow::channels::pact::ExchangeCore;
+use timely::Container;
 
-use crate::extensions::arrange::MzArrange;
-use crate::logging::initialize::ReachabilityEventRegion;
-use crate::logging::{EventQueue, LogCollection, LogVariant, TimelyLog};
-use crate::row_spine::{RowRowBatcher, RowRowBuilder};
-use crate::typedefs::{FlatKeyValBatcherDefault, RowRowSpine};
+use crate::extensions::arrange::MzArrangeCore;
+use crate::logging::initialize::ReachabilityEvent;
+use crate::logging::{consolidate_and_pack, EventQueue, LogCollection, LogVariant, TimelyLog};
+use crate::row_spine::RowRowBuilder;
+use crate::typedefs::RowRowSpine;
 
 /// Constructs the logging dataflow for reachability logs.
 ///
@@ -43,7 +39,7 @@ use crate::typedefs::{FlatKeyValBatcherDefault, RowRowSpine};
 pub(super) fn construct<A: Allocate>(
     worker: &mut timely::worker::Worker<A>,
     config: &LoggingConfig,
-    event_queue: EventQueue<FlatStack<ReachabilityEventRegion>>,
+    event_queue: EventQueue<Column<(Duration, ReachabilityEvent)>>,
 ) -> BTreeMap<LogVariant, LogCollection> {
     let interval_ms = std::cmp::max(1, config.interval.as_millis());
     let worker_index = worker.index();
@@ -52,16 +48,9 @@ pub(super) fn construct<A: Allocate>(
     // A dataflow for multiple log-derived arrangements.
     let traces = worker.dataflow_named("Dataflow: timely reachability logging", move |scope| {
         let enable_logging = config.enable_logging;
-        type UpdatesKey = (
-            bool,
-            OwnedRegionOpinion<Vec<usize>>,
-            usize,
-            usize,
-            Option<Timestamp>,
-        );
-        type UpdatesRegion = <((UpdatesKey, ()), Timestamp, Diff) as MzRegionPreference>::Region;
+        type UpdatesKey = (bool, Vec<usize>, usize, usize, Option<Timestamp>);
 
-        type CB = CapacityContainerBuilder<FlatStack<UpdatesRegion>>;
+        type CB = ColumnBuilder<((UpdatesKey, ()), Timestamp, Diff)>;
         let (updates, token) = Some(event_queue.link).mz_replay::<_, CB, _>(
             scope,
             "reachability logs",
@@ -76,7 +65,7 @@ pub(super) fn construct<A: Allocate>(
                 for (time, (addr, massaged)) in data.iter() {
                     let time_ms = ((time.as_millis() / interval_ms) + 1) * interval_ms;
                     let time_ms: Timestamp = time_ms.try_into().expect("must fit");
-                    for (source, port, update_type, ts, diff) in massaged {
+                    for (source, port, update_type, ts, diff) in massaged.into_iter() {
                         let datum = (update_type, addr, source, port, ts);
                         session.give(((datum, ()), time_ms, diff));
                     }
@@ -85,62 +74,37 @@ pub(super) fn construct<A: Allocate>(
         );
 
         // Restrict results by those logs that are meant to be active.
-        let logs_active = vec![LogVariant::Timely(TimelyLog::Reachability)];
+        let logs_active = [LogVariant::Timely(TimelyLog::Reachability)];
 
-        let updates = consolidate_pact::<
-            FlatKeyValBatcherDefault<UpdatesKey, (), Timestamp, Diff, _>,
-            _,
-            _,
-            _,
-            _,
-        >(&updates, Pipeline, "Consolidate Timely reachability")
-        .container::<FlatStack<UpdatesRegion>>();
+        let mut addr_row = Row::default();
+        let updates = consolidate_and_pack::<_, Col2ValBatcher<UpdatesKey, _, _, _>, ColumnBuilder<_>, _, _>(
+            &updates,
+            TimelyLog::Reachability,
+            move |((datum, ()), time, diff), packer, session| {
+                let (update_type, addr, source, port, ts) = datum;
+                let update_type = if *update_type { "source" } else { "target" };
+                addr_row.packer().push_list(
+                    IntoIterator::into_iter(addr)
+                        .chain(std::iter::once(source))
+                        .map(|id| Datum::UInt64(u64::cast_from(*id))),
+                );
+                let data = packer.pack_slice(&[
+                    addr_row.iter().next().unwrap(),
+                    Datum::UInt64(u64::cast_from(*port)),
+                    Datum::UInt64(u64::cast_from(worker_index)),
+                    Datum::String(update_type),
+                    Datum::from(*ts),
+                ]);
+                session.give((data, time, diff));
+            }
+        );
 
         let mut result = BTreeMap::new();
         for variant in logs_active {
             if config.index_logs.contains_key(&variant) {
-                let key = variant.index_by();
-                let (_, value) = permutation_for_arrangement(
-                    &key.iter()
-                        .cloned()
-                        .map(MirScalarExpr::Column)
-                        .collect::<Vec<_>>(),
-                    variant.desc().arity(),
-                );
-
-                let updates = updates
-                    .map(
-                        move |(((update_type, addr, source, port, ts), _), time, diff)| {
-                            let row_arena = RowArena::default();
-                            let update_type = if update_type { "source" } else { "target" };
-                            let binding = SharedRow::get();
-                            let mut row_builder = binding.borrow_mut();
-                            row_builder.packer().push_list(
-                                addr.iter()
-                                    .copied()
-                                    .chain_one(source)
-                                    .map(|id| Datum::UInt64(u64::cast_from(id))),
-                            );
-                            let datums = &[
-                                row_arena.push_unary_row(row_builder.clone()),
-                                Datum::UInt64(u64::cast_from(port)),
-                                Datum::UInt64(u64::cast_from(worker_index)),
-                                Datum::String(update_type),
-                                Datum::from(ts.clone()),
-                            ];
-                            row_builder.packer().extend(key.iter().map(|k| datums[*k]));
-                            let key_row = row_builder.clone();
-                            row_builder
-                                .packer()
-                                .extend(value.iter().map(|k| datums[*k]));
-                            let value_row = row_builder.clone();
-                            ((key_row, value_row), time, diff)
-                        },
-                    )
-                    .as_collection();
-
                 let trace = updates
-                    .mz_arrange::<RowRowBatcher<_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                        ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<Row, Row, Timestamp, Diff>),
                         &format!("Arrange {variant:?}"),
                     )
                     .trace;
diff --git a/src/compute/src/logging/timely.rs b/src/compute/src/logging/timely.rs
index 61fb4df11f758..f98f493af8fda 100644
--- a/src/compute/src/logging/timely.rs
+++ b/src/compute/src/logging/timely.rs
@@ -15,33 +15,31 @@ use std::rc::Rc;
 use std::time::Duration;
 
 use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
-use differential_dataflow::AsCollection;
 use mz_compute_client::logging::LoggingConfig;
 use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, Timestamp};
-use mz_timely_util::operator::consolidate_pact;
+use mz_timely_util::containers::{columnar_exchange, Col2ValBatcher, ColumnBuilder};
 use mz_timely_util::replay::MzReplay;
-use serde::{Deserialize, Serialize};
 use timely::communication::Allocate;
 use timely::container::columnation::{Columnation, CopyRegion};
 use timely::container::CapacityContainerBuilder;
-use timely::dataflow::channels::pact::Pipeline;
+use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
 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;
 use timely::logging::{
     ChannelsEvent, MessagesEvent, OperatesEvent, ParkEvent, ScheduleEvent, ShutdownEvent,
     TimelyEvent,
 };
+use timely::Container;
 use tracing::error;
 
-use crate::extensions::arrange::MzArrange;
+use crate::extensions::arrange::MzArrangeCore;
 use crate::logging::compute::{ComputeEvent, DataflowShutdown};
+use crate::logging::{consolidate_and_pack, LogCollection};
 use crate::logging::{EventQueue, LogVariant, SharedLoggingState, TimelyLog};
-use crate::logging::{LogCollection, PermutedRowPacker};
-use crate::row_spine::{RowRowBatcher, RowRowBuilder};
-use crate::typedefs::{KeyValBatcher, RowRowSpine};
+use crate::row_spine::RowRowBuilder;
+use crate::typedefs::{KeyBatcher, KeyValBatcher, RowRowSpine};
 
 /// Constructs the logging dataflow for timely logs.
 ///
@@ -61,20 +59,22 @@ pub(super) fn construct<A: Allocate>(
     let dataflow_index = worker.next_dataflow_index();
 
     worker.dataflow_named("Dataflow: timely logging", move |scope| {
-        let (mut logs, token) = Some(event_queue.link)
-            .mz_replay::<_, CapacityContainerBuilder<_>, _>(
+        let enable_logging = config.enable_logging;
+        let (logs, token) =
+            Some(event_queue.link).mz_replay::<_, CapacityContainerBuilder<_>, _>(
                 scope,
                 "timely logs",
                 config.interval,
                 event_queue.activator,
-                |mut session, data| session.give_iterator(data.iter()),
+                move |mut session, data| {
+                    // If logging is disabled, we still need to install the indexes, but we can leave them
+                    // empty. We do so by immediately filtering all logs events.
+                    if enable_logging {
+                        session.give_iterator(data.iter())
+                    }
+                },
             );
-
-        // If logging is disabled, we still need to install the indexes, but we can leave them
-        // empty. We do so by immediately filtering all logs events.
-        if !config.enable_logging {
-            logs = logs.filter(|_| false);
-        }
+        let logs = logs.container::<Vec<_>>();
 
         // Build a demux operator that splits the replayed event stream up into the separate
         // logging streams.
@@ -145,181 +145,162 @@ pub(super) fn construct<A: Allocate>(
         // We pre-arrange the logging streams to force a consolidation and reduce the amount of
         // updates that reach `Row` encoding.
 
-        let packer = PermutedRowPacker::new(TimelyLog::Operates);
-        let operates = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+        let operates = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
             &operates,
-            Pipeline,
-            "Consolidate Timely operates",
-        )
-        .as_collection()
-        .map(move |(id, name)| {
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(id)),
-                Datum::UInt64(u64::cast_from(worker_id)),
-                Datum::String(&name),
-            ])
-        });
-
-        let packer = PermutedRowPacker::new(TimelyLog::Channels);
-        let channels = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+            TimelyLog::Operates,
+            move |((id, name), time, diff), packer, session| {
+                let data = packer.pack_slice(&[
+                    Datum::UInt64(u64::cast_from(*id)),
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                    Datum::String(name),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
+
+        let channels = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
             &channels,
-            Pipeline,
-            "Consolidate Timely operates",
-        )
-        .as_collection()
-        .map(move |(datum, ())| {
-            let (source_node, source_port) = datum.source;
-            let (target_node, target_port) = datum.target;
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(datum.id)),
-                Datum::UInt64(u64::cast_from(worker_id)),
-                Datum::UInt64(u64::cast_from(source_node)),
-                Datum::UInt64(u64::cast_from(source_port)),
-                Datum::UInt64(u64::cast_from(target_node)),
-                Datum::UInt64(u64::cast_from(target_port)),
-            ])
-        });
-
-        let packer = PermutedRowPacker::new(TimelyLog::Addresses);
-        let addresses = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+            TimelyLog::Channels,
+            move |((datum, ()), time, diff), packer, session| {
+                let (source_node, source_port) = datum.source;
+                let (target_node, target_port) = datum.target;
+                let data = packer.pack_slice(&[
+                    Datum::UInt64(u64::cast_from(datum.id)),
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                    Datum::UInt64(u64::cast_from(source_node)),
+                    Datum::UInt64(u64::cast_from(source_port)),
+                    Datum::UInt64(u64::cast_from(target_node)),
+                    Datum::UInt64(u64::cast_from(target_port)),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
+
+        let addresses = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
             &addresses,
-            Pipeline,
-            "Consolidate Timely addresses",
-        )
-        .as_collection()
-        .map({
-            move |(id, address)| {
-                packer.pack_by_index(|packer, index| match index {
-                    0 => packer.push(Datum::UInt64(u64::cast_from(id))),
+            TimelyLog::Addresses,
+            move |((id, address), time, diff), packer, session| {
+                let data = packer.pack_by_index(|packer, index| match index {
+                    0 => packer.push(Datum::UInt64(u64::cast_from(*id))),
                     1 => packer.push(Datum::UInt64(u64::cast_from(worker_id))),
                     2 => {
                         packer.push_list(address.iter().map(|i| Datum::UInt64(u64::cast_from(*i))))
                     }
                     _ => unreachable!("Addresses relation has three columns"),
-                })
-            }
-        });
+                });
+                session.give((data, time, diff));
+            },
+        );
 
-        let packer = PermutedRowPacker::new(TimelyLog::Parks);
-        let parks = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+        let parks = consolidate_and_pack::<_, KeyBatcher<_, _, _>, ColumnBuilder<_>, _, _>(
             &parks,
-            Pipeline,
-            "Consolidate Timely parks",
-        )
-        .as_collection()
-        .map(move |(datum, ())| {
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(worker_id)),
-                Datum::UInt64(u64::try_from(datum.duration_pow).expect("duration too big")),
-                datum
-                    .requested_pow
-                    .map(|v| Datum::UInt64(v.try_into().expect("requested too big")))
-                    .unwrap_or(Datum::Null),
-            ])
-        });
-
-        let packer = PermutedRowPacker::new(TimelyLog::BatchesSent);
-        let batches_sent = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+            TimelyLog::Parks,
+            move |((datum, ()), time, diff), packer, session| {
+                let data = packer.pack_slice(&[
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                    Datum::UInt64(datum.duration_pow),
+                    datum
+                        .requested_pow
+                        .map(Datum::UInt64)
+                        .unwrap_or(Datum::Null),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
+
+        let batches_sent = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
             &batches_sent,
-            Pipeline,
-            "Consolidate Timely batches sent",
-        )
-        .as_collection()
-        .map(move |(datum, ())| {
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(datum.channel)),
-                Datum::UInt64(u64::cast_from(worker_id)),
-                Datum::UInt64(u64::cast_from(datum.worker)),
-            ])
-        });
-
-        let packer = PermutedRowPacker::new(TimelyLog::BatchesReceived);
-        let batches_received = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+            TimelyLog::BatchesSent,
+            move |((datum, ()), time, diff), packer, session| {
+                let data = packer.pack_slice(&[
+                    Datum::UInt64(u64::cast_from(datum.channel)),
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                    Datum::UInt64(u64::cast_from(datum.worker)),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
+
+        let batches_received = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
             &batches_received,
-            Pipeline,
-            "Consolidate Timely batches received",
-        )
-        .as_collection()
-        .map(move |(datum, ())| {
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(datum.channel)),
-                Datum::UInt64(u64::cast_from(datum.worker)),
-                Datum::UInt64(u64::cast_from(worker_id)),
-            ])
-        });
-
-        let packer = PermutedRowPacker::new(TimelyLog::MessagesSent);
-        let messages_sent = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+            TimelyLog::BatchesReceived,
+            move |((datum, ()), time, diff), packer, session| {
+                let data = packer.pack_slice(&[
+                    Datum::UInt64(u64::cast_from(datum.channel)),
+                    Datum::UInt64(u64::cast_from(datum.worker)),
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
+
+
+        let messages_sent = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
             &messages_sent,
-            Pipeline,
-            "Consolidate Timely messages sent",
-        )
-        .as_collection()
-        .map(move |(datum, ())| {
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(datum.channel)),
-                Datum::UInt64(u64::cast_from(worker_id)),
-                Datum::UInt64(u64::cast_from(datum.worker)),
-            ])
-        });
-
-        let packer = PermutedRowPacker::new(TimelyLog::MessagesReceived);
-        let messages_received = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+            TimelyLog::MessagesSent,
+            move |((datum, ()), time, diff), packer, session| {
+                let data = packer.pack_slice(&[
+                    Datum::UInt64(u64::cast_from(datum.channel)),
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                    Datum::UInt64(u64::cast_from(datum.worker)),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
+
+        let messages_received = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
             &messages_received,
-            Pipeline,
-            "Consolidate Timely messages received",
-        )
-        .as_collection()
-        .map(move |(datum, ())| {
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(datum.channel)),
-                Datum::UInt64(u64::cast_from(datum.worker)),
-                Datum::UInt64(u64::cast_from(worker_id)),
-            ])
-        });
-
-        let packer = PermutedRowPacker::new(TimelyLog::Elapsed);
-        let elapsed = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
+            TimelyLog::MessagesReceived,
+            move |((datum, ()), time, diff), packer, session| {
+                let data = packer.pack_slice(&[
+                    Datum::UInt64(u64::cast_from(datum.channel)),
+                    Datum::UInt64(u64::cast_from(datum.worker)),
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
+
+        let elapsed = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
             &schedules_duration,
-            Pipeline,
-            "Consolidate Timely duration",
-        )
-        .as_collection()
-        .map(move |(operator, _)| {
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(operator)),
-                Datum::UInt64(u64::cast_from(worker_id)),
-            ])
-        });
+            TimelyLog::Elapsed,
+            move |((operator, ()), time, diff), packer, session| {
+                let data = packer.pack_slice(&[Datum::UInt64(u64::cast_from(*operator)),
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
 
-        let packer = PermutedRowPacker::new(TimelyLog::Histogram);
-        let histogram = consolidate_pact::<KeyValBatcher<_, _, _, _>, _, _, _, _>(
-            &schedules_histogram,
-            Pipeline,
-            "Consolidate Timely histogram",
-        )
-        .as_collection()
-        .map(move |(datum, _)| {
-            packer.pack_slice(&[
-                Datum::UInt64(u64::cast_from(datum.operator)),
-                Datum::UInt64(u64::cast_from(worker_id)),
-                Datum::UInt64(u64::try_from(datum.duration_pow).expect("duration too big")),
-            ])
-        });
 
-        use TimelyLog::*;
-        let logs = [
-            (Operates, operates),
-            (Channels, channels),
-            (Elapsed, elapsed),
-            (Histogram, histogram),
-            (Addresses, addresses),
-            (Parks, parks),
-            (MessagesSent, messages_sent),
-            (MessagesReceived, messages_received),
-            (BatchesSent, batches_sent),
-            (BatchesReceived, batches_received),
-        ];
+        let histogram = consolidate_and_pack::<_, KeyValBatcher<_, _, _, _>, ColumnBuilder<_>, _, _>(
+            &schedules_histogram,
+            TimelyLog::Histogram,
+            move |((datum, ()), time, diff), packer, session| {
+                let data = packer.pack_slice(&[
+                    Datum::UInt64(u64::cast_from(datum.operator)),
+                    Datum::UInt64(u64::cast_from(worker_id)),
+                    Datum::UInt64(datum.duration_pow),
+                ]);
+                session.give((data, time, diff));
+            },
+        );
+
+        let logs = {
+            use TimelyLog::*;
+            [
+                (Operates, operates),
+                (Channels, channels),
+                (Elapsed, elapsed),
+                (Histogram, histogram),
+                (Addresses, addresses),
+                (Parks, parks),
+                (MessagesSent, messages_sent),
+                (MessagesReceived, messages_received),
+                (BatchesSent, batches_sent),
+                (BatchesReceived, batches_received),
+            ]
+        };
 
         // Build the output arrangements.
         let mut result = BTreeMap::new();
@@ -327,7 +308,8 @@ pub(super) fn construct<A: Allocate>(
             let variant = LogVariant::Timely(variant);
             if config.index_logs.contains_key(&variant) {
                 let trace = collection
-                    .mz_arrange::<RowRowBatcher<_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                        ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<mz_repr::Row, mz_repr::Row, Timestamp, Diff>),
                         &format!("Arrange {variant:?}"),
                     )
                     .trace;
@@ -403,7 +385,7 @@ struct DemuxOutput<'a> {
     schedules_histogram: OutputSession<'a, (ScheduleHistogramDatum, ())>,
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 struct ChannelDatum {
     id: usize,
     source: (usize, usize),
@@ -414,17 +396,17 @@ impl Columnation for ChannelDatum {
     type InnerRegion = CopyRegion<Self>;
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 struct ParkDatum {
-    duration_pow: u128,
-    requested_pow: Option<u128>,
+    duration_pow: u64,
+    requested_pow: Option<u64>,
 }
 
 impl Columnation for ParkDatum {
     type InnerRegion = CopyRegion<Self>;
 }
 
-#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 struct MessageDatum {
     channel: usize,
     worker: usize,
@@ -434,10 +416,10 @@ impl Columnation for MessageDatum {
     type InnerRegion = CopyRegion<Self>;
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 struct ScheduleHistogramDatum {
     operator: usize,
-    duration_pow: u128,
+    duration_pow: u64,
 }
 
 impl Columnation for ScheduleHistogramDatum {
@@ -534,8 +516,7 @@ impl DemuxHandler<'_, '_> {
 
         // Retract schedules information for the operator
         if let Some(schedules) = self.state.schedules_data.remove(&event.id) {
-            for (bucket, (count, elapsed_ns)) in schedules
-                .into_iter()
+            for (bucket, (count, elapsed_ns)) in IntoIterator::into_iter(schedules)
                 .enumerate()
                 .filter(|(_, (count, _))| *count != 0)
             {
@@ -566,7 +547,7 @@ impl DemuxHandler<'_, '_> {
     fn handle_dataflow_shutdown(&mut self, dataflow_index: usize) {
         // Notify compute logging about the shutdown.
         if let Some(logger) = &self.shared_state.compute_logger {
-            logger.log(ComputeEvent::DataflowShutdown(DataflowShutdown {
+            logger.log(&ComputeEvent::DataflowShutdown(DataflowShutdown {
                 dataflow_index,
             }));
         }
@@ -635,13 +616,16 @@ impl DemuxHandler<'_, '_> {
             }
             ParkEvent::Unpark => {
                 let Some(park) = self.state.last_park.take() else {
-                    error!("unpark without a preceeding park");
+                    error!("unpark without a preceding park");
                     return;
                 };
 
                 let duration_ns = self.time.saturating_sub(park.time).as_nanos();
-                let duration_pow = duration_ns.next_power_of_two();
-                let requested_pow = park.requested.map(|r| r.as_nanos().next_power_of_two());
+                let duration_pow =
+                    u64::try_from(duration_ns.next_power_of_two()).expect("must fit");
+                let requested_pow = park
+                    .requested
+                    .map(|r| u64::try_from(r.as_nanos().next_power_of_two()).expect("must fit"));
 
                 let ts = self.ts();
                 let datum = ParkDatum {
@@ -706,7 +690,7 @@ impl DemuxHandler<'_, '_> {
 
                 let elapsed_ns = self.time.saturating_sub(start_time).as_nanos();
                 let elapsed_diff = Diff::try_from(elapsed_ns).expect("must fit");
-                let elapsed_pow = elapsed_ns.next_power_of_two();
+                let elapsed_pow = u64::try_from(elapsed_ns.next_power_of_two()).expect("must fit");
 
                 let ts = self.ts();
                 let datum = event.id;
diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs
index c5dd45ed23e08..83d2cd12d1f9f 100644
--- a/src/compute/src/render.rs
+++ b/src/compute/src/render.rs
@@ -956,7 +956,7 @@ where
             // ActiveComputeState can't have a catalog reference, so we'll need to capture the names
             // in some other structure and have that structure impl ExprHumanizer
             let metadata = if should_compute_lir_metadata {
-                let operator: Box<str> = node.expr.humanize(&DummyHumanizer).into();
+                let operator = node.expr.humanize(&DummyHumanizer);
                 let operator_id_start = self.scope.peek_identifier();
                 Some((operator, operator_id_start))
             } else {
@@ -983,8 +983,7 @@ where
         }
 
         if let Some(lir_mapping_metadata) = lir_mapping_metadata {
-            let mapping: Box<[(LirId, LirMetadata)]> = lir_mapping_metadata.into();
-            self.log_lir_mapping(object_id, mapping);
+            self.log_lir_mapping(object_id, lir_mapping_metadata);
         }
 
         collections
@@ -1186,16 +1185,16 @@ where
 
     fn log_dataflow_global_id(&self, dataflow_index: usize, global_id: GlobalId) {
         if let Some(logger) = &self.compute_logger {
-            logger.log(ComputeEvent::DataflowGlobal(DataflowGlobal {
+            logger.log(&ComputeEvent::DataflowGlobal(DataflowGlobal {
                 dataflow_index,
                 global_id,
             }));
         }
     }
 
-    fn log_lir_mapping(&self, global_id: GlobalId, mapping: Box<[(LirId, LirMetadata)]>) {
+    fn log_lir_mapping(&self, global_id: GlobalId, mapping: Vec<(LirId, LirMetadata)>) {
         if let Some(logger) = &self.compute_logger {
-            logger.log(ComputeEvent::LirMapping(LirMapping { global_id, mapping }));
+            logger.log(&ComputeEvent::LirMapping(LirMapping { global_id, mapping }));
         }
     }
 
diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs
index a4440192356ca..ffb763b764be3 100644
--- a/src/compute/src/server.rs
+++ b/src/compute/src/server.rs
@@ -744,7 +744,7 @@ impl<'w, A: Allocate + 'static> Worker<'w, A> {
             for (_, peek) in std::mem::take(&mut compute_state.pending_peeks) {
                 // Log dropping the peek request.
                 if let Some(logger) = compute_state.compute_logger.as_mut() {
-                    logger.log(peek.as_log_event(false));
+                    logger.log(&peek.as_log_event(false));
                 }
             }
 
diff --git a/src/compute/src/sink/copy_to_s3_oneshot.rs b/src/compute/src/sink/copy_to_s3_oneshot.rs
index 4fb0617132a1a..e338e4ebf628a 100644
--- a/src/compute/src/sink/copy_to_s3_oneshot.rs
+++ b/src/compute/src/sink/copy_to_s3_oneshot.rs
@@ -23,6 +23,7 @@ use mz_storage_types::controller::CollectionMetadata;
 use mz_storage_types::errors::DataflowError;
 use mz_timely_util::operator::consolidate_pact;
 use timely::dataflow::channels::pact::{Exchange, Pipeline};
+use timely::dataflow::operators::core::Map;
 use timely::dataflow::operators::Operator;
 use timely::dataflow::Scope;
 use timely::progress::Antichain;
@@ -66,7 +67,7 @@ where
         // is necessary to ensure the files written from each compute replica are identical.
         // While this is not technically guaranteed, the current implementation uses a FIFO channel.
         // In the storage copy_to operator we assert the ordering of rows to detect any regressions.
-        let input = consolidate_pact::<KeyBatcher<_, _, _>, _, _, _, _>(
+        let input = consolidate_pact::<KeyBatcher<_, _, _>, _, _>(
             &sinked_collection
                 .map(move |row| {
                     let batch = row.hashed() % batch_count;
@@ -75,11 +76,12 @@ where
                 .inner,
             Exchange::new(move |(((_, batch), _), _, _)| *batch),
             "Consolidated COPY TO S3 input",
-        )
-        .as_collection();
+        );
+        // TODO: We're converting a stream of region-allocated data to a stream of vectors.
+        let input = input.map(Clone::clone).as_collection();
 
         // We need to consolidate the error collection to ensure we don't act on retracted errors.
-        let error = consolidate_pact::<KeyBatcher<_, _, _>, _, _, _, _>(
+        let error = consolidate_pact::<KeyBatcher<_, _, _>, _, _>(
             &err_collection
                 .map(move |row| {
                     let batch = row.hashed() % batch_count;
@@ -88,8 +90,8 @@ where
                 .inner,
             Exchange::new(move |(((_, batch), _), _, _)| *batch),
             "Consolidated COPY TO S3 errors",
-        )
-        .container::<Vec<_>>();
+        );
+
         // We can only propagate the one error back to the client, so filter the error
         // collection to the first error that is before the sink 'up_to' to avoid
         // sending the full error collection to the next operator. We ensure we find the
@@ -102,7 +104,9 @@ where
                     while let Some((time, data)) = input.next() {
                         if !up_to.less_equal(time.time()) && !received_one {
                             received_one = true;
-                            output.session(&time).give_iterator(data.drain(..1));
+                            output
+                                .session(&time)
+                                .give_iterator(data.iter().next().cloned().into_iter());
                         }
                     }
                 }
diff --git a/src/repr/Cargo.toml b/src/repr/Cargo.toml
index d09a22d1697b5..4a4f4ff354640 100644
--- a/src/repr/Cargo.toml
+++ b/src/repr/Cargo.toml
@@ -31,7 +31,7 @@ arrow = { version = "53.3.0", default-features = false }
 bitflags = "1.3.2"
 bytes = "1.3.0"
 cfg-if = "1.0.0"
-columnar = "0.2.0"
+columnar = "0.2.2"
 columnation = "0.1.0"
 chrono = { version = "0.4.35", default-features = false, features = ["serde", "std"] }
 compact_bytes = "0.1.2"
diff --git a/src/repr/src/global_id.rs b/src/repr/src/global_id.rs
index b570a245b549f..a09def03252ba 100644
--- a/src/repr/src/global_id.rs
+++ b/src/repr/src/global_id.rs
@@ -11,6 +11,7 @@ use std::fmt;
 use std::str::FromStr;
 
 use anyhow::{anyhow, Error};
+use columnar::Columnar;
 use columnation::{Columnation, CopyRegion};
 use mz_lowertest::MzReflect;
 use mz_ore::id_gen::AtomicIdGen;
@@ -50,6 +51,7 @@ include!(concat!(env!("OUT_DIR"), "/mz_repr.global_id.rs"));
     Serialize,
     Deserialize,
     MzReflect,
+    Columnar,
 )]
 pub enum GlobalId {
     /// System namespace.
diff --git a/src/repr/src/timestamp.rs b/src/repr/src/timestamp.rs
index c5e689e2b11f5..e00efa747f497 100644
--- a/src/repr/src/timestamp.rs
+++ b/src/repr/src/timestamp.rs
@@ -11,6 +11,7 @@ use std::convert::TryFrom;
 use std::num::TryFromIntError;
 use std::time::Duration;
 
+use columnar::Columnar;
 use dec::TryFromDecimalError;
 use mz_proto::{RustType, TryFromProtoError};
 use proptest_derive::Arbitrary;
@@ -34,7 +35,9 @@ include!(concat!(env!("OUT_DIR"), "/mz_repr.timestamp.rs"));
     Hash,
     Default,
     Arbitrary,
+    Columnar,
 )]
+#[columnar(derive(PartialEq, Eq, PartialOrd, Ord))]
 pub struct Timestamp {
     /// note no `pub`.
     internal: u64,
diff --git a/src/timely-util/Cargo.toml b/src/timely-util/Cargo.toml
index 071b181bb7b9f..9d895e52c755b 100644
--- a/src/timely-util/Cargo.toml
+++ b/src/timely-util/Cargo.toml
@@ -12,6 +12,8 @@ workspace = true
 [dependencies]
 ahash = { version = "0.8.11", default-features = false }
 bincode = "1.3.3"
+bytemuck = "1.21.0"
+columnar = "0.2.2"
 columnation = "0.1.0"
 differential-dataflow = "0.13.3"
 either = "1"
diff --git a/src/timely-util/src/containers.rs b/src/timely-util/src/containers.rs
index 9431900d8edfd..8de4cc20e4cc8 100644
--- a/src/timely-util/src/containers.rs
+++ b/src/timely-util/src/containers.rs
@@ -9,5 +9,433 @@
 
 //! Reusable containers.
 
+use std::hash::Hash;
+
+use columnar::Columnar;
+use differential_dataflow::trace::implementations::merge_batcher::{ColMerger, MergeBatcher};
+use differential_dataflow::Hashable;
+use timely::container::columnation::TimelyStack;
+
 pub mod array;
 pub mod stack;
+
+pub use container::Column;
+
+mod container {
+    use columnar::bytes::serialization::decode;
+    use columnar::common::IterOwn;
+    use columnar::Columnar;
+    use columnar::Container as _;
+    use columnar::{AsBytes, Clear, FromBytes, Index, Len};
+    use mz_ore::cast::CastFrom;
+    use timely::bytes::arc::Bytes;
+    use timely::container::PushInto;
+    use timely::dataflow::channels::ContainerBytes;
+    use timely::Container;
+
+    /// A container based on a columnar store, encoded in aligned bytes.
+    ///
+    /// The type can represent typed data, bytes from Timely, or an aligned allocation. The name
+    /// is singular to express that the preferred format is [`Column::Align`]. The [`Column::Typed`]
+    /// variant is used to construct the container, and it owns potentially multiple columns of data.
+    pub enum Column<C: Columnar> {
+        /// The typed variant of the container.
+        Typed(C::Container),
+        /// The binary variant of the container.
+        Bytes(Bytes),
+        /// Relocated, aligned binary data, if `Bytes` doesn't work for some reason.
+        ///
+        /// Reasons could include misalignment, cloning of data, or wanting
+        /// to release the `Bytes` as a scarce resource.
+        Align(Box<[u64]>),
+    }
+
+    impl<C: Columnar> Column<C> {
+        /// Borrows the container as a reference.
+        fn borrow(&self) -> <C::Container as columnar::Container<C>>::Borrowed<'_> {
+            match self {
+                Column::Typed(t) => t.borrow(),
+                Column::Bytes(b) => {
+                    <<C::Container as columnar::Container<C>>::Borrowed<'_>>::from_bytes(
+                        &mut decode(bytemuck::cast_slice(b)),
+                    )
+                }
+                Column::Align(a) => {
+                    <<C::Container as columnar::Container<C>>::Borrowed<'_>>::from_bytes(
+                        &mut decode(a),
+                    )
+                }
+            }
+        }
+    }
+
+    impl<C: Columnar> Default for Column<C> {
+        fn default() -> Self {
+            Self::Typed(Default::default())
+        }
+    }
+
+    impl<C: Columnar> Clone for Column<C>
+    where
+        C::Container: Clone,
+    {
+        fn clone(&self) -> Self {
+            match self {
+                // Typed stays typed, although we would have the option to move to aligned data.
+                // If we did it might be confusing why we couldn't push into a cloned column.
+                Column::Typed(t) => Column::Typed(t.clone()),
+                Column::Bytes(b) => {
+                    assert_eq!(b.len() % 8, 0);
+                    let mut alloc: Vec<u64> = vec![0; b.len() / 8];
+                    bytemuck::cast_slice_mut(&mut alloc[..]).copy_from_slice(&b[..]);
+                    Self::Align(alloc.into())
+                }
+                Column::Align(a) => Column::Align(a.clone()),
+            }
+        }
+    }
+
+    impl<C: Columnar> Container for Column<C> {
+        type ItemRef<'a> = C::Ref<'a>;
+        type Item<'a> = C::Ref<'a>;
+
+        fn len(&self) -> usize {
+            self.borrow().len()
+        }
+
+        // This sets the `Bytes` variant to be an empty `Typed` variant, appropriate for pushing into.
+        fn clear(&mut self) {
+            match self {
+                Column::Typed(t) => t.clear(),
+                Column::Bytes(_) => *self = Column::Typed(Default::default()),
+                Column::Align(_) => *self = Column::Typed(Default::default()),
+            }
+        }
+
+        type Iter<'a> = IterOwn<<C::Container as columnar::Container<C>>::Borrowed<'a>>;
+
+        fn iter(&self) -> Self::Iter<'_> {
+            self.borrow().into_iter()
+        }
+
+        type DrainIter<'a> = IterOwn<<C::Container as columnar::Container<C>>::Borrowed<'a>>;
+
+        fn drain(&mut self) -> Self::DrainIter<'_> {
+            self.borrow().into_iter()
+        }
+    }
+
+    impl<C: Columnar, T> PushInto<T> for Column<C>
+    where
+        C::Container: columnar::Push<T>,
+    {
+        #[inline]
+        fn push_into(&mut self, item: T) {
+            use columnar::Push;
+            match self {
+                Column::Typed(t) => t.push(item),
+                Column::Align(_) | Column::Bytes(_) => {
+                    // We really oughtn't be calling this in this case.
+                    // We could convert to owned, but need more constraints on `C`.
+                    unimplemented!("Pushing into Column::Bytes without first clearing");
+                }
+            }
+        }
+    }
+
+    impl<C: Columnar> ContainerBytes for Column<C> {
+        fn from_bytes(bytes: Bytes) -> Self {
+            // Our expectation / hope is that `bytes` is `u64` aligned and sized.
+            // If the alignment is borked, we can relocate. If the size is borked,
+            // not sure what we do in that case. An incorrect size indicates a problem
+            // of `into_bytes`, or a failure of the communication layer, both of which
+            // are unrecoverable.
+            assert_eq!(bytes.len() % 8, 0);
+            if let Ok(_) = bytemuck::try_cast_slice::<_, u64>(&bytes) {
+                Self::Bytes(bytes)
+            } else {
+                // We failed to cast the slice, so we'll reallocate.
+                let mut alloc: Vec<u64> = vec![0; bytes.len() / 8];
+                bytemuck::cast_slice_mut(&mut alloc[..]).copy_from_slice(&bytes[..]);
+                Self::Align(alloc.into())
+            }
+        }
+
+        fn length_in_bytes(&self) -> usize {
+            match self {
+                Column::Typed(t) => 8 * t.borrow().length_in_words(),
+                Column::Bytes(b) => b.len(),
+                Column::Align(a) => 8 * a.len(),
+            }
+        }
+
+        fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
+            match self {
+                Column::Typed(t) => {
+                    use columnar::Container;
+                    // Columnar data is serialized as a sequence of `u64` values, with each `[u8]` slice
+                    // serialize as first its length in bytes, and then as many `u64` values as needed.
+                    // Padding should be added, but only for alignment; no specific values are required.
+                    for (align, bytes) in t.borrow().as_bytes() {
+                        assert!(align <= 8);
+                        let length = u64::cast_from(bytes.len());
+                        writer
+                            .write_all(bytemuck::cast_slice(std::slice::from_ref(&length)))
+                            .unwrap();
+                        writer.write_all(bytes).unwrap();
+                        let padding = usize::cast_from((8 - (length % 8)) % 8);
+                        writer.write_all(&[0; 8][..padding]).unwrap();
+                    }
+                }
+                Column::Bytes(b) => writer.write_all(b).unwrap(),
+                Column::Align(a) => writer.write_all(bytemuck::cast_slice(a)).unwrap(),
+            }
+        }
+    }
+}
+
+pub use builder::ColumnBuilder;
+mod builder {
+    use std::collections::VecDeque;
+
+    use columnar::{AsBytes, Clear, Columnar, Len, Push};
+    use timely::container::PushInto;
+    use timely::container::{ContainerBuilder, LengthPreservingContainerBuilder};
+
+    use super::Column;
+
+    /// A container builder for `Column<C>`.
+    pub struct ColumnBuilder<C: Columnar> {
+        /// Container that we're writing to.
+        current: C::Container,
+        /// Finished container that we presented to callers of extract/finish.
+        ///
+        /// We don't recycle the column because for extract, it's not typed, and after calls
+        /// to finish it'll be `None`.
+        finished: Option<Column<C>>,
+        /// Completed containers pending to be sent.
+        pending: VecDeque<Column<C>>,
+    }
+
+    impl<C: Columnar, T> PushInto<T> for ColumnBuilder<C>
+    where
+        C::Container: Push<T>,
+    {
+        #[inline]
+        fn push_into(&mut self, item: T) {
+            self.current.push(item);
+            // If there is less than 10% slop with 2MB backing allocations, mint a container.
+            use columnar::Container;
+            let words = self.current.borrow().length_in_words();
+            let round = (words + ((1 << 18) - 1)) & !((1 << 18) - 1);
+            if round - words < round / 10 {
+                let mut alloc = Vec::with_capacity(round);
+                columnar::bytes::serialization::encode(
+                    &mut alloc,
+                    self.current.borrow().as_bytes(),
+                );
+                self.pending
+                    .push_back(Column::Align(alloc.into_boxed_slice()));
+                self.current.clear();
+            }
+        }
+    }
+
+    impl<C: Columnar> Default for ColumnBuilder<C> {
+        fn default() -> Self {
+            ColumnBuilder {
+                current: Default::default(),
+                finished: None,
+                pending: Default::default(),
+            }
+        }
+    }
+
+    impl<C: Columnar> ContainerBuilder for ColumnBuilder<C>
+    where
+        C::Container: Clone,
+    {
+        type Container = Column<C>;
+
+        #[inline]
+        fn extract(&mut self) -> Option<&mut Self::Container> {
+            if let Some(container) = self.pending.pop_front() {
+                self.finished = Some(container);
+                self.finished.as_mut()
+            } else {
+                None
+            }
+        }
+
+        #[inline]
+        fn finish(&mut self) -> Option<&mut Self::Container> {
+            if !self.current.is_empty() {
+                self.pending
+                    .push_back(Column::Typed(std::mem::take(&mut self.current)));
+            }
+            self.finished = self.pending.pop_front();
+            self.finished.as_mut()
+        }
+    }
+
+    impl<C: Columnar> LengthPreservingContainerBuilder for ColumnBuilder<C> where C::Container: Clone {}
+}
+
+/// A batcher for columnar storage.
+pub type Col2ValBatcher<K, V, T, R> = MergeBatcher<
+    Column<((K, V), T, R)>,
+    batcher::Chunker<TimelyStack<((K, V), T, R)>>,
+    ColMerger<(K, V), T, R>,
+>;
+pub type Col2KeyBatcher<K, T, R> = Col2ValBatcher<K, (), T, R>;
+
+/// An exchange function for columnar tuples of the form `((K, V), T, D)`. Rust has a hard
+/// time to figure out the lifetimes of the elements when specified as a closure, so we rather
+/// specify it as a function.
+#[inline(always)]
+pub fn columnar_exchange<K, V, T, D>(((k, _), _, _): &<((K, V), T, D) as Columnar>::Ref<'_>) -> u64
+where
+    K: Columnar,
+    for<'a> K::Ref<'a>: Hash,
+    V: Columnar,
+    D: Columnar,
+    T: Columnar,
+{
+    k.hashed()
+}
+
+/// Types for consolidating, merging, and extracting columnar update collections.
+pub mod batcher {
+    use std::collections::VecDeque;
+
+    use columnar::Columnar;
+    use differential_dataflow::difference::Semigroup;
+    use timely::container::{ContainerBuilder, PushInto};
+    use timely::Container;
+
+    use crate::containers::Column;
+
+    #[derive(Default)]
+    pub struct Chunker<C> {
+        /// Buffer into which we'll consolidate.
+        ///
+        /// Also the buffer where we'll stage responses to `extract` and `finish`.
+        /// When these calls return, the buffer is available for reuse.
+        target: C,
+        /// Consolidated buffers ready to go.
+        ready: VecDeque<C>,
+    }
+
+    impl<C: Container + Clone + 'static> ContainerBuilder for Chunker<C> {
+        type Container = C;
+
+        fn extract(&mut self) -> Option<&mut Self::Container> {
+            if let Some(ready) = self.ready.pop_front() {
+                self.target = ready;
+                Some(&mut self.target)
+            } else {
+                None
+            }
+        }
+
+        fn finish(&mut self) -> Option<&mut Self::Container> {
+            self.extract()
+        }
+    }
+
+    impl<'a, D, T, R, C2> PushInto<&'a mut Column<(D, T, R)>> for Chunker<C2>
+    where
+        D: Columnar,
+        for<'b> D::Ref<'b>: Ord + Copy,
+        T: Columnar,
+        for<'b> T::Ref<'b>: Ord + Copy,
+        R: Columnar + Semigroup + for<'b> Semigroup<R::Ref<'b>>,
+        for<'b> R::Ref<'b>: Ord,
+        C2: Container + for<'b> PushInto<&'b (D, T, R)>,
+    {
+        fn push_into(&mut self, container: &'a mut Column<(D, T, R)>) {
+            // Sort input data
+            // TODO: consider `Vec<usize>` that we retain, containing indexes.
+            let mut permutation = Vec::with_capacity(container.len());
+            permutation.extend(container.drain());
+            permutation.sort();
+
+            self.target.clear();
+            // Iterate over the data, accumulating diffs for like keys.
+            let mut iter = permutation.drain(..);
+            if let Some((data, time, diff)) = iter.next() {
+                let mut owned_data = D::into_owned(data);
+                let mut owned_time = T::into_owned(time);
+
+                let mut prev_data = data;
+                let mut prev_time = time;
+                let mut prev_diff = <R as Columnar>::into_owned(diff);
+
+                for (data, time, diff) in iter {
+                    if (&prev_data, &prev_time) == (&data, &time) {
+                        prev_diff.plus_equals(&diff);
+                    } else {
+                        if !prev_diff.is_zero() {
+                            D::copy_from(&mut owned_data, prev_data);
+                            T::copy_from(&mut owned_time, prev_time);
+                            let tuple = (owned_data, owned_time, prev_diff);
+                            self.target.push_into(&tuple);
+                            (owned_data, owned_time, prev_diff) = tuple;
+                        }
+                        prev_data = data;
+                        prev_time = time;
+                        R::copy_from(&mut prev_diff, diff);
+                    }
+                }
+
+                if !prev_diff.is_zero() {
+                    D::copy_from(&mut owned_data, prev_data);
+                    T::copy_from(&mut owned_time, prev_time);
+                    let tuple = (owned_data, owned_time, prev_diff);
+                    self.target.push_into(&tuple);
+                }
+            }
+
+            if !self.target.is_empty() {
+                self.ready.push_back(std::mem::take(&mut self.target));
+            }
+        }
+    }
+}
+
+pub use provided_builder::ProvidedBuilder;
+
+mod provided_builder {
+    use timely::container::ContainerBuilder;
+    use timely::Container;
+
+    /// A container builder that doesn't support pushing elements, and is only suitable for pushing
+    /// whole containers at Timely sessions. See [`give_container`] for more information.
+    ///
+    ///  [`give_container`]: timely::dataflow::channels::pushers::buffer::Session::give_container
+    pub struct ProvidedBuilder<C> {
+        _marker: std::marker::PhantomData<C>,
+    }
+
+    impl<C> Default for ProvidedBuilder<C> {
+        fn default() -> Self {
+            Self {
+                _marker: std::marker::PhantomData,
+            }
+        }
+    }
+
+    impl<C: Container + Clone + 'static> ContainerBuilder for ProvidedBuilder<C> {
+        type Container = C;
+
+        #[inline(always)]
+        fn extract(&mut self) -> Option<&mut Self::Container> {
+            None
+        }
+
+        #[inline(always)]
+        fn finish(&mut self) -> Option<&mut Self::Container> {
+            None
+        }
+    }
+}
diff --git a/src/timely-util/src/operator.rs b/src/timely-util/src/operator.rs
index 1fa5b9a1503be..8254974de00af 100644
--- a/src/timely-util/src/operator.rs
+++ b/src/timely-util/src/operator.rs
@@ -21,10 +21,11 @@ use differential_dataflow::logging::DifferentialEventBuilder;
 use differential_dataflow::trace::{Batcher, Builder, Description};
 use differential_dataflow::{AsCollection, Collection, Hashable};
 use timely::container::columnation::{Columnation, TimelyStack};
-use timely::container::{ContainerBuilder, PushInto, SizableContainer};
+use timely::container::{ContainerBuilder, PushInto};
 use timely::dataflow::channels::pact::{Exchange, ParallelizationContract, Pipeline};
 use timely::dataflow::channels::pushers::Tee;
 use timely::dataflow::channels::ContainerBytes;
+use timely::dataflow::operators::core::Map;
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder as OperatorBuilderRc;
 use timely::dataflow::operators::generic::operator::{self, Operator};
 use timely::dataflow::operators::generic::{InputHandleCore, OperatorInfo, OutputHandleCore};
@@ -706,10 +707,9 @@ where
                 data.hash(&mut h);
                 h.finish()
             });
-            // Access to `arrange_core` is OK because we specify the trace and don't hold on to it.
-            consolidate_pact::<Ba, _, _, _, _>(&self.map(|k| (k, ())).inner, exchange, name)
+            consolidate_pact::<Ba, _, _>(&self.map(|k| (k, ())).inner, exchange, name)
+                .map(|((k, ()), time, diff)| (k.clone(), time.clone(), diff.clone()))
                 .as_collection()
-                .map(|(k, ())| k)
         } else {
             self
         }
@@ -729,9 +729,9 @@ where
         let exchange =
             Exchange::new(move |update: &((D1, ()), G::Timestamp, R)| (update.0).0.hashed());
 
-        consolidate_pact::<Ba, _, _, _, _>(&self.map(|k| (k, ())).inner, exchange, name)
+        consolidate_pact::<Ba, _, _>(&self.map(|k| (k, ())).inner, exchange, name)
+            .map(|((k, ()), time, diff)| (k.clone(), time.clone(), diff.clone()))
             .as_collection()
-            .map(|(k, ())| k)
     }
 }
 
@@ -773,19 +773,17 @@ where
 /// The data are accumulated in place, each held back until their timestamp has completed.
 ///
 /// This serves as a low-level building-block for more user-friendly functions.
-pub fn consolidate_pact<B, P, G, CI, CO>(
-    stream: &StreamCore<G, CI>,
+pub fn consolidate_pact<Ba, P, G>(
+    stream: &StreamCore<G, Ba::Input>,
     pact: P,
     name: &str,
-) -> StreamCore<G, CO>
+) -> StreamCore<G, Ba::Output>
 where
     G: Scope,
-    B: Batcher<Input = CI, Time = G::Timestamp> + 'static,
-    B::Output: Container,
-    for<'a> CO: PushInto<<B::Output as Container>::Item<'a>>,
-    P: ParallelizationContract<G::Timestamp, CI>,
-    CI: Container + Data,
-    CO: SizableContainer + Data,
+    Ba: Batcher<Time = G::Timestamp> + 'static,
+    Ba::Input: Container + Clone + 'static,
+    Ba::Output: Container + Clone,
+    P: ParallelizationContract<G::Timestamp, Ba::Input>,
 {
     stream.unary_frontier(pact, name, |_cap, info| {
         // Acquire a logger for arrange events.
@@ -797,7 +795,7 @@ where
                 .map(Into::into)
         };
 
-        let mut batcher = B::new(logger, info.global_id);
+        let mut batcher = Ba::new(logger, info.global_id);
         // Capabilities for the lower envelope of updates in `batcher`.
         let mut capabilities = Antichain::<Capability<G::Timestamp>>::new();
         let mut prev_frontier = Antichain::from_elem(G::Timestamp::minimum());
@@ -834,7 +832,7 @@ where
                             let mut session = output.session(&capabilities.elements()[index]);
                             // Extract updates not in advance of `upper`.
                             let output =
-                                batcher.seal::<ConsolidateBuilder<_, B::Output, CO>>(upper.clone());
+                                batcher.seal::<ConsolidateBuilder<_, Ba::Output>>(upper.clone());
                             for mut batch in output {
                                 session.give_container(&mut batch);
                             }
@@ -870,25 +868,21 @@ where
 }
 
 /// A builder that wraps a session for direct output to a stream.
-struct ConsolidateBuilder<T, I, O> {
-    buffer: Vec<O>,
+struct ConsolidateBuilder<T, I> {
     _marker: PhantomData<(T, I)>,
 }
 
-impl<T, I, O> Builder for ConsolidateBuilder<T, I, O>
+impl<T, I> Builder for ConsolidateBuilder<T, I>
 where
     T: Timestamp,
     I: Container,
-    O: SizableContainer,
-    for<'a> O: PushInto<I::Item<'a>>,
 {
     type Input = I;
     type Time = T;
-    type Output = Vec<O>;
+    type Output = Vec<I>;
 
     fn new() -> Self {
         Self {
-            buffer: Vec::default(),
             _marker: PhantomData,
         }
     }
@@ -897,35 +891,15 @@ where
         Self::new()
     }
 
-    fn push(&mut self, chunk: &mut Self::Input) {
-        // TODO(mh): This is less efficient than it could be because it extracts each item
-        // individually and then pushes it. However, it is not a regression over the previous
-        // implementation. In the future, we want to either clone many elements in one go,
-        // or ensure that `Vec<Input>` == `Output`, which would avoid looking at the container
-        // contents at all.
-        'element: for element in chunk.drain() {
-            if let Some(last) = self.buffer.last_mut() {
-                if !last.at_capacity() {
-                    last.push_into(element);
-                    continue 'element;
-                }
-            }
-            let mut new = O::default();
-            new.ensure_capacity(&mut None);
-            new.push_into(element);
-            self.buffer.push(new);
-        }
+    fn push(&mut self, _chunk: &mut Self::Input) {
+        unimplemented!("ConsolidateBuilder::push")
     }
 
     fn done(self, _: Description<Self::Time>) -> Self::Output {
-        self.buffer
+        unimplemented!("ConsolidateBuilder::done")
     }
 
-    fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {
-        let mut builder = Self::new();
-        for mut input in chain.drain(..) {
-            builder.push(&mut input);
-        }
-        builder.done(description)
+    fn seal(chain: &mut Vec<Self::Input>, _description: Description<Self::Time>) -> Self::Output {
+        std::mem::take(chain)
     }
 }