nx_sketch_tuple.nx source
↩ module page · 328 lines · 11249 B
1// sketch_tuple.nx -- Tuple sketch (Theta + per-hash auxiliary data).
2//
3// Extends the Theta sketch family by attaching a SCALAR value to each
4// hash slot. Use cases (matching DataSketches Tuple docs):
5// - sum-of-revenue across distinct users (one (hash, revenue) per user)
6// - max-recency across distinct items
7// - count-of-events-per-distinct-user
8//
9// Multiple inserts of the same hash combine via a CALLER-CHOSEN reducer:
10// NX_TUPLE_RED_SUM — add (default for revenue / counts)
11// NX_TUPLE_RED_MAX — keep larger (for recency / peak)
12// NX_TUPLE_RED_MIN — keep smaller (for first-seen / earliest)
13// NX_TUPLE_RED_REPLACE— overwrite (last-write-wins)
14//
15// THETA + AGGREGATE ESTIMATOR:
16// E[sum_over_set] ≈ (sum_over_sample) * HASH_MAX / theta
17// where the sum_over_sample is over the values < theta currently
18// stored. Theta-corrected like Theta cardinality estimator.
19//
20// MERGE (capability beyond DataSketches Tuple-merge spec):
21// - theta_out = min(theta_a, theta_b)
22// - hashes_out = K smallest across union, filtered to < theta_out
23// - value_out for each surviving hash = reducer(a.val, b.val) if both
24// have it, else whichever side had it.
25
26// nx_safety_envelope:
27// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
28// sil_target: SIL1
29// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
30// verdict: NOT_YET_EVALUATED
31
32import "nx_syscalls.nx"
33import "nx_murmur3.nx"
34import "nx_sketch_types.nx"
35
36const NX_TUPLE_HASH_MAX: i64 = 4294967296
37
38const NX_TUPLE_RED_SUM: i64 = 0
39const NX_TUPLE_RED_MAX: i64 = 1
40const NX_TUPLE_RED_MIN: i64 = 2
41const NX_TUPLE_RED_REPLACE: i64 = 3
42
43struct TupleEntry {
44 hash: i64,
45 value: i64,
46}
47
48struct TupleSketch {
49 entries: *TupleEntry, // sorted ascending by hash
50 k: i64,
51 n_items: i64,
52 theta: i64,
53 seed: i64,
54 reducer: i64,
55}
56
57// === reducer ======================================================
58
59func nx_tuple_reduce(reducer: i64, a: i64, b: i64) -> i64 {
60 if reducer == NX_TUPLE_RED_SUM { return a + b }
61 if reducer == NX_TUPLE_RED_MAX {
62 if a > b { return a }
63 return b
64 }
65 if reducer == NX_TUPLE_RED_MIN {
66 if a < b { return a }
67 return b
68 }
69 return b // REPLACE: take new value
70}
71
72// === construction =================================================
73
74func nx_tuple_alloc(k: i64, seed: i64, reducer: i64) -> *TupleSketch {
75 if k < 16 { return 0 as *TupleSketch }
76 if k > 65536 { return 0 as *TupleSketch }
77 if reducer < 0 { return 0 as *TupleSketch }
78 if reducer > NX_TUPLE_RED_REPLACE { return 0 as *TupleSketch }
79 let raw: *u8 = sys_mmap(56)
80 let t: *TupleSketch = raw as *TupleSketch
81 let ent_raw: *u8 = sys_mmap(k * 16)
82 t.entries = ent_raw as *TupleEntry
83 t.k = k
84 t.n_items = 0
85 t.theta = NX_TUPLE_HASH_MAX
86 t.seed = seed
87 t.reducer = reducer
88 return t
89}
90
91func nx_tuple_entry_at(t: *TupleSketch, i: i64) -> *TupleEntry {
92 return (t.entries as i64 + i * 16) as *TupleEntry
93}
94
95// === add (key + value) ============================================
96
97func nx_tuple_add(t: *TupleSketch, key: *u8, key_len: i64, value: i64) -> i64 {
98 let raw_h: i64 = murmur3_32(t.seed, key, key_len) & 0xFFFFFFFF
99 var h: i64 = raw_h
100 if h == 0 { h = 1 }
101 if h >= t.theta { return 0 }
102
103 // Binary search for h in entries[].
104 var lo: i64 = 0
105 var hi: i64 = t.n_items
106 while lo < hi {
107 let mid: i64 = (lo + hi) / 2
108 let e: *TupleEntry = nx_tuple_entry_at(t, mid)
109 if e.hash == h {
110 // Existing -- reduce.
111 e.value = nx_tuple_reduce(t.reducer, e.value, value)
112 return 0
113 }
114 if e.hash < h { lo = mid + 1 }
115 if e.hash > h { hi = mid }
116 }
117 // New entry at index lo.
118 if t.n_items < t.k {
119 var j: i64 = t.n_items
120 while j > lo {
121 let src: *TupleEntry = nx_tuple_entry_at(t, j - 1)
122 let dst: *TupleEntry = nx_tuple_entry_at(t, j)
123 dst.hash = src.hash
124 dst.value = src.value
125 j = j - 1
126 }
127 let slot: *TupleEntry = nx_tuple_entry_at(t, lo)
128 slot.hash = h
129 slot.value = value
130 t.n_items = t.n_items + 1
131 if t.n_items == t.k {
132 let last: *TupleEntry = nx_tuple_entry_at(t, t.k - 1)
133 t.theta = last.hash
134 }
135 return 0
136 }
137 // Full: replace max if h < max.
138 let max_entry: *TupleEntry = nx_tuple_entry_at(t, t.k - 1)
139 if h >= max_entry.hash { return 0 }
140 // Drop max, insert h at position lo.
141 var j2: i64 = t.k - 1
142 while j2 > lo {
143 let src: *TupleEntry = nx_tuple_entry_at(t, j2 - 1)
144 let dst: *TupleEntry = nx_tuple_entry_at(t, j2)
145 dst.hash = src.hash
146 dst.value = src.value
147 j2 = j2 - 1
148 }
149 let slot2: *TupleEntry = nx_tuple_entry_at(t, lo)
150 slot2.hash = h
151 slot2.value = value
152 let new_last: *TupleEntry = nx_tuple_entry_at(t, t.k - 1)
153 t.theta = new_last.hash
154 return 0
155}
156
157// === aggregate (theta-corrected sum-over-distinct-set) ============
158//
159// E[sum_over_set] = (sum_over_sample_under_theta) * HASH_MAX / theta
160// The sum_over_sample excludes the kth_smallest hash (the border)
161// to match the unbiased estimator used in nx_theta_estimate.
162
163func nx_tuple_aggregate(t: *TupleSketch) -> i64 {
164 if t.n_items == 0 { return 0 }
165 if t.theta == NX_TUPLE_HASH_MAX {
166 // No down-sampling -- exact sum.
167 var sum: i64 = 0
168 var i: i64 = 0
169 while i < t.n_items {
170 let e: *TupleEntry = nx_tuple_entry_at(t, i)
171 sum = sum + e.value
172 i = i + 1
173 }
174 return sum
175 }
176 var sum: i64 = 0
177 var i: i64 = 0
178 while i < t.n_items {
179 let e: *TupleEntry = nx_tuple_entry_at(t, i)
180 if e.hash < t.theta {
181 sum = sum + e.value
182 }
183 i = i + 1
184 }
185 return (sum * NX_TUPLE_HASH_MAX) / t.theta
186}
187
188// Cardinality estimate (same shape as Theta).
189func nx_tuple_cardinality(t: *TupleSketch) -> i64 {
190 if t.n_items == 0 { return 0 }
191 if t.theta == NX_TUPLE_HASH_MAX { return t.n_items }
192 var below: i64 = 0
193 var i: i64 = 0
194 while i < t.n_items {
195 let e: *TupleEntry = nx_tuple_entry_at(t, i)
196 if e.hash < t.theta { below = below + 1 }
197 i = i + 1
198 }
199 if below == 0 { return 0 }
200 return (below * NX_TUPLE_HASH_MAX) / t.theta
201}
202
203// === typed query ==================================================
204
205func nx_tuple_stddev_rel_ppb(k: i64) -> i64 {
206 if k <= 16 { return 258000000 }
207 if k <= 64 { return 126000000 }
208 if k <= 256 { return 62700000 }
209 if k <= 1024 { return 31300000 }
210 if k <= 4096 { return 15600000 }
211 if k <= 16384 { return 7810000 }
212 return 3900000
213}
214
215func nx_tuple_query_aggregate(t: *TupleSketch) -> *ApproxI64 {
216 let agg: i64 = nx_tuple_aggregate(t)
217 return nx_approx_new(agg, NX_ENV_REL_STDDEV,
218 nx_tuple_stddev_rel_ppb(t.k),
219 682700000,
220 NX_MATURITY_REFERENCE_IMPL,
221 NX_ADV_HONEST)
222}
223
224func nx_tuple_query_cardinality(t: *TupleSketch) -> *ApproxI64 {
225 let card: i64 = nx_tuple_cardinality(t)
226 return nx_approx_new(card, NX_ENV_REL_STDDEV,
227 nx_tuple_stddev_rel_ppb(t.k),
228 682700000,
229 NX_MATURITY_REFERENCE_IMPL,
230 NX_ADV_HONEST)
231}
232
233// === merge ========================================================
234//
235// Two-pointer linear merge. For hashes present in both, combine
236// values via the (matching) reducer. theta_out = min.
237
238func nx_tuple_min2(a: i64, b: i64) -> i64 {
239 if a < b { return a }
240 return b
241}
242
243func nx_tuple_merge(a: *TupleSketch, b: *TupleSketch) -> *TupleSketch {
244 if a.k != b.k { return 0 as *TupleSketch }
245 if a.seed != b.seed { return 0 as *TupleSketch }
246 if a.reducer != b.reducer { return 0 as *TupleSketch }
247 let out: *TupleSketch = nx_tuple_alloc(a.k, a.seed, a.reducer)
248 out.theta = nx_tuple_min2(a.theta, b.theta)
249
250 var i: i64 = 0
251 var j: i64 = 0
252 var w: i64 = 0
253 var done: i64 = 0
254 while done == 0 {
255 if w >= out.k { done = 1 }
256 if done == 0 {
257 let a_done: i64 = i >= a.n_items
258 let b_done: i64 = j >= b.n_items
259 if a_done == 1 {
260 if b_done == 1 { done = 1 }
261 if b_done == 0 {
262 let eb: *TupleEntry = nx_tuple_entry_at(b, j)
263 j = j + 1
264 if eb.hash < out.theta {
265 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
266 slot.hash = eb.hash
267 slot.value = eb.value
268 w = w + 1
269 }
270 }
271 }
272 if done == 0 {
273 if a_done == 0 {
274 if b_done == 1 {
275 let ea: *TupleEntry = nx_tuple_entry_at(a, i)
276 i = i + 1
277 if ea.hash < out.theta {
278 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
279 slot.hash = ea.hash
280 slot.value = ea.value
281 w = w + 1
282 }
283 }
284 if b_done == 0 {
285 let ea: *TupleEntry = nx_tuple_entry_at(a, i)
286 let eb: *TupleEntry = nx_tuple_entry_at(b, j)
287 if ea.hash == eb.hash {
288 i = i + 1
289 j = j + 1
290 if ea.hash < out.theta {
291 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
292 slot.hash = ea.hash
293 slot.value = nx_tuple_reduce(out.reducer, ea.value, eb.value)
294 w = w + 1
295 }
296 }
297 if ea.hash < eb.hash {
298 i = i + 1
299 if ea.hash < out.theta {
300 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
301 slot.hash = ea.hash
302 slot.value = ea.value
303 w = w + 1
304 }
305 }
306 if ea.hash > eb.hash {
307 j = j + 1
308 if eb.hash < out.theta {
309 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
310 slot.hash = eb.hash
311 slot.value = eb.value
312 w = w + 1
313 }
314 }
315 }
316 }
317 }
318 }
319 }
320 out.n_items = w
321 return out
322}
323
324// === introspection ================================================
325
326func nx_tuple_memory_bytes(t: *TupleSketch) -> i64 {
327 return 56 + t.k * 16
328}