From 96579d26d6bc75cc8ea6447d812e51311c938b29 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 7 Aug 2026 17:17:49 -0400 Subject: [PATCH 1/4] Move the enter wrapper's cursor to the cursor module The `enter` wrapper had two halves. One presents a wrapped trace and batch, rewriting descriptions and compaction frontiers through `Refines`; it names no keys and no values. The other navigates the wrapped batch with a cursor. This moves the second half to `trace/cursor/wrappers/enter.rs`, leaving `trace/wrappers/enter.rs` free of any cursor mention. A reader who declines cursors keeps the wrapper and supplies its own module beside this one. The cursor reaches the wrapped batch through a new `inner()` accessor. It sits in its own unbounded impl block because the `Cursor` impl does not know that the wrapped storage is a `BatchReader`. Co-Authored-By: Claude Opus 5 (1M context) --- differential-dataflow/src/trace/cursor/mod.rs | 1 + .../src/trace/cursor/wrappers/enter.rs | 81 +++++++++++++++++ .../src/trace/cursor/wrappers/mod.rs | 9 ++ .../src/trace/wrappers/enter.rs | 86 +++---------------- 4 files changed, 101 insertions(+), 76 deletions(-) create mode 100644 differential-dataflow/src/trace/cursor/wrappers/enter.rs create mode 100644 differential-dataflow/src/trace/cursor/wrappers/mod.rs diff --git a/differential-dataflow/src/trace/cursor/mod.rs b/differential-dataflow/src/trace/cursor/mod.rs index 4c893de22..2b563ea37 100644 --- a/differential-dataflow/src/trace/cursor/mod.rs +++ b/differential-dataflow/src/trace/cursor/mod.rs @@ -6,6 +6,7 @@ //! supports efficient seeking (via the `seek_key` and `seek_val` methods). pub mod cursor_list; +pub mod wrappers; pub use self::cursor_list::CursorList; diff --git a/differential-dataflow/src/trace/cursor/wrappers/enter.rs b/differential-dataflow/src/trace/cursor/wrappers/enter.rs new file mode 100644 index 000000000..dec50e2bc --- /dev/null +++ b/differential-dataflow/src/trace/cursor/wrappers/enter.rs @@ -0,0 +1,81 @@ +//! Cursor for the `enter` batch wrapper. + +use timely::progress::timestamp::Refines; + +use crate::lattice::Lattice; +use crate::trace::implementations::BatchContainer; +use crate::trace::wrappers::enter::BatchEnter; +use crate::trace::{BatchReader, Navigable}; +use crate::trace::cursor::Cursor; + +impl Navigable for BatchEnter +where + B: BatchReader + Navigable, + TInner: Refines+Lattice, + TInner: Refines<::Time>, +{ + type Cursor = BatchCursorEnter; + + fn cursor(&self) -> Self::Cursor { + BatchCursorEnter::new(self.inner().cursor()) + } +} + +/// Wrapper to provide cursor to nested scope. +pub struct BatchCursorEnter { + phantom: ::std::marker::PhantomData, + cursor: C, +} + +impl BatchCursorEnter { + fn new(cursor: C) -> Self { + BatchCursorEnter { + phantom: ::std::marker::PhantomData, + cursor, + } + } +} + +impl Cursor for BatchCursorEnter +where + TInner: Refines+Lattice, +{ + type Storage = BatchEnter; + + type Key<'a> = C::Key<'a>; + type ValOwn = C::ValOwn; + type Val<'a> = C::Val<'a>; + type KeyContainer = C::KeyContainer; + type ValContainer = C::ValContainer; + type DiffContainer = C::DiffContainer; + type Diff = C::Diff; + type DiffGat<'a> = C::DiffGat<'a>; + type TimeContainer = Vec; + type Time = as BatchContainer>::Owned; + type TimeGat<'a> = as BatchContainer>::ReadItem<'a>; + + #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage.inner()) } + #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage.inner()) } + + #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage.inner()) } + #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage.inner()) } + + #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_key(storage.inner()) } + #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_val(storage.inner()) } + + #[inline] + fn map_times)>(&mut self, storage: &Self::Storage, mut logic: L) { + self.cursor.map_times(storage.inner(), |time, diff| { + logic(&TInner::to_inner(C::owned_time(time)), diff) + }) + } + + #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage.inner()) } + #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage.inner(), key) } + + #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage.inner()) } + #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage.inner(), val) } + + #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage.inner()) } + #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage.inner()) } +} diff --git a/differential-dataflow/src/trace/cursor/wrappers/mod.rs b/differential-dataflow/src/trace/cursor/wrappers/mod.rs new file mode 100644 index 000000000..c77543d68 --- /dev/null +++ b/differential-dataflow/src/trace/cursor/wrappers/mod.rs @@ -0,0 +1,9 @@ +//! Cursor implementations for the trace wrappers. +//! +//! The wrappers themselves live in [`crate::trace::wrappers`] and are cursor-free: they present +//! wrapped traces and batches, and expose their time semantics over owned times. This module +//! supplies one way to read through those wrappers, by forwarding to the cursor of the wrapped +//! batch and applying the wrapper's time rule to each time it produces. Another read strategy +//! would supply its own module here, and reuse the same time rules. + +pub mod enter; diff --git a/differential-dataflow/src/trace/wrappers/enter.rs b/differential-dataflow/src/trace/wrappers/enter.rs index f63f44db6..b168ab6e6 100644 --- a/differential-dataflow/src/trace/wrappers/enter.rs +++ b/differential-dataflow/src/trace/wrappers/enter.rs @@ -5,8 +5,7 @@ use timely::progress::timestamp::Refines; use timely::progress::{Antichain, frontier::AntichainRef}; use crate::lattice::Lattice; -use crate::trace::{BatchReader, Description, Navigable, TraceReader}; -use crate::trace::cursor::Cursor; +use crate::trace::{BatchReader, Description, TraceReader}; /// Wrapper to provide trace to nested scope. pub struct TraceEnter { @@ -102,19 +101,6 @@ pub struct BatchEnter { description: Description, } -impl Navigable for BatchEnter -where - B: BatchReader + Navigable, - TInner: Refines+Lattice, - TInner: Refines<::Time>, -{ - type Cursor = BatchCursorEnter; - - fn cursor(&self) -> Self::Cursor { - BatchCursorEnter::new(self.batch.cursor()) - } -} - impl BatchReader for BatchEnter where B: BatchReader, @@ -125,6 +111,15 @@ where fn description(&self) -> &Description { &self.description } } +impl BatchEnter { + /// The wrapped batch, whose times are those of the containing scope. + /// + /// Each of its times enters the nested scope as `TInner::to_inner(time)`; that rule is the + /// whole of the wrapper's read-side semantics, and any reader of the wrapped batch must + /// apply it. + pub fn inner(&self) -> &B { &self.batch } +} + impl BatchEnter where B: BatchReader, @@ -142,64 +137,3 @@ where } } } - -use crate::trace::implementations::BatchContainer; - -/// Wrapper to provide cursor to nested scope. -pub struct BatchCursorEnter { - phantom: ::std::marker::PhantomData, - cursor: C, -} - -impl BatchCursorEnter { - fn new(cursor: C) -> Self { - BatchCursorEnter { - phantom: ::std::marker::PhantomData, - cursor, - } - } -} - -impl Cursor for BatchCursorEnter -where - TInner: Refines+Lattice, -{ - type Storage = BatchEnter; - - type Key<'a> = C::Key<'a>; - type ValOwn = C::ValOwn; - type Val<'a> = C::Val<'a>; - type KeyContainer = C::KeyContainer; - type ValContainer = C::ValContainer; - type DiffContainer = C::DiffContainer; - type Diff = C::Diff; - type DiffGat<'a> = C::DiffGat<'a>; - type TimeContainer = Vec; - type Time = as BatchContainer>::Owned; - type TimeGat<'a> = as BatchContainer>::ReadItem<'a>; - - #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(&storage.batch) } - #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(&storage.batch) } - - #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(&storage.batch) } - #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(&storage.batch) } - - #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_key(&storage.batch) } - #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_val(&storage.batch) } - - #[inline] - fn map_times)>(&mut self, storage: &Self::Storage, mut logic: L) { - self.cursor.map_times(&storage.batch, |time, diff| { - logic(&TInner::to_inner(C::owned_time(time)), diff) - }) - } - - #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(&storage.batch) } - #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(&storage.batch, key) } - - #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(&storage.batch) } - #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(&storage.batch, val) } - - #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(&storage.batch) } - #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(&storage.batch) } -} From 814670abf3122e3738887f79bda102e5439b4a1e Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 7 Aug 2026 17:23:28 -0400 Subject: [PATCH 2/4] Move the frontier wrapper's cursor to the cursor module As with `enter`, the wrapper splits into a half that presents a wrapped trace and batch and a half that navigates it. The second half moves to `trace/cursor/wrappers/frontier.rs`. The time semantics stay with the batch, as `advance_time`, which advances a time by `since` and reports whether `until` suppresses it. Any reader of the wrapped batch applies that rule, whether or not it reads through a cursor. The cursor no longer keeps its own copies of the two frontiers, so it holds nothing but the cursor it forwards to. Its bound strengthens to `Storage: BatchReader