nx_sketch_kmv.nx source
↩ module page · 316 lines · 10241 B
1// sketch_kmv.nx -- KMV (K-Minimum-Values) sketch.
2//
3// Bar-Yossef et al. 2002 + Beyer 2007. Maintain the K smallest
4// 32-bit hash values seen. Direct ancestor of DataSketches'
5// Theta sketches; we ship the foundation primitive first then
6// build Theta on top.
7//
8// CAPABILITIES (set-operation cardinality):
9// - cardinality: (K-1) * 2^32 / kth_smallest when |set| > K
10// total_inserted_unique when |set| <= K
11// - union: K smallest across A.values ∪ B.values
12// - Jaccard: |A.kmins ∩ B.kmins| / |A.kmins ∪ B.kmins|
13// unbiased estimator of true Jaccard
14//
15// Error: rel_stddev ~ 1/sqrt(K-1) at 1-sigma.
16// For K=4096: 1.56% stddev. For K=16384: 0.78%.
17//
18// COMPLEMENTS HLL:
19// - HLL: better space per accuracy point for pure cardinality.
20// - KMV: SAME sketch supports union AND intersection (Jaccard).
21// The trade-off is well-studied; both ship in DataSketches
22// (HLL.java and Theta.java) -- we ship the same axis pair.
23//
24// Hash domain: 32-bit (murmur3_32). We store hashes as i64 to
25// keep arithmetic simple; the value space is [0, 2^32). Hash 0
26// is mapped to 1 (matches Cuckoo's empty-sentinel pattern; KMV
27// uses the special "all-slots-filled-with-max" sentinel via
28// the n_items counter instead, but mapping 0->1 avoids edge
29// cases in the cardinality formula).
30
31// nx_safety_envelope:
32// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
33// sil_target: SIL1
34// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
35// verdict: NOT_YET_EVALUATED
36
37import "nx_syscalls.nx"
38import "nx_murmur3.nx"
39import "nx_sketch_types.nx"
40
41const NX_KMV_K_MIN: i64 = 16
42const NX_KMV_K_MAX: i64 = 65536
43
44// 2^32 in i64 -- the hash domain size used in cardinality estimation.
45const NX_KMV_HASH_MAX: i64 = 4294967296
46
47struct Kmv {
48 values: *i64, // sorted ascending; max at index n_items-1
49 k: i64,
50 n_items: i64, // = min(unique_inserts, k)
51 seed: i64,
52}
53
54// === construction =================================================
55
56func nx_kmv_alloc(k: i64, seed: i64) -> *Kmv {
57 if k < NX_KMV_K_MIN { return 0 as *Kmv }
58 if k > NX_KMV_K_MAX { return 0 as *Kmv }
59 let raw: *u8 = sys_mmap(40)
60 let kmv: *Kmv = raw as *Kmv
61 kmv.k = k
62 kmv.n_items = 0
63 let values_raw: *u8 = sys_mmap(k * 8)
64 kmv.values = values_raw as *i64
65 kmv.seed = seed
66 return kmv
67}
68
69// === hash helper =================================================
70
71func nx_kmv_hash(kmv: *Kmv, key: *u8, len: i64) -> i64 {
72 let h: i64 = murmur3_32(kmv.seed, key, len) & 0xFFFFFFFF
73 if h == 0 { return 1 } // remap 0 -> 1
74 return h
75}
76
77// === insertion =================================================
78//
79// Binary-search for position; if hash already present, no-op
80// (KMV is a SET of hashes). Else if set not full, insert and
81// shift tail right. Else compare with current max (values[k-1]);
82// if hash < max, replace max + sift down to maintain sort.
83
84func nx_kmv_add(kmv: *Kmv, key: *u8, len: i64) -> i64 {
85 let h: i64 = nx_kmv_hash(kmv, key, len)
86
87 // Binary search for h in values[0..n_items).
88 var lo: i64 = 0
89 var hi: i64 = kmv.n_items
90 while lo < hi {
91 let mid: i64 = (lo + hi) / 2
92 let v: i64 = kmv.values[mid]
93 if v == h { return 0 } // already in set
94 if v < h { lo = mid + 1 }
95 if v > h { hi = mid }
96 }
97 // `lo` is the insertion index.
98
99 if kmv.n_items < kmv.k {
100 // Shift values[lo..n_items) right by one, insert.
101 var j: i64 = kmv.n_items
102 while j > lo {
103 kmv.values[j] = kmv.values[j - 1]
104 j = j - 1
105 }
106 kmv.values[lo] = h
107 kmv.n_items = kmv.n_items + 1
108 return 0
109 }
110 // Full set: only insert if h < current max (values[k-1]).
111 let cur_max: i64 = kmv.values[kmv.k - 1]
112 if h >= cur_max { return 0 }
113 // Shift values[lo..k-1) right by one (overwriting old max), insert h.
114 var j2: i64 = kmv.k - 1
115 while j2 > lo {
116 kmv.values[j2] = kmv.values[j2 - 1]
117 j2 = j2 - 1
118 }
119 kmv.values[lo] = h
120 return 0
121}
122
123// Insert pre-hashed i64 value directly (for testing + union).
124func nx_kmv_add_hash(kmv: *Kmv, raw_h: i64) -> i64 {
125 var h: i64 = raw_h & 0xFFFFFFFF
126 if h == 0 { h = 1 }
127 var lo: i64 = 0
128 var hi: i64 = kmv.n_items
129 while lo < hi {
130 let mid: i64 = (lo + hi) / 2
131 let v: i64 = kmv.values[mid]
132 if v == h { return 0 }
133 if v < h { lo = mid + 1 }
134 if v > h { hi = mid }
135 }
136 if kmv.n_items < kmv.k {
137 var j: i64 = kmv.n_items
138 while j > lo {
139 kmv.values[j] = kmv.values[j - 1]
140 j = j - 1
141 }
142 kmv.values[lo] = h
143 kmv.n_items = kmv.n_items + 1
144 return 0
145 }
146 let cur_max: i64 = kmv.values[kmv.k - 1]
147 if h >= cur_max { return 0 }
148 var j2: i64 = kmv.k - 1
149 while j2 > lo {
150 kmv.values[j2] = kmv.values[j2 - 1]
151 j2 = j2 - 1
152 }
153 kmv.values[lo] = h
154 return 0
155}
156
157// === cardinality estimation ====================================
158//
159// Bar-Yossef 2002 estimator:
160// if n_items < k: cardinality = n_items (exact)
161// else: cardinality = (k - 1) * HASH_MAX / kth_min
162// kth_min is values[k-1] (the largest of the k smallest).
163//
164// For HASH_MAX = 2^32 = 4294967296, the multiplication
165// (k-1) * 2^32 fits in i64 well past k=2^30.
166
167func nx_kmv_estimate(kmv: *Kmv) -> i64 {
168 if kmv.n_items < kmv.k { return kmv.n_items }
169 let kth: i64 = kmv.values[kmv.k - 1]
170 if kth == 0 { return 0 }
171 return ((kmv.k - 1) * NX_KMV_HASH_MAX) / kth
172}
173
174// === typed query ===============================================
175//
176// rel_stddev = 1 / sqrt(k - 1). Tabulate by k class.
177
178func nx_kmv_stddev_rel_ppb(k: i64) -> i64 {
179 if k <= 16 { return 258000000 } // 1/sqrt(15) = 0.258
180 if k <= 64 { return 126000000 } // 1/sqrt(63) = 0.126
181 if k <= 256 { return 62700000 } // 1/sqrt(255)
182 if k <= 1024 { return 31300000 } // 1/sqrt(1023)
183 if k <= 4096 { return 15600000 } // 1/sqrt(4095) = 0.0156
184 if k <= 16384 { return 7810000 } // 1/sqrt(16383)
185 return 3900000 // 1/sqrt(65535)
186}
187
188func nx_kmv_query(kmv: *Kmv) -> *ApproxI64 {
189 let est: i64 = nx_kmv_estimate(kmv)
190 return nx_approx_new(est, NX_ENV_REL_STDDEV,
191 nx_kmv_stddev_rel_ppb(kmv.k),
192 682700000, // 0.6827 = 1-sigma
193 NX_MATURITY_REFERENCE_IMPL,
194 NX_ADV_HONEST)
195}
196
197// === union =====================================================
198//
199// Pre: a.k == b.k and a.seed == b.seed.
200// Result: new KMV holding the K smallest across A ∪ B.
201//
202// Linear merge of two sorted streams, K-bounded.
203
204func nx_kmv_union(a: *Kmv, b: *Kmv) -> *Kmv {
205 if a.k != b.k { return 0 as *Kmv }
206 if a.seed != b.seed { return 0 as *Kmv }
207 let out: *Kmv = nx_kmv_alloc(a.k, a.seed)
208 var i: i64 = 0
209 var j: i64 = 0
210 var w: i64 = 0
211 while w < out.k {
212 let a_done: i64 = i >= a.n_items
213 let b_done: i64 = j >= b.n_items
214 if a_done == 1 {
215 if b_done == 1 { w = out.k; return out }
216 out.values[w] = b.values[j]
217 j = j + 1
218 w = w + 1
219 }
220 if a_done == 0 {
221 if b_done == 1 {
222 out.values[w] = a.values[i]
223 i = i + 1
224 w = w + 1
225 }
226 if b_done == 0 {
227 let va: i64 = a.values[i]
228 let vb: i64 = b.values[j]
229 if va < vb {
230 out.values[w] = va
231 i = i + 1
232 w = w + 1
233 }
234 if va == vb {
235 out.values[w] = va
236 i = i + 1
237 j = j + 1
238 w = w + 1
239 }
240 if va > vb {
241 out.values[w] = vb
242 j = j + 1
243 w = w + 1
244 }
245 }
246 }
247 // Early-exit when both sides exhausted.
248 if i >= a.n_items {
249 if j >= b.n_items { w = out.k }
250 }
251 }
252 // Set n_items based on whether we filled the cap or ran out.
253 // If both sources exhausted before w == k, n_items == w.
254 // Otherwise we wrote exactly k values.
255 out.n_items = w
256 return out
257}
258
259// === Jaccard similarity ========================================
260//
261// J(A, B) = |A ∩ B| / |A ∪ B|
262// ≈ |kmins(A) ∩ kmins(B)| / |kmins(A) ∪ kmins(B)|
263// where intersection/union are computed over the K-min sets.
264//
265// Two-pointer linear scan over sorted hash lists. Returns
266// Jaccard in parts-per-million (0 = disjoint, 1_000_000 = identical).
267
268func nx_kmv_jaccard_ppm(a: *Kmv, b: *Kmv) -> i64 {
269 if a.k != b.k { return 0 }
270 if a.seed != b.seed { return 0 }
271 if a.n_items == 0 {
272 if b.n_items == 0 { return 1000000 } // both empty: defined as identical
273 return 0
274 }
275 if b.n_items == 0 { return 0 }
276 var i: i64 = 0
277 var j: i64 = 0
278 var intersect: i64 = 0
279 var unite: i64 = 0
280 while i < a.n_items {
281 if j >= b.n_items { i = a.n_items }
282 if i < a.n_items {
283 if j < b.n_items {
284 let va: i64 = a.values[i]
285 let vb: i64 = b.values[j]
286 if va == vb {
287 intersect = intersect + 1
288 unite = unite + 1
289 i = i + 1
290 j = j + 1
291 }
292 if va < vb {
293 unite = unite + 1
294 i = i + 1
295 }
296 if va > vb {
297 unite = unite + 1
298 j = j + 1
299 }
300 }
301 }
302 }
303 // Drain remaining b.
304 while j < b.n_items {
305 unite = unite + 1
306 j = j + 1
307 }
308 if unite == 0 { return 0 }
309 return (intersect * 1000000) / unite
310}
311
312// === introspection ==============================================
313
314func nx_kmv_memory_bytes(kmv: *Kmv) -> i64 {
315 return 40 + kmv.k * 8
316}