2010YOUY01 commented on code in PR #23816:
URL: https://github.com/apache/datafusion/pull/23816#discussion_r3780810351
##########
datafusion/functions-aggregate/src/hyperloglog.rs:
##########
@@ -70,107 +80,174 @@ impl<T> HyperLogLog<T>
where
T: Hash + ?Sized,
{
- /// Creates a new, empty HyperLogLog.
+ /// Creates a new, empty HyperLogLog with the default precision (14).
pub fn new() -> Self {
- let registers = [0; NUM_REGISTERS];
- Self::new_with_registers(registers)
+ Self::with_precision(DEFAULT_HLL_P)
}
- /// Creates a HyperLogLog from already populated registers
- /// note that this method should not be invoked in untrusted environment
- /// because the internal structure of registers are not examined.
- pub(crate) fn new_with_registers(registers: [u8; NUM_REGISTERS]) -> Self {
+ /// Creates a new, empty HyperLogLog with the given precision `p`.
+ ///
+ /// The number of registers is `2^p`. Supported range:
`HLL_P_MIN..=DEFAULT_HLL_P`.
+ #[inline(always)]
+ pub fn with_precision(p: usize) -> Self {
+ assert!(
+ (HLL_P_MIN..=DEFAULT_HLL_P).contains(&p),
+ "HLL precision must be in {HLL_P_MIN}..={DEFAULT_HLL_P}, got {p}",
+ );
+ let q = 64 - p;
+ let p_mask = ((1_usize << p) as u64) - 1;
Self {
- registers,
+ p,
+ q,
+ p_mask,
+ phantom: PhantomData,
+ registers: [0u8; 1 << DEFAULT_HLL_P],
+ }
+ }
+
+ /// Creates a HyperLogLog from already populated registers.
+ ///
+ /// The precision is inferred from the register slice length, which must be
+ /// a power of two in the range `2^HLL_P_MIN..=2^DEFAULT_HLL_P`.
+ ///
+ /// Note that this method should not be invoked in an untrusted environment
+ /// because the internal structure of registers is not examined.
+ pub(crate) fn from_registers(v: &[u8]) -> Self {
+ let len = v.len();
+ assert!(
+ len.is_power_of_two(),
+ "register slice length must be a power of two, got {len}",
+ );
+ let p = len.ilog2() as usize;
+ assert!(
+ (HLL_P_MIN..=DEFAULT_HLL_P).contains(&p),
+ "inferred precision {p} is outside {HLL_P_MIN}..={DEFAULT_HLL_P}",
+ );
+ let q = 64 - p;
+ let p_mask = (len as u64) - 1;
+ let mut registers = [0u8; 1 << DEFAULT_HLL_P];
+ registers[..len].copy_from_slice(v);
+ Self {
+ p,
+ q,
+ p_mask,
phantom: PhantomData,
+ registers,
}
}
- /// The HLL hash state is shared through `datafusion_common::hash_utils`
- /// so sketches remain compatible across accumulators.
+ /// The precision of this sketch.
#[inline]
- fn hash_value(&self, obj: &T) -> u64 {
- HLL_HASH_STATE.hash_one(obj)
+ pub(crate) fn precision(&self) -> usize {
+ self.p
+ }
+
+ /// Heap bytes used by the register buffer (not captured by `size_of_val`).
+ /// Always 0 — registers are stored inline in the struct.
+ pub(crate) fn register_heap_size(&self) -> usize {
+ 0
}
/// Adds an element to the HyperLogLog.
+ #[cfg(test)]
pub fn add(&mut self, obj: &T) {
- let hash = self.hash_value(obj);
+ let hash = HLL_HASH_STATE.hash_one(obj);
self.add_hashed(hash);
}
- /// Adds a pre-computed hash value directly to the HyperLogLog.
- ///
- /// The hash should be computed using [`HLL_HASH_STATE`], the same hasher
used
- /// by [`Self::add`].
- #[inline]
+ /// Adds a pre-computed hash. Hash must be produced by [`HLL_HASH_STATE`].
+ #[inline(always)]
pub(crate) fn add_hashed(&mut self, hash: u64) {
- let index = (hash & HLL_P_MASK) as usize;
- let p = ((hash >> HLL_P) | (1_u64 << HLL_Q)).trailing_zeros() + 1;
- self.registers[index] = self.registers[index].max(p as u8);
+ let index = (hash & self.p_mask) as usize;
+ let rho = ((hash >> self.p) | (1_u64 << self.q)).trailing_zeros() + 1;
+ self.registers[index] = self.registers[index].max(rho as u8);
+ }
+
+ /// Adds a slice of pre-computed hashes. Hashes must use
[`HLL_HASH_STATE`].
+ pub(crate) fn add_hashed_slice(&mut self, hashes: &[u64]) {
+ // Default precision: use module-level constants so LLVM emits
immediates.
+ // Other precisions: hoist struct fields outside the loop.
+ if self.p == DEFAULT_HLL_P {
+ for &hash in hashes {
+ let index = (hash & DEFAULT_MASK) as usize;
+ let rho =
+ ((hash >> DEFAULT_HLL_P) | (1_u64 <<
DEFAULT_Q)).trailing_zeros() + 1;
+ self.registers[index] = self.registers[index].max(rho as u8);
+ }
+ } else {
+ let (p, q, p_mask) = (self.p, self.q, self.p_mask);
+ for &hash in hashes {
+ let index = (hash & p_mask) as usize;
+ let rho = ((hash >> p) | (1_u64 << q)).trailing_zeros() + 1;
+ self.registers[index] = self.registers[index].max(rho as u8);
+ }
+ }
}
- /// Get the register histogram (each value in register index into
- /// the histogram; u32 is enough because we only have 2**14=16384 registers
#[inline]
- fn get_histogram(&self) -> [u32; HLL_Q + 2] {
- let mut histogram = [0; HLL_Q + 2];
- // hopefully this can be unrolled
- for r in self.registers {
- histogram[r as usize] += 1;
+ fn get_histogram(&self) -> [u32; 64 - HLL_P_MIN + 2] {
+ let mut histogram = [0u32; 64 - HLL_P_MIN + 2];
+ for r in &self.registers[..1 << self.p] {
+ histogram[*r as usize] += 1;
}
histogram
}
- /// Merge the other [`HyperLogLog`] into this one
+ /// Merge the other [`HyperLogLog`] into this one.
pub fn merge(&mut self, other: &HyperLogLog<T>) {
- assert!(
- self.registers.len() == other.registers.len(),
- "unexpected got unequal register size, expect {}, got {}",
- self.registers.len(),
- other.registers.len()
+ assert_eq!(
+ self.p, other.p,
+ "cannot merge HLL sketches with different precisions ({} vs {})",
+ self.p, other.p
);
- for i in 0..self.registers.len() {
+ let n = 1 << self.p;
+ for i in 0..n {
self.registers[i] = self.registers[i].max(other.registers[i]);
}
}
/// Guess the number of unique elements seen by the HyperLogLog.
pub fn count(&self) -> usize {
- count_from_histogram(&self.get_histogram())
+ count_from_histogram(&self.get_histogram()[..self.q + 2], self.p)
}
}
-/// Compute `index` and `rho` (register value) for a precomputed hash, exactly
as
-/// [`HyperLogLog::add_hashed`] does.
+/// Compute `index` and `rho` (register value) for a precomputed hash at a
given
+/// precision, exactly as [`HyperLogLog::add_hashed`] does.
#[inline]
-pub(crate) fn register_for_hash(hash: u64) -> (usize, u8) {
- let index = (hash & HLL_P_MASK) as usize;
- let rho = (((hash >> HLL_P) | (1_u64 << HLL_Q)).trailing_zeros() + 1) as
u8;
+pub(crate) fn register_for_hash(hash: u64, p: usize) -> (usize, u8) {
+ let q = 64 - p;
+ let p_mask: u64 = ((1_usize << p) as u64) - 1;
+ let index = (hash & p_mask) as usize;
+ let rho = (((hash >> p) | (1_u64 << q)).trailing_zeros() + 1) as u8;
(index, rho)
}
/// Estimate the cardinality of a set of precomputed hashes without
-/// materializing a full [`NUM_REGISTERS`]-byte register array.
+/// materializing a full register array.
///
/// This is equivalent to adding every hash to a fresh [`HyperLogLog`] via
/// [`HyperLogLog::add_hashed`] and calling [`HyperLogLog::count`], but only
does
/// work proportional to the number of hashes. It is used to cheaply estimate
the
/// many small groups produced by a high-cardinality `GROUP BY`, where
allocating
-/// and scanning a 16 KiB sketch per group would dominate the runtime.
+/// and scanning a sketch per group would dominate the runtime.
///
/// `hashes` may contain duplicates (duplicate hashes are idempotent).
-pub(crate) fn count_from_hashes(hashes: &[u64]) -> usize {
+pub(crate) fn count_from_hashes(hashes: &[u64], p: usize) -> usize {
Review Comment:
leaking precision to all internal APIs seem not necessary
probably we can do something like
```
enum GroupHllBuffer {
/// Distinct hashes seen so far. May contain duplicates between
compactions.
Sparse(Vec<u64>),
Dense(Box<HyperLogLog<u8>>),
}
struct GroupHLL {
hll_buffer: GroupHLLBuffer,
p: usize,
}
```
and move all the implementation to the new struct
##########
datafusion/functions-aggregate/src/hyperloglog.rs:
##########
@@ -70,107 +80,174 @@ impl<T> HyperLogLog<T>
where
T: Hash + ?Sized,
{
- /// Creates a new, empty HyperLogLog.
+ /// Creates a new, empty HyperLogLog with the default precision (14).
pub fn new() -> Self {
- let registers = [0; NUM_REGISTERS];
- Self::new_with_registers(registers)
+ Self::with_precision(DEFAULT_HLL_P)
}
- /// Creates a HyperLogLog from already populated registers
- /// note that this method should not be invoked in untrusted environment
- /// because the internal structure of registers are not examined.
- pub(crate) fn new_with_registers(registers: [u8; NUM_REGISTERS]) -> Self {
+ /// Creates a new, empty HyperLogLog with the given precision `p`.
+ ///
+ /// The number of registers is `2^p`. Supported range:
`HLL_P_MIN..=DEFAULT_HLL_P`.
+ #[inline(always)]
+ pub fn with_precision(p: usize) -> Self {
+ assert!(
+ (HLL_P_MIN..=DEFAULT_HLL_P).contains(&p),
+ "HLL precision must be in {HLL_P_MIN}..={DEFAULT_HLL_P}, got {p}",
+ );
+ let q = 64 - p;
+ let p_mask = ((1_usize << p) as u64) - 1;
Self {
- registers,
+ p,
+ q,
+ p_mask,
+ phantom: PhantomData,
+ registers: [0u8; 1 << DEFAULT_HLL_P],
+ }
+ }
+
+ /// Creates a HyperLogLog from already populated registers.
+ ///
+ /// The precision is inferred from the register slice length, which must be
+ /// a power of two in the range `2^HLL_P_MIN..=2^DEFAULT_HLL_P`.
+ ///
+ /// Note that this method should not be invoked in an untrusted environment
+ /// because the internal structure of registers is not examined.
+ pub(crate) fn from_registers(v: &[u8]) -> Self {
+ let len = v.len();
+ assert!(
+ len.is_power_of_two(),
+ "register slice length must be a power of two, got {len}",
+ );
+ let p = len.ilog2() as usize;
+ assert!(
+ (HLL_P_MIN..=DEFAULT_HLL_P).contains(&p),
+ "inferred precision {p} is outside {HLL_P_MIN}..={DEFAULT_HLL_P}",
+ );
+ let q = 64 - p;
+ let p_mask = (len as u64) - 1;
+ let mut registers = [0u8; 1 << DEFAULT_HLL_P];
+ registers[..len].copy_from_slice(v);
+ Self {
+ p,
+ q,
+ p_mask,
phantom: PhantomData,
+ registers,
}
}
- /// The HLL hash state is shared through `datafusion_common::hash_utils`
- /// so sketches remain compatible across accumulators.
+ /// The precision of this sketch.
#[inline]
- fn hash_value(&self, obj: &T) -> u64 {
- HLL_HASH_STATE.hash_one(obj)
+ pub(crate) fn precision(&self) -> usize {
+ self.p
+ }
+
+ /// Heap bytes used by the register buffer (not captured by `size_of_val`).
+ /// Always 0 — registers are stored inline in the struct.
+ pub(crate) fn register_heap_size(&self) -> usize {
+ 0
}
/// Adds an element to the HyperLogLog.
+ #[cfg(test)]
pub fn add(&mut self, obj: &T) {
- let hash = self.hash_value(obj);
+ let hash = HLL_HASH_STATE.hash_one(obj);
self.add_hashed(hash);
}
- /// Adds a pre-computed hash value directly to the HyperLogLog.
- ///
- /// The hash should be computed using [`HLL_HASH_STATE`], the same hasher
used
- /// by [`Self::add`].
- #[inline]
+ /// Adds a pre-computed hash. Hash must be produced by [`HLL_HASH_STATE`].
+ #[inline(always)]
pub(crate) fn add_hashed(&mut self, hash: u64) {
- let index = (hash & HLL_P_MASK) as usize;
- let p = ((hash >> HLL_P) | (1_u64 << HLL_Q)).trailing_zeros() + 1;
- self.registers[index] = self.registers[index].max(p as u8);
+ let index = (hash & self.p_mask) as usize;
+ let rho = ((hash >> self.p) | (1_u64 << self.q)).trailing_zeros() + 1;
+ self.registers[index] = self.registers[index].max(rho as u8);
+ }
+
+ /// Adds a slice of pre-computed hashes. Hashes must use
[`HLL_HASH_STATE`].
+ pub(crate) fn add_hashed_slice(&mut self, hashes: &[u64]) {
Review Comment:
I think small micro-bench regression is acceptable if we want additional
feature, this performance hack is unnecessary, could we remove that to make the
implementation simpler?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
