linkrs_storage/
stats.rs

1//! Storage statistics helpers: HyperLogLog cardinality estimator.
2//!
3//! Precision target is P=6 (M=64 registers): a small 64-byte footprint per
4//! column with an expected relative error around 13%. Callers needing
5//! tighter bounds should document the measured error next to the estimate.
6//!
7//! Hashing is deterministic FNV-1a so persisted registers stay comparable
8//! across processes and restarts.
9
10use std::io::{Read, Write};
11
12use linkrs_core::{StorageResult, Value};
13
14pub const HLL_P: u32 = 6;
15pub const HLL_M: usize = 64;
16
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub struct HyperLogLog {
19    registers: [u8; HLL_M],
20}
21
22impl Default for HyperLogLog {
23    fn default() -> Self {
24        Self::new()
25    }
26}
27
28impl HyperLogLog {
29    pub fn new() -> Self {
30        Self {
31            registers: [0; HLL_M],
32        }
33    }
34
35    pub fn add_value(&mut self, value: &Value) {
36        match value {
37            Value::Double(f) if f.is_nan() => (),
38            Value::Float(f) if f.is_nan() => (),
39            Value::Null(_) | Value::Empty => (),
40            Value::Double(f) => self.add_hash(f.to_bits()),
41            Value::Float(f) => self.add_hash(f.to_bits() as u64),
42            _ => {
43                self.add_hash(fnv1a_value(value));
44            }
45        }
46    }
47
48    pub fn add_hash(&mut self, hash: u64) {
49        let idx = (hash & (HLL_M as u64 - 1)) as usize;
50        let w = hash >> HLL_P;
51        let rank = (w.leading_zeros() - HLL_P + 1).min(64 - HLL_P) as u8;
52        if rank > self.registers[idx] {
53            self.registers[idx] = rank;
54        }
55    }
56
57    pub fn estimate(&self) -> u64 {
58        let m = HLL_M as f64;
59        let alpha = 0.673;
60        let mut sum = 0.0;
61        let mut zeros = 0u32;
62        for r in &self.registers {
63            sum += 2.0f64.powi(-(*r as i32));
64            if *r == 0 {
65                zeros += 1;
66            }
67        }
68        let raw = alpha * m * m / sum;
69        let estimate = if raw <= 2.5 * m && zeros > 0 {
70            m * (m / zeros as f64).ln()
71        } else {
72            raw
73        };
74        estimate.round() as u64
75    }
76
77    pub fn merge(&mut self, other: &Self) {
78        for (a, b) in self.registers.iter_mut().zip(other.registers.iter()) {
79            if *b > *a {
80                *a = *b;
81            }
82        }
83    }
84
85    pub fn registers(&self) -> &[u8; HLL_M] {
86        &self.registers
87    }
88
89    pub fn serialize(&self, writer: &mut impl Write) -> StorageResult<usize> {
90        writer.write_all(&self.registers)?;
91        Ok(HLL_M)
92    }
93
94    pub fn deserialize(reader: &mut impl Read) -> StorageResult<Self> {
95        let mut registers = [0u8; HLL_M];
96        reader.read_exact(&mut registers)?;
97        Ok(Self { registers })
98    }
99}
100
101/// Deterministic 64-bit FNV-1a over a stable per-type encoding.
102///
103/// Replaces the process-seeded default hasher so HyperLogLog registers can be
104/// persisted and merged across restarts without changing the estimate.
105fn fnv1a_value(value: &Value) -> u64 {
106    const OFFSET: u64 = 0xcbf29ce484222325;
107    const PRIME: u64 = 0x100000001b3;
108    let mut hash = OFFSET;
109    let mut mix = |bytes: &[u8]| {
110        for b in bytes {
111            hash ^= *b as u64;
112            hash = hash.wrapping_mul(PRIME);
113        }
114    };
115    // Discriminant first so different types never collide on equal payloads.
116    mix(&[value_discriminant(value)]);
117    match value {
118        Value::Bool(b) => mix(&[*b as u8]),
119        Value::SmallInt(v) => mix(&v.to_le_bytes()),
120        Value::Int(v) => mix(&v.to_le_bytes()),
121        Value::BigInt(v) => mix(&v.to_le_bytes()),
122        Value::String(s) => mix(s.as_bytes()),
123        Value::FixedString(s) => mix(s.as_bytes()),
124        Value::Blob(b) => mix(b),
125        Value::Date(d) => {
126            mix(&d.year.to_le_bytes());
127            mix(&d.month.to_le_bytes());
128            mix(&d.day.to_le_bytes());
129        }
130        Value::Time(t) => {
131            mix(&t.hour.to_le_bytes());
132            mix(&t.minute.to_le_bytes());
133            mix(&t.sec.to_le_bytes());
134            mix(&t.microsec.to_le_bytes());
135        }
136        Value::DateTime(dt) => {
137            mix(&dt.year.to_le_bytes());
138            mix(&dt.month.to_le_bytes());
139            mix(&dt.day.to_le_bytes());
140            mix(&dt.hour.to_le_bytes());
141            mix(&dt.minute.to_le_bytes());
142            mix(&dt.sec.to_le_bytes());
143            mix(&dt.microsec.to_le_bytes());
144        }
145        Value::Uuid(u) => mix(u.as_bytes()),
146        Value::Json(j) => mix(j.as_str().as_bytes()),
147        Value::JsonB(j) => mix(j.to_json_string().as_bytes()),
148        _ => {
149            if let Ok(bytes) = postcard::to_allocvec(value) {
150                mix(&bytes);
151            } else {
152                mix(&value.estimated_size().to_le_bytes());
153            }
154        }
155    }
156    // FNV-1a leaves sequential inputs correlated in the low bits that back
157    // the register index, so run a splitmix64 finalizer for avalanche while
158    // staying dependency-free and deterministic.
159    mix64(hash)
160}
161
162/// SplitMix64 finalizer: strong bit avalanche for a 64-bit hash.
163fn mix64(mut x: u64) -> u64 {
164    x ^= x >> 30;
165    x = x.wrapping_mul(0xbf58476d1ce4e5b9);
166    x ^= x >> 27;
167    x = x.wrapping_mul(0x94d049bb133111eb);
168    x ^= x >> 31;
169    x
170}
171
172/// Stable per-type discriminant for the FNV-1a encoding.
173fn value_discriminant(value: &Value) -> u8 {
174    use linkrs_core::value::Value::*;
175    match value {
176        Null(_) => 0,
177        Empty => 1,
178        Bool(_) => 2,
179        SmallInt(_) => 3,
180        Int(_) => 4,
181        BigInt(_) => 5,
182        Float(_) => 6,
183        Double(_) => 7,
184        String(_) => 8,
185        FixedString(_) => 9,
186        Blob(_) => 10,
187        Date(_) => 11,
188        Time(_) => 12,
189        DateTime(_) => 13,
190        Uuid(_) => 14,
191        Json(_) => 15,
192        JsonB(_) => 16,
193        _ => 255,
194    }
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200
201    #[test]
202    fn empty_estimate_is_zero() {
203        assert_eq!(HyperLogLog::new().estimate(), 0);
204    }
205
206    #[test]
207    fn single_value_estimate_near_one() {
208        let mut hll = HyperLogLog::new();
209        hll.add_value(&Value::Int(42));
210        let est = hll.estimate();
211        assert!((1..=3).contains(&est), "estimate={}", est);
212    }
213
214    #[test]
215    fn merge_union_matches_full() {
216        let mut a = HyperLogLog::new();
217        let mut b = HyperLogLog::new();
218        let mut full = HyperLogLog::new();
219        for i in 0..1000i32 {
220            let v = Value::Int(i);
221            full.add_value(&v);
222            if i % 2 == 0 {
223                a.add_value(&v);
224            } else {
225                b.add_value(&v);
226            }
227        }
228        a.merge(&b);
229        let merged = a.estimate() as f64;
230        let expected = full.estimate() as f64;
231        let err = (merged - expected).abs() / expected.max(1.0);
232        assert!(err < 0.15, "merged={} expected={}", merged, expected);
233    }
234
235    #[test]
236    fn roundtrip() {
237        let mut hll = HyperLogLog::new();
238        hll.add_value(&Value::string("hello"));
239        let mut buf = Vec::new();
240        hll.serialize(&mut buf).unwrap();
241        assert_eq!(buf.len(), 64);
242        let back = HyperLogLog::deserialize(&mut &buf[..]).unwrap();
243        assert_eq!(back, hll);
244    }
245
246    #[test]
247    fn nan_skipped() {
248        let mut hll = HyperLogLog::new();
249        hll.add_value(&Value::Double(f64::NAN));
250        assert_eq!(hll.estimate(), 0);
251    }
252
253    #[test]
254    fn deterministic_registers_across_instances() {
255        let values: Vec<Value> = (0..200i32).map(Value::Int).collect();
256        let mut a = HyperLogLog::new();
257        let mut b = HyperLogLog::new();
258        for v in &values {
259            a.add_value(v);
260        }
261        for v in values.iter().rev() {
262            b.add_value(v);
263        }
264        assert_eq!(a.registers(), b.registers());
265        assert_eq!(a.estimate(), b.estimate());
266    }
267}
linkrs_storage/stats.rs

linkrs_storage/
stats.rs
