Remove use of borrow_as()

Author: frankmcsherry

differential-dataflow/examples/columnar.rs

@@ -638,6 +638,9 @@ pub mod dd_builder {
             // Owned key and val would need to be members of `self`, as this method can be called multiple times,
             // and we need to correctly cache last for reasons of correctness, not just performance.
 
+            let mut key_con = L::KeyContainer::with_capacity(1);
+            let mut val_con = L::ValContainer::with_capacity(1);
+
             for ((key,val),time,diff) in chunk.drain() {
                 // It would be great to avoid.
                 let key  = <layout::Key<L> as Columnar>::into_owned(key);
@@ -646,16 +649,19 @@ pub mod dd_builder {
                 let time = <layout::Time<L> as Columnar>::into_owned(time);
                 let diff = <layout::Diff<L> as Columnar>::into_owned(diff);
 
+                key_con.clear(); key_con.push_own(&key);
+                val_con.clear(); val_con.push_own(&val);
+
                 // Pre-load the first update.
                 if self.result.keys.is_empty() {
                     self.result.vals.vals.push_own(&val);
                     self.result.keys.push_own(&key);
                     self.staging.push(time, diff);
                 }
                 // Perhaps this is a continuation of an already received key.
-                else if self.result.keys.last().map(|k| L::KeyContainer::borrow_as(&key).eq(&k)).unwrap_or(false) {
+                else if self.result.keys.last() == key_con.get(0) {
                     // Perhaps this is a continuation of an already received value.
-                    if self.result.vals.vals.last().map(|v| L::ValContainer::borrow_as(&val).eq(&v)).unwrap_or(false) {
+                    if self.result.vals.vals.last() == val_con.get(0) {
                         self.staging.push(time, diff);
                     } else {
                         // New value; complete representation of prior value.
@@ -739,20 +745,24 @@ pub mod dd_builder {
             // Owned key and val would need to be members of `self`, as this method can be called multiple times,
             // and we need to correctly cache last for reasons of correctness, not just performance.
 
+            let mut key_con = L::KeyContainer::with_capacity(1);
+
             for ((key,_val),time,diff) in chunk.drain() {
                 // It would be great to avoid.
                 let key  = <layout::Key<L> as Columnar>::into_owned(key);
                 // These feel fine (wrt the other versions)
                 let time = <layout::Time<L> as Columnar>::into_owned(time);
                 let diff = <layout::Diff<L> as Columnar>::into_owned(diff);
 
+                key_con.clear(); key_con.push_own(&key);
+
                 // Pre-load the first update.
                 if self.result.keys.is_empty() {
                     self.result.keys.push_own(&key);
                     self.staging.push(time, diff);
                 }
                 // Perhaps this is a continuation of an already received key.
-                else if self.result.keys.last().map(|k| L::KeyContainer::borrow_as(&key).eq(&k)).unwrap_or(false) {
+                else if self.result.keys.last() == key_con.get(0) {
                     self.staging.push(time, diff);
                 } else {
                     // New key; complete representation of prior key.

differential-dataflow/src/operators/arrange/upsert.rs

@@ -233,14 +233,17 @@ where
                                 // new stuff that we add.
                                 let (mut trace_cursor, trace_storage) = reader_local.cursor();
                                 let mut builder = Bu::new();
+                                let mut key_con = Tr::KeyContainer::with_capacity(1);
                                 for (key, mut list) in to_process {
 
+                                    key_con.clear(); key_con.push_own(&key);
+
                                     // The prior value associated with the key.
                                     let mut prev_value: Option<Tr::ValOwn> = None;
 
                                     // Attempt to find the key in the trace.
-                                    trace_cursor.seek_key(&trace_storage, Tr::KeyContainer::borrow_as(&key));
-                                    if trace_cursor.get_key(&trace_storage).map(|k| k.eq(&Tr::KeyContainer::borrow_as(&key))).unwrap_or(false) {
+                                    trace_cursor.seek_key(&trace_storage, key_con.index(0));
+                                    if trace_cursor.get_key(&trace_storage).map(|k| k.eq(&key_con.index(0))).unwrap_or(false) {
                                         // Determine the prior value associated with the key.
                                         while let Some(val) = trace_cursor.get_val(&trace_storage) {
                                             let mut count = 0;

differential-dataflow/src/operators/reduce.rs

@@ -431,11 +431,16 @@ where
                         // We first extract those times from this list that lie in the interval we will process.
                         sort_dedup(&mut interesting);
                         // `exposed` contains interesting (key, time)s now below `upper_limit`
-                        let exposed = {
-                            let (exposed, new_interesting) = interesting.drain(..).partition(|(_, time)| !upper_limit.less_equal(time));
-                            interesting = new_interesting;
-                            exposed
-                        };
+                        let mut exposed_keys = T1::KeyContainer::with_capacity(0);
+                        let mut exposed_time = T1::TimeContainer::with_capacity(0);
+                        // Keep pairs greater or equal to `upper_limit`, and "expose" other pairs.
+                        interesting.retain(|(key, time)| {
+                            if upper_limit.less_equal(time) { true } else {
+                                exposed_keys.push_own(key);
+                                exposed_time.push_own(time);
+                                false
+                            }
+                        });
 
                         // Prepare an output buffer and builder for each capability.
                         //
@@ -471,12 +476,10 @@ where
                         // indicates whether more data remain. We move through `exposed` using (index) `exposed_position`.
                         // There could perhaps be a less provocative variable name.
                         let mut exposed_position = 0;
-                        while batch_cursor.key_valid(batch_storage) || exposed_position < exposed.len() {
-
-                            use std::borrow::Borrow;
+                        while batch_cursor.key_valid(batch_storage) || exposed_position < exposed_keys.len() {
 
                             // Determine the next key we will work on; could be synthetic, could be from a batch.
-                            let key1 = exposed.get(exposed_position).map(|x| T1::KeyContainer::borrow_as(&x.0));
+                            let key1 = exposed_keys.get(exposed_position);
                             let key2 = batch_cursor.get_key(batch_storage);
                             let key = match (key1, key2) {
                                 (Some(key1), Some(key2)) => ::std::cmp::min(key1, key2),
@@ -492,8 +495,8 @@ where
                             interesting_times.clear();
 
                             // Populate `interesting_times` with synthetic interesting times (below `upper_limit`) for this key.
-                            while exposed.get(exposed_position).map(|x| x.0.borrow()).map(|k| key.eq(&T1::KeyContainer::borrow_as(&k))).unwrap_or(false) {
-                                interesting_times.push(exposed[exposed_position].1.clone());
+                            while exposed_keys.get(exposed_position) == Some(key) {
+                                interesting_times.push(T1::owned_time(exposed_time.index(exposed_position)));
                                 exposed_position += 1;
                             }
 

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

@@ -207,10 +207,15 @@ pub mod layers {
         ///
         /// Tracked independently to account for duplicate compression.
         total: usize,
+
+        /// Time container to stage singleton times for evaluation.
+        time_con: T,
+        /// Diff container to stage singleton times for evaluation.
+        diff_con: D,
     }
 
     impl<T: BatchContainer, D: BatchContainer> Default for UpdsBuilder<T, D> {
-        fn default() -> Self { Self { stash: Vec::default(), total: 0, } }
+        fn default() -> Self { Self { stash: Vec::default(), total: 0, time_con: BatchContainer::with_capacity(1), diff_con: BatchContainer::with_capacity(1) } }
     }
 
 
@@ -230,34 +235,28 @@ pub mod layers {
         pub fn seal<O: for<'a> BatchContainer<ReadItem<'a> = usize>>(&mut self, upds: &mut Upds<O, T, D>) -> bool {
             use crate::consolidation;
             consolidation::consolidate(&mut self.stash);
-            if !self.stash.is_empty() {
-                // If there is a single element, equal to a just-prior recorded update,
-                // we push nothing and report an unincremented offset to encode this case.
-                let time_diff = upds.times.last().zip(upds.diffs.last());
-                let last_eq = self.stash.last().zip(time_diff).map(|((t1, d1), (t2, d2))| {
-                    let t1 = T::borrow_as(t1);
-                    let d1 = D::borrow_as(d1);
-                    t1.eq(&t2) && d1.eq(&d2)
-                });
-                if self.stash.len() == 1 && last_eq.unwrap_or(false) {
-                    // Just clear out the stash, as we won't drain it here.
+            // If everything consolidates away, return false.
+            if self.stash.is_empty() { return false; }
+            // If there is a singleton, we may be able to optimize.
+            if self.stash.len() == 1 {
+                let (time, diff) = self.stash.last().unwrap();
+                self.time_con.clear(); self.time_con.push_own(time);
+                self.diff_con.clear(); self.diff_con.push_own(diff);
+                if upds.times.last() == self.time_con.get(0) && upds.diffs.last() == self.diff_con.get(0) {
                     self.total += 1;
                     self.stash.clear();
                     upds.offs.push_ref(upds.times.len());
+                    return true;
                 }
-                else {
-                    // Conventional; move `stash` into `updates`.
-                    self.total += self.stash.len();
-                    for (time, diff) in self.stash.drain(..) {
-                        upds.times.push_own(&time);
-                        upds.diffs.push_own(&diff);
-                    }
-                    upds.offs.push_ref(upds.times.len());
-                }
-                true
-            } else {
-                false
             }
+            // Conventional; move `stash` into `updates`.
+            self.total += self.stash.len();
+            for (time, diff) in self.stash.drain(..) {
+                upds.times.push_own(&time);
+                upds.diffs.push_own(&diff);
+            }
+            upds.offs.push_ref(upds.times.len());
+            true
         }
 
         /// Completes the building and returns the total updates sealed.

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

@@ -204,6 +204,13 @@ mod val_batch {
             self.key_count += 1;
         }
 
+        /// Inserts a reference to an owned key, inefficiently. Should be removed.
+        fn insert_key_own(&mut self, key: &layout::Key<L>, offset: Option<usize>) {
+            let mut key_con = L::KeyContainer::with_capacity(1);
+            key_con.push_own(&key);
+            self.insert_key(key_con.index(0), offset)
+        }
+
         /// Indicates both the desired location and the hash signature of the key.
         fn desired_location<K: Hashable>(&self, key: &K) -> usize {
             if self.divisor == 0 { 0 }
@@ -608,9 +615,8 @@ mod val_batch {
                 // we push nothing and report an unincremented offset to encode this case.
                 let time_diff = self.result.times.last().zip(self.result.diffs.last());
                 let last_eq = self.update_stash.last().zip(time_diff).map(|((t1, d1), (t2, d2))| {
-                    let t1 = L::TimeContainer::borrow_as(t1);
-                    let d1 = L::DiffContainer::borrow_as(d1);
-                    t1.eq(&t2) && d1.eq(&d2)
+                    // TODO: The use of `into_owned` is a work-around for not having reference types.
+                    *t1 == L::TimeContainer::into_owned(t2) && *d1 == L::DiffContainer::into_owned(d2)
                 });
                 if self.update_stash.len() == 1 && last_eq.unwrap_or(false) {
                     // Just clear out update_stash, as we won't drain it here.
@@ -767,9 +773,8 @@ mod val_batch {
         /// to recover the singleton to push it into `updates` to join the second update.
         fn push_update(&mut self, time: layout::Time<L>, diff: layout::Diff<L>) {
             // If a just-pushed update exactly equals `(time, diff)` we can avoid pushing it.
-            let t1 = L::TimeContainer::borrow_as(&time);
-            let d1 = L::DiffContainer::borrow_as(&diff);
-            if self.result.times.last().map(|t| t == t1).unwrap_or(false) && self.result.diffs.last().map(|d| d == d1).unwrap_or(false) {
+            // TODO: The use of `into_owned` is a bandage for not having references we can compare.
+            if self.result.times.last().map(|t| L::TimeContainer::into_owned(t) == time).unwrap_or(false) && self.result.diffs.last().map(|d| L::DiffContainer::into_owned(d) == diff).unwrap_or(false) {
                 assert!(self.singleton.is_none());
                 self.singleton = Some((time, diff));
             }
@@ -849,7 +854,7 @@ mod val_batch {
                     self.push_update(time, diff);
                     self.result.vals.push_into(val);
                     // Insert the key, but with no specified offset.
-                    self.result.insert_key(L::KeyContainer::borrow_as(&key), None);
+                    self.result.insert_key_own(&key, None);
                 }
             }
         }

dogsdogsdogs/src/operators/half_join.rs

@@ -331,13 +331,21 @@ where
     let (mut cursor, storage) = trace.cursor();
     let mut yielded = false;
 
+    let mut key_con = Tr::KeyContainer::with_capacity(1);
+    let mut time_con = Tr::TimeContainer::with_capacity(1);
+    for time in frontier.iter() {
+        time_con.push_own(time);
+    }
+
     // Process proposals one at a time, stopping if we should yield.
     for ((ref key, ref val1, ref time), ref initial, ref mut diff1) in proposals.iter_mut() {
         // Use TOTAL ORDER to allow the release of `time`.
         yielded = yielded || yield_function(timer, *work);
-        if !yielded && !frontier.iter().any(|t| comparison(Tr::TimeContainer::borrow_as(t), initial)) {
-            cursor.seek_key(&storage, Tr::KeyContainer::borrow_as(key));
-            if cursor.get_key(&storage) == Some(Tr::KeyContainer::borrow_as(key)) {
+
+        if !yielded && !(0 .. time_con.len()).any(|i| comparison(time_con.index(i), initial)) {
+            key_con.clear(); key_con.push_own(&key);
+            cursor.seek_key(&storage, key_con.index(0));
+            if cursor.get_key(&storage) == key_con.get(0) {
                 while let Some(val2) = cursor.get_val(&storage) {
                     cursor.map_times(&storage, |t, d| {
                         if comparison(t, initial) {

dogsdogsdogs/src/operators/lookup_map.rs

@@ -89,12 +89,14 @@ where
                     });
 
                     let (mut cursor, storage) = trace.cursor();
-
+                    // Key container to stage keys for comparison.
+                    let mut key_con = Tr::KeyContainer::with_capacity(1);
                     for &mut (ref prefix, ref time, ref mut diff) in prefixes.iter_mut() {
                         if !input2.frontier.less_equal(time) {
                             logic2(prefix, &mut key1);
-                            cursor.seek_key(&storage, Tr::KeyContainer::borrow_as(&key1));
-                            if cursor.get_key(&storage) == Some(Tr::KeyContainer::borrow_as(&key1)) {
+                            key_con.clear(); key_con.push_own(&key1);
+                            cursor.seek_key(&storage, key_con.index(1));
+                            if cursor.get_key(&storage) == Some(key_con.index(1)) {
                                 while let Some(value) = cursor.get_val(&storage) {
                                     let mut count = Tr::Diff::zero();
                                     cursor.map_times(&storage, |t, d| {