diff --git a/CHANGELOG.md b/CHANGELOG.md index 67d8965..8dd97ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,7 @@ All significant changes to this project will be documented in this file. * T-Digest rejects truncated serialized payloads before allocating, and updating a deserialized digest no longer allows its buffered state to grow without bound. * Compact HLL4 images now restore all register values correctly. * `HllSketch::lower_bound` now uses the number of non-zero registers as a floor in HLL mode, matching Java, C++, and Go and avoiding a bound below the distinct count already proven by register hits. +* `HllUnion` now keeps a single HLL-mode input's estimate stable when copying or downsampling it and keeps confidence bounds consistent across HLL4, HLL6, and HLL8 result types, matching Java and C++. * HLL, Theta, and Tuple deserializers now return `InvalidData` for malformed payload sizes and entry counts instead of risking oversized allocations or decoding failures. * Malformed CPC images now return `InvalidData` instead of panicking. * Seeded deserializers now return `InvalidData` rather than panicking when the caller supplies a seed whose hash is the reserved zero value. diff --git a/datasketches/src/hll/array4.rs b/datasketches/src/hll/array4.rs index 17327fb..a076aa1 100644 --- a/datasketches/src/hll/array4.rs +++ b/datasketches/src/hll/array4.rs @@ -28,7 +28,8 @@ use crate::common::NumStdDev; use crate::error::Error; use crate::hll::Coupon; use crate::hll::aux_map::AuxMap; -use crate::hll::estimator::HipEstimator; +use crate::hll::estimator::EstimateState; +use crate::hll::estimator::Estimator; use crate::hll::serialization::COMPACT_FLAG_MASK; use crate::hll::serialization::COUPON_SIZE_BYTES; use crate::hll::serialization::CUR_MODE_HLL; @@ -70,8 +71,7 @@ pub struct Array4 { num_at_cur_min: u32, /// Exception table for values >= 15 after cur_min offset aux_map: Option, - /// HIP estimator for cardinality estimation - estimator: HipEstimator, + estimator: Estimator, } impl Array4 { @@ -84,7 +84,7 @@ impl Array4 { cur_min: 0, num_at_cur_min, aux_map: None, - estimator: HipEstimator::new(lg_config_k), + estimator: Estimator::new(lg_config_k), } } @@ -125,9 +125,9 @@ impl Array4 { 1 << self.lg_config_k } - /// Get the current HIP accumulator value - pub(super) fn hip_accum(&self) -> f64 { - self.estimator.hip_accum() + /// Returns the estimate state independently from register-derived cached values. + pub(super) fn estimate_state(&self) -> EstimateState { + self.estimator.estimate_state() } /// Set raw 4-bit value in slot @@ -270,7 +270,7 @@ impl Array4 { self.num_at_cur_min = num_at_new; } - /// Get the current cardinality estimate using HIP estimator + /// Returns the current cardinality estimate. pub fn estimate(&self) -> f64 { // Array4 tracks cur_min and num_at_cur_min dynamically self.estimator @@ -297,11 +297,9 @@ impl Array4 { ) } - /// Set the HIP accumulator value - /// - /// This is used when promoting from coupon modes to carry forward the estimate - pub fn set_hip_accum(&mut self, value: f64) { - self.estimator.set_hip_accum(value); + /// Restores estimate state after copying or transforming the same logical sketch. + pub(super) fn restore_estimate_state(&mut self, state: EstimateState) { + self.estimator.restore_estimate_state(state); } /// Check if the sketch is empty (all slots are zero) @@ -322,7 +320,7 @@ impl Array4 { let k = 1usize << lg_config_k; let num_bytes = 1usize << (lg_config_k - 1); // k/2 bytes for 4-bit packing - // Read HIP estimator values from preamble + // Read estimator values from preamble let hip_accum = cursor .read_f64_le() .map_err(insufficient_data("hip_accum"))?; @@ -404,12 +402,7 @@ impl Array4 { aux_map = Some(aux); } - // Create estimator and restore state - let mut estimator = HipEstimator::new(lg_config_k); - estimator.set_hip_accum(hip_accum); - estimator.set_kxq0(kxq0); - estimator.set_kxq1(kxq1); - estimator.set_out_of_order(ooo); + let estimator = Estimator::from_serialized(hip_accum, kxq0, kxq1, ooo); Ok(Self { lg_config_k, @@ -449,7 +442,7 @@ impl Array4 { // COMPACT_FLAG_MASK is always set: aux map entries are written as a compact sequential // list of populated entries only. let mut flags = COMPACT_FLAG_MASK; - if self.estimator.is_out_of_order() { + if self.estimator.uses_composite_estimate() { flags |= OUT_OF_ORDER_FLAG_MASK; } bytes.write_u8(flags); @@ -460,7 +453,7 @@ impl Array4 { // Mode byte: HLL mode with HLL4 type bytes.write_u8(encode_mode_byte(CUR_MODE_HLL, TGT_HLL4)); - // Write HIP estimator values + // Write estimator values bytes.write_f64_le(self.estimator.hip_accum()); bytes.write_f64_le(self.estimator.kxq0()); bytes.write_f64_le(self.estimator.kxq1()); diff --git a/datasketches/src/hll/array6.rs b/datasketches/src/hll/array6.rs index 69bfa22..5ab5657 100644 --- a/datasketches/src/hll/array6.rs +++ b/datasketches/src/hll/array6.rs @@ -28,7 +28,8 @@ use crate::codec::family::Family; use crate::common::NumStdDev; use crate::error::Error; use crate::hll::Coupon; -use crate::hll::estimator::HipEstimator; +use crate::hll::estimator::EstimateState; +use crate::hll::estimator::Estimator; use crate::hll::serialization::CUR_MODE_HLL; use crate::hll::serialization::HLL_PREAMBLE_SIZE; use crate::hll::serialization::HLL_PREINTS; @@ -47,8 +48,7 @@ pub struct Array6 { bytes: Box<[u8]>, /// Count of slots with value 0 num_zeros: u32, - /// HIP estimator for cardinality estimation - estimator: HipEstimator, + estimator: Estimator, } impl Array6 { @@ -60,7 +60,7 @@ impl Array6 { lg_config_k, bytes: vec![0u8; num_bytes].into_boxed_slice(), num_zeros: k, - estimator: HipEstimator::new(lg_config_k), + estimator: Estimator::new(lg_config_k), } } @@ -91,9 +91,9 @@ impl Array6 { 1 << self.lg_config_k } - /// Get the current HIP accumulator value - pub(super) fn hip_accum(&self) -> f64 { - self.estimator.hip_accum() + /// Returns the estimate state independently from register-derived cached values. + pub(super) fn estimate_state(&self) -> EstimateState { + self.estimator.estimate_state() } /// Set value in a slot (6-bit value) @@ -145,7 +145,7 @@ impl Array6 { } } - /// Get the current cardinality estimate using HIP estimator + /// Returns the current cardinality estimate. pub fn estimate(&self) -> f64 { // Array6 doesn't use cur_min (always 0), so num_at_cur_min = num_zeros self.estimator.estimate(self.lg_config_k, 0, self.num_zeros) @@ -163,11 +163,9 @@ impl Array6 { .lower_bound(self.lg_config_k, 0, self.num_zeros, num_std_dev) } - /// Set the HIP accumulator value - /// - /// This is used when promoting from coupon modes to carry forward the estimate - pub fn set_hip_accum(&mut self, value: f64) { - self.estimator.set_hip_accum(value); + /// Restores estimate state after copying or transforming the same logical sketch. + pub(super) fn restore_estimate_state(&mut self, state: EstimateState) { + self.estimator.restore_estimate_state(state); } /// Check if the sketch is empty (all slots are zero) @@ -186,7 +184,7 @@ impl Array6 { let k = 1 << lg_config_k; let num_bytes = num_bytes_for_k(k); - // Read HIP estimator values from preamble + // Read estimator values from preamble let hip_accum = cursor .read_f64_le() .map_err(insufficient_data("hip_accum"))?; @@ -218,12 +216,7 @@ impl Array6 { .read_exact(&mut data) .map_err(insufficient_data("data"))?; - // Create estimator and restore state - let mut estimator = HipEstimator::new(lg_config_k); - estimator.set_hip_accum(hip_accum); - estimator.set_kxq0(kxq0); - estimator.set_kxq1(kxq1); - estimator.set_out_of_order(ooo); + let estimator = Estimator::from_serialized(hip_accum, kxq0, kxq1, ooo); Ok(Self { lg_config_k, @@ -251,7 +244,7 @@ impl Array6 { // Write flags let mut flags = 0u8; - if self.estimator.is_out_of_order() { + if self.estimator.uses_composite_estimate() { flags |= OUT_OF_ORDER_FLAG_MASK; } bytes.write_u8(flags); @@ -262,7 +255,7 @@ impl Array6 { // Mode byte: HLL mode with HLL6 type bytes.write_u8(encode_mode_byte(CUR_MODE_HLL, TGT_HLL6)); - // Write HIP estimator values + // Write estimator values bytes.write_f64_le(self.estimator.hip_accum()); bytes.write_f64_le(self.estimator.kxq0()); bytes.write_f64_le(self.estimator.kxq1()); diff --git a/datasketches/src/hll/array8.rs b/datasketches/src/hll/array8.rs index 0cc1e2a..8a3b61c 100644 --- a/datasketches/src/hll/array8.rs +++ b/datasketches/src/hll/array8.rs @@ -27,7 +27,8 @@ use crate::codec::family::Family; use crate::common::NumStdDev; use crate::error::Error; use crate::hll::Coupon; -use crate::hll::estimator::HipEstimator; +use crate::hll::estimator::EstimateState; +use crate::hll::estimator::Estimator; use crate::hll::serialization::CUR_MODE_HLL; use crate::hll::serialization::HLL_PREAMBLE_SIZE; use crate::hll::serialization::HLL_PREINTS; @@ -44,8 +45,7 @@ pub struct Array8 { bytes: Box<[u8]>, /// Count of slots with value 0 num_zeros: u32, - /// HIP estimator for cardinality estimation - estimator: HipEstimator, + estimator: Estimator, } impl Array8 { @@ -56,7 +56,7 @@ impl Array8 { lg_config_k, bytes: vec![0u8; k as usize].into_boxed_slice(), num_zeros: k, - estimator: HipEstimator::new(lg_config_k), + estimator: Estimator::new(lg_config_k), } } @@ -99,7 +99,7 @@ impl Array8 { } } - /// Get the current cardinality estimate using HIP estimator + /// Returns the current cardinality estimate. pub fn estimate(&self) -> f64 { // Array8 doesn't use cur_min (always 0), so num_at_cur_min = num_zeros self.estimator.estimate(self.lg_config_k, 0, self.num_zeros) @@ -117,13 +117,6 @@ impl Array8 { .lower_bound(self.lg_config_k, 0, self.num_zeros, num_std_dev) } - /// Set the HIP accumulator value - /// - /// This is used when promoting from coupon modes to carry forward the estimate - pub fn set_hip_accum(&mut self, value: f64) { - self.estimator.set_hip_accum(value); - } - /// Check if the sketch is empty (all slots are zero) pub fn is_empty(&self) -> bool { self.num_zeros == (1 << self.lg_config_k) @@ -139,9 +132,14 @@ impl Array8 { 1 << self.lg_config_k } - /// Get the current HIP accumulator value - pub(super) fn hip_accum(&self) -> f64 { - self.estimator.hip_accum() + /// Returns the estimate state independently from register-derived cached values. + pub(super) fn estimate_state(&self) -> EstimateState { + self.estimator.estimate_state() + } + + /// Restores estimate state after copying or transforming the same logical sketch. + pub(super) fn restore_estimate_state(&mut self, state: EstimateState) { + self.estimator.restore_estimate_state(state); } /// Directly set a register value @@ -154,17 +152,17 @@ impl Array8 { /// Rebuild estimator state from current register values /// - /// Recomputes num_zeros, kxq0, kxq1, and marks estimator as out-of-order. + /// Recomputes num_zeros, kxq0, and kxq1, then switches to composite estimation. /// Should be called after bulk register modifications. pub(super) fn rebuild_estimator_from_registers(&mut self) { self.rebuild_cached_values(); - self.estimator.set_out_of_order(true); + self.estimator.invalidate_hip(); } /// Merge another Array8 with the same lg_k /// - /// Performs register-by-register max merge. Marks estimator as - /// out-of-order since HIP cannot be maintained during bulk operations. + /// Performs register-by-register max merge. HIP is invalidated because the register update + /// order is unavailable. /// /// # Panics /// @@ -181,7 +179,7 @@ impl Array8 { } self.rebuild_cached_values(); - self.estimator.set_out_of_order(true); + self.estimator.invalidate_hip(); } /// Merge an array with larger lg_k (downsampling) @@ -219,7 +217,7 @@ impl Array8 { } self.rebuild_cached_values(); - self.estimator.set_out_of_order(true); + self.estimator.invalidate_hip(); } /// Rebuild cached values after bulk modifications @@ -245,8 +243,7 @@ impl Array8 { } } - self.estimator.set_kxq0(kxq0_sum); - self.estimator.set_kxq1(kxq1_sum); + self.estimator.restore_kxq(kxq0_sum, kxq1_sum); } /// Deserialize Array8 from HLL mode bytes @@ -259,7 +256,7 @@ impl Array8 { ) -> Result { let k = 1usize << lg_config_k; - // Read HIP estimator values from preamble + // Read estimator values from preamble let hip_accum = cursor .read_f64_le() .map_err(insufficient_data("hip_accum"))?; @@ -291,12 +288,7 @@ impl Array8 { .read_exact(&mut data) .map_err(insufficient_data("data"))?; - // Create estimator and restore state - let mut estimator = HipEstimator::new(lg_config_k); - estimator.set_hip_accum(hip_accum); - estimator.set_kxq0(kxq0); - estimator.set_kxq1(kxq1); - estimator.set_out_of_order(ooo); + let estimator = Estimator::from_serialized(hip_accum, kxq0, kxq1, ooo); Ok(Self { lg_config_k, @@ -323,7 +315,7 @@ impl Array8 { // Write flags let mut flags = 0u8; - if self.estimator.is_out_of_order() { + if self.estimator.uses_composite_estimate() { flags |= OUT_OF_ORDER_FLAG_MASK; } bytes.write_u8(flags); @@ -334,7 +326,7 @@ impl Array8 { // Mode byte: HLL mode with HLL8 type bytes.write_u8(encode_mode_byte(CUR_MODE_HLL, TGT_HLL8)); - // Write HIP estimator values + // Write estimator values bytes.write_f64_le(self.estimator.hip_accum()); bytes.write_f64_le(self.estimator.kxq0()); bytes.write_f64_le(self.estimator.kxq1()); @@ -537,8 +529,8 @@ mod tests { assert_eq!(dst.get(2), 35, "dst[2] updated to larger value"); assert_eq!(dst.get(3), 40, "dst[3] got new value"); - // Verify estimator marked as OOO - assert!(dst.estimator.is_out_of_order()); + // Bulk merges require composite estimation. + assert!(dst.estimator.uses_composite_estimate()); // Verify num_zeros updated (should be 12: 16 - 4 non-zero) assert_eq!(dst.num_zeros, 12); @@ -567,8 +559,8 @@ mod tests { assert_eq!(dst.get(0), 25, "dst[0] = max(10, 15, 25)"); assert_eq!(dst.get(1), 30, "dst[1] = max(20, 18, 30)"); - // Verify estimator marked as OOO - assert!(dst.estimator.is_out_of_order()); + // Bulk merges require composite estimation. + assert!(dst.estimator.uses_composite_estimate()); } #[test] diff --git a/datasketches/src/hll/composite_interpolation.rs b/datasketches/src/hll/composite_interpolation.rs index 9f2083e..e5e13e3 100644 --- a/datasketches/src/hll/composite_interpolation.rs +++ b/datasketches/src/hll/composite_interpolation.rs @@ -15,11 +15,10 @@ // specific language governing permissions and limitations // under the License. -//! Composite interpolation tables for HLL out-of-order estimation +//! Composite interpolation tables for HLL estimation. //! -//! These tables are used with cubic interpolation to provide accurate -//! cardinality estimates when the HLL sketch is in out-of-order mode -//! (after deserialization or merging). +//! These tables correct raw HyperLogLog estimates when the register update history is unavailable, +//! such as after merging independent sketches. //! //! Currently, this module contains tables for common lg_k values (4-12). The full C++ //! implementation has tables for lg_k 4-21. Additional tables can be found at: diff --git a/datasketches/src/hll/estimator.rs b/datasketches/src/hll/estimator.rs index 903e780..f607e8e 100644 --- a/datasketches/src/hll/estimator.rs +++ b/datasketches/src/hll/estimator.rs @@ -15,12 +15,10 @@ // specific language governing permissions and limitations // under the License. -//! HIP (Historical Inverse Probability) Estimator for HyperLogLog +//! Cardinality estimation for HLL-mode sketches. //! -//! The HIP estimator provides improved cardinality estimation by maintaining -//! an accumulator that tracks the historical sequence of register updates. -//! This is more accurate than the standard HLL estimator, especially for -//! moderate cardinalities. +//! Sequential register updates use HIP (Historical Inverse Probability). Bulk merges use the +//! composite estimator because register values do not retain their update order. use crate::common::NumStdDev; use crate::common::inv_pow2::inv_pow2; @@ -28,37 +26,56 @@ use crate::hll::composite_interpolation; use crate::hll::cubic_interpolation; use crate::hll::harmonic_numbers; -/// HIP estimator with KxQ registers for improved cardinality estimation -/// -/// This struct encapsulates all estimation-related state and logic, -/// allowing it to be composed into Array4, Array6, and Array8. +/// Selects the estimate that is valid for the current register history. +#[derive(Debug, Clone, Copy, PartialEq)] +pub(super) enum EstimateState { + /// The register updates have a known order, so the HIP accumulator is valid. + Hip(f64), + /// A bulk merge lost the update order, so the estimate must come from the registers. + Composite, +} + +/// Cardinality estimator shared by Array4, Array6, and Array8. /// -/// The estimator supports two modes: -/// * **In-order mode**: Uses HIP (Historical Inverse Probability) accumulator for accurate -/// sequential updates -/// * **Out-of-order mode**: Uses composite estimator (raw HLL + linear counting) after -/// deserialization or merging +/// Sequential updates use HIP. Bulk merges use the composite estimator because register values do +/// not retain update order. Both modes maintain KxQ as a cache derived from the current registers. #[derive(Debug, Clone, PartialEq)] -pub struct HipEstimator { - /// HIP estimator accumulator - hip_accum: f64, +pub struct Estimator { + /// Estimate selected by the register history. + estimate_state: EstimateState, /// KxQ register for values < 32 (larger inverse powers) kxq0: f64, /// KxQ register for values >= 32 (tiny inverse powers) kxq1: f64, - /// Out-of-order flag: when true, HIP updates are skipped - out_of_order: bool, } -impl HipEstimator { - /// Create a new HIP estimator for a sketch with 2^lg_config_k registers +impl Estimator { + /// Creates an estimator for a sketch with 2^lg_config_k registers. pub fn new(lg_config_k: u8) -> Self { let k = 1 << lg_config_k; Self { - hip_accum: 0.0, + estimate_state: EstimateState::Hip(0.0), kxq0: k as f64, // All registers start at 0, so kxq0 = k * (1/2^0) = k kxq1: 0.0, - out_of_order: false, + } + } + + /// Restores estimator fields from an HLL serialization preamble. + pub(super) fn from_serialized( + hip_accum: f64, + kxq0: f64, + kxq1: f64, + out_of_order: bool, + ) -> Self { + let estimate_state = if out_of_order { + EstimateState::Composite + } else { + EstimateState::Hip(hip_accum) + }; + Self { + estimate_state, + kxq0, + kxq1, } } @@ -68,7 +85,7 @@ impl HipEstimator { /// /// # Algorithm /// - /// 1. Update HIP accumulator (unless out-of-order) + /// 1. Update the HIP accumulator when it remains valid /// 2. Update KxQ registers (always) /// /// The KxQ registers are split for numerical precision: @@ -77,13 +94,12 @@ impl HipEstimator { pub fn update(&mut self, lg_config_k: u8, old_value: u8, new_value: u8) { let k = (1 << lg_config_k) as f64; - // Update HIP accumulator FIRST (unless out-of-order) - // When out-of-order (from deserialization or merge), HIP is invalid - if !self.out_of_order { - self.hip_accum += k / (self.kxq0 + self.kxq1); + // Update HIP accumulator FIRST when the register history is still available. + if let EstimateState::Hip(hip_accum) = &mut self.estimate_state { + *hip_accum += k / (self.kxq0 + self.kxq1); } - // Always update KxQ registers (regardless of OOO flag) + // Always update KxQ because it depends only on the current registers. self.update_kxq(old_value, new_value); } @@ -106,7 +122,7 @@ impl HipEstimator { /// Get the current cardinality estimate /// - /// Dispatches to either HIP or composite estimator based on out-of-order flag. + /// Dispatches to the estimate selected by the register history. /// /// # Arguments /// @@ -114,10 +130,11 @@ impl HipEstimator { /// * `cur_min`: Current minimum register value (for Array4, 0 for Array6/8) /// * `num_at_cur_min`: Number of registers at cur_min value pub fn estimate(&self, lg_config_k: u8, cur_min: u8, num_at_cur_min: u32) -> f64 { - if self.out_of_order { - self.get_composite_estimate(lg_config_k, cur_min, num_at_cur_min) - } else { - self.hip_accum + match self.estimate_state { + EstimateState::Hip(hip_accum) => hip_accum, + EstimateState::Composite => { + self.composite_estimate(lg_config_k, cur_min, num_at_cur_min) + } } } @@ -139,7 +156,12 @@ impl HipEstimator { num_std_dev: NumStdDev, ) -> f64 { let estimate = self.estimate(lg_config_k, cur_min, num_at_cur_min); - let rse = get_rel_err(lg_config_k, true, self.out_of_order, num_std_dev); + let rse = relative_error( + lg_config_k, + true, + self.uses_composite_estimate(), + num_std_dev, + ); // RSE is negative for upper bounds, so (1 + rse) < 1, making bound > estimate estimate / (1.0 + rse) } @@ -165,7 +187,12 @@ impl HipEstimator { num_std_dev: NumStdDev, ) -> f64 { let estimate = self.estimate(lg_config_k, cur_min, num_at_cur_min); - let rse = get_rel_err(lg_config_k, false, self.out_of_order, num_std_dev); + let rse = relative_error( + lg_config_k, + false, + self.uses_composite_estimate(), + num_std_dev, + ); let config_k = 1u32 << lg_config_k; let num_nonzero_registers = if cur_min == 0 { config_k - num_at_cur_min @@ -181,7 +208,7 @@ impl HipEstimator { /// Formula: correctionFactor * k^2 / (kxq0 + kxq1) /// /// Uses lg_k-specific correction factors for small k. - fn get_raw_estimate(&self, lg_config_k: u8) -> f64 { + fn raw_estimate(&self, lg_config_k: u8) -> f64 { let k = (1 << lg_config_k) as f64; // Correction factors from empirical analysis @@ -198,7 +225,7 @@ impl HipEstimator { /// Get linear counting (bitmap) estimate for small cardinalities /// /// Uses harmonic numbers to estimate based on empty registers. - fn get_bitmap_estimate(&self, lg_config_k: u8, cur_min: u8, num_at_cur_min: u32) -> f64 { + fn bitmap_estimate(&self, lg_config_k: u8, cur_min: u8, num_at_cur_min: u32) -> f64 { let k = 1 << lg_config_k; // Number of unhit (empty) buckets @@ -215,11 +242,11 @@ impl HipEstimator { /// Get composite estimate (blends raw HLL and linear counting) /// - /// This is the primary estimator used when in out-of-order mode. - /// It uses cubic interpolation on raw HLL estimate, then blends + /// This estimate is used when the register update history is unavailable. + /// It uses cubic interpolation on the raw HLL estimate, then blends /// with linear counting for small cardinalities. - fn get_composite_estimate(&self, lg_config_k: u8, cur_min: u8, num_at_cur_min: u32) -> f64 { - let raw_est = self.get_raw_estimate(lg_config_k); + fn composite_estimate(&self, lg_config_k: u8, cur_min: u8, num_at_cur_min: u32) -> f64 { + let raw_estimate = self.raw_estimate(lg_config_k); // Get composite interpolation table let x_arr = composite_interpolation::get_x_arr(lg_config_k); @@ -227,35 +254,36 @@ impl HipEstimator { let y_stride = composite_interpolation::get_y_stride(lg_config_k) as f64; // Handle edge cases - if raw_est < x_arr[0] { + if raw_estimate < x_arr[0] { return 0.0; } let x_arr_len_m1 = x_arr_len - 1; // Above interpolation range: extrapolate linearly - if raw_est > x_arr[x_arr_len_m1] { + if raw_estimate > x_arr[x_arr_len_m1] { let final_y = y_stride * (x_arr_len_m1 as f64); let factor = final_y / x_arr[x_arr_len_m1]; - return raw_est * factor; + return raw_estimate * factor; } // Interpolate using cubic interpolation - let adj_est = cubic_interpolation::using_x_arr_and_y_stride(x_arr, y_stride, raw_est); + let adjusted_estimate = + cubic_interpolation::using_x_arr_and_y_stride(x_arr, y_stride, raw_estimate); // Avoid linear counting if estimate is high // (threshold: 3*k ensures we're above potential linear counting instability) let k = 1 << lg_config_k; - if adj_est > (3 * k) as f64 { - return adj_est; + if adjusted_estimate > (3 * k) as f64 { + return adjusted_estimate; } // Get linear counting estimate - let lin_est = self.get_bitmap_estimate(lg_config_k, cur_min, num_at_cur_min); + let linear_estimate = self.bitmap_estimate(lg_config_k, cur_min, num_at_cur_min); // Blend estimates based on crossover threshold // Use average to reduce bias from threshold comparison - let avg_est = (adj_est + lin_est) / 2.0; + let average_estimate = (adjusted_estimate + linear_estimate) / 2.0; // Crossover thresholds (empirically determined) let crossover = match lg_config_k { @@ -266,16 +294,19 @@ impl HipEstimator { let threshold = crossover * (k as f64); - if avg_est > threshold { - adj_est + if average_estimate > threshold { + adjusted_estimate } else { - lin_est + linear_estimate } } /// Get the HIP accumulator value pub fn hip_accum(&self) -> f64 { - self.hip_accum + match self.estimate_state { + EstimateState::Hip(hip_accum) => hip_accum, + EstimateState::Composite => 0.0, + } } /// Get the kxq0 register value @@ -288,38 +319,30 @@ impl HipEstimator { self.kxq1 } - /// Check if this estimator is in out-of-order mode - pub fn is_out_of_order(&self) -> bool { - self.out_of_order + /// Returns whether estimates are derived from registers rather than HIP history. + pub fn uses_composite_estimate(&self) -> bool { + matches!(self.estimate_state, EstimateState::Composite) } - /// Set the out-of-order flag - /// - /// This should be set to true when: - /// * Deserializing a sketch from bytes - /// * After a merge/union operation - pub fn set_out_of_order(&mut self, ooo: bool) { - self.out_of_order = ooo; - if ooo { - // When going out-of-order, invalidate HIP accumulator - // (it will be recomputed if needed via composite estimator) - self.hip_accum = 0.0; - } + /// Returns the estimate state independently from register-derived KxQ values. + pub(super) fn estimate_state(&self) -> EstimateState { + self.estimate_state } - /// Set the HIP accumulator directly - pub fn set_hip_accum(&mut self, value: f64) { - self.hip_accum = value; + /// Restores estimate state after copying or transforming the same logical sketch. + pub(super) fn restore_estimate_state(&mut self, state: EstimateState) { + self.estimate_state = state; } - /// Set the kxq0 register directly - pub fn set_kxq0(&mut self, value: f64) { - self.kxq0 = value; + /// Invalidates HIP after registers from independent histories are merged. + pub(super) fn invalidate_hip(&mut self) { + self.estimate_state = EstimateState::Composite; } - /// Set the kxq1 register directly - pub fn set_kxq1(&mut self, value: f64) { - self.kxq1 = value; + /// Replaces register-derived KxQ values after a bulk register operation. + pub(super) fn restore_kxq(&mut self, kxq0: f64, kxq1: f64) { + self.kxq0 = kxq0; + self.kxq1 = kxq1; } } @@ -331,22 +354,27 @@ impl HipEstimator { /// /// * `lg_config_k`: Log2 of number of registers (must be 4-21) /// * `upper_bound`: Whether computing upper bound (vs lower bound) -/// * `ooo`: Whether sketch is out-of-order (merged/deserialized) +/// * `composite`: Whether the estimate is derived from registers rather than HIP history /// * `num_std_dev`: Number of standard deviations (1, 2, or 3) /// /// # Returns /// /// Relative error factor to apply to estimate -fn get_rel_err(lg_config_k: u8, upper_bound: bool, ooo: bool, num_std_dev: NumStdDev) -> f64 { +fn relative_error( + lg_config_k: u8, + upper_bound: bool, + composite: bool, + num_std_dev: NumStdDev, +) -> f64 { // For lg_k > 12, use analytical formula with RSE factors if lg_config_k > 12 { // RSE factors from Apache DataSketches C++ implementation // HLL_HIP_RSE_FACTOR = sqrt(ln(2)) ≈ 0.8325546 // HLL_NON_HIP_RSE_FACTOR = sqrt((3 * ln(2)) - 1) ≈ 1.03896 - let rse_factor = if ooo { - 1.03896 // Non-HIP (out-of-order) + let rse_factor = if composite { + 1.03896 // Composite } else { - 0.8325546 // HIP (in-order) + 0.8325546 // HIP }; let k = (1 << lg_config_k) as f64; @@ -359,8 +387,8 @@ fn get_rel_err(lg_config_k: u8, upper_bound: bool, ooo: bool, num_std_dev: NumSt // Tables are indexed by: ((lg_k - 4) * 3) + (num_std_dev - 1) let idx = ((lg_config_k as usize) - 4) * 3 + ((num_std_dev as usize) - 1); - // Select the appropriate table based on ooo and upper_bound flags - match (ooo, upper_bound) { + // Select the appropriate table based on estimator and bound direction. + match (composite, upper_bound) { (false, false) => HIP_LB[idx], // Case 0: HIP, Lower Bound (false, true) => HIP_UB[idx], // Case 1: HIP, Upper Bound (true, false) => NON_HIP_LB[idx], // Case 2: Non-HIP, Lower Bound @@ -435,7 +463,7 @@ static HIP_UB: [f64; 27] = [ -0.037896952, //12 ]; -/// Non-HIP (out-of-order) Lower Bound errors for lg_k 4-12, std_dev 1-3 +/// Composite (non-HIP) lower-bound errors for lg_k 4-12, std_dev 1-3. /// Q(.84134), Q(.97725), Q(.99865) quantiles static NON_HIP_LB: [f64; 27] = [ 0.254409839, @@ -467,7 +495,7 @@ static NON_HIP_LB: [f64; 27] = [ 0.049677541, //12 ]; -/// Non-HIP (out-of-order) Upper Bound errors for lg_k 4-12, std_dev 1-3 +/// Composite (non-HIP) upper-bound errors for lg_k 4-12, std_dev 1-3. /// Q(.15866), Q(.02275), Q(.00135) quantiles static NON_HIP_UB: [f64; 27] = [ -0.256980172, @@ -510,7 +538,7 @@ mod tests { #[test] fn lower_bound_is_clamped_to_the_non_zero_register_count() { let lg_config_k = 4; - let mut estimator = HipEstimator::new(lg_config_k); + let mut estimator = Estimator::new(lg_config_k); for _ in 0..8 { estimator.update(lg_config_k, 0, 1); } @@ -524,7 +552,7 @@ mod tests { #[test] fn lower_bound_is_clamped_to_config_k_when_every_register_is_hit() { let lg_config_k = 4; - let mut estimator = HipEstimator::new(lg_config_k); + let mut estimator = Estimator::new(lg_config_k); for _ in 0..16 { estimator.update(lg_config_k, 0, 1); } @@ -537,17 +565,17 @@ mod tests { #[test] fn test_estimator_initialization() { - let est = HipEstimator::new(10); // 1024 registers + let est = Estimator::new(10); // 1024 registers assert_eq!(est.hip_accum(), 0.0); assert_eq!(est.kxq0(), 1024.0); // All zeros = 1.0 each assert_eq!(est.kxq1(), 0.0); - assert!(!est.is_out_of_order()); + assert!(!est.uses_composite_estimate()); } #[test] fn test_estimator_update() { - let mut est = HipEstimator::new(8); // 256 registers + let mut est = Estimator::new(8); // 256 registers // Update from 0 to 10 est.update(8, 0, 10); @@ -562,7 +590,7 @@ mod tests { #[test] fn test_kxq_split() { - let mut est = HipEstimator::new(8); + let mut est = Estimator::new(8); // Update to value < 32 (goes to kxq0) est.update(8, 0, 10); @@ -582,36 +610,21 @@ mod tests { } #[test] - fn test_out_of_order_flag() { - let mut est = HipEstimator::new(10); + fn test_invalidate_hip() { + let mut est = Estimator::new(10); - // Normal update - est.update(8, 0, 5); + est.update(10, 0, 5); let hip_normal = est.hip_accum(); assert_that!(hip_normal, gt(0.0)); - // Set out-of-order - est.set_out_of_order(true); - assert!(est.is_out_of_order()); - assert_eq!(est.hip_accum(), 0.0); // HIP invalidated + est.invalidate_hip(); + assert!(est.uses_composite_estimate()); + assert_eq!(est.hip_accum(), 0.0); - // Update while OOO - HIP should not change, but kxq should + // Register-derived state continues to update after HIP is invalidated. let kxq0_before = est.kxq0(); - est.update(8, 5, 10); - assert_eq!(est.hip_accum(), 0.0); // HIP still 0 - assert_ne!(est.kxq0(), kxq0_before); // kxq changed - } - - #[test] - fn test_setters() { - let mut est = HipEstimator::new(10); - - est.set_hip_accum(123.45); - est.set_kxq0(678.9); - est.set_kxq1(0.0012); - - assert_eq!(est.hip_accum(), 123.45); - assert_eq!(est.kxq0(), 678.9); - assert_eq!(est.kxq1(), 0.0012); + est.update(10, 5, 10); + assert_eq!(est.hip_accum(), 0.0); + assert_ne!(est.kxq0(), kxq0_before); } } diff --git a/datasketches/src/hll/serialization.rs b/datasketches/src/hll/serialization.rs index 9f4eeeb..cacb6cb 100644 --- a/datasketches/src/hll/serialization.rs +++ b/datasketches/src/hll/serialization.rs @@ -29,7 +29,7 @@ pub const EMPTY_FLAG_MASK: u8 = 4; /// /// HLL register arrays have the same layout in compact and updatable images. pub const COMPACT_FLAG_MASK: u8 = 8; -/// Flag indicating out-of-order mode (HIP estimator invalid) +/// Flag indicating that HIP history is unavailable and composite estimation is required. pub const OUT_OF_ORDER_FLAG_MASK: u8 = 16; /// Preamble size for LIST mode (8 bytes = 2 ints) diff --git a/datasketches/src/hll/sketch.rs b/datasketches/src/hll/sketch.rs index 8e924e2..4e0ed71 100644 --- a/datasketches/src/hll/sketch.rs +++ b/datasketches/src/hll/sketch.rs @@ -37,6 +37,7 @@ use crate::hll::array4::AuxFormat; use crate::hll::array6::Array6; use crate::hll::array8::Array8; use crate::hll::container::Container; +use crate::hll::estimator::EstimateState; use crate::hll::hash_set::HashSet; use crate::hll::list::List; use crate::hll::mode::Mode; @@ -487,7 +488,7 @@ fn promote_container_to_array(container: &Container, hll_type: HllType, lg_confi for coupon in container.iter() { array.update(coupon); } - array.set_hip_accum(container.estimate()); + array.restore_estimate_state(EstimateState::Hip(container.estimate())); Mode::Array4(array) } HllType::Hll6 => { @@ -495,7 +496,7 @@ fn promote_container_to_array(container: &Container, hll_type: HllType, lg_confi for coupon in container.iter() { array.update(coupon); } - array.set_hip_accum(container.estimate()); + array.restore_estimate_state(EstimateState::Hip(container.estimate())); Mode::Array6(array) } HllType::Hll8 => { @@ -503,7 +504,7 @@ fn promote_container_to_array(container: &Container, hll_type: HllType, lg_confi for coupon in container.iter() { array.update(coupon); } - array.set_hip_accum(container.estimate()); + array.restore_estimate_state(EstimateState::Hip(container.estimate())); Mode::Array8(array) } } diff --git a/datasketches/src/hll/union.rs b/datasketches/src/hll/union.rs index 5602695..eb98382 100644 --- a/datasketches/src/hll/union.rs +++ b/datasketches/src/hll/union.rs @@ -38,6 +38,7 @@ use crate::hll::HllType; use crate::hll::array4::Array4; use crate::hll::array6::Array6; use crate::hll::array8::Array8; +use crate::hll::estimator::EstimateState; use crate::hll::mode::Mode; /// An HLL union for combining multiple HLL sketches. @@ -424,14 +425,16 @@ fn merge_array_into_array8(dst_array8: &mut Array8, dst_lg_k: u8, src_mode: &Mod } } -/// Extract HIP accumulator from an array mode -fn get_array_hip_accum(mode: &Mode) -> f64 { +/// Extract estimate state from an array mode. +fn get_array_estimate_state(mode: &Mode) -> EstimateState { match mode { - Mode::Array8(src) => src.hip_accum(), - Mode::Array6(src) => src.hip_accum(), - Mode::Array4(src) => src.hip_accum(), + Mode::Array8(src) => src.estimate_state(), + Mode::Array6(src) => src.estimate_state(), + Mode::Array4(src) => src.estimate_state(), Mode::List { .. } | Mode::Set { .. } => { - unreachable!("get_array_hip_accum called with non-array mode; List/Set not supported"); + unreachable!( + "get_array_estimate_state called with non-array mode; List/Set not supported" + ); } } } @@ -520,8 +523,9 @@ fn merge_array_with_downsample(dst: &mut Array8, dst_lg_k: u8, src_mode: &Mode, /// Convert Array8 to a different HLL type /// /// Creates a new sketch with the requested type by copying register values -/// from the Array8 source. Preserves the HIP accumulator. +/// from the Array8 source. Preserves the estimate state. fn convert_array8_to_type(src: &Array8, lg_config_k: u8, target_type: HllType) -> HllSketch { + let estimate_state = src.estimate_state(); match target_type { HllType::Hll8 => HllSketch::from_mode(lg_config_k, Mode::Array8(src.clone())), HllType::Hll6 => { @@ -534,12 +538,7 @@ fn convert_array8_to_type(src: &Array8, lg_config_k: u8, target_type: HllType) - array6.update(coupon); } } - - let src_est = src.estimate(); - let arr6_est = array6.estimate(); - if src_est > arr6_est { - array6.set_hip_accum(src_est); - } + array6.restore_estimate_state(estimate_state); HllSketch::from_mode(lg_config_k, Mode::Array6(array6)) } @@ -552,12 +551,7 @@ fn convert_array8_to_type(src: &Array8, lg_config_k: u8, target_type: HllType) - array4.update(coupon); } } - - let src_est = src.estimate(); - let arr4_est = array4.estimate(); - if src_est > arr4_est { - array4.set_hip_accum(src_est); - } + array4.restore_estimate_state(estimate_state); HllSketch::from_mode(lg_config_k, Mode::Array4(array4)) } @@ -578,11 +572,12 @@ fn copy_array46_via_coupons(dst: &mut Array8, num_registers: usize, get_value: i /// Copy or downsample a source array to create a new Array8 /// /// Directly copies if src_lg_k <= tgt_lg_k, downsamples otherwise. -/// Result is marked as out-of-order and HIP accumulator is preserved. +/// The source estimate state carries over because the result represents the same logical sketch. fn copy_or_downsample(src_mode: &Mode, src_lg_k: u8, tgt_lg_k: u8) -> Array8 { - if src_lg_k <= tgt_lg_k { + let estimate_state = get_array_estimate_state(src_mode); + + let mut result = if src_lg_k <= tgt_lg_k { let mut result = Array8::new(src_lg_k); - let src_hip = get_array_hip_accum(src_mode); match src_mode { Mode::Array8(src) => { @@ -601,12 +596,14 @@ fn copy_or_downsample(src_mode: &Mode, src_lg_k: u8, tgt_lg_k: u8) -> Array8 { } } - result.set_hip_accum(src_hip); result } else { // Downsample from src to tgt let mut result = Array8::new(tgt_lg_k); merge_array_with_downsample(&mut result, tgt_lg_k, src_mode, src_lg_k); result - } + }; + + result.restore_estimate_state(estimate_state); + result } diff --git a/tests-integration/tests/hll_test/bounds.rs b/tests-integration/tests/hll_test/bounds.rs index 159c01f..2dc70da 100644 --- a/tests-integration/tests/hll_test/bounds.rs +++ b/tests-integration/tests/hll_test/bounds.rs @@ -70,7 +70,7 @@ fn hll_mode_lower_bound_matches_cross_language_register_floor() { near(9.946968965192236, 1e-9) ); - // The same floor applies when a union selects the out-of-order estimator. + // The same floor applies when a union selects the composite estimator. let sketch = hll_mode_union(7, HllType::Hll4, 40); assert_eq!(sketch.lower_bound(NumStdDev::Three), 34.0); } diff --git a/tests-integration/tests/hll_test/union.rs b/tests-integration/tests/hll_test/union.rs index b6b8038..7a6409a 100644 --- a/tests-integration/tests/hll_test/union.rs +++ b/tests-integration/tests/hll_test/union.rs @@ -15,18 +15,6 @@ // specific language governing permissions and limitations // under the License. -//! HyperLogLog Union Integration Tests -//! -//! These tests verify the public API behavior of HllUnion, focusing on: -//! * Basic union operations -//! * Mode transitions and mixed-mode unions -//! * Different HLL types and lg_k values -//! * Bounds and statistical properties -//! * Mathematical properties (commutativity, associativity, idempotency) -//! * Reset and reuse patterns -//! -//! This mirrors the testing strategy used in hll_update_test.rs - use datasketches::common::NumStdDev; use datasketches::error::ErrorKind; use datasketches::hll::HllSketch; @@ -59,6 +47,28 @@ fn assert_estimate_within(estimate: f64, expected: f64, relative_error: f64) { ); } +fn assert_same_estimate_and_bounds(actual: &HllSketch, expected: &HllSketch) { + let actual_type = actual.target_type(); + let expected_type = expected.target_type(); + assert_eq!( + actual.estimate(), + expected.estimate(), + "{actual_type:?} and {expected_type:?} estimates", + ); + for num_std_dev in [NumStdDev::One, NumStdDev::Two, NumStdDev::Three] { + assert_eq!( + actual.lower_bound(num_std_dev), + expected.lower_bound(num_std_dev), + "{actual_type:?} and {expected_type:?} lower bounds at {num_std_dev:?}", + ); + assert_eq!( + actual.upper_bound(num_std_dev), + expected.upper_bound(num_std_dev), + "{actual_type:?} and {expected_type:?} upper bounds at {num_std_dev:?}", + ); + } +} + fn serialize_flat_union(first: &HllSketch, second: &HllSketch, third: &HllSketch) -> Vec { let mut union = HllUnion::new(8).unwrap(); union.update(first); @@ -235,6 +245,10 @@ fn test_union_mixed_hll_types() { assert_eq!(result6.target_type(), HllType::Hll6); assert_eq!(result8.target_type(), HllType::Hll8); + // Target types change only the register encoding. + assert_same_estimate_and_bounds(&result4, &result8); + assert_same_estimate_and_bounds(&result6, &result8); + // Should estimate ~7,000 unique values (0-6,999) for (result, type_name) in [ (result4.estimate(), "Hll4"), @@ -553,22 +567,19 @@ fn test_union_associativity() { } #[test] -fn test_union_idempotency() { - // Verify A∪A = A - let mut sketch = HllSketch::new(12, HllType::Hll8).unwrap(); - for i in 0..1000 { - sketch.update(i); - } +fn test_union_repeated_input_is_stable() { + let sketch = make_hll_sketch(HllType::Hll8, 12, 0, 1_000); let mut union = HllUnion::new(12).unwrap(); union.update(&sketch); - let est1 = union.estimate(); + assert_eq!(union.estimate(), sketch.estimate()); - // Union with itself union.update(&sketch); - let est2 = union.estimate(); + let merged_estimate = union.estimate(); + assert_estimate_within(merged_estimate, sketch.estimate(), 0.02); - assert_eq!(est1, est2); + union.update(&sketch); + assert_eq!(union.estimate(), merged_estimate); } #[test] @@ -708,3 +719,29 @@ fn test_union_estimated_size() { union.update(&sketch); assert_eq!(union.estimated_size(), 1120); } + +#[test] +fn test_single_hll_input_preserves_estimate_and_bounds() { + for hll_type in HLL_TYPES { + let sketch = make_hll_sketch(hll_type, 8, 0, 1_000); + let mut union = HllUnion::new(10).unwrap(); + union.update(&sketch); + + assert_eq!(union.estimate(), sketch.estimate(), "{hll_type:?} estimate"); + for target_type in HLL_TYPES { + assert_same_estimate_and_bounds(&union.to_sketch(target_type), &sketch); + } + } +} + +#[test] +fn test_downsampling_single_hll_input_preserves_estimate() { + for hll_type in HLL_TYPES { + let sketch = make_hll_sketch(hll_type, 12, 0, 10_000); + let mut union = HllUnion::new(8).unwrap(); + union.update(&sketch); + + assert_eq!(union.lg_config_k(), 8, "{hll_type:?} lg_config_k"); + assert_eq!(union.estimate(), sketch.estimate(), "{hll_type:?} estimate"); + } +}