You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datasketches.apache.org by ti...@apache.org on 2026/07/24 23:33:37 UTC
(datasketches-rust) branch main updated: feat: support tuple intersection (#154)
This is an automated email from the ASF dual-hosted git repository.
tisonkun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datasketches-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 9a409c8 feat: support tuple intersection (#154)
9a409c8 is described below
commit 9a409c8df6c439617267b10436d8bb7a7038b7aa
Author: Yijun Zhao <ar...@gmail.com>
AuthorDate: Sat Jul 25 07:33:31 2026 +0800
feat: support tuple intersection (#154)
Co-authored-by: tison <wa...@gmail.com>
---
datasketches/src/lib.rs | 1 -
datasketches/src/theta/hash_table.rs | 17 +-
datasketches/src/theta/intersection.rs | 185 ++----------
datasketches/src/thetacommon/intersection.rs | 388 ++++++++++++++++++++++++++
datasketches/src/thetacommon/mod.rs | 1 +
datasketches/src/tuple/hash_table.rs | 7 +
datasketches/src/tuple/intersection.rs | 384 +++++++++++++++++++++++++
datasketches/src/tuple/mod.rs | 2 +
datasketches/tests/tuple_intersection_test.rs | 343 +++++++++++++++++++++++
datasketches/tests/tuple_sketch_test.rs | 48 ++--
10 files changed, 1187 insertions(+), 189 deletions(-)
diff --git a/datasketches/src/lib.rs b/datasketches/src/lib.rs
index deeb271..9c69ba0 100644
--- a/datasketches/src/lib.rs
+++ b/datasketches/src/lib.rs
@@ -46,7 +46,6 @@ pub mod tdigest;
#[cfg(feature = "theta")]
pub mod theta;
#[cfg(any(feature = "theta", feature = "tuple"))]
-#[allow(dead_code)] // some utilities are only used for certain sketches
pub mod thetacommon;
#[cfg(feature = "tuple")]
pub mod tuple;
diff --git a/datasketches/src/theta/hash_table.rs b/datasketches/src/theta/hash_table.rs
index 5dd83ee..8bd2aa9 100644
--- a/datasketches/src/theta/hash_table.rs
+++ b/datasketches/src/theta/hash_table.rs
@@ -75,16 +75,6 @@ impl ThetaHashTable {
}
})
}
-
- /// Get iterator over entries.
- pub fn iter(&self) -> impl Iterator<Item = u64> + '_ {
- self.iter_entries().map(RawHashTableEntry::hash)
- }
-
- /// Returns true if the given hash exists in the table.
- pub fn contains_hash(&self, hash: u64) -> bool {
- self.get_entry(hash).is_some()
- }
}
#[cfg(test)]
@@ -97,6 +87,13 @@ mod tests {
use crate::thetacommon::hash_table::starting_sub_multiple;
use crate::thetacommon::hash_table::starting_theta_from_sampling_probability;
+ impl ThetaHashTable {
+ /// Get iterator over entries.
+ pub fn iter(&self) -> impl Iterator<Item = u64> + '_ {
+ self.iter_entries().map(RawHashTableEntry::hash)
+ }
+ }
+
#[test]
fn test_new_hash_table() {
let table = ThetaHashTable::new(8, ResizeFactor::X8, 1.0, DEFAULT_UPDATE_SEED);
diff --git a/datasketches/src/theta/intersection.rs b/datasketches/src/theta/intersection.rs
index 6383643..9d317f0 100644
--- a/datasketches/src/theta/intersection.rs
+++ b/datasketches/src/theta/intersection.rs
@@ -15,14 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-use crate::common::ResizeFactor;
use crate::error::Error;
use crate::hash::DEFAULT_UPDATE_SEED;
use crate::theta::CompactThetaSketch;
use crate::theta::ThetaSketchView;
-use crate::theta::hash_table::ThetaHashTable;
-use crate::thetacommon::constants::HASH_TABLE_REBUILD_THRESHOLD;
-use crate::thetacommon::constants::MAX_THETA;
+use crate::theta::hash_table::ThetaEntry;
+use crate::thetacommon::intersection::RawThetaIntersection;
+use crate::thetacommon::intersection::RawThetaIntersectionPolicy;
/// Stateful intersection operator for Theta sketches.
///
@@ -30,24 +29,21 @@ use crate::thetacommon::constants::MAX_THETA;
/// [`has_result`](Self::has_result) to check.
#[derive(Debug)]
pub struct ThetaIntersection {
- is_valid: bool,
- table: ThetaHashTable,
+ raw: RawThetaIntersection<ThetaEntry, NoopIntersectionPolicy>,
+}
+
+#[derive(Debug)]
+struct NoopIntersectionPolicy;
+
+impl RawThetaIntersectionPolicy<ThetaEntry> for NoopIntersectionPolicy {
+ fn merge(&self, _existing: &mut ThetaEntry, _incoming: ThetaEntry) {}
}
impl ThetaIntersection {
/// Creates a new intersection operator for the given `seed`.
pub fn new(seed: u64) -> Self {
Self {
- is_valid: false,
- table: ThetaHashTable::from_raw_parts(
- 0,
- 0,
- ResizeFactor::X1,
- 1.0,
- MAX_THETA,
- seed,
- false,
- ),
+ raw: RawThetaIntersection::new(seed, NoopIntersectionPolicy),
}
}
@@ -62,144 +58,12 @@ impl ThetaIntersection {
/// and every update can reduce the current set to leave the overlapping
/// subset only.
pub fn update<S: ThetaSketchView>(&mut self, sketch: &S) -> Result<(), Error> {
- let new_default_table = |table: &ThetaHashTable| {
- ThetaHashTable::from_raw_parts(
- 0,
- 0,
- ResizeFactor::X1,
- 1.0,
- table.theta(),
- table.hash_seed(),
- table.is_empty(),
- )
- };
-
- if self.table.is_empty() {
- return Ok(());
- }
-
- if !sketch.is_empty() && sketch.seed_hash() != self.table.seed_hash() {
- return Err(Error::invalid_argument(format!(
- "incompatible seed hash: expected {}, got {}",
- self.table.seed_hash(),
- sketch.seed_hash()
- )));
- }
-
- if sketch.is_empty() {
- self.table.set_empty(true);
- }
-
- self.table.set_theta(if self.table.is_empty() {
- MAX_THETA
- } else {
- self.table.theta().min(sketch.theta())
- });
-
- if self.is_valid && self.table.num_retained() == 0 {
- return Ok(());
- }
-
- if sketch.num_retained() == 0 {
- self.is_valid = true;
- self.table = new_default_table(&self.table);
- return Ok(());
- }
-
- // first update, copy or move incoming sketch
- if !self.is_valid {
- self.is_valid = true;
- let lg_size = ThetaHashTable::lg_size_from_count_for_rebuild(
- sketch.num_retained(),
- HASH_TABLE_REBUILD_THRESHOLD,
- );
- self.table = ThetaHashTable::from_raw_parts(
- lg_size,
- lg_size - 1,
- ResizeFactor::X1,
- 1.0,
- self.table.theta(),
- self.table.hash_seed(),
- self.table.is_empty(),
- );
- for entry in sketch.iter() {
- let hash = entry.hash();
- if !self.table.try_insert_hash(hash) {
- return Err(Error::invalid_argument(
- "Insert entries from sketch fail, possibly corrupted input sketch",
- ));
- }
- }
- // Safety check.
- if self.table.num_retained() != sketch.num_retained() {
- return Err(Error::invalid_argument(
- "num entries mismatch, possibly corrupted input sketch",
- ));
- }
- } else {
- let max_matches = self.table.num_retained().min(sketch.num_retained());
- let mut matched_entries = Vec::with_capacity(max_matches);
- let mut count = 0;
- for entry in sketch.iter() {
- let hash = entry.hash();
- if hash < self.table.theta() {
- if self.table.contains_hash(hash) {
- if matched_entries.len() == max_matches {
- return Err(Error::invalid_argument(
- "max matches exceeded, possibly corrupted input sketch",
- ));
- }
- matched_entries.push(hash);
- }
- } else if sketch.is_ordered() {
- break; // early stop for ordered sketches
- }
- count += 1;
- }
- // Safety check.
- if count > sketch.num_retained() {
- return Err(Error::invalid_argument(
- "more keys than expected, possibly corrupted input sketch",
- ));
- } else if !sketch.is_ordered() && count < sketch.num_retained() {
- return Err(Error::invalid_argument(
- "fewer keys than expected, possibly corrupted input sketch",
- ));
- }
- if matched_entries.is_empty() {
- self.table = new_default_table(&self.table);
- if self.table.theta() == MAX_THETA {
- self.table.set_empty(true);
- }
- } else {
- let lg_size = ThetaHashTable::lg_size_from_count_for_rebuild(
- matched_entries.len(),
- HASH_TABLE_REBUILD_THRESHOLD,
- );
- self.table = ThetaHashTable::from_raw_parts(
- lg_size,
- lg_size - 1,
- ResizeFactor::X1,
- 1.0,
- self.table.theta(),
- self.table.hash_seed(),
- self.table.is_empty(),
- );
- for hash in matched_entries {
- if !self.table.try_insert_hash(hash) {
- return Err(Error::invalid_argument(
- "duplicate key, possibly corrupted input sketch",
- ));
- }
- }
- }
- }
- Ok(())
+ self.raw.update(sketch)
}
/// Returns whether this operator has received at least one update.
pub fn has_result(&self) -> bool {
- self.is_valid
+ self.raw.has_result()
}
/// Returns the intersection result as a compact theta sketch.
@@ -209,19 +73,20 @@ impl ThetaIntersection {
/// Panics if called before the first [`update`](Self::update).
pub fn to_sketch(&self, ordered: bool) -> CompactThetaSketch {
assert!(
- self.is_valid,
+ self.raw.has_result(),
"ThetaIntersection::to_sketch() called before first update()"
);
- let mut hashes: Vec<u64> = self.table.iter().collect();
- if ordered {
- hashes.sort_unstable();
- }
+ let parts = self.raw.result(ordered);
CompactThetaSketch::from_parts(
- hashes,
- self.table.theta(),
- self.table.seed_hash(),
- ordered,
- self.table.is_empty(),
+ parts
+ .entries
+ .into_iter()
+ .map(|entry| entry.hash())
+ .collect(),
+ parts.theta,
+ parts.seed_hash,
+ parts.ordered,
+ parts.empty,
)
}
}
diff --git a/datasketches/src/thetacommon/intersection.rs b/datasketches/src/thetacommon/intersection.rs
new file mode 100644
index 0000000..c2bb2be
--- /dev/null
+++ b/datasketches/src/thetacommon/intersection.rs
@@ -0,0 +1,388 @@
+// 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 crate::common::ResizeFactor;
+use crate::error::Error;
+use crate::thetacommon::RawHashTableEntry;
+use crate::thetacommon::RawThetaSketchView;
+use crate::thetacommon::constants::HASH_TABLE_REBUILD_THRESHOLD;
+use crate::thetacommon::constants::MAX_THETA;
+use crate::thetacommon::hash_table::RawCompactParts;
+use crate::thetacommon::hash_table::RawHashTable;
+
+/// Merges an incoming entry into an existing entry with the same hash.
+///
+/// For plain Theta entries there is nothing to merge (the entry is only a hash); tuple entries
+/// combine their summaries.
+pub trait RawThetaIntersectionPolicy<E> {
+ fn merge(&self, existing: &mut E, incoming: E);
+}
+
+/// Generic state machine shared by Theta and Tuple intersections.
+///
+/// `E` is the retained entry type. `P` defines how equal-hash entries are combined; it is only
+/// exercised for keys present in both the running intersection and the incoming sketch.
+#[derive(Debug)]
+pub struct RawThetaIntersection<E, P> {
+ table: RawHashTable<E>,
+ policy: P,
+ is_valid: bool,
+}
+
+impl<E, P> RawThetaIntersection<E, P>
+where
+ E: RawHashTableEntry,
+{
+ /// Creates a new intersection operator for the given `seed` and entry-merge `policy`.
+ pub fn new(seed: u64, policy: P) -> Self {
+ Self {
+ is_valid: false,
+ table: RawHashTable::from_raw_parts(
+ 0,
+ 0,
+ ResizeFactor::X1,
+ 1.0,
+ MAX_THETA,
+ seed,
+ false,
+ ),
+ policy,
+ }
+ }
+
+ /// Updates the intersection with a given sketch.
+ ///
+ /// The intersection can be viewed as starting from the "universe" set, and every update
+ /// reduces the current set to the keys it shares with `sketch`.
+ pub fn update<S>(&mut self, sketch: &S) -> Result<(), Error>
+ where
+ S: RawThetaSketchView<E>,
+ E: Clone,
+ P: RawThetaIntersectionPolicy<E>,
+ {
+ let new_default_table = |table: &RawHashTable<E>| {
+ RawHashTable::from_raw_parts(
+ 0,
+ 0,
+ ResizeFactor::X1,
+ 1.0,
+ table.theta(),
+ table.hash_seed(),
+ table.is_empty(),
+ )
+ };
+
+ if self.table.is_empty() {
+ return Ok(());
+ }
+
+ if !sketch.is_empty() && sketch.seed_hash() != self.table.seed_hash() {
+ return Err(Error::invalid_argument(format!(
+ "incompatible seed hash: expected {}, got {}",
+ self.table.seed_hash(),
+ sketch.seed_hash()
+ )));
+ }
+
+ if sketch.is_empty() {
+ self.table.set_empty(true);
+ }
+
+ self.table.set_theta(if self.table.is_empty() {
+ MAX_THETA
+ } else {
+ self.table.theta().min(sketch.theta())
+ });
+
+ if self.is_valid && self.table.num_retained() == 0 {
+ return Ok(());
+ }
+
+ if sketch.num_retained() == 0 {
+ self.is_valid = true;
+ self.table = new_default_table(&self.table);
+ return Ok(());
+ }
+
+ // first update, copy incoming entries
+ if !self.is_valid {
+ self.is_valid = true;
+ let lg_size = RawHashTable::<E>::lg_size_from_count_for_rebuild(
+ sketch.num_retained(),
+ HASH_TABLE_REBUILD_THRESHOLD,
+ );
+ // num_retained >= 1 here (the zero case returned early above), so lg_size >= 1 and
+ // lg_size - 1 below cannot underflow.
+ debug_assert!(lg_size >= 1);
+ self.table = RawHashTable::from_raw_parts(
+ lg_size,
+ lg_size - 1,
+ ResizeFactor::X1,
+ 1.0,
+ self.table.theta(),
+ self.table.hash_seed(),
+ self.table.is_empty(),
+ );
+ for entry in sketch.iter() {
+ let hash = entry.hash();
+ if !self.table.upsert_entry(hash, |existing| match existing {
+ Some(_) => None,
+ None => Some(entry),
+ }) {
+ return Err(Error::invalid_argument(
+ "Insert entries from sketch fail, possibly corrupted input sketch",
+ ));
+ }
+ }
+ // Safety check.
+ if self.table.num_retained() != sketch.num_retained() {
+ return Err(Error::invalid_argument(
+ "num entries mismatch, possibly corrupted input sketch",
+ ));
+ }
+ } else {
+ let max_matches = self.table.num_retained().min(sketch.num_retained());
+ let mut matched_entries = Vec::with_capacity(max_matches);
+ let mut count = 0;
+ for entry in sketch.iter() {
+ let hash = entry.hash();
+ if hash < self.table.theta() {
+ if let Some(existing) = self.table.get_entry(hash) {
+ if matched_entries.len() == max_matches {
+ return Err(Error::invalid_argument(
+ "max matches exceeded, possibly corrupted input sketch",
+ ));
+ }
+ let mut merged = existing.clone();
+ self.policy.merge(&mut merged, entry);
+ matched_entries.push(merged);
+ }
+ } else if sketch.is_ordered() {
+ break; // early stop for ordered sketches
+ }
+ count += 1;
+ }
+ // Safety check.
+ if count > sketch.num_retained() {
+ return Err(Error::invalid_argument(
+ "more keys than expected, possibly corrupted input sketch",
+ ));
+ } else if !sketch.is_ordered() && count < sketch.num_retained() {
+ return Err(Error::invalid_argument(
+ "fewer keys than expected, possibly corrupted input sketch",
+ ));
+ }
+ if matched_entries.is_empty() {
+ self.table = new_default_table(&self.table);
+ if self.table.theta() == MAX_THETA {
+ self.table.set_empty(true);
+ }
+ } else {
+ let lg_size = RawHashTable::<E>::lg_size_from_count_for_rebuild(
+ matched_entries.len(),
+ HASH_TABLE_REBUILD_THRESHOLD,
+ );
+ // matched_entries is non-empty here (the empty case is handled above), so
+ // lg_size >= 1 and lg_size - 1 below cannot underflow.
+ debug_assert!(lg_size >= 1);
+ self.table = RawHashTable::from_raw_parts(
+ lg_size,
+ lg_size - 1,
+ ResizeFactor::X1,
+ 1.0,
+ self.table.theta(),
+ self.table.hash_seed(),
+ self.table.is_empty(),
+ );
+ for entry in matched_entries {
+ let hash = entry.hash();
+ if !self.table.upsert_entry(hash, |existing| match existing {
+ Some(_) => None,
+ None => Some(entry),
+ }) {
+ return Err(Error::invalid_argument(
+ "duplicate key, possibly corrupted input sketch",
+ ));
+ }
+ }
+ }
+ }
+ Ok(())
+ }
+
+ /// Returns whether this operator has received at least one update.
+ pub fn has_result(&self) -> bool {
+ self.is_valid
+ }
+
+ /// Return the current intersection state as raw compact-sketch parts.
+ pub fn result(&self, ordered: bool) -> RawCompactParts<E>
+ where
+ E: Clone,
+ {
+ let mut entries: Vec<E> = self.table.iter_entries().cloned().collect();
+ if ordered {
+ entries.sort_unstable_by_key(RawHashTableEntry::hash);
+ }
+ RawCompactParts {
+ entries,
+ theta: self.table.theta(),
+ seed_hash: self.table.seed_hash(),
+ ordered,
+ empty: self.table.is_empty(),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::hash::DEFAULT_UPDATE_SEED;
+
+ #[derive(Clone, Debug, Eq, PartialEq)]
+ struct TestEntry {
+ hash: u64,
+ summary: u64,
+ }
+
+ impl RawHashTableEntry for TestEntry {
+ fn hash(&self) -> u64 {
+ self.hash
+ }
+ }
+
+ struct TestSketch {
+ entries: Vec<TestEntry>,
+ }
+
+ impl TestSketch {
+ fn of_hashes(hashes: &[u64]) -> Self {
+ Self {
+ entries: hashes
+ .iter()
+ .map(|&hash| TestEntry { hash, summary: 1 })
+ .collect(),
+ }
+ }
+ }
+
+ impl RawThetaSketchView<TestEntry> for TestSketch {
+ fn seed_hash(&self) -> u16 {
+ crate::hash::compute_seed_hash(DEFAULT_UPDATE_SEED)
+ }
+
+ fn theta(&self) -> u64 {
+ MAX_THETA
+ }
+
+ fn is_empty(&self) -> bool {
+ self.entries.is_empty()
+ }
+
+ fn is_ordered(&self) -> bool {
+ false
+ }
+
+ fn iter(&self) -> impl Iterator<Item = TestEntry> + '_ {
+ self.entries.iter().cloned()
+ }
+
+ fn num_retained(&self) -> usize {
+ self.entries.len()
+ }
+ }
+
+ struct SumPolicy;
+
+ impl RawThetaIntersectionPolicy<TestEntry> for SumPolicy {
+ fn merge(&self, existing: &mut TestEntry, incoming: TestEntry) {
+ existing.summary += incoming.summary;
+ }
+ }
+
+ #[test]
+ fn first_update_copies_entries() {
+ let mut intersection = RawThetaIntersection::new(DEFAULT_UPDATE_SEED, SumPolicy);
+ assert!(!intersection.has_result());
+
+ intersection
+ .update(&TestSketch::of_hashes(&[1, 2, 3]))
+ .unwrap();
+
+ assert!(intersection.has_result());
+ let parts = intersection.result(true);
+ assert_eq!(
+ parts.entries,
+ vec![
+ TestEntry {
+ hash: 1,
+ summary: 1
+ },
+ TestEntry {
+ hash: 2,
+ summary: 1
+ },
+ TestEntry {
+ hash: 3,
+ summary: 1
+ },
+ ]
+ );
+ }
+
+ #[test]
+ fn second_update_keeps_matches_and_merges_with_policy() {
+ let mut intersection = RawThetaIntersection::new(DEFAULT_UPDATE_SEED, SumPolicy);
+ intersection
+ .update(&TestSketch::of_hashes(&[1, 2, 3]))
+ .unwrap();
+ intersection
+ .update(&TestSketch::of_hashes(&[2, 3, 4]))
+ .unwrap();
+
+ let parts = intersection.result(true);
+ assert_eq!(
+ parts.entries,
+ vec![
+ TestEntry {
+ hash: 2,
+ summary: 2
+ },
+ TestEntry {
+ hash: 3,
+ summary: 2
+ },
+ ]
+ );
+ }
+
+ #[test]
+ fn disjoint_second_update_empties_intersection() {
+ let mut intersection = RawThetaIntersection::new(DEFAULT_UPDATE_SEED, SumPolicy);
+ intersection
+ .update(&TestSketch::of_hashes(&[1, 2, 3]))
+ .unwrap();
+ intersection
+ .update(&TestSketch::of_hashes(&[4, 5]))
+ .unwrap();
+
+ let parts = intersection.result(true);
+ assert!(parts.entries.is_empty());
+ assert!(parts.empty);
+ assert_eq!(parts.theta, MAX_THETA);
+ }
+}
diff --git a/datasketches/src/thetacommon/mod.rs b/datasketches/src/thetacommon/mod.rs
index e0bed52..11c9346 100644
--- a/datasketches/src/thetacommon/mod.rs
+++ b/datasketches/src/thetacommon/mod.rs
@@ -20,6 +20,7 @@
pub(crate) mod binomial_bounds;
pub(crate) mod constants;
pub(crate) mod hash_table;
+pub(crate) mod intersection;
pub(crate) mod union;
/// An entry retained by a Theta sketch family hash table.
diff --git a/datasketches/src/tuple/hash_table.rs b/datasketches/src/tuple/hash_table.rs
index 04775d7..138c940 100644
--- a/datasketches/src/tuple/hash_table.rs
+++ b/datasketches/src/tuple/hash_table.rs
@@ -20,6 +20,7 @@ use std::num::NonZeroU64;
use crate::thetacommon::RawHashTableEntry;
use crate::thetacommon::hash_table::RawHashTable;
+use crate::thetacommon::intersection::RawThetaIntersectionPolicy;
use crate::thetacommon::union::RawThetaUnionPolicy;
use crate::tuple::SummaryCombinePolicy;
@@ -113,6 +114,12 @@ impl<P: SummaryCombinePolicy> RawThetaUnionPolicy<TupleEntry<P::Summary>> for P
}
}
+impl<P: SummaryCombinePolicy> RawThetaIntersectionPolicy<TupleEntry<P::Summary>> for P {
+ fn merge(&self, existing: &mut TupleEntry<P::Summary>, incoming: TupleEntry<P::Summary>) {
+ self.combine(&mut existing.summary, &incoming.summary);
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/datasketches/src/tuple/intersection.rs b/datasketches/src/tuple/intersection.rs
new file mode 100644
index 0000000..736cd94
--- /dev/null
+++ b/datasketches/src/tuple/intersection.rs
@@ -0,0 +1,384 @@
+// 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.
+
+//! Tuple sketch intersection.
+//!
+//! [`TupleIntersection`] computes the intersection (set AND) of Tuple sketches. It reuses the raw
+//! intersection state machine (`RawThetaIntersection`) that also drives the Theta intersection;
+//! the only Tuple-specific addition is that for each key retained in both the running result and
+//! the incoming sketch, the two summaries are combined with a [`SummaryCombinePolicy`].
+//!
+//! Unlike the union there is no default policy: how to combine the summaries of keys present in
+//! both inputs is application-specific, so a policy must always be supplied.
+
+use crate::error::Error;
+use crate::hash::DEFAULT_UPDATE_SEED;
+use crate::thetacommon::intersection::RawThetaIntersection;
+use crate::tuple::hash_table::TupleEntry;
+use crate::tuple::policy::SummaryCombinePolicy;
+use crate::tuple::sketch::CompactTupleSketch;
+use crate::tuple::sketch::TupleSketchView;
+
+/// Stateful intersection operator for Tuple sketches.
+///
+/// `P` is the [`SummaryCombinePolicy`] applied to keys present in more than one input. There is no
+/// default policy (see the module docs), so one must be supplied at construction.
+///
+/// Before the first [`update`](Self::update), the result is undefined; use
+/// [`has_result`](Self::has_result) to check.
+///
+/// # Examples
+///
+/// ```
+/// use datasketches::tuple::DefaultUpdatePolicy;
+/// use datasketches::tuple::SummaryCombinePolicy;
+/// use datasketches::tuple::SummaryPolicy;
+/// use datasketches::tuple::TupleIntersection;
+/// use datasketches::tuple::TupleSketchBuilder;
+///
+/// // Sum the summaries of keys that appear in both inputs.
+/// #[derive(Default)]
+/// struct SumPolicy;
+/// impl SummaryPolicy for SumPolicy {
+/// type Summary = u64;
+///
+/// fn create(&self) -> Self::Summary {
+/// 0
+/// }
+/// }
+/// impl SummaryCombinePolicy for SumPolicy {
+/// fn combine(&self, summary: &mut Self::Summary, other: &Self::Summary) {
+/// *summary += *other;
+/// }
+/// }
+///
+/// let update_policy = DefaultUpdatePolicy::<u64>::default();
+/// let mut a = TupleSketchBuilder::new(update_policy).build();
+/// a.update("shared", 3);
+/// a.update("only_a", 1);
+///
+/// let mut b = TupleSketchBuilder::new(update_policy).build();
+/// b.update("shared", 4);
+/// b.update("only_b", 1);
+///
+/// let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+/// intersection.update(&a).unwrap();
+/// intersection.update(&b).unwrap();
+///
+/// let result = intersection.to_sketch(true);
+/// assert_eq!(result.num_retained(), 1); // only "shared"
+/// assert_eq!(result.iter().next().unwrap().1, &7); // 3 + 4
+/// ```
+#[derive(Debug)]
+pub struct TupleIntersection<P>
+where
+ P: SummaryCombinePolicy,
+{
+ raw: RawThetaIntersection<TupleEntry<P::Summary>, P>,
+}
+
+impl<P> TupleIntersection<P>
+where
+ P: SummaryCombinePolicy,
+{
+ /// Creates a new intersection operator for the given `seed` and combine `policy`.
+ pub fn new(seed: u64, policy: P) -> Self {
+ Self {
+ raw: RawThetaIntersection::new(seed, policy),
+ }
+ }
+
+ /// Creates a new intersection operator with the default seed and the given combine `policy`.
+ pub fn new_with_default_seed(policy: P) -> Self {
+ Self::new(DEFAULT_UPDATE_SEED, policy)
+ }
+
+ /// Updates the intersection with a given sketch.
+ ///
+ /// The intersection can be viewed as starting from the "universe" set, and every update reduces
+ /// the current set to the keys it shares with `sketch`. Summaries of shared keys are combined
+ /// via the policy.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if `sketch` (when non-empty) has a different seed hash, or if the input
+ /// appears corrupted (entry counts do not match what the sketch reports).
+ pub fn update<V>(&mut self, sketch: &V) -> Result<(), Error>
+ where
+ V: TupleSketchView<P::Summary>,
+ P::Summary: Clone,
+ {
+ self.raw.update(sketch)
+ }
+
+ /// Returns whether this operator has received at least one update.
+ pub fn has_result(&self) -> bool {
+ self.raw.has_result()
+ }
+
+ /// Returns the intersection result as a compact Tuple sketch.
+ ///
+ /// If `ordered` is true, retained entries are sorted ascending by hash.
+ ///
+ /// # Panics
+ ///
+ /// Panics if called before the first [`update`](Self::update).
+ pub fn to_sketch(&self, ordered: bool) -> CompactTupleSketch<P::Summary>
+ where
+ P::Summary: Clone,
+ {
+ assert!(
+ self.raw.has_result(),
+ "TupleIntersection::to_sketch() called before first update()"
+ );
+ let parts = self.raw.result(ordered);
+ CompactTupleSketch::from_parts(
+ parts.entries,
+ parts.theta,
+ parts.seed_hash,
+ parts.ordered,
+ parts.empty,
+ )
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::tuple::DefaultUpdatePolicy;
+ use crate::tuple::SummaryPolicy;
+ use crate::tuple::TupleSketchBuilder;
+
+ #[derive(Debug, Default, Clone, Copy)]
+ struct SumPolicy;
+
+ impl SummaryPolicy for SumPolicy {
+ type Summary = u64;
+
+ fn create(&self) -> Self::Summary {
+ 0
+ }
+ }
+
+ impl SummaryCombinePolicy for SumPolicy {
+ fn combine(&self, summary: &mut Self::Summary, other: &Self::Summary) {
+ *summary += *other;
+ }
+ }
+
+ fn default_sketch_builder() -> TupleSketchBuilder<DefaultUpdatePolicy<u64>> {
+ TupleSketchBuilder::new(DefaultUpdatePolicy::<u64>::default())
+ }
+
+ fn sorted_entries(sketch: &CompactTupleSketch<u64>) -> Vec<(u64, u64)> {
+ let mut entries: Vec<(u64, u64)> = sketch.iter().map(|(h, &s)| (h, s)).collect();
+ entries.sort_unstable();
+ entries
+ }
+
+ #[test]
+ fn intersection_of_overlapping_sketches() {
+ let mut a = default_sketch_builder().build();
+ for i in 0..1000 {
+ a.update(i, 1u64);
+ }
+ let mut b = default_sketch_builder().build();
+ for i in 500..1500 {
+ b.update(i, 1u64);
+ }
+
+ let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ intersection.update(&a).unwrap();
+ intersection.update(&b).unwrap();
+
+ let result = intersection.to_sketch(true);
+ // Keys 500..1000 are shared (exact mode), each summary is 1 + 1 = 2.
+ assert_eq!(result.num_retained(), 500);
+ assert_eq!(result.estimate(), 500.0);
+ assert!(result.iter().all(|(_, &s)| s == 2));
+ }
+
+ #[test]
+ fn intersection_combines_summaries_of_shared_keys() {
+ let mut a = default_sketch_builder().build();
+ a.update("shared", 3u64);
+ a.update("only_a", 100u64);
+ let mut b = default_sketch_builder().build();
+ b.update("shared", 4u64);
+ b.update("only_b", 200u64);
+
+ let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ intersection.update(&a).unwrap();
+ intersection.update(&b).unwrap();
+
+ let result = intersection.to_sketch(true);
+ assert_eq!(sorted_entries(&result).len(), 1);
+ assert_eq!(result.iter().next().unwrap().1, &7); // 3 + 4
+ }
+
+ #[test]
+ fn intersection_is_order_independent() {
+ let mut a = default_sketch_builder().build();
+ for i in 0..1000 {
+ a.update(i, 1u64);
+ }
+ let mut b = default_sketch_builder().build();
+ for i in 500..1500 {
+ b.update(i, 1u64);
+ }
+
+ let mut a_then_b = TupleIntersection::new_with_default_seed(SumPolicy);
+ a_then_b.update(&a).unwrap();
+ a_then_b.update(&b).unwrap();
+
+ let mut b_then_a = TupleIntersection::new_with_default_seed(SumPolicy);
+ b_then_a.update(&b).unwrap();
+ b_then_a.update(&a).unwrap();
+
+ assert_eq!(
+ sorted_entries(&a_then_b.to_sketch(true)),
+ sorted_entries(&b_then_a.to_sketch(true))
+ );
+ }
+
+ #[test]
+ fn intersection_accepts_updatable_and_compact_inputs() {
+ let mut a = default_sketch_builder().build();
+ for i in 0..1000 {
+ a.update(i, 1u64);
+ }
+ let mut b = default_sketch_builder().build();
+ for i in 500..1500 {
+ b.update(i, 1u64);
+ }
+ let b_compact = b.compact(true);
+
+ let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ intersection.update(&a).unwrap();
+ intersection.update(&b_compact).unwrap();
+
+ assert_eq!(intersection.to_sketch(true).num_retained(), 500);
+ }
+
+ #[test]
+ fn intersection_with_disjoint_sketches_is_empty() {
+ let mut a = default_sketch_builder().build();
+ for i in 0..1000 {
+ a.update(i, 1u64);
+ }
+ let mut b = default_sketch_builder().build();
+ for i in 1000..2000 {
+ b.update(i, 1u64);
+ }
+
+ let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ intersection.update(&a).unwrap();
+ intersection.update(&b).unwrap();
+
+ let result = intersection.to_sketch(true);
+ assert_eq!(result.num_retained(), 0);
+ assert_eq!(result.estimate(), 0.0);
+ }
+
+ #[test]
+ fn intersection_with_empty_input_is_empty() {
+ let mut a = default_sketch_builder().build();
+ for i in 0..1000 {
+ a.update(i, 1u64);
+ }
+ let empty = default_sketch_builder().build();
+
+ let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ intersection.update(&a).unwrap();
+ intersection.update(&empty).unwrap();
+
+ let result = intersection.to_sketch(true);
+ assert!(result.is_empty());
+ assert_eq!(result.num_retained(), 0);
+ assert_eq!(result.estimate(), 0.0);
+ }
+
+ #[test]
+ fn intersection_single_update_returns_input() {
+ let mut a = default_sketch_builder().build();
+ for i in 0..100 {
+ a.update(i, 5u64);
+ }
+
+ let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ intersection.update(&a).unwrap();
+
+ let result = intersection.to_sketch(true);
+ assert_eq!(result.num_retained(), 100);
+ // A single update copies the input unchanged (summaries not combined with anything).
+ assert!(result.iter().all(|(_, &s)| s == 5));
+ }
+
+ #[test]
+ fn has_result_reflects_first_update() {
+ let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ assert!(!intersection.has_result());
+
+ let mut a = default_sketch_builder().build();
+ a.update(1, 1u64);
+ intersection.update(&a).unwrap();
+ assert!(intersection.has_result());
+ }
+
+ #[test]
+ #[should_panic(expected = "before first update")]
+ fn result_before_update_panics() {
+ let intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ let _ = intersection.to_sketch(true);
+ }
+
+ #[test]
+ fn intersection_rejects_seed_mismatch() {
+ let mut a = default_sketch_builder().seed(1).build();
+ a.update(1, 1u64);
+
+ let mut intersection = TupleIntersection::new(2, SumPolicy);
+ let err = intersection.update(&a).unwrap_err();
+ assert_eq!(err.kind(), crate::error::ErrorKind::InvalidArgument);
+ }
+
+ #[test]
+ fn intersection_in_estimation_mode_estimates_within_bounds() {
+ let mut a = default_sketch_builder().lg_k(8).build();
+ for i in 0..50000 {
+ a.update(i, 1u64);
+ }
+ let mut b = default_sketch_builder().lg_k(8).build();
+ for i in 25000..75000 {
+ b.update(i, 1u64);
+ }
+
+ let mut intersection = TupleIntersection::new_with_default_seed(SumPolicy);
+ intersection.update(&a).unwrap();
+ intersection.update(&b).unwrap();
+
+ let result = intersection.to_sketch(true);
+ assert!(result.is_estimation_mode());
+ // True intersection size is 25000 (keys 25000..50000).
+ let lower = result.lower_bound(crate::common::NumStdDev::Three);
+ let upper = result.upper_bound(crate::common::NumStdDev::Three);
+ assert!(
+ lower <= 25000.0 && 25000.0 <= upper,
+ "expected 25000 in [{lower}, {upper}]"
+ );
+ }
+}
diff --git a/datasketches/src/tuple/mod.rs b/datasketches/src/tuple/mod.rs
index b941efa..6701f42 100644
--- a/datasketches/src/tuple/mod.rs
+++ b/datasketches/src/tuple/mod.rs
@@ -39,12 +39,14 @@
//! ```
mod hash_table;
+mod intersection;
mod policy;
mod serialization;
mod sketch;
mod union;
pub use self::hash_table::TupleEntry;
+pub use self::intersection::TupleIntersection;
pub use self::policy::DefaultUnionPolicy;
pub use self::policy::DefaultUpdatePolicy;
pub use self::policy::SummaryCombinePolicy;
diff --git a/datasketches/tests/tuple_intersection_test.rs b/datasketches/tests/tuple_intersection_test.rs
new file mode 100644
index 0000000..9407a88
--- /dev/null
+++ b/datasketches/tests/tuple_intersection_test.rs
@@ -0,0 +1,343 @@
+// 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.
+
+#![cfg(feature = "tuple")]
+
+use datasketches::tuple::CompactTupleSketch;
+use datasketches::tuple::DefaultUpdatePolicy;
+use datasketches::tuple::SummaryCombinePolicy;
+use datasketches::tuple::SummaryPolicy;
+use datasketches::tuple::TupleIntersection;
+use datasketches::tuple::TupleSketch;
+use datasketches::tuple::TupleSketchBuilder;
+
+#[derive(Debug, Default, Clone, Copy)]
+struct SumPolicy;
+
+impl SummaryPolicy for SumPolicy {
+ type Summary = u64;
+
+ fn create(&self) -> Self::Summary {
+ 0
+ }
+}
+
+impl SummaryCombinePolicy for SumPolicy {
+ fn combine(&self, summary: &mut Self::Summary, other: &Self::Summary) {
+ *summary += *other;
+ }
+}
+
+fn default_sketch_builder() -> TupleSketchBuilder<DefaultUpdatePolicy<u64>> {
+ TupleSketchBuilder::new(DefaultUpdatePolicy::<u64>::default())
+}
+
+fn sketch_with_range(start: u64, count: u64) -> TupleSketch<DefaultUpdatePolicy<u64>> {
+ let mut sketch = default_sketch_builder().build();
+ for i in 0..count {
+ sketch.update(start + i, 1u64);
+ }
+ sketch
+}
+
+#[test]
+fn test_has_result_state_machine() {
+ let mut a = default_sketch_builder().build();
+ a.update("x", 1u64);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ assert!(!i.has_result());
+ i.update(&a).unwrap();
+ assert!(i.has_result());
+ assert!(i.to_sketch(true).estimate() >= 1.0);
+}
+
+#[test]
+fn test_result_before_update_panics() {
+ let i = TupleIntersection::new(123, SumPolicy);
+ let result = std::panic::catch_unwind(|| {
+ let _ = i.to_sketch(true);
+ });
+ assert!(result.is_err());
+}
+
+#[test]
+fn test_update_accepts_compact_sketch() {
+ let mut a = default_sketch_builder().build();
+ a.update("x", 1u64);
+ a.update("y", 1u64);
+
+ let mut b = default_sketch_builder().build();
+ b.update("y", 1u64);
+ b.update("z", 1u64);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&a.compact(true)).unwrap();
+ i.update(&b).unwrap();
+
+ let r = i.to_sketch(true);
+ assert!(r.estimate() == 1.0);
+ assert!(r.is_ordered());
+
+ let mut c = default_sketch_builder().build();
+ c.update("a", 1u64);
+ c.update("b", 1u64);
+ c.update("c", 1u64);
+
+ i.update(&c.compact(false)).unwrap();
+
+ let r = i.to_sketch(false);
+ assert!(r.estimate() == 0.0);
+ assert!(!r.is_ordered());
+}
+
+#[test]
+fn test_seed_mismatch_behaviour_for_empty_sketch() {
+ let empty_other_seed = default_sketch_builder().seed(2).build();
+ let mut i = TupleIntersection::new(1, SumPolicy);
+
+ i.update(&empty_other_seed).unwrap();
+ assert!(i.has_result());
+ let r = i.to_sketch(true);
+ assert!(r.is_empty());
+}
+
+#[test]
+fn test_seed_mismatch_behaviour() {
+ let mut one_other_seed = default_sketch_builder().seed(2).build();
+ one_other_seed.update("value", 1u64);
+ let mut i = TupleIntersection::new(1, SumPolicy);
+
+ assert!(i.update(&one_other_seed).is_err());
+}
+
+#[test]
+fn test_terminal_empty_state_ignores_future_updates() {
+ let empty = default_sketch_builder().build();
+
+ let mut non_empty = default_sketch_builder().build();
+ non_empty.update("x", 1u64);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&empty).unwrap();
+ i.update(&non_empty).unwrap();
+
+ let r = i.to_sketch(true);
+ assert!(r.is_empty());
+}
+
+#[test]
+fn test_to_sketch_unordered_is_not_ordered() {
+ let mut a = default_sketch_builder().build();
+ for i in 0..64 {
+ a.update(i, 1u64);
+ }
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&a).unwrap();
+
+ let r = i.to_sketch(false);
+ assert!(!r.is_ordered());
+}
+
+#[test]
+fn test_empty_update_twice() {
+ let empty = default_sketch_builder().build();
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+
+ i.update(&empty).unwrap();
+ let r1 = i.to_sketch(true);
+ assert_eq!(r1.num_retained(), 0);
+ assert!(r1.is_empty());
+ assert!(!r1.is_estimation_mode());
+ assert_eq!(r1.estimate(), 0.0);
+
+ i.update(&empty).unwrap();
+ let r2 = i.to_sketch(true);
+ assert_eq!(r2.num_retained(), 0);
+ assert!(r2.is_empty());
+ assert!(!r2.is_estimation_mode());
+ assert_eq!(r2.estimate(), 0.0);
+}
+
+#[test]
+fn test_non_empty_no_retained_keys() {
+ let mut s = default_sketch_builder().sampling_probability(0.001).build();
+ s.update(1u64, 1u64);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s).unwrap();
+ let r1 = i.to_sketch(true);
+ assert_eq!(r1.num_retained(), 0);
+ assert!(!r1.is_empty());
+ assert!(r1.is_estimation_mode());
+ assert!((r1.theta() - 0.001).abs() < 1e-10);
+ assert_eq!(r1.estimate(), 0.0);
+
+ i.update(&s).unwrap();
+ let r2 = i.to_sketch(true);
+ assert_eq!(r2.num_retained(), 0);
+ assert!(!r2.is_empty());
+ assert!(r2.is_estimation_mode());
+ assert!((r2.theta() - 0.001).abs() < 1e-10);
+ assert_eq!(r2.estimate(), 0.0);
+}
+
+#[test]
+fn test_exact_half_overlap_unordered() {
+ let s1 = sketch_with_range(0, 1000);
+ let s2 = sketch_with_range(500, 1000);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s1).unwrap();
+ i.update(&s2).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(!r.is_empty());
+ assert!(!r.is_estimation_mode());
+ assert_eq!(r.estimate(), 500.0);
+}
+
+#[test]
+fn test_exact_half_overlap_ordered() {
+ let s1 = sketch_with_range(0, 1000);
+ let s2 = sketch_with_range(500, 1000);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s1.compact(true)).unwrap();
+ i.update(&s2.compact(true)).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(!r.is_empty());
+ assert!(!r.is_estimation_mode());
+ assert_eq!(r.estimate(), 500.0);
+}
+
+#[test]
+fn test_exact_disjoint_unordered() {
+ let s1 = sketch_with_range(0, 1000);
+ let s2 = sketch_with_range(1000, 1000);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s1).unwrap();
+ i.update(&s2).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(r.is_empty());
+ assert!(!r.is_estimation_mode());
+ assert_eq!(r.estimate(), 0.0);
+}
+
+#[test]
+fn test_exact_disjoint_ordered() {
+ let s1 = sketch_with_range(0, 1000);
+ let s2 = sketch_with_range(1000, 1000);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s1.compact(true)).unwrap();
+ i.update(&s2.compact(true)).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(r.is_empty());
+ assert!(!r.is_estimation_mode());
+ assert_eq!(r.estimate(), 0.0);
+}
+
+#[test]
+fn test_estimation_half_overlap_unordered() {
+ let s1 = sketch_with_range(0, 10000);
+ let s2 = sketch_with_range(5000, 10000);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s1).unwrap();
+ i.update(&s2).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(!r.is_empty());
+ assert!(r.is_estimation_mode());
+ assert!((r.estimate() - 5000.0).abs() <= 5000.0 * 0.02);
+}
+
+#[test]
+fn test_estimation_half_overlap_ordered() {
+ let s1 = sketch_with_range(0, 10000);
+ let s2 = sketch_with_range(5000, 10000);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s1.compact(true)).unwrap();
+ i.update(&s2.compact(true)).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(!r.is_empty());
+ assert!(r.is_estimation_mode());
+ assert!((r.estimate() - 5000.0).abs() <= 5000.0 * 0.02);
+}
+
+#[test]
+fn test_estimation_half_overlap_ordered_deserialized_compact() {
+ let s1 = sketch_with_range(0, 10000);
+ let s2 = sketch_with_range(5000, 10000);
+ let c1 = CompactTupleSketch::<u64>::deserialize(&s1.compact(true).serialize()).unwrap();
+ let c2 = CompactTupleSketch::<u64>::deserialize(&s2.compact(true).serialize()).unwrap();
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&c1).unwrap();
+ i.update(&c2).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(!r.is_empty());
+ assert!(r.is_estimation_mode());
+ assert!((r.estimate() - 5000.0).abs() <= 5000.0 * 0.02);
+}
+
+#[test]
+fn test_estimation_disjoint_unordered() {
+ let s1 = sketch_with_range(0, 10000);
+ let s2 = sketch_with_range(10000, 10000);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s1).unwrap();
+ i.update(&s2).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(!r.is_empty());
+ assert!(r.is_estimation_mode());
+ assert_eq!(r.estimate(), 0.0);
+}
+
+#[test]
+fn test_estimation_disjoint_ordered() {
+ let s1 = sketch_with_range(0, 10000);
+ let s2 = sketch_with_range(10000, 10000);
+
+ let mut i = TupleIntersection::new_with_default_seed(SumPolicy);
+ i.update(&s1.compact(true)).unwrap();
+ i.update(&s2.compact(true)).unwrap();
+ let r = i.to_sketch(true);
+
+ assert!(!r.is_empty());
+ assert!(r.is_estimation_mode());
+ assert_eq!(r.estimate(), 0.0);
+}
+
+#[test]
+fn test_seed_mismatch_non_empty_returns_error() {
+ let mut s = default_sketch_builder().build();
+ s.update(1u64, 1u64);
+
+ let mut i = TupleIntersection::new(123, SumPolicy);
+ assert!(i.update(&s).is_err());
+}
diff --git a/datasketches/tests/tuple_sketch_test.rs b/datasketches/tests/tuple_sketch_test.rs
index fb5468e..ca69e02 100644
--- a/datasketches/tests/tuple_sketch_test.rs
+++ b/datasketches/tests/tuple_sketch_test.rs
@@ -22,13 +22,13 @@ use datasketches::hash_value;
use datasketches::tuple::DefaultUpdatePolicy;
use datasketches::tuple::TupleSketchBuilder;
-fn builder() -> TupleSketchBuilder<DefaultUpdatePolicy<u64>> {
+fn default_sketch_builder() -> TupleSketchBuilder<DefaultUpdatePolicy<u64>> {
TupleSketchBuilder::new(DefaultUpdatePolicy::<u64>::default())
}
#[test]
fn test_basic_update() {
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
assert!(sketch.is_empty());
assert_eq!(sketch.estimate(), 0.0);
@@ -42,7 +42,7 @@ fn test_basic_update() {
#[test]
fn test_summary_accumulates_per_key() {
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
for _ in 0..5 {
sketch.update("same_key", 2u64);
}
@@ -54,7 +54,7 @@ fn test_summary_accumulates_per_key() {
#[test]
fn test_update_various_types() {
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
sketch.update("string", 1u64);
sketch.update(42i64, 1u64);
@@ -69,7 +69,7 @@ fn test_update_various_types() {
assert!(!sketch.is_empty());
assert_eq!(sketch.estimate(), 5.0);
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
sketch.update("string", 1u64);
sketch.update(42i64, 1u64);
@@ -87,7 +87,7 @@ fn test_update_various_types() {
#[test]
fn test_duplicate_updates() {
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
for _ in 0..100 {
sketch.update("same_value", 1u64);
@@ -98,7 +98,7 @@ fn test_duplicate_updates() {
#[test]
fn test_theta_reduction() {
- let mut sketch = builder().lg_k(5).build(); // Small k to trigger theta reduction
+ let mut sketch = default_sketch_builder().lg_k(5).build(); // Small k to trigger theta reduction
assert!(!sketch.is_estimation_mode());
// Insert many values to trigger theta reduction
@@ -112,7 +112,7 @@ fn test_theta_reduction() {
#[test]
fn test_trim() {
- let mut sketch = builder().lg_k(5).build();
+ let mut sketch = default_sketch_builder().lg_k(5).build();
// Insert many values
for i in 0..1000 {
@@ -130,7 +130,7 @@ fn test_trim() {
#[test]
fn test_reset() {
- let mut sketch = builder().lg_k(5).build();
+ let mut sketch = default_sketch_builder().lg_k(5).build();
// Insert many values
for i in 0..1000 {
@@ -153,7 +153,7 @@ fn test_reset() {
#[test]
fn test_iterator() {
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
sketch.update("value1", 1u64);
sketch.update("value2", 1u64);
@@ -165,7 +165,7 @@ fn test_iterator() {
#[test]
fn test_bounds_empty_sketch() {
- let sketch = builder().lg_k(12).build();
+ let sketch = default_sketch_builder().lg_k(12).build();
assert!(sketch.is_empty());
assert!(!sketch.is_estimation_mode());
assert_eq!(sketch.theta(), 1.0);
@@ -180,7 +180,7 @@ fn test_bounds_empty_sketch() {
#[test]
fn test_bounds_exact_mode() {
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
for i in 0..2000 {
sketch.update(i, 1u64);
}
@@ -194,7 +194,7 @@ fn test_bounds_exact_mode() {
#[test]
fn test_bounds_estimation_mode() {
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
let n = 10000;
for i in 0..n {
sketch.update(i, 1u64);
@@ -236,7 +236,10 @@ fn test_bounds_estimation_mode() {
#[test]
fn test_bounds_with_sampling() {
- let mut sketch = builder().lg_k(12).sampling_probability(0.5).build();
+ let mut sketch = default_sketch_builder()
+ .lg_k(12)
+ .sampling_probability(0.5)
+ .build();
for i in 0..1000 {
sketch.update(i, 1u64);
@@ -256,7 +259,7 @@ fn test_bounds_with_sampling() {
#[test]
fn test_bounds_all_num_std_devs() {
- let mut sketch = builder().lg_k(12).build();
+ let mut sketch = default_sketch_builder().lg_k(12).build();
for i in 0..10000 {
sketch.update(i, 1u64);
}
@@ -278,7 +281,10 @@ fn test_bounds_all_num_std_devs() {
#[test]
fn test_bounds_empty_estimation_mode() {
// Create a sketch with sampling probability < 1.0 to force estimation mode
- let sketch = builder().lg_k(12).sampling_probability(0.1).build();
+ let sketch = default_sketch_builder()
+ .lg_k(12)
+ .sampling_probability(0.1)
+ .build();
// The sketch is empty but theta < 1.0, so it's in estimation mode.
// When empty, both bounds should return 0.0 (matching the Java/Theta behavior).
@@ -293,13 +299,19 @@ fn test_bounds_empty_estimation_mode() {
fn test_compact_preserves_logical_non_empty_after_screened_update() {
let screened_value = (0u64..)
.find(|candidate| {
- let mut sketch = builder().lg_k(12).sampling_probability(0.5).build();
+ let mut sketch = default_sketch_builder()
+ .lg_k(12)
+ .sampling_probability(0.5)
+ .build();
sketch.update(*candidate, 1u64);
!sketch.is_empty() && sketch.num_retained() == 0
})
.expect("failed to find a value screened out by the sampling theta");
- let mut sketch = builder().lg_k(12).sampling_probability(0.5).build();
+ let mut sketch = default_sketch_builder()
+ .lg_k(12)
+ .sampling_probability(0.5)
+ .build();
sketch.update(screened_value, 1u64);
assert!(!sketch.is_empty());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org