Encourage Cursor::{get_key, get_val}; Update CursorList

Author: frankmcsherry

differential-dataflow/src/operators/join.rs

@@ -661,11 +661,12 @@ where
 
         let mut thinker = JoinThinker::new();
 
-        while batch.key_valid(batch_storage) && trace.key_valid(trace_storage) && effort < *fuel {
+        while let (Some(batch_key), Some(trace_key), true) = (batch.get_key(batch_storage), trace.get_key(trace_storage), effort < *fuel) {
+        // while batch.key_valid(batch_storage) && trace.key_valid(trace_storage) && effort < *fuel {
 
-            match trace.key(trace_storage).cmp(&batch.key(batch_storage)) {
-                Ordering::Less => trace.seek_key(trace_storage, batch.key(batch_storage)),
-                Ordering::Greater => batch.seek_key(batch_storage, trace.key(trace_storage)),
+            match trace_key.cmp(&batch_key) {
+                Ordering::Less => trace.seek_key(trace_storage, batch_key),
+                Ordering::Greater => batch.seek_key(batch_storage, trace_key),
                 Ordering::Equal => {
 
                     use crate::IntoOwned;
@@ -679,7 +680,7 @@ where
 
                     // populate `temp` with the results in the best way we know how.
                     thinker.think(|v1,v2,t,r1,r2| {
-                        let key = batch.key(batch_storage);
+                        let key = batch_key;
                         logic(key, v1, v2, &t, r1, r2, &mut session);
                     });
 

differential-dataflow/src/operators/mod.rs

@@ -45,9 +45,9 @@ impl<'a, C: Cursor> EditList<'a, C> {
         L: Fn(C::TimeGat<'_>)->C::Time,
     {
         self.clear();
-        while cursor.val_valid(storage) {
+        while let Some(val) = cursor.get_val(storage) {
             cursor.map_times(storage, |time1, diff1| self.push(logic(time1), diff1.into_owned()));
-            self.seal(cursor.val(storage));
+            self.seal(val);
             cursor.step_val(storage);
         }
     }

differential-dataflow/src/trace/cursor/cursor_list.rs

@@ -26,58 +26,68 @@ impl<C: Cursor> CursorList<C> {
         result
     }
 
-    // Initialize min_key with the indices of cursors with the minimum key.
-    //
-    // This method scans the current keys of each cursor, and tracks the indices
-    // of cursors whose key equals the minimum valid key seen so far. As it goes,
-    // if it observes an improved key it clears the current list, updates the
-    // minimum key, and continues.
-    //
-    // Once finished, it invokes `minimize_vals()` to ensure the value cursor is
-    // in a consistent state as well.
+    /// Initialize min_key with the indices of cursors with the minimum key.
+    ///
+    /// This method scans the current keys of each cursor, and tracks the indices
+    /// of cursors whose key equals the minimum valid key seen so far. As it goes,
+    /// if it observes an improved key it clears the current list, updates the
+    /// minimum key, and continues.
+    ///
+    /// Once finished, it invokes `minimize_vals()` to ensure the value cursor is
+    /// in a consistent state as well.
     fn minimize_keys(&mut self, storage: &[C::Storage]) {
 
         self.min_key.clear();
 
-        // Determine the index of the cursor with minimum key.
-        let mut min_key_opt = None;
-        for (index, cursor) in self.cursors.iter().enumerate() {
-            let key = cursor.get_key(&storage[index]);
-            if key.is_some() {
-                if min_key_opt.is_none() || key.lt(&min_key_opt) {
-                    min_key_opt = key;
-                    self.min_key.clear();
-                }
-                if key.eq(&min_key_opt) {
-                    self.min_key.push(index);
+        // We'll visit each non-`None` key, maintaining the indexes of the least keys in `self.min_key`.
+        let mut iter = self.cursors.iter().enumerate().flat_map(|(idx, cur)| cur.get_key(&storage[idx]).map(|key| (idx, key)));
+        if let Some((idx, key)) = iter.next() {
+            let mut min_key = key;
+            self.min_key.push(idx);
+            for (idx, key) in iter {
+                match key.cmp(&min_key) {
+                    std::cmp::Ordering::Less => {
+                        self.min_key.clear();
+                        self.min_key.push(idx);
+                        min_key = key;
+                    }
+                    std::cmp::Ordering::Equal => {
+                        self.min_key.push(idx);
+                    }
+                    std::cmp::Ordering::Greater => { }
                 }
             }
         }
 
         self.minimize_vals(storage);
     }
 
-    // Initialize min_val with the indices of minimum key cursors with the minimum value.
-    //
-    // This method scans the current values of cursor with minimum keys, and tracks the
-    // indices of cursors whose value equals the minimum valid value seen so far. As it
-    // goes, if it observes an improved value it clears the current list, updates the minimum
-    // value, and continues.
+    /// Initialize min_val with the indices of minimum key cursors with the minimum value.
+    ///
+    /// This method scans the current values of cursor with minimum keys, and tracks the
+    /// indices of cursors whose value equals the minimum valid value seen so far. As it
+    /// goes, if it observes an improved value it clears the current list, updates the minimum
+    /// value, and continues.
     fn minimize_vals(&mut self, storage: &[C::Storage]) {
 
         self.min_val.clear();
 
-        // Determine the index of the cursor with minimum value.
-        let mut min_val = None;
-        for &index in self.min_key.iter() {
-            let val = self.cursors[index].get_val(&storage[index]);
-            if val.is_some() {
-                if min_val.is_none() || val.lt(&min_val) {
-                    min_val = val;
-                    self.min_val.clear();
-                }
-                if val.eq(&min_val) {
-                    self.min_val.push(index);
+        // We'll visit each non-`None` value, maintaining the indexes of the least values in `self.min_val`.
+        let mut iter = self.min_key.iter().cloned().flat_map(|idx| self.cursors[idx].get_val(&storage[idx]).map(|val| (idx, val)));
+        if let Some((idx, val)) = iter.next() {
+            let mut min_val = val;
+            self.min_val.push(idx);
+            for (idx, val) in iter {
+                match val.cmp(&min_val) {
+                    std::cmp::Ordering::Less => {
+                        self.min_val.clear();
+                        self.min_val.push(idx);
+                        min_val = val;
+                    }
+                    std::cmp::Ordering::Equal => {
+                        self.min_val.push(idx);
+                    }
+                    std::cmp::Ordering::Greater => { }
                 }
             }
         }

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

@@ -81,14 +81,14 @@ pub trait Cursor {
     {
         let mut out = Vec::new();
         self.rewind_keys(storage);
-        while self.key_valid(storage) {
+        while let Some(key) = self.get_key(storage) {
             self.rewind_vals(storage);
-            while self.val_valid(storage) {
+            while let Some(val) = self.get_val(storage) {
                 let mut kv_out = Vec::new();
                 self.map_times(storage, |ts, r| {
                     kv_out.push((ts.into_owned(), r.into_owned()));
                 });
-                out.push(((self.key(storage).into_owned(), self.val(storage).into_owned()), kv_out));
+                out.push(((key.into_owned(), val.into_owned()), kv_out));
                 self.step_val(storage);
             }
             self.step_key(storage);