Switch some dataflow fragments to columnar

Change the dataflow fragment for `specialized_arrange` and in the linear
join preparation phase to use columnar data on edges.

Depends on:
* TimelyDataflow/differential-dataflow#562
* TimelyDataflow/timely-dataflow#630

Signed-off-by: Moritz Hoffmann <mh@materialize.com>

Author: antiguru

Cargo.lock

@@ -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]

src/adapter-types/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"
@@ -78,7 +78,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"

src/adapter/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"

src/catalog/Cargo.toml

@@ -13,15 +13,15 @@ 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_"] }
 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 }

src/cluster/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"

src/compute-client/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 }
 

src/compute-types/Cargo.toml

@@ -50,7 +50,7 @@ pub const LINEAR_JOIN_YIELDING: Config<&str> = Config::new(
 /// Enable lgalloc for columnation.
 pub const ENABLE_COLUMNATION_LGALLOC: Config<bool> = Config::new(
     "enable_columnation_lgalloc",
-    false,
+    true,
     "Enable allocating regions from lgalloc.",
 );
 

src/compute-types/src/dyncfgs.rs

@@ -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"] }

src/compute/Cargo.toml

@@ -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 {