Remove use of timely_sort crate (TimelyDataflow#313)

frankmcsherry · web-flow · commit b168506cd74a · 2021-03-09T19:42:15.000-05:00
* remove use of timely_sort crate

* correct tests
diff --git a/Cargo.toml b/Cargo.toml
@@ -29,7 +29,6 @@ serde = "1.0"
 serde_derive = "1.0"
 abomonation = "0.7"
 abomonation_derive = "0.5"
-timely_sort="0.1.6"
 #timely = { version = "0.11", default-features = false }
 timely = { git = "https://github.com/TimelyDataflow/timely-dataflow", default-features = false }
 #timely = { path = "../timely-dataflow/timely/", default-features = false }
diff --git a/dogsdogsdogs/Cargo.toml b/dogsdogsdogs/Cargo.toml
@@ -8,7 +8,6 @@ license = "MIT"
 abomonation = "0.7"
 abomonation_derive = "0.5"
 timely = { git = "https://github.com/TimelyDataflow/timely-dataflow", default-features = false }
-timely_sort = "0.1.6"
 differential-dataflow = { path = "../", default-features = false }
 serde = "1"
 serde_derive = "1"
diff --git a/dogsdogsdogs/src/operators/lookup_map.rs b/dogsdogsdogs/src/operators/lookup_map.rs
@@ -6,8 +6,6 @@ use timely::dataflow::channels::pact::{Pipeline, Exchange};
 use timely::dataflow::operators::Operator;
 use timely::progress::Antichain;
 
-use timely_sort::Unsigned;
-
 use differential_dataflow::{ExchangeData, Collection, AsCollection, Hashable};
 use differential_dataflow::difference::{Semigroup, Monoid};
 use differential_dataflow::lattice::Lattice;
@@ -58,7 +56,7 @@ where
     let mut key: Tr::Key = supplied_key0;
     let exchange = Exchange::new(move |update: &(D,G::Timestamp,R)| {
         logic1(&update.0, &mut key);
-        key.hashed().as_u64()
+        key.hashed().into()
     });
 
     let mut key1: Tr::Key = supplied_key1;
diff --git a/src/hashable.rs b/src/hashable.rs
@@ -15,11 +15,6 @@
 //! distributed integers can perhaps do something simpler (like report their own value).
 
 use std::hash::Hasher;
-use std::ops::Deref;
-
-use abomonation::Abomonation;
-
-use timely_sort::Unsigned;
 
 /// Types with a `hashed` method, producing an unsigned output of some type.
 ///
@@ -28,7 +23,7 @@ use timely_sort::Unsigned;
 /// can take advantage of the smaller size.
 pub trait Hashable {
     /// The type of the output value.
-    type Output: Unsigned+Copy;
+    type Output: Into<u64>+Copy;
     /// A well-distributed integer derived from the data.
     fn hashed(&self) -> Self::Output;
 }
@@ -41,139 +36,3 @@ impl<T: ::std::hash::Hash> Hashable for T {
         h.finish()
     }
 }
-
-/// A marker trait for types whose `Ord` implementation orders first by `hashed()`.
-///
-/// Types implementing this trait *must* implement `Ord` and satisfy the property that two values
-/// with different hashes have the same order as their hashes. This trait allows implementations
-/// that sort by hash value to rely on the `Ord` implementation of the type.
-pub trait HashOrdered : Ord+Hashable { }
-impl<T: Ord+Hashable> HashOrdered for OrdWrapper<T> { }
-impl<T: Ord+Hashable> HashOrdered for HashableWrapper<T> { }
-impl<T: Unsigned+Copy> HashOrdered for UnsignedWrapper<T> { }
-
-// It would be great to use the macros for these, but I couldn't figure out how to get it
-// to work with constraints (i.e. `Hashable`) on the generic parameters.
-impl<T: Ord+Hashable+Abomonation> Abomonation for OrdWrapper<T> {
-    #[inline] unsafe fn entomb<W: ::std::io::Write>(&self, write: &mut W) -> ::std::io::Result<()> {
-        self.item.entomb(write)
-    }
-    #[inline] unsafe fn exhume<'a,'b>(&'a mut self, mut bytes: &'b mut [u8]) -> Option<&'b mut [u8]> {
-        let temp = bytes;
-        bytes = self.item.exhume(temp)?;
-        Some(bytes)
-    }
-}
-
-// It would be great to use the macros for these, but I couldn't figure out how to get it
-// to work with constraints (i.e. `Hashable`) on the generic parameters.
-impl<T: Hashable+Abomonation> Abomonation for HashableWrapper<T> {
-
-    #[inline] unsafe fn entomb<W: ::std::io::Write>(&self, write: &mut W) -> ::std::io::Result<()> {
-        self.item.entomb(write)
-    }
-    #[inline] unsafe fn exhume<'a,'b>(&'a mut self, mut bytes: &'b mut [u8]) -> Option<&'b mut [u8]> {
-        let temp = bytes;
-        bytes = self.item.exhume(temp)?;
-        Some(bytes)
-    }
-}
-
-impl<T: Unsigned+Copy+Hashable+Abomonation> Abomonation for UnsignedWrapper<T> {
-
-    #[inline] unsafe fn entomb<W: ::std::io::Write>(&self, write: &mut W) -> ::std::io::Result<()> {
-        self.item.entomb(write)
-    }
-    #[inline] unsafe fn exhume<'a,'b>(&'a mut self, mut bytes: &'b mut [u8]) -> Option<&'b mut [u8]> {
-        let temp = bytes;
-        bytes = self.item.exhume(temp)?;
-        Some(bytes)
-    }
-}
-
-
-/// A wrapper around hashable types that ensures an implementation of `Ord` that compares
-/// hash values first.
-#[derive(Clone, Eq, PartialEq, Debug, Default)]
-pub struct OrdWrapper<T: Ord+Hashable> {
-    /// The item, so you can grab it.
-    pub item: T
-}
-
-impl<T: Ord+Hashable> PartialOrd for OrdWrapper<T> {
-    #[inline]
-    fn partial_cmp(&self, other: &Self) -> Option<::std::cmp::Ordering> {
-        (self.item.hashed(), &self.item).partial_cmp(&(other.item.hashed(), &other.item))
-    }
-}
-impl<T: Ord+Hashable> Ord for OrdWrapper<T> {
-    #[inline]
-    fn cmp(&self, other: &Self) -> ::std::cmp::Ordering {
-        (self.item.hashed(), &self.item).cmp(&(other.item.hashed(), &other.item))
-    }
-}
-
-impl<T: Ord+Hashable> Hashable for OrdWrapper<T> {
-    type Output = T::Output;
-    fn hashed(&self) -> T::Output { self.item.hashed() }
-}
-
-impl<T: Ord+Hashable> Deref for OrdWrapper<T> {
-    type Target = T;
-    fn deref(&self) -> &T { &self.item }
-}
-
-
-/// Wrapper to stash hash value with the actual value.
-#[derive(Clone, Default, Ord, PartialOrd, Eq, PartialEq, Debug, Copy)]
-pub struct HashableWrapper<T: Hashable> {
-    hash: T::Output,
-    /// The item, for reference.
-    pub item: T,
-}
-
-impl<T: Hashable> Hashable for HashableWrapper<T> {
-    type Output = T::Output;
-    #[inline]
-    fn hashed(&self) -> T::Output { self.hash }
-}
-
-impl<T: Hashable> Deref for HashableWrapper<T> {
-    type Target = T;
-    #[inline]
-    fn deref(&self) -> &T { &self.item }
-}
-
-impl<T: Hashable> From<T> for HashableWrapper<T> {
-    #[inline]
-    fn from(item: T) -> HashableWrapper<T> {
-        HashableWrapper {
-            hash: item.hashed(),
-            item,
-        }
-    }
-}
-
-/// A wrapper around an unsigned integer, providing `hashed` as the value itself.
-#[derive(Clone, Ord, PartialOrd, Eq, PartialEq, Default, Debug, Copy)]
-pub struct UnsignedWrapper<T: Unsigned+Copy> {
-    /// The item.
-    pub item: T,
-}
-
-impl<T: Unsigned+Copy> Hashable for UnsignedWrapper<T> {
-    type Output = T;
-    #[inline]
-    fn hashed(&self) -> Self::Output { self.item }
-}
-
-impl<T: Unsigned+Copy> Deref for UnsignedWrapper<T> {
-    type Target = T;
-    #[inline]
-    fn deref(&self) -> &T { &self.item }
-}
-
-impl<T: Unsigned+Copy> From<T> for UnsignedWrapper<T> {
-    #[inline]
-    fn from(item: T) -> Self { UnsignedWrapper { item } }
-}
diff --git a/src/lib.rs b/src/lib.rs
@@ -92,7 +92,6 @@ impl<T: timely::ExchangeData + Ord + Debug> ExchangeData for T { }
 
 extern crate fnv;
 extern crate timely;
-extern crate timely_sort;
 
 #[macro_use]
 extern crate abomonation_derive;
diff --git a/src/operators/arrange/agent.rs b/src/operators/arrange/agent.rs
@@ -187,7 +187,6 @@ where
     /// use differential_dataflow::operators::reduce::Reduce;
     /// use differential_dataflow::trace::Trace;
     /// use differential_dataflow::trace::implementations::ord::OrdValSpine;
-    /// use differential_dataflow::hashable::OrdWrapper;
     ///
     /// fn main() {
     ///     ::timely::execute(Config::thread(), |worker| {
@@ -247,7 +246,6 @@ where
     /// use differential_dataflow::operators::reduce::Reduce;
     /// use differential_dataflow::trace::Trace;
     /// use differential_dataflow::trace::implementations::ord::OrdValSpine;
-    /// use differential_dataflow::hashable::OrdWrapper;
     ///
     /// fn main() {
     ///     ::timely::execute(Config::thread(), |worker| {
@@ -361,7 +359,6 @@ where
     /// use differential_dataflow::trace::Trace;
     /// use differential_dataflow::trace::TraceReader;
     /// use differential_dataflow::trace::implementations::ord::OrdValSpine;
-    /// use differential_dataflow::hashable::OrdWrapper;
     /// use differential_dataflow::input::Input;
     ///
     /// fn main() {
diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs
@@ -26,8 +26,6 @@ use timely::progress::Timestamp;
 use timely::progress::{Antichain, frontier::AntichainRef};
 use timely::dataflow::operators::Capability;
 
-use timely_sort::Unsigned;
-
 use ::{Data, ExchangeData, Collection, AsCollection, Hashable};
 use ::difference::Semigroup;
 use lattice::Lattice;
@@ -274,7 +272,7 @@ where
         Tr: 'static,
     {
         // while the arrangement is already correctly distributed, the query stream may not be.
-        let exchange = Exchange::new(move |update: &(Tr::Key,G::Timestamp)| update.0.hashed().as_u64());
+        let exchange = Exchange::new(move |update: &(Tr::Key,G::Timestamp)| update.0.hashed().into());
         queries.binary_frontier(&self.stream, exchange, Pipeline, "TraceQuery", move |_capability, _info| {
 
             let mut trace = Some(self.trace.clone());
@@ -484,7 +482,7 @@ where
         Tr::Batch: Batch<K, V, G::Timestamp, R>,
         Tr::Cursor: Cursor<K, V, G::Timestamp, R>,
     {
-        let exchange = Exchange::new(move |update: &((K,V),G::Timestamp,R)| (update.0).0.hashed().as_u64());
+        let exchange = Exchange::new(move |update: &((K,V),G::Timestamp,R)| (update.0).0.hashed().into());
         self.arrange_core(exchange, name)
     }
 
diff --git a/src/operators/arrange/upsert.rs b/src/operators/arrange/upsert.rs
@@ -115,8 +115,6 @@ use timely::progress::Timestamp;
 use timely::progress::Antichain;
 use timely::dataflow::operators::Capability;
 
-use timely_sort::Unsigned;
-
 use ::{ExchangeData, Hashable};
 use lattice::Lattice;
 use trace::{Trace, TraceReader, Batch, Cursor};
@@ -157,7 +155,7 @@ where
 
         let reader = &mut reader;
 
-        let exchange = Exchange::new(move |update: &(Tr::Key,Option<Tr::Val>,G::Timestamp)| (update.0).hashed().as_u64());
+        let exchange = Exchange::new(move |update: &(Tr::Key,Option<Tr::Val>,G::Timestamp)| (update.0).hashed().into());
 
         stream.unary_frontier(exchange, name, move |_capability, info| {
 
diff --git a/src/operators/join.rs b/src/operators/join.rs
@@ -237,7 +237,6 @@ pub trait JoinCore<G: Scope, K: 'static, V: 'static, R: Semigroup> where G::Time
     /// use differential_dataflow::operators::join::JoinCore;
     /// use differential_dataflow::trace::Trace;
     /// use differential_dataflow::trace::implementations::ord::OrdValSpine;
-    /// use differential_dataflow::hashable::OrdWrapper;
     ///
     /// fn main() {
     ///     ::timely::example(|scope| {
diff --git a/src/operators/reduce.rs b/src/operators/reduce.rs
@@ -253,7 +253,6 @@ pub trait ReduceCore<G: Scope, K: Data, V: Data, R: Semigroup> where G::Timestam
     /// use differential_dataflow::operators::reduce::ReduceCore;
     /// use differential_dataflow::trace::Trace;
     /// use differential_dataflow::trace::implementations::ord::OrdValSpine;
-    /// use differential_dataflow::hashable::OrdWrapper;
     ///
     /// fn main() {
     ///     ::timely::example(|scope| {
diff --git a/tests/trace.rs b/tests/trace.rs
@@ -6,27 +6,25 @@ use std::rc::Rc;
 use timely::dataflow::operators::generic::OperatorInfo;
 use timely::progress::{Antichain, frontier::AntichainRef};
 
-use differential_dataflow::hashable::UnsignedWrapper;
-
 use differential_dataflow::trace::implementations::ord::OrdValBatch;
 use differential_dataflow::trace::{Trace, TraceReader, Batch, Batcher};
 use differential_dataflow::trace::cursor::CursorDebug;
 use differential_dataflow::trace::implementations::spine_fueled::Spine;
 
 pub type OrdValSpine<K, V, T, R> = Spine<K, V, T, R, Rc<OrdValBatch<K, V, T, R>>>;
 
-type IntegerTrace = OrdValSpine<UnsignedWrapper<u64>, u64, usize, i64>;
+type IntegerTrace = OrdValSpine<u64, u64, usize, i64>;
 
-fn get_trace() -> Spine<UnsignedWrapper<u64>, u64, usize, i64, Rc<OrdValBatch<UnsignedWrapper<u64>, u64, usize, i64>>> {
+fn get_trace() -> Spine<u64, u64, usize, i64, Rc<OrdValBatch<u64, u64, usize, i64>>> {
     let op_info = OperatorInfo::new(0, 0, &[]);
     let mut trace = IntegerTrace::new(op_info, None, None);
     {
-        let mut batcher = <<IntegerTrace as TraceReader>::Batch as Batch<UnsignedWrapper<u64>, u64, usize, i64>>::Batcher::new();
+        let mut batcher = <<IntegerTrace as TraceReader>::Batch as Batch<u64, u64, usize, i64>>::Batcher::new();
 
         batcher.push_batch(&mut vec![
-            ((1.into(), 2), 0, 1),
-            ((2.into(), 3), 1, 1),
-            ((2.into(), 3), 2, -1),
+            ((1, 2), 0, 1),
+            ((2, 3), 1, 1),
+            ((2, 3), 2, -1),
         ]);
 
         let batch_ts = &[1, 2, 3];
@@ -44,21 +42,21 @@ fn test_trace() {
 
     let (mut cursor1, storage1) = trace.cursor_through(AntichainRef::new(&[1])).unwrap();
     let vec_1 = cursor1.to_vec(&storage1);
-    assert_eq!(vec_1, vec![((1.into(), 2), vec![(0, 1)])]);
+    assert_eq!(vec_1, vec![((1, 2), vec![(0, 1)])]);
 
     let (mut cursor2, storage2) = trace.cursor_through(AntichainRef::new(&[2])).unwrap();
     let vec_2 = cursor2.to_vec(&storage2);
     println!("--> {:?}", vec_2);
     assert_eq!(vec_2, vec![
-               ((1.into(), 2), vec![(0, 1)]),
-               ((2.into(), 3), vec![(1, 1)]),
+               ((1, 2), vec![(0, 1)]),
+               ((2, 3), vec![(1, 1)]),
     ]);
 
     let (mut cursor3, storage3) = trace.cursor_through(AntichainRef::new(&[3])).unwrap();
     let vec_3 = cursor3.to_vec(&storage3);
     assert_eq!(vec_3, vec![
-               ((1.into(), 2), vec![(0, 1)]),
-               ((2.into(), 3), vec![(1, 1), (2, -1)]),
+               ((1, 2), vec![(0, 1)]),
+               ((2, 3), vec![(1, 1), (2, -1)]),
     ]);
 
     let (mut cursor4, storage4) = trace.cursor();
