datasketches 0.4.0

// 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 types.
//!
//! This module provides [`TupleSketch`] (mutable) and [`CompactTupleSketch`] (immutable),
//! the Tuple sketch analogues of the Theta sketch. Each retained key carries a user-defined summary
//! created by a [`SummaryPolicy`] and updated through one or more [`SummaryUpdatePolicy`]
//! implementations.

use std::hash::Hash;
use std::slice;

use crate::codec::SketchBytes;
use crate::codec::SketchSlice;
use crate::codec::assert::ensure_preamble_longs_in_range;
use crate::codec::assert::insufficient_data;
use crate::codec::family::Family;
use crate::common::NumStdDev;
use crate::common::ResizeFactor;
use crate::error::Error;
use crate::error::ErrorKind;
use crate::hash::DEFAULT_UPDATE_SEED;
use crate::hash::check_seed_hash;
use crate::hash::compute_seed_hash;
use crate::thetacommon::EntrySketch;
use crate::thetacommon::KeySketch;
use crate::thetacommon::SketchScalars;
use crate::thetacommon::binomial_bounds;
use crate::thetacommon::constants::DEFAULT_LG_K;
use crate::thetacommon::constants::FLAGS_IS_COMPACT;
use crate::thetacommon::constants::FLAGS_IS_EMPTY;
use crate::thetacommon::constants::FLAGS_IS_ORDERED;
use crate::thetacommon::constants::FLAGS_IS_READ_ONLY;
use crate::thetacommon::constants::MAX_LG_K;
use crate::thetacommon::constants::MAX_THETA;
use crate::thetacommon::constants::MIN_LG_K;
use crate::thetacommon::hash_table::SketchHashTableIter;
use crate::tuple::hash_table::TupleEntry;
use crate::tuple::hash_table::TupleHashTable;
use crate::tuple::policy::SummaryPolicy;
use crate::tuple::policy::SummaryUpdatePolicy;
use crate::tuple::serialization::SERIAL_VERSION;
use crate::tuple::serialization::SERIAL_VERSION_LEGACY;
use crate::tuple::serialization::SKETCH_TYPE;
use crate::tuple::serialization::SKETCH_TYPE_LEGACY;
use crate::tuple::serialization::TupleSummaryValue;

/// Read-only view of a mutable or compact Tuple sketch.
///
/// The view borrows the sketch without exposing its update policy. It can inspect keys without
/// requiring `S: Clone`; set operations that retain summaries require `S: Clone` when invoked.
///
/// # Examples
///
/// ```
/// use datasketches::tuple::DefaultUpdatePolicy;
/// use datasketches::tuple::TupleSketchBuilder;
///
/// let mut sketch = TupleSketchBuilder::new(DefaultUpdatePolicy::<u64>::default()).build();
/// sketch.update("apple", 1);
/// let view = sketch.as_view();
/// assert_eq!(view.iter().next().unwrap().1, &1);
/// ```
#[derive(Debug)]
pub struct TupleSketchView<'a, S>(TupleSketchViewState<'a, S>);

#[derive(Debug)]
enum TupleSketchViewState<'a, S> {
    Mutable(&'a TupleHashTable<S>),
    Compact(&'a CompactTupleSketch<S>),
}

enum TupleSketchIter<'a, S> {
    Mutable(SketchHashTableIter<'a, TupleEntry<S>>),
    Compact(slice::Iter<'a, TupleEntry<S>>),
}

impl<'a, S> Iterator for TupleSketchIter<'a, S> {
    type Item = (u64, &'a S);

    fn next(&mut self) -> Option<Self::Item> {
        match self {
            Self::Mutable(iter) => iter.next().map(|entry| (entry.hash(), entry.summary())),
            Self::Compact(iter) => iter.next().map(|entry| (entry.hash(), entry.summary())),
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        match self {
            Self::Mutable(iter) => iter.size_hint(),
            Self::Compact(iter) => iter.size_hint(),
        }
    }
}

impl<S> Clone for TupleSketchView<'_, S> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<S> Copy for TupleSketchView<'_, S> {}

impl<S> Clone for TupleSketchViewState<'_, S> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<S> Copy for TupleSketchViewState<'_, S> {}

impl<'a, S> TupleSketchView<'a, S> {
    /// Returns the 16-bit seed hash.
    pub fn seed_hash(&self) -> u16 {
        match self.0 {
            TupleSketchViewState::Mutable(table) => table.seed_hash(),
            TupleSketchViewState::Compact(sketch) => sketch.seed_hash(),
        }
    }

    /// Returns theta as a `u64` threshold.
    pub fn theta64(&self) -> u64 {
        match self.0 {
            TupleSketchViewState::Mutable(table) => table.theta(),
            TupleSketchViewState::Compact(sketch) => sketch.theta64(),
        }
    }

    /// Returns whether the viewed sketch has not received any updates.
    pub fn is_empty(&self) -> bool {
        match self.0 {
            TupleSketchViewState::Mutable(table) => table.is_empty(),
            TupleSketchViewState::Compact(sketch) => sketch.is_empty(),
        }
    }

    /// Returns whether retained entries are ordered by ascending hash.
    pub fn is_ordered(&self) -> bool {
        match self.0 {
            TupleSketchViewState::Mutable(_) => false,
            TupleSketchViewState::Compact(sketch) => sketch.is_ordered(),
        }
    }

    /// Returns an iterator over retained hashes and borrowed summaries.
    pub fn iter(self) -> impl Iterator<Item = (u64, &'a S)> + 'a {
        match self.0 {
            TupleSketchViewState::Mutable(table) => TupleSketchIter::Mutable(table.iter_entries()),
            TupleSketchViewState::Compact(sketch) => {
                TupleSketchIter::Compact(sketch.entries.iter())
            }
        }
    }

    /// Returns the number of retained entries.
    pub fn num_retained(&self) -> usize {
        match self.0 {
            TupleSketchViewState::Mutable(table) => table.num_retained(),
            TupleSketchViewState::Compact(sketch) => sketch.num_retained(),
        }
    }
}

impl<S> KeySketch for TupleSketchView<'_, S> {
    fn scalars(self) -> SketchScalars {
        SketchScalars {
            seed_hash: self.seed_hash(),
            theta: self.theta64(),
            empty: self.is_empty(),
            ordered: self.is_ordered(),
            num_retained: self.num_retained(),
        }
    }

    fn hashes(self) -> impl Iterator<Item = u64> {
        self.iter().map(|(hash, _)| hash)
    }
}

impl<'a, S> EntrySketch for TupleSketchView<'a, S>
where
    S: Clone + 'a,
{
    type Entry = TupleEntry<S>;

    fn entries(self) -> impl Iterator<Item = Self::Entry> {
        self.iter()
            .map(|(hash, summary)| TupleEntry::new(hash, summary.clone()))
    }
}

impl<'a, P> From<&'a TupleSketch<P>> for TupleSketchView<'a, P::Summary>
where
    P: SummaryPolicy,
{
    fn from(sketch: &'a TupleSketch<P>) -> Self {
        Self(TupleSketchViewState::Mutable(&sketch.table))
    }
}

impl<'a, S> From<&'a CompactTupleSketch<S>> for TupleSketchView<'a, S> {
    fn from(sketch: &'a CompactTupleSketch<S>) -> Self {
        Self(TupleSketchViewState::Compact(sketch))
    }
}

/// Mutable Tuple sketch for building from input data.
///
/// `P` defines how summaries are created. The summary retained alongside each key is
/// [`P::Summary`](SummaryPolicy::Summary), while each accepted update type is selected by a
/// [`SummaryUpdatePolicy<U>`] implementation.
///
/// # Examples
///
/// ```
/// use datasketches::tuple::DefaultUpdatePolicy;
/// use datasketches::tuple::TupleSketchBuilder;
///
/// let policy = DefaultUpdatePolicy::<u64>::default();
/// let mut sketch = TupleSketchBuilder::new(policy).build();
/// sketch.update("apple", 1);
/// sketch.update("apple", 1);
/// assert!(sketch.estimate() >= 1.0);
/// assert_eq!(sketch.num_retained(), 1);
/// ```
#[derive(Debug)]
pub struct TupleSketch<P>
where
    P: SummaryPolicy,
{
    table: TupleHashTable<P::Summary>,
    policy: P,
}

impl<P> TupleSketch<P>
where
    P: SummaryPolicy,
{
    /// Returns a read-only view accepted by Tuple set operations.
    pub fn as_view(&self) -> TupleSketchView<'_, P::Summary> {
        self.into()
    }

    /// Updates the sketch with a key and a value accepted by the policy.
    ///
    /// If the key is new, the policy creates a summary and folds in `value`; if the key already
    /// exists, `value` is folded into the retained summary. Updates screened out by theta do not
    /// change any summary.
    ///
    /// # Examples
    ///
    /// ```
    /// use datasketches::tuple::DefaultUpdatePolicy;
    /// use datasketches::tuple::TupleSketchBuilder;
    ///
    /// let policy = DefaultUpdatePolicy::<u64>::default();
    /// let mut sketch = TupleSketchBuilder::new(policy).build();
    /// sketch.update(42, 5);
    /// ```
    pub fn update<U>(&mut self, key: impl Hash, value: U)
    where
        P: SummaryUpdatePolicy<U>,
    {
        let policy = &self.policy;
        self.table.try_insert(key, |existing| match existing {
            Some(summary) => {
                policy.update(summary, value);
                None
            }
            None => {
                let mut summary = policy.create();
                policy.update(&mut summary, value);
                Some(summary)
            }
        });
    }

    /// Returns the cardinality (distinct key count) estimate.
    pub fn estimate(&self) -> f64 {
        if self.is_empty() {
            return 0.0;
        }
        let num_retained = self.table.num_retained() as f64;
        let theta = self.table.theta() as f64 / MAX_THETA as f64;
        num_retained / theta
    }

    /// Returns theta as a fraction in `[0.0, 1.0]`.
    pub fn theta(&self) -> f64 {
        self.table.theta() as f64 / MAX_THETA as f64
    }

    /// Returns theta as a `u64`.
    pub fn theta64(&self) -> u64 {
        self.table.theta()
    }

    /// Returns the 16-bit seed hash.
    pub fn seed_hash(&self) -> u16 {
        self.table.seed_hash()
    }

    /// Returns `true` if the sketch is empty.
    pub fn is_empty(&self) -> bool {
        self.table.is_empty()
    }

    /// Returns `true` if the sketch is in estimation mode.
    pub fn is_estimation_mode(&self) -> bool {
        self.table.theta() < MAX_THETA
    }

    /// Returns the number of retained entries.
    pub fn num_retained(&self) -> usize {
        self.table.num_retained()
    }

    /// Returns the configured `lg_k`.
    pub fn lg_k(&self) -> u8 {
        self.table.lg_nom_size()
    }

    /// Trims the sketch to the capacity configured by `lg_k`.
    pub fn trim(&mut self) {
        self.table.trim();
    }

    /// Resets the sketch to the empty state.
    pub fn reset(&mut self) {
        self.table.reset();
    }

    /// Returns an iterator over retained entries as `(hash, &summary)` pairs.
    pub fn iter(&self) -> impl Iterator<Item = (u64, &P::Summary)> + '_ {
        self.table.iter()
    }

    /// Returns the approximate lower error bound given the number of standard deviations.
    pub fn lower_bound(&self, num_std_dev: NumStdDev) -> f64 {
        if !self.is_estimation_mode() {
            return self.num_retained() as f64;
        }
        binomial_bounds::lower_bound(self.num_retained() as u64, self.theta(), num_std_dev)
            .expect("theta should always be valid")
    }

    /// Returns the approximate upper error bound given the number of standard deviations.
    pub fn upper_bound(&self, num_std_dev: NumStdDev) -> f64 {
        if !self.is_estimation_mode() {
            return self.num_retained() as f64;
        }
        binomial_bounds::upper_bound(
            self.num_retained() as u64,
            self.theta(),
            num_std_dev,
            self.is_empty(),
        )
        .expect("theta should always be valid")
    }

    /// Returns the estimated size of the sketch in bytes.
    pub fn estimated_size(&self) -> usize {
        size_of::<Self>() + self.table.estimated_size()
    }
}

impl<P> TupleSketch<P>
where
    P: SummaryPolicy,
    P::Summary: Clone,
{
    /// Returns this sketch in compact, immutable form.
    ///
    /// If `ordered` is `true`, retained entries are sorted by hash in ascending order.
    ///
    /// # Examples
    ///
    /// ```
    /// use datasketches::tuple::DefaultUpdatePolicy;
    /// use datasketches::tuple::TupleSketchBuilder;
    ///
    /// let policy = DefaultUpdatePolicy::<u64>::default();
    /// let mut sketch = TupleSketchBuilder::new(policy).build();
    /// sketch.update("apple", 1);
    /// let compact = sketch.compact(true);
    /// assert_eq!(compact.num_retained(), 1);
    /// ```
    pub fn compact(&self, ordered: bool) -> CompactTupleSketch<P::Summary> {
        let parts = self.table.to_compact_parts(ordered);
        CompactTupleSketch::from_parts(
            parts.entries,
            parts.theta,
            parts.seed_hash,
            parts.ordered,
            parts.empty,
        )
    }
}

/// Compact (immutable) Tuple sketch.
///
/// This is the serialization-friendly form: a compact array of retained [`TupleEntry`] values
/// (hash plus summary) plus theta and a 16-bit seed hash. It can be ordered (sorted ascending by
/// hash) or unordered.
#[derive(Clone, Debug)]
pub struct CompactTupleSketch<S> {
    entries: Vec<TupleEntry<S>>,
    theta: u64,
    seed_hash: u16,
    ordered: bool,
    empty: bool,
}

impl<S> CompactTupleSketch<S> {
    pub(super) fn from_parts(
        entries: Vec<TupleEntry<S>>,
        theta: u64,
        seed_hash: u16,
        ordered: bool,
        empty: bool,
    ) -> Self {
        Self {
            entries,
            theta,
            seed_hash,
            ordered,
            empty,
        }
    }

    /// Returns a read-only view accepted by Tuple set operations.
    pub fn as_view(&self) -> TupleSketchView<'_, S> {
        self.into()
    }

    /// Returns the cardinality (distinct key count) estimate.
    pub fn estimate(&self) -> f64 {
        if self.is_empty() {
            return 0.0;
        }
        let num_retained = self.num_retained() as f64;
        if self.theta == MAX_THETA {
            return num_retained;
        }
        let theta = self.theta as f64 / MAX_THETA as f64;
        num_retained / theta
    }

    /// Returns theta as a fraction (0.0 to 1.0).
    pub fn theta(&self) -> f64 {
        self.theta as f64 / MAX_THETA as f64
    }

    /// Returns theta as `u64`.
    pub fn theta64(&self) -> u64 {
        self.theta
    }

    /// Returns `true` if the sketch is empty.
    pub fn is_empty(&self) -> bool {
        self.empty
    }

    /// Returns `true` if the sketch is in estimation mode.
    pub fn is_estimation_mode(&self) -> bool {
        self.theta < MAX_THETA
    }

    /// Returns the number of retained entries.
    pub fn num_retained(&self) -> usize {
        self.entries.len()
    }

    /// Returns `true` if retained entries are ordered (sorted ascending by hash).
    pub fn is_ordered(&self) -> bool {
        self.ordered
    }

    /// Returns the 16-bit seed hash.
    pub fn seed_hash(&self) -> u16 {
        self.seed_hash
    }

    /// Returns an iterator over retained entries as `(hash, &summary)` pairs.
    pub fn iter(&self) -> impl Iterator<Item = (u64, &S)> + '_ {
        self.entries
            .iter()
            .map(|entry| (entry.hash(), entry.summary()))
    }

    /// Returns the approximate lower error bound given the number of standard deviations.
    pub fn lower_bound(&self, num_std_dev: NumStdDev) -> f64 {
        if !self.is_estimation_mode() {
            return self.num_retained() as f64;
        }
        binomial_bounds::lower_bound(self.num_retained() as u64, self.theta(), num_std_dev)
            .expect("compact theta should always be valid")
    }

    /// Returns the approximate upper error bound given the number of standard deviations.
    pub fn upper_bound(&self, num_std_dev: NumStdDev) -> f64 {
        if !self.is_estimation_mode() {
            return self.num_retained() as f64;
        }
        binomial_bounds::upper_bound(
            self.num_retained() as u64,
            self.theta(),
            num_std_dev,
            self.is_empty(),
        )
        .expect("compact theta should always be valid")
    }

    /// Returns the estimated size of the sketch in bytes.
    pub fn estimated_size(&self) -> usize {
        size_of::<Self>() + self.entries.capacity() * size_of::<TupleEntry<S>>()
    }

    fn preamble_longs(&self) -> u8 {
        if self.is_estimation_mode() {
            3
        } else if self.is_empty() || self.entries.len() == 1 {
            1
        } else {
            2
        }
    }

    /// Serializes this sketch into the compact Tuple binary format.
    ///
    /// Each summary is encoded by its [`TupleSummaryValue`] implementation. The layout matches the
    /// Java/C++ Tuple sketches, so the output can be read by those implementations given a
    /// compatible summary encoding.
    ///
    /// # Examples
    ///
    /// ```
    /// use datasketches::tuple::DefaultUpdatePolicy;
    /// use datasketches::tuple::TupleSketchBuilder;
    ///
    /// let policy = DefaultUpdatePolicy::<u64>::default();
    /// let mut sketch = TupleSketchBuilder::new(policy).build();
    /// sketch.update("apple", 1);
    /// let bytes = sketch.compact(true).serialize();
    /// assert!(!bytes.is_empty());
    /// ```
    pub fn serialize(&self) -> Vec<u8>
    where
        S: TupleSummaryValue,
    {
        let pre_longs = self.preamble_longs();
        let entries_size: usize = self
            .entries
            .iter()
            .map(|entry| 8 + entry.summary().serialize_size())
            .sum();
        let mut bytes = SketchBytes::with_capacity(8 * pre_longs as usize + entries_size);

        bytes.write_u8(pre_longs);
        bytes.write_u8(SERIAL_VERSION);
        bytes.write_u8(Family::TUPLE.id);
        bytes.write_u8(SKETCH_TYPE);
        bytes.write_u8(0); // unused

        let mut flags = FLAGS_IS_READ_ONLY | FLAGS_IS_COMPACT;
        if self.is_empty() {
            flags |= FLAGS_IS_EMPTY;
        }
        if self.is_ordered() {
            flags |= FLAGS_IS_ORDERED;
        }
        bytes.write_u8(flags);
        bytes.write_u16_le(self.seed_hash);

        if pre_longs > 1 {
            bytes.write_u32_le(self.entries.len() as u32);
            bytes.write_u32_le(0); // unused
        }
        if self.is_estimation_mode() {
            bytes.write_u64_le(self.theta);
        }

        for entry in &self.entries {
            bytes.write_u64_le(entry.hash());
            entry.summary().serialize_value(&mut bytes);
        }
        bytes.into_bytes()
    }

    /// Deserializes a compact Tuple sketch using the default seed.
    pub fn deserialize(bytes: &[u8]) -> Result<Self, Error>
    where
        S: TupleSummaryValue,
    {
        Self::deserialize_with_seed(bytes, DEFAULT_UPDATE_SEED)
    }

    /// Deserializes a compact Tuple sketch using the provided expected `seed`.
    ///
    /// # Errors
    ///
    /// Returns an error if the bytes are truncated, the family/serial version/sketch type are
    /// unexpected, the seed hash does not match (for non-empty sketches), or an entry is corrupted.
    pub fn deserialize_with_seed(bytes: &[u8], seed: u64) -> Result<Self, Error>
    where
        S: TupleSummaryValue,
    {
        let mut cursor = SketchSlice::new(bytes);
        let pre_longs = cursor
            .read_u8()
            .map_err(insufficient_data("preamble_longs"))?;
        let ser_ver = cursor
            .read_u8()
            .map_err(insufficient_data("serial_version"))?;
        let family_id = cursor.read_u8().map_err(insufficient_data("family_id"))?;
        let sketch_type = cursor.read_u8().map_err(insufficient_data("sketch_type"))?;
        cursor.read_u8().map_err(insufficient_data("<unused>"))?;
        let flags = cursor.read_u8().map_err(insufficient_data("flags"))?;
        let seed_hash = cursor
            .read_u16_le()
            .map_err(insufficient_data("seed_hash"))?;

        Family::TUPLE.validate_id(family_id)?;
        ensure_preamble_longs_in_range(
            Family::TUPLE.min_pre_longs..=Family::TUPLE.max_pre_longs,
            pre_longs,
        )?;
        if ser_ver != SERIAL_VERSION && ser_ver != SERIAL_VERSION_LEGACY {
            return Err(Error::deserial(format!(
                "unsupported serial version: expected {} or {}, got {ser_ver}",
                SERIAL_VERSION, SERIAL_VERSION_LEGACY,
            )));
        }
        if sketch_type != SKETCH_TYPE && sketch_type != SKETCH_TYPE_LEGACY {
            return Err(Error::deserial(format!(
                "unsupported sketch type: expected {} or {}, got {sketch_type}",
                SKETCH_TYPE, SKETCH_TYPE_LEGACY,
            )));
        }

        let empty = (flags & FLAGS_IS_EMPTY) != 0;
        let ordered = (flags & FLAGS_IS_ORDERED) != 0;

        if empty {
            return Ok(Self::from_parts(
                vec![],
                MAX_THETA,
                seed_hash,
                ordered,
                true,
            ));
        }

        check_seed_hash(
            compute_seed_hash(seed),
            seed_hash,
            "deserialized CompactTupleSketch",
            ErrorKind::InvalidData,
        )?;

        let mut theta = MAX_THETA;
        let num_entries = if pre_longs == 1 {
            1usize
        } else {
            let n = cursor
                .read_u32_le()
                .map_err(insufficient_data("num_entries"))? as usize;
            cursor
                .read_u32_le()
                .map_err(insufficient_data("<unused_u32>"))?;
            if pre_longs > 2 {
                theta = cursor.read_u64_le().map_err(insufficient_data("theta"))?;
            }
            n
        };

        let mut entries = Vec::with_capacity(num_entries);
        for _ in 0..num_entries {
            let hash = cursor
                .read_u64_le()
                .map_err(insufficient_data("entry_hash"))?;
            if hash == 0 || hash >= theta {
                return Err(Error::deserial("corrupted: invalid retained hash value"));
            }
            let summary = S::deserialize_value(&mut cursor)?;
            entries.push(TupleEntry::new(hash, summary));
        }

        Ok(Self::from_parts(entries, theta, seed_hash, ordered, false))
    }
}

/// Builder for [`TupleSketch`].
///
/// Every builder carries a concrete [`SummaryPolicy`]. Use
/// [`DefaultUpdatePolicy`](crate::tuple::DefaultUpdatePolicy) for default-constructed additive
/// summaries, or supply a custom policy.
#[derive(Debug)]
pub struct TupleSketchBuilder<P>
where
    P: SummaryPolicy,
{
    lg_k: u8,
    resize_factor: ResizeFactor,
    sampling_probability: f32,
    seed: u64,
    policy: P,
}

impl<P> TupleSketchBuilder<P>
where
    P: SummaryPolicy,
{
    /// Creates a builder with the given summary policy.
    ///
    /// # Examples
    ///
    /// ```
    /// use datasketches::tuple::SummaryPolicy;
    /// use datasketches::tuple::SummaryUpdatePolicy;
    /// use datasketches::tuple::TupleSketchBuilder;
    ///
    /// struct MaxPolicy;
    ///
    /// impl SummaryPolicy for MaxPolicy {
    ///     type Summary = u64;
    ///
    ///     fn create(&self) -> Self::Summary {
    ///         0
    ///     }
    /// }
    ///
    /// impl SummaryUpdatePolicy<u64> for MaxPolicy {
    ///     fn update(&self, summary: &mut Self::Summary, value: u64) {
    ///         *summary = (*summary).max(value);
    ///     }
    /// }
    ///
    /// let mut sketch = TupleSketchBuilder::new(MaxPolicy).build();
    /// sketch.update("k", 3);
    /// sketch.update("k", 7);
    /// ```
    pub fn new(policy: P) -> Self {
        Self {
            lg_k: DEFAULT_LG_K,
            resize_factor: ResizeFactor::X8,
            sampling_probability: 1.0,
            seed: DEFAULT_UPDATE_SEED,
            policy,
        }
    }

    /// Sets `lg_k`, the base-2 logarithm of the nominal capacity.
    ///
    /// # Panics
    ///
    /// Panics if `lg_k` is outside `[5, 26]`.
    pub fn lg_k(mut self, lg_k: u8) -> Self {
        assert!(
            (MIN_LG_K..=MAX_LG_K).contains(&lg_k),
            "lg_k must be in [{MIN_LG_K}, {MAX_LG_K}], got {lg_k}"
        );
        self.lg_k = lg_k;
        self
    }

    /// Sets the resize factor.
    pub fn resize_factor(mut self, factor: ResizeFactor) -> Self {
        self.resize_factor = factor;
        self
    }

    /// Sets the sampling probability.
    ///
    /// # Panics
    ///
    /// Panics if `probability` is outside `(0.0, 1.0]`.
    pub fn sampling_probability(mut self, probability: f32) -> Self {
        assert!(
            (0.0..=1.0).contains(&probability) && probability > 0.0,
            "sampling_probability must be in (0.0, 1.0], got {probability}"
        );
        self.sampling_probability = probability;
        self
    }

    /// Sets the hash seed.
    pub fn seed(mut self, seed: u64) -> Self {
        self.seed = seed;
        self
    }

    /// Builds a [`TupleSketch`] using the supplied policy.
    pub fn build(self) -> TupleSketch<P> {
        TupleSketch {
            table: TupleHashTable::new(
                self.lg_k,
                self.resize_factor,
                self.sampling_probability,
                self.seed,
            ),
            policy: self.policy,
        }
    }
}