Update columnar, timely, differential

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.219"
-timely = "0.24.0"
+timely = "0.25.1"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
 tracing = "0.1.37"
 

src/adapter/Cargo.toml

@@ -20,7 +20,7 @@ bytesize = "2.1.0"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 dec = "0.4.8"
 derivative = "2.2.0"
-differential-dataflow = "0.17.0"
+differential-dataflow = "0.18.0"
 enum-kinds = "0.5.1"
 fail = { version = "0.5.1", features = ["failpoints"] }
 futures = "0.3.31"
@@ -86,7 +86,7 @@ serde_plain = "1.0.2"
 sha2 = "0.10.9"
 smallvec = { version = "1.15.1", features = ["union"] }
 static_assertions = "1.1"
-timely = "0.24.0"
+timely = "0.25.1"
 tokio = { version = "1.44.1", features = ["rt", "time"] }
 tokio-postgres = { version = "0.7.8" }
 tokio-stream = "0.1.17"

src/catalog/Cargo.toml

@@ -19,7 +19,7 @@ bytesize = "2.1.0"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 clap = { version = "4.5.23", features = ["derive"] }
 derivative = "2.2.0"
-differential-dataflow = "0.17.0"
+differential-dataflow = "0.18.0"
 futures = "0.3.31"
 ipnet = "2.11.0"
 itertools = "0.14.0"
@@ -64,7 +64,7 @@ serde_plain = "1.0.2"
 static_assertions = "1.1"
 sha2 = "0.10.9"
 thiserror = "2.0.17"
-timely = "0.24.0"
+timely = "0.25.1"
 tokio = { version = "1.44.1" }
 tracing = "0.1.37"
 uuid = "1.18.1"

src/cluster/Cargo.toml

@@ -13,15 +13,15 @@ workspace = true
 anyhow = "1.0.100"
 async-trait = "0.1.89"
 crossbeam-channel = "0.5.15"
-differential-dataflow = "0.17.0"
+differential-dataflow = "0.18.0"
 futures = "0.3.31"
 lgalloc = "0.6.0"
 mz-cluster-client = { path = "../cluster-client" }
 mz-ore = { path = "../ore", features = ["async", "process", "tracing"] }
 mz-service = { path = "../service" }
 rand = "0.8.5"
 regex = "1.11.3"
-timely = "0.24.0"
+timely = "0.25.1"
 tokio = { version = "1.44.1", features = ["fs", "rt", "sync", "net"] }
 tracing = "0.1.37"
 uuid = "1.18.1"

src/compute-client/Cargo.toml

@@ -15,7 +15,7 @@ async-trait = "0.1.89"
 bytesize = "2.1.0"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 derivative = "2.2.0"
-differential-dataflow = "0.17.0"
+differential-dataflow = "0.18.0"
 futures = "0.3.31"
 mz-build-info = { path = "../build-info" }
 mz-cluster-client = { path = "../cluster-client" }
@@ -36,7 +36,7 @@ prometheus = { version = "0.14.0", default-features = false }
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.145"
 thiserror = "2.0.17"
-timely = "0.24.0"
+timely = "0.25.1"
 tokio = "1.44.1"
 tracing = "0.1.37"
 uuid = { version = "1.18.1", features = ["serde", "v4"] }

src/compute-types/Cargo.toml

@@ -10,17 +10,17 @@ publish = false
 workspace = true
 
 [dependencies]
-columnar = "0.10.2"
+columnar = "0.11.0"
 columnation = "0.1.0"
-differential-dataflow = "0.17.0"
+differential-dataflow = "0.18.0"
 itertools = "0.14.0"
 mz-dyncfg = { path = "../dyncfg" }
 mz-expr = { path = "../expr" }
 mz-ore = { path = "../ore", features = ["tracing", "metrics"] }
 mz-repr = { path = "../repr", features = ["tracing"] }
 mz-storage-types = { path = "../storage-types" }
 serde = { version = "1.0.219", features = ["derive"] }
-timely = "0.24.0"
+timely = "0.25.1"
 tracing = "0.1.37"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
 

src/compute/Cargo.toml

@@ -13,11 +13,11 @@ workspace = true
 anyhow = "1.0.100"
 async-stream = "0.3.6"
 bytesize = "2.1.0"
-columnar = "0.10.2"
+columnar = "0.11.0"
 crossbeam-channel = "0.5.15"
 dec = { version = "0.4.8", features = ["serde"] }
-differential-dataflow = "0.17.0"
-differential-dogs3 = "0.17.0"
+differential-dataflow = "0.18.0"
+differential-dogs3 = "0.18.0"
 futures = "0.3.31"
 itertools = "0.14.0"
 lgalloc = "0.6"
@@ -41,7 +41,7 @@ prometheus = { version = "0.14.0", default-features = false }
 scopeguard = "1.2.0"
 serde = { version = "1.0.219", features = ["derive"] }
 smallvec = { version = "1.15.1", features = ["serde", "union"] }
-timely = "0.24.0"
+timely = "0.25.1"
 tokio = { version = "1.44.1", features = ["fs", "rt", "sync", "net"] }
 tracing = "0.1.37"
 uuid = { version = "1.18.1", features = ["serde", "v4"] }

src/compute/src/command_channel.rs

@@ -128,12 +128,12 @@ pub fn render<A: Allocate>(timely_worker: &mut TimelyWorker<A>) -> (Sender, Rece
             .sink(
                 Exchange::new(|(idx, _, _)| u64::cast_from(*idx)),
                 "command_channel::sink",
-                move |input| {
-                    while let Some((_cap, data)) = input.next() {
+                move |(input, _)| {
+                    input.for_each(|_time, data| {
                         for (_idx, cmd, nonce) in data.drain(..) {
                             let _ = output_tx.send((cmd, nonce));
                         }
-                    }
+                    });
                 },
             );
         }

src/compute/src/extensions/arrange.rs

@@ -16,7 +16,7 @@ use differential_dataflow::operators::arrange::arrangement::arrange_core;
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
 use differential_dataflow::trace::implementations::spine_fueled::Spine;
 use differential_dataflow::trace::{Batch, Batcher, Builder, Trace, TraceReader};
-use differential_dataflow::{Collection, Data, ExchangeData, Hashable};
+use differential_dataflow::{Collection, Data, ExchangeData, Hashable, VecCollection};
 use timely::Container;
 use timely::dataflow::channels::pact::{Exchange, ParallelizationContract, Pipeline};
 use timely::dataflow::operators::Operator;
@@ -112,7 +112,7 @@ where
     }
 }
 
-impl<G, K, V, R> MzArrange for Collection<G, (K, V), R>
+impl<G, K, V, R> MzArrange for VecCollection<G, (K, V), R>
 where
     G: Scope,
     G::Timestamp: Lattice,
@@ -138,7 +138,7 @@ where
     }
 }
 
-impl<G, K, V, R, C> MzArrangeCore for Collection<G, (K, V), R, C>
+impl<G, C> MzArrangeCore for Collection<G, C>
 where
     G: Scope,
     G::Timestamp: Lattice,
@@ -167,10 +167,10 @@ where
 /// A specialized collection where data only has a key, but no associated value.
 ///
 /// Created by calling `collection.into()`.
-pub struct KeyCollection<G: Scope, K, R = usize>(Collection<G, K, R>);
+pub struct KeyCollection<G: Scope, K, R = usize>(VecCollection<G, K, R>);
 
-impl<G: Scope, K, R: Semigroup> From<Collection<G, K, R>> for KeyCollection<G, K, R> {
-    fn from(value: Collection<G, K, R>) -> Self {
+impl<G: Scope, K, R: Semigroup> From<VecCollection<G, K, R>> for KeyCollection<G, K, R> {
+    fn from(value: VecCollection<G, K, R>) -> Self {
         KeyCollection(value)
     }
 }
@@ -271,13 +271,13 @@ where
             let mut batches = BTreeMap::new();
 
             move |input, output| {
-                while let Some((time, data)) = input.next() {
+                input.for_each(|time, data| {
                     batches.extend(
                         data.iter()
                             .map(|batch| (Rc::as_ptr(batch), Rc::downgrade(batch))),
                     );
                     output.session(&time).give_container(data);
-                }
+                });
                 let Some(trace) = trace.upgrade() else {
                     return;
                 };