diff --git a/differential-dataflow/examples/cursors.rs b/differential-dataflow/examples/cursors.rs index 4855ae7ef..8dcfc739d 100644 --- a/differential-dataflow/examples/cursors.rs +++ b/differential-dataflow/examples/cursors.rs @@ -34,11 +34,11 @@ use std::collections::BTreeMap; use timely::dataflow::operators::probe::Handle; -use timely::progress::frontier::AntichainRef; +use timely::progress::frontier::{Antichain, AntichainRef}; use timely::dataflow::operators::Probe; use differential_dataflow::input::Input; -use differential_dataflow::trace::cursor::Cursor; +use differential_dataflow::trace::cursor::{Cursor, cursor_list}; use differential_dataflow::trace::TraceReader; type Node = u32; @@ -93,7 +93,8 @@ fn main() { } /* Return trace content after the last round. */ - let (mut cursor, storage) = graph_trace.cursor(); + let batches = graph_trace.batches_through(Antichain::new().borrow()).unwrap(); + let (mut cursor, storage) = cursor_list(batches); cursor.to_vec(&storage, |k| k.clone(), |v| v.clone()) }) .unwrap().join(); diff --git a/differential-dataflow/examples/multitemporal.rs b/differential-dataflow/examples/multitemporal.rs index 63040c6f6..0850745c1 100644 --- a/differential-dataflow/examples/multitemporal.rs +++ b/differential-dataflow/examples/multitemporal.rs @@ -3,11 +3,12 @@ use std::io::BufRead; use timely::dataflow::ProbeHandle; use timely::dataflow::operators::vec::unordered_input::UnorderedInput; use timely::dataflow::operators::Probe; -use timely::progress::frontier::AntichainRef; +use timely::progress::frontier::{Antichain, AntichainRef}; use timely::PartialOrder; use differential_dataflow::AsCollection; use differential_dataflow::trace::{Cursor, TraceReader}; +use differential_dataflow::trace::cursor::cursor_list; use pair::Pair; @@ -98,7 +99,8 @@ fn main() { else { println!("Report at {:?}", query_time); // enumerate the contents of `trace` at `query_time`. - let (mut cursor, storage) = trace.cursor(); + let batches = trace.batches_through(Antichain::new().borrow()).unwrap(); + let (mut cursor, storage) = cursor_list(batches); while let Some(key) = cursor.get_key(&storage) { while let Some(_val) = cursor.get_val(&storage) { let mut sum = 0; diff --git a/differential-dataflow/src/operators/arrange/upsert.rs b/differential-dataflow/src/operators/arrange/upsert.rs index 33dcc4cc8..8d9d32a6c 100644 --- a/differential-dataflow/src/operators/arrange/upsert.rs +++ b/differential-dataflow/src/operators/arrange/upsert.rs @@ -231,7 +231,8 @@ where // Prepare a cursor to the existing arrangement, and a batch builder for // new stuff that we add. - let (mut trace_cursor, trace_storage) = reader_local.cursor(); + let batches = reader_local.batches_through(Antichain::new().borrow()).unwrap(); + let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(batches); let mut builder = Bu::new(); let mut key_con = as Cursor>::KeyContainer::with_capacity(1); for (key, mut list) in to_process { diff --git a/differential-dataflow/src/operators/count.rs b/differential-dataflow/src/operators/count.rs index 842e828de..9a9506afc 100644 --- a/differential-dataflow/src/operators/count.rs +++ b/differential-dataflow/src/operators/count.rs @@ -97,7 +97,7 @@ where let mut session = output.session(&capability); let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage); - let (mut trace_cursor, trace_storage) = trace.cursor_through(lower_limit.borrow()).unwrap(); + let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap()); while let Some(key) = batch_cursor.get_key(&batch_storage) { let mut count: Option> = None; diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs index d33f52b63..0d03d6b20 100644 --- a/differential-dataflow/src/operators/join.rs +++ b/differential-dataflow/src/operators/join.rs @@ -105,7 +105,7 @@ where // initial work for the two traces, and before the operator is constructed. // Acknowledged frontier for each input. - // These two are used exclusively to track batch boundaries on which we may want/need to call `cursor_through`. + // These two are used exclusively to track batch boundaries on which we may want/need to call `batches_through`. // They will drive our physical compaction of each trace, and we want to maintain at all times that each is beyond // the physical compaction frontier of their corresponding trace. // Should we ever *drop* a trace, these are 1. much harder to maintain correctly, but 2. no longer used. diff --git a/differential-dataflow/src/operators/threshold.rs b/differential-dataflow/src/operators/threshold.rs index 2909a0c53..d46906638 100644 --- a/differential-dataflow/src/operators/threshold.rs +++ b/differential-dataflow/src/operators/threshold.rs @@ -147,7 +147,7 @@ where let mut session = output.session(&capability); let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage); - let (mut trace_cursor, trace_storage) = trace.cursor_through(lower_limit.borrow()).unwrap(); + let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap()); while let Some(key) = batch_cursor.get_key(&batch_storage) { let mut count: Option> = None; diff --git a/differential-dataflow/src/trace/cursor/mod.rs b/differential-dataflow/src/trace/cursor/mod.rs index 283b0771e..4c893de22 100644 --- a/differential-dataflow/src/trace/cursor/mod.rs +++ b/differential-dataflow/src/trace/cursor/mod.rs @@ -19,9 +19,7 @@ use crate::trace::implementations::containers::BatchContainer; /// This is the entry point for accessing batch data through cursors, and the place that opinions /// about keys and values are introduced (via the `Cursor` associated type). Cut-and-merge assembly /// is the trace's concern: [`TraceReader::batches_through`](crate::trace::TraceReader::batches_through) -/// selects the batches and the defaulted -/// [`TraceReader::cursor_through`](crate::trace::TraceReader::cursor_through) builds a [`CursorList`] -/// over their per-batch cursors. +/// selects the batches, and [`cursor_list()`] merges their cursors. pub trait Navigable { /// The cursor type. @@ -53,8 +51,7 @@ pub type BatchTimeGat<'a, Tr> = as Cursor>::TimeGat<'a>; /// Assembles a merged cursor over a sequence of batches. /// /// The batches become the cursor's storage and are returned alongside the cursor; they must be kept -/// alive and handed to the cursor's navigation methods. This is the shared assembly behind -/// `TraceReader::cursor_through` and the per-round input cursors in `reduce` / `count` / `threshold`. +/// alive and handed to the cursor's navigation methods. pub fn cursor_list(batches: Vec) -> (CursorList, Vec) { let cursors = batches.iter().map(|batch| batch.cursor()).collect::>(); let cursor = CursorList::new(cursors, &batches); diff --git a/differential-dataflow/src/trace/implementations/spine_fueled.rs b/differential-dataflow/src/trace/implementations/spine_fueled.rs index f4e6ad036..a6fc7b904 100644 --- a/differential-dataflow/src/trace/implementations/spine_fueled.rs +++ b/differential-dataflow/src/trace/implementations/spine_fueled.rs @@ -177,7 +177,7 @@ impl TraceReader for Spine { let include_upper = PartialOrder::less_equal(&batch.upper().borrow(), &upper); if include_lower != include_upper && upper != batch.lower().borrow() { - panic!("`cursor_through`: `upper` straddles batch"); + panic!("`batches_through`: `upper` straddles batch"); } // include pending batches diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index c6f864f53..1b5439f87 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -22,7 +22,6 @@ use crate::logging::Logger; pub use self::cursor::Cursor; pub use self::cursor::Navigable; pub use self::cursor::{BatchCursor, BatchKey, BatchVal, BatchValOwn, BatchDiff, BatchDiffGat, BatchTimeGat}; -use self::cursor::CursorList; pub use self::description::Description; /// A type used to express how much effort a trace should exert even in the absence of updates. @@ -50,35 +49,12 @@ pub trait TraceReader { /// Acquires the non-empty sequence of batches covering updates at times not greater or equal to an /// element of `upper`. /// - /// This is the sole primitive each `TraceReader` must implement to expose its contents: the - /// `cursor` and `cursor_through` methods assemble a [`CursorList`] over these batches' cursors, - /// for which the returned `Vec` serves as storage (the cursor borrows from it). - /// /// This method is expected to work if called with an `upper` that (i) was an observed bound in batches from /// the trace, and (ii) the trace has not been advanced beyond `upper`. Practically, the implementation should /// be expected to look for a "clean cut" using `upper`, and if it finds such a cut can return the batches. This - /// should allow `upper` such as `&[]` as used by `self.cursor()`, though it is difficult to imagine other uses. + /// should allow `upper` such as `&[]`, used to acquire all batches, though it is difficult to imagine other uses. fn batches_through(&mut self, upper: AntichainRef) -> Option>; - /// Provides a cursor over updates contained in the trace. - fn cursor(&mut self) -> (CursorList<::Cursor>, Vec) where Self::Batch: Navigable { - if let Some(cursor) = self.cursor_through(Antichain::new().borrow()) { - cursor - } - else { - panic!("unable to acquire complete cursor for trace; is it closed?"); - } - } - - /// Acquires a cursor to the restriction of the collection's contents to updates at times not greater or - /// equal to an element of `upper`. - /// - /// The cursor is a [`CursorList`] that merges the cursors of the batches returned by - /// [`batches_through`](TraceReader::batches_through); see that method for the contract on `upper`. - fn cursor_through(&mut self, upper: AntichainRef) -> Option<(CursorList<::Cursor>, Vec)> where Self::Batch: Navigable { - Some(self::cursor::cursor_list(self.batches_through(upper)?)) - } - /// Advances the frontier that constrains logical compaction. /// /// Logical compaction is the ability of the trace to change the times of the updates it contains. @@ -123,16 +99,15 @@ pub trait TraceReader { /// Reports the physical compaction frontier. /// /// All batches containing updates beyond this frontier will not be merged with other batches. This allows - /// the caller to create a cursor through any frontier beyond the physical compaction frontier, with the - /// `cursor_through()` method. This functionality is primarily of interest to the `join` operator, and any + /// the caller to acquire the batches through any frontier beyond the physical compaction frontier, with the + /// `batches_through()` method. This functionality is primarily of interest to the `join` operator, and any /// other operators who need to take notice of the physical structure of update batches. fn get_physical_compaction(&mut self) -> AntichainRef<'_, Self::Time>; /// Maps logic across the non-empty sequence of batches in the trace. /// /// This is currently used only to extract historical data to prime late-starting operators who want to reproduce - /// the stream of batches moving past the trace. It could also be a fine basis for a default implementation of the - /// cursor methods, as they (by default) just move through batches accumulating cursors into a cursor list. + /// the stream of batches moving past the trace. fn map_batches(&self, f: F); /// Reads the upper frontier of committed times. @@ -165,11 +140,8 @@ pub trait TraceReader { /// An append-only collection of `(key, val, time, diff)` tuples. /// -/// The trace must pretend to look like a collection of `(Key, Val, Time, isize)` tuples, but is permitted -/// to introduce new types `KeyRef`, `ValRef`, and `TimeRef` which can be dereference to the types above. -/// -/// The trace must be constructable from, and navigable by the `Key`, `Val`, `Time` types, but does not need -/// to return them. +/// The trace itself is opinionated only about `Time`, which bounds its contents and drives its compaction. +/// Key, value, and diff opinions live on the batches' cursors, and are reached through [`Navigable`]. pub trait Trace : TraceReader { /// Allocates a new empty trace. @@ -215,9 +187,6 @@ pub trait Trace : TraceReader { pub trait BatchReader : Sized { /// The timestamp type of the batch's updates. - /// - /// A batch carries only time; navigating its contents is the separate, optional [`Navigable`] - /// capability, which an operator requests with a `Self: Navigable` bound when it needs a cursor. type Time: Timestamp + Lattice; /// The number of updates in the batch. diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs index d30e8ef2d..2b263113f 100644 --- a/differential-dataflow/tests/trace.rs +++ b/differential-dataflow/tests/trace.rs @@ -4,7 +4,7 @@ use timely::progress::{Antichain, frontier::AntichainRef}; use differential_dataflow::trace::implementations::{ValBatcher, ValBuilder, ValSpine}; use differential_dataflow::trace::{Trace, TraceReader, Batcher, Builder}; -use differential_dataflow::trace::cursor::Cursor; +use differential_dataflow::trace::cursor::{Cursor, cursor_list}; type IntegerTrace = ValSpine; type IntegerBuilder = ValBuilder; @@ -34,11 +34,11 @@ fn get_trace() -> ValSpine { fn test_trace() { let mut trace = get_trace(); - let (mut cursor1, storage1) = trace.cursor_through(AntichainRef::new(&[1])).unwrap(); + let (mut cursor1, storage1) = cursor_list(trace.batches_through(AntichainRef::new(&[1])).unwrap()); let vec_1 = cursor1.to_vec(&storage1, |k| k.clone(), |v| v.clone()); assert_eq!(vec_1, vec![((1, 2), vec![(0, 1)])]); - let (mut cursor2, storage2) = trace.cursor_through(AntichainRef::new(&[2])).unwrap(); + let (mut cursor2, storage2) = cursor_list(trace.batches_through(AntichainRef::new(&[2])).unwrap()); let vec_2 = cursor2.to_vec(&storage2, |k| k.clone(), |v| v.clone()); println!("--> {:?}", vec_2); assert_eq!(vec_2, vec![ @@ -46,14 +46,15 @@ fn test_trace() { ((2, 3), vec![(1, 1)]), ]); - let (mut cursor3, storage3) = trace.cursor_through(AntichainRef::new(&[3])).unwrap(); + let (mut cursor3, storage3) = cursor_list(trace.batches_through(AntichainRef::new(&[3])).unwrap()); let vec_3 = cursor3.to_vec(&storage3, |k| k.clone(), |v| v.clone()); assert_eq!(vec_3, vec![ ((1, 2), vec![(0, 1)]), ((2, 3), vec![(1, 1), (2, -1)]), ]); - let (mut cursor4, storage4) = trace.cursor(); + let batches = trace.batches_through(Antichain::new().borrow()).unwrap(); + let (mut cursor4, storage4) = cursor_list(batches); let vec_4 = cursor4.to_vec(&storage4, |k| k.clone(), |v| v.clone()); assert_eq!(vec_4, vec_3); } diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs index c5a9ae9b9..d3d9db914 100644 --- a/dogsdogsdogs/src/operators/half_join.rs +++ b/dogsdogsdogs/src/operators/half_join.rs @@ -49,6 +49,7 @@ use differential_dataflow::difference::{Monoid, Semigroup}; use differential_dataflow::lattice::Lattice; use differential_dataflow::operators::arrange::Arranged; use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchTimeGat, BatchVal, Cursor, Navigable, TraceReader}; +use differential_dataflow::trace::cursor::cursor_list; use differential_dataflow::consolidation::{consolidate, consolidate_updates}; use differential_dataflow::trace::implementations::BatchContainer; @@ -327,7 +328,8 @@ where // Sort requests by key for in-order cursor traversal. consolidate_updates(proposals); - let (mut cursor, storage) = trace.cursor(); + let batches = trace.batches_through(Antichain::new().borrow()).unwrap(); + let (mut cursor, storage) = cursor_list(batches); let mut yielded = false; let mut key_con = as Cursor>::KeyContainer::with_capacity(1); diff --git a/dogsdogsdogs/src/operators/half_join2.rs b/dogsdogsdogs/src/operators/half_join2.rs index e9deb66ce..102a0f108 100644 --- a/dogsdogsdogs/src/operators/half_join2.rs +++ b/dogsdogsdogs/src/operators/half_join2.rs @@ -34,6 +34,7 @@ use differential_dataflow::difference::{Monoid, Semigroup}; use differential_dataflow::lattice::Lattice; use differential_dataflow::operators::arrange::Arranged; use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchTimeGat, BatchVal, Cursor, Navigable, TraceReader}; +use differential_dataflow::trace::cursor::cursor_list; use differential_dataflow::consolidation::{consolidate, consolidate_updates}; use differential_dataflow::trace::implementations::BatchContainer; @@ -263,7 +264,8 @@ where let mut builders = (0..blob.caps.len()).map(|_| CB::default()).collect::>(); - let (mut cursor, storage) = trace.cursor(); + let batches = trace.batches_through(Antichain::new().borrow()).unwrap(); + let (mut cursor, storage) = cursor_list(batches); let mut key_con = as Cursor>::KeyContainer::with_capacity(1); let mut removals: ChangeBatch = ChangeBatch::new(); diff --git a/dogsdogsdogs/src/operators/lookup_map.rs b/dogsdogsdogs/src/operators/lookup_map.rs index 25cda232f..dc2a3f494 100644 --- a/dogsdogsdogs/src/operators/lookup_map.rs +++ b/dogsdogsdogs/src/operators/lookup_map.rs @@ -9,6 +9,7 @@ use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable} use differential_dataflow::difference::{IsZero, Semigroup, Monoid}; use differential_dataflow::operators::arrange::Arranged; use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, BatchVal, Cursor, Navigable, TraceReader}; +use differential_dataflow::trace::cursor::cursor_list; use differential_dataflow::trace::implementations::BatchContainer; /// Proposes extensions to a stream of prefixes. @@ -88,7 +89,8 @@ where key1.cmp(&key2) }); - let (mut cursor, storage) = trace.cursor(); + let batches = trace.batches_through(Antichain::new().borrow()).unwrap(); + let (mut cursor, storage) = cursor_list(batches); // Key container to stage keys for comparison. let mut key_con = as Cursor>::KeyContainer::with_capacity(1); for &mut (ref prefix, ref time, ref mut diff) in prefixes.iter_mut() {