Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Created generic for hashers #3

Closed
wants to merge 8 commits into from
Closed

Created generic for hashers #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
5ed5509
Created generic for hashers
LucaCappelletti94 Aug 20, 2024
1f41b6e
Merge branch 'main' into hasher
LucaCappelletti94 Aug 20, 2024
8c875c8
Added default value
LucaCappelletti94 Aug 21, 2024
ab83d23
Formatted code
LucaCappelletti94 Aug 22, 2024
e9aab75
Updated serde for variant with hasher generic
LucaCappelletti94 Sep 17, 2024
2c9aa3a
Updated test suite for serde
LucaCappelletti94 Sep 17, 2024
0e1a805
Resolved clippy code smell
LucaCappelletti94 Sep 17, 2024
f72d3c8
Formatted document
LucaCappelletti94 Sep 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 34 additions & 26 deletions src/hyperloglog.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,39 +5,41 @@
//! 1. https://github.com/crepererum/pdatastructs.rs/blob/3997ed50f6b6871c9e53c4c5e0f48f431405fc63/src/hyperloglog.rs
//! 2. https://github.com/apache/arrow-datafusion/blob/f203d863f5c8bc9f133f6dd9b2e34e57ac3cdddc/datafusion/physical-expr/src/aggregate/hyperloglog.rs

use std::hash::Hash;

use ahash::RandomState;
use ahash::AHasher;
use std::hash::{Hash, Hasher};

/// By default, we use 2**14 registers like redis
const DEFAULT_P: usize = 14_usize;

/// Fixed seed
const SEED: RandomState = RandomState::with_seeds(
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removing these will cause incompatiable changes.

We need to use fixed seed.

0x355e438b4b1478c7_u64,
0xd0e8453cd135b473_u64,
0xf7b252066a57836a_u64,
0xb8a829e3713c09bf_u64,
);

/// HyperLogLog is a probabilistic data structure used to estimate the cardinality of a multiset.
///
/// Note: We don't make HyperLogLog as static struct by keeping `PhantomData<T>`
/// Callers should take care of its hash function to be unchanged.
/// P is the bucket number, must be [4, 18]
/// Q = 64 - P
/// Register num is 1 << P
#[derive(Clone, Debug, Eq, PartialEq)]
#[derive(Clone, Debug)]
#[cfg_attr(feature = "mem_dbg", derive(mem_dbg::MemSize, mem_dbg::MemDbg))]
pub struct HyperLogLog<const P: usize = DEFAULT_P> {
pub struct HyperLogLog<H = AHasher, const P: usize = DEFAULT_P> {
pub(crate) registers: Vec<u8>,
_hasher: std::marker::PhantomData<H>,
}

impl<const P: usize> Default for HyperLogLog<P> {
impl<H: Default + Hasher, const P: usize> Default for HyperLogLog<H, P> {
fn default() -> Self {
Self::new()
}
}

impl<const P: usize> HyperLogLog<P> {
impl<H, const P: usize> PartialEq for HyperLogLog<H, P> {
fn eq(&self, other: &Self) -> bool {
self.registers == other.registers
}
}

impl<H, const P: usize> Eq for HyperLogLog<H, P> {}

impl<H: Default + Hasher, const P: usize> HyperLogLog<H, P> {
/// note that this method should not be invoked in untrusted environment
pub fn new() -> Self {
assert!(
Expand All @@ -48,13 +50,17 @@ impl<const P: usize> HyperLogLog<P> {

Self {
registers: vec![0; 1 << P],
_hasher: std::marker::PhantomData,
}
}

pub fn with_registers(registers: Vec<u8>) -> Self {
assert_eq!(registers.len(), Self::number_registers());

Self { registers }
Self {
registers,
_hasher: std::marker::PhantomData,
}
}

/// Adds an hash to the HyperLogLog.
Expand All @@ -69,7 +75,9 @@ impl<const P: usize> HyperLogLog<P> {
/// Adds an object to the HyperLogLog.
/// Though we could pass different types into this method, caller should notice that
pub fn add_object<T: Hash>(&mut self, obj: &T) {
let hash = SEED.hash_one(obj);
let mut hasher = H::default();
obj.hash(&mut hasher);
let hash = hasher.finish();
self.add_hash(hash);
}

Expand Down Expand Up @@ -191,7 +199,7 @@ fn hll_tau(x: f64) -> f64 {

#[cfg(test)]
mod tests {
use crate::HyperLogLog;
use super::*;

const P: usize = 14;
const NUM_REGISTERS: usize = 1 << P;
Expand All @@ -217,7 +225,7 @@ mod tests {

macro_rules! sized_number_test {
($SIZE: expr, $T: tt) => {{
let mut hll = HyperLogLog::<P>::new();
let mut hll = HyperLogLog::<AHasher, P>::new();
for i in 0..$SIZE {
hll.add_object(&(i as $T));
}
Expand Down Expand Up @@ -246,13 +254,13 @@ mod tests {

#[test]
fn test_empty() {
let hll = HyperLogLog::<P>::new();
let hll = HyperLogLog::<AHasher, P>::new();
assert_eq!(hll.count(), 0);
}

#[test]
fn test_one() {
let mut hll = HyperLogLog::<P>::new();
let mut hll = HyperLogLog::<AHasher, P>::new();
hll.add_hash(1);
assert_eq!(hll.count(), 1);
}
Expand Down Expand Up @@ -284,19 +292,19 @@ mod tests {

#[test]
fn test_empty_merge() {
let mut hll = HyperLogLog::<P>::new();
hll.merge(&HyperLogLog::<P>::new());
let mut hll = HyperLogLog::<AHasher, P>::new();
hll.merge(&HyperLogLog::<AHasher, P>::new());
assert_eq!(hll.count(), 0);
}

#[test]
fn test_merge_overlapped() {
let mut hll = HyperLogLog::<P>::new();
let mut hll = HyperLogLog::<AHasher, P>::new();
for i in 0..1000 {
hll.add_object(&i);
}

let other = HyperLogLog::<P>::new();
let other = HyperLogLog::<AHasher, P>::new();
for i in 0..1000 {
hll.add_object(&i);
}
Expand All @@ -307,7 +315,7 @@ mod tests {

#[test]
fn test_repetition() {
let mut hll = HyperLogLog::<P>::new();
let mut hll = HyperLogLog::<AHasher, P>::new();
for i in 0..1_000_000 {
hll.add_object(&(i % 1000));
}
Expand Down
33 changes: 19 additions & 14 deletions src/serde.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use crate::HyperLogLog;
use core::hash::Hasher;

#[derive(serde::Serialize, borsh::BorshSerialize)]
enum HyperLogLogVariantRef<'a, const P: usize> {
Expand All @@ -14,30 +15,33 @@ enum HyperLogLogVariant<const P: usize> {
Full(Vec<u8>),
}

impl<const P: usize> From<HyperLogLogVariant<P>> for HyperLogLog<P> {
impl<H: Default + Hasher, const P: usize> From<HyperLogLogVariant<P>> for HyperLogLog<H, P> {
fn from(value: HyperLogLogVariant<P>) -> Self {
match value {
HyperLogLogVariant::Empty => HyperLogLog::<P>::new(),
HyperLogLogVariant::Empty => Self::new(),
HyperLogLogVariant::Sparse { data } => {
let mut registers = vec![0; 1 << P];
for (index, val) in data {
registers[index as usize] = val;
}

HyperLogLog::<P> { registers }
Self::with_registers(registers)
}
HyperLogLogVariant::Full(registers) => HyperLogLog::<P> { registers },
HyperLogLogVariant::Full(registers) => Self::with_registers(registers),
}
}
}

impl<'a, const P: usize> From<&'a HyperLogLog<P>> for HyperLogLogVariantRef<'a, P> {
fn from(hll: &'a HyperLogLog<P>) -> Self {
let none_empty_registers = HyperLogLog::<P>::number_registers() - hll.num_empty_registers();
impl<'a, H: Default + Hasher, const P: usize> From<&'a HyperLogLog<H, P>>
for HyperLogLogVariantRef<'a, P>
{
fn from(hll: &'a HyperLogLog<H, P>) -> Self {
let none_empty_registers =
HyperLogLog::<H, P>::number_registers() - hll.num_empty_registers();

if none_empty_registers == 0 {
HyperLogLogVariantRef::Empty
} else if none_empty_registers * 3 <= HyperLogLog::<P>::number_registers() {
} else if none_empty_registers * 3 <= HyperLogLog::<H, P>::number_registers() {
// If the number of empty registers is larger enough, we can use sparse serialize to reduce the binary size
// each register in sparse format will occupy 3 bytes, 2 for register index and 1 for register value.
let sparse_data: Vec<(u16, u8)> = hll
Expand All @@ -60,7 +64,7 @@ impl<'a, const P: usize> From<&'a HyperLogLog<P>> for HyperLogLogVariantRef<'a,
}
}

impl<const P: usize> serde::Serialize for HyperLogLog<P> {
impl<H: Default + Hasher, const P: usize> serde::Serialize for HyperLogLog<H, P> {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
Expand All @@ -70,7 +74,7 @@ impl<const P: usize> serde::Serialize for HyperLogLog<P> {
}
}

impl<'de, const P: usize> serde::Deserialize<'de> for HyperLogLog<P> {
impl<'de, H: Default + Hasher, const P: usize> serde::Deserialize<'de> for HyperLogLog<H, P> {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
Expand All @@ -80,14 +84,14 @@ impl<'de, const P: usize> serde::Deserialize<'de> for HyperLogLog<P> {
}
}

impl<const P: usize> borsh::BorshSerialize for HyperLogLog<P> {
impl<H: Default + Hasher, const P: usize> borsh::BorshSerialize for HyperLogLog<H, P> {
fn serialize<W: std::io::prelude::Write>(&self, writer: &mut W) -> std::io::Result<()> {
let v: HyperLogLogVariantRef<'_, P> = self.into();
v.serialize(writer)
}
}

impl<const P: usize> borsh::BorshDeserialize for HyperLogLog<P> {
impl<H: Default + Hasher, const P: usize> borsh::BorshDeserialize for HyperLogLog<H, P> {
fn deserialize_reader<R: std::io::prelude::Read>(reader: &mut R) -> std::io::Result<Self> {
let v = HyperLogLogVariant::<P>::deserialize_reader(reader)?;
Ok(v.into())
Expand All @@ -97,19 +101,20 @@ impl<const P: usize> borsh::BorshDeserialize for HyperLogLog<P> {
#[cfg(test)]
mod tests {
use crate::HyperLogLog;
use ahash::AHasher;

const P: usize = 14;
#[test]
fn test_serde() {
let mut hll = HyperLogLog::<P>::new();
let mut hll = HyperLogLog::<AHasher, P>::new();
json_serde_equal(&hll);

for i in 0..100000 {
hll.add_object(&(i % 200));
}
json_serde_equal(&hll);

let hll = HyperLogLog::<P>::with_registers(vec![1; 1 << P]);
let hll = HyperLogLog::<AHasher, P>::with_registers(vec![1; 1 << P]);
json_serde_equal(&hll);
}

Expand Down
Loading