diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index 3aab9716d..22d2efc99 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -30,11 +30,9 @@ use timely::dataflow::operators::Capability; use crate::{Data, VecCollection, AsCollection}; use crate::difference::Semigroup; use crate::lattice::Lattice; -use crate::trace::{self, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchTimeGat, BatchVal, BatchValOwn}; +use crate::trace::{self, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchVal, BatchValOwn}; use trace::wrappers::enter::{TraceEnter, BatchEnter,}; -use trace::wrappers::enter_at::TraceEnter as TraceEnterAt; -use trace::wrappers::enter_at::BatchEnter as BatchEnterAt; use super::TraceAgent; @@ -92,26 +90,6 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { } } - /// Brings an arranged collection into a nested scope. - /// - /// This method produces a proxy trace handle that uses the same backing data, but acts as if the timestamps - /// have all been extended with an additional coordinate with the default value. The resulting collection does - /// not vary with the new timestamp coordinate. - pub fn enter_at<'inner, TInner, F, P>(self, child: Scope<'inner, TInner>, logic: F, prior: P) -> Arranged<'inner, TraceEnterAt> - where - Tr::Batch: Navigable, - TInner: Refines+Lattice+'static, - F: FnMut(BatchKey<'_, Tr>, BatchVal<'_, Tr>, BatchTimeGat<'_, Tr>)->TInner+Clone+'static, - P: FnMut(&TInner)->Tr::Time+Clone+'static, - { - let logic1 = logic.clone(); - let logic2 = logic.clone(); - Arranged { - trace: TraceEnterAt::make_from(self.trace, logic1, prior), - stream: self.stream.enter(child).map(move |bw| BatchEnterAt::make_from(bw, logic2.clone())), - } - } - /// Extracts a collection of any container from the stream of batches. /// /// This method is like `self.stream.flat_map`, except that it produces containers diff --git a/differential-dataflow/src/trace/wrappers/enter_at.rs b/differential-dataflow/src/trace/wrappers/enter_at.rs deleted file mode 100644 index 777f6e084..000000000 --- a/differential-dataflow/src/trace/wrappers/enter_at.rs +++ /dev/null @@ -1,237 +0,0 @@ -//! Wrappers to provide trace access to nested scopes. - -use timely::progress::timestamp::Refines; -use timely::progress::{Antichain, frontier::AntichainRef}; - -use crate::lattice::Lattice; -use crate::trace::{BatchKey, BatchReader, BatchTimeGat, BatchVal, Description, Navigable, TraceReader}; -use crate::trace::cursor::Cursor; - -/// Wrapper to provide trace to nested scope. -/// -/// Each wrapped update is presented with a timestamp determined by `logic`. -/// -/// At the same time, we require a method `prior` that can "invert" timestamps, -/// and which will be applied to compaction frontiers as they are communicated -/// back to the wrapped traces. A better explanation is pending, and until that -/// happens use this construct at your own peril! -pub struct TraceEnter { - trace: Tr, - stash1: Antichain, - stash2: Antichain, - logic: F, - prior: G, -} - -impl Clone for TraceEnter -where - Tr: TraceReader+Clone, - F: Clone, - G: Clone, -{ - fn clone(&self) -> Self { - TraceEnter { - trace: self.trace.clone(), - stash1: Antichain::new(), - stash2: Antichain::new(), - logic: self.logic.clone(), - prior: self.prior.clone(), - } - } -} - -impl TraceReader for TraceEnter -where - Tr: TraceReader, - TInner: Refines+Lattice, - F: 'static, - F: FnMut(BatchKey<'_, Tr>, BatchVal<'_, Tr>, BatchTimeGat<'_, Tr>)->TInner+Clone, - G: FnMut(&TInner)->Tr::Time+Clone+'static, -{ - type Time = TInner; - type Batch = BatchEnter; - - fn map_batches(&self, mut f: F2) { - let logic = self.logic.clone(); - self.trace.map_batches(|batch| { - f(&Self::Batch::make_from(batch.clone(), logic.clone())); - }) - } - - fn set_logical_compaction(&mut self, frontier: AntichainRef<'_, TInner>) { - self.stash1.clear(); - for time in frontier.iter() { - self.stash1.insert((self.prior)(time)); - } - self.trace.set_logical_compaction(self.stash1.borrow()); - } - fn get_logical_compaction(&mut self) -> AntichainRef<'_, TInner> { - self.stash2.clear(); - for time in self.trace.get_logical_compaction().iter() { - self.stash2.insert(TInner::to_inner(time.clone())); - } - self.stash2.borrow() - } - - fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, TInner>) { - self.stash1.clear(); - for time in frontier.iter() { - self.stash1.insert((self.prior)(time)); - } - self.trace.set_physical_compaction(self.stash1.borrow()); - } - fn get_physical_compaction(&mut self) -> AntichainRef<'_, TInner> { - self.stash2.clear(); - for time in self.trace.get_physical_compaction().iter() { - self.stash2.insert(TInner::to_inner(time.clone())); - } - self.stash2.borrow() - } - - fn batches_through(&mut self, upper: AntichainRef) -> Option> { - self.stash1.clear(); - for time in upper.iter() { - self.stash1.insert(time.clone().to_outer()); - } - let logic = self.logic.clone(); - let storage = self.trace.batches_through(self.stash1.borrow())?; - Some(storage.into_iter().map(|batch| BatchEnter::make_from(batch, logic.clone())).collect()) - } -} - -impl TraceEnter -where - Tr: TraceReader, - TInner: Refines+Lattice, -{ - /// Makes a new trace wrapper - pub fn make_from(trace: Tr, logic: F, prior: G) -> Self { - TraceEnter { - trace, - stash1: Antichain::new(), - stash2: Antichain::new(), - logic, - prior, - } - } -} - - -/// Wrapper to provide batch to nested scope. -#[derive(Clone)] -pub struct BatchEnter { - batch: B, - description: Description, - logic: F, -} - -impl Navigable for BatchEnter -where - B: BatchReader + Navigable, - TInner: Refines+Lattice, - TInner: Refines<::Time>, - F: FnMut(::Key<'_>, ::Val<'_>, ::TimeGat<'_>)->TInner+Clone, -{ - type Cursor = BatchCursorEnter; - - fn cursor(&self) -> Self::Cursor { - BatchCursorEnter::new(self.batch.cursor(), self.logic.clone()) - } -} - -impl BatchReader for BatchEnter -where - B: BatchReader + Navigable, - TInner: Refines+Lattice, - F: FnMut(::Key<'_>, ::Val<'_>, ::TimeGat<'_>)->TInner+Clone, -{ - type Time = TInner; - fn len(&self) -> usize { self.batch.len() } - fn description(&self) -> &Description { &self.description } -} - -impl BatchEnter -where - B: BatchReader, - TInner: Refines+Lattice, -{ - /// Makes a new batch wrapper - pub fn make_from(batch: B, logic: F) -> Self { - let lower: Vec<_> = batch.description().lower().elements().iter().map(|x| TInner::to_inner(x.clone())).collect(); - let upper: Vec<_> = batch.description().upper().elements().iter().map(|x| TInner::to_inner(x.clone())).collect(); - let since: Vec<_> = batch.description().since().elements().iter().map(|x| TInner::to_inner(x.clone())).collect(); - - BatchEnter { - batch, - description: Description::new(Antichain::from(lower), Antichain::from(upper), Antichain::from(since)), - logic, - } - } -} - -use crate::trace::implementations::BatchContainer; - -/// Wrapper to provide cursor to nested scope. -pub struct BatchCursorEnter { - phantom: ::std::marker::PhantomData, - cursor: C, - logic: F, -} - -impl BatchCursorEnter { - fn new(cursor: C, logic: F) -> Self { - BatchCursorEnter { - phantom: ::std::marker::PhantomData, - cursor, - logic, - } - } -} - -impl Cursor for BatchCursorEnter -where - TInner: Refines+Lattice, - F: FnMut(C::Key<'_>, C::Val<'_>, C::TimeGat<'_>)->TInner, -{ - 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) { - let key = self.key(storage); - let val = self.val(storage); - let logic2 = &mut self.logic; - self.cursor.map_times(&storage.batch, |time, diff| { - logic(&logic2(key, val, 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) } -} diff --git a/differential-dataflow/src/trace/wrappers/mod.rs b/differential-dataflow/src/trace/wrappers/mod.rs index ddebc66fb..71157cee2 100644 --- a/differential-dataflow/src/trace/wrappers/mod.rs +++ b/differential-dataflow/src/trace/wrappers/mod.rs @@ -1,5 +1,4 @@ //! Wrappers around trace implementations, providing derived views of updates. pub mod enter; -pub mod enter_at; pub mod frontier;