Cursors migration

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();

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;

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 = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
                                 for (key, mut list) in to_process {

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<BatchDiff<Tr>> = None;

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.

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<BatchDiff<Tr>> = None;

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> = <BatchCursor<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<B: crate::trace::BatchReader + Navigable>(batches: Vec<B>) -> (CursorList<B::Cursor>, Vec<B>) {
     let cursors = batches.iter().map(|batch| batch.cursor()).collect::<Vec<_>>();
     let cursor = CursorList::new(cursors, &batches);

differential-dataflow/src/trace/implementations/spine_fueled.rs

@@ -177,7 +177,7 @@ impl<B: Batch+Clone+'static> TraceReader for Spine<B> {
                 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

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<Self::Time>) -> Option<Vec<Self::Batch>>;
 
-    /// Provides a cursor over updates contained in the trace.
-    fn cursor(&mut self) -> (CursorList<<Self::Batch as Navigable>::Cursor>, Vec<Self::Batch>) 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<Self::Time>) -> Option<(CursorList<<Self::Batch as Navigable>::Cursor>, Vec<Self::Batch>)> 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<F: FnMut(&Self::Batch)>(&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<Batch: Batch> {
 
     /// Allocates a new empty trace.
@@ -215,9 +187,6 @@ pub trait Trace : TraceReader<Batch: Batch> {
 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.

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<u64, u64, usize, i64>;
 type IntegerBuilder = ValBuilder<u64, u64, usize, i64>;
@@ -34,26 +34,27 @@ fn get_trace() -> ValSpine<u64, u64, usize, i64> {
 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![
                ((1, 2), vec![(0, 1)]),
                ((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);
 }

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 = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);

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::<Vec<_>>();
 
-                    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 = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
                     let mut removals: ChangeBatch<Tr::Time> = ChangeBatch::new();
 

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 = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
                     for &mut (ref prefix, ref time, ref mut diff) in prefixes.iter_mut() {