rete_core/dict.rs
1//! Front-coded, restart-indexed dictionary sections (SPEC.md §5.1).
2//!
3//! A section holds the UTF-8 terms of one kind (shared / subjects / objects /
4//! predicates / graphs), sorted and assigned dense 1-based IDs. Terms are stored
5//! in runs of `R`; each run starts with a full term and front-codes the rest.
6
7use std::sync::atomic::{AtomicBool, Ordering};
8use std::sync::Arc;
9
10use crate::chunk_cache::{CacheEntry, ChunkCache};
11use crate::varint::{read_uvarint, write_uvarint};
12
13/// Default restart interval: a full term every `R` entries.
14pub const DEFAULT_RESTART_INTERVAL: u32 = 16;
15
16/// Reserved ID meaning "no such term".
17pub const ABSENT: u32 = 0;
18
19/// How many not-yet-resident chunks a windowed resolution
20/// ([`ChunkedSection::decode_and_walk`]) faults concurrently in one group. It
21/// bounds the transient set of decoded bodies held at once (this many `Arc`s,
22/// each ≤ one chunk), so the parallel decode's memory is a small constant on top
23/// of the `--memory-budget-mb` cache, independent of the window size or budget.
24/// Also the group size, so a larger value means fewer (larger) parallel
25/// dispatches — amortizing the per-dispatch cost on the tight-budget path where
26/// real decoding happens.
27#[cfg(feature = "parallel")]
28const PAR_DECODE_FANOUT: usize = 32;
29
30/// A small, dedicated rayon pool for parallel chunk decode. Separate from (and
31/// smaller than) the global pool so the number of decode worker threads — and
32/// thus their allocator arenas, the dominant driver of the resident-set bump
33/// under parallel decode — is a small constant, not the machine's full core
34/// count. Decode is memory-bandwidth/CPU bound and a handful of threads saturate
35/// it, so this bounds peak RSS without materially costing throughput. Built once.
36#[cfg(feature = "parallel")]
37fn decode_pool() -> &'static rayon::ThreadPool {
38 use std::sync::OnceLock;
39 static POOL: OnceLock<rayon::ThreadPool> = OnceLock::new();
40 POOL.get_or_init(|| {
41 let n = std::thread::available_parallelism()
42 .map(|n| n.get())
43 .unwrap_or(4)
44 .clamp(1, 8);
45 rayon::ThreadPoolBuilder::new()
46 .num_threads(n)
47 .thread_name(|i| format!("rete-decode-{i}"))
48 .build()
49 .expect("build dedicated chunk-decode thread pool")
50 })
51}
52
53#[derive(Debug, thiserror::Error)]
54pub enum DictError {
55 #[error("malformed dictionary section: {0}")]
56 Malformed(&'static str),
57}
58
59/// Build a dictionary section from terms (any order; sorted + deduped here).
60#[derive(Default)]
61pub struct DictSectionBuilder {
62 terms: Vec<String>,
63 restart_interval: u32,
64}
65
66/// Terms per restart run, overridable via `RETE_DICT_RESTART_INTERVAL`.
67///
68/// The interval sets random-access granularity as well as front-coding gain:
69/// chunks are cut on whole-run boundaries, so with very large literals (e.g.
70/// base64-embedded images) a single run dwarfs the 64 KiB chunk budget and one
71/// term lookup drags in ~`r` terms' worth of bytes. Setting `1` makes every term
72/// its own restart — direct seek, larger restart table, no front-coding (which
73/// is worthless for base64 anyway, since such terms share no prefix).
74pub fn env_restart_interval() -> u32 {
75 static R: std::sync::OnceLock<u32> = std::sync::OnceLock::new();
76 *R.get_or_init(|| {
77 std::env::var("RETE_DICT_RESTART_INTERVAL")
78 .ok()
79 .and_then(|v| v.parse::<u32>().ok())
80 .filter(|&v| v >= 1)
81 .unwrap_or(DEFAULT_RESTART_INTERVAL)
82 })
83}
84
85impl DictSectionBuilder {
86 pub fn new() -> Self {
87 Self {
88 terms: Vec::new(),
89 restart_interval: env_restart_interval(),
90 }
91 }
92
93 pub fn with_restart_interval(mut self, r: u32) -> Self {
94 assert!(r >= 1, "restart interval must be >= 1");
95 self.restart_interval = r;
96 self
97 }
98
99 pub fn push(&mut self, term: impl Into<String>) {
100 self.terms.push(term.into());
101 }
102
103 /// Serialize to bytes. Terms are sorted and deduped; the resulting IDs are
104 /// `1..=n` in sorted order.
105 pub fn build(mut self) -> Vec<u8> {
106 self.terms.sort_unstable();
107 self.terms.dedup();
108 let r = self.restart_interval as usize;
109 let n = self.terms.len();
110 let num_restarts = n.div_ceil(r);
111
112 // Encode body and capture each run's starting offset (relative to body).
113 let mut body = Vec::new();
114 let mut restart_offsets = Vec::with_capacity(num_restarts);
115 let mut prev = "";
116 for (i, term) in self.terms.iter().enumerate() {
117 if i % r == 0 {
118 restart_offsets.push(body.len() as u64);
119 // restart entry: shared = 0, full term
120 write_uvarint(&mut body, 0);
121 write_uvarint(&mut body, term.len() as u64);
122 body.extend_from_slice(term.as_bytes());
123 } else {
124 let shared = common_prefix_len(prev, term);
125 let suffix = &term.as_bytes()[shared..];
126 write_uvarint(&mut body, shared as u64);
127 write_uvarint(&mut body, suffix.len() as u64);
128 body.extend_from_slice(suffix);
129 }
130 prev = term;
131 }
132
133 // header || restart-offset table || body
134 let mut out = Vec::new();
135 write_uvarint(&mut out, n as u64);
136 write_uvarint(&mut out, self.restart_interval as u64);
137 write_uvarint(&mut out, num_restarts as u64);
138 for off in &restart_offsets {
139 write_uvarint(&mut out, *off);
140 }
141 out.extend_from_slice(&body);
142 out
143 }
144}
145
146/// Parsed section metadata (the restart table) — cache this once and reuse it
147/// across lookups instead of re-parsing the section header every time.
148#[derive(Debug, Clone)]
149pub struct SectionMeta {
150 pub term_count: u32,
151 pub restart_interval: u32,
152 /// Absolute offsets into the section bytes for each run start.
153 ///
154 /// `u64`, not `usize`: a section may exceed 4 GiB and wasm32 is a 32-bit
155 /// target, so a `usize` offset silently truncates there (see #70). A
156 /// dictionary carrying embedded media is exactly this case — WikiArt's is
157 /// 23.4 GB.
158 pub restart_offsets: Vec<u64>,
159}
160
161/// Parse only the header/restart table of a section.
162pub fn parse_meta(bytes: &[u8]) -> Result<SectionMeta, DictError> {
163 let mut pos = 0;
164 let take = |pos: &mut usize| -> Result<u64, DictError> {
165 let (v, n) =
166 read_uvarint(&bytes[*pos..]).ok_or(DictError::Malformed("truncated header"))?;
167 *pos += n;
168 Ok(v)
169 };
170 let term_count = take(&mut pos)? as u32;
171 let restart_interval = take(&mut pos)? as u32;
172 let num_restarts = take(&mut pos)? as usize;
173 if restart_interval == 0 {
174 return Err(DictError::Malformed("zero restart interval"));
175 }
176 // `num_restarts` is untrusted; each restart is ≥1 byte, so cap the
177 // pre-allocation at the buffer length to avoid an OOM on a bogus count.
178 let mut rel = Vec::with_capacity(num_restarts.min(bytes.len()));
179 for _ in 0..num_restarts {
180 rel.push(take(&mut pos)?);
181 }
182 let body_start = pos as u64;
183 Ok(SectionMeta {
184 term_count,
185 restart_interval,
186 restart_offsets: rel.into_iter().map(|o| body_start + o).collect(),
187 })
188}
189
190/// Decode the front-coded entry at `pos` *into* `buf`: truncate `buf` to the
191/// entry's shared-prefix length and append its suffix bytes. For a restart
192/// entry the stored shared length is 0, so the same decode works for both
193/// entry kinds. Returns the next entry's position; `None` on malformed bytes
194/// (including a shared length longer than the previous term). Allocation-free
195/// after `buf`'s first growth — this is the hot path of every term resolution.
196#[inline]
197fn entry_into(bytes: &[u8], pos: usize, buf: &mut Vec<u8>) -> Option<usize> {
198 let (shared, n1) = read_uvarint(bytes.get(pos..)?)?;
199 let p = pos + n1;
200 let (suf, n2) = read_uvarint(bytes.get(p..)?)?;
201 let start = p + n2;
202 let end = start
203 .checked_add(suf as usize)
204 .filter(|&e| e <= bytes.len())?;
205 if shared as usize > buf.len() {
206 return None;
207 }
208 buf.truncate(shared as usize);
209 buf.extend_from_slice(&bytes[start..end]);
210 Some(end)
211}
212
213/// Decode a run's restart (full-term) entry at `off` into `buf`.
214fn run_entry_into(bytes: &[u8], off: usize, buf: &mut Vec<u8>) -> Option<usize> {
215 buf.clear(); // a restart entry stands alone; stale prefix bytes must not leak
216 entry_into(bytes, off, buf)
217}
218
219/// Resolve `id` (1-based) to its term using cached metadata. Returns `None` on
220/// any inconsistency in untrusted metadata/bytes rather than panicking.
221pub fn section_term(bytes: &[u8], meta: &SectionMeta, id: u32) -> Option<String> {
222 if id == ABSENT || id > meta.term_count {
223 return None;
224 }
225 let idx = (id - 1) as usize;
226 let run = idx / meta.restart_interval as usize;
227 let steps = idx % meta.restart_interval as usize;
228 let mut buf = Vec::new();
229 let mut pos = run_entry_into(bytes, *meta.restart_offsets.get(run)? as usize, &mut buf)?;
230 for _ in 0..steps {
231 pos = entry_into(bytes, pos, &mut buf)?;
232 }
233 Some(String::from_utf8_lossy(&buf).into_owned())
234}
235
236/// Resolve `term` to its ID using cached metadata.
237pub fn section_id(bytes: &[u8], meta: &SectionMeta, term: &str) -> Option<u32> {
238 let mut buf = Vec::new();
239 // Binary search restart runs by their first (full) term.
240 let mut lo = 0usize;
241 let mut hi = meta.restart_offsets.len();
242 while lo < hi {
243 let mid = (lo + hi) / 2;
244 run_entry_into(bytes, meta.restart_offsets[mid] as usize, &mut buf)?;
245 if buf.as_slice() <= term.as_bytes() {
246 lo = mid + 1;
247 } else {
248 hi = mid;
249 }
250 }
251 if lo == 0 {
252 return None; // smaller than every term
253 }
254 let run = lo - 1;
255 let mut pos = run_entry_into(bytes, meta.restart_offsets[run] as usize, &mut buf)?;
256 let base_id = (run * meta.restart_interval as usize) as u32 + 1;
257 // saturating_sub: corrupt metadata where run*interval > term_count must not
258 // underflow-panic.
259 let run_len = meta.restart_interval.min(
260 meta.term_count
261 .saturating_sub(run as u32 * meta.restart_interval),
262 );
263 for step in 0..run_len {
264 if buf.as_slice() == term.as_bytes() {
265 return Some(base_id + step);
266 }
267 if buf.as_slice() > term.as_bytes() {
268 return None;
269 }
270 if step + 1 < run_len {
271 pos = entry_into(bytes, pos, &mut buf)?;
272 }
273 }
274 None
275}
276
277// --- chunked sections ---------------------------------------------------------
278//
279// A section's body is split (on run boundaries) into **chunks** that are
280// compressed and fetched independently (format v0.2), so a remote client
281// resolves a term with one chunk fault instead of downloading the whole
282// section. Local sections are the degenerate case: one pre-set chunk holding
283// the entire serialized section — a single code path serves both.
284
285/// Fetches one chunk's decompressed body slice on demand (`None` = the fetch
286/// failed; the section records it and the lookup misses — callers over remote
287/// data must check [`ChunkedSection::load_incomplete`] after evaluating).
288pub type ChunkLoader = Box<dyn Fn(usize) -> Option<Vec<u8>> + Send + Sync>;
289
290/// Fetches **many** chunks in one round trip: given ascending chunk indices,
291/// returns each chunk's decompressed body in the same order. The ranged
292/// reader implements this by coalescing byte-adjacent chunk ranges into
293/// single range reads — a full-dictionary sweep (export, dump) costs a few
294/// requests per section instead of one per chunk. `None` = the batch failed;
295/// callers fall back to the per-chunk [`ChunkLoader`].
296pub type ChunkBulkLoader = Box<dyn Fn(&[usize]) -> Option<Vec<Vec<u8>>> + Send + Sync>;
297
298/// One chunk: a run-aligned slice of the section body. `body_start` is the
299/// offset (in the section's coordinate space — the same space
300/// [`SectionMeta::restart_offsets`] uses) where `data[0]` sits.
301pub struct SectionChunk {
302 first_run: usize,
303 /// The chunk's **routing key** — the only thing the chunk-level binary
304 /// search in [`ChunkedSection::id`] compares against. It is a *separator*,
305 /// **not a term**:
306 ///
307 /// ```text
308 /// last_term(chunk i-1) < key(i) <= first_term(chunk i)
309 /// ```
310 ///
311 /// and chunk 0's key is empty (`b"" <= anything`, and nothing routes before
312 /// chunk 0 anyway). Writers store the *shortest* such string — see
313 /// [`shortest_separator`] — which is why a file's chunk directory is a few
314 /// bytes per chunk instead of a full copy of every boundary term.
315 ///
316 /// Files written before that change carry the boundary term verbatim; the
317 /// verbatim first term is the degenerate separator, so both read the same
318 /// way and no version check is needed. Nothing may reconstruct a term from
319 /// this field, compare it for equality with a term, or report it as one: a
320 /// key that is *not* a separator (a truncation, a hash) mis-routes silently
321 /// — `term(id)` and every dump keep working because they route by
322 /// `first_run`, and only `id(term)` lies.
323 ///
324 /// Unused (empty) for the single local chunk.
325 key: Vec<u8>,
326 body_start: u64,
327}
328
329impl SectionChunk {
330 /// A chunk descriptor: routing metadata only. The decompressed body (and
331 /// its derived run-offset table) lives in the section's [`ChunkCache`],
332 /// keyed by `(section_index, chunk_index)`, faulted in on first touch and
333 /// evictable — not held here. `key` is the routing separator, **not a
334 /// term** (see the `key` field).
335 pub fn new(first_run: usize, key: Vec<u8>, body_start: u64) -> Self {
336 SectionChunk {
337 first_run,
338 key,
339 body_start,
340 }
341 }
342}
343
344/// Scan a chunk's decompressed body for the byte offset of each run start (see
345/// [`SectionChunk::run_offsets`]). O(terms-in-chunk), run once per faulted chunk.
346fn chunk_run_offsets(data: &[u8], restart_interval: usize) -> Vec<usize> {
347 let mut offs = vec![0usize];
348 let mut pos = 0usize;
349 let mut buf = Vec::new();
350 let mut count = 0usize;
351 while pos < data.len() {
352 let Some(next) = entry_into(data, pos, &mut buf) else {
353 break;
354 };
355 count += 1;
356 pos = next;
357 if pos < data.len() && count.is_multiple_of(restart_interval) {
358 offs.push(pos);
359 }
360 }
361 offs
362}
363
364/// The first (full) term of the run starting at `off`, as raw bytes. `None`
365/// on malformed bytes.
366pub fn run_first_term(bytes: &[u8], off: usize) -> Option<Vec<u8>> {
367 let mut buf = Vec::new();
368 run_entry_into(bytes, off, &mut buf)?;
369 Some(buf)
370}
371
372/// The LAST term encoded in `bytes[run_off..end]`, where `run_off` is a run
373/// (restart) boundary and `end` is the end of that run's chunk. Walks the
374/// front-coded entries forward, so it decodes at most `restart_interval` of
375/// them. `None` on malformed bytes.
376///
377/// Writers use it for one thing: the previous chunk's last term, the lower
378/// bound of the next chunk's [`shortest_separator`].
379pub fn run_last_term(bytes: &[u8], run_off: usize, end: usize) -> Option<Vec<u8>> {
380 let lim = bytes.get(..end)?;
381 let mut buf = Vec::new();
382 let mut pos = run_entry_into(lim, run_off, &mut buf)?;
383 while pos < end {
384 pos = entry_into(lim, pos, &mut buf)?;
385 }
386 Some(buf)
387}
388
389/// The **shortest** byte string `s` with `prev_last < s <= first`: `first`
390/// truncated one byte past where it first differs from `prev_last`.
391///
392/// This is the routing key a chunk directory stores (`SectionChunk::key`).
393/// `first` is the chunk's first term and `prev_last`
394/// the previous chunk's last term, so `prev_last < first` holds for any sorted
395/// section; if it does not (malformed input), the full `first` is returned —
396/// always a valid separator, just not a short one.
397///
398/// It really is the shortest. Let `k` be the length of the common prefix of
399/// `prev_last` and `first`; the result has length `k + 1`. No shorter `s`
400/// exists: any `s` with `|s| <= k` shares that prefix's bytes with both, so it
401/// is either a prefix of `prev_last` (hence `<= prev_last`), or it first
402/// differs from the shared prefix — below it, making `s < prev_last`, or above
403/// it, making `s > first`. Every case is outside `(prev_last, first]`.
404pub fn shortest_separator(prev_last: &[u8], first: &[u8]) -> Vec<u8> {
405 let shared = prev_last
406 .iter()
407 .zip(first)
408 .take_while(|(a, b)| a == b)
409 .count();
410 if shared >= first.len() {
411 // `first` is a prefix of `prev_last`, i.e. `first <= prev_last`: not a
412 // sorted pair. Keep the whole term.
413 return first.to_vec();
414 }
415 first[..shared + 1].to_vec()
416}
417
418/// A dictionary section whose body is served chunk-by-chunk: metadata + chunk
419/// directory always present, chunk bytes local or faulted in on first touch.
420pub struct ChunkedSection {
421 meta: SectionMeta,
422 chunks: Vec<SectionChunk>,
423 loader: Option<ChunkLoader>,
424 bulk: Option<ChunkBulkLoader>,
425 failed: AtomicBool,
426 /// Decompressed chunk bodies (and their derived run tables) live here, keyed
427 /// by `(section_index, chunk_index)`, faulted in on first touch and
428 /// evictable. Shared across a [`Dictionary`]'s four sections. Phase 0 caps
429 /// it at unlimited, so residency is identical to the old per-chunk
430 /// `OnceLock`s.
431 cache: Arc<ChunkCache>,
432 /// This section's discriminant in the shared cache's key space (0 shared,
433 /// 1 subject-only, 2 object-only, 3 predicates).
434 section_index: u8,
435}
436
437impl ChunkedSection {
438 /// A local section: the whole serialized section (header + body) as one
439 /// resident chunk at coordinate 0, so the absolute restart offsets index it
440 /// directly. Malformed bytes degrade to an empty section (no panics on
441 /// untrusted files), matching the previous reader behavior. The body is
442 /// stored in `cache` at `(section_index, 0)` (unlimited cap ⇒ resident for
443 /// the section's lifetime, exactly as before).
444 pub fn local(section_bytes: Vec<u8>, cache: Arc<ChunkCache>, section_index: u8) -> Self {
445 let meta = parse_meta(§ion_bytes).unwrap_or(SectionMeta {
446 term_count: 0,
447 restart_interval: 1,
448 restart_offsets: Vec::new(),
449 });
450 cache.insert((section_index, 0), Arc::from(section_bytes));
451 ChunkedSection {
452 meta,
453 chunks: vec![SectionChunk::new(0, Vec::new(), 0)],
454 loader: None,
455 bulk: None,
456 failed: AtomicBool::new(false),
457 cache,
458 section_index,
459 }
460 }
461
462 /// A section from parsed parts: metadata + chunk list, with an optional
463 /// loader for non-resident chunks (the remote lazy-open path) — resident
464 /// chunk lists (a locally-decoded chunked section) supply their bodies
465 /// through `resident` instead.
466 pub fn from_parts(
467 meta: SectionMeta,
468 chunks: Vec<SectionChunk>,
469 loader: Option<ChunkLoader>,
470 cache: Arc<ChunkCache>,
471 section_index: u8,
472 ) -> Self {
473 ChunkedSection {
474 meta,
475 chunks,
476 loader,
477 bulk: None,
478 failed: AtomicBool::new(false),
479 cache,
480 section_index,
481 }
482 }
483
484 /// A section whose chunk bodies are already decoded (the local chunked-file
485 /// open path): the descriptors and their bodies arrive together and the
486 /// bodies are seeded into the cache. No loader — every chunk is resident.
487 pub fn resident(
488 meta: SectionMeta,
489 chunks: Vec<(SectionChunk, Vec<u8>)>,
490 cache: Arc<ChunkCache>,
491 section_index: u8,
492 ) -> Self {
493 let descriptors = chunks
494 .into_iter()
495 .enumerate()
496 .map(|(ci, (chunk, body))| {
497 cache.insert((section_index, ci as u32), Arc::from(body));
498 chunk
499 })
500 .collect();
501 ChunkedSection {
502 meta,
503 chunks: descriptors,
504 loader: None,
505 bulk: None,
506 failed: AtomicBool::new(false),
507 cache,
508 section_index,
509 }
510 }
511
512 /// Attach a batched chunk fetcher (see [`ChunkBulkLoader`]): full-section
513 /// sweeps ([`prefetch_all`](Self::prefetch_all)) go through it instead of
514 /// faulting chunk by chunk.
515 pub fn with_bulk_loader(mut self, bulk: ChunkBulkLoader) -> Self {
516 self.bulk = Some(bulk);
517 self
518 }
519
520 /// Batch-fault every unloaded chunk through the bulk loader, if one is
521 /// attached and at least two chunks are missing. Callers about to sweep
522 /// the whole section (export/dump term resolution) call this once; a
523 /// failed batch leaves the chunks unloaded for the per-chunk loader to
524 /// retry (and record failures) lookup by lookup.
525 pub fn prefetch_all(&self) {
526 self.prefetch_chunks(&(0..self.chunks.len()).collect::<Vec<_>>());
527 }
528
529 /// Batch-fault a *specific* set of chunks (the subset a bounded query's
530 /// output touches) through the bulk loader, coalescing byte-adjacent ranges
531 /// into single reads. `cis` should be ascending and deduped; chunks already
532 /// resident are skipped. Like [`prefetch_all`](Self::prefetch_all) a single
533 /// missing chunk is left for the per-chunk loader, and a failed batch leaves
534 /// the chunks unloaded for that loader to retry (and record failures).
535 pub fn prefetch_chunks(&self, cis: &[usize]) {
536 let Some(bulk) = &self.bulk else { return };
537 let missing: Vec<usize> = cis
538 .iter()
539 .copied()
540 .filter(|&ci| {
541 ci < self.chunks.len() && !self.cache.contains((self.section_index, ci as u32))
542 })
543 .collect();
544 if missing.len() < 2 {
545 return;
546 }
547 if let Some(bodies) = bulk(&missing) {
548 if bodies.len() == missing.len() {
549 for (&ci, body) in missing.iter().zip(bodies) {
550 self.cache
551 .insert((self.section_index, ci as u32), Arc::from(body));
552 }
553 }
554 }
555 }
556
557 pub fn meta(&self) -> &SectionMeta {
558 &self.meta
559 }
560
561 pub fn term_count(&self) -> u32 {
562 self.meta.term_count
563 }
564
565 /// The shared chunk cache backing this section (all four sections of a
566 /// [`crate::dictionary::Dictionary`] share one instance) — the handle the
567 /// bounded-export budget (phase 2) sets a byte cap on.
568 pub(crate) fn cache(&self) -> &Arc<ChunkCache> {
569 &self.cache
570 }
571
572 /// Did any chunk fetch fail since this section was opened — or since the
573 /// last [`reset_load_failure`](Self::reset_load_failure)?
574 pub fn load_incomplete(&self) -> bool {
575 self.failed.load(Ordering::Relaxed)
576 }
577
578 /// Forget recorded fetch failures (see `GraphIndex::reset_load_failure` —
579 /// the per-query reset for resident sessions). Failed chunks were never
580 /// cached, so the next resolution retries them.
581 pub fn reset_load_failure(&self) {
582 self.failed.store(false, Ordering::Relaxed);
583 }
584
585 /// The cache entry for chunk `ci`, faulting its body in through the loader
586 /// on a miss. Returns an owned [`CacheEntry`] handle (an `Arc`) the caller
587 /// holds for the duration of one decode — valid even if a later eviction
588 /// drops the map slot (the handout invariant).
589 ///
590 /// A FAILED fetch records the failure and caches nothing (returns `None`),
591 /// so a later resolution retries the chunk — a transient network error must
592 /// not permanently poison a resident session. A local section with no
593 /// loader returns an empty body for a missing chunk, matching the old
594 /// `get_or_init(Vec::new)` behavior.
595 fn chunk_entry(&self, ci: usize) -> Option<Arc<CacheEntry>> {
596 let key = (self.section_index, ci as u32);
597 if let Some(e) = self.cache.get(key) {
598 return Some(e);
599 }
600 match &self.loader {
601 Some(load) => match load(ci) {
602 Some(bytes) => Some(self.cache.insert(key, Arc::from(bytes))),
603 None => {
604 self.failed.store(true, Ordering::Relaxed);
605 None
606 }
607 },
608 None => Some(self.cache.insert(key, Arc::from(Vec::new()))),
609 }
610 }
611
612 /// The chunk holding `run` (chunks ascend by `first_run`; the first chunk
613 /// always starts at run 0).
614 fn chunk_of_run(&self, run: usize) -> Option<usize> {
615 let i = self.chunks.partition_point(|c| c.first_run <= run);
616 i.checked_sub(1)
617 }
618
619 /// The chunk index holding term `id` (1-based, section-local), or `None` if
620 /// out of range. Mirrors the run→chunk math in [`term`](Self::term) so a
621 /// caller can group a set of output ids by chunk and batch-prefetch them.
622 pub fn chunk_of_id(&self, id: u32) -> Option<usize> {
623 if id == ABSENT || id > self.meta.term_count {
624 return None;
625 }
626 let run = (id - 1) as usize / self.meta.restart_interval as usize;
627 self.chunk_of_run(run)
628 }
629
630 /// The byte offset (into chunk `ci`'s decompressed body) of `run`.
631 /// Full/local sections use the section-wide restart table (unchanged
632 /// behavior); a *lite* remote section (empty `restart_offsets`) derives it
633 /// from the chunk itself — that per-chunk run table is cached inside the
634 /// `entry` (evicted with the body), so the open never holds the whole table.
635 fn run_off_in_chunk(
636 &self,
637 ci: usize,
638 run: usize,
639 entry: &CacheEntry,
640 ri: usize,
641 ) -> Option<usize> {
642 let chunk = &self.chunks[ci];
643 if self.meta.restart_offsets.is_empty() {
644 entry
645 .runs(|b| chunk_run_offsets(b, ri))
646 .get(run.checked_sub(chunk.first_run)?)
647 .copied()
648 } else {
649 self.meta
650 .restart_offsets
651 .get(run)?
652 .checked_sub(chunk.body_start)
653 .map(|o| o as usize)
654 }
655 }
656
657 /// One past the last run index held by chunk `ci` (its run range is
658 /// `[first_run, run_end)`).
659 fn run_end_of_chunk(&self, ci: usize, entry: &CacheEntry, ri: usize) -> usize {
660 if let Some(next) = self.chunks.get(ci + 1) {
661 return next.first_run;
662 }
663 let chunk = &self.chunks[ci];
664 if self.meta.restart_offsets.is_empty() {
665 chunk.first_run + entry.runs(|b| chunk_run_offsets(b, ri)).len()
666 } else {
667 self.meta.restart_offsets.len()
668 }
669 }
670
671 /// Resolve `id` (1-based) to its term. One chunk fault at most.
672 /// Chunk index holding term `id` — [`term`](Self::term)'s routing without
673 /// faulting the chunk. `None` for an absent/out-of-range id.
674 pub(crate) fn chunk_of_term(&self, id: u32) -> Option<usize> {
675 if id == ABSENT || id > self.meta.term_count {
676 return None;
677 }
678 self.chunk_of_run((id - 1) as usize / self.meta.restart_interval as usize)
679 }
680
681 pub fn term(&self, id: u32) -> Option<String> {
682 if id == ABSENT || id > self.meta.term_count {
683 return None;
684 }
685 let idx = (id - 1) as usize;
686 let ri = self.meta.restart_interval as usize;
687 let run = idx / ri;
688 let steps = idx % ri;
689 let ci = self.chunk_of_run(run)?;
690 let entry = self.chunk_entry(ci)?;
691 let bytes: &[u8] = entry.body();
692 let off = self.run_off_in_chunk(ci, run, &entry, ri)?;
693 let mut buf = Vec::new();
694 let mut pos = run_entry_into(bytes, off, &mut buf)?;
695 for _ in 0..steps {
696 pos = entry_into(bytes, pos, &mut buf)?;
697 }
698 Some(String::from_utf8_lossy(&buf).into_owned())
699 }
700
701 /// Resolve many ids of THIS section in one chunk-ordered, single-pass walk,
702 /// filling `out[slot]` for each `(id, slot)` in `jobs`.
703 ///
704 /// `jobs` is sorted in place by id. Because chunks and runs are *contiguous
705 /// ascending* id ranges, sorting by id also groups the jobs by chunk and by
706 /// run — so each chunk is faulted **once** and each run walked **once**
707 /// forward, picking off every requested id as the decode passes its
708 /// position. This is where the redundant per-id re-walk that
709 /// [`term`](Self::term) does (decode from the run start for *every* id)
710 /// disappears: within a run the walk advances monotonically across all the
711 /// window's ids that live there.
712 ///
713 /// For any id this fills `out[slot]` with exactly what [`term`](Self::term)
714 /// returns for that id (the same decoded bytes, the same
715 /// `String::from_utf8_lossy`), and leaves `out[slot]` untouched where
716 /// `term(id)` is `None` — an absent/out-of-range id, a failed chunk fault,
717 /// or malformed bytes. Duplicate ids each fill their own slot. It is the
718 /// batch twin of `term`, used by the export window
719 /// ([`Dictionary::resolve_window`](crate::dictionary::Dictionary::resolve_window))
720 /// so a window's scattered object ids touch each chunk once instead of
721 /// re-faulting and re-walking per row.
722 ///
723 /// [`crate::dictionary::Dictionary::resolve_window`]: the sole caller.
724 pub fn resolve_into(&self, jobs: &mut [(u32, usize)], out: &mut [Option<String>]) {
725 let ri = self.meta.restart_interval as usize;
726 if ri == 0 {
727 return;
728 }
729 let tc = self.meta.term_count;
730 // Sort by id: chunks and runs are contiguous ascending id ranges, so
731 // this groups jobs by chunk and by run in one pass.
732 jobs.sort_unstable_by_key(|&(id, _)| id);
733 // Partition the sorted jobs into contiguous per-chunk segments
734 // `(chunk, [start, end))`. Every job of a chunk is one contiguous slice
735 // (chunks are ascending id ranges, `jobs` is id-sorted). Absent (id
736 // `ABSENT` = 0) and out-of-range (`id > tc`) ids sort to the ends and
737 // belong to no chunk — they form no segment, so their slots stay None,
738 // exactly as `term(id) == None` leaves them.
739 let mut segments: Vec<(usize, usize, usize)> = Vec::new();
740 let mut i = 0;
741 while i < jobs.len() {
742 let id0 = jobs[i].0;
743 if id0 == ABSENT || id0 > tc {
744 i += 1;
745 continue;
746 }
747 let Some(ci) = self.chunk_of_run((id0 - 1) as usize / ri) else {
748 i += 1;
749 continue;
750 };
751 let start = i;
752 i += 1;
753 while i < jobs.len() {
754 let id = jobs[i].0;
755 if id == ABSENT || id > tc {
756 i += 1;
757 continue;
758 }
759 if self.chunk_of_run((id - 1) as usize / ri) != Some(ci) {
760 break; // first id of the next chunk
761 }
762 i += 1;
763 }
764 segments.push((ci, start, i));
765 }
766 if segments.is_empty() {
767 return;
768 }
769 self.decode_and_walk(&segments, jobs, out, ri, tc);
770 }
771
772 /// Fault each segment's chunk and fill its slots. The chunk decodes are
773 /// independent and CPU-bound (zstd), so with the `parallel` feature a group
774 /// whose chunks are **not yet resident** is decoded concurrently on a
775 /// bounded pool before its (serial) fill; without it, one at a time — the
776 /// exact behavior this refactor extracted. Either way each chunk is faulted
777 /// once and its slots filled in segment order, so the output is byte-for-byte
778 /// identical.
779 ///
780 /// **Parallel decode runs only where there is decode work.** A group whose
781 /// chunks are already resident (a generous `--memory-budget-mb` after the
782 /// window prefetch warmed them, or a chunk re-touched by a later window) is
783 /// walked serially with **no** parallel dispatch — so the single-pass /
784 /// large-budget export keeps its serial speed and the rayon fan-out only
785 /// pays for itself in the tight-budget regime, where the whole dictionary
786 /// does not fit and chunks are genuinely re-decoded. Skipping the dispatch
787 /// on cache-warm windows is what keeps a large budget from regressing.
788 ///
789 /// The bound matters for `--memory-budget-mb`: a decoding group holds at
790 /// most [`PAR_DECODE_FANOUT`] decoded bodies at once (a small constant,
791 /// independent of the window size or the budget) and each is held behind its
792 /// cache `Arc` for the group's fill — so even a budget that evicts a body
793 /// the moment it lands cannot force a re-decode before the fill consumes it.
794 /// The decode runs on a small dedicated pool ([`decode_pool`]), so the number
795 /// of worker threads (and their allocator arenas) is bounded regardless of
796 /// the machine's core count.
797 #[cfg(feature = "parallel")]
798 fn decode_and_walk(
799 &self,
800 segments: &[(usize, usize, usize)],
801 jobs: &[(u32, usize)],
802 out: &mut [Option<String>],
803 ri: usize,
804 tc: u32,
805 ) {
806 use rayon::prelude::*;
807 for group in segments.chunks(PAR_DECODE_FANOUT) {
808 let has_miss = group
809 .iter()
810 .any(|&(ci, _, _)| !self.cache.contains((self.section_index, ci as u32)));
811 if has_miss {
812 // Real decode work: fault this group's chunks in parallel on the
813 // bounded pool (order preserved so each entry lines up with its
814 // segment), then fill serially. A failed fault yields `None`
815 // (recorded in `self.failed`), same as the serial path.
816 let entries: Vec<Option<Arc<CacheEntry>>> = decode_pool().install(|| {
817 group
818 .par_iter()
819 .map(|&(ci, _, _)| self.chunk_entry(ci))
820 .collect()
821 });
822 for (&(ci, s, e), entry) in group.iter().zip(entries) {
823 if let Some(entry) = entry {
824 self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
825 }
826 }
827 } else {
828 // Cache-warm: no decode to parallelize, so avoid the dispatch.
829 for &(ci, s, e) in group {
830 if let Some(entry) = self.chunk_entry(ci) {
831 self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
832 }
833 }
834 }
835 }
836 }
837
838 /// Serial fallback for builds without the `parallel` feature (wasm, and
839 /// `--no-default-features`): fault and walk one chunk at a time.
840 #[cfg(not(feature = "parallel"))]
841 fn decode_and_walk(
842 &self,
843 segments: &[(usize, usize, usize)],
844 jobs: &[(u32, usize)],
845 out: &mut [Option<String>],
846 ri: usize,
847 tc: u32,
848 ) {
849 for &(ci, s, e) in segments {
850 if let Some(entry) = self.chunk_entry(ci) {
851 self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
852 }
853 }
854 }
855
856 /// Fill the output slots owned by one chunk: `chunk_jobs` is the id-sorted
857 /// slice of `(id, slot)` jobs that all live in chunk `ci`, and `entry` is
858 /// that chunk's already-faulted body. Walks the chunk's front-coded runs
859 /// once each, forward, picking off every requested id as the decode passes
860 /// it — the single-pass twin of `term(id)`, writing exactly the bytes
861 /// `term(id)` would for each id and leaving a slot None where `term(id)` is
862 /// None (a malformed run entry, past which no larger id in that run decodes).
863 fn walk_chunk(
864 &self,
865 ci: usize,
866 entry: &CacheEntry,
867 chunk_jobs: &[(u32, usize)],
868 out: &mut [Option<String>],
869 ri: usize,
870 tc: u32,
871 ) {
872 let bytes: &[u8] = entry.body();
873 let mut k = 0;
874 while k < chunk_jobs.len() {
875 let id = chunk_jobs[k].0;
876 if id == ABSENT || id > tc {
877 k += 1;
878 continue;
879 }
880 let run = (id - 1) as usize / ri;
881 // Decode this run's restart entry. Any failure here means
882 // `term(id)` is None for the whole run — skip every job in it.
883 let run_start = self.run_off_in_chunk(ci, run, entry, ri).and_then(|off| {
884 let mut buf = Vec::new();
885 run_entry_into(bytes, off, &mut buf).map(|pos| (buf, pos))
886 });
887 let Some((mut buf, mut pos)) = run_start else {
888 while k < chunk_jobs.len() {
889 let id = chunk_jobs[k].0;
890 if id == ABSENT || id > tc {
891 k += 1;
892 continue;
893 }
894 if (id - 1) as usize / ri != run {
895 break;
896 }
897 k += 1;
898 }
899 continue;
900 };
901 // `buf` holds the run's term #`cur_step`; advance monotonically.
902 let mut cur_step = 0usize;
903 let mut broken = false;
904 while k < chunk_jobs.len() {
905 let id = chunk_jobs[k].0;
906 if id == ABSENT || id > tc {
907 k += 1;
908 continue;
909 }
910 if (id - 1) as usize / ri != run {
911 break; // next run
912 }
913 let steps = (id - 1) as usize % ri;
914 if !broken {
915 while cur_step < steps {
916 match entry_into(bytes, pos, &mut buf) {
917 Some(next) => {
918 pos = next;
919 cur_step += 1;
920 }
921 None => {
922 // Malformed past this point: `term(id)` is None
923 // here and for every larger id in the run (they
924 // all decode through this entry).
925 broken = true;
926 break;
927 }
928 }
929 }
930 if !broken && cur_step == steps {
931 out[chunk_jobs[k].1] = Some(String::from_utf8_lossy(&buf).into_owned());
932 }
933 }
934 k += 1;
935 }
936 }
937 }
938
939 /// Resolve `term` to its ID. Chunk-level binary search runs on the (local)
940 /// chunk directory, so this also costs at most one chunk fault.
941 pub fn id(&self, term: &str) -> Option<u32> {
942 if self.chunks.is_empty() {
943 return None;
944 }
945 let ri = self.meta.restart_interval as usize;
946 // Pick the chunk: the last one whose ROUTING KEY is <= `term` (the
947 // single local chunk skips the search — its key is unset).
948 //
949 // This is the ONLY read of [`SectionChunk::key`] in the codebase, and
950 // all it needs is the separator invariant `last(i-1) < key(i) <=
951 // first(i)`: `term` lands in chunk `i` exactly when `key(i) <= term <
952 // key(i+1)`, which covers every term of chunk `i` and nothing in any
953 // other chunk. A verbatim first term satisfies it too, which is why
954 // files written either way route identically.
955 //
956 // With separators the search can land on chunk `i` for a `term` in the
957 // gap `key(i) <= term < first(i)`; the run search below then hits its
958 // `lo == first_run` branch and answers `None`, which is correct — such
959 // a term is strictly between two chunks, i.e. absent.
960 let ci = if self.chunks.len() == 1 {
961 0
962 } else {
963 let i = self
964 .chunks
965 .partition_point(|c| c.key.as_slice() <= term.as_bytes());
966 i.checked_sub(1)?
967 };
968 let first_run = self.chunks[ci].first_run;
969 let entry = self.chunk_entry(ci)?;
970 let bytes: &[u8] = entry.body();
971
972 // Binary search this chunk's runs by their first (full) term.
973 let run_end = self.run_end_of_chunk(ci, &entry, ri);
974 let mut buf = Vec::new();
975 let mut lo = first_run;
976 let mut hi = run_end;
977 while lo < hi {
978 let mid = (lo + hi) / 2;
979 let off = self.run_off_in_chunk(ci, mid, &entry, ri)?;
980 run_entry_into(bytes, off, &mut buf)?;
981 if buf.as_slice() <= term.as_bytes() {
982 lo = mid + 1;
983 } else {
984 hi = mid;
985 }
986 }
987 if lo == first_run {
988 // Smaller than every term in (and before) this chunk. Unreachable
989 // when the key is a verbatim first term; reachable, and correct,
990 // when it is a separator: `key(ci) <= term < first(ci)` puts the
991 // term strictly between two chunks.
992 return None;
993 }
994 let run = lo - 1;
995 let off = self.run_off_in_chunk(ci, run, &entry, ri)?;
996 let mut pos = run_entry_into(bytes, off, &mut buf)?;
997 let base_id = (run * ri) as u32 + 1;
998 // saturating_sub: corrupt metadata must not underflow-panic.
999 let run_len = self.meta.restart_interval.min(
1000 self.meta
1001 .term_count
1002 .saturating_sub(run as u32 * self.meta.restart_interval),
1003 );
1004 for step in 0..run_len {
1005 if buf.as_slice() == term.as_bytes() {
1006 return Some(base_id + step);
1007 }
1008 if buf.as_slice() > term.as_bytes() {
1009 return None;
1010 }
1011 if step + 1 < run_len {
1012 pos = entry_into(bytes, pos, &mut buf)?;
1013 }
1014 }
1015 None
1016 }
1017
1018 /// The full serialized section (header + body), for the file writer. The
1019 /// local single-chunk case returns the stored bytes; a chunked section
1020 /// re-assembles them (header re-encoded from the metadata).
1021 pub fn raw_section_bytes(&self) -> Vec<u8> {
1022 if self.chunks.len() == 1 && self.chunks[0].body_start == 0 {
1023 if let Some(entry) = self.cache.get((self.section_index, 0)) {
1024 return entry.body().to_vec();
1025 }
1026 }
1027 let mut out = encode_section_header(&self.meta);
1028 for ci in 0..self.chunks.len() {
1029 if let Some(entry) = self.chunk_entry(ci) {
1030 out.extend_from_slice(entry.body());
1031 }
1032 }
1033 out
1034 }
1035}
1036
1037/// Re-encode a section header (term_count, interval, restart table) from its
1038/// parsed metadata — the inverse of [`parse_meta`]. Restart offsets are stored
1039/// body-relative on disk; `meta` holds them absolute, so the body start is
1040/// re-derived as the first run's offset.
1041pub fn encode_section_header(meta: &SectionMeta) -> Vec<u8> {
1042 let body_start = meta.restart_offsets.first().copied().unwrap_or(0);
1043 let mut out = Vec::new();
1044 write_uvarint(&mut out, meta.term_count as u64);
1045 write_uvarint(&mut out, meta.restart_interval as u64);
1046 write_uvarint(&mut out, meta.restart_offsets.len() as u64);
1047 for off in &meta.restart_offsets {
1048 write_uvarint(&mut out, off.saturating_sub(body_start));
1049 }
1050 out
1051}
1052
1053/// A parsed, read-only dictionary section (parses its metadata on construction).
1054pub struct DictSection<'a> {
1055 bytes: &'a [u8],
1056 meta: SectionMeta,
1057}
1058
1059impl<'a> DictSection<'a> {
1060 pub fn parse(bytes: &'a [u8]) -> Result<Self, DictError> {
1061 Ok(Self {
1062 bytes,
1063 meta: parse_meta(bytes)?,
1064 })
1065 }
1066
1067 pub fn len(&self) -> u32 {
1068 self.meta.term_count
1069 }
1070
1071 pub fn is_empty(&self) -> bool {
1072 self.meta.term_count == 0
1073 }
1074
1075 /// Resolve `id` (1-based) to its term, or `None` if out of range.
1076 pub fn term(&self, id: u32) -> Option<String> {
1077 section_term(self.bytes, &self.meta, id)
1078 }
1079
1080 /// Resolve `term` to its ID, or `None` if absent.
1081 pub fn id(&self, term: &str) -> Option<u32> {
1082 section_id(self.bytes, &self.meta, term)
1083 }
1084}
1085
1086fn common_prefix_len(a: &str, b: &str) -> usize {
1087 a.bytes().zip(b.bytes()).take_while(|(x, y)| x == y).count()
1088}
1089
1090#[cfg(test)]
1091mod tests {
1092 use super::*;
1093
1094 /// Build a *lite* (empty section-wide restart table), genuinely multi-chunk
1095 /// [`ChunkedSection`] from `terms`: the exact shape a ranged open produces,
1096 /// so `resolve_into`'s per-chunk run-table branch and its cross-chunk /
1097 /// cross-run walk are all exercised. `runs_per_chunk` chops the runs into
1098 /// several chunks regardless of size, so a tiny fixture still spans many.
1099 fn lite_multichunk(terms: &[&str], ri: u32, runs_per_chunk: usize) -> (ChunkedSection, u32) {
1100 let mut b = DictSectionBuilder::new().with_restart_interval(ri);
1101 for t in terms {
1102 b.push(*t);
1103 }
1104 let raw = b.build();
1105 let full_meta = parse_meta(&raw).unwrap();
1106 let n_runs = full_meta.restart_offsets.len();
1107 let mut chunks = Vec::new();
1108 let mut bodies: Vec<Vec<u8>> = Vec::new();
1109 let mut r = 0;
1110 while r < n_runs {
1111 let first_run = r;
1112 let start = full_meta.restart_offsets[r] as usize;
1113 let r2 = (r + runs_per_chunk).min(n_runs);
1114 let end = if r2 < n_runs {
1115 full_meta.restart_offsets[r2] as usize
1116 } else {
1117 raw.len()
1118 };
1119 chunks.push(SectionChunk::new(first_run, Vec::new(), start as u64));
1120 bodies.push(raw[start..end].to_vec());
1121 r = r2;
1122 }
1123 assert!(chunks.len() > 1, "fixture must be multi-chunk");
1124 // Lite: no section-wide restart table -> run offsets come from each
1125 // decoded chunk body, cached in its cache entry (the ranged path).
1126 let lite_meta = SectionMeta {
1127 term_count: full_meta.term_count,
1128 restart_interval: full_meta.restart_interval,
1129 restart_offsets: Vec::new(),
1130 };
1131 let loader: ChunkLoader = Box::new(move |ci| bodies.get(ci).cloned());
1132 let sec = ChunkedSection::from_parts(
1133 lite_meta,
1134 chunks,
1135 Some(loader),
1136 ChunkCache::unlimited_arc(),
1137 0,
1138 );
1139 (sec, full_meta.term_count)
1140 }
1141
1142 /// The windowed batch resolver must fill exactly what a per-id `term()`
1143 /// would, for a scrambled, duplicated, out-of-range id set spread across
1144 /// many chunks and runs — the byte-identity guarantee `dump_filtered_each`
1145 /// rests on (phase 1). Also cross-checked against the ground-truth
1146 /// `section_term` decoder.
1147 #[test]
1148 fn resolve_into_matches_term_across_chunks_and_runs() {
1149 // Shared prefixes (front-coding), a long literal (a run that dwarfs its
1150 // neighbors), blank nodes — 240 distinct terms.
1151 let owned: Vec<String> = (0..240)
1152 .map(|i| match i % 4 {
1153 0 => format!("<http://example.org/entity/{i:05}>"),
1154 1 => format!("<http://example.org/entity/{i:05}/sub/leaf>"),
1155 2 => format!("\"literal value {} padded {}\"", i, "x".repeat(i % 50)),
1156 _ => format!("_:b{i:05}"),
1157 })
1158 .collect();
1159 let terms: Vec<&str> = owned.iter().map(String::as_str).collect();
1160
1161 for (ri, rpc) in [(1u32, 5usize), (4, 3), (16, 2)] {
1162 let (sec, count) = lite_multichunk(&terms, ri, rpc);
1163 let raw = {
1164 let mut b = DictSectionBuilder::new().with_restart_interval(ri);
1165 for t in &terms {
1166 b.push(*t);
1167 }
1168 b.build()
1169 };
1170 let full_meta = parse_meta(&raw).unwrap();
1171
1172 // A scrambled probe set: every id once, a run of duplicates, some
1173 // out-of-range ids (0 and past the end) that must stay `None`.
1174 let mut probe: Vec<u32> = (1..=count).collect();
1175 probe.extend([1, 1, count, count, ABSENT, count + 3, count + 100]);
1176 // Deterministic shuffle so chunk order != probe order.
1177 let mut state = 0xDEAD_BEEF_1234_5678u64;
1178 for k in (1..probe.len()).rev() {
1179 state ^= state << 13;
1180 state ^= state >> 7;
1181 state ^= state << 17;
1182 probe.swap(k, (state % (k as u64 + 1)) as usize);
1183 }
1184
1185 let mut jobs: Vec<(u32, usize)> = probe.iter().map(|&id| (id, 0)).collect();
1186 for (slot, j) in jobs.iter_mut().enumerate() {
1187 j.1 = slot;
1188 }
1189 let mut out: Vec<Option<String>> = vec![None; probe.len()];
1190 sec.resolve_into(&mut jobs, &mut out);
1191
1192 for (slot, &id) in probe.iter().enumerate() {
1193 let via_term = sec.term(id);
1194 assert_eq!(
1195 out[slot], via_term,
1196 "resolve_into(id={id}) != term(id) at ri={ri} rpc={rpc}"
1197 );
1198 // Ground truth for in-range ids.
1199 let truth = section_term(&raw, &full_meta, id);
1200 assert_eq!(out[slot], truth, "resolve_into(id={id}) != section_term");
1201 }
1202 }
1203 }
1204
1205 fn sample() -> Vec<String> {
1206 // Deliberately unsorted, with shared prefixes and a duplicate.
1207 [
1208 "http://ex.org/Alice",
1209 "http://ex.org/Bob",
1210 "http://ex.org/Alan",
1211 "http://ex.org/knows",
1212 "http://ex.org/Alice", // dup
1213 "zeta",
1214 ]
1215 .iter()
1216 .map(|s| s.to_string())
1217 .collect()
1218 }
1219
1220 #[test]
1221 fn round_trip_all_ids_and_terms() {
1222 for r in [1u32, 2, 16, 1000] {
1223 let mut b = DictSectionBuilder::new().with_restart_interval(r);
1224 for t in sample() {
1225 b.push(t);
1226 }
1227 let bytes = b.build();
1228 let sec = DictSection::parse(&bytes).unwrap();
1229
1230 // Expected sorted unique set.
1231 let mut expected = sample();
1232 expected.sort();
1233 expected.dedup();
1234 assert_eq!(sec.len() as usize, expected.len());
1235
1236 for (i, term) in expected.iter().enumerate() {
1237 let id = (i + 1) as u32;
1238 assert_eq!(
1239 sec.term(id).as_deref(),
1240 Some(term.as_str()),
1241 "term({id}) r={r}"
1242 );
1243 assert_eq!(sec.id(term), Some(id), "id({term}) r={r}");
1244 }
1245 }
1246 }
1247
1248 #[test]
1249 fn lookups_for_absent_terms() {
1250 let mut b = DictSectionBuilder::new();
1251 for t in sample() {
1252 b.push(t);
1253 }
1254 let bytes = b.build();
1255 let sec = DictSection::parse(&bytes).unwrap();
1256 assert_eq!(sec.id("aaa-before-everything"), None);
1257 assert_eq!(sec.id("zzz-after-everything"), None);
1258 assert_eq!(sec.id("http://ex.org/Alic"), None); // prefix, not present
1259 assert_eq!(sec.term(0), None);
1260 assert_eq!(sec.term(9999), None);
1261 }
1262
1263 /// Property-style stress of the front-coded decode paths: a large
1264 /// deterministic-pseudo-random term pool (IRIs with heavy shared prefixes,
1265 /// literals, blank nodes — duplicates likely) must round-trip every id and
1266 /// term across restart intervals, and near-miss probes (a stored term ± a
1267 /// suffix/truncation, including run-boundary terms) must resolve to `None`.
1268 /// Term resolution is the engine's hot path, so this is the test that
1269 /// guards the buffer-reuse decode.
1270 #[test]
1271 fn randomized_round_trip_and_near_misses_across_restart_intervals() {
1272 let mut state = 0x9E37_79B9_7F4A_7C15u64;
1273 let mut next = move || {
1274 state ^= state << 13;
1275 state ^= state >> 7;
1276 state ^= state << 17;
1277 state
1278 };
1279 let pool: Vec<String> = (0..1500)
1280 .map(|i| {
1281 let n = next();
1282 match n % 4 {
1283 0 => format!("<http://example.org/entity/{n:x}>"),
1284 1 => format!("\"literal value {} with spaces\"", n % 300), // dups likely
1285 2 => format!("<http://example.org/entity/{}/sub/{i}>", n % 64), // long shared prefixes
1286 _ => format!("_:b{}", n % 256),
1287 }
1288 })
1289 .collect();
1290 let mut expected = pool.clone();
1291 expected.sort();
1292 expected.dedup();
1293
1294 for r in [1u32, 3, 16, 64] {
1295 let mut b = DictSectionBuilder::new().with_restart_interval(r);
1296 for t in &pool {
1297 b.push(t.clone());
1298 }
1299 let bytes = b.build();
1300 let sec = DictSection::parse(&bytes).unwrap();
1301 assert_eq!(sec.len() as usize, expected.len(), "r={r}");
1302
1303 for (i, term) in expected.iter().enumerate() {
1304 let id = (i + 1) as u32;
1305 assert_eq!(
1306 sec.term(id).as_deref(),
1307 Some(term.as_str()),
1308 "term({id}) r={r}"
1309 );
1310 assert_eq!(sec.id(term), Some(id), "id({term}) r={r}");
1311 }
1312 // Near misses around run boundaries (and a sample of the rest):
1313 // an appended suffix or a truncation is never a stored term unless
1314 // it happens to collide with one.
1315 for (i, term) in expected.iter().enumerate() {
1316 let near_boundary = (i as u32) % r <= 1;
1317 if !near_boundary && i % 37 != 0 {
1318 continue;
1319 }
1320 let longer = format!("{term}\u{1}");
1321 assert_eq!(sec.id(&longer), None, "near-miss long r={r}");
1322 let mut shorter = term.clone();
1323 shorter.pop();
1324 if !shorter.is_empty() && expected.binary_search(&shorter).is_err() {
1325 assert_eq!(sec.id(&shorter), None, "near-miss short {shorter:?} r={r}");
1326 }
1327 }
1328 assert_eq!(sec.term(expected.len() as u32 + 1), None, "past-end r={r}");
1329 }
1330 }
1331}