vendor streaming_algorithm

Author: alecmocatta

Cargo.toml

@@ -44,6 +44,7 @@ amadeus-commoncrawl = { version = "=0.3.7", path = "amadeus-commoncrawl", option
 amadeus-parquet = { version = "=0.3.7", path = "amadeus-parquet", optional = true }
 amadeus-postgres = { version = "=0.3.7", path = "amadeus-postgres", optional = true }
 amadeus-serde = { version = "=0.3.7", path = "amadeus-serde", optional = true }
+amadeus-streaming = { version = "=0.3.7", path = "amadeus-streaming" }
 async-channel = "1.1"
 bincode = { version = "1.3", optional = true }
 constellation-rs = { version = "0.2.0-alpha.2", default-features = false, optional = true }
@@ -72,7 +73,6 @@ doc-comment = "0.3"
 either = { version = "1.5", features = ["serde"] }
 rand = "0.7"
 serde_json = "1.0"
-streaming_algorithms = "0.3"
 tokio = { version = "0.2", features = ["macros", "time"] }
 
 [target.'cfg(target_arch = "wasm32")'.dev-dependencies]

amadeus-core/Cargo.toml

@@ -19,6 +19,7 @@ azure-devops = { project = "alecmocatta/amadeus", pipeline = "tests", build = "2
 maintenance = { status = "actively-developed" }
 
 [dependencies]
+amadeus-streaming = { version = "=0.3.7", path = "../amadeus-streaming" }
 async-trait = "0.1"
 derive-new = "0.5"
 educe = "0.4"
@@ -33,7 +34,6 @@ rand = "0.7"
 replace_with = "0.1"
 serde = { version = "1.0", features = ["derive"] }
 serde_closure = "0.3"
-streaming_algorithms = "0.3"
 sum = { version = "0.1", features = ["futures", "serde"] }
 tokio = { version = "0.2", features = ["blocking", "rt-core"] }
 walkdir = "2.2"

amadeus-core/src/par_sink/sample.rs

@@ -1,10 +1,10 @@
 #![allow(clippy::type_complexity)]
 
+use amadeus_streaming::{HyperLogLogMagnitude, SampleUnstable as SASampleUnstable, Top};
 use derive_new::new;
 use rand::thread_rng;
 use serde::{Deserialize, Serialize};
 use std::hash::Hash;
-use streaming_algorithms::{HyperLogLogMagnitude, SampleUnstable as SASampleUnstable, Top};
 
 use super::{
 	folder_par_sink, FolderSync, FolderSyncReducer, ParallelPipe, ParallelSink, SumFolder, SumZeroFolder

amadeus-core/src/par_stream.rs

@@ -296,7 +296,7 @@ macro_rules! stream {
 			#[inline]
 			async fn most_frequent<P>(
 				self, pool: &P, n: usize, probability: f64, tolerance: f64,
-			) -> ::streaming_algorithms::Top<Self::Item, usize>
+			) -> ::amadeus_streaming::Top<Self::Item, usize>
 			where
 				P: $pool,
 				Self::Item: Hash + Eq + Clone + $send + 'static,
@@ -313,7 +313,7 @@ macro_rules! stream {
 			#[inline]
 			async fn most_distinct<P, A, B>(
 				self, pool: &P, n: usize, probability: f64, tolerance: f64, error_rate: f64,
-			) -> ::streaming_algorithms::Top<A, streaming_algorithms::HyperLogLogMagnitude<B>>
+			) -> ::amadeus_streaming::Top<A, amadeus_streaming::HyperLogLogMagnitude<B>>
 			where
 				P: $pool,
 				Self: $stream<Item = (A, B)> + Sized,
@@ -337,7 +337,7 @@ macro_rules! stream {
 			#[inline]
 			async fn sample_unstable<P>(
 				self, pool: &P, samples: usize,
-			) -> ::streaming_algorithms::SampleUnstable<Self::Item>
+			) -> ::amadeus_streaming::SampleUnstable<Self::Item>
 			where
 				P: $pool,
 				Self::Item: $send + 'static,

amadeus-streaming/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "amadeus-streaming"
+version = "0.3.7"
+license = "Apache-2.0"
+authors = ["Alec Mocatta <[email protected]>"]
+categories = ["data-structures", "algorithms", "science"]
+keywords = ["streaming-algorithm", "probabilistic", "sketch", "data-structure", "hyperloglog"]
+description = """
+SIMD-accelerated implementations of various streaming algorithms, including Count–min sketch, Top k, HyperLogLog, Reservoir sampling.
+"""
+repository = "https://github.com/alecmocatta/amadeus"
+homepage = "https://github.com/alecmocatta/amadeus"
+documentation = "https://docs.rs/amadeus"
+readme = "README.md"
+edition = "2018"
+
+[badges]
+azure-devops = { project = "alecmocatta/amadeus", pipeline = "tests", build = "26" }
+maintenance = { status = "actively-developed" }
+
+[features]
+nightly = ["packed_simd"]
+
+[dependencies]
+twox-hash = "1.1"
+serde = { version = "1.0", features = ["derive"] }
+rand = { version = "0.7", features = ["small_rng"] }
+packed_simd = { version = "0.3", features = ["into_bits"], optional = true }
+
+[build-dependencies]
+rustversion = "1.0"

amadeus-streaming/README.md

@@ -0,0 +1,32 @@
+# amadeus-streaming
+
+SIMD-accelerated implementations of various [streaming algorithms](https://en.wikipedia.org/wiki/Streaming_algorithm).
+
+This library is a work in progress. PRs are very welcome! Currently implemented algorithms include:
+
+ * Count–min sketch
+ * Top k (Count–min sketch plus a doubly linked hashmap to track heavy hitters / top k keys when ordered by aggregated value)
+ * HyperLogLog
+ * Reservoir sampling
+
+A goal of this library is to enable composition of these algorithms; for example Top k + HyperLogLog to enable an approximate version of something akin to `SELECT key FROM table GROUP BY key ORDER BY COUNT(DISTINCT value) DESC LIMIT k`.
+
+Run your application with `RUSTFLAGS="-C target-cpu=native"` and the `nightly` feature to benefit from the SIMD-acceleration like so:
+
+```bash
+RUSTFLAGS="-C target-cpu=native" cargo run --features "streaming_algorithms/nightly" --release
+```
+
+See [this gist](https://gist.github.com/debasishg/8172796) for a good list of further algorithms to be implemented. Other resources are [Probabilistic data structures – Wikipedia](https://en.wikipedia.org/wiki/Category:Probabilistic_data_structures), [DataSketches – A similar Java library originating at Yahoo](https://datasketches.github.io/), and [Algebird  – A similar Java library originating at Twitter](https://github.com/twitter/algebird).
+
+As these implementations are often in hot code paths, unsafe is used, albeit only when necessary to a) achieve the asymptotically optimal algorithm or b) mitigate an observed bottleneck.
+
+## License
+Licensed under either of
+
+ * Apache License, Version 2.0, ([LICENSE-APACHE.txt](LICENSE-APACHE.txt) or http://www.apache.org/licenses/LICENSE-2.0)
+ * MIT license ([LICENSE-MIT.txt](LICENSE-MIT.txt) or http://opensource.org/licenses/MIT)
+
+at your option.
+
+Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.

amadeus-streaming/build.rs

@@ -0,0 +1,12 @@
+fn main() {
+	println!("cargo:rerun-if-changed=build.rs");
+
+	nightly();
+}
+
+#[rustversion::nightly]
+fn nightly() {
+	println!("cargo:rustc-cfg=nightly");
+}
+#[rustversion::not(nightly)]
+fn nightly() {}