diff --git a/CHANGELOG.md b/CHANGELOG.md index cbc6000b..8f254ac5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,10 +6,12 @@ All significant changes to this project will be documented in this file. ### New features +* Add `HllUnion::serialize` to serialize union results directly. * Add immutable `XorFilter` and `XorFilterBuilder` behind the `xor` feature, with 8-, 16-, and 32-bit fingerprints, consuming builder configuration with complete-input construction, bounded seeded retries and compact temporary storage, optional cardinality metadata for imported filters, standard serialization for all three widths (Java-compatible for 8/16-bit filters), and raw parts for application-defined storage. ### Improvements +* `HllUnion` is faster when adopting HLL8 inputs without downsampling or merging multiple dense inputs before querying or serializing the result. * `KllSketch::rank` is faster for repeated queries on compacted sketches. * Repeated scalar `TDigest::rank` queries are faster, including small digests and data with repeated values. * `TDigest` and `TDigestMut` CDF and PMF batches are faster for many split points; batches with sparse split points keep the original binary-search behavior. diff --git a/benchmarks/hll/mod.rs b/benchmarks/hll/mod.rs index 1dea33be..9e440b27 100644 --- a/benchmarks/hll/mod.rs +++ b/benchmarks/hll/mod.rs @@ -15,4 +15,6 @@ // specific language governing permissions and limitations // under the License. +mod serde; +mod union; mod update; diff --git a/benchmarks/hll/serde.rs b/benchmarks/hll/serde.rs new file mode 100644 index 00000000..8bafba49 --- /dev/null +++ b/benchmarks/hll/serde.rs @@ -0,0 +1,41 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use datasketches::hll::HllType; +use datasketches::hll::HllUnion; +use divan::Bencher; +use divan::black_box; + +fn union(rows: u64) -> HllUnion { + let mut union = HllUnion::new(18).unwrap(); + for value in 0..rows { + union.update_value(value); + } + union +} + +#[divan::bench(args = [0, 4, 100, 80_000])] +fn serialize_union(bencher: Bencher, rows: u64) { + let union = union(rows); + bencher.bench_local(|| black_box(&union).serialize()); +} + +#[divan::bench(args = [0, 4, 100, 80_000])] +fn serialize_union_via_sketch(bencher: Bencher, rows: u64) { + let union = union(rows); + bencher.bench_local(|| black_box(&union).to_sketch(HllType::Hll8).serialize()); +} diff --git a/benchmarks/hll/union.rs b/benchmarks/hll/union.rs new file mode 100644 index 00000000..c6be59c8 --- /dev/null +++ b/benchmarks/hll/union.rs @@ -0,0 +1,112 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use datasketches::hll::HllSketch; +use datasketches::hll::HllType; +use datasketches::hll::HllUnion; +use divan::Bencher; +use divan::black_box; + +fn partials(lg_k: u8, count: usize, rows: usize) -> Vec { + (0..count) + .map(|partial| { + let mut sketch = HllSketch::new(lg_k, HllType::Hll8).unwrap(); + for value in partial * rows..(partial + 1) * rows { + sketch.update(value as u64); + } + sketch + }) + .collect() +} + +#[divan::bench(args = [12, 18])] +fn empty_from_dense(bencher: Bencher, lg_k: u8) { + let source = partials(lg_k, 1, 1 << lg_k).pop().unwrap(); + bencher.bench_local(|| { + let mut union = HllUnion::new(lg_k).unwrap(); + union.update(black_box(&source)); + black_box(union) + }); +} + +#[divan::bench(args = [12, 18])] +fn recreated_small_partials(bencher: Bencher, lg_k: u8) { + let partials = partials(lg_k, 400, 200); + bencher.bench_local(|| { + let mut sketch = HllSketch::new(lg_k, HllType::Hll8).unwrap(); + for partial in black_box(&partials) { + let mut union = HllUnion::new(lg_k).unwrap(); + union.update(&sketch); + union.update(partial); + sketch = union.to_sketch(HllType::Hll8); + } + black_box(sketch.estimate()); + black_box(sketch) + }); +} + +#[divan::bench(args = [12, 18])] +fn persistent_small_partials(bencher: Bencher, lg_k: u8) { + let partials = partials(lg_k, 400, 200); + bencher.bench_local(|| { + let mut union = HllUnion::new(lg_k).unwrap(); + for partial in black_box(&partials) { + union.update(partial); + } + black_box(union.estimate()); + black_box(union.to_sketch(HllType::Hll8)) + }); +} + +#[divan::bench(args = [12, 18])] +fn dense_partials(bencher: Bencher, lg_k: u8) { + let partials = partials(lg_k, 16, (1 << lg_k) / 4); + bencher.bench_local(|| { + let mut union = HllUnion::new(lg_k).unwrap(); + for partial in black_box(&partials) { + union.update(partial); + } + black_box(union.estimate()); + black_box(union.to_sketch(HllType::Hll8)) + }); +} + +#[divan::bench(args = [12, 18])] +fn dense_partials_with_queries(bencher: Bencher, lg_k: u8) { + let partials = partials(lg_k, 16, (1 << lg_k) / 4); + bencher.bench_local(|| { + let mut union = HllUnion::new(lg_k).unwrap(); + for partial in black_box(&partials) { + union.update(partial); + black_box(union.estimate()); + } + black_box(union.to_sketch(HllType::Hll8)) + }); +} + +#[divan::bench(args = [12, 18])] +fn downsampled_dense_partials(bencher: Bencher, lg_k: u8) { + let partials = partials(lg_k + 1, 16, (1 << lg_k) / 2); + bencher.bench_local(|| { + let mut union = HllUnion::new(lg_k).unwrap(); + for partial in black_box(&partials) { + union.update(partial); + } + black_box(union.estimate()); + black_box(union.to_sketch(HllType::Hll8)) + }); +} diff --git a/datasketches/src/hll/array8.rs b/datasketches/src/hll/array8.rs index 1269d171..65861044 100644 --- a/datasketches/src/hll/array8.rs +++ b/datasketches/src/hll/array8.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::sync::OnceLock; + use crate::codec::SketchBytes; use crate::codec::SketchSlice; use crate::codec::assert::insufficient_data; @@ -33,14 +35,30 @@ use crate::hll::serialization::TGT_HLL8; use crate::hll::serialization::encode_mode_byte; /// Registers stored one per byte. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone)] pub struct Array8 { lg_config_k: u8, bytes: Box<[u8]>, + /// Absent estimation data means a bulk merge invalidated HIP and the register-derived fields. + /// Shared queries initialize it once; exclusive updates can maintain or invalidate it. + estimation: OnceLock, +} + +/// Zero-register count and estimator used for cardinality queries. +#[derive(Debug, Clone, PartialEq)] +struct Estimation { num_zeros: u32, estimator: Estimator, } +impl PartialEq for Array8 { + fn eq(&self, other: &Self) -> bool { + self.lg_config_k == other.lg_config_k + && self.bytes == other.bytes + && self.estimation() == other.estimation() + } +} + impl Array8 { pub fn new(lg_config_k: u8) -> Self { let k = 1 << lg_config_k; @@ -48,8 +66,10 @@ impl Array8 { Self { lg_config_k, bytes: vec![0u8; k as usize].into_boxed_slice(), - num_zeros: k, - estimator: Estimator::new(lg_config_k), + estimation: OnceLock::from(Estimation { + num_zeros: k, + estimator: Estimator::new(lg_config_k), + }), } } @@ -71,39 +91,58 @@ impl Array8 { let old_value = self.get(slot); if new_value > old_value { - self.estimator - .update(self.lg_config_k, old_value, new_value); - - self.put(slot, new_value); - - if old_value == 0 { - self.num_zeros -= 1; + if let Some(estimation) = self.estimation.get_mut() { + estimation + .estimator + .update(self.lg_config_k, old_value, new_value); + if old_value == 0 { + estimation.num_zeros -= 1; + } } + self.put(slot, new_value); } } pub fn estimate(&self) -> f64 { + let Estimation { + num_zeros, + estimator, + } = self.estimation(); // 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) + estimator.estimate(self.lg_config_k, 0, *num_zeros) } pub fn composite_estimate(&self) -> f64 { - self.estimator - .composite_estimate(self.lg_config_k, 0, self.num_zeros) + let Estimation { + num_zeros, + estimator, + } = self.estimation(); + estimator.composite_estimate(self.lg_config_k, 0, *num_zeros) } pub fn upper_bound(&self, num_std_dev: NumStdDev) -> f64 { - self.estimator - .upper_bound(self.lg_config_k, 0, self.num_zeros, num_std_dev) + let Estimation { + num_zeros, + estimator, + } = self.estimation(); + estimator.upper_bound(self.lg_config_k, 0, *num_zeros, num_std_dev) } pub fn lower_bound(&self, num_std_dev: NumStdDev) -> f64 { - self.estimator - .lower_bound(self.lg_config_k, 0, self.num_zeros, num_std_dev) + let Estimation { + num_zeros, + estimator, + } = self.estimation(); + estimator.lower_bound(self.lg_config_k, 0, *num_zeros, num_std_dev) } pub fn is_empty(&self) -> bool { - self.num_zeros == (1 << self.lg_config_k) + if let Some(estimation) = self.estimation.get() { + return estimation.num_zeros == (1 << self.lg_config_k); + } + // Union checks emptiness between merges. A nonzero register answers that without + // rebuilding all estimator fields and defeating deferred reconstruction. + self.bytes.iter().all(|&value| value == 0) } pub fn values(&self) -> &[u8] { @@ -115,22 +154,31 @@ impl Array8 { } pub fn estimate_state(&self) -> EstimateState { - self.estimator.estimate_state() + self.estimation + .get() + .map_or(EstimateState::Composite, |estimation| { + estimation.estimator.estimate_state() + }) } pub fn restore_estimate_state(&mut self, state: EstimateState) { - self.estimator.restore_estimate_state(state); + if let Some(estimation) = self.estimation.get_mut() { + estimation.estimator.restore_estimate_state(state); + } else if let EstimateState::Hip(_) = state { + let mut estimation = self.rebuild_estimation(); + estimation.estimator.restore_estimate_state(state); + self.estimation = OnceLock::from(estimation); + } } - /// Bypasses estimator updates. Call `rebuild_estimator_from_registers` after all writes. + /// Bypasses estimator updates. Call `invalidate_estimator` after all writes. pub fn set_register(&mut self, slot: usize, value: u8) { self.bytes[slot] = value; } - /// Rebuilds register-derived caches and invalidates HIP after bulk writes. - pub fn rebuild_estimator_from_registers(&mut self) { - self.rebuild_cached_values(); - self.estimator.invalidate_hip(); + /// Invalidates HIP and defers rebuilding register-derived fields until a query needs them. + pub fn invalidate_estimator(&mut self) { + self.estimation.take(); } /// Merges registers and invalidates HIP because their update order is unavailable. @@ -145,8 +193,7 @@ impl Array8 { self.bytes[i] = self.bytes[i].max(val); } - self.rebuild_cached_values(); - self.estimator.invalidate_hip(); + self.invalidate_estimator(); } /// Register values come from a separate hash word, so folding slot indices needs no @@ -169,13 +216,15 @@ impl Array8 { self.bytes[dst_slot] = self.bytes[dst_slot].max(val); } - self.rebuild_cached_values(); - self.estimator.invalidate_hip(); + self.invalidate_estimator(); } - /// Rebuilds register-derived caches without changing the estimate state. - fn rebuild_cached_values(&mut self) { - self.num_zeros = self.bytes.iter().filter(|&&v| v == 0).count() as u32; + fn estimation(&self) -> &Estimation { + self.estimation.get_or_init(|| self.rebuild_estimation()) + } + + fn rebuild_estimation(&self) -> Estimation { + let num_zeros = self.bytes.iter().filter(|&&v| v == 0).count() as u32; let mut kxq0_sum = 0.0; let mut kxq1_sum = 0.0; @@ -190,7 +239,13 @@ impl Array8 { } } - self.estimator.restore_kxq(kxq0_sum, kxq1_sum); + let mut estimator = Estimator::new(self.lg_config_k); + estimator.restore_kxq(kxq0_sum, kxq1_sum); + estimator.invalidate_hip(); + Estimation { + num_zeros, + estimator, + } } pub fn deserialize_registers( @@ -236,12 +291,15 @@ impl Array8 { Ok(Self { lg_config_k, bytes: data.into_boxed_slice(), - num_zeros, - estimator, + estimation: OnceLock::from(Estimation { + num_zeros, + estimator, + }), }) } pub fn serialize(&self, lg_config_k: u8) -> Vec { + let estimation = self.estimation(); let k = 1 << lg_config_k; let total_size = HLL_PREAMBLE_SIZE + k as usize; let mut bytes = SketchBytes::with_capacity(total_size); @@ -253,7 +311,7 @@ impl Array8 { bytes.write_u8(0); // unused for HLL mode let mut flags = 0u8; - if self.estimator.uses_composite_estimate() { + if estimation.estimator.uses_composite_estimate() { flags |= OUT_OF_ORDER_FLAG_MASK; } bytes.write_u8(flags); @@ -263,12 +321,12 @@ impl Array8 { bytes.write_u8(encode_mode_byte(CUR_MODE_HLL, TGT_HLL8)); - bytes.write_f64_le(self.estimator.hip_accum()); - bytes.write_f64_le(self.estimator.kxq0()); - bytes.write_f64_le(self.estimator.kxq1()); + bytes.write_f64_le(estimation.estimator.hip_accum()); + bytes.write_f64_le(estimation.estimator.kxq0()); + bytes.write_f64_le(estimation.estimator.kxq1()); // Write num_at_cur_min (num_zeros for Array8) - bytes.write_u32_le(self.num_zeros); + bytes.write_u32_le(estimation.num_zeros); // Write aux_count (always 0 for Array8) bytes.write_u32_le(0); @@ -398,11 +456,11 @@ mod tests { arr.update(Coupon::pack(1, 50)); // value >= 32, goes to kxq1 // Initial kxq0 = 256 (all zeros = 1.0 each) - assert_that!(arr.estimator.kxq0(), lt(256.0)); + assert_that!(arr.estimation().estimator.kxq0(), lt(256.0)); // kxq1 should have a positive value (from 1/2^50) - assert_that!(arr.estimator.kxq1(), gt(0.0)); - assert_that!(arr.estimator.kxq1(), lt(1e-10)); + assert_that!(arr.estimation().estimator.kxq1(), gt(0.0)); + assert_that!(arr.estimation().estimator.kxq1(), lt(1e-10)); } #[test] @@ -442,9 +500,9 @@ mod tests { assert_eq!(dst.get(3), 40, "dst[3] got new value"); // Bulk merges require composite estimation. - assert!(dst.estimator.uses_composite_estimate()); + assert!(dst.estimation().estimator.uses_composite_estimate()); - assert_eq!(dst.num_zeros, 12); + assert_eq!(dst.estimation().num_zeros, 12); } #[test] @@ -468,7 +526,7 @@ mod tests { assert_eq!(dst.get(1), 30, "dst[1] = max(20, 18, 30)"); // Bulk merges require composite estimation. - assert!(dst.estimator.uses_composite_estimate()); + assert!(dst.estimation().estimator.uses_composite_estimate()); } #[test] @@ -490,18 +548,28 @@ mod tests { } #[test] - fn test_rebuild_cached_values() { + fn test_rebuild_estimation() { let mut arr = Array8::new(4); // 16 slots arr.put(0, 10); arr.put(1, 20); arr.put(2, 30); - arr.num_zeros = 999; - - arr.rebuild_cached_values(); + arr.put(3, 31); + arr.put(4, 32); + arr.put(5, 63); + arr.invalidate_estimator(); - assert_eq!(arr.num_zeros, 13); + let estimation = arr.estimation(); + assert_eq!(estimation.num_zeros, 10); + assert_eq!( + estimation.estimator.kxq0(), + 10.0 + 2_f64.powi(-10) + 2_f64.powi(-20) + 2_f64.powi(-30) + 2_f64.powi(-31) + ); + assert_eq!( + estimation.estimator.kxq1(), + 2_f64.powi(-32) + 2_f64.powi(-63) + ); } #[test] diff --git a/datasketches/src/hll/estimator.rs b/datasketches/src/hll/estimator.rs index a72fcca0..631d31ff 100644 --- a/datasketches/src/hll/estimator.rs +++ b/datasketches/src/hll/estimator.rs @@ -42,7 +42,7 @@ pub enum EstimateState { /// precision when large and small contributions coexist. #[derive(Debug, Clone, PartialEq)] pub struct Estimator { - estimate_state: EstimateState, + state: EstimateState, /// KxQ register for values < 32 (larger inverse powers) kxq0: f64, /// KxQ register for values >= 32 (tiny inverse powers) @@ -53,7 +53,7 @@ impl Estimator { pub fn new(lg_config_k: u8) -> Self { let k = 1 << lg_config_k; Self { - estimate_state: EstimateState::Hip(0.0), + state: EstimateState::Hip(0.0), kxq0: k as f64, // All registers start at 0, so kxq0 = k * (1/2^0) = k kxq1: 0.0, } @@ -61,16 +61,12 @@ impl Estimator { /// Restores estimator fields from an HLL serialization preamble. pub fn from_serialized(hip_accum: f64, kxq0: f64, kxq1: f64, out_of_order: bool) -> Self { - let estimate_state = if out_of_order { + let state = if out_of_order { EstimateState::Composite } else { EstimateState::Hip(hip_accum) }; - Self { - estimate_state, - kxq0, - kxq1, - } + Self { state, kxq0, kxq1 } } /// Applies a register increase; requires `new_value > old_value`. @@ -78,7 +74,7 @@ impl Estimator { let k = (1 << lg_config_k) as f64; // HIP must use the KxQ sums from before this register update. - if let EstimateState::Hip(hip_accum) = &mut self.estimate_state { + if let EstimateState::Hip(hip_accum) = &mut self.state { *hip_accum += k / (self.kxq0 + self.kxq1); } @@ -103,7 +99,7 @@ impl Estimator { /// For `Array6`/`Array8`, pass zero for `cur_min` and the zero-register count for /// `num_at_cur_min`. `Array4` tracks its actual minimum and its count. pub fn estimate(&self, lg_config_k: u8, cur_min: u8, num_at_cur_min: u32) -> f64 { - match self.estimate_state { + match self.state { EstimateState::Hip(hip_accum) => hip_accum, EstimateState::Composite => { self.composite_estimate(lg_config_k, cur_min, num_at_cur_min) @@ -234,7 +230,7 @@ impl Estimator { } pub fn hip_accum(&self) -> f64 { - match self.estimate_state { + match self.state { EstimateState::Hip(hip_accum) => hip_accum, EstimateState::Composite => 0.0, } @@ -249,21 +245,21 @@ impl Estimator { } pub fn uses_composite_estimate(&self) -> bool { - matches!(self.estimate_state, EstimateState::Composite) + matches!(self.state, EstimateState::Composite) } pub fn estimate_state(&self) -> EstimateState { - self.estimate_state + self.state } /// Restores estimate state after copying or transforming the same logical sketch. pub fn restore_estimate_state(&mut self, state: EstimateState) { - self.estimate_state = state; + self.state = state; } /// Invalidates HIP after registers from independent histories are merged. pub fn invalidate_hip(&mut self) { - self.estimate_state = EstimateState::Composite; + self.state = EstimateState::Composite; } /// Replaces register-derived KxQ values after a bulk register operation. diff --git a/datasketches/src/hll/union.rs b/datasketches/src/hll/union.rs index b245fa71..f942fc00 100644 --- a/datasketches/src/hll/union.rs +++ b/datasketches/src/hll/union.rs @@ -166,15 +166,10 @@ impl HllUnion { match self.gadget.mode() { Mode::Array8(old_gadget) => { - merge_array_with_downsample( - &mut new_array, - src_lg_k, - &Mode::Array8(old_gadget.clone()), - dst_lg_k, - ); + new_array.merge_array_with_downsample(old_gadget.values(), dst_lg_k); } _ => { - unreachable!("gadget mode changed unexpectedly; should never be Array4/Array6") + unreachable!("gadget mode changed unexpectedly; should never be Array4/Array6"); } } @@ -186,7 +181,7 @@ impl HllUnion { merge_array_into_array8(dst_array, dst_lg_k, src_mode, src_lg_k); } _ => { - unreachable!("gadget mode changed unexpectedly; should never be Array4/Array6") + unreachable!("gadget mode changed unexpectedly; should never be Array4/Array6"); } } } @@ -249,6 +244,29 @@ impl HllUnion { } } + /// Serializes the union's result as an [`HllType::Hll8`] sketch. + /// + /// Produces the same bytes as `to_sketch(HllType::Hll8).serialize()` without copying an + /// intermediate sketch. The image retains the effective `lg_k` and estimate, but not the + /// union's configured maximum `lg_k`. It can be read with [`HllSketch::deserialize`]. + /// + /// # Examples + /// + /// ``` + /// use datasketches::hll::HllSketch; + /// use datasketches::hll::HllUnion; + /// + /// let mut union = HllUnion::new(12).unwrap(); + /// union.update_value("apple"); + /// let mut partial = HllSketch::deserialize(&union.serialize()).unwrap(); + /// partial.update("banana"); + /// union.update(&partial); + /// assert!((union.estimate() - 2.0).abs() < 0.01); + /// ``` + pub fn serialize(&self) -> Vec { + self.gadget.serialize() + } + /// Returns the union's current effective `lg_k`. pub fn lg_config_k(&self) -> u8 { self.gadget.lg_config_k() @@ -468,7 +486,7 @@ fn merge_array46_same_lgk(dst: &mut Array8, num_registers: usize, get_value: imp dst.set_register(slot, val); } } - dst.rebuild_estimator_from_registers(); + dst.invalidate_estimator(); } /// Merging independent register histories invalidates the HIP accumulator. @@ -506,7 +524,7 @@ fn merge_array46_with_downsample( } } } - dst.rebuild_estimator_from_registers(); + dst.invalidate_estimator(); } /// Multiple source registers map to each destination register via masking. @@ -586,17 +604,17 @@ fn copy_or_downsample(src_mode: &Mode, src_lg_k: u8, tgt_lg_k: u8) -> Array8 { 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); - match src_mode { - Mode::Array8(src) => { - result.merge_array_same_lgk(src.values()); - } + Mode::Array8(src) => src.clone(), Mode::Array6(src) => { + let mut result = Array8::new(src_lg_k); copy_array46_via_coupons(&mut result, src.num_registers(), |slot| src.get(slot)); + result } Mode::Array4(src) => { + let mut result = Array8::new(src_lg_k); copy_array46_via_coupons(&mut result, src.num_registers(), |slot| src.get(slot)); + result } Mode::List { .. } | Mode::Set { .. } => { unreachable!( @@ -604,8 +622,6 @@ fn copy_or_downsample(src_mode: &Mode, src_lg_k: u8, tgt_lg_k: u8) -> Array8 { ); } } - - result } else { let mut result = Array8::new(tgt_lg_k); merge_array_with_downsample(&mut result, tgt_lg_k, src_mode, src_lg_k); diff --git a/tests-integration/tests/hll_test/union.rs b/tests-integration/tests/hll_test/union.rs index c61a1716..5f807d8e 100644 --- a/tests-integration/tests/hll_test/union.rs +++ b/tests-integration/tests/hll_test/union.rs @@ -745,3 +745,116 @@ fn test_downsampling_single_hll_input_preserves_estimate() { assert_eq!(union.estimate(), sketch.estimate(), "{hll_type:?} estimate"); } } + +#[test] +fn test_copied_hll8_input_preserves_subsequent_updates() { + for lg_max_k in [10, 12] { + for merged in [false, true] { + let mut source = make_hll_sketch(HllType::Hll8, 10, 0, 2_000); + if merged { + let mut union = HllUnion::new(10).unwrap(); + union.update(&source); + union.update(&make_hll_sketch(HllType::Hll8, 10, 1_000, 3_000)); + source = union.to_sketch(HllType::Hll8); + } + let mut union = HllUnion::new(lg_max_k).unwrap(); + union.update(&source); + let mut expected = source.clone(); + + // Copying must retain both the estimate history and the caches used by later updates. + for value in 2_000..5_000_u64 { + union.update_value(value); + expected.update(value); + } + let result = union.to_sketch(HllType::Hll8); + assert_same_estimate_and_bounds(&result, &expected); + assert_eq!(result.serialize(), expected.serialize()); + assert_eq!(union.lg_config_k(), source.lg_config_k()); + } + } +} + +#[test] +fn test_union_query_timing_does_not_change_mixed_updates() { + let mut queried = HllUnion::new(12).unwrap(); + let mut deferred = queried.clone(); + let inputs = [ + (HllType::Hll8, 12), + (HllType::Hll4, 12), + (HllType::Hll6, 14), + (HllType::Hll8, 10), + (HllType::Hll4, 8), + ]; + for (index, (hll_type, lg_k)) in inputs.into_iter().enumerate() { + let start = index as u64 * 10_000; + let source = make_hll_sketch(hll_type, lg_k, start, start + 5_000); + queried.update(&source); + deferred.update(&source); + assert!(!deferred.is_empty()); + + // Queries refresh one accumulator between merges; the other keeps accepting updates. + assert_same_estimate_and_bounds( + &queried.to_sketch(HllType::Hll8), + &queried.to_sketch(HllType::Hll4), + ); + let restored = HllSketch::deserialize(&queried.serialize()).unwrap(); + assert_eq!(restored.composite_estimate(), queried.composite_estimate()); + for value in start + 5_000..start + 5_100 { + queried.update_value(value); + deferred.update_value(value); + } + } + + assert_eq!(queried.serialize(), deferred.serialize()); + assert_same_estimate_and_bounds( + &queried.to_sketch(HllType::Hll8), + &deferred.to_sketch(HllType::Hll8), + ); + + // A copied result can be merged again before either its source or the copy is queried. + let source = make_hll_sketch(HllType::Hll8, 8, 50_000, 55_000); + deferred.update(&source); + let result = deferred.to_sketch(HllType::Hll8); + let mut copied = HllUnion::new(12).unwrap(); + copied.update(&result); + for value in 55_000..56_000_u64 { + copied.update_value(value); + deferred.update_value(value); + } + assert_eq!(copied.serialize(), deferred.serialize()); +} + +#[test] +fn test_union_supports_shared_queries_after_dense_merges() { + let mut union = HllUnion::new(12).unwrap(); + for index in 0..3 { + union.update(&make_hll_sketch( + HllType::Hll8, + 12, + index * 5_000, + (index + 1) * 5_000, + )); + } + let expected = union.to_sketch(HllType::Hll8); + let bytes = expected.serialize(); + let barrier = std::sync::Barrier::new(4); + + std::thread::scope(|scope| { + for _ in 0..4 { + scope.spawn(|| { + barrier.wait(); + assert_eq!(union.estimate(), expected.estimate()); + assert_eq!(union.composite_estimate(), expected.composite_estimate()); + assert_eq!( + union.lower_bound(NumStdDev::Two), + expected.lower_bound(NumStdDev::Two) + ); + assert_eq!( + union.upper_bound(NumStdDev::Two), + expected.upper_bound(NumStdDev::Two) + ); + assert_eq!(union.serialize(), bytes); + }); + } + }); +} diff --git a/tests-integration/tests/serde_tests/hll.rs b/tests-integration/tests/serde_tests/hll.rs index 6f4485b2..ed0dd41b 100644 --- a/tests-integration/tests/serde_tests/hll.rs +++ b/tests-integration/tests/serde_tests/hll.rs @@ -21,6 +21,7 @@ use std::path::PathBuf; use datasketches::hash::value::natural_extend; use datasketches::hll::HllSketch; use datasketches::hll::HllType; +use datasketches::hll::HllUnion; use googletest::assert_that; use googletest::prelude::all; use googletest::prelude::ge; @@ -30,6 +31,40 @@ use googletest::prelude::near; use crate::serialization_test_data; +#[test] +fn union_serialization_preserves_result_and_subsequent_updates() { + for hll_type in [HllType::Hll4, HllType::Hll6, HllType::Hll8] { + for count in [0, 4, 100, 10_000_u64] { + let mut source = HllSketch::new(12, hll_type).unwrap(); + for value in 0..count { + source.update(value); + } + for lg_max_k in [8, 12, 14] { + let mut union = HllUnion::new(lg_max_k).unwrap(); + union.update(&source); + let bytes = union.serialize(); + assert_eq!(bytes, union.to_sketch(HllType::Hll8).serialize()); + + let mut restored = HllSketch::deserialize(&bytes).unwrap(); + assert_eq!(restored.target_type(), HllType::Hll8); + assert_eq!(restored.lg_config_k(), union.lg_config_k()); + assert_eq!(restored.estimate(), union.estimate()); + + for value in count..count + 1_000 { + union.update_value(value); + restored.update(value); + } + assert_eq!(union.to_sketch(HllType::Hll8), restored); + + union.reset(); + let empty = HllSketch::deserialize(&union.serialize()).unwrap(); + assert!(empty.is_empty()); + assert_eq!(empty.lg_config_k(), lg_max_k); + } + } + } +} + fn test_sketch_file(path: PathBuf, expected_cardinality: usize, expected_lg_k: u8) { let expected = expected_cardinality as f64;