Move the wrapper cursors to the cursor module

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;
 

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<B, TInner> Navigable for BatchEnter<B, TInner>
+where
+    B: BatchReader + Navigable,
+    TInner: Refines<B::Time>+Lattice,
+    TInner: Refines<<B::Cursor as Cursor>::Time>,
+{
+    type Cursor = BatchCursorEnter<B::Cursor, TInner>;
+
+    fn cursor(&self) -> Self::Cursor {
+        BatchCursorEnter::new(self.inner().cursor())
+    }
+}
+
+/// Wrapper to provide cursor to nested scope.
+pub struct BatchCursorEnter<C, TInner> {
+    phantom: ::std::marker::PhantomData<TInner>,
+    cursor: C,
+}
+
+impl<C, TInner> BatchCursorEnter<C, TInner> {
+    fn new(cursor: C) -> Self {
+        BatchCursorEnter {
+            phantom: ::std::marker::PhantomData,
+            cursor,
+        }
+    }
+}
+
+impl<TInner, C: Cursor> Cursor for BatchCursorEnter<C, TInner>
+where
+    TInner: Refines<C::Time>+Lattice,
+{
+    type Storage = BatchEnter<C::Storage, TInner>;
+
+    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<TInner>;
+    type Time = <Vec<TInner> as BatchContainer>::Owned;
+    type TimeGat<'a> = <Vec<TInner> 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::Key<'a>> { self.cursor.get_key(storage.inner()) }
+    #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(storage.inner()) }
+
+    #[inline]
+    fn map_times<L: FnMut(&TInner, Self::DiffGat<'_>)>(&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()) }
+}

differential-dataflow/src/trace/cursor/wrappers/frontier.rs

@@ -0,0 +1,74 @@
+//! Cursor for the `frontier` batch wrapper.
+
+use crate::trace::implementations::BatchContainer;
+use crate::trace::wrappers::frontier::BatchFrontier;
+use crate::trace::{BatchReader, Navigable};
+use crate::trace::cursor::Cursor;
+
+impl<B> Navigable for BatchFrontier<B>
+where
+    B: BatchReader + Navigable,
+    B::Cursor: Cursor<Time = B::Time>,
+{
+    type Cursor = BatchCursorFrontier<B::Cursor>;
+
+    fn cursor(&self) -> Self::Cursor {
+        BatchCursorFrontier { cursor: self.inner().cursor() }
+    }
+}
+
+/// Wrapper to provide cursor to nested scope.
+///
+/// The wrapper's `since` and `until` frontiers stay with the batch, which applies them through
+/// [`BatchFrontier::advance_time`]; the cursor holds only the cursor it forwards to.
+pub struct BatchCursorFrontier<C> {
+    cursor: C,
+}
+
+impl<C> Cursor for BatchCursorFrontier<C>
+where
+    C: Cursor<Storage: BatchReader<Time = C::Time>>,
+{
+    type Storage = BatchFrontier<C::Storage>;
+
+    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<C::Time>;
+    type Time = <Vec<C::Time> as BatchContainer>::Owned;
+    type TimeGat<'a> = <Vec<C::Time> 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::Key<'a>> { self.cursor.get_key(storage.inner()) }
+    #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(storage.inner()) }
+
+    #[inline]
+    fn map_times<L: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, mut logic: L) {
+        let mut temp: C::Time = <C::Time as timely::progress::Timestamp>::minimum();
+        self.cursor.map_times(storage.inner(), |time, diff| {
+            C::clone_time_onto(time, &mut temp);
+            if storage.advance_time(&mut temp) {
+                logic(&temp, 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()) }
+}

differential-dataflow/src/trace/cursor/wrappers/mod.rs

@@ -0,0 +1,10 @@
+//! 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;
+pub mod frontier;

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<Tr: TraceReader, TInner> {
@@ -102,19 +101,6 @@ pub struct BatchEnter<B, TInner> {
     description: Description<TInner>,
 }
 
-impl<B, TInner> Navigable for BatchEnter<B, TInner>
-where
-    B: BatchReader + Navigable,
-    TInner: Refines<B::Time>+Lattice,
-    TInner: Refines<<B::Cursor as Cursor>::Time>,
-{
-    type Cursor = BatchCursorEnter<B::Cursor, TInner>;
-
-    fn cursor(&self) -> Self::Cursor {
-        BatchCursorEnter::new(self.batch.cursor())
-    }
-}
-
 impl<B, TInner> BatchReader for BatchEnter<B, TInner>
 where
     B: BatchReader,
@@ -125,6 +111,15 @@ where
     fn description(&self) -> &Description<TInner> { &self.description }
 }
 
+impl<B, TInner> BatchEnter<B, TInner> {
+    /// 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<B, TInner> BatchEnter<B, TInner>
 where
     B: BatchReader,
@@ -142,64 +137,3 @@ where
         }
     }
 }
-
-use crate::trace::implementations::BatchContainer;
-
-/// Wrapper to provide cursor to nested scope.
-pub struct BatchCursorEnter<C, TInner> {
-    phantom: ::std::marker::PhantomData<TInner>,
-    cursor: C,
-}
-
-impl<C, TInner> BatchCursorEnter<C, TInner> {
-    fn new(cursor: C) -> Self {
-        BatchCursorEnter {
-            phantom: ::std::marker::PhantomData,
-            cursor,
-        }
-    }
-}
-
-impl<TInner, C: Cursor> Cursor for BatchCursorEnter<C, TInner>
-where
-    TInner: Refines<C::Time>+Lattice,
-{
-    type Storage = BatchEnter<C::Storage, TInner>;
-
-    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<TInner>;
-    type Time = <Vec<TInner> as BatchContainer>::Owned;
-    type TimeGat<'a> = <Vec<TInner> 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::Key<'a>> { self.cursor.get_key(&storage.batch) }
-    #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(&storage.batch) }
-
-    #[inline]
-    fn map_times<L: FnMut(&TInner, Self::DiffGat<'_>)>(&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) }
-}