diff --git a/Cargo.lock b/Cargo.lock
index 2b6f8db62e51d..2693ff55694f1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2381,9 +2381,9 @@ dependencies = [
 
 [[package]]
 name = "differential-dataflow"
-version = "0.13.4"
+version = "0.13.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75b93af605b7f82fbf6e671a5bb0f940b385e57254a0af59ce6dfb98b8c4b302"
+checksum = "62e90a3a356ef9af92925918d36f5206cb74ef834a1ec39c1ed9631d00736686"
 dependencies = [
  "columnar",
  "fnv",
@@ -2393,9 +2393,9 @@ dependencies = [
 
 [[package]]
 name = "differential-dogs3"
-version = "0.1.4"
+version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "17ba29145a1df1bdc3da1142eeb991f0866620c79ce9d85e83a3837f29112ba0"
+checksum = "6ecafb7f2d23b07b36a442274be75c4090cfd225b49918aceeba577b119780ea"
 dependencies = [
  "differential-dataflow",
  "serde",
@@ -10434,9 +10434,9 @@ dependencies = [
 
 [[package]]
 name = "timely"
-version = "0.17.0"
+version = "0.17.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0506b6506ef04c371ea6151942df60e309e7f5e710458a6b533e364ee0b3cab3"
+checksum = "1502aad02b725c0efcd64bac7f2b1b05bf402c2ada6d74cd95b6b9ac252d62ed"
 dependencies = [
  "bincode",
  "byteorder",
diff --git a/src/adapter-types/Cargo.toml b/src/adapter-types/Cargo.toml
index d1cdc0ae56507..4c66606fcb08e 100644
--- a/src/adapter-types/Cargo.toml
+++ b/src/adapter-types/Cargo.toml
@@ -15,7 +15,7 @@ mz-ore = { path = "../ore" }
 mz-repr = { path = "../repr" }
 mz-storage-types = { path = "../storage-types" }
 serde = "1.0.152"
-timely = "0.17.0"
+timely = "0.17.1"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
 
 [package.metadata.cargo-udeps.ignore]
diff --git a/src/adapter/Cargo.toml b/src/adapter/Cargo.toml
index 273be043fa7a3..c58b17196ca29 100644
--- a/src/adapter/Cargo.toml
+++ b/src/adapter/Cargo.toml
@@ -18,7 +18,7 @@ bytesize = "1.1.0"
 chrono = { version = "0.4.35", default-features = false, features = ["std"] }
 dec = "0.4.8"
 derivative = "2.2.0"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 enum-kinds = "0.5.1"
 fail = { version = "0.5.1", features = ["failpoints"] }
 futures = "0.3.25"
@@ -81,7 +81,7 @@ serde_plain = "1.0.1"
 sha2 = "0.10.6"
 smallvec = { version = "1.10.0", features = ["union"] }
 static_assertions = "1.1"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = ["rt", "time"] }
 tokio-postgres = { version = "0.7.8" }
 tracing = "0.1.37"
diff --git a/src/catalog/Cargo.toml b/src/catalog/Cargo.toml
index 6d1e030684df6..61be0b675f8e1 100644
--- a/src/catalog/Cargo.toml
+++ b/src/catalog/Cargo.toml
@@ -18,7 +18,7 @@ bytesize = "1.1.0"
 chrono = { version = "0.4.35", default-features = false, features = ["std"] }
 clap = { version = "4.5.23", features = ["derive"] }
 derivative = "2.2.0"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 ipnet = "2.5.0"
 itertools = "0.12.1"
@@ -60,7 +60,7 @@ serde_plain = "1.0.1"
 static_assertions = "1.1"
 sha2 = "0.10.6"
 thiserror = "1.0.37"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0" }
 tracing = "0.1.37"
 uuid = "1.2.2"
diff --git a/src/cluster/Cargo.toml b/src/cluster/Cargo.toml
index 43ca3a040c8c2..11832fc8b08f9 100644
--- a/src/cluster/Cargo.toml
+++ b/src/cluster/Cargo.toml
@@ -13,7 +13,7 @@ workspace = true
 anyhow = "1.0.66"
 async-trait = "0.1.83"
 crossbeam-channel = "0.5.8"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 mz-cluster-client = { path = "../cluster-client" }
 mz-ore = { path = "../ore", features = ["async", "process", "tracing_"] }
@@ -21,7 +21,7 @@ mz-persist-client = { path = "../persist-client" }
 mz-service = { path = "../service" }
 mz-txn-wal = { path = "../txn-wal" }
 regex = "1.7.0"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = ["fs", "rt", "sync", "net"] }
 tracing = "0.1.37"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
diff --git a/src/compute-client/Cargo.toml b/src/compute-client/Cargo.toml
index c7ed7aa902e90..1b783735aeb51 100644
--- a/src/compute-client/Cargo.toml
+++ b/src/compute-client/Cargo.toml
@@ -15,7 +15,7 @@ async-trait = "0.1.83"
 bytesize = "1.1.0"
 crossbeam-channel = "0.5.8"
 derivative = "2.2.0"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 http = "1.1.0"
 mz-build-info = { path = "../build-info" }
@@ -43,7 +43,7 @@ prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = "1.0.125"
 thiserror = "1.0.37"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = "1.38.0"
 tokio-stream = "0.1.11"
 tonic = "0.12.1"
diff --git a/src/compute-types/Cargo.toml b/src/compute-types/Cargo.toml
index 4e368867b8edd..78a7e2bf78f2e 100644
--- a/src/compute-types/Cargo.toml
+++ b/src/compute-types/Cargo.toml
@@ -12,7 +12,7 @@ workspace = true
 [dependencies]
 columnar = "0.2.2"
 columnation = "0.1.0"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 itertools = "0.12.1"
 mz-dyncfg = { path = "../dyncfg" }
 mz-expr = { path = "../expr" }
@@ -24,7 +24,7 @@ proptest = { version = "1.6.0", default-features = false, features = ["std"] }
 proptest-derive = { version = "0.5.1", features = ["boxed_union"] }
 prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 serde = { version = "1.0.152", features = ["derive"] }
-timely = "0.17.0"
+timely = "0.17.1"
 tracing = "0.1.37"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
 
diff --git a/src/compute/Cargo.toml b/src/compute/Cargo.toml
index e972dbb2f1782..c00ea84eedcc6 100644
--- a/src/compute/Cargo.toml
+++ b/src/compute/Cargo.toml
@@ -16,8 +16,8 @@ bytesize = "1.1.0"
 columnar = "0.2.2"
 crossbeam-channel = "0.5.8"
 dec = { version = "0.4.8", features = ["serde"] }
-differential-dataflow = "0.13.4"
-differential-dogs3 = "0.1.4"
+differential-dataflow = "0.13.5"
+differential-dogs3 = "0.1.5"
 futures = "0.3.25"
 itertools = "0.12.1"
 lgalloc = "0.4"
@@ -39,7 +39,7 @@ prometheus = { version = "0.13.3", default-features = false }
 scopeguard = "1.1.0"
 serde = { version = "1.0.152", features = ["derive"] }
 smallvec = { version = "1.10.0", features = ["serde", "union"] }
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = ["fs", "rt", "sync", "net"] }
 tracing = "0.1.37"
 uuid = { version = "1.7.0", features = ["serde", "v4"] }
diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs
index 83d2cd12d1f9f..72c7953bed533 100644
--- a/src/compute/src/render.rs
+++ b/src/compute/src/render.rs
@@ -110,6 +110,7 @@ use std::rc::{Rc, Weak};
 use std::sync::Arc;
 use std::task::Poll;
 
+use columnar::Columnar;
 use differential_dataflow::dynamic::pointstamp::PointStamp;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
@@ -498,6 +499,7 @@ impl<'g, G, T> Context<Child<'g, G, T>>
 where
     G: Scope<Timestamp = mz_repr::Timestamp>,
     T: Refines<G::Timestamp> + RenderTimestamp,
+    <T as Columnar>::Container: Clone + Send,
 {
     pub(crate) fn import_index(
         &mut self,
@@ -647,6 +649,7 @@ impl<'g, G, T> Context<Child<'g, G, T>>
 where
     G: Scope<Timestamp = mz_repr::Timestamp>,
     T: RenderTimestamp,
+    <T as Columnar>::Container: Clone + Send,
 {
     pub(crate) fn export_index_iterative(
         &self,
@@ -889,6 +892,8 @@ impl<G> Context<G>
 where
     G: Scope,
     G::Timestamp: RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
 {
     /// Renders a non-recursive plan to a differential dataflow, producing the collection of
     /// results.
@@ -1293,7 +1298,11 @@ where
 
 #[allow(dead_code)] // Some of the methods on this trait are unused, but useful to have.
 /// A timestamp type that can be used for operations within MZ's dataflow layer.
-pub trait RenderTimestamp: Timestamp + Lattice + Refines<mz_repr::Timestamp> + Columnation {
+pub trait RenderTimestamp:
+    Timestamp + Lattice + Refines<mz_repr::Timestamp> + Columnation + Columnar
+where
+    <Self as Columnar>::Container: Clone + Send,
+{
     /// The system timestamp component of the timestamp.
     ///
     /// This is useful for manipulating the system time, as when delaying
@@ -1429,6 +1438,7 @@ impl<S> WithStartSignal for MzArrangementImport<S>
 where
     S: Scope,
     S::Timestamp: RenderTimestamp,
+    <S::Timestamp as Columnar>::Container: Clone + Send,
 {
     fn with_start_signal(self, signal: StartSignal) -> Self {
         match self {
@@ -1443,6 +1453,7 @@ impl<S, Tr> WithStartSignal for Arranged<S, Tr>
 where
     S: Scope,
     S::Timestamp: RenderTimestamp,
+    <S::Timestamp as Columnar>::Container: Clone + Send,
     Tr: TraceReader + Clone,
 {
     fn with_start_signal(self, signal: StartSignal) -> Self {
diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs
index cae8662236652..577afa3861439 100644
--- a/src/compute/src/render/context.rs
+++ b/src/compute/src/render/context.rs
@@ -14,37 +14,40 @@ use std::collections::BTreeMap;
 use std::rc::Weak;
 use std::sync::mpsc;
 
+use columnar::Columnar;
 use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::trace::cursor::IntoOwned;
 use differential_dataflow::trace::{BatchReader, Cursor, TraceReader};
-use differential_dataflow::{Collection, Data};
+use differential_dataflow::{AsCollection, Collection, Data};
 use mz_compute_types::dataflows::DataflowDescription;
 use mz_compute_types::plan::{AvailableCollections, LirId};
 use mz_expr::{Id, MapFilterProject, MirScalarExpr};
-use mz_repr::fixed_length::{FromDatumIter, ToDatumIter};
+use mz_repr::fixed_length::ToDatumIter;
 use mz_repr::{DatumVec, DatumVecBorrow, Diff, GlobalId, Row, RowArena, SharedRow};
 use mz_storage_types::controller::CollectionMetadata;
 use mz_storage_types::errors::DataflowError;
+use mz_timely_util::containers::{columnar_exchange, Col2ValBatcher, ColumnBuilder};
 use mz_timely_util::operator::{CollectionExt, StreamExt};
 use timely::container::columnation::Columnation;
 use timely::container::CapacityContainerBuilder;
-use timely::dataflow::channels::pact::Pipeline;
+use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
 use timely::dataflow::operators::generic::OutputHandleCore;
 use timely::dataflow::operators::Capability;
 use timely::dataflow::scopes::Child;
 use timely::dataflow::{Scope, ScopeParent};
 use timely::progress::timestamp::Refines;
 use timely::progress::{Antichain, Timestamp};
+use timely::Container;
 use tracing::error;
 
 use crate::arrangement::manager::SpecializedTraceHandle;
 use crate::compute_state::{ComputeState, HydrationEvent};
-use crate::extensions::arrange::{KeyCollection, MzArrange};
+use crate::extensions::arrange::{KeyCollection, MzArrange, MzArrangeCore};
 use crate::render::errors::ErrorLogger;
 use crate::render::{LinearJoinSpec, RenderTimestamp};
-use crate::row_spine::{RowRowBatcher, RowRowBuilder};
+use crate::row_spine::RowRowBuilder;
 use crate::typedefs::{
     ErrAgent, ErrBatcher, ErrBuilder, ErrEnter, ErrSpine, RowRowAgent, RowRowEnter, RowRowSpine,
 };
@@ -919,10 +922,11 @@ where
 
 impl<S, T> CollectionBundle<S, T>
 where
-    T: timely::progress::Timestamp + Lattice + Columnation,
+    T: Timestamp + Lattice + Columnation,
     S: Scope,
-    S::Timestamp:
-        Refines<T> + Lattice + timely::progress::Timestamp + crate::render::RenderTimestamp,
+    S::Timestamp: Refines<T> + RenderTimestamp,
+    <S::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <S::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
 {
     /// Presents `self` as a stream of updates, having been subjected to `mfp`.
     ///
@@ -1046,7 +1050,8 @@ where
                     .collection
                     .clone()
                     .expect("Collection constructed above");
-                let (oks, errs_keyed) = Self::specialized_arrange(&name, oks, &key, &thinning);
+                let (oks, errs_keyed) =
+                    Self::arrange_collection(&name, oks, key.clone(), thinning.clone());
                 let errs: KeyCollection<_, _, _> = errs.concat(&errs_keyed).into();
                 let errs = errs.mz_arrange::<ErrBatcher<_, _>, ErrBuilder<_, _>, ErrSpine<_, _>>(
                     &format!("{}-errors", name),
@@ -1058,50 +1063,60 @@ where
         self
     }
 
-    /// Builds a specialized arrangement to provided types. The specialization for key and
-    /// value types of the arrangement is based on the bit length derived from the corresponding
-    /// type descriptions.
-    fn specialized_arrange(
+    /// Builds an arrangement from a collection, using the specified key and value thinning.
+    ///
+    /// The arrangement's key is based on the `key` expressions, and the value the input with
+    /// the `thinning` applied to it. It selects which of the input columns are included in the
+    /// value of the arrangement. The thinning is in support of permuting arrangements such that
+    /// columns in the key are not included in the value.
+    fn arrange_collection(
         name: &String,
         oks: Collection<S, Row, i64>,
-        key: &Vec<MirScalarExpr>,
-        thinning: &Vec<usize>,
+        key: Vec<MirScalarExpr>,
+        thinning: Vec<usize>,
     ) -> (MzArrangement<S>, Collection<S, DataflowError, i64>) {
-        // Catch-all: Just use RowRow.
+        // The following `unary_fallible` implements a `map_fallible`, but produces columnar updates
+        // for the ok stream. The `map_fallible` cannot be used here because the closure cannot
+        // return references, which is what we need to push into columnar streams. Instead, we use
+        // a bespoke operator that also optimizes reuse of allocations across individual updates.
         let (oks, errs) = oks
-            .map_fallible::<CapacityContainerBuilder<_>, CapacityContainerBuilder<_>, _, _, _>(
+            .inner
+            .unary_fallible::<ColumnBuilder<((Row, Row), S::Timestamp, i64)>, _, _, _>(
+                Pipeline,
                 "FormArrangementKey",
-                specialized_arrangement_key(key.clone(), thinning.clone()),
+                move |_, _| {
+                    Box::new(move |input, ok, err| {
+                        let mut key_buf = Row::default();
+                        let mut val_buf = Row::default();
+                        let mut datums = DatumVec::new();
+                        let temp_storage = RowArena::new();
+                        while let Some((time, data)) = input.next() {
+                            let mut ok_session = ok.session_with_builder(&time);
+                            let mut err_session = err.session(&time);
+                            for (row, time, diff) in data.iter() {
+                                temp_storage.clear();
+                                let datums = datums.borrow_with(row);
+                                let key_iter = key.iter().map(|k| k.eval(&datums, &temp_storage));
+                                match key_buf.packer().try_extend(key_iter) {
+                                    Ok(()) => {
+                                        let val_datum_iter = thinning.iter().map(|c| datums[*c]);
+                                        val_buf.packer().extend(val_datum_iter);
+                                        ok_session.give(((&*key_buf, &*val_buf), time, diff));
+                                    }
+                                    Err(e) => {
+                                        err_session.give((e.into(), time.clone(), *diff));
+                                    }
+                                }
+                            }
+                        }
+                    })
+                },
             );
-        let oks =
-            oks.mz_arrange::<RowRowBatcher<_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(name);
-        (MzArrangement::RowRow(oks), errs)
-    }
-}
-
-/// Obtains a function that maps input rows to (key, value) pairs according to
-/// the given key and thinning expressions. This function allows for specialization
-/// of key and value types and is intended to use to form arrangement keys.
-fn specialized_arrangement_key<K, V>(
-    key: Vec<MirScalarExpr>,
-    thinning: Vec<usize>,
-) -> impl FnMut(Row) -> Result<(K, V), DataflowError>
-where
-    K: Columnation + Data + FromDatumIter,
-    V: Columnation + Data + FromDatumIter,
-{
-    let mut key_buf = K::default();
-    let mut val_buf = V::default();
-    let mut datums = DatumVec::new();
-    move |row| {
-        // TODO: Consider reusing the `row` allocation; probably in *next* invocation.
-        let datums = datums.borrow_with(&row);
-        let temp_storage = RowArena::new();
-        let val_datum_iter = thinning.iter().map(|c| datums[*c]);
-        Ok::<(K, V), DataflowError>((
-            key_buf.try_from_datum_iter(key.iter().map(|k| k.eval(&datums, &temp_storage)))?,
-            val_buf.from_datum_iter(val_datum_iter),
-        ))
+        let oks = oks
+            .mz_arrange_core::<_, Col2ValBatcher<_, _,_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<Row, Row, S::Timestamp, Diff>),name
+            );
+        (MzArrangement::RowRow(oks), errs.as_collection())
     }
 }
 
diff --git a/src/compute/src/render/flat_map.rs b/src/compute/src/render/flat_map.rs
index 566395c64ffcb..cacbd65d7d042 100644
--- a/src/compute/src/render/flat_map.rs
+++ b/src/compute/src/render/flat_map.rs
@@ -7,6 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use columnar::Columnar;
 use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use mz_expr::MfpPlan;
 use mz_expr::{MapFilterProject, MirScalarExpr, TableFunc};
@@ -26,6 +27,7 @@ impl<G> Context<G>
 where
     G: Scope,
     G::Timestamp: crate::render::RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
 {
     /// Applies a `TableFunc` to every row, followed by an `mfp`.
     pub fn render_flat_map(
@@ -130,6 +132,7 @@ fn drain_through_mfp<T>(
     >,
 ) where
     T: crate::render::RenderTimestamp,
+    <T as Columnar>::Container: Clone + Send,
 {
     let temp_storage = RowArena::new();
     let binding = SharedRow::get();
diff --git a/src/compute/src/render/join/delta_join.rs b/src/compute/src/render/join/delta_join.rs
index 7501e9f124e02..b68115fd772d0 100644
--- a/src/compute/src/render/join/delta_join.rs
+++ b/src/compute/src/render/join/delta_join.rs
@@ -15,6 +15,7 @@
 
 use std::collections::{BTreeMap, BTreeSet};
 
+use columnar::Columnar;
 use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
@@ -45,7 +46,9 @@ use crate::typedefs::{RowRowAgent, RowRowEnter};
 impl<G> Context<G>
 where
     G: Scope,
-    G::Timestamp: crate::render::RenderTimestamp,
+    G::Timestamp: RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
 {
     /// Renders `MirRelationExpr:Join` using dogs^3 delta query dataflows.
     ///
@@ -326,7 +329,9 @@ fn dispatch_build_halfjoin_local<G, CF>(
 )
 where
     G: Scope,
-    G::Timestamp: crate::render::RenderTimestamp,
+    G::Timestamp: RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
     CF: Fn(&G::Timestamp, &G::Timestamp) -> bool + 'static,
 {
     match trace {
@@ -358,7 +363,9 @@ fn dispatch_build_halfjoin_trace<G, T, CF>(
 where
     G: Scope,
     T: Timestamp + Lattice + Columnation,
-    G::Timestamp: Lattice + crate::render::RenderTimestamp + Refines<T> + Columnation,
+    G::Timestamp: RenderTimestamp + Refines<T>,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
     CF: Fn(&G::Timestamp, &G::Timestamp) -> bool + 'static,
 {
     match trace {
@@ -398,7 +405,9 @@ fn build_halfjoin<G, Tr, K, CF>(
 )
 where
     G: Scope,
-    G::Timestamp: crate::render::RenderTimestamp,
+    G::Timestamp: RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
     Tr: TraceReader<Time = G::Timestamp, Diff = Diff> + Clone + 'static,
     K: ExchangeData + Hashable + Default + FromDatumIter + ToDatumIter,
     for<'a> Tr::Key<'a>: IntoOwned<'a, Owned = K>,
@@ -531,7 +540,9 @@ fn dispatch_build_update_stream_local<G>(
 ) -> (Collection<G, Row, Diff>, Collection<G, DataflowError, Diff>)
 where
     G: Scope,
-    G::Timestamp: crate::render::RenderTimestamp,
+    G::Timestamp: RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
 {
     match trace {
         MzArrangement::RowRow(inner) => build_update_stream::<_, RowRowAgent<_, _>>(
@@ -553,7 +564,9 @@ fn dispatch_build_update_stream_trace<G, T>(
 where
     G: Scope,
     T: Timestamp + Lattice + Columnation,
-    G::Timestamp: Lattice + crate::render::RenderTimestamp + Refines<T> + Columnation,
+    G::Timestamp: Lattice + RenderTimestamp + Refines<T>,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
 {
     match trace {
         MzArrangementImport::RowRow(inner) => build_update_stream::<_, RowRowEnter<_, _, _>>(
@@ -578,7 +591,9 @@ fn build_update_stream<G, Tr>(
 ) -> (Collection<G, Row, Diff>, Collection<G, DataflowError, Diff>)
 where
     G: Scope,
-    G::Timestamp: crate::render::RenderTimestamp,
+    G::Timestamp: RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
     for<'a, 'b> &'a G::Timestamp: PartialEq<Tr::TimeGat<'b>>,
     Tr: for<'a> TraceReader<Time = G::Timestamp, Diff = Diff> + Clone + 'static,
     for<'a> Tr::Key<'a>: ToDatumIter,
diff --git a/src/compute/src/render/join/linear_join.rs b/src/compute/src/render/join/linear_join.rs
index 5907adacb0b03..7d719a989854a 100644
--- a/src/compute/src/render/join/linear_join.rs
+++ b/src/compute/src/render/join/linear_join.rs
@@ -13,6 +13,7 @@
 
 use std::time::{Duration, Instant};
 
+use columnar::Columnar;
 use differential_dataflow::consolidation::ConsolidatingContainerBuilder;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::arrangement::Arranged;
@@ -25,20 +26,22 @@ use mz_dyncfg::ConfigSet;
 use mz_repr::fixed_length::ToDatumIter;
 use mz_repr::{DatumVec, Diff, Row, RowArena, SharedRow};
 use mz_storage_types::errors::DataflowError;
-use mz_timely_util::operator::CollectionExt;
+use mz_timely_util::containers::{columnar_exchange, Col2ValBatcher, ColumnBuilder};
+use mz_timely_util::operator::{CollectionExt, StreamExt};
 use timely::container::columnation::Columnation;
-use timely::container::CapacityContainerBuilder;
+use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
 use timely::dataflow::operators::OkErr;
 use timely::dataflow::scopes::Child;
 use timely::dataflow::{Scope, ScopeParent};
 use timely::progress::timestamp::{Refines, Timestamp};
 
-use crate::extensions::arrange::MzArrange;
+use crate::extensions::arrange::MzArrangeCore;
 use crate::render::context::{
     ArrangementFlavor, CollectionBundle, Context, MzArrangement, MzArrangementImport, ShutdownToken,
 };
 use crate::render::join::mz_join_core::mz_join_core;
-use crate::row_spine::{RowRowBatcher, RowRowBuilder, RowRowSpine};
+use crate::render::RenderTimestamp;
+use crate::row_spine::{RowRowBuilder, RowRowSpine};
 use crate::typedefs::{RowRowAgent, RowRowEnter};
 
 /// Available linear join implementations.
@@ -204,7 +207,9 @@ where
 impl<G, T> Context<G, T>
 where
     G: Scope,
-    G::Timestamp: Lattice + Refines<T> + Columnation,
+    G::Timestamp: Lattice + Refines<T> + RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
+    for<'a> <G::Timestamp as Columnar>::Ref<'a>: Ord + Copy,
     T: Timestamp + Lattice + Columnation,
 {
     pub(crate) fn render_join(
@@ -356,37 +361,52 @@ where
         // If we have only a streamed collection, we must first form an arrangement.
         if let JoinedFlavor::Collection(stream) = joined {
             let name = "LinearJoinKeyPreparation";
-            type CB<C> = CapacityContainerBuilder<C>;
-            let (keyed, errs) = stream.map_fallible::<CB<_>, CB<_>, _, _, _>(name, {
-                // Reuseable allocation for unpacking.
-                let mut datums = DatumVec::new();
-                move |row| {
-                    let binding = SharedRow::get();
-                    let mut row_builder = binding.borrow_mut();
-                    let temp_storage = RowArena::new();
-                    let datums_local = datums.borrow_with(&row);
-                    row_builder.packer().try_extend(
-                        stream_key
-                            .iter()
-                            .map(|e| e.eval(&datums_local, &temp_storage)),
-                    )?;
-                    let key = row_builder.clone();
-                    row_builder
-                        .packer()
-                        .extend(stream_thinning.iter().map(|e| datums_local[*e]));
-                    let value = row_builder.clone();
-                    Ok((key, value))
-                }
-            });
+            let (keyed, errs) = stream
+                .inner
+                .unary_fallible::<ColumnBuilder<((Row, Row), S::Timestamp, Diff)>, _, _, _>(
+                    Pipeline,
+                    name,
+                    |_, _| {
+                        Box::new(move |input, ok, errs| {
+                            let temp_storage = RowArena::new();
+                            let mut key_buf = Row::default();
+                            let mut val_buf = Row::default();
+                            let mut datums = DatumVec::new();
+                            while let Some((time, data)) = input.next() {
+                                let mut ok_session = ok.session_with_builder(&time);
+                                let mut err_session = errs.session(&time);
+                                for (row, time, diff) in data.iter() {
+                                    temp_storage.clear();
+                                    let datums_local = datums.borrow_with(row);
+                                    let datums = stream_key
+                                        .iter()
+                                        .map(|e| e.eval(&datums_local, &temp_storage));
+                                    let result = key_buf.packer().try_extend(datums);
+                                    match result {
+                                        Ok(()) => {
+                                            val_buf.packer().extend(
+                                                stream_thinning.iter().map(|e| datums_local[*e]),
+                                            );
+                                            ok_session.give(((&key_buf, &val_buf), time, diff));
+                                        }
+                                        Err(e) => {
+                                            err_session.give((e.into(), time.clone(), *diff));
+                                        }
+                                    }
+                                }
+                            }
+                        })
+                    },
+                );
 
-            errors.push(errs);
+            errors.push(errs.as_collection());
 
             // TODO(vmarcos): We should implement further arrangement specialization here (database-issues#6659).
             // By knowing how types propagate through joins we could specialize intermediate
             // arrangements as well, either in values or eventually in keys.
             let arranged = keyed
-                .mz_arrange::<RowRowBatcher<_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
-                    "JoinStage",
+                .mz_arrange_core::<_, Col2ValBatcher<_, _,_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                    ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<Row, Row, S::Timestamp, Diff>),"JoinStage"
                 );
             joined = JoinedFlavor::Local(MzArrangement::RowRow(arranged));
         }
diff --git a/src/compute/src/render/sinks.rs b/src/compute/src/render/sinks.rs
index 8ddd5eb2b40b0..fd7f835c93787 100644
--- a/src/compute/src/render/sinks.rs
+++ b/src/compute/src/render/sinks.rs
@@ -9,10 +9,7 @@
 
 //! Logic related to the creation of dataflow sinks.
 
-use std::any::Any;
-use std::collections::{BTreeMap, BTreeSet};
-use std::rc::{Rc, Weak};
-
+use columnar::Columnar;
 use differential_dataflow::Collection;
 use mz_compute_types::sinks::{ComputeSinkConnection, ComputeSinkDesc};
 use mz_expr::{permutation_for_arrangement, EvalError, MapFilterProject};
@@ -23,6 +20,9 @@ use mz_repr::{Diff, GlobalId, Row};
 use mz_storage_types::controller::CollectionMetadata;
 use mz_storage_types::errors::DataflowError;
 use mz_timely_util::operator::CollectionExt;
+use std::any::Any;
+use std::collections::{BTreeMap, BTreeSet};
+use std::rc::{Rc, Weak};
 use timely::container::CapacityContainerBuilder;
 use timely::dataflow::scopes::Child;
 use timely::dataflow::Scope;
@@ -37,6 +37,8 @@ impl<'g, G, T> Context<Child<'g, G, T>>
 where
     G: Scope<Timestamp = mz_repr::Timestamp>,
     T: RenderTimestamp,
+    <T as Columnar>::Container: Clone + Send,
+    for<'a> <T as Columnar>::Ref<'a>: Ord + Copy,
 {
     /// Export the sink described by `sink` from the rendering context.
     pub(crate) fn export_sink(
diff --git a/src/compute/src/render/top_k.rs b/src/compute/src/render/top_k.rs
index c26e12113dfbb..1ef56ff27699e 100644
--- a/src/compute/src/render/top_k.rs
+++ b/src/compute/src/render/top_k.rs
@@ -11,10 +11,7 @@
 //!
 //! Consult [TopKPlan] documentation for details.
 
-use std::cell::RefCell;
-use std::collections::BTreeMap;
-use std::rc::Rc;
-
+use columnar::Columnar;
 use differential_dataflow::hashable::Hashable;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
@@ -31,6 +28,9 @@ use mz_ore::soft_assert_or_log;
 use mz_repr::{Datum, DatumVec, Diff, Row, ScalarType, SharedRow};
 use mz_storage_types::errors::DataflowError;
 use mz_timely_util::operator::CollectionExt;
+use std::cell::RefCell;
+use std::collections::BTreeMap;
+use std::rc::Rc;
 use timely::container::columnation::Columnation;
 use timely::container::{CapacityContainerBuilder, PushInto};
 use timely::dataflow::channels::pact::Pipeline;
@@ -53,6 +53,7 @@ impl<G> Context<G>
 where
     G: Scope,
     G::Timestamp: crate::render::RenderTimestamp,
+    <G::Timestamp as Columnar>::Container: Clone + Send,
 {
     pub(crate) fn render_topk(
         &self,
diff --git a/src/controller/Cargo.toml b/src/controller/Cargo.toml
index d73475b4d1d47..94573db9cdf6e 100644
--- a/src/controller/Cargo.toml
+++ b/src/controller/Cargo.toml
@@ -32,7 +32,7 @@ mz-txn-wal = { path = "../txn-wal" }
 regex = "1.7.0"
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = "1.0.125"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = "1.38.0"
 tracing = "0.1.37"
 uuid = { version = "1.7.0" }
diff --git a/src/durable-cache/Cargo.toml b/src/durable-cache/Cargo.toml
index 336e15e10234c..8cf12fd518a1f 100644
--- a/src/durable-cache/Cargo.toml
+++ b/src/durable-cache/Cargo.toml
@@ -12,7 +12,7 @@ workspace = true
 [dependencies]
 async-trait = "0.1.83"
 bytes = { version = "1.3.0" }
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 itertools = { version = "0.12.1" }
 mz-ore = { path = "../ore", features = ["process"] }
@@ -23,7 +23,7 @@ mz-timely-util = { path = "../timely-util" }
 prometheus = { version = "0.13.3", default-features = false }
 prost = { version = "0.13.1", features = ["no-recursion-limit"] }
 serde = { version = "1.0.152", features = ["derive", "rc"] }
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", default-features = false, features = ["rt", "rt-multi-thread"] }
 tracing = "0.1.37"
 uuid = { version = "1.7.0", features = ["v4"] }
diff --git a/src/environmentd/Cargo.toml b/src/environmentd/Cargo.toml
index fd2ae958e2d8a..aca06758e552d 100644
--- a/src/environmentd/Cargo.toml
+++ b/src/environmentd/Cargo.toml
@@ -145,7 +145,7 @@ reqwest = { version = "0.11.13", features = ["blocking"] }
 serde_json = "1.0.125"
 serde_urlencoded = "0.7.1"
 similar-asserts = "1.4"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio-postgres = { version = "0.7.8", features = ["with-chrono-0_4"] }
 
 [build-dependencies]
diff --git a/src/expr/Cargo.toml b/src/expr/Cargo.toml
index 48de6ee02ab19..53f5338b1abcd 100644
--- a/src/expr/Cargo.toml
+++ b/src/expr/Cargo.toml
@@ -59,7 +59,7 @@ serde_json = "1.0.125"
 sha1 = "0.10.5"
 sha2 = "0.10.6"
 subtle = "2.4.1"
-timely = "0.17.0"
+timely = "0.17.1"
 tracing = "0.1.37"
 uncased = "0.9.7"
 uuid = { version = "1.7.0", features = ["v5"] }
diff --git a/src/interchange/Cargo.toml b/src/interchange/Cargo.toml
index f9b9580f10e55..253e48da4ecaa 100644
--- a/src/interchange/Cargo.toml
+++ b/src/interchange/Cargo.toml
@@ -20,7 +20,7 @@ byteorder = "1.4.3"
 bytes = "1.3.0"
 chrono = { version = "0.4.35", default-features = false, features = ["std"] }
 clap = { version = "4.5.23", features = ["derive"] }
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 itertools = "0.12.1"
 maplit = "1.0.2"
 mz-avro = { path = "../avro", features = ["snappy"] }
@@ -33,7 +33,7 @@ prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 prost-reflect = "0.14.5"
 seahash = "4"
 serde_json = "1.0.125"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = ["macros", "net", "rt", "rt-multi-thread", "time"] }
 tracing = "0.1.37"
 uuid = { version = "1.7.0", features = ["serde"] }
diff --git a/src/persist-cli/Cargo.toml b/src/persist-cli/Cargo.toml
index 6e73b01894abd..8a50819d9c8f4 100644
--- a/src/persist-cli/Cargo.toml
+++ b/src/persist-cli/Cargo.toml
@@ -23,7 +23,7 @@ async-trait = "0.1.83"
 axum = "0.7.5"
 bytes = { version = "1.3.0", features = ["serde"] }
 clap = { version = "4.5.23", features = ["derive", "env"] }
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 humantime = "2.1.0"
 mz-http-util = { path = "../http-util" }
@@ -40,7 +40,7 @@ num_enum = "0.5.7"
 prometheus = { version = "0.13.3", default-features = false }
 serde = { version = "1.0.152", features = ["derive", "rc"] }
 serde_json = "1.0.125"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", default-features = false, features = ["macros", "sync", "rt", "rt-multi-thread", "time"] }
 tracing = "0.1.37"
 url = "2.3.1"
diff --git a/src/persist-client/Cargo.toml b/src/persist-client/Cargo.toml
index f705fb120e49d..a32f7c13e2e58 100644
--- a/src/persist-client/Cargo.toml
+++ b/src/persist-client/Cargo.toml
@@ -35,7 +35,7 @@ async-stream = "0.3.3"
 async-trait = "0.1.83"
 bytes = { version = "1.3.0", features = ["serde"] }
 clap = { version = "4.5.23", features = ["derive"] }
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 futures-util = "0.3"
 h2 = "0.3.13"
@@ -59,7 +59,7 @@ sentry-tracing = "0.29.1"
 semver = { version = "1.0.16", features = ["serde"] }
 serde = { version = "1.0.152", features = ["derive", "rc"] }
 serde_json = "1.0.125"
-timely = "0.17.0"
+timely = "0.17.1"
 thiserror = "1.0.37"
 tokio = { version = "1.38.0", default-features = false, features = ["macros", "sync", "rt", "rt-multi-thread", "time"] }
 tokio-metrics = "0.3.0"
diff --git a/src/persist-types/Cargo.toml b/src/persist-types/Cargo.toml
index fb268f82042b8..c2ababe117482 100644
--- a/src/persist-types/Cargo.toml
+++ b/src/persist-types/Cargo.toml
@@ -26,7 +26,7 @@ proptest-derive = { version = "0.5.1", features = ["boxed_union"] }
 prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = { version = "1.0.125" }
-timely = "0.17.0"
+timely = "0.17.1"
 tracing = "0.1.37"
 uuid = { version = "1.7.0", features = ["v4"] }
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
diff --git a/src/persist/Cargo.toml b/src/persist/Cargo.toml
index 39e8d8d4e20d3..f96142142a7d9 100644
--- a/src/persist/Cargo.toml
+++ b/src/persist/Cargo.toml
@@ -37,7 +37,7 @@ azure_core = "0.21.0"
 base64 = "0.13.1"
 bytes = "1.3.0"
 deadpool-postgres = "0.10.3"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 fail = { version = "0.5.1", features = ["failpoints"] }
 futures-util = "0.3.25"
 md-5 = "0.10.5"
@@ -58,7 +58,7 @@ proptest-derive = { version = "0.5.1", features = ["boxed_union"] }
 prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 rand = { version = "0.8.5", features = ["small_rng"] }
 serde = { version = "1.0.152", features = ["derive"] }
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", default-features = false, features = ["fs", "macros", "sync", "rt", "rt-multi-thread"] }
 tokio-postgres = { version = "0.7.8" }
 tracing = "0.1.37"
diff --git a/src/repr/Cargo.toml b/src/repr/Cargo.toml
index 440808c72e692..3cfb5d22c5577 100644
--- a/src/repr/Cargo.toml
+++ b/src/repr/Cargo.toml
@@ -36,7 +36,7 @@ columnation = "0.1.0"
 chrono = { version = "0.4.35", default-features = false, features = ["serde", "std"] }
 compact_bytes = "0.1.2"
 dec = "0.4.8"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 enum-kinds = "0.5.1"
 flatcontainer = "0.5.0"
 hex = "0.4.3"
@@ -68,7 +68,7 @@ serde_json = { version = "1.0.125", features = ["arbitrary_precision", "preserve
 smallvec = { version = "1.10.0", features = ["serde", "union"] }
 static_assertions = "1.1"
 strsim = "0.11.1"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio-postgres = { version = "0.7.8" }
 tracing-core = "0.1.30"
 url = { version = "2.3.1", features = ["serde"] }
diff --git a/src/repr/src/row.rs b/src/repr/src/row.rs
index cfef8f425f9e3..3218da60ee95c 100644
--- a/src/repr/src/row.rs
+++ b/src/repr/src/row.rs
@@ -2824,6 +2824,7 @@ impl RowArena {
         Ok(self.push_unary_row(row))
     }
 
+    /// Clear the contents of the arena.
     pub fn clear(&self) {
         self.inner.borrow_mut().clear();
     }
diff --git a/src/service/Cargo.toml b/src/service/Cargo.toml
index 095502c93f3d8..ea9158512b42a 100644
--- a/src/service/Cargo.toml
+++ b/src/service/Cargo.toml
@@ -35,7 +35,7 @@ prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 semver = "1.0.16"
 serde = { version = "1.0.152", features = ["derive"] }
 sysinfo = "0.27.2"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = "1.38.0"
 tokio-stream = "0.1.11"
 tonic = "0.12.1"
diff --git a/src/storage-client/Cargo.toml b/src/storage-client/Cargo.toml
index 25d7a30777e3a..bb41eaa58df71 100644
--- a/src/storage-client/Cargo.toml
+++ b/src/storage-client/Cargo.toml
@@ -13,7 +13,7 @@ workspace = true
 anyhow = "1.0.66"
 async-trait = "0.1.83"
 chrono = { version = "0.4.35", default-features = false, features = ["std"] }
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 http = "1.1.0"
 itertools = { version = "0.12.1" }
@@ -46,7 +46,7 @@ serde = { version = "1.0.152", features = ["derive"] }
 serde_json = { version = "1.0.125" }
 smallvec = { version = "1.10.0", features = ["serde", "union"] }
 static_assertions = "1.1"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = [
     "fs",
     "rt",
diff --git a/src/storage-controller/Cargo.toml b/src/storage-controller/Cargo.toml
index 73a5e83ae0e2a..0c7c2864c339b 100644
--- a/src/storage-controller/Cargo.toml
+++ b/src/storage-controller/Cargo.toml
@@ -15,7 +15,7 @@ async-trait = "0.1.83"
 bytes = "1.3.0"
 chrono = { version = "0.4.35", default-features = false, features = ["std"] }
 derivative = "2.2.0"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 itertools = { version = "0.12.1" }
 mz-build-info = { path = "../build-info" }
@@ -38,7 +38,7 @@ proptest = { version = "1.6.0", default-features = false, features = ["std"] }
 prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = { version = "1.0.125" }
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = ["fs", "rt", "sync", "test-util", "time"] }
 tokio-postgres = { version = "0.7.8", features = ["serde"] }
 tokio-stream = "0.1.11"
diff --git a/src/storage-operators/Cargo.toml b/src/storage-operators/Cargo.toml
index 2b89e41024f20..f26651a0be045 100644
--- a/src/storage-operators/Cargo.toml
+++ b/src/storage-operators/Cargo.toml
@@ -47,7 +47,7 @@ reqwest = { version = "0.11.13", features = ["stream"] }
 sentry = { version = "0.29.1" }
 serde = { version = "1.0.152", features = ["derive"] }
 smallvec = { version = "1.10.0", features = ["union"] }
-timely = "0.17.0"
+timely = "0.17.1"
 thiserror = "1.0.37"
 tokio = { version = "1.38.0", features = ["fs", "rt", "sync", "test-util", "time"] }
 tokio-stream = "0.1.11"
diff --git a/src/storage-types/Cargo.toml b/src/storage-types/Cargo.toml
index ab09f4d1be935..2db6559e5ae88 100644
--- a/src/storage-types/Cargo.toml
+++ b/src/storage-types/Cargo.toml
@@ -25,7 +25,7 @@ bytes = "1.3.0"
 columnation = "0.1.0"
 dec = "0.4.8"
 derivative = "2.2.0"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 hex = "0.4.3"
 http = "1.1.0"
 itertools = { version = "0.12.1" }
@@ -62,7 +62,7 @@ regex = "1.7.0"
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = { version = "1.0.125", features = ["preserve_order"] }
 thiserror = "1.0.37"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = ["fs", "rt", "sync", "test-util", "time"] }
 tokio-postgres = { version = "0.7.8", features = ["serde"] }
 tracing = "0.1.37"
diff --git a/src/storage/Cargo.toml b/src/storage/Cargo.toml
index 0bd2af8dacca5..428fa6c2da1a8 100644
--- a/src/storage/Cargo.toml
+++ b/src/storage/Cargo.toml
@@ -26,7 +26,7 @@ columnation = "0.1.0"
 crossbeam-channel = "0.5.8"
 csv-core = { version = "0.1.10" }
 dec = "0.4.8"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 fail = { version = "0.5.1", features = ["failpoints"] }
 futures = "0.3.25"
 indexmap = { version = "2.0.0", default-features = false, features = ["std"] }
@@ -77,7 +77,7 @@ serde = { version = "1.0.152", features = ["derive"] }
 serde_json = { version = "1.0.125" }
 serde_bytes = { version = "0.11.14" }
 sha2 = "0.10.6"
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = ["fs", "rt", "sync", "test-util"] }
 tokio-postgres = { version = "0.7.8", features = ["serde"] }
 tokio-stream = "0.1.11"
diff --git a/src/timely-util/Cargo.toml b/src/timely-util/Cargo.toml
index 88901d8931402..65dbc6bb8d307 100644
--- a/src/timely-util/Cargo.toml
+++ b/src/timely-util/Cargo.toml
@@ -15,7 +15,7 @@ bincode = "1.3.3"
 bytemuck = "1.21.0"
 columnar = "0.2.2"
 columnation = "0.1.0"
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 either = "1"
 futures-util = "0.3.25"
 lgalloc = "0.4"
@@ -23,7 +23,7 @@ mz-ore = { path = "../ore", features = ["async", "process", "tracing_", "test"]
 num-traits = "0.2"
 proptest = { version = "1.6.0", default-features = false, features = ["std"] }
 serde = { version = "1.0.152", features = ["derive"] }
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", features = ["macros", "rt-multi-thread", "time"] }
 tracing = "0.1.37"
 uuid = { version = "1.7.0", features = ["serde", "v4"] }
diff --git a/src/transform/Cargo.toml b/src/transform/Cargo.toml
index 2718a2b83b3c8..ab4409769a28c 100644
--- a/src/transform/Cargo.toml
+++ b/src/transform/Cargo.toml
@@ -10,7 +10,7 @@ publish = false
 workspace = true
 
 [dependencies]
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 enum-kinds = "0.5.1"
 itertools = "0.12.1"
 mz-compute-types = { path = "../compute-types" }
diff --git a/src/txn-wal/Cargo.toml b/src/txn-wal/Cargo.toml
index 6e7e2bdff4a10..e06a7e76356b6 100644
--- a/src/txn-wal/Cargo.toml
+++ b/src/txn-wal/Cargo.toml
@@ -12,7 +12,7 @@ workspace = true
 [dependencies]
 async-trait = "0.1.83"
 bytes = { version = "1.3.0" }
-differential-dataflow = "0.13.4"
+differential-dataflow = "0.13.5"
 futures = "0.3.25"
 itertools = { version = "0.12.1" }
 mz-ore = { path = "../ore", features = ["process"] }
@@ -23,7 +23,7 @@ mz-timely-util = { path = "../timely-util" }
 prometheus = { version = "0.13.3", default-features = false }
 prost = { version = "0.13.2", features = ["no-recursion-limit"] }
 serde = { version = "1.0.152", features = ["derive", "rc"] }
-timely = "0.17.0"
+timely = "0.17.1"
 tokio = { version = "1.38.0", default-features = false, features = ["rt", "rt-multi-thread"] }
 tracing = "0.1.37"
 uuid = { version = "1.7.0", features = ["v4"] }