nx_sketch_kll.nx source
↩ module page · 317 lines · 10292 B
1// sketch_kll.nx -- compactor-hierarchy quantile sketch.
2//
3// MRL99-style (Manku-Rajagopalan-Lindsay 1999) "compactor cascade":
4// maintain levels 0, 1, 2, ... each holding up to k items. Items
5// at level h carry weight 2^h. When level h fills, sort it, flip
6// a coin, promote either even-indexed or odd-indexed items to
7// level h+1 (the other half discarded).
8//
9// Karnin-Lang-Liberty 2016 (FOCS) tightens this via geometric
10// per-level capacity decay; we use uniform capacity here for
11// simpler code at slightly looser bound. Rank-error declared
12// conservatively as ~1/sqrt(k) at conf 0.95.
13//
14// Versus Reservoir (sketch_reservoir.nx): KLL/MRL has tighter
15// rank-error at fixed memory because the weighted-promotion
16// cascade preserves rank information that uniform reservoir
17// sampling discards. At k=200, MRL eps ~ 0.07 vs Reservoir
18// eps ~ 0.085 (similar but slightly tighter); the win compounds
19// at higher k.
20//
21// LOSSLESS-LANGUAGE DISCIPLINE (doc 20):
22// Same NX_ENV_RANK_ERROR envelope as Reservoir; substrate doesn't
23// double-count. Multiple primitives sharing an envelope kind is
24// fine -- the discriminator is about the SHAPE of the error, not
25// uniqueness.
26
27// nx_safety_envelope:
28// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
29// sil_target: SIL1
30// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
31// verdict: NOT_YET_EVALUATED
32
33import "nx_syscalls.nx"
34import "nx_sketch_types.nx"
35
36const NX_KLL_K_MIN: i64 = 8
37const NX_KLL_K_MAX: i64 = 10000
38const NX_KLL_MAX_LEVELS: i64 = 16
39
40// LCG constants (same as Reservoir).
41const NX_KLL_LCG_A: i64 = 1103515245
42const NX_KLL_LCG_C: i64 = 12345
43const NX_KLL_LCG_MOD: i64 = 0x7FFFFFFF
44
45struct Kll {
46 k: i64,
47 n_levels: i64, // levels currently allocated
48 level_items: *i64, // contiguous: max_levels * k entries
49 level_count: *i64, // current item count per level
50 total_items: i64,
51 min_val: i64,
52 max_val: i64,
53 rng_state: i64,
54}
55
56// === construction =================================================
57
58func nx_kll_alloc(k: i64, seed: i64) -> *Kll {
59 if k < NX_KLL_K_MIN { return 0 as *Kll }
60 if k > NX_KLL_K_MAX { return 0 as *Kll }
61 let raw: *u8 = sys_mmap(64)
62 let s: *Kll = raw as *Kll
63 s.k = k
64 s.n_levels = 1 // level 0 always present
65 let items_bytes: i64 = NX_KLL_MAX_LEVELS * k * 8
66 let items_raw: *u8 = sys_mmap(items_bytes)
67 s.level_items = items_raw as *i64
68 let count_raw: *u8 = sys_mmap(NX_KLL_MAX_LEVELS * 8)
69 s.level_count = count_raw as *i64
70 var i: i64 = 0
71 while i < NX_KLL_MAX_LEVELS {
72 s.level_count[i] = 0
73 i = i + 1
74 }
75 s.total_items = 0
76 // Sentinels for min/max -- use first add to bootstrap.
77 s.min_val = 0x7FFFFFFFFFFFFFFF
78 s.max_val = -1 - 0x7FFFFFFFFFFFFFFF // i64 min
79 s.rng_state = seed | 1
80 return s
81}
82
83// LCG state mutation.
84func nx_kll_rng_next(s: *Kll) -> i64 {
85 let next: i64 = ((s.rng_state * NX_KLL_LCG_A) + NX_KLL_LCG_C) & NX_KLL_LCG_MOD
86 s.rng_state = next
87 return next
88}
89
90// === level-buffer access helpers =================================
91// Buffer for level h starts at level_items[h * k].
92
93func nx_kll_level_addr(s: *Kll, level: i64) -> *i64 {
94 return (s.level_items as i64 + level * s.k * 8) as *i64
95}
96
97func nx_kll_sort_level(s: *Kll, level: i64) -> i64 {
98 let count: i64 = s.level_count[level]
99 let buf: *i64 = nx_kll_level_addr(s, level)
100 var i: i64 = 1
101 while i < count {
102 let cur: i64 = buf[i]
103 var j: i64 = i - 1
104 var done: i64 = 0
105 while done == 0 {
106 if j < 0 { done = 1 }
107 if done == 0 {
108 let prev: i64 = buf[j]
109 if prev <= cur { done = 1 }
110 if done == 0 {
111 buf[j + 1] = prev
112 j = j - 1
113 }
114 }
115 }
116 buf[j + 1] = cur
117 i = i + 1
118 }
119 return 0
120}
121
122// === compact: sort + coin-flip + promote half ====================
123
124func nx_kll_compact(s: *Kll, level: i64) -> i64 {
125 if level >= NX_KLL_MAX_LEVELS - 1 {
126 // No room to promote -- drop the level (rare; happens only
127 // at extreme stream sizes).
128 s.level_count[level] = 0
129 return 0
130 }
131 nx_kll_sort_level(s, level)
132 let count: i64 = s.level_count[level]
133 let buf: *i64 = nx_kll_level_addr(s, level)
134 // Coin flip: 0 = take even indices (0, 2, ...), 1 = take odd.
135 let pick_odd: i64 = nx_kll_rng_next(s) & 1
136 // Promote chosen half to level+1.
137 let next_level: i64 = level + 1
138 let next_buf: *i64 = nx_kll_level_addr(s, next_level)
139 let next_count_before: i64 = s.level_count[next_level]
140 var i: i64 = pick_odd // start at 0 or 1
141 var promoted: i64 = 0
142 while i < count {
143 let dst_idx: i64 = next_count_before + promoted
144 if dst_idx < s.k {
145 next_buf[dst_idx] = buf[i]
146 promoted = promoted + 1
147 }
148 i = i + 2
149 }
150 s.level_count[next_level] = next_count_before + promoted
151 s.level_count[level] = 0
152 // Grow n_levels if we just pushed into a new level.
153 if next_level >= s.n_levels {
154 s.n_levels = next_level + 1
155 }
156 // Cascade if next level is now full.
157 if s.level_count[next_level] >= s.k {
158 nx_kll_compact(s, next_level)
159 }
160 return 0
161}
162
163// === add ==========================================================
164
165func nx_kll_add(s: *Kll, value: i64) -> i64 {
166 if value < s.min_val { s.min_val = value }
167 if value > s.max_val { s.max_val = value }
168 s.total_items = s.total_items + 1
169 let buf: *i64 = nx_kll_level_addr(s, 0)
170 let count: i64 = s.level_count[0]
171 buf[count] = value
172 s.level_count[0] = count + 1
173 if s.level_count[0] >= s.k {
174 nx_kll_compact(s, 0)
175 }
176 return 0
177}
178
179// === total weight (for normalisation) ============================
180
181func nx_kll_total_weight(s: *Kll) -> i64 {
182 var total: i64 = 0
183 var h: i64 = 0
184 while h < s.n_levels {
185 total = total + s.level_count[h] * (1 << h)
186 h = h + 1
187 }
188 return total
189}
190
191// === build sorted weighted view + query ==========================
192//
193// Concatenate every level's items with weight=2^h, sort by value,
194// scan accumulating cumulative weight, return value at target rank.
195// Sort is insertion-style over the combined view (O(N log N) at
196// query but N is small -- bounded by k * log(total_items)).
197
198func nx_kll_quantile(s: *Kll, p_milli: i64) -> i64 {
199 if s.total_items == 0 { return 0 }
200 let total_weight: i64 = nx_kll_total_weight(s)
201 if total_weight == 0 { return 0 }
202 // Materialise view: collect (value, weight) pairs. Cap entry
203 // count at max_levels * k.
204 let view_cap: i64 = NX_KLL_MAX_LEVELS * s.k
205 let val_raw: *u8 = sys_mmap(view_cap * 8)
206 let wt_raw: *u8 = sys_mmap(view_cap * 8)
207 let view_vals: *i64 = val_raw as *i64
208 let view_wts: *i64 = wt_raw as *i64
209 var n: i64 = 0
210 var h: i64 = 0
211 while h < s.n_levels {
212 let buf: *i64 = nx_kll_level_addr(s, h)
213 let cnt: i64 = s.level_count[h]
214 let w: i64 = 1 << h
215 var i: i64 = 0
216 while i < cnt {
217 view_vals[n] = buf[i]
218 view_wts[n] = w
219 n = n + 1
220 i = i + 1
221 }
222 h = h + 1
223 }
224 // Sort the view by value (insertion sort; n is small).
225 var i: i64 = 1
226 while i < n {
227 let cur_v: i64 = view_vals[i]
228 let cur_w: i64 = view_wts[i]
229 var j: i64 = i - 1
230 var done: i64 = 0
231 while done == 0 {
232 if j < 0 { done = 1 }
233 if done == 0 {
234 if view_vals[j] <= cur_v { done = 1 }
235 if done == 0 {
236 view_vals[j + 1] = view_vals[j]
237 view_wts[j + 1] = view_wts[j]
238 j = j - 1
239 }
240 }
241 }
242 view_vals[j + 1] = cur_v
243 view_wts[j + 1] = cur_w
244 i = i + 1
245 }
246 // Walk sorted view, find target rank.
247 let target: i64 = (p_milli * total_weight) / 1000
248 var cum: i64 = 0
249 var k: i64 = 0
250 while k < n {
251 cum = cum + view_wts[k]
252 if cum >= target { return view_vals[k] }
253 k = k + 1
254 }
255 return view_vals[n - 1]
256}
257
258// === rank-of-value ===============================================
259
260func nx_kll_rank(s: *Kll, value: i64) -> i64 {
261 if s.total_items == 0 { return 0 }
262 if value < s.min_val { return 0 }
263 if value >= s.max_val { return 1000 }
264 let total_weight: i64 = nx_kll_total_weight(s)
265 if total_weight == 0 { return 0 }
266 // Sum weights of items with value <= query value.
267 var cum: i64 = 0
268 var h: i64 = 0
269 while h < s.n_levels {
270 let buf: *i64 = nx_kll_level_addr(s, h)
271 let cnt: i64 = s.level_count[h]
272 let w: i64 = 1 << h
273 var i: i64 = 0
274 while i < cnt {
275 if buf[i] <= value {
276 cum = cum + w
277 }
278 i = i + 1
279 }
280 h = h + 1
281 }
282 return (cum * 1000) / total_weight
283}
284
285// === error bound + query =========================================
286//
287// MRL/KLL rank-error: ~1.36/sqrt(k) at 95% confidence (Hoeffding-
288// style conservative bound, same shape as Reservoir but k is the
289// per-level capacity, not the total reservoir size).
290
291func nx_kll_rank_error_ppb(k: i64) -> i64 {
292 if k <= 8 { return 480000000 } // 0.48
293 if k <= 32 { return 240000000 }
294 if k <= 128 { return 120000000 } // 0.12
295 if k <= 512 { return 60000000 } // 0.06
296 if k <= 2048 { return 30000000 } // 0.03
297 return 15000000 // 0.015
298}
299
300func nx_kll_query_quantile(s: *Kll, p_milli: i64) -> *ApproxI64 {
301 let v: i64 = nx_kll_quantile(s, p_milli)
302 return nx_approx_new(v, NX_ENV_RANK_ERROR,
303 nx_kll_rank_error_ppb(s.k),
304 950000000,
305 NX_MATURITY_REFERENCE_IMPL,
306 NX_ADV_HONEST)
307}
308
309// === memory introspection ========================================
310
311func nx_kll_memory_bytes(s: *Kll) -> i64 {
312 return 64 + NX_KLL_MAX_LEVELS * s.k * 8 + NX_KLL_MAX_LEVELS * 8
313}
314
315func nx_kll_levels_used(s: *Kll) -> i64 {
316 return s.n_levels
317}