code wiki / _hdl_build / nx_analyst_store.nx
nx_analyst_store.nx source
↩ module page · 181 lines · 8879 B
1// nx_analyst_store.nx -- BRIDGE: run the analytics stack on REAL STORED DATA (the "make it real" step -- the
2// analyst pointed at seg_store records, not just hand-built in-memory columns). Scans multi-valued records
3// under a key, parses a TAB-field of each record's value into an i64 column, and hands it to the analytics
4// (nx_analyst_report / nx_analyst_data / nx_dataframe). This is what turns the analytics LIBRARY into an
5// analyst you point at your actual sovereign store. license_tier: ORIGINAL
6import "nx_syscalls.nx"
7import "nx_seg_store.nx"
8import "_hdl_build/nx_analyst_report.nx"
9
10// parse the tab-delimited field `fidx` of val[0..vlen) as i64 (leading '-' allowed; stops at non-digit).
11func asr_field_i64(val: *u8, vlen: i64, fidx: i64) -> i64 {
12 var i: i64 = 0
13 var f: i64 = 0
14 while f < fidx {
15 var found: i64 = 0
16 while found == 0 {
17 if i >= vlen { return 0 }
18 if val[i] == (9 as u8) { found = 1 }
19 i = i + 1
20 }
21 f = f + 1
22 }
23 var neg: i64 = 0
24 if i < vlen { if val[i] == (45 as u8) { neg = 1; i = i + 1 } }
25 var n: i64 = 0
26 var go: i64 = 1
27 while go == 1 {
28 if i >= vlen { go = 0 } else {
29 let c: i64 = val[i] as i64
30 if c < 48 { go = 0 } else { if c > 57 { go = 0 } else { n = n * 10 + (c - 48); i = i + 1 } }
31 }
32 }
33 if neg == 1 { return 0 - n }
34 return n
35}
36
37// ★★SEG_STORE CONTRACT, corrected 2026-07-23 (F1001). ss_scan retains at most SS_VER_WINDOW (256) versions
38// of ONE key and on overflow the window SLIDES -- the OLDEST drop out. So a 10,000-row dataset stored as
39// versions of one key does not "cap at 256": it silently returns the NEWEST 256, and every statistic
40// computed on it describes a recency slice the caller never asked for. That is not a ceiling, it is silent
41// data loss -- and the old note here ("honest about the ceiling: it returns what the store returns") was
42// wrong, because the caller had no way to learn it had happened.
43// Two fixes: (a) every load now REPORTS whether it touched the window, and (b) a SHARDED loader spreads
44// rows over <keybase>:0, <keybase>:1 ... so real datasets never go near it.
45// ⚠Fail-closed at EQUALITY: a load returning exactly SS_VER_WINDOW rows cannot be distinguished from a
46// truncated one, so it is flagged. Under-trusting that boundary beats over-trusting a truncated load.
47// scratch buffers for ss_scan must hold SS_VER_WINDOW entries whatever the caller's max_n is -- ss_scan
48// fills the CALLER's arrays up to its own window, so sizing them by max_n alone under-allocates for any
49// max_n < SS_VER_WINDOW (latent out-of-bounds write; found 2026-07-23 reading this for F1001).
50func asr_scan_slots(max_n: i64) -> i64 {
51 var s: i64 = max_n + 16
52 if s < SS_VER_SLOTS { s = SS_VER_SLOTS }
53 return s
54}
55
56// load field `fidx` of every record under `key` into out_col; returns row count (<=max_n).
57// flags[0] is set to 1 when the scan reached SS_VER_WINDOW -- i.e. the rows returned may be a TRUNCATED
58// newest-first slice rather than the whole key. Pass 0 as *i64 if you genuinely do not care.
59func asr_load_field_win(prefix: *u8, key: *u8, fidx: i64, out_col: *i64, max_n: i64, flags: *i64) -> i64 {
60 let slots: i64 = asr_scan_slots(max_n)
61 let kinds: *i64 = sys_mmap(8 * slots) as *i64
62 let ptrs: *i64 = sys_mmap(8 * slots) as *i64
63 let lens: *i64 = sys_mmap(8 * slots) as *i64
64 let cnt: i64 = ss_scan(prefix, key, kinds, ptrs, lens)
65 if (flags as i64) != 0 { if cnt >= SS_VER_WINDOW { flags[0] = 1 } }
66 var m: i64 = cnt
67 if m > max_n { m = max_n }
68 var i: i64 = 0
69 while i < m {
70 let v: *u8 = ptrs[i] as *u8
71 out_col[i] = asr_field_i64(v, lens[i], fidx)
72 i = i + 1
73 }
74 return m
75}
76// unchanged signature for existing callers (rule 19); the window-aware form is above.
77func asr_load_field(prefix: *u8, key: *u8, fidx: i64, out_col: *i64, max_n: i64) -> i64 {
78 return asr_load_field_win(prefix, key, fidx, out_col, max_n, 0 as *i64)
79}
80
81// "<keybase>:<idx>" -- the shard key convention. Returns bytes written (out is NUL-terminated).
82func asr_shard_key(keybase: *u8, idx: i64, out: *u8) -> i64 {
83 var o: i64 = 0
84 while keybase[o] != (0 as u8) { out[o] = keybase[o]; o = o + 1 }
85 out[o] = 58 as u8
86 o = o + 1
87 var m: i64 = idx
88 let t: *u8 = sys_mmap(24)
89 var k: i64 = 0
90 if m == 0 { t[0] = 48 as u8; k = 1 }
91 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
92 while k > 0 { k = k - 1; out[o] = t[k]; o = o + 1 }
93 out[o] = 0 as u8
94 return o
95}
96
97// THE SCALE PATH: load field `fidx` across shard keys <keybase>:0, <keybase>:1, ... concatenated IN SHARD
98// ORDER, so row order is stable and reproducible. Stops at the first empty shard. flags[0] is set when any
99// individual shard hit the version window (that shard was itself over-filled -- reshard it smaller) or when
100// maxshards was exhausted while the last shard was still returning rows (more data remains unread).
101func asr_load_field_sharded(prefix: *u8, keybase: *u8, fidx: i64, out_col: *i64, max_n: i64, maxshards: i64, flags: *i64) -> i64 {
102 let kb: *u8 = sys_mmap(256)
103 let sf: *i64 = sys_mmap(8) as *i64
104 var total: i64 = 0
105 var s: i64 = 0
106 var more: i64 = 1
107 while s < maxshards {
108 if more == 1 {
109 if total < max_n {
110 asr_shard_key(keybase, s, kb)
111 sf[0] = 0
112 let room: i64 = max_n - total
113 let got: i64 = asr_load_field_win(prefix, kb, fidx, (((out_col as i64) + total * 8) as *i64), room, sf)
114 if sf[0] == 1 { if (flags as i64) != 0 { flags[0] = 1 } }
115 if got == 0 { more = 0 } else { total = total + got }
116 } else { more = 0 }
117 }
118 s = s + 1
119 }
120 // exhausted the shard budget while the last shard still had rows => data remains unread
121 if more == 1 { if (flags as i64) != 0 { flags[0] = 1 } }
122 return total
123}
124
125// full "analyze a stored dataset": load `nfields` tab-fields of the records under `key` into columns, then
126// run the dataset analyst against target_field. names = array of field-name pointers. Returns report bytes.
127func asr_analyze_stored(prefix: *u8, key: *u8, field_names: *i64, nfields: i64, target_field: i64, n_max: i64, out: *u8, cap: i64) -> i64 {
128 if nfields <= 0 { return 0 }
129 if nfields > 32 { return 0 }
130 let cols: *i64 = sys_mmap(8 * nfields) as *i64
131 let fl: *i64 = sys_mmap(8) as *i64
132 fl[0] = 0
133 var rows: i64 = 0
134 var f: i64 = 0
135 while f < nfields {
136 let colf: *i64 = sys_mmap(8 * n_max) as *i64
137 rows = asr_load_field_win(prefix, key, f, colf, n_max, fl)
138 cols[f] = colf as i64
139 f = f + 1
140 }
141 if rows <= 0 { return 0 }
142 // A report computed on a truncated slice must SAY SO before its first number. Silence here is how a
143 // recency slice of 256 rows gets read as "the dataset".
144 var o: i64 = 0
145 if fl[0] == 1 {
146 o = arx_cat(out, o, "!! TRUNCATED LOAD -- this key reached the seg_store version window (" as *u8)
147 o = arx_catn(out, o, SS_VER_WINDOW)
148 o = arx_cat(out, o, "); the rows below are the NEWEST slice, NOT the whole dataset. Every statistic in this report describes that slice only. Reshard the data across <key>:0/<key>:1/... and use asr_analyze_stored_sharded.\n\n" as *u8)
149 }
150 if cap - o < 8192 { return o }
151 let r: i64 = ar_analyze(cols, field_names, nfields, rows, target_field, (((out as i64) + o) as *u8), cap - o)
152 if r <= 0 { return o }
153 return o + r
154}
155
156// THE SCALE FRONT DOOR: same analysis, but over rows spread across <keybase>:0, <keybase>:1, ... so the
157// dataset size is bounded by n_max and the shard budget rather than by one key's version window.
158func asr_analyze_stored_sharded(prefix: *u8, keybase: *u8, field_names: *i64, nfields: i64, target_field: i64, n_max: i64, maxshards: i64, out: *u8, cap: i64) -> i64 {
159 if nfields <= 0 { return 0 }
160 if nfields > 32 { return 0 }
161 let cols: *i64 = sys_mmap(8 * nfields) as *i64
162 let fl: *i64 = sys_mmap(8) as *i64
163 fl[0] = 0
164 var rows: i64 = 0
165 var f: i64 = 0
166 while f < nfields {
167 let colf: *i64 = sys_mmap(8 * n_max) as *i64
168 rows = asr_load_field_sharded(prefix, keybase, f, colf, n_max, maxshards, fl)
169 cols[f] = colf as i64
170 f = f + 1
171 }
172 if rows <= 0 { return 0 }
173 var o: i64 = 0
174 if fl[0] == 1 {
175 o = arx_cat(out, o, "!! INCOMPLETE SHARDED LOAD -- a shard reached the version window, or the shard budget ran out with rows still unread. The rows below are a SUBSET. Reshard smaller or raise maxshards.\n\n" as *u8)
176 }
177 if cap - o < 8192 { return o }
178 let r: i64 = ar_analyze(cols, field_names, nfields, rows, target_field, (((out as i64) + o) as *u8), cap - o)
179 if r <= 0 { return o }
180 return o + r
181}