From 84c5b93984b43cbda0f91d96706b5281a89ecdec Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 08:10:18 +0000 Subject: [PATCH 1/7] trace: add Arc-backed spine variants for cross-thread sharing Add `arc_blanket_impls`, a mirror of `rc_blanket_impls` using `Arc`, and `ArcOrdValSpine`/`ArcOrdKeySpine` with their `ArcOrdValBuilder`/`ArcOrdKeyBuilder` alongside the existing `Rc` spines and builders. Also re-export them as `ArcValSpine`/`ArcKeySpine`/`ArcValBuilder`/`ArcKeyBuilder`. The addition is purely additive. The default `OrdValSpine`/`OrdKeySpine` and `Val`/`Key` aliases remain `Rc`-backed, so existing users are unaffected. An `Arc`'d batch whose contents are `Send + Sync` can be handed to and read from a thread other than the one maintaining the trace, which the `Rc` variants cannot do. Atomic reference counting is marginally more expensive, so `Rc` stays the default and callers opt into `Arc` only when they need the sharing. A test reads an `Arc`'d batch from another thread. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../src/trace/implementations/mod.rs | 7 ++ .../src/trace/implementations/ord_neu.rs | 14 +++ differential-dataflow/src/trace/mod.rs | 119 ++++++++++++++++++ differential-dataflow/tests/trace.rs | 31 +++++ 4 files changed, 171 insertions(+) diff --git a/differential-dataflow/src/trace/implementations/mod.rs b/differential-dataflow/src/trace/implementations/mod.rs index 9b69cb12d..70041d3d4 100644 --- a/differential-dataflow/src/trace/implementations/mod.rs +++ b/differential-dataflow/src/trace/implementations/mod.rs @@ -53,6 +53,13 @@ pub use self::ord_neu::OrdKeySpine as KeySpine; pub use self::ord_neu::OrdKeyBatcher as KeyBatcher; pub use self::ord_neu::RcOrdKeyBuilder as KeyBuilder; +// `Arc`-backed variants of the `Val`/`Key` aliases, whose batches can be shared across threads. +// The batcher is shared with the `Rc` variants, since batchers do not reference-count batches. +pub use self::ord_neu::ArcOrdValSpine as ArcValSpine; +pub use self::ord_neu::ArcOrdValBuilder as ArcValBuilder; +pub use self::ord_neu::ArcOrdKeySpine as ArcKeySpine; +pub use self::ord_neu::ArcOrdKeyBuilder as ArcKeyBuilder; + use std::convert::TryInto; use serde::{Deserialize, Serialize}; diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index cf0c22a5f..1c5f58232 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -9,11 +9,13 @@ //! and should consume fewer resources (computation and memory) when it applies. use std::rc::Rc; +use std::sync::Arc; use crate::trace::implementations::spine_fueled::Spine; use crate::trace::implementations::merge_batcher::MergeBatcher; use crate::trace::implementations::merge_batcher::vec::VecMerger; use crate::trace::rc_blanket_impls::RcBuilder; +use crate::trace::arc_blanket_impls::ArcBuilder; use super::{Layout, Vector}; @@ -22,17 +24,29 @@ pub use self::key_batch::{OrdKeyBatch, OrdKeyBuilder}; /// A trace implementation using a spine of ordered lists. pub type OrdValSpine = Spine>>>; +/// An `Arc`-backed variant of [`OrdValSpine`]. +/// +/// Its batches are `Arc`'d rather than `Rc`'d, so a batch whose contents are `Send + Sync` can be +/// read from a thread other than the one maintaining the trace. Prefer [`OrdValSpine`] unless that +/// cross-thread sharing is needed, since atomic reference counting is marginally more expensive. +pub type ArcOrdValSpine = Spine>>>; /// A batcher using ordered lists. pub type OrdValBatcher = MergeBatcher>; /// A builder using ordered lists. pub type RcOrdValBuilder = RcBuilder, Vec<((K,V),T,R)>>>; +/// An `Arc`-backed variant of [`RcOrdValBuilder`], pairing with [`ArcOrdValSpine`]. +pub type ArcOrdValBuilder = ArcBuilder, Vec<((K,V),T,R)>>>; /// A trace implementation using a spine of ordered lists. pub type OrdKeySpine = Spine>>>; +/// An `Arc`-backed variant of [`OrdKeySpine`], readable from other threads. See [`ArcOrdValSpine`]. +pub type ArcOrdKeySpine = Spine>>>; /// A batcher for ordered lists. pub type OrdKeyBatcher = MergeBatcher>; /// A builder for ordered lists. pub type RcOrdKeyBuilder = RcBuilder, Vec<((K,()),T,R)>>>; +/// An `Arc`-backed variant of [`RcOrdKeyBuilder`], pairing with [`ArcOrdKeySpine`]. +pub type ArcOrdKeyBuilder = ArcBuilder, Vec<((K,()),T,R)>>>; pub use layers::{Vals, Upds}; /// Layers are containers of lists of some type. diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index c6f864f53..579b6c377 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -437,3 +437,122 @@ pub mod rc_blanket_impls { fn done(self) -> Rc { Rc::new(self.merger.done()) } } } + +/// Blanket implementations for atomically reference counted batches. +/// +/// These mirror [`rc_blanket_impls`], but use `Arc` so that batches whose contents +/// are `Send + Sync` can be shared across threads. This enables reading a trace's +/// batches from outside the worker that maintains the trace, for example to import +/// a snapshot of an arrangement into a dataflow running elsewhere. +pub mod arc_blanket_impls { + + use std::sync::Arc; + + use timely::progress::{Antichain, frontier::AntichainRef}; + use super::{Batch, BatchReader, Builder, Merger, Navigable, Cursor, Description}; + + impl Navigable for Arc { + /// The type used to enumerate the batch's contents. + type Cursor = ArcBatchCursor; + /// Acquires a cursor to the batch's contents. + fn cursor(&self) -> Self::Cursor { + ArcBatchCursor::new((**self).cursor()) + } + } + + impl BatchReader for Arc { + + type Time = B::Time; + /// The number of updates in the batch. + fn len(&self) -> usize { (**self).len() } + /// Describes the times of the updates in the batch. + fn description(&self) -> &Description { (**self).description() } + } + + /// Wrapper to provide cursor to nested scope. + pub struct ArcBatchCursor { + cursor: C, + } + + impl ArcBatchCursor { + fn new(cursor: C) -> Self { + ArcBatchCursor { + cursor, + } + } + } + + impl Cursor for ArcBatchCursor { + + type Storage = Arc; + + type Key<'a> = C::Key<'a>; + type ValOwn = C::ValOwn; + type Val<'a> = C::Val<'a>; + type Time = C::Time; + type TimeGat<'a> = C::TimeGat<'a>; + type Diff = C::Diff; + type DiffGat<'a> = C::DiffGat<'a>; + type KeyContainer = C::KeyContainer; + type ValContainer = C::ValContainer; + type TimeContainer = C::TimeContainer; + type DiffContainer = C::DiffContainer; + + #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) } + #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) } + + #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage) } + #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage) } + + #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_key(storage) } + #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_val(storage) } + + #[inline] + fn map_times, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L) { + self.cursor.map_times(storage, logic) + } + + #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage) } + #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage, key) } + + #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage) } + #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage, val) } + + #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage) } + #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage) } + } + + /// An immutable collection of updates. + impl Batch for Arc { + type Merger = ArcMerger; + fn empty(lower: Antichain, upper: Antichain) -> Self { + Arc::new(B::empty(lower, upper)) + } + } + + /// Wrapper type for building atomically reference counted batches. + pub struct ArcBuilder { builder: B } + + /// Functionality for building batches from ordered update sequences. + impl Builder for ArcBuilder { + type Input = B::Input; + type Time = B::Time; + type Output = Arc; + fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self { ArcBuilder { builder: B::with_capacity(keys, vals, upds) } } + fn push(&mut self, input: &mut Self::Input) { self.builder.push(input) } + fn done(self, description: Description) -> Arc { Arc::new(self.builder.done(description)) } + fn seal(chain: &mut Vec, description: Description) -> Self::Output { + Arc::new(B::seal(chain, description)) + } + } + + /// Wrapper type for merging atomically reference counted batches. + pub struct ArcMerger { merger: B::Merger } + + /// Represents a merge in progress. + impl Merger> for ArcMerger { + fn new(source1: &Arc, source2: &Arc, compaction_frontier: AntichainRef) -> Self { ArcMerger { merger: B::begin_merge(source1, source2, compaction_frontier) } } + fn work(&mut self, source1: &Arc, source2: &Arc, fuel: &mut isize) { self.merger.work(source1, source2, fuel) } + fn done(self) -> Arc { Arc::new(self.merger.done()) } + } +} diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs index d30e8ef2d..1863e7bb5 100644 --- a/differential-dataflow/tests/trace.rs +++ b/differential-dataflow/tests/trace.rs @@ -57,3 +57,34 @@ fn test_trace() { let vec_4 = cursor4.to_vec(&storage4, |k| k.clone(), |v| v.clone()); assert_eq!(vec_4, vec_3); } + +/// Batches of the `Arc`-backed spines can be handed to another thread, which can hold and read +/// them independently. This is the property that enables sharing a trace's contents outside the +/// worker that maintains it. The default `Rc`-backed spines do not have it, by design. +#[test] +fn test_batches_read_from_other_thread() { + use differential_dataflow::trace::Navigable; + use differential_dataflow::trace::implementations::ord_neu::{ArcOrdValBuilder, OrdValBatcher}; + + fn assert_send_sync(_: &T) {} + + let mut batcher = OrdValBatcher::::new(None, 0); + batcher.push_into(vec![ + ((1, 2), 0, 1), + ((2, 3), 1, 1), + ]); + let (mut chain, description) = batcher.seal(Antichain::from_elem(2)); + let batch = ArcOrdValBuilder::::seal(&mut chain, description); + + assert_send_sync(&batch); + + let read = std::thread::spawn(move || { + let mut cursor = batch.cursor(); + cursor.to_vec(&batch, |k| k.clone(), |v| v.clone()) + }).join().unwrap(); + + assert_eq!(read, vec![ + ((1, 2), vec![(0, 1)]), + ((2, 3), vec![(1, 1)]), + ]); +} From 5b06554f09cd8b32a62b8ea3f490e0ce9b449e3b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 08:10:33 +0000 Subject: [PATCH 2/7] arrange: share arrangements across timely runtimes Add operators::arrange::sharing, a primitive for reading an arrangement from outside the worker that maintains it, built on the Arc-backed spines. Arranged::publish attaches a publisher operator to an arrangement on the owning worker and returns a Published, whose handle() mints Clone + Send SharedTraceHandles. A handle implements TraceReader, so downstream operators drive its compaction and acquire cursors as with any trace handle. SharedTraceHandle::snapshot_at serves a consistent point or full-scan read from any thread, blocking until the publication frontier passes the query time. SharedTraceHandle::import replays the shared arrangement into another scope, so a second timely runtime can consume it as an ordinary Arranged. The publisher forwards the meet of reader holds to its own TraceAgent, the sole writer of the trace's compaction, never the empty meet of zero holds (which would irreversibly release the trace). Handle clones carry independent holds, so two consumers of one import cannot release each other's holds. Compaction is applied to the agent outside the shared lock, since physical compaction can run an unbounded merge. The published since is the frontier the trace will actually hold after forwarding, so a handle registering concurrently cannot latch a floor below the trace's real compaction. A late importer registering after the publisher closed seeds its own terminal frontier, so it drains and releases rather than hanging. Tests run a cross-thread snapshot and a cross-dataflow import through a shared handle. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../src/operators/arrange/mod.rs | 1 + .../src/operators/arrange/sharing.rs | 613 ++++++++++++++++++ differential-dataflow/tests/sharing.rs | 159 +++++ 3 files changed, 773 insertions(+) create mode 100644 differential-dataflow/src/operators/arrange/sharing.rs create mode 100644 differential-dataflow/tests/sharing.rs diff --git a/differential-dataflow/src/operators/arrange/mod.rs b/differential-dataflow/src/operators/arrange/mod.rs index fedbfdc23..2dbda1960 100644 --- a/differential-dataflow/src/operators/arrange/mod.rs +++ b/differential-dataflow/src/operators/arrange/mod.rs @@ -63,6 +63,7 @@ type TraceAgentQueueWriter = Weak<(Activator, RefCell>)>; pub mod writer; pub mod agent; pub mod arrangement; +pub mod sharing; pub mod upsert; diff --git a/differential-dataflow/src/operators/arrange/sharing.rs b/differential-dataflow/src/operators/arrange/sharing.rs new file mode 100644 index 000000000..2b6dffa0c --- /dev/null +++ b/differential-dataflow/src/operators/arrange/sharing.rs @@ -0,0 +1,613 @@ +//! Sharing arrangements across timely runtimes. +//! +//! An arrangement is normally readable only from the worker that maintains it: its batches are +//! reference counted with `Rc` and its trace handle is `Rc>`, both pinned to one +//! thread. This module lets a worker publish an arrangement whose batches are `Arc`'d (and whose +//! contents are `Send + Sync`) through a *publication point*, from which readers on other threads +//! take consistent snapshots or import the arrangement into a second timely runtime. +//! +//! The unit that crosses the thread boundary is not the [`Spine`](super::super::super::trace::implementations::spine_fueled::Spine), +//! which holds thread-local state and has a single writer, but the spine's *contents*: a chain of +//! immutable `Arc`'d batches together with the trace's `since` and `upper` frontiers. Because +//! batches are immutable, a chain plus frontiers is a self-describing, consistent view. When the +//! publishing worker later merges batches, a reader holding an older chain is unaffected: its `Arc`s +//! keep the pre-merge batches alive until it drops them. +//! +//! ## Pieces +//! +//! * [`Arranged::publish`] attaches a publisher to an arrangement on the owning worker and returns a +//! [`Published`] whose [`Published::handle`] hands out `Clone + Send` [`SharedTraceHandle`]s. +//! * [`SharedTraceHandle`] implements [`TraceReader`], so it drives compaction and cursors like any +//! trace handle. [`SharedTraceHandle::snapshot_at`] serves a point or full-scan read from any +//! thread. [`SharedTraceHandle::import`] replays the shared arrangement into another scope. +//! +//! ## Compaction +//! +//! Every reader registers logical and physical holds. The publisher forwards their *meet* (the +//! greatest lower bound, so the trace never compacts past the least reader's hold) to its own +//! `TraceAgent`, which is the sole writer of the trace's compaction frontiers. With no readers the +//! publisher holds compaction at the published `since` rather than forwarding the empty meet, which +//! would be the empty frontier and would irreversibly release the trace. + +use std::collections::{BTreeMap, VecDeque}; +use std::sync::{Arc, Condvar, Mutex}; + +use timely::dataflow::operators::generic::{source, Operator}; +use timely::dataflow::operators::CapabilitySet; +use timely::dataflow::Scope; +use timely::progress::{frontier::AntichainRef, Antichain}; +use timely::scheduling::activate::SyncActivator; + +use crate::lattice::{antichain_meet, Lattice}; +use crate::trace::cursor::{cursor_list, CursorList, Navigable}; +use crate::trace::{BatchReader, TraceReader}; + +use super::{Arranged, TraceAgent, TraceReplayInstruction}; + +/// The queue and wakeup for one importer registered against a publication point. +struct ImportQueue { + /// Replay instructions the publisher appends and the importer drains, mirroring the local + /// arrange replay queue. Batches carry a hint time that lower-bounds their updates. + instructions: VecDeque>, + /// Wakes the importer's source operator on the reader's worker when new instructions arrive. + activator: SyncActivator, +} + +/// State shared between a publisher (on the owning worker) and its readers (on any thread). +/// +/// The `chain`, `since`, and `upper` are always updated together under the lock, so every reader +/// observes a frontier-consistent view. +struct SharedTraceState { + /// The published chain, sourced from `map_batches`: contiguous descriptions including the + /// seal-only empty batches that never travel on the arrangement stream. Coverage is at least + /// `upper` (within a worker step it may briefly run ahead, never behind). + chain: Vec, + /// Logical compaction frontier of the published view. Reads at times not beyond `since` are not + /// accurate. A snapshot must pick a time at or beyond it. + since: Antichain, + /// Seal frontier: the join of the chain's batch uppers. Batches strictly below `upper` are + /// complete and readable. + upper: Antichain, + /// Per-registration logical holds. The publisher forwards their meet, falling back to `since` + /// when empty (never the destructive empty meet of zero holds). + logical_holds: BTreeMap>, + /// Per-registration physical holds, forwarded independently of the logical holds. + physical_holds: BTreeMap>, + /// Importer queues, keyed by registration id. A handle may back several registrations, so this + /// is keyed separately from any handle. + queues: BTreeMap>, + /// Monotonic source of registration ids for holds and queues. + next_id: usize, + /// Set when the publisher drops. A terminal empty frontier is enqueued to each importer, so + /// readers close only after draining what was already published. + closed: bool, +} + +impl SharedTraceState +where + Tr::Time: Lattice + Clone, +{ + /// The meet of `holds`, or `fallback` when there are none. + /// + /// Never returns the empty meet of zero holds, which is the empty frontier and would tell the + /// trace to compact everything, permanently releasing the publisher's capability. + fn compaction_target( + holds: &BTreeMap>, + fallback: &Antichain, + ) -> Antichain { + let mut iter = holds.values(); + match iter.next() { + None => fallback.clone(), + Some(first) => { + let mut acc = first.clone(); + for hold in iter { + acc = antichain_meet(&acc.borrow()[..], &hold.borrow()[..]); + } + acc + } + } + } +} + +/// A publication point paired with its live [`Condvar`], so peek waiters can block for `upper` to +/// advance without a lost-wakeup race: the publisher signals under the same lock it uses to move +/// `upper`. +struct SharedTrace { + state: Mutex>, + upper_changed: Condvar, +} + +type SharedTraceRef = Arc>; + +/// The result of publishing an arrangement. Holding it keeps the publication point registered; +/// dropping it does not stop the publisher (the publisher lives with its dataflow), but no further +/// handles can be minted from it. +pub struct Published { + shared: SharedTraceRef, +} + +impl Published +where + Tr::Time: Lattice + Clone, +{ + /// Hands out a `Clone + Send` handle to the published arrangement. + /// + /// The handle registers a logical hold at the current published `since`, so the arrangement + /// will not compact past it until the handle (and all its clones) drop. + pub fn handle(&self) -> SharedTraceHandle { + SharedTraceHandle::register(Arc::clone(&self.shared)) + } +} + +/// A `Clone + Send` reader of a published arrangement. +/// +/// Implements [`TraceReader`], so downstream operators drive its compaction and acquire cursors as +/// with any trace handle. Each clone carries an independent registration: cloning mints a fresh id +/// and copies the source's holds, so two consumers of one import cannot release each other's holds. +pub struct SharedTraceHandle { + shared: SharedTraceRef, + /// This handle's hold registration id. + id: usize, + /// This handle's own logical frontier, mirrored into `logical_holds[id]`. Kept locally so + /// `get_logical_compaction` can return a borrow. + logical: Antichain, + /// This handle's own physical frontier, mirrored into `physical_holds[id]`. + physical: Antichain, +} + +impl SharedTraceHandle +where + Tr::Time: Lattice + Clone, +{ + /// Registers a fresh hold at the current published `since` and returns a handle for it. + fn register(shared: SharedTraceRef) -> Self { + let (id, since) = { + let mut state = shared.state.lock().expect("shared trace poisoned"); + let id = state.next_id; + state.next_id += 1; + let since = state.since.clone(); + state.logical_holds.insert(id, since.clone()); + state.physical_holds.insert(id, since.clone()); + (id, since) + }; + Self { + shared, + id, + logical: since.clone(), + physical: since, + } + } + + /// Takes a consistent snapshot of the published arrangement as of a time at or beyond `since` + /// and strictly below `upper`, blocking until `upper` passes `time` (or the publisher closes). + /// + /// Works from any thread. Returns `None` if the publisher closed before `upper` passed `time`. + pub fn snapshot_at(&self, time: &Tr::Time) -> Option> { + let mut state = self.shared.state.lock().expect("shared trace poisoned"); + // `upper` not less-equal `time` means all updates at `time` are sealed. + while state.upper.less_equal(time) { + if state.closed { + return None; + } + state = self + .shared + .upper_changed + .wait(state) + .expect("shared trace poisoned"); + } + Some(TraceSnapshot { + chain: state.chain.clone(), + since: state.since.clone(), + upper: state.upper.clone(), + }) + } + + /// Writes this handle's logical or physical hold into the shared registry. + fn update_hold(&self, logical: bool) { + let mut state = self.shared.state.lock().expect("shared trace poisoned"); + if logical { + state.logical_holds.insert(self.id, self.logical.clone()); + } else { + state.physical_holds.insert(self.id, self.physical.clone()); + } + } +} + +impl Clone for SharedTraceHandle +where + Tr::Time: Lattice + Clone, +{ + fn clone(&self) -> Self { + // A clone must be an independent hold: `import` returns `Arranged { trace: handle.clone() }` + // and `Arranged` is itself `Clone`, so distinct downstream operators drive compaction on + // distinct clones. Sharing one hold slot would let the faster operator release the slower + // one's hold. This mirrors `TraceAgent::clone`, which registers an independent counted hold. + let id = { + let mut state = self.shared.state.lock().expect("shared trace poisoned"); + let id = state.next_id; + state.next_id += 1; + state.logical_holds.insert(id, self.logical.clone()); + state.physical_holds.insert(id, self.physical.clone()); + id + }; + Self { + shared: Arc::clone(&self.shared), + id, + logical: self.logical.clone(), + physical: self.physical.clone(), + } + } +} + +impl Drop for SharedTraceHandle { + fn drop(&mut self) { + if let Ok(mut state) = self.shared.state.lock() { + state.logical_holds.remove(&self.id); + state.physical_holds.remove(&self.id); + } + } +} + +impl TraceReader for SharedTraceHandle +where + Tr::Time: Lattice + Clone, +{ + type Time = Tr::Time; + type Batch = Tr::Batch; + + fn batches_through(&mut self, upper: AntichainRef) -> Option> { + let state = self.shared.state.lock().expect("shared trace poisoned"); + // A clean cut of the published chain: all non-empty batches whose upper is not beyond + // `upper`, and none whose lower is beyond `upper`. Empty batches are dropped, as + // `Spine::batches_through` does. The published chain is totally ordered by description, so a + // straddling non-empty batch would be a bug upstream, not something to handle here. + let mut out = Vec::new(); + for batch in state.chain.iter() { + // A batch whose lower is beyond the cut, and everything after it in the totally + // ordered chain, lies past `upper`. Empty batches never carry updates to read. + if timely::PartialOrder::less_equal(&upper, &batch.lower().borrow()) { + break; + } + if !batch.is_empty() { + out.push(batch.clone()); + } + } + Some(out) + } + + fn set_logical_compaction(&mut self, frontier: AntichainRef) { + self.logical = frontier.to_owned(); + self.update_hold(true); + } + + fn get_logical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { + self.logical.borrow() + } + + fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, Tr::Time>) { + self.physical = frontier.to_owned(); + self.update_hold(false); + } + + fn get_physical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { + self.physical.borrow() + } + + fn map_batches(&self, mut f: F) { + let state = self.shared.state.lock().expect("shared trace poisoned"); + for batch in state.chain.iter() { + f(batch); + } + } +} + +/// Smallest time, used only to satisfy the borrow in the cut check for empty lower frontiers. +fn batch_min() -> Tr::Time { + ::minimum() +} + +/// An owned, consistent snapshot of a published arrangement: an immutable chain plus its frontiers. +/// +/// Holding it pins the chain's batches, keeping their memory alive even as the publishing worker +/// merges. Dropping it releases them, from whatever thread drops last. +pub struct TraceSnapshot { + chain: Vec, + since: Antichain, + upper: Antichain, +} + +impl TraceSnapshot { + /// The logical compaction frontier of the snapshot. Times not beyond it are not accurate. + pub fn since(&self) -> AntichainRef<'_, Tr::Time> { + self.since.borrow() + } + /// The seal frontier of the snapshot. Times strictly below it are complete. + pub fn upper(&self) -> AntichainRef<'_, Tr::Time> { + self.upper.borrow() + } + /// A cursor merging the snapshot's batch cursors, with the batches as its storage. + pub fn cursor(&self) -> (CursorList<::Cursor>, Vec) + where + Tr::Batch: Navigable, + { + cursor_list(self.chain.clone()) + } +} + +impl<'scope, Tr> Arranged<'scope, TraceAgent> +where + Tr: crate::trace::Trace + 'static, + Tr::Batch: Send + Sync, + Tr::Time: Lattice + Clone + Send + Sync, +{ + /// Publishes this arrangement through a publication point on the owning worker. + /// + /// Attaches a publisher operator to the arrangement stream. On each activation the publisher + /// refreshes the published chain, `since`, and `upper` from the trace, appends newly arrived + /// batches to importer queues, and forwards the meet of reader holds to the trace's compaction. + /// The returned [`Published`] mints [`SharedTraceHandle`]s that read the arrangement from any + /// thread. + pub fn publish(&self) -> Published { + self.publish_named("PublishShared") + } + + /// [`Arranged::publish`], with a name for the publisher operator. + pub fn publish_named(&self, name: &str) -> Published { + // The publisher owns a `TraceAgent` clone: its read capability is the aggregate lease for + // all readers, so the trace cannot compact or drop out from under them. + let mut agent = self.trace.clone(); + + let initial_since = agent.get_logical_compaction().to_owned(); + let shared: SharedTraceRef = Arc::new(SharedTrace { + state: Mutex::new(SharedTraceState { + chain: Vec::new(), + since: initial_since, + // NOTE: `Antichain::from_elem(minimum)`, never `Antichain::new()`. The empty + // frontier reads as "complete through the end of time", making every snapshot wait + // vacuously true and returning empty results instead of blocking. + upper: Antichain::from_elem(batch_min::()), + logical_holds: BTreeMap::new(), + physical_holds: BTreeMap::new(), + queues: BTreeMap::new(), + next_id: 0, + closed: false, + }), + upper_changed: Condvar::new(), + }); + + let publisher = Publisher { + shared: Arc::clone(&shared), + }; + + let sink_shared = Arc::clone(&shared); + self.stream.clone().sink( + timely::dataflow::channels::pact::Pipeline, + name, + move |(input, _frontier)| { + // Keep `publisher` alive with the operator, so operator (dataflow) drop closes the + // publication point. + let _publisher = &publisher; + + // Batches arriving on the stream, each with a capability time that lower-bounds the + // batch's updates. Empty seal batches do not travel the stream; they are picked up + // from the trace below. + let mut arrived: Vec<(Tr::Batch, Tr::Time)> = Vec::new(); + input.for_each(|cap, data| { + let hint = cap.time().clone(); + for batch in data.drain(..) { + arrived.push((batch, hint.clone())); + } + }); + + // Refresh the chain and frontiers from the trace, the authoritative source. + let mut chain = Vec::new(); + let mut upper = Antichain::new(); + agent.map_batches(|batch| { + chain.push(batch.clone()); + upper = batch.upper().to_owned(); + }); + if chain.is_empty() { + upper = Antichain::from_elem(batch_min::()); + } + let since = agent.get_logical_compaction().to_owned(); + + let (logical_target, physical_target) = { + let mut state = sink_shared.state.lock().expect("shared trace poisoned"); + + for (batch, hint) in arrived.drain(..) { + for queue in state.queues.values_mut() { + queue.instructions.push_back(TraceReplayInstruction::Batch( + batch.clone(), + Some(hint.clone()), + )); + } + } + + let upper_advanced = state.upper != upper; + if upper_advanced { + for queue in state.queues.values_mut() { + queue + .instructions + .push_back(TraceReplayInstruction::Frontier(upper.clone())); + } + } + + let logical = + SharedTraceState::::compaction_target(&state.logical_holds, &since); + let physical = + SharedTraceState::::compaction_target(&state.physical_holds, &since); + + state.chain = chain; + // Publish the `since` the trace will actually hold after we forward `logical` + // below, which is `join(since, logical)` because agent compaction only advances + // and is irreversible. Publishing the raw pre-forward `since` would let a handle + // registering in this window latch a floor below the trace's real compaction, + // an anti-conservative `since` that claims accuracy at already-merged times. + state.since = since.join(&logical); + state.upper = upper; + + // Wake importers and any peek waiters. + for queue in state.queues.values() { + let _ = queue.activator.activate(); + } + if upper_advanced { + sink_shared.upper_changed.notify_all(); + } + + (logical, physical) + }; + + // Apply compaction to the agent OUTSIDE the lock: `set_physical_compaction` can run + // an unbounded merge synchronously, which must not block concurrent readers. + agent.set_logical_compaction(logical_target.borrow()); + agent.set_physical_compaction(physical_target.borrow()); + }, + ); + + Published { shared } + } +} + +/// Guard that marks the publication point closed when the publisher operator drops, waking readers +/// so they drain and shut down. +struct Publisher { + shared: SharedTraceRef, +} + +impl Drop for Publisher { + fn drop(&mut self) { + if let Ok(mut state) = self.shared.state.lock() { + state.closed = true; + let empty = Antichain::new(); + for queue in state.queues.values_mut() { + queue + .instructions + .push_back(TraceReplayInstruction::Frontier(empty.clone())); + let _ = queue.activator.activate(); + } + } + self.shared.upper_changed.notify_all(); + } +} + +impl SharedTraceHandle +where + Tr: 'static, + Tr::Time: Lattice + Clone, + Tr::Batch: Navigable, +{ + /// Imports the published arrangement into `scope`, returning an [`Arranged`] backed by this + /// shared trace. + /// + /// The importer registers a replay queue seeded with the current chain and drains it as the + /// publisher appends. The registration is owned by the source operator, so dropping the import + /// dataflow deregisters it and releases its holds even while other handle clones and the reader + /// worker live on. + pub fn import<'scope>( + &self, + scope: Scope<'scope, Tr::Time>, + name: &str, + ) -> Arranged<'scope, SharedTraceHandle> { + let shared = Arc::clone(&self.shared); + let trace = self.clone(); + + let stream = source(scope, name, move |capability, info| { + let activator = scope.worker().sync_activator_for(info.address.to_vec()); + + // Register under one lock acquisition: mint an id, seed the queue with the current + // chain (hint `minimum`, as the local replay does for historical batches) followed by + // the current upper, and install the queue. Later batches append; earlier ones are + // seeded. Nothing is missed or duplicated. + let reg_id = { + let mut state = shared.state.lock().expect("shared trace poisoned"); + let reg_id = state.next_id; + state.next_id += 1; + let mut instructions = VecDeque::new(); + for batch in state.chain.iter() { + instructions.push_back(TraceReplayInstruction::Batch( + batch.clone(), + Some(batch_min::()), + )); + } + instructions.push_back(TraceReplayInstruction::Frontier(state.upper.clone())); + // If the publisher already closed, its one-shot terminal frontier has been and gone, + // so seed our own. Otherwise a late importer would drain the chain and then wait + // forever for a frontier that never arrives, leaking its capability. Mirrors the + // `state.closed` guard in `snapshot_at`. + if state.closed { + instructions.push_back(TraceReplayInstruction::Frontier(Antichain::new())); + } + state.queues.insert( + reg_id, + ImportQueue { + instructions, + activator, + }, + ); + reg_id + }; + + let mut capabilities = Some(CapabilitySet::new()); + capabilities.as_mut().unwrap().insert(capability); + + // Deregisters the queue when the source operator (and thus this closure) drops. + let _guard = QueueGuard { + shared: Arc::clone(&shared), + reg_id, + }; + + move |output| { + let _guard = &_guard; + let mut drained = Vec::new(); + { + let mut state = shared.state.lock().expect("shared trace poisoned"); + if let Some(queue) = state.queues.get_mut(®_id) { + drained.extend(queue.instructions.drain(..)); + } + } + + if let Some(caps) = capabilities.as_mut() { + for instruction in drained { + match instruction { + TraceReplayInstruction::Frontier(frontier) => { + if frontier.is_empty() { + // Terminal frontier: publisher closed, no more capabilities. + capabilities = None; + break; + } + caps.downgrade(&frontier.borrow()[..]); + } + TraceReplayInstruction::Batch(batch, hint) => { + if let Some(time) = hint { + if !batch.is_empty() { + // Emit under a capability delayed to the batch's hint, a + // lower bound on its times. Never the batch upper: that + // would let the frontier pass updates still in the batch. + let cap = caps.delayed(&time); + output.session(&cap).give(batch); + } + } + } + } + } + } + } + }); + + Arranged { stream, trace } + } +} + +/// Deregisters an importer's replay queue when its source operator drops. +struct QueueGuard { + shared: SharedTraceRef, + reg_id: usize, +} + +impl Drop for QueueGuard { + fn drop(&mut self) { + if let Ok(mut state) = self.shared.state.lock() { + state.queues.remove(&self.reg_id); + } + } +} diff --git a/differential-dataflow/tests/sharing.rs b/differential-dataflow/tests/sharing.rs new file mode 100644 index 000000000..ee6d892af --- /dev/null +++ b/differential-dataflow/tests/sharing.rs @@ -0,0 +1,159 @@ +//! Sharing arrangements across threads and runtimes. + +use std::sync::mpsc; + +use timely::dataflow::operators::capture::Extract; +use timely::dataflow::operators::{Capture, Probe}; +use timely::dataflow::ProbeHandle; + +use differential_dataflow::input::InputSession; +use differential_dataflow::operators::arrange::sharing::SharedTraceHandle; +use differential_dataflow::operators::arrange::Arrange; +use differential_dataflow::trace::cursor::Cursor; +use differential_dataflow::trace::implementations::{ + ArcValBuilder, ArcValSpine, ValBatcher, ValBuilder, ValSpine, +}; + +// The shared spine is `Arc`-backed, since `publish` requires `Send + Sync` batches. The reduce +// output below stays on the default `Rc`-backed `ValSpine`/`ValBuilder`, as it is not shared. +type Spine = ArcValSpine; +type Handle = SharedTraceHandle; + +/// A worker on one thread publishes an arrangement. A separate thread holds a `Send` handle, waits +/// for the publication frontier to pass a time, snapshots, and reads the collection at that time. +#[test] +fn snapshot_from_another_thread() { + let (handle_tx, handle_rx) = mpsc::channel::(); + // The reader raises this once it has its snapshot, so the publishing worker knows it can stop + // stepping. A retained trace handle keeps the dataflow from quiescing, so the worker never + // finishes on its own. + let done = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let reader_done = std::sync::Arc::clone(&done); + + // Reader thread: receive the handle, snapshot as of time 2, accumulate. + let reader = std::thread::spawn(move || { + let handle = handle_rx.recv().unwrap(); + let snapshot = handle.snapshot_at(&2).expect("publisher closed early"); + reader_done.store(true, std::sync::atomic::Ordering::SeqCst); + let (mut cursor, storage) = snapshot.cursor(); + // Accumulate `(key, val) -> diff` for all times <= 2. + let mut acc: Vec<((u64, u64), isize)> = Vec::new(); + while cursor.key_valid(&storage) { + let key = *cursor.key(&storage); + while cursor.val_valid(&storage) { + let val = *cursor.val(&storage); + let mut sum = 0isize; + cursor.map_times(&storage, |t, d| { + if *t <= 2 { + sum += *d; + } + }); + if sum != 0 { + acc.push(((key, val), sum)); + } + cursor.step_val(&storage); + } + cursor.step_key(&storage); + } + acc.sort(); + acc + }); + + timely::execute_directly(move |worker| { + let mut input = InputSession::::new(); + let published = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + arranged.publish() + }); + handle_tx.send(published.handle()).unwrap(); + + // Time 0: (1,10)+1, (2,20)+1. Time 1: retract (2,20). Time 2: (3,30)+1. + input.advance_to(0); + input.insert((1, 10)); + input.insert((2, 20)); + input.advance_to(1); + input.remove((2, 20)); + input.advance_to(2); + input.insert((3, 30)); + input.advance_to(3); + input.flush(); + + // Step until the reader has taken its snapshot. The publisher advances `upper` as it steps, + // which unblocks the reader's `snapshot_at`. + while !done.load(std::sync::atomic::Ordering::SeqCst) { + worker.step(); + } + }); + + let got = reader.join().unwrap(); + // As of time 2: (1,10) present, (2,20) inserted then retracted, (3,30) inserted at 2. + assert_eq!(got, vec![((1, 10), 1), ((3, 30), 1)]); +} + +/// A publisher in one dataflow, an importer in another dataflow of the same worker, connected only +/// through a `SharedTraceHandle`. The importer reduces the shared arrangement and captures the +/// result, which must match a direct reduction of the input. +#[test] +fn import_through_shared_handle() { + let (capture_tx, capture_rx) = mpsc::channel(); + + timely::execute_directly(move |worker| { + let mut input = InputSession::::new(); + + // Publisher dataflow. + let handle = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + arranged.publish().handle() + }); + + // Importer dataflow: import the shared arrangement, count values per key, capture. + let mut probe = ProbeHandle::new(); + worker.dataflow(|scope| { + let imported = handle.import(scope, "Import"); + let counted = imported + .reduce_abelian::<_, ValBuilder<_, _, _, _>, ValSpine, _, _>( + "Count", + |_key, input, output| { + let count = input.iter().map(|&(_, d)| d).sum::() as u64; + output.push((count, 1)); + }, + |vec, key, upds| { + vec.clear(); + vec.extend(upds.drain(..).map(|(v, t, r)| ((*key, v), t, r))); + }, + ) + .as_collection(|k, v| (*k, *v)); + counted + .inner + .probe_with(&mut probe) + .capture_into(capture_tx.clone()); + }); + + // Key 1 has two values (10, 20), key 2 has one value (30), all at time 0. + input.advance_to(0); + input.insert((1, 10)); + input.insert((1, 20)); + input.insert((2, 30)); + input.advance_to(1); + input.flush(); + + // Step until the imported-and-reduced output has sealed time 0. The worker closure then + // returns, dropping the dataflows and flushing the capture. + while probe.less_than(&1) { + worker.step(); + } + }); + + // Reduce output at time 0: key 1 -> count 2, key 2 -> count 1. + let mut results: Vec<((u64, u64), u64, isize)> = capture_rx + .extract() + .into_iter() + .flat_map(|(_, data)| data) + .collect(); + results.sort(); + assert_eq!(results, vec![((1, 2), 0, 1), ((2, 1), 0, 1)]); +} From f497181fd11d7cbc81eeda246032236d4440abef Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 19:25:16 +0000 Subject: [PATCH 3/7] sharing: gate snapshot_at on since, fail-stop straddle, document pin/peers work Corrections from adversarial review and the two-runtime compute integration design: - snapshot_at returns None when since has advanced beyond the requested time, so a point read past compaction fails safe instead of returning coalesced results. Mirrors the single-runtime peek path's since gate. - batches_through asserts a non-empty batch does not straddle the requested cut, matching Spine::batches_through, so an import cannot silently return updates at times not before the cut and corrupt a downstream cursor_through. - TODO documenting that publishing currently pins compaction: the publisher forwards its own frozen hold, which never advances and pins the trace's effective since at publish time. The fix is design-laden (publishing must carry no independent floor) and is left to the integration with tests. - TODO documenting the equal-peers assertion import should carry. Read holds registered by import are correct and intended. Existing sharing and trace tests pass. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../src/operators/arrange/sharing.rs | 53 +++++++++++++++++-- 1 file changed, 48 insertions(+), 5 deletions(-) diff --git a/differential-dataflow/src/operators/arrange/sharing.rs b/differential-dataflow/src/operators/arrange/sharing.rs index 2b6dffa0c..eb34e7a1f 100644 --- a/differential-dataflow/src/operators/arrange/sharing.rs +++ b/differential-dataflow/src/operators/arrange/sharing.rs @@ -178,10 +178,15 @@ where } } - /// Takes a consistent snapshot of the published arrangement as of a time at or beyond `since` - /// and strictly below `upper`, blocking until `upper` passes `time` (or the publisher closes). + /// Takes a consistent snapshot of the published arrangement as of `time`, blocking until `upper` + /// passes `time`. /// - /// Works from any thread. Returns `None` if the publisher closed before `upper` passed `time`. + /// Works from any thread. Returns `None` when the snapshot cannot serve `time`, which is either + /// the publisher closed before `upper` passed `time`, or compaction has advanced `since` beyond + /// `time` so the accumulation at `time` is no longer accurate. The gate on `since` mirrors the + /// single-runtime peek path, which errors when the compaction frontier is beyond the read time + /// rather than returning coalesced results. A caller that needs to tell the two `None` cases + /// apart should inspect `since` before the read. pub fn snapshot_at(&self, time: &Tr::Time) -> Option> { let mut state = self.shared.state.lock().expect("shared trace poisoned"); // `upper` not less-equal `time` means all updates at `time` are sealed. @@ -195,6 +200,11 @@ where .wait(state) .expect("shared trace poisoned"); } + // `since` beyond `time` means times at `time` have been coalesced and a read there would be + // inaccurate. Fail to `None` rather than serve stale data. + if !state.since.less_equal(time) { + return None; + } Some(TraceSnapshot { chain: state.chain.clone(), since: state.since.clone(), @@ -259,8 +269,7 @@ where let state = self.shared.state.lock().expect("shared trace poisoned"); // A clean cut of the published chain: all non-empty batches whose upper is not beyond // `upper`, and none whose lower is beyond `upper`. Empty batches are dropped, as - // `Spine::batches_through` does. The published chain is totally ordered by description, so a - // straddling non-empty batch would be a bug upstream, not something to handle here. + // `Spine::batches_through` does. let mut out = Vec::new(); for batch in state.chain.iter() { // A batch whose lower is beyond the cut, and everything after it in the totally @@ -269,6 +278,15 @@ where break; } if !batch.is_empty() { + // Fail-stop on a batch that straddles the cut (`lower < upper < batch.upper()`), + // matching `Spine::batches_through`. Returning it would hand back updates at times + // not before `upper`, corrupting a downstream `cursor_through` consumer such as + // `join`. The published chain is totally ordered by description, so this cut is + // clean unless a caller requested a frontier that is not batch-aligned. + assert!( + timely::PartialOrder::less_equal(&batch.upper().borrow(), &upper), + "batches_through: upper straddles batch" + ); out.push(batch.clone()); } } @@ -409,6 +427,25 @@ where if chain.is_empty() { upper = Antichain::from_elem(batch_min::()); } + // TODO(sharing): this pins compaction. `get_logical_compaction` returns the + // publisher agent's OWN hold, which starts at the publish-time `since` and, with no + // readers, `compaction_target` returns it unchanged (see below) and + // `set_logical_compaction` re-applies it as a no-op. So the publisher agent's hold + // never advances, and because that agent registers a counted hold in the trace's + // `TraceBox`, the trace's effective compaction is pinned at the publish-time `since` + // for the life of the arrangement, regardless of the writer/controller advancing a + // different handle. Merely publishing then prevents the trace from ever compacting + // or merging. + // + // The fix is that publishing must carry no independent compaction floor. The + // trace's writer (and, in Materialize, the controller) drive `since`, and only a + // live importer's own `as_of` hold may hold the trace back. Candidate mechanisms, + // to be settled with the integration and covered by tests: derive the published + // `since` from the trace's batch descriptions rather than the agent's own hold, and + // either drop the publisher's holding agent entirely (relying on the writer handle + // plus importer holds to keep the trace live) or continuously advance the + // publisher's hold to the writer-driven frontier so it never constrains. An + // importer's registered hold (correct and intended) stays as-is. let since = agent.get_logical_compaction().to_owned(); let (logical_target, physical_target) = { @@ -508,6 +545,12 @@ where scope: Scope<'scope, Tr::Time>, name: &str, ) -> Arranged<'scope, SharedTraceHandle> { + // TODO(sharing): assert the importing scope's total peer count equals the publisher's. + // Pairwise import (importer worker `i` reads publisher worker `i`) is sound only when both + // sides shard by the same `key.hashed() % peers`, which requires equal total peers + // (workers-per-process times processes), not just equal workers-per-process. Record the + // publisher's peer count at publish time in `SharedTrace` and assert `scope.peers()` matches + // here, so a misconfiguration fails loudly instead of silently reading the wrong shard. let shared = Arc::clone(&self.shared); let trace = self.clone(); From c442118d4e0f17deb55541839c93aeb262d3d23a Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Mon, 20 Jul 2026 23:03:29 +0200 Subject: [PATCH 4/7] fix(sharing): import asserts equal total peers Pairwise import (importer worker i reads publisher worker i) is sound only when both runtimes shard keys by the same total peer count. Record the publisher's peers at publish time and assert scope.peers() matches it in import, so a mismatch fails loudly instead of silently reading the wrong shard. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/operators/arrange/sharing.rs | 24 ++++++++--- differential-dataflow/tests/sharing.rs | 41 ++++++++++++++++++- 2 files changed, 58 insertions(+), 7 deletions(-) diff --git a/differential-dataflow/src/operators/arrange/sharing.rs b/differential-dataflow/src/operators/arrange/sharing.rs index eb34e7a1f..83b882c4b 100644 --- a/differential-dataflow/src/operators/arrange/sharing.rs +++ b/differential-dataflow/src/operators/arrange/sharing.rs @@ -115,6 +115,12 @@ where struct SharedTrace { state: Mutex>, upper_changed: Condvar, + /// Total peer count (workers-per-process times processes) of the scope that published this + /// arrangement. Set once at publish time and never mutated afterward, so `import` reads it + /// without taking `state`'s lock. Pairwise import (importer worker `i` reads publisher worker + /// `i`) is sound only when an importing scope shards keys the same way, which requires this to + /// match the importing scope's own `peers()`. + peers: usize, } type SharedTraceRef = Arc>; @@ -391,6 +397,7 @@ where closed: false, }), upper_changed: Condvar::new(), + peers: self.stream.scope().peers(), }); let publisher = Publisher { @@ -536,6 +543,11 @@ where /// Imports the published arrangement into `scope`, returning an [`Arranged`] backed by this /// shared trace. /// + /// Requires `scope`'s total peer count (workers-per-process times processes) to equal the + /// publisher's, panicking otherwise. Pairwise import (importer worker `i` reads publisher + /// worker `i`) is sound only when both sides shard by the same `key.hashed() % peers`; a + /// mismatched peer count would silently read the wrong shard instead of failing loudly. + /// /// The importer registers a replay queue seeded with the current chain and drains it as the /// publisher appends. The registration is owned by the source operator, so dropping the import /// dataflow deregisters it and releases its holds even while other handle clones and the reader @@ -545,12 +557,12 @@ where scope: Scope<'scope, Tr::Time>, name: &str, ) -> Arranged<'scope, SharedTraceHandle> { - // TODO(sharing): assert the importing scope's total peer count equals the publisher's. - // Pairwise import (importer worker `i` reads publisher worker `i`) is sound only when both - // sides shard by the same `key.hashed() % peers`, which requires equal total peers - // (workers-per-process times processes), not just equal workers-per-process. Record the - // publisher's peer count at publish time in `SharedTrace` and assert `scope.peers()` matches - // here, so a misconfiguration fails loudly instead of silently reading the wrong shard. + assert_eq!( + scope.peers(), + self.shared.peers, + "shared-trace import requires equal total peers (workers_per_process * num_processes)" + ); + let shared = Arc::clone(&self.shared); let trace = self.clone(); diff --git a/differential-dataflow/tests/sharing.rs b/differential-dataflow/tests/sharing.rs index ee6d892af..140cb3aef 100644 --- a/differential-dataflow/tests/sharing.rs +++ b/differential-dataflow/tests/sharing.rs @@ -1,6 +1,6 @@ //! Sharing arrangements across threads and runtimes. -use std::sync::mpsc; +use std::sync::{mpsc, Mutex}; use timely::dataflow::operators::capture::Extract; use timely::dataflow::operators::{Capture, Probe}; @@ -157,3 +157,42 @@ fn import_through_shared_handle() { results.sort(); assert_eq!(results, vec![((1, 2), 0, 1), ((2, 1), 0, 1)]); } + +/// A publisher on a two-worker runtime (`peers() == 2`) hands its handle to an importer on a +/// single-threaded runtime (`peers() == 1`). Pairwise import assumes both sides shard keys the +/// same way, which requires equal total peers, so `import` must assert and panic rather than +/// silently reading the wrong shard. +#[test] +#[should_panic(expected = "peers")] +fn import_asserts_equal_peers() { + let (handle_tx, handle_rx) = mpsc::channel::(); + let handle_tx = Mutex::new(handle_tx); + + // Publisher runtime: two worker threads, so the publishing scope's `peers()` is 2. + timely::execute(timely::Config::process(2), move |worker| { + let mut input = InputSession::::new(); + let published = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + arranged.publish() + }); + // Only one worker needs to hand out a handle; the others publish redundantly (mirroring + // real SPMD dataflows) but nobody reads their handles. + if worker.index() == 0 { + handle_tx.lock().unwrap().send(published.handle()).unwrap(); + } + }) + .expect("publisher runtime failed to start"); + + let handle = handle_rx.recv().expect("publisher did not send a handle"); + + // Importer runtime: single-threaded (`execute_directly` never spawns worker threads), so + // `peers()` is 1, mismatching the publisher's 2. `import` runs on this same thread, so its + // panic unwinds directly into the test rather than being swallowed at a thread boundary. + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let _imported = handle.import(scope, "Import"); + }); + }); +} From 63ce1ba65ece59e4ae30340da5980cd2ac2dccd4 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Mon, 20 Jul 2026 23:29:51 +0200 Subject: [PATCH 5/7] fix(sharing): publishing no longer pins compaction Publishing an arrangement through the shared-trace primitive froze the trace's logical compaction at the publish-time since for the life of the arrangement, even with zero readers, because the publisher's own agent hold pinned the trace-box meet and never advanced. Source the writer-driven frontier from the trace box's accumulated holds with the publisher's own contribution removed, and advance the publisher's hold to it so it follows the writer instead of pinning. The per-batch since cannot serve as this source: it only advances when the Spine compacts, which the publisher's own hold prevents. Publish exactly the trace's real compaction so a live importer hold keeps its own frontier readable. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/operators/arrange/sharing.rs | 107 ++++++++++------ differential-dataflow/tests/sharing.rs | 116 ++++++++++++++++++ 2 files changed, 188 insertions(+), 35 deletions(-) diff --git a/differential-dataflow/src/operators/arrange/sharing.rs b/differential-dataflow/src/operators/arrange/sharing.rs index 83b882c4b..b1c11b374 100644 --- a/differential-dataflow/src/operators/arrange/sharing.rs +++ b/differential-dataflow/src/operators/arrange/sharing.rs @@ -25,9 +25,11 @@ //! //! Every reader registers logical and physical holds. The publisher forwards their *meet* (the //! greatest lower bound, so the trace never compacts past the least reader's hold) to its own -//! `TraceAgent`, which is the sole writer of the trace's compaction frontiers. With no readers the -//! publisher holds compaction at the published `since` rather than forwarding the empty meet, which -//! would be the empty frontier and would irreversibly release the trace. +//! `TraceAgent`, which is the sole writer of the trace's compaction frontiers. Publishing itself +//! carries no compaction floor: with no readers the publisher advances its hold to the +//! writer-driven frontier (the meet of the other agents' holds), so the trace compacts and merges +//! as the writer advances. The publisher never forwards the empty frontier, which would +//! irreversibly release the trace. use std::collections::{BTreeMap, VecDeque}; use std::sync::{Arc, Condvar, Mutex}; @@ -35,7 +37,8 @@ use std::sync::{Arc, Condvar, Mutex}; use timely::dataflow::operators::generic::{source, Operator}; use timely::dataflow::operators::CapabilitySet; use timely::dataflow::Scope; -use timely::progress::{frontier::AntichainRef, Antichain}; +use timely::progress::frontier::{AntichainRef, MutableAntichain}; +use timely::progress::Antichain; use timely::scheduling::activate::SyncActivator; use crate::lattice::{antichain_meet, Lattice}; @@ -68,8 +71,8 @@ struct SharedTraceState { /// Seal frontier: the join of the chain's batch uppers. Batches strictly below `upper` are /// complete and readable. upper: Antichain, - /// Per-registration logical holds. The publisher forwards their meet, falling back to `since` - /// when empty (never the destructive empty meet of zero holds). + /// Per-registration logical holds. The publisher forwards their meet, falling back to the + /// writer-driven frontier when empty (never the destructive empty meet of zero holds). logical_holds: BTreeMap>, /// Per-registration physical holds, forwarded independently of the logical holds. physical_holds: BTreeMap>, @@ -434,26 +437,48 @@ where if chain.is_empty() { upper = Antichain::from_elem(batch_min::()); } - // TODO(sharing): this pins compaction. `get_logical_compaction` returns the - // publisher agent's OWN hold, which starts at the publish-time `since` and, with no - // readers, `compaction_target` returns it unchanged (see below) and - // `set_logical_compaction` re-applies it as a no-op. So the publisher agent's hold - // never advances, and because that agent registers a counted hold in the trace's - // `TraceBox`, the trace's effective compaction is pinned at the publish-time `since` - // for the life of the arrangement, regardless of the writer/controller advancing a - // different handle. Merely publishing then prevents the trace from ever compacting - // or merging. + // Contract: publishing carries no independent compaction floor. The trace's writer + // (and, in Materialize, the controller) drive `since` through their own trace + // handles. Only a live importer's registered hold may hold the trace back, and it + // releases on drop. The publisher keeps a holding agent solely so importer holds + // have somewhere to forward to, so that hold must FOLLOW the writer rather than pin + // the trace. // - // The fix is that publishing must carry no independent compaction floor. The - // trace's writer (and, in Materialize, the controller) drive `since`, and only a - // live importer's own `as_of` hold may hold the trace back. Candidate mechanisms, - // to be settled with the integration and covered by tests: derive the published - // `since` from the trace's batch descriptions rather than the agent's own hold, and - // either drop the publisher's holding agent entirely (relying on the writer handle - // plus importer holds to keep the trace live) or continuously advance the - // publisher's hold to the writer-driven frontier so it never constrains. An - // importer's registered hold (correct and intended) stays as-is. - let since = agent.get_logical_compaction().to_owned(); + // The writer-driven frontier is the meet of all other agents' holds, i.e. the + // frontier the trace would compact to if the publisher were not holding it. The + // `TraceBox` accumulates every agent's hold in a `MutableAntichain`; cloning it and + // subtracting the publisher's own contribution yields that meet. Advancing the + // publisher's hold to it lets the trace compact and merge as the writer advances. + // + // NOTE: the per-batch `since` from `map_batches` cannot serve as this source. Those + // frontiers advance only when the Spine actually compacts, which the publisher's own + // pinning hold prevents, so they stay frozen at the publish-time `since`. Reading + // the trace-box meet breaks that circularity. + let publisher_logical = agent.get_logical_compaction().to_owned(); + let publisher_physical = agent.get_physical_compaction().to_owned(); + // The writer-driven frontier for one dimension: the meet of the accumulated holds + // with the publisher's own contribution removed. An empty result means the + // publisher is the sole holder. Never forward the empty frontier: it would compact + // everything and release the publisher's capability, so fall back to the publisher's + // current hold and keep the current floor. + let others_meet = + |accumulated: &MutableAntichain, own: &Antichain| { + let mut others = accumulated.clone(); + others.update_iter(own.iter().map(|t| (t.clone(), -1))); + if others.frontier().is_empty() { + own.clone() + } else { + others.frontier().to_owned() + } + }; + let (writer_logical, writer_physical) = { + let trace_box = agent.trace_box_unstable(); + let trace_box = trace_box.borrow(); + ( + others_meet(&trace_box.logical_compaction, &publisher_logical), + others_meet(&trace_box.physical_compaction, &publisher_physical), + ) + }; let (logical_target, physical_target) = { let mut state = sink_shared.state.lock().expect("shared trace poisoned"); @@ -476,18 +501,30 @@ where } } - let logical = - SharedTraceState::::compaction_target(&state.logical_holds, &since); - let physical = - SharedTraceState::::compaction_target(&state.physical_holds, &since); + // Fall back to the writer-driven frontier (not the publisher's frozen hold) when + // there are no reader holds, so with zero readers the target follows the writer. + let logical = SharedTraceState::::compaction_target( + &state.logical_holds, + &writer_logical, + ); + let physical = SharedTraceState::::compaction_target( + &state.physical_holds, + &writer_physical, + ); state.chain = chain; - // Publish the `since` the trace will actually hold after we forward `logical` - // below, which is `join(since, logical)` because agent compaction only advances - // and is irreversible. Publishing the raw pre-forward `since` would let a handle - // registering in this window latch a floor below the trace's real compaction, - // an anti-conservative `since` that claims accuracy at already-merged times. - state.since = since.join(&logical); + // Publish the trace's real logical compaction after we forward `logical` below. + // Agent compaction only advances (joins), so the publisher's hold becomes + // `join(publisher_logical, logical)`. The trace's real compaction is the meet of + // every agent's hold: the publisher's post-forward hold and the writer-driven + // meet of the others. Publishing exactly that keeps the gate in step with the + // trace, so a reader hold keeps its own frontier readable rather than being + // raced past by the writer. It is never below the real compaction (it equals + // it), so a handle registering in this window cannot latch an anti-conservative + // `since` that claims accuracy at already-merged times. + let publisher_after = publisher_logical.join(&logical); + state.since = + antichain_meet(&publisher_after.borrow()[..], &writer_logical.borrow()[..]); state.upper = upper; // Wake importers and any peek waiters. diff --git a/differential-dataflow/tests/sharing.rs b/differential-dataflow/tests/sharing.rs index 140cb3aef..4e5841fae 100644 --- a/differential-dataflow/tests/sharing.rs +++ b/differential-dataflow/tests/sharing.rs @@ -6,6 +6,8 @@ use timely::dataflow::operators::capture::Extract; use timely::dataflow::operators::{Capture, Probe}; use timely::dataflow::ProbeHandle; +use timely::progress::Antichain; + use differential_dataflow::input::InputSession; use differential_dataflow::operators::arrange::sharing::SharedTraceHandle; use differential_dataflow::operators::arrange::Arrange; @@ -13,6 +15,7 @@ use differential_dataflow::trace::cursor::Cursor; use differential_dataflow::trace::implementations::{ ArcValBuilder, ArcValSpine, ValBatcher, ValBuilder, ValSpine, }; +use differential_dataflow::trace::TraceReader; // The shared spine is `Arc`-backed, since `publish` requires `Send + Sync` batches. The reduce // output below stays on the default `Rc`-backed `ValSpine`/`ValBuilder`, as it is not shared. @@ -158,6 +161,119 @@ fn import_through_shared_handle() { assert_eq!(results, vec![((1, 2), 0, 1), ((2, 1), 0, 1)]); } +/// Feeds `input` a fresh update at `time` and steps the worker, so the publisher operator +/// reactivates and republishes its `since` from the trace. The publisher only runs on stream +/// input, so a bare compaction advance on another handle is invisible until the next batch. +fn tick( + worker: &mut timely::worker::Worker, + input: &mut InputSession, + time: u64, +) { + input.advance_to(time); + input.insert((time, time)); + input.advance_to(time + 1); + input.flush(); + for _ in 0..20 { + worker.step(); + } +} + +/// Publishing must not pin compaction. With no registered reader holds, as the trace's writer +/// advances logical (and physical) compaction the publisher's own hold must follow, so the trace +/// actually compacts. We keep a writer handle on the original trace, advance its compaction to +/// `10`, tick the publisher, and assert the published `since` advanced: a fresh snapshot rejects a +/// read at `5` (a compacted time) while still serving `10`. Before the fix the publisher pins the +/// trace at the publish-time frontier, so `since` stays at `0` and the read at `5` is served. +#[test] +fn publish_without_readers_does_not_pin_compaction() { + timely::execute_directly(move |worker| { + let mut input = InputSession::::new(); + // Keep a writer handle (a plain `TraceAgent` clone) alongside the publication. This is the + // only hold besides the publisher's own agent. Crucially, no `SharedTraceHandle` is minted, + // so `state.logical_holds` stays empty: zero registered reader holds. + let (mut writer, published) = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + (arranged.trace.clone(), arranged.publish()) + }); + + // Seed some updates and let the publisher settle. + for t in 0..5 { + tick(worker, &mut input, t); + } + + // The writer advances its logical and physical compaction to 10, as a controller would, + // then a fresh batch reactivates the publisher so it republishes `since`. + writer.set_logical_compaction(Antichain::from_elem(10).borrow()); + writer.set_physical_compaction(Antichain::from_elem(10).borrow()); + tick(worker, &mut input, 10); + + // The published `since` followed the writer to 10: a fresh handle cannot snapshot at the + // compacted time 5, but can at 10. + let handle = published.handle(); + assert!( + handle.snapshot_at(&5).is_none(), + "snapshot at a compacted time must be rejected (since did not advance)" + ); + assert!( + handle.snapshot_at(&10).is_some(), + "snapshot at the compaction frontier must succeed" + ); + }); +} + +/// A live reader hold holds the trace back at its own frontier, and releasing it (drop) lets the +/// trace follow the writer. A reader registered at `since` 0 keeps time 0 readable even as the +/// writer advances to 10: the publisher forwards the reader's hold, so the trace stays at 0 and the +/// published `since` stays at 0. After the reader drops, a further tick lets the publisher follow +/// the writer, and a read at the now-compacted time 5 is rejected. This guards the fix on the "with +/// holds" side. Before the fix the publisher pins the trace regardless, so the post-drop compaction +/// never happens. +#[test] +fn import_hold_pins_then_releases() { + timely::execute_directly(move |worker| { + let mut input = InputSession::::new(); + let (mut writer, published) = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + (arranged.trace.clone(), arranged.publish()) + }); + + // A live reader hold registered at the publish-time `since` (0). + let reader = published.handle(); + + for t in 0..5 { + tick(worker, &mut input, t); + } + + // The writer advances to 10, but the live reader hold (0) holds the trace back: the + // publisher forwards the reader's hold, so the published `since` stays at 0 and a read at 0 + // still succeeds. + writer.set_logical_compaction(Antichain::from_elem(10).borrow()); + writer.set_physical_compaction(Antichain::from_elem(10).borrow()); + tick(worker, &mut input, 10); + assert!( + reader.snapshot_at(&0).is_some(), + "a live reader hold must keep its frontier readable" + ); + + // Drop the reader. Now only the publisher holds, and it follows the writer, so the trace + // compacts to 10. A handle minted afterward (so it does not itself hold the trace at 0) + // observes the advanced `since`: a read at the compacted time 5 is rejected. Before the fix + // the publisher pins the trace even with no readers, so `since` stays at 0 and this read + // succeeds. + drop(reader); + tick(worker, &mut input, 11); + let observer = published.handle(); + assert!( + observer.snapshot_at(&5).is_none(), + "after the reader drops, the trace must compact to the writer frontier" + ); + }); +} + /// A publisher on a two-worker runtime (`peers() == 2`) hands its handle to an importer on a /// single-threaded runtime (`peers() == 1`). Pairwise import assumes both sides shard keys the /// same way, which requires equal total peers, so `import` must assert and panic rather than From b57e366bef921adf411a32ca59bba07bd0616f7c Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Tue, 21 Jul 2026 21:19:41 +0200 Subject: [PATCH 6/7] sharing: add SharedTraceHandle::import_snapshot_at for single-time reads Imports a published arrangement as a static snapshot at `as_of`, emitting the sealed chain once (when `upper` reaches `as_of`) through a TraceFrontier/ BatchFrontier wrapper advanced to `as_of` and bounded by `until`, then releasing the capability. Unlike the live `import`, there is no forward change stream, so the forward-batch capability handling and the lack of `as_of` coalescing (both unsound for a read at `as_of`) do not apply. Times are advanced on read, so the shared Arc batches are reused, never re-arranged. Co-Authored-By: Claude Opus 4.8 --- .../src/operators/arrange/sharing.rs | 120 ++++++++++++++++++ 1 file changed, 120 insertions(+) diff --git a/differential-dataflow/src/operators/arrange/sharing.rs b/differential-dataflow/src/operators/arrange/sharing.rs index b1c11b374..727e06135 100644 --- a/differential-dataflow/src/operators/arrange/sharing.rs +++ b/differential-dataflow/src/operators/arrange/sharing.rs @@ -43,6 +43,7 @@ use timely::scheduling::activate::SyncActivator; use crate::lattice::{antichain_meet, Lattice}; use crate::trace::cursor::{cursor_list, CursorList, Navigable}; +use crate::trace::wrappers::frontier::{BatchFrontier, TraceFrontier}; use crate::trace::{BatchReader, TraceReader}; use super::{Arranged, TraceAgent, TraceReplayInstruction}; @@ -688,6 +689,125 @@ where Arranged { stream, trace } } + + /// Imports a static snapshot of the published arrangement at `as_of`, for single-time reads. + /// + /// Unlike [`Self::import`], which replays the live change stream, this emits the published chain + /// exactly once (when the arrangement has sealed through `as_of`) and then releases its + /// capability, so the imported collection is complete for a single logical time. Times are + /// presented through a [`TraceFrontier`]/[`BatchFrontier`] wrapper advanced to `as_of` and + /// bounded by `until`, so a downstream reader sees the accumulation at `as_of` with no + /// pre-`as_of` retractions. This is what a one-shot peek or its query dataflow needs, and it + /// sidesteps the live import's forward-batch capability handling entirely: there is no forward + /// stream. + /// + /// The wrapper advances times on read, so the shared `Arc` batches are reused as-is, never + /// re-arranged. + /// + /// The snapshot is emitted only once `upper` is strictly beyond `as_of` (the same gate as + /// [`Self::snapshot_at`]), so the accumulation at `as_of` is final. Until then the operator holds + /// its capability at the minimum time and re-checks on each publisher activation, so the output + /// frontier does not advance and a downstream result read waits rather than observing a partial + /// (often empty) snapshot. The caller's read hold must keep `since` at or below `as_of`, matching + /// [`Self::snapshot_at`]; a `since` past `as_of` would make the read inaccurate. + /// + /// Requires `scope`'s total peer count to equal the publisher's, as [`Self::import`] does. + pub fn import_snapshot_at<'scope>( + &self, + scope: Scope<'scope, Tr::Time>, + name: &str, + as_of: Antichain, + until: Antichain, + ) -> Arranged<'scope, TraceFrontier>> { + assert_eq!( + scope.peers(), + self.shared.peers, + "shared-trace import requires equal total peers (workers_per_process * num_processes)" + ); + + let trace = TraceFrontier::make_from(self.clone(), as_of.borrow(), until.borrow()); + let shared = Arc::clone(&self.shared); + + let stream = source(scope, name, move |capability, info| { + let activator = scope.worker().sync_activator_for(info.address.to_vec()); + + // Register an (instruction-less) queue purely to receive the publisher's activations on + // each `upper` advance and on close, so we re-check the seal condition without polling. + let reg_id = { + let mut state = shared.state.lock().expect("shared trace poisoned"); + let reg_id = state.next_id; + state.next_id += 1; + state.queues.insert( + reg_id, + ImportQueue { + instructions: VecDeque::new(), + activator, + }, + ); + reg_id + }; + + // Deregisters the queue when the source operator (and thus this closure) drops. + let _guard = QueueGuard { + shared: Arc::clone(&shared), + reg_id, + }; + + let mut capabilities = Some(CapabilitySet::new()); + capabilities.as_mut().unwrap().insert(capability); + + move |output| { + let _guard = &_guard; + let Some(caps) = capabilities.as_mut() else { + // The snapshot was already emitted and the capability released. + return; + }; + + let (chain, upper, closed) = { + let mut state = shared.state.lock().expect("shared trace poisoned"); + // Drain any queued instructions to reset the coalescing activator flag; this + // path reads `chain`/`upper` directly rather than replaying instructions. + if let Some(queue) = state.queues.get_mut(®_id) { + queue.instructions.clear(); + } + (state.chain.clone(), state.upper.clone(), state.closed) + }; + + // Emit once `upper` is strictly beyond every `as_of` time, mirroring the wait in + // `snapshot_at`: only then is the accumulation at `as_of` final (all updates at times + // not beyond `as_of` are sealed, so advancing them to `as_of` yields a fixed answer). + // Keying on `as_of` rather than `until` is deliberate: `until` may be the empty + // (unbounded) frontier for a long-lived index, which a live `upper` never reaches. A + // premature emit at `upper == as_of` would ship an incomplete (often empty) snapshot + // and then close, so the read must wait for `upper` to pass `as_of`. If the publisher + // has closed we emit whatever has been sealed. + let sealed = closed || as_of.iter().all(|t| !upper.less_equal(t)); + if !sealed { + // Not yet sealed. Hold the capability and wait for the next publisher activation. + return; + } + + for batch in chain { + if !batch.is_empty() { + // Emit whole `Arc` batches under a capability at the minimum time, exactly as + // the live `import` seeds historical batches. The `BatchFrontier` wrapper + // advances the batches' times to `as_of` on read, so the emitted capability + // (minimum) is always at or below the presented times. + let cap = caps.delayed(&batch_min::()); + output + .session(&cap) + .give(BatchFrontier::make_from(batch, as_of.borrow(), until.borrow())); + } + } + + // Release the capability: the single-time snapshot is complete, so the output + // frontier advances to the empty antichain. + capabilities = None; + } + }); + + Arranged { stream, trace } + } } /// Deregisters an importer's replay queue when its source operator drops. From c6f92e966f8176f1baa4d22db528b4158f61bee3 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Wed, 22 Jul 2026 11:29:58 +0200 Subject: [PATCH 7/7] agent: expose TraceBox accumulated compaction frontiers Add public TraceBox::logical_compaction()/physical_compaction() accessors returning the accumulated MutableAntichain holds of all referees. This lets an external referee (a cross-runtime arrangement publisher living in a downstream crate) compute the frontier the trace would compact to without its own hold, which previously required reading the pub(crate) fields from inside this crate. Co-Authored-By: Claude Opus 4.8 --- differential-dataflow/src/operators/arrange/agent.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/differential-dataflow/src/operators/arrange/agent.rs b/differential-dataflow/src/operators/arrange/agent.rs index a310be8bb..f383b04e3 100644 --- a/differential-dataflow/src/operators/arrange/agent.rs +++ b/differential-dataflow/src/operators/arrange/agent.rs @@ -568,6 +568,15 @@ pub mod trace_box { /// /// This is used to inspect batches for purposes of resource accounting in external systems. pub fn trace(&self) -> &Tr { &self.trace } + /// The accumulated logical-compaction holds of all referees of the trace. + /// + /// This is the `MutableAntichain` whose frontier the trace compacts to. Exposed so an + /// external referee (for example a cross-runtime publisher) can compute the frontier the + /// trace would compact to without its own hold, by cloning this and subtracting that hold. + pub fn logical_compaction(&self) -> &MutableAntichain { &self.logical_compaction } + /// The accumulated physical-compaction holds of all referees of the trace. See + /// [`TraceBox::logical_compaction`]. + pub fn physical_compaction(&self) -> &MutableAntichain { &self.physical_compaction } /// Replaces elements of `lower` with those of `upper`. #[inline] pub fn adjust_logical_compaction(&mut self, lower: AntichainRef, upper: AntichainRef) {