sketch_tuple.nx source
↩ module page · 322 lines · 11311 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
26import "syscalls.nx"
27import "murmur3.nx"
28import "sketch_types.nx"
29
30const NX_TUPLE_HASH_MAX: i64 = 4294967296
31
32const NX_TUPLE_RED_SUM: i64 = 0
33const NX_TUPLE_RED_MAX: i64 = 1
34const NX_TUPLE_RED_MIN: i64 = 2
35const NX_TUPLE_RED_REPLACE: i64 = 3
36
37struct TupleEntry {
38 hash: i64,
39 value: i64,
40}
41
42struct TupleSketch {
43 entries: *TupleEntry, // sorted ascending by hash
44 k: i64,
45 n_items: i64,
46 theta: i64,
47 seed: i64,
48 reducer: i64,
49}
50
51// === reducer ======================================================
52
53func nx_tuple_reduce(reducer: i64, a: i64, b: i64) -> i64 {
54 if reducer == NX_TUPLE_RED_SUM { return a + b }
55 if reducer == NX_TUPLE_RED_MAX {
56 if a > b { return a }
57 return b
58 }
59 if reducer == NX_TUPLE_RED_MIN {
60 if a < b { return a }
61 return b
62 }
63 return b // REPLACE: take new value
64}
65
66// === construction =================================================
67
68func nx_tuple_alloc(k: i64, seed: i64, reducer: i64) -> *TupleSketch {
69 if k < 16 { return 0 as *TupleSketch }
70 if k > 65536 { return 0 as *TupleSketch }
71 if reducer < 0 { return 0 as *TupleSketch }
72 if reducer > NX_TUPLE_RED_REPLACE { return 0 as *TupleSketch }
73 let raw: *u8 = sys_mmap(56)
74 let t: *TupleSketch = raw as *TupleSketch
75 let ent_raw: *u8 = sys_mmap(k * 16)
76 t.entries = ent_raw as *TupleEntry
77 t.k = k
78 t.n_items = 0
79 t.theta = NX_TUPLE_HASH_MAX
80 t.seed = seed
81 t.reducer = reducer
82 return t
83}
84
85func nx_tuple_entry_at(t: *TupleSketch, i: i64) -> *TupleEntry {
86 return (t.entries as i64 + i * 16) as *TupleEntry
87}
88
89// === add (key + value) ============================================
90
91func nx_tuple_add(t: *TupleSketch, key: *u8, key_len: i64, value: i64) -> i64 {
92 let raw_h: i64 = murmur3_32(t.seed, key, key_len) & 0xFFFFFFFF
93 var h: i64 = raw_h
94 if h == 0 { h = 1 }
95 if h >= t.theta { return 0 }
96
97 // Binary search for h in entries[].
98 var lo: i64 = 0
99 var hi: i64 = t.n_items
100 while lo < hi {
101 let mid: i64 = (lo + hi) / 2
102 let e: *TupleEntry = nx_tuple_entry_at(t, mid)
103 if e.hash == h {
104 // Existing -- reduce.
105 e.value = nx_tuple_reduce(t.reducer, e.value, value)
106 return 0
107 }
108 if e.hash < h { lo = mid + 1 }
109 if e.hash > h { hi = mid }
110 }
111 // New entry at index lo.
112 if t.n_items < t.k {
113 var j: i64 = t.n_items
114 while j > lo {
115 let src: *TupleEntry = nx_tuple_entry_at(t, j - 1)
116 let dst: *TupleEntry = nx_tuple_entry_at(t, j)
117 dst.hash = src.hash
118 dst.value = src.value
119 j = j - 1
120 }
121 let slot: *TupleEntry = nx_tuple_entry_at(t, lo)
122 slot.hash = h
123 slot.value = value
124 t.n_items = t.n_items + 1
125 if t.n_items == t.k {
126 let last: *TupleEntry = nx_tuple_entry_at(t, t.k - 1)
127 t.theta = last.hash
128 }
129 return 0
130 }
131 // Full: replace max if h < max.
132 let max_entry: *TupleEntry = nx_tuple_entry_at(t, t.k - 1)
133 if h >= max_entry.hash { return 0 }
134 // Drop max, insert h at position lo.
135 var j2: i64 = t.k - 1
136 while j2 > lo {
137 let src: *TupleEntry = nx_tuple_entry_at(t, j2 - 1)
138 let dst: *TupleEntry = nx_tuple_entry_at(t, j2)
139 dst.hash = src.hash
140 dst.value = src.value
141 j2 = j2 - 1
142 }
143 let slot2: *TupleEntry = nx_tuple_entry_at(t, lo)
144 slot2.hash = h
145 slot2.value = value
146 let new_last: *TupleEntry = nx_tuple_entry_at(t, t.k - 1)
147 t.theta = new_last.hash
148 return 0
149}
150
151// === aggregate (theta-corrected sum-over-distinct-set) ============
152//
153// E[sum_over_set] = (sum_over_sample_under_theta) * HASH_MAX / theta
154// The sum_over_sample excludes the kth_smallest hash (the border)
155// to match the unbiased estimator used in nx_theta_estimate.
156
157func nx_tuple_aggregate(t: *TupleSketch) -> i64 {
158 if t.n_items == 0 { return 0 }
159 if t.theta == NX_TUPLE_HASH_MAX {
160 // No down-sampling -- exact sum.
161 var sum: i64 = 0
162 var i: i64 = 0
163 while i < t.n_items {
164 let e: *TupleEntry = nx_tuple_entry_at(t, i)
165 sum = sum + e.value
166 i = i + 1
167 }
168 return sum
169 }
170 var sum: i64 = 0
171 var i: i64 = 0
172 while i < t.n_items {
173 let e: *TupleEntry = nx_tuple_entry_at(t, i)
174 if e.hash < t.theta {
175 sum = sum + e.value
176 }
177 i = i + 1
178 }
179 return (sum * NX_TUPLE_HASH_MAX) / t.theta
180}
181
182// Cardinality estimate (same shape as Theta).
183func nx_tuple_cardinality(t: *TupleSketch) -> i64 {
184 if t.n_items == 0 { return 0 }
185 if t.theta == NX_TUPLE_HASH_MAX { return t.n_items }
186 var below: i64 = 0
187 var i: i64 = 0
188 while i < t.n_items {
189 let e: *TupleEntry = nx_tuple_entry_at(t, i)
190 if e.hash < t.theta { below = below + 1 }
191 i = i + 1
192 }
193 if below == 0 { return 0 }
194 return (below * NX_TUPLE_HASH_MAX) / t.theta
195}
196
197// === typed query ==================================================
198
199func nx_tuple_stddev_rel_ppb(k: i64) -> i64 {
200 if k <= 16 { return 258000000 }
201 if k <= 64 { return 126000000 }
202 if k <= 256 { return 62700000 }
203 if k <= 1024 { return 31300000 }
204 if k <= 4096 { return 15600000 }
205 if k <= 16384 { return 7810000 }
206 return 3900000
207}
208
209func nx_tuple_query_aggregate(t: *TupleSketch) -> *ApproxI64 {
210 let agg: i64 = nx_tuple_aggregate(t)
211 return nx_approx_new(agg, NX_ENV_REL_STDDEV,
212 nx_tuple_stddev_rel_ppb(t.k),
213 682700000,
214 NX_MATURITY_REFERENCE_IMPL,
215 NX_ADV_HONEST)
216}
217
218func nx_tuple_query_cardinality(t: *TupleSketch) -> *ApproxI64 {
219 let card: i64 = nx_tuple_cardinality(t)
220 return nx_approx_new(card, NX_ENV_REL_STDDEV,
221 nx_tuple_stddev_rel_ppb(t.k),
222 682700000,
223 NX_MATURITY_REFERENCE_IMPL,
224 NX_ADV_HONEST)
225}
226
227// === merge ========================================================
228//
229// Two-pointer linear merge. For hashes present in both, combine
230// values via the (matching) reducer. theta_out = min.
231
232func nx_tuple_min2(a: i64, b: i64) -> i64 {
233 if a < b { return a }
234 return b
235}
236
237func nx_tuple_merge(a: *TupleSketch, b: *TupleSketch) -> *TupleSketch {
238 if a.k != b.k { return 0 as *TupleSketch }
239 if a.seed != b.seed { return 0 as *TupleSketch }
240 if a.reducer != b.reducer { return 0 as *TupleSketch }
241 let out: *TupleSketch = nx_tuple_alloc(a.k, a.seed, a.reducer)
242 out.theta = nx_tuple_min2(a.theta, b.theta)
243
244 var i: i64 = 0
245 var j: i64 = 0
246 var w: i64 = 0
247 var done: i64 = 0
248 while done == 0 {
249 if w >= out.k { done = 1 }
250 if done == 0 {
251 let a_done: i64 = i >= a.n_items
252 let b_done: i64 = j >= b.n_items
253 if a_done == 1 {
254 if b_done == 1 { done = 1 }
255 if b_done == 0 {
256 let eb: *TupleEntry = nx_tuple_entry_at(b, j)
257 j = j + 1
258 if eb.hash < out.theta {
259 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
260 slot.hash = eb.hash
261 slot.value = eb.value
262 w = w + 1
263 }
264 }
265 }
266 if done == 0 {
267 if a_done == 0 {
268 if b_done == 1 {
269 let ea: *TupleEntry = nx_tuple_entry_at(a, i)
270 i = i + 1
271 if ea.hash < out.theta {
272 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
273 slot.hash = ea.hash
274 slot.value = ea.value
275 w = w + 1
276 }
277 }
278 if b_done == 0 {
279 let ea: *TupleEntry = nx_tuple_entry_at(a, i)
280 let eb: *TupleEntry = nx_tuple_entry_at(b, j)
281 if ea.hash == eb.hash {
282 i = i + 1
283 j = j + 1
284 if ea.hash < out.theta {
285 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
286 slot.hash = ea.hash
287 slot.value = nx_tuple_reduce(out.reducer, ea.value, eb.value)
288 w = w + 1
289 }
290 }
291 if ea.hash < eb.hash {
292 i = i + 1
293 if ea.hash < out.theta {
294 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
295 slot.hash = ea.hash
296 slot.value = ea.value
297 w = w + 1
298 }
299 }
300 if ea.hash > eb.hash {
301 j = j + 1
302 if eb.hash < out.theta {
303 let slot: *TupleEntry = nx_tuple_entry_at(out, w)
304 slot.hash = eb.hash
305 slot.value = eb.value
306 w = w + 1
307 }
308 }
309 }
310 }
311 }
312 }
313 }
314 out.n_items = w
315 return out
316}
317
318// === introspection ================================================
319
320func nx_tuple_memory_bytes(t: *TupleSketch) -> i64 {
321 return 56 + t.k * 16
322}