From 0ca907e97da7c325ac5ae9362406dcefca266164 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 19 Jul 2026 16:33:07 +0000 Subject: [PATCH 1/4] row-spine: assert batch types are Send + Sync Batches shared across threads, for example behind an Arc to serve reads from outside the worker that maintains the trace, require their contents to be Send + Sync. This already holds for all row-spine batch types because the backing containers bottom out in Vecs, lgalloc regions, and CompactBytes. Add a compile-time assertion so it stays that way. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- src/row-spine/Cargo.toml | 3 +++ src/row-spine/src/lib.rs | 20 +++++++++++++++++++- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/src/row-spine/Cargo.toml b/src/row-spine/Cargo.toml index 998f8dbc92f61..9ce7b79fab11c 100644 --- a/src/row-spine/Cargo.toml +++ b/src/row-spine/Cargo.toml @@ -19,5 +19,8 @@ mz-repr = { path = "../repr" } mz-timely-util = { path = "../timely-util", default-features = false } timely.workspace = true +[dev-dependencies] +mz-ore = { path = "../ore", default-features = false, features = ["columnation"] } + [features] default = ["mz-ore/default", "mz-timely-util/default"] diff --git a/src/row-spine/src/lib.rs b/src/row-spine/src/lib.rs index 2eb38d7b149ed..1da0ce3529bfc 100644 --- a/src/row-spine/src/lib.rs +++ b/src/row-spine/src/lib.rs @@ -155,10 +155,28 @@ mod spines { #[cfg(test)] mod tests { use crate::DatumContainer; + use crate::spines::{RowLayout, RowRowLayout, RowValLayout}; use differential_dataflow::trace::implementations::BatchContainer; + use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatch, OrdValBatch}; use mz_repr::adt::date::Date; use mz_repr::adt::interval::Interval; - use mz_repr::{Datum, Row, SqlScalarType}; + use mz_repr::{Datum, Diff, Row, SqlScalarType, Timestamp}; + use mz_timely_util::columnation::ColumnationStack; + + fn assert_send_sync() {} + + /// The batch types backing our spines must stay `Send + Sync`, so that batches + /// can be shared across threads (for example behind an `Arc`) to serve reads + /// from outside the worker that maintains the trace. This holds because the + /// backing containers bottom out in `Vec`s, lgalloc regions, and `CompactBytes`, + /// all of which are thread-safe. + #[mz_ore::test] + fn batches_are_send_sync() { + assert_send_sync::>>(); + assert_send_sync::>>(); + assert_send_sync::>>(); + assert_send_sync::>(); + } #[mz_ore::test] #[cfg_attr(miri, ignore)] // unsupported operation: integer-to-pointer casts and `ptr::with_exposed_provenance` are not supported From 3c148d8a401dd4910ad6e174d23ca7c7883aa667 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 03:55:37 +0000 Subject: [PATCH 2/4] spines: migrate production arrangements from Rc to Arc batches Switch Materialize's spine typedefs to Arc'd batches so arrangements can be read across thread boundaries, the prerequisite for sharing a trace between adjacent timely runtimes. - mz-row-spine: RowRowSpine, RowValSpine, RowSpine, ValRowSpine and their builders move from Rc/RcBuilder to Arc/ArcBuilder. ErrSpine and the generic spines are aliases of these and follow. - mz-compute typedefs: ColValSpine, ColKeySpine and builders likewise. - The arrangement-size logger held batches by Rc weak reference; it now uses Arc. The storage sink's stock OrdValSpine builder alias is renamed. The batch contents (lgalloc regions, CompactBytes, columnation stacks) were already Send + Sync, asserted in mz-row-spine, so only the wrapper changes. differential-dataflow is patched to the fork carrying arc_blanket_impls and ArcBuilder via a sibling-checkout path patch. This is prototype plumbing until an upstream release carries those, at which point the patch repoints at a git rev or drops. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- Cargo.lock | 12 ++++-------- Cargo.toml | 6 ++++++ src/compute/src/extensions/arrange.rs | 15 ++++++++------- src/compute/src/typedefs.rs | 12 ++++++------ src/row-spine/src/lib.rs | 24 ++++++++++++------------ src/storage/src/render/sinks.rs | 4 ++-- 6 files changed, 38 insertions(+), 35 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b98141d47d7a2..069318ef1051c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3093,9 +3093,7 @@ checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" [[package]] name = "differential-dataflow" -version = "0.25.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7140f9ee3bac19b6db1c4f589ff241148a72bdbf9eb10b51dddb0977de527813" +version = "0.25.1" dependencies = [ "columnar", "columnation", @@ -3108,9 +3106,7 @@ dependencies = [ [[package]] name = "differential-dogs3" -version = "0.25.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33b117006f84a58286d95be98c96d3973d016a399241c6e07004a2a8365c545a" +version = "0.25.1" dependencies = [ "differential-dataflow", "serde", @@ -10732,7 +10728,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.14.0", + "itertools 0.10.5", "log", "multimap", "petgraph", @@ -10753,7 +10749,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.117", diff --git a/Cargo.toml b/Cargo.toml index feb27cb95bf1c..1da4cc9a5815f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -644,6 +644,12 @@ debug-assertions = true # merged), after which point it becomes impossible to build that historical # version of Materialize. [patch.crates-io] +# Arc'd batches for cross-thread arrangement sharing, from a sibling checkout of +# the fork. Prototype plumbing: repoint at a git rev (or drop the patch) once an +# upstream differential-dataflow release carries the `arc_blanket_impls` module. +differential-dataflow = { path = "../differential-dataflow/differential-dataflow" } +differential-dogs3 = { path = "../differential-dataflow/dogsdogsdogs" } + # Waiting on https://github.com/sfackler/rust-postgres/pull/752. postgres = { git = "https://github.com/MaterializeInc/rust-postgres" } tokio-postgres = { git = "https://github.com/MaterializeInc/rust-postgres" } diff --git a/src/compute/src/extensions/arrange.rs b/src/compute/src/extensions/arrange.rs index c9616c83a924c..fcacb4fb24032 100644 --- a/src/compute/src/extensions/arrange.rs +++ b/src/compute/src/extensions/arrange.rs @@ -8,7 +8,8 @@ // by the Apache License, Version 2.0. use std::collections::BTreeMap; -use std::rc::{Rc, Weak}; +use std::rc::Rc; +use std::sync::{Arc, Weak}; use differential_dataflow::difference::Semigroup; use differential_dataflow::lattice::Lattice; @@ -241,9 +242,9 @@ pub trait ArrangementSize { /// * `logic`: Closure that calculates the heap size/capacity/allocations for a batch. The return /// value are size and capacity in bytes, and number of allocations, all in absolute values. fn log_arrangement_size_inner<'scope, B, L>( - arranged: Arranged<'scope, TraceAgent>>>, + arranged: Arranged<'scope, TraceAgent>>>, mut logic: L, -) -> Arranged<'scope, TraceAgent>>> +) -> Arranged<'scope, TraceAgent>>> where B: Batch + 'static, L: FnMut(&B) -> (usize, usize, usize) + 'static, @@ -282,8 +283,8 @@ where input.for_each(|time, data| { for batch in data.iter() { batches - .entry(Rc::as_ptr(batch)) - .or_insert_with(|| (Rc::downgrade(batch), logic(batch))); + .entry(Arc::as_ptr(batch)) + .or_insert_with(|| (Arc::downgrade(batch), logic(batch))); } output.session(&time).give_container(data); }); @@ -293,8 +294,8 @@ where trace.borrow().trace().map_batches(|batch| { batches - .entry(Rc::as_ptr(batch)) - .or_insert_with(|| (Rc::downgrade(batch), logic(batch))); + .entry(Arc::as_ptr(batch)) + .or_insert_with(|| (Arc::downgrade(batch), logic(batch))); }); let (mut size, mut capacity, mut allocations) = (0, 0, 0); diff --git a/src/compute/src/typedefs.rs b/src/compute/src/typedefs.rs index 3b7f33af5ea68..af9c867ec0586 100644 --- a/src/compute/src/typedefs.rs +++ b/src/compute/src/typedefs.rs @@ -29,7 +29,7 @@ pub use crate::typedefs::spines::{ColKeySpine, ColValSpine}; pub use mz_row_spine::{RowRowSpine, RowSpine, RowValBatcher, RowValSpine}; pub(crate) mod spines { - use std::rc::Rc; + use std::sync::Arc; use columnation::Columnation; use differential_dataflow::trace::implementations::ord_neu::{ @@ -37,7 +37,7 @@ pub(crate) mod spines { }; use differential_dataflow::trace::implementations::spine_fueled::Spine; use differential_dataflow::trace::implementations::{Layout, Update}; - use differential_dataflow::trace::rc_blanket_impls::RcBuilder; + use differential_dataflow::trace::arc_blanket_impls::ArcBuilder; use mz_timely_util::columnation::ColumnationStack; use mz_row_spine::OffsetOptimized; @@ -45,16 +45,16 @@ pub(crate) mod spines { use crate::typedefs::{KeyBatcher, KeyValBatcher}; /// A spine for generic keys and values. - pub type ColValSpine = Spine>>>; + pub type ColValSpine = Spine>>>; pub type ColValBatcher = KeyValBatcher; pub type ColValBuilder = - RcBuilder, ColumnationStack<((K, V), T, R)>>>; + ArcBuilder, ColumnationStack<((K, V), T, R)>>>; /// A spine for generic keys - pub type ColKeySpine = Spine>>>; + pub type ColKeySpine = Spine>>>; pub type ColKeyBatcher = KeyBatcher; pub type ColKeyBuilder = - RcBuilder, ColumnationStack<((K, ()), T, R)>>>; + ArcBuilder, ColumnationStack<((K, ()), T, R)>>>; /// A layout based on chunked timely stacks pub struct MzStack { diff --git a/src/row-spine/src/lib.rs b/src/row-spine/src/lib.rs index 1da0ce3529bfc..8b6227c283182 100644 --- a/src/row-spine/src/lib.rs +++ b/src/row-spine/src/lib.rs @@ -29,7 +29,7 @@ pub static DICTIONARY_COMPRESSION: std::sync::atomic::AtomicBool = /// Spines specialized to contain `Row` types in keys and values. mod spines { - use std::rc::Rc; + use std::sync::Arc; use columnation::Columnation; use differential_dataflow::trace::implementations::Layout; @@ -37,7 +37,7 @@ mod spines { use differential_dataflow::trace::implementations::merge_batcher::MergeBatcher; use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatch, OrdValBatch}; use differential_dataflow::trace::implementations::spine_fueled::Spine; - use differential_dataflow::trace::rc_blanket_impls::RcBuilder; + use differential_dataflow::trace::arc_blanket_impls::ArcBuilder; use mz_repr::Row; use mz_timely_util::columnation::{ColInternalMerger, ColumnationStack}; @@ -48,9 +48,9 @@ mod spines { type KeyValBatcher = MergeBatcher>; type KeyBatcher = KeyValBatcher; - pub type RowRowSpine = Spine>>>; + pub type RowRowSpine = Spine>>>; pub type RowRowBatcher = KeyValBatcher; - pub type RowRowBuilder = RcBuilder>; + pub type RowRowBuilder = ArcBuilder>; /// `RowRowBuilder` variant that consumes [`Column`] chunks. Pairs with /// [`Col2ValPagedBatcher`] for the spillable arrange path. Installs a @@ -61,21 +61,21 @@ mod spines { /// [`Col2ValPagedBatcher`]: mz_timely_util::columnar::Col2ValPagedBatcher /// [`Column`]: mz_timely_util::columnar::Column pub type RowRowColPagedBuilder = - RcBuilder>; + ArcBuilder>; - pub type RowValSpine = Spine>>>; + pub type RowValSpine = Spine>>>; pub type RowValBatcher = KeyValBatcher; pub type RowValBuilder = - RcBuilder>; + ArcBuilder>; - pub type RowSpine = Spine>>>; + pub type RowSpine = Spine>>>; pub type RowBatcher = KeyBatcher; - pub type RowBuilder = RcBuilder>; + pub type RowBuilder = ArcBuilder>; - pub type ValRowSpine = Spine>>>; + pub type ValRowSpine = Spine>>>; pub type ValRowBatcher = KeyValBatcher; pub type ValRowBuilder = - RcBuilder>; + ArcBuilder>; /// `ValRowBuilder` variant that consumes [`Column`] chunks. Pairs with /// `Col2ValPagedBatcher` for the spillable arrange path where @@ -86,7 +86,7 @@ mod spines { /// /// [`Column`]: mz_timely_util::columnar::Column pub type ValRowColPagedBuilder = - RcBuilder>; + ArcBuilder>; /// A layout based on timely stacks pub struct RowRowLayout> { diff --git a/src/storage/src/render/sinks.rs b/src/storage/src/render/sinks.rs index e27212700e331..086da824ac659 100644 --- a/src/storage/src/render/sinks.rs +++ b/src/storage/src/render/sinks.rs @@ -15,7 +15,7 @@ use std::time::{Duration, Instant}; use differential_dataflow::operators::arrange::{Arrange, Arranged, TraceAgent}; use differential_dataflow::trace::TraceReader; use differential_dataflow::trace::implementations::ord_neu::{ - OrdValBatcher, OrdValSpine, RcOrdValBuilder, + OrdValBatcher, OrdValSpine, ArcOrdValBuilder, }; use differential_dataflow::{AsCollection, Hashable, VecCollection}; use mz_persist_client::operators::shard_source::SnapshotMode; @@ -148,7 +148,7 @@ fn arrange_sink_input<'scope>( // Allow access to `arrange_named` because we cannot access Mz's wrapper // from here. TODO(database-issues#5046): Revisit with cluster unification. #[allow(clippy::disallowed_methods)] - let Arranged {stream, trace: _} = keyed.arrange_named::, RcOrdValBuilder<_, _, _, _>, OrdValSpine<_, _, _, _>>("Arrange Sink"); + let Arranged {stream, trace: _} = keyed.arrange_named::, ArcOrdValBuilder<_, _, _, _>, OrdValSpine<_, _, _, _>>("Arrange Sink"); stream } From 5265025fa41a3dd6c8c12ae51192fb8f8ad5b7c6 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 08:15:23 +0000 Subject: [PATCH 3/4] storage: use ArcOrdValSpine for the sink arrangement The stock differential OrdValSpine is Rc-backed. The sink trace uses the Arc-backed ArcOrdValSpine variant so its batches match the Arc migration of the other production spines. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- src/storage/src/render/sinks.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/storage/src/render/sinks.rs b/src/storage/src/render/sinks.rs index 086da824ac659..4fb0054081115 100644 --- a/src/storage/src/render/sinks.rs +++ b/src/storage/src/render/sinks.rs @@ -15,7 +15,7 @@ use std::time::{Duration, Instant}; use differential_dataflow::operators::arrange::{Arrange, Arranged, TraceAgent}; use differential_dataflow::trace::TraceReader; use differential_dataflow::trace::implementations::ord_neu::{ - OrdValBatcher, OrdValSpine, ArcOrdValBuilder, + ArcOrdValBuilder, ArcOrdValSpine, OrdValBatcher, }; use differential_dataflow::{AsCollection, Hashable, VecCollection}; use mz_persist_client::operators::shard_source::SnapshotMode; @@ -35,7 +35,7 @@ use crate::storage_state::StorageState; /// The concrete trace type produced internally when arranging a sink's input. /// The sink never sees this directly — only the batches flowing through it — /// but it's the anchor for the batch type in [`SinkBatchStream`]. -pub(crate) type SinkTrace = TraceAgent, Row, Timestamp, Diff>>; +pub(crate) type SinkTrace = TraceAgent, Row, Timestamp, Diff>>; /// Stream of arrangement batches handed to [`SinkRender::render_sink`]. /// @@ -148,7 +148,7 @@ fn arrange_sink_input<'scope>( // Allow access to `arrange_named` because we cannot access Mz's wrapper // from here. TODO(database-issues#5046): Revisit with cluster unification. #[allow(clippy::disallowed_methods)] - let Arranged {stream, trace: _} = keyed.arrange_named::, ArcOrdValBuilder<_, _, _, _>, OrdValSpine<_, _, _, _>>("Arrange Sink"); + let Arranged {stream, trace: _} = keyed.arrange_named::, ArcOrdValBuilder<_, _, _, _>, ArcOrdValSpine<_, _, _, _>>("Arrange Sink"); stream } From d97c0a19c8e44cd9879be5e63f69b7ec56c3f9c5 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 08:20:50 +0000 Subject: [PATCH 4/4] build: point differential-dataflow patch at the fork PR branch Repoint the [patch.crates-io] entry for differential-dataflow and differential-dogs3 from the local sibling checkout to the fork branch (antiguru/differential-dataflow, the Arc-batches PR). Prototype plumbing until that lands upstream and a released version carries arc_blanket_impls. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- Cargo.lock | 6 ++++-- Cargo.toml | 10 +++++----- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 069318ef1051c..48f37d004ce8b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3094,6 +3094,7 @@ checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" [[package]] name = "differential-dataflow" version = "0.25.1" +source = "git+https://github.com/antiguru/differential-dataflow?branch=claude%2Fspines-differential-arc-j93mho#5b06554f09cd8b32a62b8ea3f490e0ce9b449e3b" dependencies = [ "columnar", "columnation", @@ -3107,6 +3108,7 @@ dependencies = [ [[package]] name = "differential-dogs3" version = "0.25.1" +source = "git+https://github.com/antiguru/differential-dataflow?branch=claude%2Fspines-differential-arc-j93mho#5b06554f09cd8b32a62b8ea3f490e0ce9b449e3b" dependencies = [ "differential-dataflow", "serde", @@ -10728,7 +10730,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.10.5", + "itertools 0.14.0", "log", "multimap", "petgraph", @@ -10749,7 +10751,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.117", diff --git a/Cargo.toml b/Cargo.toml index 1da4cc9a5815f..08ae4258215b0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -644,11 +644,11 @@ debug-assertions = true # merged), after which point it becomes impossible to build that historical # version of Materialize. [patch.crates-io] -# Arc'd batches for cross-thread arrangement sharing, from a sibling checkout of -# the fork. Prototype plumbing: repoint at a git rev (or drop the patch) once an -# upstream differential-dataflow release carries the `arc_blanket_impls` module. -differential-dataflow = { path = "../differential-dataflow/differential-dataflow" } -differential-dogs3 = { path = "../differential-dataflow/dogsdogsdogs" } +# Arc'd batches for cross-thread arrangement sharing, from the branch behind +# TimelyDataflow/differential-dataflow#807. Prototype plumbing: repoint at a +# released version (and drop the patch) once that PR lands upstream. +differential-dataflow = { git = "https://github.com/antiguru/differential-dataflow", branch = "claude/spines-differential-arc-j93mho" } +differential-dogs3 = { git = "https://github.com/antiguru/differential-dataflow", branch = "claude/spines-differential-arc-j93mho" } # Waiting on https://github.com/sfackler/rust-postgres/pull/752. postgres = { git = "https://github.com/MaterializeInc/rust-postgres" }