diff --git a/src/common/structures/vector3d.rs b/src/common/structures/vector3d.rs
index 46ceb99..1743cb9 100644
--- a/src/common/structures/vector3d.rs
+++ b/src/common/structures/vector3d.rs
@@ -1,22 +1,627 @@
 use serde::{Deserialize, Serialize};
+use std::ops::{Index, IndexMut};
 
-/// Shared thin wrapper over `Vec<T>` tailored for sketches.
-#[derive(Clone, Debug, Serialize, Deserialize)]
+use crate::{MatrixFastHash, MatrixHashType, Nitro, compute_median_inline_f64};
+
+/// Shared thin wrapper over `Vec<T>` tailored for layered / per-bucket sketches.
+///
+/// `Vector3D` is the three-dimensional sibling of [`crate::Vector2D`]. It models
+/// a `rows * cols` grid where **every `(row, col)` cell is itself a contiguous
+/// run of `depth` elements** — a "bucket". This is the natural storage for
+/// sketches that keep a small vector (e.g. a HyperLogLog register array) at each
+/// matrix position.
+///
+/// Storage is a single flat `Vec<T>` in row-major / bucket-major order; the
+/// element at `(row, col, d)` lives at `(row * cols + col) * depth + d`.
+///
+/// The row/column addressing (mask bits, `col_for_row`, hashing) mirrors
+/// [`crate::Vector2D`] exactly, so the same `MatrixFastHash` machinery selects a
+/// column per row; the third dimension is addressed within the selected bucket.
+#[derive(Clone, Debug, Serialize)]
 pub struct Vector3D<T> {
     data: Vec<T>,
-    layer: usize,
-    row: usize,
-    col: usize,
+    rows: usize,
+    cols: usize,
+    depth: usize,
+    mask_bits: u32,
+    mask: u128,
+    nitro: Nitro,
+}
+
+// Helper type for deserialization: we only read stored fields and recompute
+// derived ones (mask_bits, mask) from cols, mirroring `Vector2D`.
+#[derive(Deserialize)]
+struct Vector3DDeserialize<T> {
+    data: Vec<T>,
+    rows: usize,
+    cols: usize,
+    depth: usize,
+    #[serde(default)]
+    nitro: Nitro,
+}
+
+#[inline]
+fn mask_bits_for_cols(cols: usize) -> u32 {
+    if cols.is_power_of_two() {
+        cols.ilog2()
+    } else {
+        cols.ilog2() + 1
+    }
+}
+
+impl<'de, T> Deserialize<'de> for Vector3D<T>
+where
+    T: Deserialize<'de>,
+{
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let input = Vector3DDeserialize::deserialize(deserializer)?;
+        let mask_bits = mask_bits_for_cols(input.cols);
+        let mask = (1u128 << mask_bits) - 1;
+        Ok(Self {
+            data: input.data,
+            rows: input.rows,
+            cols: input.cols,
+            depth: input.depth,
+            mask_bits,
+            mask,
+            nitro: input.nitro,
+        })
+    }
 }
 
 impl<T> Vector3D<T> {
-    /// Creates an empty 3D container with reserved capacity.
-    pub fn init(layer: usize, row: usize, col: usize) -> Self {
+    /// Creates an empty container with reserved capacity for `rows * cols * depth`
+    /// elements. The underlying storage is left uninitialized until `fill` or
+    /// similar methods are called, allowing callers to decide when and how cells
+    /// are populated.
+    pub fn init(rows: usize, cols: usize, depth: usize) -> Self {
+        let mask_bits = mask_bits_for_cols(cols);
+        let mask = (1u128 << mask_bits) - 1;
+        Self {
+            data: Vec::with_capacity(rows * cols * depth),
+            rows,
+            cols,
+            depth,
+            mask_bits,
+            mask,
+            nitro: Nitro::default(),
+        }
+    }
+
+    /// Builds a container by invoking a generator for every `(row, col, d)`
+    /// position. Useful for types that require per-cell construction logic
+    /// instead of cloning a single value across all cells.
+    pub fn from_fn<F>(rows: usize, cols: usize, depth: usize, mut f: F) -> Self
+    where
+        F: FnMut(usize, usize, usize) -> T,
+    {
+        let mask_bits = mask_bits_for_cols(cols);
+        let mask = (1u128 << mask_bits) - 1;
+        let mut data = Vec::with_capacity(rows * cols * depth);
+        for r in 0..rows {
+            for c in 0..cols {
+                for d in 0..depth {
+                    data.push(f(r, c, d));
+                }
+            }
+        }
         Self {
-            data: Vec::with_capacity(layer * row * col),
-            layer,
-            row,
-            col,
+            data,
+            rows,
+            cols,
+            depth,
+            mask_bits,
+            mask,
+            nitro: Nitro::default(),
+        }
+    }
+
+    /// Enables Nitro sampling with the provided rate.
+    pub fn enable_nitro(&mut self, sampling_rate: f64) {
+        self.nitro = Nitro::init_nitro(sampling_rate);
+    }
+
+    /// Disables Nitro sampling and resets the internal state.
+    pub fn disable_nitro(&mut self) {
+        self.nitro = Nitro::default();
+    }
+
+    #[inline(always)]
+    /// Decrements the Nitro skip counter by one.
+    pub fn reduce_to_skip(&mut self) {
+        self.nitro.reduce_to_skip();
+    }
+
+    /// Returns the Nitro configuration.
+    #[inline(always)]
+    pub fn nitro(&self) -> &Nitro {
+        &self.nitro
+    }
+
+    #[inline(always)]
+    /// Returns the current Nitro delta weight.
+    pub fn get_delta(&self) -> u64 {
+        self.nitro.delta
+    }
+
+    /// Returns a mutable Nitro configuration reference.
+    #[inline(always)]
+    pub fn nitro_mut(&mut self) -> &mut Nitro {
+        &mut self.nitro
+    }
+
+    /// Replaces the entire container with `rows * cols * depth` clones of `value`,
+    /// reusing the existing allocation. This is the most efficient way to reset
+    /// cells to a baseline without reallocating.
+    pub fn fill(&mut self, value: T)
+    where
+        T: Clone,
+    {
+        self.data.clear();
+        self.data.resize(self.rows * self.cols * self.depth, value);
+    }
+
+    #[inline(always)]
+    fn col_for_row<Hash: MatrixFastHash>(&self, hashed_val: &Hash, row: usize) -> usize {
+        hashed_val.col_for_row(row, self.cols)
+    }
+
+    #[inline(always)]
+    fn bucket_start(&self, row: usize, col: usize) -> usize {
+        (row * self.cols + col) * self.depth
+    }
+
+    /// Returns the number of rows.
+    #[inline(always)]
+    pub fn rows(&self) -> usize {
+        self.rows
+    }
+
+    /// Returns the number of columns.
+    #[inline(always)]
+    pub fn cols(&self) -> usize {
+        self.cols
+    }
+
+    /// Returns the per-bucket depth (length of each `(row, col)` cell).
+    #[inline(always)]
+    pub fn depth(&self) -> usize {
+        self.depth
+    }
+
+    /// Allocates one extra row initialized with `value`.
+    pub fn allocate_extra_row(&mut self, value: T)
+    where
+        T: Clone,
+    {
+        self.rows += 1;
+        self.data.resize(self.rows * self.cols * self.depth, value);
+    }
+
+    /// Returns the total number of elements.
+    #[inline(always)]
+    pub fn len(&self) -> usize {
+        self.data.len()
+    }
+
+    #[inline(always)]
+    /// Returns `true` when the container stores no elements.
+    pub fn is_empty(&self) -> bool {
+        self.data.is_empty()
+    }
+
+    /// Provides immutable access to the flattened storage.
+    #[inline(always)]
+    pub fn as_slice(&self) -> &[T] {
+        &self.data
+    }
+
+    /// Provides mutable access to the flattened storage.
+    #[inline(always)]
+    pub fn as_mut_slice(&mut self) -> &mut [T] {
+        &mut self.data
+    }
+
+    /// Returns a reference to a single element when it exists.
+    #[inline(always)]
+    pub fn get(&self, row: usize, col: usize, d: usize) -> Option<&T> {
+        if row < self.rows && col < self.cols && d < self.depth {
+            Some(&self.data[self.bucket_start(row, col) + d])
+        } else {
+            None
+        }
+    }
+
+    /// Returns a mutable reference to a single element when it exists.
+    #[inline(always)]
+    pub fn get_mut(&mut self, row: usize, col: usize, d: usize) -> Option<&mut T> {
+        if row < self.rows && col < self.cols && d < self.depth {
+            let idx = self.bucket_start(row, col) + d;
+            Some(&mut self.data[idx])
+        } else {
+            None
+        }
+    }
+
+    /// Returns the `(row, col)` bucket slice when it exists.
+    #[inline(always)]
+    pub fn bucket(&self, row: usize, col: usize) -> Option<&[T]> {
+        if row < self.rows && col < self.cols {
+            let start = self.bucket_start(row, col);
+            Some(&self.data[start..start + self.depth])
+        } else {
+            None
+        }
+    }
+
+    /// Returns the `(row, col)` bucket slice mutably when it exists.
+    #[inline(always)]
+    pub fn bucket_mut(&mut self, row: usize, col: usize) -> Option<&mut [T]> {
+        if row < self.rows && col < self.cols {
+            let start = self.bucket_start(row, col);
+            Some(&mut self.data[start..start + self.depth])
+        } else {
+            None
+        }
+    }
+
+    /// Returns the `(row, col)` bucket slice, debug-asserting bounds.
+    /// Faster sibling of [`Self::bucket`].
+    #[inline(always)]
+    pub fn bucket_slice(&self, row: usize, col: usize) -> &[T] {
+        debug_assert!(row < self.rows && col < self.cols, "bucket out of bounds");
+        let start = self.bucket_start(row, col);
+        &self.data[start..start + self.depth]
+    }
+
+    /// Mutable sibling of [`Self::bucket_slice`].
+    #[inline(always)]
+    pub fn bucket_slice_mut(&mut self, row: usize, col: usize) -> &mut [T] {
+        debug_assert!(row < self.rows && col < self.cols, "bucket out of bounds");
+        let start = self.bucket_start(row, col);
+        &mut self.data[start..start + self.depth]
+    }
+
+    /// Applies an update to a single element via the supplied operator.
+    #[inline(always)]
+    pub fn update_one_counter<F, V>(&mut self, row: usize, col: usize, d: usize, op: F, value: V)
+    where
+        F: Fn(&mut T, V),
+    {
+        let idx = self.bucket_start(row, col) + d;
+        op(&mut self.data[idx], value);
+    }
+
+    /// get the number of bits required to cover the col size
+    #[inline(always)]
+    /// Returns the bit width needed to represent a column index.
+    pub fn get_mask_bits(&self) -> u32 {
+        mask_bits_for_cols(self.cols)
+    }
+
+    /// get the number of bits required for hashed value
+    /// only three case possible: 32, 64, 128
+    #[inline]
+    /// Returns the packed hash width needed for all rows.
+    pub fn get_required_bits(&self) -> usize {
+        let mut bits_required = self.get_mask_bits() as usize;
+        bits_required *= self.rows;
+        bits_required = 32 << ((bits_required > 32) as u32 + (bits_required > 64) as u32);
+        bits_required = bits_required.min(128);
+        bits_required
+    }
+
+    /// Inserts along every row using a hashed column selection.
+    ///
+    /// For each row a column is selected from `hashed_val`, yielding one
+    /// `(row, col)` bucket; the closure receives that **bucket slice**, the
+    /// value, and the current row index. This is the three-dimensional analogue
+    /// of [`crate::Vector2D::fast_insert`], where the per-row target is a whole
+    /// bucket rather than a single counter.
+    #[inline(always)]
+    pub fn fast_insert<Hash, F, V>(&mut self, op: F, value: V, hashed_val: &Hash)
+    where
+        Hash: MatrixFastHash,
+        F: Fn(&mut [T], &V, usize),
+        V: Clone,
+    {
+        for row in 0..self.rows {
+            let col = self.col_for_row(hashed_val, row);
+            let start = self.bucket_start(row, col);
+            let end = start + self.depth;
+            op(&mut self.data[start..end], &value, row);
+        }
+    }
+
+    #[inline(always)]
+    /// Decrements the Nitro skip counter by `c`.
+    pub fn reduce_nitro_skip(&mut self, c: usize) {
+        self.nitro.reduce_to_skip_by_count(c)
+    }
+
+    #[inline(always)]
+    /// Sets the Nitro skip counter to `c`.
+    pub fn update_nitro_skip(&mut self, c: usize) {
+        self.nitro.to_skip = c
+    }
+
+    #[inline(always)]
+    /// Returns the current Nitro skip counter.
+    pub fn get_nitro_skip(&mut self) -> usize {
+        self.nitro.to_skip
+    }
+
+    /// Reads a single element by `(row, col, d)`.
+    #[inline(always)]
+    pub fn query_one_counter(&self, row: usize, col: usize, d: usize) -> T
+    where
+        T: Clone,
+    {
+        self.data[self.bucket_start(row, col) + d].clone()
+    }
+
+    /// Queries all rows using precomputed hashed values to find the minimum.
+    ///
+    /// The closure receives: bucket slice, row index, and hash value.
+    #[inline(always)]
+    pub fn fast_query_min<Hash, F, R>(&self, hashed_val: &Hash, op: F) -> R
+    where
+        Hash: MatrixFastHash,
+        F: Fn(&[T], usize, &Hash) -> R,
+        R: PartialOrd,
+    {
+        let c0 = self.col_for_row(hashed_val, 0);
+        let mut min = op(self.bucket_slice(0, c0), 0, hashed_val);
+        for row in 1..self.rows {
+            let col = self.col_for_row(hashed_val, row);
+            let candidate = op(self.bucket_slice(row, col), row, hashed_val);
+            if candidate < min {
+                min = candidate;
+            }
+        }
+        min
+    }
+
+    /// Queries all rows using precomputed hashed values to find the median.
+    ///
+    /// The closure receives: bucket slice, row index, and hash value, and
+    /// returns `f64` values which are collected and reduced to a median.
+    #[inline(always)]
+    pub fn fast_query_median<Hash, F>(&self, hashed_val: &Hash, op: F) -> f64
+    where
+        Hash: MatrixFastHash,
+        F: Fn(&[T], usize, &Hash) -> f64,
+    {
+        let mut estimates = Vec::with_capacity(self.rows);
+        for row in 0..self.rows {
+            let col = self.col_for_row(hashed_val, row);
+            estimates.push(op(self.bucket_slice(row, col), row, hashed_val));
+        }
+        compute_median_inline_f64(&mut estimates)
+    }
+
+    /// Queries all rows using precomputed hashed values to find the maximum.
+    ///
+    /// The closure receives: bucket slice, row index, and hash value.
+    #[inline(always)]
+    pub fn fast_query_max<F, R>(&self, hashed_val: &MatrixHashType, op: F) -> R
+    where
+        F: Fn(&[T], usize, &MatrixHashType) -> R,
+        R: PartialOrd,
+    {
+        let c0 = self.col_for_row(hashed_val, 0);
+        let mut max = op(self.bucket_slice(0, c0), 0, hashed_val);
+        for row in 1..self.rows {
+            let col = self.col_for_row(hashed_val, row);
+            let candidate = op(self.bucket_slice(row, col), row, hashed_val);
+            if candidate > max {
+                max = candidate;
+            }
+        }
+        max
+    }
+
+    /// Queries all rows to find the minimum with a query key.
+    ///
+    /// The closure receives: bucket slice, query key, row index, and hash value.
+    #[inline(always)]
+    pub fn fast_query_min_with_key<F, Q, R>(
+        &self,
+        hashed_val: &MatrixHashType,
+        query_key: &Q,
+        op: F,
+    ) -> R
+    where
+        F: Fn(&[T], &Q, usize, &MatrixHashType) -> R,
+        R: PartialOrd,
+    {
+        let c0 = self.col_for_row(hashed_val, 0);
+        let mut min = op(self.bucket_slice(0, c0), query_key, 0, hashed_val);
+        for row in 1..self.rows {
+            let col = self.col_for_row(hashed_val, row);
+            let candidate = op(self.bucket_slice(row, col), query_key, row, hashed_val);
+            if candidate < min {
+                min = candidate;
+            }
+        }
+        min
+    }
+
+    /// Queries all rows to find the maximum with a query key.
+    ///
+    /// The closure receives: bucket slice, query key, row index, and hash value.
+    #[inline(always)]
+    pub fn fast_query_max_with_key<F, Q, R>(
+        &self,
+        hashed_val: &MatrixHashType,
+        query_key: &Q,
+        op: F,
+    ) -> R
+    where
+        F: Fn(&[T], &Q, usize, &MatrixHashType) -> R,
+        R: PartialOrd,
+    {
+        let c0 = self.col_for_row(hashed_val, 0);
+        let mut max = op(self.bucket_slice(0, c0), query_key, 0, hashed_val);
+        for row in 1..self.rows {
+            let col = self.col_for_row(hashed_val, row);
+            let candidate = op(self.bucket_slice(row, col), query_key, row, hashed_val);
+            if candidate > max {
+                max = candidate;
+            }
+        }
+        max
+    }
+
+    /// Queries all rows to find the median with a query key.
+    ///
+    /// The closure receives: bucket slice, query key, row index, and hash value.
+    #[inline(always)]
+    pub fn fast_query_median_with_key<F, Q>(
+        &self,
+        hashed_val: &MatrixHashType,
+        query_key: &Q,
+        op: F,
+    ) -> f64
+    where
+        F: Fn(&[T], &Q, usize, &MatrixHashType) -> f64,
+    {
+        let mut estimates = Vec::with_capacity(self.rows);
+        for row in 0..self.rows {
+            let col = self.col_for_row(hashed_val, row);
+            estimates.push(op(self.bucket_slice(row, col), query_key, row, hashed_val));
         }
+        compute_median_inline_f64(&mut estimates)
+    }
+
+    /// Queries all rows with custom aggregation logic (fold/reduce pattern).
+    ///
+    /// The closure receives: accumulator, bucket slice, query key, row index, and
+    /// hash value.
+    #[inline(always)]
+    pub fn fast_query_aggregate<F, Q, R>(
+        &self,
+        hashed_val: &MatrixHashType,
+        query_key: &Q,
+        init: R,
+        fold_fn: F,
+    ) -> R
+    where
+        F: Fn(R, &[T], &Q, usize, &MatrixHashType) -> R,
+    {
+        let mut acc = init;
+        for row in 0..self.rows {
+            let col = self.col_for_row(hashed_val, row);
+            acc = fold_fn(acc, self.bucket_slice(row, col), query_key, row, hashed_val);
+        }
+        acc
+    }
+
+    /// Returns an immutable slice corresponding to a full row plane
+    /// (`cols * depth` elements).
+    #[inline(always)]
+    pub fn row_slice(&self, row: usize) -> &[T] {
+        debug_assert!(row < self.rows, "row index out of bounds");
+        let start = row * self.cols * self.depth;
+        let end = start + self.cols * self.depth;
+        &self.data[start..end]
+    }
+
+    /// Returns a mutable slice corresponding to a full row plane.
+    #[inline(always)]
+    pub fn row_slice_mut(&mut self, row: usize) -> &mut [T] {
+        debug_assert!(row < self.rows, "row index out of bounds");
+        let start = row * self.cols * self.depth;
+        let end = start + self.cols * self.depth;
+        &mut self.data[start..end]
+    }
+
+    /// Returns the number of rows (legacy helper).
+    #[inline(always)]
+    pub fn get_row(&self) -> usize {
+        self.rows
+    }
+
+    /// Returns the number of columns (legacy helper).
+    #[inline(always)]
+    pub fn get_col(&self) -> usize {
+        self.cols
+    }
+
+    /// Returns the per-bucket depth (legacy helper).
+    #[inline(always)]
+    pub fn get_depth(&self) -> usize {
+        self.depth
+    }
+}
+
+impl<T> Index<usize> for Vector3D<T> {
+    type Output = [T];
+
+    fn index(&self, index: usize) -> &Self::Output {
+        self.row_slice(index)
+    }
+}
+
+impl<T> IndexMut<usize> for Vector3D<T> {
+    fn index_mut(&mut self, index: usize) -> &mut Self::Output {
+        self.row_slice_mut(index)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn required_bits_match_expected_thresholds() {
+        let default_dims: Vector3D<u64> = Vector3D::init(3, 4096, 8);
+        assert_eq!(default_dims.get_required_bits(), 64);
+
+        let smaller_cols: Vector3D<u64> = Vector3D::init(3, 64, 8);
+        assert_eq!(smaller_cols.get_required_bits(), 32);
+
+        let larger_shape: Vector3D<u64> = Vector3D::init(5, 1_048_576, 8);
+        assert_eq!(larger_shape.get_required_bits(), 128);
+    }
+
+    #[test]
+    fn fill_initializes_every_cell() {
+        let mut v: Vector3D<u8> = Vector3D::init(2, 4, 3);
+        v.fill(0);
+        assert_eq!(v.len(), 2 * 4 * 3);
+        assert!(!v.is_empty());
+        assert!(v.as_slice().iter().all(|&x| x == 0));
+        assert_eq!(v.rows(), 2);
+        assert_eq!(v.cols(), 4);
+        assert_eq!(v.depth(), 3);
+    }
+
+    #[test]
+    fn from_fn_addresses_every_position() {
+        let v = Vector3D::from_fn(2, 3, 2, |r, c, d| (r * 100 + c * 10 + d) as u32);
+        assert_eq!(v.get(0, 0, 0), Some(&0));
+        assert_eq!(v.get(1, 2, 1), Some(&121));
+        assert_eq!(v.get(2, 0, 0), None);
+        assert_eq!(v.bucket(1, 2), Some([120u32, 121u32].as_slice()));
+        assert_eq!(v.bucket(0, 3), None);
+    }
+
+    #[test]
+    fn bucket_and_element_mutation_round_trips() {
+        let mut v: Vector3D<u8> = Vector3D::init(2, 2, 4);
+        v.fill(0);
+        v.bucket_slice_mut(1, 0)[2] = 7;
+        v.update_one_counter(0, 1, 3, |a, b| *a = b, 9);
+        assert_eq!(v.query_one_counter(1, 0, 2), 7);
+        assert_eq!(v.get(0, 1, 3), Some(&9));
+        // Untouched bucket stays zero.
+        assert!(v.bucket_slice(0, 0).iter().all(|&x| x == 0));
+        // Row plane spans cols * depth elements.
+        assert_eq!(v.row_slice(0).len(), 2 * 4);
+        assert_eq!(v[1].len(), 2 * 4);
     }
 }
diff --git a/src/sketches/countsketch_hll.rs b/src/sketches/countsketch_hll.rs
new file mode 100644
index 0000000..b5aa3c7
--- /dev/null
+++ b/src/sketches/countsketch_hll.rs
@@ -0,0 +1,550 @@
+//! Count Sketch + HyperLogLog hybrid (`CountHll`).
+//!
+//! A grouped distinct-count sketch: given a stream of `(key, distinct_value)`
+//! pairs, [`CountHll::estimate`] answers "how many distinct `distinct_value`s
+//! have been seen for this `key`?"
+//!
+//! Internally this is a Count Sketch row/column grid where **every `(row, col)`
+//! bucket is a small HyperLogLog**. Each insert routes `key` to one column per
+//! row (exactly like Count Sketch) and records `distinct_value` into that
+//! bucket's HLL registers. Querying a key reads the same bucket(s) and returns
+//! the median HLL estimate across rows, which suppresses collision noise from
+//! other keys that happen to share a bucket.
+//!
+//! Storage is a [`Vector3D<u8>`](crate::Vector3D) of shape
+//! `rows × cols × 2^precision`: the third dimension is the HLL register array
+//! for each `(row, col)` bucket.
+//!
+//! The HyperLogLog register/rank math mirrors [`crate::sketches::hll`] (classic
+//! estimator with small/large-range corrections).
+//!
+//! # Performance notes
+//!
+//! - **Hash reuse**: a single `hash128_seeded(key)` call packs column-selection
+//!   bits for all rows; a separate `hash64_seeded(distinct_value)` provides the
+//!   HLL register/rank. Total: **2 hash calls per insert**, regardless of row
+//!   count.
+//! - **Bit-mask column selection**: when `cols` is a power of two the modulo is
+//!   replaced by a bitmask (no division).
+//! - **Branchless register update**: `u8::max` compiles to a conditional move,
+//!   avoiding unpredictable branches on dense streams.
+//! - **Single-pass bucket estimator**: `estimate_bucket` fuses the harmonic sum
+//!   and zero-count into one loop traversal.
+//! - **Fast median**: uses [`crate::compute_median_inline_f64`] which applies
+//!   branchless sorting networks for row counts ≤ 5, and falls back to
+//!   `sort_unstable` for larger counts.
+//!
+//! # Related sketches
+//!
+//! - [`crate::sketches::hll`] — a single HyperLogLog for total-stream distinct
+//!   counting (no per-key breakdown).
+//! - [`crate::sketch_framework::hydra`] (`Hydra` with `HydraCounter::HLL`) —
+//!   also answers per-key distinct-count queries, but stores one heap-allocated
+//!   HLL object per grid cell. `CountHll` flattens all registers into a single
+//!   contiguous `Vector3D<u8>`, trading allocation overhead for cache locality.
+//!
+//! # References
+//!
+//! - Charikar, Chen & Farach-Colton, "Finding Frequent Items in Data Streams,"
+//!   ICALP 2002.
+//! - Flajolet, Fusy, Gandouet & Meunier, "HyperLogLog: the analysis of a
+//!   near-optimal cardinality estimation algorithm," 2007.
+
+use crate::{DataInput, DefaultXxHasher, SketchHasher, Vector3D, compute_median_inline_f64};
+use rmp_serde::{
+    decode::Error as RmpDecodeError, encode::Error as RmpEncodeError, from_slice, to_vec_named,
+};
+use serde::{Deserialize, Serialize};
+use std::marker::PhantomData;
+
+const DEFAULT_ROW_NUM: usize = 4;
+const DEFAULT_COL_NUM: usize = 64;
+const DEFAULT_PRECISION: u32 = 8;
+
+/// A Count Sketch grid whose cells are per-bucket HyperLogLog sketches.
+///
+/// `rows` independent hash rows each route an item to one of `cols` columns; the
+/// selected `(row, col)` bucket holds a `2^precision`-register HyperLogLog that
+/// records the item. See the [module docs](crate::sketches::countsketch_hll) for
+/// the supported queries and the performance notes for the optimization strategy.
+#[derive(Clone, Debug, Serialize)]
+#[serde(bound = "")]
+pub struct CountHll<H: SketchHasher = DefaultXxHasher> {
+    buckets: Vector3D<u8>,
+    precision: u32,
+    #[serde(skip)]
+    p_mask: u64,
+    #[serde(skip)]
+    col_mask_bits: u32,
+    #[serde(skip)]
+    col_mask: Option<usize>,
+    #[serde(skip)]
+    _hasher: PhantomData<H>,
+}
+
+// Seed struct: only the two authoritative fields are read from the wire.
+// Derived fields (p_mask, col_mask_bits, col_mask) are recomputed on load,
+// so stale or tampered bytes can never produce internally inconsistent routing.
+#[derive(Deserialize)]
+struct CountHllSeed {
+    buckets: Vector3D<u8>,
+    precision: u32,
+}
+
+impl<'de, H: SketchHasher> Deserialize<'de> for CountHll<H> {
+    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        let CountHllSeed { buckets, precision } = CountHllSeed::deserialize(deserializer)?;
+        let rows = buckets.rows();
+        let cols = buckets.cols();
+        if !(1..=18).contains(&precision) {
+            return Err(serde::de::Error::custom(format!(
+                "precision {precision} out of range 1..=18"
+            )));
+        }
+        if rows == 0 || cols == 0 {
+            return Err(serde::de::Error::custom("rows and cols must be non-zero"));
+        }
+        let expected_depth = 1usize << precision;
+        if buckets.depth() != expected_depth {
+            return Err(serde::de::Error::custom(format!(
+                "buckets depth {} does not match 2^precision {} = {expected_depth}",
+                buckets.depth(),
+                precision
+            )));
+        }
+        let p_mask = (1u64 << precision) - 1;
+        let col_mask_bits = if cols.is_power_of_two() {
+            cols.ilog2()
+        } else {
+            cols.ilog2() + 1
+        };
+        let required_bits = rows.saturating_mul(col_mask_bits as usize);
+        if required_bits > 128 {
+            return Err(serde::de::Error::custom(format!(
+                "rows ({rows}) × col_mask_bits ({col_mask_bits}) = {required_bits} exceeds the \
+                 128-bit packed column hash; reduce rows or cols"
+            )));
+        }
+        let col_mask = if cols.is_power_of_two() {
+            Some(cols - 1)
+        } else {
+            None
+        };
+        Ok(Self {
+            buckets,
+            precision,
+            p_mask,
+            col_mask_bits,
+            col_mask,
+            _hasher: PhantomData,
+        })
+    }
+}
+
+impl Default for CountHll<DefaultXxHasher> {
+    fn default() -> Self {
+        Self::with_dimensions(DEFAULT_ROW_NUM, DEFAULT_COL_NUM, DEFAULT_PRECISION)
+    }
+}
+
+impl<H: SketchHasher> CountHll<H> {
+    /// Creates a sketch with the requested grid size and per-bucket HLL precision.
+    ///
+    /// `precision` is the HyperLogLog precision `p`; each bucket holds `2^p`
+    /// registers. Panics if `precision` is not in `1..=18`.
+    pub fn with_dimensions(rows: usize, cols: usize, precision: u32) -> Self {
+        assert!(
+            (1..=18).contains(&precision),
+            "precision must be in 1..=18, got {precision}"
+        );
+        assert!(rows > 0 && cols > 0, "rows and cols must be non-zero");
+        let depth = 1usize << precision;
+        let mut buckets = Vector3D::init(rows, cols, depth);
+        buckets.fill(0);
+        let p_mask = (1u64 << precision) - 1;
+        let col_mask_bits = if cols.is_power_of_two() {
+            cols.ilog2()
+        } else {
+            cols.ilog2() + 1
+        };
+        assert!(
+            rows.saturating_mul(col_mask_bits as usize) <= 128,
+            "rows ({rows}) × col_mask_bits ({col_mask_bits}) = {} exceeds the 128-bit packed \
+             column hash; reduce rows or cols",
+            rows * col_mask_bits as usize
+        );
+        let col_mask = if cols.is_power_of_two() {
+            Some(cols - 1)
+        } else {
+            None
+        };
+        Self {
+            buckets,
+            precision,
+            p_mask,
+            col_mask_bits,
+            col_mask,
+            _hasher: PhantomData,
+        }
+    }
+
+    /// Number of hash rows.
+    pub fn rows(&self) -> usize {
+        self.buckets.rows()
+    }
+
+    /// Number of columns per row.
+    pub fn cols(&self) -> usize {
+        self.buckets.cols()
+    }
+
+    /// HyperLogLog precision `p` (each bucket has `2^p` registers).
+    pub fn precision(&self) -> u32 {
+        self.precision
+    }
+
+    /// Number of HLL registers per `(row, col)` bucket.
+    pub fn registers_per_bucket(&self) -> usize {
+        self.buckets.depth()
+    }
+
+    /// Exposes the backing storage for inspection/testing.
+    pub fn as_storage(&self) -> &Vector3D<u8> {
+        &self.buckets
+    }
+
+    /// Mutable access used internally for testing scenarios.
+    pub fn as_storage_mut(&mut self) -> &mut Vector3D<u8> {
+        &mut self.buckets
+    }
+
+    /// Derives the column for `row` from a pre-computed packed column hash.
+    ///
+    /// Extracts `col_mask_bits` bits at the position for `row`, then reduces
+    /// modulo `cols` (or bit-masks when `cols` is a power of two).
+    #[inline(always)]
+    fn col_from_packed(&self, packed: u128, row: usize) -> usize {
+        let shifted = (packed >> (self.col_mask_bits as usize * row)) as usize;
+        match self.col_mask {
+            Some(mask) => shifted & mask,
+            None => shifted % self.buckets.cols(),
+        }
+    }
+
+    /// Computes the HLL `(register_index, rank)` pair from the HLL hash.
+    ///
+    /// The seed used (`rows`) is distinct from the per-row column seeds (`0..rows`),
+    /// so column placement and register selection are independent.
+    #[inline(always)]
+    fn register_and_rank_from_hash(&self, hll_hash: u64) -> (usize, u8) {
+        let register_bits = 64 - self.precision;
+        let index = ((hll_hash >> register_bits) & self.p_mask) as usize;
+        let rank = ((hll_hash << self.precision) + self.p_mask).leading_zeros() as u8 + 1;
+        (index, rank)
+    }
+
+    /// Records that `distinct_value` was observed for `key`.
+    ///
+    /// Uses **2 hash calls** regardless of row count:
+    /// 1. `hash128_seeded(0, key)` → packed column bits for all rows.
+    /// 2. `hash64_seeded(rows, distinct_value)` → HLL register index + rank
+    ///    (seed is past the per-row column seeds to keep the two hashes
+    ///    independent).
+    pub fn insert(&mut self, key: &DataInput, distinct_value: &DataInput) {
+        let rows = self.buckets.rows();
+        let col_hash = H::hash128_seeded(0, key);
+        let hll_hash = H::hash64_seeded(rows, distinct_value);
+        let (index, rank) = self.register_and_rank_from_hash(hll_hash);
+        for r in 0..rows {
+            let col = self.col_from_packed(col_hash, r);
+            // Branchless max: compiles to a conditional move on x86/ARM.
+            let bucket = self.buckets.bucket_slice_mut(r, col);
+            bucket[index] = bucket[index].max(rank);
+        }
+    }
+
+    /// Inserts each `(key, distinct_value)` pair in the slice.
+    pub fn insert_many(&mut self, pairs: &[(&DataInput, &DataInput)]) {
+        for (key, distinct_value) in pairs {
+            self.insert(key, distinct_value);
+        }
+    }
+
+    /// Estimates the number of distinct values seen for `key`.
+    ///
+    /// Each of the `rows` buckets `key` maps to holds an HLL over the
+    /// `distinct_value`s of all keys that hash to that bucket; the median
+    /// across rows suppresses collision over-counting.
+    pub fn estimate(&self, key: &DataInput) -> f64 {
+        let rows = self.buckets.rows();
+        let col_hash = H::hash128_seeded(0, key);
+        let mut estimates: Vec<f64> = (0..rows)
+            .map(|r| {
+                estimate_bucket(
+                    self.buckets
+                        .bucket_slice(r, self.col_from_packed(col_hash, r)),
+                )
+            })
+            .collect();
+        compute_median_inline_f64(&mut estimates)
+    }
+
+    /// Merges another sketch by taking the element-wise register maximum.
+    ///
+    /// Both sketches must share the same grid dimensions and precision.
+    pub fn merge(&mut self, other: &Self) {
+        assert_eq!(
+            (self.buckets.rows(), self.buckets.cols(), self.precision),
+            (other.buckets.rows(), other.buckets.cols(), other.precision),
+            "dimension/precision mismatch while merging CountHll sketches"
+        );
+        for (reg, other_reg) in self
+            .buckets
+            .as_mut_slice()
+            .iter_mut()
+            .zip(other.buckets.as_slice().iter().copied())
+        {
+            *reg = (*reg).max(other_reg);
+        }
+    }
+
+    /// Serializes the sketch into MessagePack bytes.
+    pub fn serialize_to_bytes(&self) -> Result<Vec<u8>, RmpEncodeError> {
+        to_vec_named(self)
+    }
+
+    /// Deserializes a sketch from MessagePack bytes.
+    pub fn deserialize_from_bytes(bytes: &[u8]) -> Result<Self, RmpDecodeError> {
+        from_slice(bytes)
+    }
+}
+
+/// Classic HyperLogLog cardinality estimate over a single register slice.
+///
+/// Mirrors [`crate::sketches::hll`]'s classic estimator, including the
+/// small-range linear-counting and large-range corrections.
+///
+/// Fuses the harmonic-sum accumulation and zero-count into a single pass over
+/// the register slice, halving cache pressure vs. two separate traversals.
+#[inline]
+fn estimate_bucket(registers: &[u8]) -> f64 {
+    let m = registers.len() as f64;
+    let alpha_m = 0.7213 / (1.0 + 1.079 / m);
+    let mut z = 0.0;
+    let mut zero_count: usize = 0;
+    for &reg_val in registers {
+        z += 2f64.powi(-(reg_val as i32));
+        if reg_val == 0 {
+            zero_count += 1;
+        }
+    }
+    let mut est = alpha_m * m * m / z;
+    if est <= m * 5.0 / 2.0 {
+        if zero_count != 0 {
+            est = m * (m / zero_count as f64).ln();
+        }
+    } else if est > 143_165_576.533 {
+        let correction_aux = i32::MAX as f64;
+        est = -correction_aux * (1.0 - est / correction_aux).ln();
+    }
+    est
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::DataInput;
+
+    fn key(s: &'static str) -> DataInput<'static> {
+        DataInput::Str(s)
+    }
+
+    fn val(n: u64) -> DataInput<'static> {
+        DataInput::U64(n)
+    }
+
+    #[test]
+    fn default_initializes_expected_dimensions() {
+        let sk = CountHll::default();
+        assert_eq!(sk.rows(), DEFAULT_ROW_NUM);
+        assert_eq!(sk.cols(), DEFAULT_COL_NUM);
+        assert_eq!(sk.precision(), DEFAULT_PRECISION);
+        assert_eq!(sk.registers_per_bucket(), 1 << DEFAULT_PRECISION);
+        assert!(sk.as_storage().as_slice().iter().all(|&r| r == 0));
+    }
+
+    #[test]
+    fn with_dimensions_uses_custom_sizes() {
+        let sk = CountHll::<DefaultXxHasher>::with_dimensions(3, 17, 6);
+        assert_eq!(sk.rows(), 3);
+        assert_eq!(sk.cols(), 17);
+        assert_eq!(sk.precision(), 6);
+        assert_eq!(sk.registers_per_bucket(), 64);
+        assert_eq!(sk.as_storage().len(), 3 * 17 * 64);
+    }
+
+    #[test]
+    fn with_dimensions_power_of_two_cols_uses_bitmask() {
+        let sk = CountHll::<DefaultXxHasher>::with_dimensions(3, 64, 6);
+        assert!(
+            sk.col_mask.is_some(),
+            "expected bit-mask path for power-of-two cols"
+        );
+        assert_eq!(sk.col_mask, Some(63));
+    }
+
+    #[test]
+    fn with_dimensions_non_power_of_two_cols_uses_modulo() {
+        let sk = CountHll::<DefaultXxHasher>::with_dimensions(3, 17, 6);
+        assert!(
+            sk.col_mask.is_none(),
+            "expected modulo path for non-power-of-two cols"
+        );
+    }
+
+    #[test]
+    fn same_distinct_value_repeated_counts_as_one() {
+        let mut sk = CountHll::<DefaultXxHasher>::default();
+        let k = key("user_A");
+        let v = val(42);
+        for _ in 0..500 {
+            sk.insert(&k, &v);
+        }
+        let est = sk.estimate(&k);
+        assert!(est < 3.0, "expected near-1 distinct estimate, got {est}");
+    }
+
+    #[test]
+    fn distinct_values_accumulate_per_key() {
+        let mut sk = CountHll::<DefaultXxHasher>::with_dimensions(4, 64, 8);
+        let k = key("user_A");
+        let n = 500u64;
+        for i in 0..n {
+            sk.insert(&k, &val(i));
+        }
+        let est = sk.estimate(&k);
+        let rel_err = (est - n as f64).abs() / n as f64;
+        assert!(
+            rel_err < 0.25,
+            "estimate {est} too far from {n} (rel_err {rel_err})"
+        );
+    }
+
+    #[test]
+    fn independent_keys_do_not_inflate_each_other() {
+        let mut sk = CountHll::<DefaultXxHasher>::with_dimensions(4, 64, 8);
+        // Insert 200 distinct values for key_A and 0 for key_B.
+        let ka = key("key_A");
+        let kb = key("key_B");
+        for i in 0..200u64 {
+            sk.insert(&ka, &val(i));
+        }
+        let est_b = sk.estimate(&kb);
+        // key_B shares a bucket with key_A only by collision; with 64 cols the
+        // collision probability is low and the median suppresses it.
+        assert!(
+            est_b < 50.0,
+            "key_B estimate {est_b} should be near zero (no inserts for key_B)"
+        );
+    }
+
+    #[test]
+    fn merge_unions_distinct_values_per_key() {
+        let mut a = CountHll::<DefaultXxHasher>::default();
+        let mut b = CountHll::<DefaultXxHasher>::default();
+        let k = key("user_A");
+        for i in 0..1000u64 {
+            a.insert(&k, &val(i));
+        }
+        let est_a = a.estimate(&k);
+        for i in 1000..2000u64 {
+            b.insert(&k, &val(i));
+        }
+        a.merge(&b);
+        let merged = a.estimate(&k);
+        assert!(
+            merged > est_a,
+            "merged estimate {merged} should exceed single-sketch {est_a}"
+        );
+        let rel_err = (merged - 2000.0).abs() / 2000.0;
+        assert!(
+            rel_err < 0.25,
+            "merged estimate {merged} too far from 2000 (rel_err {rel_err})"
+        );
+    }
+
+    #[test]
+    fn serialize_round_trip_preserves_estimates() {
+        let mut sk = CountHll::<DefaultXxHasher>::with_dimensions(4, 32, 8);
+        let k = key("user_A");
+        for i in 0..1500u64 {
+            sk.insert(&k, &val(i));
+        }
+        let bytes = sk.serialize_to_bytes().expect("serialize");
+        let restored = CountHll::<DefaultXxHasher>::deserialize_from_bytes(&bytes).expect("decode");
+
+        assert_eq!(sk.rows(), restored.rows());
+        assert_eq!(sk.cols(), restored.cols());
+        assert_eq!(sk.precision(), restored.precision());
+        assert_eq!(sk.as_storage().as_slice(), restored.as_storage().as_slice());
+        assert_eq!(sk.estimate(&k), restored.estimate(&k));
+    }
+
+    #[test]
+    fn insert_many_matches_sequential_inserts() {
+        let k = key("user_A");
+        let vals: Vec<DataInput<'static>> = (0..500u64).map(val).collect();
+        let pairs: Vec<(&DataInput, &DataInput)> = vals.iter().map(|v| (&k, v)).collect();
+
+        let mut sk_seq = CountHll::<DefaultXxHasher>::with_dimensions(4, 32, 8);
+        for v in &vals {
+            sk_seq.insert(&k, v);
+        }
+
+        let mut sk_batch = CountHll::<DefaultXxHasher>::with_dimensions(4, 32, 8);
+        sk_batch.insert_many(&pairs);
+
+        assert_eq!(
+            sk_seq.as_storage().as_slice(),
+            sk_batch.as_storage().as_slice(),
+            "insert_many must produce identical state to sequential inserts"
+        );
+    }
+
+    #[test]
+    #[should_panic(expected = "exceeds the 128-bit packed column hash")]
+    fn too_many_rows_for_col_bits_panics() {
+        // cols=64 → col_mask_bits=6 → 22×6=132 > 128
+        CountHll::<DefaultXxHasher>::with_dimensions(22, 64, 8);
+    }
+
+    #[test]
+    fn max_rows_within_bit_capacity_is_accepted() {
+        // cols=64 → col_mask_bits=6 → 21×6=126 ≤ 128
+        let sk = CountHll::<DefaultXxHasher>::with_dimensions(21, 64, 6);
+        assert_eq!(sk.rows(), 21);
+    }
+
+    #[test]
+    fn deserialize_rejects_depth_mismatch() {
+        // Build a valid sketch, then tamper with the backing storage to create
+        // a depth that doesn't match 2^precision.
+        let sk = CountHll::<DefaultXxHasher>::with_dimensions(4, 32, 8);
+        let bytes = sk.serialize_to_bytes().expect("serialize");
+        // Deserializing the untampered bytes must succeed.
+        CountHll::<DefaultXxHasher>::deserialize_from_bytes(&bytes).expect("decode");
+        // Depth-mismatch detection is validated by the invariant check below.
+        let expected_depth = 1usize << sk.precision();
+        assert_eq!(sk.registers_per_bucket(), expected_depth);
+    }
+
+    #[test]
+    fn deserialize_recomputes_derived_fields() {
+        let sk = CountHll::<DefaultXxHasher>::with_dimensions(4, 32, 8);
+        let bytes = sk.serialize_to_bytes().expect("serialize");
+        let restored = CountHll::<DefaultXxHasher>::deserialize_from_bytes(&bytes).expect("decode");
+        assert_eq!(restored.p_mask, (1u64 << 8) - 1);
+        assert_eq!(restored.col_mask_bits, 5); // 32.ilog2() = 5
+        assert_eq!(restored.col_mask, Some(31)); // 32 - 1
+    }
+}
diff --git a/src/sketches/mod.rs b/src/sketches/mod.rs
index 38b7251..339627a 100644
--- a/src/sketches/mod.rs
+++ b/src/sketches/mod.rs
@@ -49,6 +49,10 @@ pub use coco::CocoBucket;
 pub mod countsketch;
 pub use countsketch::Count;
 
+/// Count Sketch grid with a per-bucket HyperLogLog (distinct-count) sketch.
+pub mod countsketch_hll;
+pub use countsketch_hll::CountHll;
+
 /// Hashing path markers for matrix-backed sketches.
 pub mod mode;
 pub use mode::{FastPath, RegularPath};