nx_warc_index.nx source
↩ module page · 94 lines · 5482 B
1// nx_warc_index.nx -- C3: the WARC -> INVERTED-INDEX ingest pipeline = the heart of "ingest Common Crawl into our
2// index". Walks WARC 'response' records (Common Crawl / ISO 28500), extracts each record's content, and two-pass
3// indexes it into nx_search_inverted -> queryable. This is what runs at NAS scale over CC WET/WARC segments (composing
4// gz-inflate + posting-compression + sharding + write-read-decoupling, all gated separately). The WARC record walk is
5// inlined here (byte-identical framing to the canonical nx_warc_reader) to keep a single clean syscalls lineage --
6// nx_warc_reader carries its own main() + the nx_syscalls/syscalls.nx alias, and importing it alongside
7// nx_search_inverted trips the large-graph double-import/double-main limitation (filed). Built + gated NOW on an
8// in-memory WARC fixture; the live CC fetch + gz-inflate run on the NAS. license_tier: ORIGINAL
9import "nx_search_inverted.nx"
10
11func wi_lc(c: i64) -> i64 { if c >= 0x41 { if c <= 0x5a { return c + 0x20 } } return c }
12func wi_starts(buf: *u8, off: i64, n: i64, lit: *u8, litlen: i64) -> i64 {
13 if off + litlen > n { return 0 }
14 var i: i64 = 0; while i < litlen { if (buf[off+i] as i64) != (lit[i] as i64) { return 0 } i = i + 1 }
15 return 1
16}
17func wi_starts_ci(buf: *u8, off: i64, n: i64, lit: *u8, litlen: i64) -> i64 {
18 if off + litlen > n { return 0 }
19 var i: i64 = 0; while i < litlen { if wi_lc(buf[off+i] as i64) != (lit[i] as i64) { return 0 } i = i + 1 }
20 return 1
21}
22func wi_blankline(buf: *u8, off: i64, n: i64) -> i64 {
23 var i: i64 = off
24 while i + 3 < n { if (buf[i] as i64)==0x0d { if (buf[i+1] as i64)==0x0a { if (buf[i+2] as i64)==0x0d { if (buf[i+3] as i64)==0x0a { return i } } } } i = i + 1 }
25 return 0 - 1
26}
27func wi_uint(buf: *u8, off: i64, n: i64) -> i64 {
28 var v: i64 = 0; var i: i64 = off; var dg: i64 = 1
29 while dg == 1 { if i >= n { dg = 0 } else { let c: i64 = buf[i] as i64; if c < 0x30 { dg = 0 } else { if c > 0x39 { dg = 0 } else { v = v*10 + (c - 0x30); i = i + 1 } } } }
30 return v
31}
32// find header `name` (lowercase) in [hstart,hend); out2[0]=value_off, out2[1]=value_len; returns 1 if found.
33func wi_header_val(buf: *u8, hstart: i64, hend: i64, n: i64, name: *u8, namelen: i64, out2: *i64) -> i64 {
34 var i: i64 = hstart; var linestart: i64 = 1
35 while i < hend {
36 if linestart == 1 { if wi_starts_ci(buf, i, n, name, namelen) == 1 { if (buf[i+namelen] as i64) == 0x3a {
37 var vs: i64 = i + namelen + 1
38 var sp: i64 = 1; while sp == 1 { if vs < hend { if (buf[vs] as i64)==0x20 { vs=vs+1 } else { sp=0 } } else { sp=0 } }
39 var ve: i64 = vs; var fe: i64 = 0
40 while fe == 0 { if ve < hend { if (buf[ve] as i64)==0x0d { fe=1 } else { ve=ve+1 } } else { fe=1 } }
41 out2[0] = vs; out2[1] = ve - vs; return 1
42 } } }
43 if (buf[i] as i64) == 0x0a { linestart = 1 } else { linestart = 0 }
44 i = i + 1
45 }
46 return 0
47}
48// walk one WARC record at pos; fields[0/1]=type off/len, [2/3]=uri off/len, [4/5]=content off/len. returns next pos, -1 at end.
49func wi_next_record(buf: *u8, n: i64, pos: i64, fields: *i64) -> i64 {
50 var p: i64 = pos
51 var sk: i64 = 1
52 while sk == 1 { if p < n { let c: i64 = buf[p] as i64; if c==0x0d { p=p+1 } else { if c==0x0a { p=p+1 } else { sk=0 } } } else { sk=0 } }
53 if p >= n { return 0 - 1 }
54 if wi_starts(buf, p, n, "WARC/" as *u8, 5) == 0 { return 0 - 1 }
55 let bl: i64 = wi_blankline(buf, p, n)
56 if bl < 0 { return 0 - 1 }
57 let hend: i64 = bl + 2
58 let b2: *i64 = sys_mmap(32) as *i64
59 if wi_header_val(buf, p, hend, n, "warc-type" as *u8, 9, b2) == 1 { fields[0]=b2[0]; fields[1]=b2[1] } else { fields[0]=0; fields[1]=0 }
60 if wi_header_val(buf, p, hend, n, "warc-target-uri" as *u8, 15, b2) == 1 { fields[2]=b2[0]; fields[3]=b2[1] } else { fields[2]=0; fields[3]=0 }
61 var clen: i64 = 0
62 if wi_header_val(buf, p, hend, n, "content-length" as *u8, 14, b2) == 1 { clen = wi_uint(buf, b2[0], n) }
63 fields[4] = bl + 4
64 fields[5] = clen
65 var np: i64 = bl + 4 + clen
66 var sk2: i64 = 1
67 while sk2 == 1 { if np < n { let c: i64 = buf[np] as i64; if c==0x0d { np=np+1 } else { if c==0x0a { np=np+1 } else { sk2=0 } } } else { sk2=0 } }
68 return np
69}
70func wi_type_is(buf: *u8, off: i64, len: i64, lit: *u8, litlen: i64) -> i64 {
71 if len != litlen { return 0 }
72 var i: i64 = 0; while i < litlen { if wi_lc(buf[off+i] as i64) != (lit[i] as i64) { return 0 } i = i + 1 }
73 return 1
74}
75// INGEST: index every 'response' record's content into idx (two-pass). fills coff/clen[ndoc]; returns ndoc.
76func wi_build_index(warcbuf: *u8, n: i64, idx: *NxInvIndex, coff: *i64, clen: *i64, maxdoc: i64) -> i64 {
77 let fields: *i64 = sys_mmap(8*6) as *i64
78 var pos: i64 = 0; var ndoc: i64 = 0; var go: i64 = 1
79 while go == 1 {
80 let nextp: i64 = wi_next_record(warcbuf, n, pos, fields)
81 if nextp < 0 { go = 0 } else {
82 if wi_type_is(warcbuf, fields[0], fields[1], "response" as *u8, 8) == 1 {
83 if ndoc < maxdoc { coff[ndoc] = fields[4]; clen[ndoc] = fields[5]; ndoc = ndoc + 1 }
84 }
85 pos = nextp
86 }
87 }
88 var i: i64 = 0
89 while i < ndoc { nx_inv_index_row(idx, ((warcbuf as i64) + coff[i]) as *u8, clen[i], i); i = i + 1 }
90 nx_inv_finalize_offsets(idx)
91 i = 0
92 while i < ndoc { nx_inv_emit_row(idx, ((warcbuf as i64) + coff[i]) as *u8, clen[i], i); i = i + 1 }
93 return ndoc
94}