nx_sketch_space_saving.nx source
↩ module page · 226 lines · 7223 B
1// sketch_space_saving.nx -- SpaceSaving top-K heavy hitters.
2//
3// Metwally-Agrawal-El-Abbadi 2005. Bounded set of (key, count,
4// error) triples of capacity K. On add: if key tracked, increment;
5// else if room, insert with count=1, error=0; else evict the
6// minimum-count entry and install (new_key, min_count + count,
7// min_count).
8//
9// Guarantee: any item with true frequency > N/K is tracked (no
10// false negatives for heavy hitters). The reported count
11// over-estimates by at most `error`; true_count is in
12// [count - error, count] for tracked items.
13//
14// COMPLEMENTS CMS:
15// - CMS: per-query frequency for ANY key (overestimate-only).
16// - SpaceSaving: enumerate the top-K most frequent keys directly.
17// Together they cover the full frequency-counting axis.
18//
19// API takes i64 keys (caller hashes strings to i64 via murmur3 if
20// they have string-keyed streams).
21//
22// LOSSLESS-LANGUAGE DISCIPLINE (doc 20):
23// nx_ss_query returns ApproxI64 with envelope_kind = NX_ENV_ABS;
24// param_a holds the per-key error bound (count - error <= true <=
25// count); conf_ppb = 1e9 (deterministic; not probabilistic).
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_SS_K_MIN: i64 = 2
37const NX_SS_K_MAX: i64 = 100000
38
39// Counter triple: 24 bytes per entry.
40struct SsCounter {
41 key: i64,
42 count: i64,
43 error: i64,
44}
45
46struct SpaceSaving {
47 k: i64,
48 n_tracked: i64, // <= k
49 counters: *SsCounter,
50 total_count: i64,
51}
52
53// === construction =================================================
54
55func nx_ss_alloc(k: i64) -> *SpaceSaving {
56 if k < NX_SS_K_MIN { return 0 as *SpaceSaving }
57 if k > NX_SS_K_MAX { return 0 as *SpaceSaving }
58 let raw: *u8 = sys_mmap(40)
59 let s: *SpaceSaving = raw as *SpaceSaving
60 s.k = k
61 s.n_tracked = 0
62 let table_bytes: i64 = k * 24
63 let table_raw: *u8 = sys_mmap(table_bytes)
64 s.counters = table_raw as *SsCounter
65 s.total_count = 0
66 return s
67}
68
69// === find / insert helpers ========================================
70
71func nx_ss_find_key(s: *SpaceSaving, key: i64) -> i64 {
72 // Returns index of counter holding `key`, or -1.
73 var i: i64 = 0
74 while i < s.n_tracked {
75 let c: *SsCounter = (s.counters as i64 + i * 24) as *SsCounter
76 if c.key == key { return i }
77 i = i + 1
78 }
79 return -1
80}
81
82func nx_ss_find_min(s: *SpaceSaving) -> i64 {
83 // Returns index of counter with smallest count (any tie broken
84 // by lowest index).
85 if s.n_tracked == 0 { return -1 }
86 var min_idx: i64 = 0
87 let c0: *SsCounter = s.counters
88 var min_count: i64 = c0.count
89 var i: i64 = 1
90 while i < s.n_tracked {
91 let c: *SsCounter = (s.counters as i64 + i * 24) as *SsCounter
92 if c.count < min_count {
93 min_count = c.count
94 min_idx = i
95 }
96 i = i + 1
97 }
98 return min_idx
99}
100
101// === add ==========================================================
102
103func nx_ss_add(s: *SpaceSaving, key: i64, count: i64) -> i64 {
104 if count <= 0 { return 0 }
105 s.total_count = s.total_count + count
106 let pos: i64 = nx_ss_find_key(s, key)
107 if pos >= 0 {
108 let c: *SsCounter = (s.counters as i64 + pos * 24) as *SsCounter
109 c.count = c.count + count
110 return 0
111 }
112 if s.n_tracked < s.k {
113 let slot: *SsCounter = (s.counters as i64 + s.n_tracked * 24) as *SsCounter
114 slot.key = key
115 slot.count = count
116 slot.error = 0
117 s.n_tracked = s.n_tracked + 1
118 return 0
119 }
120 // Evict minimum.
121 let min_idx: i64 = nx_ss_find_min(s)
122 let m: *SsCounter = (s.counters as i64 + min_idx * 24) as *SsCounter
123 let evicted_count: i64 = m.count
124 m.key = key
125 m.count = evicted_count + count
126 m.error = evicted_count
127 return 0
128}
129
130// === query ========================================================
131
132func nx_ss_estimate(s: *SpaceSaving, key: i64) -> i64 {
133 let pos: i64 = nx_ss_find_key(s, key)
134 if pos < 0 { return 0 }
135 let c: *SsCounter = (s.counters as i64 + pos * 24) as *SsCounter
136 return c.count
137}
138
139func nx_ss_lower_bound(s: *SpaceSaving, key: i64) -> i64 {
140 let pos: i64 = nx_ss_find_key(s, key)
141 if pos < 0 { return 0 }
142 let c: *SsCounter = (s.counters as i64 + pos * 24) as *SsCounter
143 return c.count - c.error
144}
145
146func nx_ss_error(s: *SpaceSaving, key: i64) -> i64 {
147 let pos: i64 = nx_ss_find_key(s, key)
148 if pos < 0 { return 0 }
149 let c: *SsCounter = (s.counters as i64 + pos * 24) as *SsCounter
150 return c.error
151}
152
153// Maximum possible over-count: per Metwally et al. the error of
154// any tracked entry is bounded by ceil(N / k).
155func nx_ss_max_overcount(s: *SpaceSaving) -> i64 {
156 return (s.total_count + s.k - 1) / s.k
157}
158
159func nx_ss_query(s: *SpaceSaving, key: i64) -> *ApproxI64 {
160 let est: i64 = nx_ss_estimate(s, key)
161 return nx_approx_new(est, NX_ENV_ABS, nx_ss_max_overcount(s),
162 1000000000,
163 NX_MATURITY_REFERENCE_IMPL,
164 NX_ADV_HONEST)
165}
166
167// === top-K enumeration ===========================================
168//
169// Return the top-N keys by count (descending). Caller pre-allocates
170// the (keys, counts) arrays of size n. Uses simple sort-then-copy
171// (k is small).
172
173func if_min(a: i64, b: i64) -> i64 {
174 if a < b { return a }
175 return b
176}
177
178func nx_ss_top_k(s: *SpaceSaving, n: i64, out_keys: *i64, out_counts: *i64) -> i64 {
179 // Build an index array, sort by count descending, materialize.
180 let limit: i64 = if_min(n, s.n_tracked)
181 // Insertion sort over counter indices by count descending.
182 let idx_raw: *u8 = sys_mmap(s.n_tracked * 8)
183 let idx: *i64 = idx_raw as *i64
184 var i: i64 = 0
185 while i < s.n_tracked { idx[i] = i; i = i + 1 }
186 // Sort idx[0..n_tracked) by counters[idx[*]].count descending.
187 i = 1
188 while i < s.n_tracked {
189 let cur: i64 = idx[i]
190 let cur_ptr: *SsCounter = (s.counters as i64 + cur * 24) as *SsCounter
191 let cur_count: i64 = cur_ptr.count
192 var j: i64 = i - 1
193 var done: i64 = 0
194 while done == 0 {
195 if j < 0 { done = 1 }
196 if done == 0 {
197 let prev_ptr: *SsCounter = (s.counters as i64 + idx[j] * 24) as *SsCounter
198 let prev_count: i64 = prev_ptr.count
199 if prev_count >= cur_count {
200 done = 1
201 }
202 if done == 0 {
203 idx[j + 1] = idx[j]
204 j = j - 1
205 }
206 }
207 }
208 idx[j + 1] = cur
209 i = i + 1
210 }
211 i = 0
212 while i < limit {
213 let ci: i64 = idx[i]
214 let c: *SsCounter = (s.counters as i64 + ci * 24) as *SsCounter
215 out_keys[i] = c.key
216 out_counts[i] = c.count
217 i = i + 1
218 }
219 return limit
220}
221
222// === memory introspection =========================================
223
224func nx_ss_memory_bytes(s: *SpaceSaving) -> i64 {
225 return 40 + s.k * 24
226}